July 2020 - tor-commits - lists.torproject.org

[tor/master] Add unit tests for connection_describe().
by nickm＠torproject.org 16 Jul '20

16 Jul '20

commit 10d8555e7db2f4835b4fc9a0f4592e470a974c5f Author: Nick Mathewson <nickm(a)torproject.org> Date: Wed Jul 15 16:58:46 2020 -0400 Add unit tests for connection_describe(). --- src/test/test_connection.c | 107 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 107 insertions(+) diff --git a/src/test/test_connection.c b/src/test/test_connection.c index 7ced49a166..e9be04656b 100644 --- a/src/test/test_connection.c +++ b/src/test/test_connection.c @@ -10,6 +10,7 @@ #include "core/or/or.h" #include "test/test.h" +#include "app/config/config.h" #include "app/config/or_options_st.h" #include "core/mainloop/connection.h" #include "core/or/connection_edge.h" @@ -961,6 +962,111 @@ test_failed_orconn_tracker(void *arg) ; } +static void +test_conn_describe(void *arg) +{ + (void)arg; + or_options_t *options = get_options_mutable(); + options->SafeLogging_ = SAFELOG_SCRUB_ALL; + + // Let's start with a listener connection since they're simple. + connection_t *conn = connection_new(CONN_TYPE_OR_LISTENER, AF_INET); + tor_addr_parse(&conn->addr, "44.22.11.11"); + conn->port = 80; + tt_str_op(connection_describe(conn), OP_EQ, + "OR listener connection (ready) on 44.22.11.11:80"); + // If the address is unspec, we should still work. + tor_addr_make_unspec(&conn->addr); + tt_str_op(connection_describe(conn), OP_EQ, + "OR listener connection (ready) on <unset>:80"); + // Try making the address null. + tor_addr_make_null(&conn->addr, AF_INET); + tt_str_op(connection_describe(conn), OP_EQ, + "OR listener connection (ready) on 0.0.0.0:80"); + // What if the address is uninitialized? (This can happen if we log about the + // connection before we set the address.) + memset(&conn->addr, 0, sizeof(conn->addr)); + tt_str_op(connection_describe(conn), OP_EQ, + "OR listener connection (ready) on <unset>:80"); + connection_free_minimal(conn); + + // Try a unix socket. + conn = connection_new(CONN_TYPE_CONTROL_LISTENER, AF_UNIX); + conn->address = tor_strdup("/a/path/that/could/exist"); + tt_str_op(connection_describe(conn), OP_EQ, + "Control listener connection (ready) on /a/path/that/could/exist"); + connection_free_minimal(conn); + + // Try an IPv6 address. + conn = connection_new(CONN_TYPE_AP_LISTENER, AF_INET6); + tor_addr_parse(&conn->addr, "ff00::3"); + conn->port = 9050; + tt_str_op(connection_describe(conn), OP_EQ, + "Socks listener connection (ready) on [ff00::3]:9050"); + connection_free_minimal(conn); + + // Now let's mess with exit connections. They have some special issues. + options->SafeLogging_ = SAFELOG_SCRUB_NONE; + conn = connection_new(CONN_TYPE_EXIT, AF_INET); + // If address and state are unset, we should say SOMETHING. + tt_str_op(connection_describe(conn), OP_EQ, + "Exit connection (uninitialized) to <unset> (DNS lookup pending)"); + // Now suppose that the address is set but we haven't resolved the hostname. + conn->port = 443; + conn->address = tor_strdup("www.torproject.org"); + conn->state = EXIT_CONN_STATE_RESOLVING; + tt_str_op(connection_describe(conn), OP_EQ, + "Exit connection (waiting for dest info) to " + "www.torproject.org:443 (DNS lookup pending)"); + // Now give it a hostname! + tor_addr_parse(&conn->addr, "192.168.8.8"); + conn->state = EXIT_CONN_STATE_OPEN; + tt_str_op(connection_describe(conn), OP_EQ, + "Exit connection (open) to 192.168.8.8:443"); + // But what if safelogging is on? + options->SafeLogging_ = SAFELOG_SCRUB_RELAY; + tt_str_op(connection_describe(conn), OP_EQ, + "Exit connection (open) to [scrubbed]"); + connection_free_minimal(conn); + + // Now at last we look at OR addresses, which are complicated. + conn = connection_new(CONN_TYPE_OR, AF_INET6); + conn->state = OR_CONN_STATE_OPEN; + conn->port = 8080; + tor_addr_parse(&conn->addr, "[ffff:3333:1111::2]"); + // This should get scrubbed, since the lack of a set ID means we might be + // talking to a client. + tt_str_op(connection_describe(conn), OP_EQ, + "OR connection (open) with [scrubbed]"); + // But now suppose we aren't safelogging? We'll get the address then. + options->SafeLogging_ = SAFELOG_SCRUB_NONE; + tt_str_op(connection_describe(conn), OP_EQ, + "OR connection (open) with [ffff:3333:1111::2]:8080"); + // Suppose we have an ID, so we know it isn't a client. + TO_OR_CONN(conn)->identity_digest[3] = 7; + options->SafeLogging_ = SAFELOG_SCRUB_RELAY; // back to safelogging. + tt_str_op(connection_describe(conn), OP_EQ, + "OR connection (open) with [ffff:3333:1111::2]:8080 " + "ID=0000000700000000000000000000000000000000"); + // Add a 'real address' that is the same as the one we have. + tor_addr_parse(&TO_OR_CONN(conn)->real_addr, "[ffff:3333:1111::2]"); + tt_str_op(connection_describe(conn), OP_EQ, + "OR connection (open) with [ffff:3333:1111::2]:8080 " + "ID=0000000700000000000000000000000000000000"); + // Add a different 'real address' + tor_addr_parse(&TO_OR_CONN(conn)->real_addr, "[ffff:3333:1111::8]"); + tt_str_op(connection_describe(conn), OP_EQ, + "OR connection (open) with [ffff:3333:1111::8]:8080 " + "ID=0000000700000000000000000000000000000000 " + "canonical_addr=[ffff:3333:1111::2]"); + + // Clear identity_digest so that free_minimal won't complain. + memset(TO_OR_CONN(conn)->identity_digest, 0, DIGEST_LEN); + + done: + connection_free_minimal(conn); +} + #ifndef COCCI #define CONNECTION_TESTCASE(name, fork, setup) \ { #name, test_conn_##name, fork, &setup, NULL } @@ -995,5 +1101,6 @@ struct testcase_t connection_tests[] = { //CONNECTION_TESTCASE(func_suffix, TT_FORK, setup_func_pair), { "failed_orconn_tracker", test_failed_orconn_tracker, TT_FORK, NULL, NULL }, + { "describe", test_conn_describe, TT_FORK, NULL, NULL }, END_OF_TESTCASES };

1 0

[tor/master] Changes file for 40041.
by nickm＠torproject.org 16 Jul '20

16 Jul '20

commit ae8072e8ff57b7c809de825370518b6377e4478a Author: Nick Mathewson <nickm(a)torproject.org> Date: Tue Jul 14 14:41:21 2020 -0400 Changes file for 40041. Closes #40041. --- changes/ticket40041 | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/changes/ticket40041 b/changes/ticket40041 new file mode 100644 index 0000000000..cc680db7c5 --- /dev/null +++ b/changes/ticket40041 @@ -0,0 +1,9 @@ + o Minor features (logging): + - Provide more complete descriptions of our connections when logging + about them. Closes ticket 40041. + + o Code simplification and refactoring: + - Refactor our code that logs a descriptions of connections, channels, + and the peers on them, to use a single call path. This change + enables us to refactor the data types that they use, and eliminate + many confusing users of those types. Closes ticket 40041.

1 0

[onionperf/develop] Merge branch 'task-40005' into develop
by karsten＠torproject.org 16 Jul '20

16 Jul '20

commit a2ef62c2b7d1406f72ac87769ce0396f543146d8 Merge: c761ac6 c1cd60e Author: Karsten Loesing <karsten.loesing(a)gmx.net> Date: Thu Jul 16 09:55:36 2020 +0200 Merge branch 'task-40005' into develop onionperf/analysis.py | 76 ++++++++++++------------------------ onionperf/measurement.py | 2 +- onionperf/onionperf | 9 +---- onionperf/reprocessing.py | 8 ++-- onionperf/tests/test_reprocessing.py | 2 +- 5 files changed, 33 insertions(+), 64 deletions(-)

1 0

[onionperf/develop] Take out Tor summaries and the do_complete switch.
by karsten＠torproject.org 16 Jul '20

16 Jul '20

commit 95a2f8b482ee882b4ec898a7bbc76018a6d98659 Author: Karsten Loesing <karsten.loesing(a)gmx.net> Date: Tue Jul 14 10:14:22 2020 +0200 Take out Tor summaries and the do_complete switch. Implements tpo/metrics/onionperf#40005. --- onionperf/analysis.py | 70 ++++++++++++------------------------ onionperf/measurement.py | 2 +- onionperf/onionperf | 9 ++--- onionperf/reprocessing.py | 8 ++--- onionperf/tests/test_reprocessing.py | 2 +- 5 files changed, 30 insertions(+), 61 deletions(-) diff --git a/onionperf/analysis.py b/onionperf/analysis.py index 19c0192..26b00b2 100644 --- a/onionperf/analysis.py +++ b/onionperf/analysis.py @@ -29,13 +29,7 @@ class OPAnalysis(Analysis): def add_torctl_file(self, filepath): self.torctl_filepaths.append(filepath) - def get_tor_bandwidth_summary(self, node, direction): - try: - return self.json_db['data'][node]['tor']['bandwidth_summary'][direction] - except: - return None - - def analyze(self, do_complete=False, date_filter=None): + def analyze(self, date_filter=None): if self.did_analysis: return @@ -47,7 +41,7 @@ class OPAnalysis(Analysis): if len(filepaths) > 0: for filepath in filepaths: logging.info("parsing log file at {0}".format(filepath)) - parser.parse(util.DataSource(filepath), do_complete=do_complete) + parser.parse(util.DataSource(filepath), do_complete=True) if self.nickname is None: parsed_name = parser.get_name() @@ -134,7 +128,7 @@ class OPAnalysis(Analysis): class Parser(object, metaclass=ABCMeta): @abstractmethod - def parse(self, source, do_complete): + def parse(self, source, do_complete=True): pass @abstractmethod def get_data(self): @@ -270,14 +264,10 @@ class TorCtlParser(Parser): def __init__(self, date_filter=None): ''' date_filter should be given in UTC ''' - self.do_complete = False - self.bandwidth_summary = {'bytes_read':{}, 'bytes_written':{}} self.circuits_state = {} self.circuits = {} - self.circuits_summary = {'buildtimes':[], 'lifetimes':[]} self.streams_state = {} self.streams = {} - self.streams_summary = {'lifetimes':{}} self.name = None self.boot_succeeded = False self.build_timeout_last = None @@ -320,15 +310,10 @@ class TorCtlParser(Parser): data = circ.get_data() if data is not None: - if built is not None and started is not None and len(data['path']) == 3: - self.circuits_summary['buildtimes'].append(built - started) - if ended is not None and started is not None: - self.circuits_summary['lifetimes'].append(ended - started) - if self.do_complete: - self.circuits[cid] = data + self.circuits[cid] = data self.circuits_state.pop(cid) - elif self.do_complete and isinstance(event, CircMinorEvent): + elif isinstance(event, CircMinorEvent): if event.purpose != event.old_purpose or event.event != CircEvent.PURPOSE_CHANGED: key = "{0}:{1}".format(event.event, event.purpose) circ.add_event(key, arrival_dt) @@ -364,15 +349,9 @@ class TorCtlParser(Parser): data = strm.get_data() if data is not None: - if self.do_complete: - self.streams[sid] = data - self.streams_summary['lifetimes'].setdefault(stream_type, []).append(ended - started) + self.streams[sid] = data self.streams_state.pop(sid) - def __handle_bw(self, event, arrival_dt): - self.bandwidth_summary['bytes_read'][int(arrival_dt)] = event.read - self.bandwidth_summary['bytes_written'][int(arrival_dt)] = event.written - def __handle_buildtimeout(self, event, arrival_dt): self.build_timeout_last = event.timeout self.build_quantile_last = event.quantile @@ -382,8 +361,6 @@ class TorCtlParser(Parser): self.__handle_circuit(event, arrival_dt) elif isinstance(event, StreamEvent): self.__handle_stream(event, arrival_dt) - elif isinstance(event, BandwidthEvent): - self.__handle_bw(event, arrival_dt) elif isinstance(event, BuildTimeoutSetEvent): self.__handle_buildtimeout(event, arrival_dt) @@ -408,27 +385,26 @@ class TorCtlParser(Parser): elif re.search("BOOTSTRAP", line) is not None and re.search("PROGRESS=100", line) is not None: self.boot_succeeded = True - if self.do_complete or (self.do_complete is False and re.search("650\sBW", line) is not None): - # parse with stem - timestamps, sep, raw_event_str = line.partition(" 650 ") - if sep == '': - return True + # parse with stem + timestamps, sep, raw_event_str = line.partition(" 650 ") + if sep == '': + return True + + # event.arrived_at is also available but at worse granularity + unix_ts = float(timestamps.strip().split()[2]) - # event.arrived_at is also available but at worse granularity - unix_ts = float(timestamps.strip().split()[2]) + # check if we should ignore the line + line_date = datetime.datetime.utcfromtimestamp(unix_ts).date() + if not self.__is_date_valid(line_date): + return True - # check if we should ignore the line - line_date = datetime.datetime.utcfromtimestamp(unix_ts).date() - if not self.__is_date_valid(line_date): - return True + event = ControlMessage.from_str("{0} {1}".format(sep.strip(), raw_event_str)) + convert('EVENT', event) + self.__handle_event(event, unix_ts) - event = ControlMessage.from_str("{0} {1}".format(sep.strip(), raw_event_str)) - convert('EVENT', event) - self.__handle_event(event, unix_ts) return True - def parse(self, source, do_complete=False): - self.do_complete = do_complete + def parse(self, source, do_complete=True): source.open(newline='\r\n') for line in source: # ignore line parsing errors @@ -442,9 +418,7 @@ class TorCtlParser(Parser): source.close() def get_data(self): - return {'circuits': self.circuits, 'circuits_summary': self.circuits_summary, - 'streams':self.streams, 'streams_summary': self.streams_summary, - 'bandwidth_summary': self.bandwidth_summary} + return {'circuits': self.circuits, 'streams': self.streams} def get_name(self): return self.name diff --git a/onionperf/measurement.py b/onionperf/measurement.py index 9d84d4e..823f37a 100644 --- a/onionperf/measurement.py +++ b/onionperf/measurement.py @@ -157,7 +157,7 @@ def logrotate_thread_task(writables, tgen_writable, torctl_writable, docroot, ni anal.add_torctl_file(torctl_writable.rotate_file(filename_datetime=next_midnight)) # run the analysis, i.e. parse the files - anal.analyze(do_simple=False) + anal.analyze() # save the results in onionperf json format in the www docroot anal.save(output_prefix=docroot, do_compress=True, date_prefix=next_midnight.date()) diff --git a/onionperf/onionperf b/onionperf/onionperf index 5e2c2fb..36520ad 100755 --- a/onionperf/onionperf +++ b/onionperf/onionperf @@ -279,11 +279,6 @@ files generated by this script will be written""", action="store", dest="date_prefix", default=None) - analyze_parser.add_argument('-s', '--do-simple-parse', - help="""parse and export only summary statistics rather than full transfer/circuit/stream data""", - action="store_false", dest="do_complete", - default=True) - # visualize visualize_parser = sub_parser.add_parser('visualize', description=DESC_VISUALIZE, help=HELP_VISUALIZE, formatter_class=my_formatter_class) @@ -387,7 +382,7 @@ def analyze(args): analysis.add_tgen_file(args.tgen_logpath) if args.torctl_logpath is not None: analysis.add_torctl_file(args.torctl_logpath) - analysis.analyze(args.do_complete, date_filter=args.date_filter) + analysis.analyze(date_filter=args.date_filter) analysis.save(output_prefix=args.prefix, date_prefix=args.date_prefix) elif args.tgen_logpath is not None and os.path.isdir(args.tgen_logpath) and args.torctl_logpath is not None and os.path.isdir(args.torctl_logpath): @@ -396,7 +391,7 @@ def analyze(args): torctl_logs = reprocessing.collect_logs(args.torctl_logpath, '*torctl.log*') log_pairs = reprocessing.match(tgen_logs, torctl_logs, args.date_filter) logging.info("Found {0} matching log pairs to be reprocessed".format(len(log_pairs))) - reprocessing.multiprocess_logs(log_pairs, args.prefix, args.nickname, args.do_complete) + reprocessing.multiprocess_logs(log_pairs, args.prefix, args.nickname) else: logging.error("Given paths were an unrecognized mix of file and directory paths, nothing will be analyzed") diff --git a/onionperf/reprocessing.py b/onionperf/reprocessing.py index 8d8dcd4..d949dca 100644 --- a/onionperf/reprocessing.py +++ b/onionperf/reprocessing.py @@ -46,21 +46,21 @@ def match(tgen_logs, tor_logs, date_filter): return log_pairs -def analyze_func(prefix, nick, do_complete, pair): +def analyze_func(prefix, nick, pair): analysis = OPAnalysis(nickname=nick) logging.info('Analysing pair for date {0}'.format(pair[2])) analysis.add_tgen_file(pair[0]) analysis.add_torctl_file(pair[1]) - analysis.analyze(do_complete=do_complete, date_filter=pair[2]) + analysis.analyze(date_filter=pair[2]) analysis.save(output_prefix=prefix) return 1 -def multiprocess_logs(log_pairs, prefix, nick=None, do_complete=False): +def multiprocess_logs(log_pairs, prefix, nick=None): pool = Pool(cpu_count()) analyses = None try: - func = partial(analyze_func, prefix, nick, do_complete) + func = partial(analyze_func, prefix, nick) mr = pool.map_async(func, log_pairs) pool.close() while not mr.ready(): diff --git a/onionperf/tests/test_reprocessing.py b/onionperf/tests/test_reprocessing.py index 5e758d2..b51d154 100644 --- a/onionperf/tests/test_reprocessing.py +++ b/onionperf/tests/test_reprocessing.py @@ -61,7 +61,7 @@ def test_log_match_with_wrong_filter_date(): def test_analyze_func_json(): pair = (DATA_DIR + 'logs/onionperf_2019-01-10_23:59:59.tgen.log', DATA_DIR + 'logs/onionperf_2019-01-10_23:59:59.torctl.log', datetime.datetime(2019, 1, 10, 0, 0)) work_dir = tempfile.mkdtemp() - reprocessing.analyze_func(work_dir, None, False, pair) + reprocessing.analyze_func(work_dir, None, pair) json_file = os.path.join(work_dir, "2019-01-10.onionperf.analysis.json.xz") assert(os.path.exists(json_file)) shutil.rmtree(work_dir)

1 0

[onionperf/develop] Removes unused import and do_complete attribute from tor parser
by karsten＠torproject.org 16 Jul '20

16 Jul '20

commit c1cd60e05e23ef022076df9665a9d23c891561ca Author: Ana Custura <ana(a)netstat.org.uk> Date: Wed Jul 15 11:56:50 2020 +0100 Removes unused import and do_complete attribute from tor parser --- onionperf/analysis.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/onionperf/analysis.py b/onionperf/analysis.py index 26b00b2..e269f6a 100644 --- a/onionperf/analysis.py +++ b/onionperf/analysis.py @@ -10,7 +10,7 @@ from abc import ABCMeta, abstractmethod # stem imports from stem import CircEvent, CircStatus, CircPurpose, StreamStatus -from stem.response.events import CircuitEvent, CircMinorEvent, StreamEvent, BandwidthEvent, BuildTimeoutSetEvent +from stem.response.events import CircuitEvent, CircMinorEvent, StreamEvent, BuildTimeoutSetEvent from stem.response import ControlMessage, convert # tgentools imports @@ -34,14 +34,14 @@ class OPAnalysis(Analysis): return self.date_filter = date_filter - tgen_parser = TGenParser(date_filter=self.date_filter) + super().analyze(do_complete=True) torctl_parser = TorCtlParser(date_filter=self.date_filter) - for (filepaths, parser, json_db_key) in [(self.tgen_filepaths, tgen_parser, 'tgen'), (self.torctl_filepaths, torctl_parser, 'tor')]: + for (filepaths, parser, json_db_key) in [(self.torctl_filepaths, torctl_parser, 'tor')]: if len(filepaths) > 0: for filepath in filepaths: logging.info("parsing log file at {0}".format(filepath)) - parser.parse(util.DataSource(filepath), do_complete=True) + parser.parse(util.DataSource(filepath)) if self.nickname is None: parsed_name = parser.get_name() @@ -128,7 +128,7 @@ class OPAnalysis(Analysis): class Parser(object, metaclass=ABCMeta): @abstractmethod - def parse(self, source, do_complete=True): + def parse(self, source): pass @abstractmethod def get_data(self): @@ -404,7 +404,7 @@ class TorCtlParser(Parser): return True - def parse(self, source, do_complete=True): + def parse(self, source): source.open(newline='\r\n') for line in source: # ignore line parsing errors

1 0

[stem/master] Correct rebase discrepancies
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 18a3280d9cb27a81bb01d8e964449adda3dc734e Author: Damian Johnson <atagar(a)torproject.org> Date: Mon May 18 14:43:18 2020 -0700 Correct rebase discrepancies To make Illia's branch cleanly mergable I rebased onto our present master. Manually resolving the conflicts resulted in a slightly different result than he had. This delta makes us perfectly match his commit c788fd8. --- stem/client/__init__.py | 11 +++--- stem/connection.py | 25 ++++++------- stem/control.py | 63 ++++++++++++++++++--------------- stem/descriptor/remote.py | 16 ++++----- stem/interpreter/__init__.py | 4 +-- stem/interpreter/autocomplete.py | 8 ++--- stem/interpreter/commands.py | 12 +++---- stem/interpreter/help.py | 9 ++--- stem/response/__init__.py | 2 +- stem/socket.py | 45 +++++++++++------------ test/integ/connection/authentication.py | 6 +++- test/unit/control/controller.py | 6 ++-- 12 files changed, 105 insertions(+), 102 deletions(-) diff --git a/stem/client/__init__.py b/stem/client/__init__.py index 941f0ee7..8ea7b3c1 100644 --- a/stem/client/__init__.py +++ b/stem/client/__init__.py @@ -33,7 +33,7 @@ import stem.socket import stem.util.connection from types import TracebackType -from typing import Dict, Iterator, List, Optional, Sequence, Type, Union +from typing import AsyncIterator, Dict, List, Optional, Sequence, Type, Union from stem.client.cell import ( CELL_TYPE_SIZE, @@ -70,7 +70,8 @@ class Relay(object): self.link_protocol = LinkProtocol(link_protocol) self._orport = orport self._orport_buffer = b'' # unread bytes - self._circuits = {} + self._orport_lock = stem.util.CombinedReentrantAndAsyncioLock() + self._circuits = {} # type: Dict[int, stem.client.Circuit] @staticmethod async def connect(address: str, port: int, link_protocols: Sequence['stem.client.datatype.LinkProtocol'] = DEFAULT_LINK_PROTOCOLS) -> 'stem.client.Relay': # type: ignore @@ -191,7 +192,7 @@ class Relay(object): cell, self._orport_buffer = Cell.pop(self._orport_buffer, self.link_protocol) return cell - async def _msg(self, cell: 'stem.client.cell.Cell') -> Iterator['stem.client.cell.Cell']: + async def _msg(self, cell: 'stem.client.cell.Cell') -> AsyncIterator['stem.client.cell.Cell']: """ Sends a cell on the ORPort and provides the response we receive in reply. @@ -283,7 +284,7 @@ class Relay(object): return circ - async def __aiter__(self) -> Iterator['stem.client.Circuit']: + async def __aiter__(self) -> AsyncIterator['stem.client.Circuit']: async with self._orport_lock: for circ in self._circuits.values(): yield circ @@ -381,7 +382,7 @@ class Circuit(object): self.forward_digest = forward_digest self.forward_key = forward_key - async def close(self)- > None: + async def close(self) -> None: async with self.relay._orport_lock: await self.relay._orport.send(stem.client.cell.DestroyCell(self.id).pack(self.relay.link_protocol)) del self.relay._circuits[self.id] diff --git a/stem/connection.py b/stem/connection.py index de76a345..213ba010 100644 --- a/stem/connection.py +++ b/stem/connection.py @@ -159,7 +159,7 @@ import stem.util.str_tools import stem.util.system import stem.version -from typing import Any, List, Optional, Sequence, Tuple, Type, Union +from typing import Any, cast, List, Optional, Sequence, Tuple, Type, Union from stem.util import log AuthMethod = stem.util.enum.Enum('NONE', 'PASSWORD', 'COOKIE', 'SAFECOOKIE', 'UNKNOWN') @@ -227,7 +227,7 @@ COMMON_TOR_COMMANDS = ( ) -def connect(control_port: Tuple[str, Union[str, int]] = ('127.0.0.1', 'default'), control_socket: str = '/var/run/tor/control', password: Optional[str] = None, password_prompt: bool = False, chroot_path: Optional[str] = None, controller: Type = stem.control.Controller) -> Any: +def connect(control_port: Tuple[str, Union[str, int]] = ('127.0.0.1', 'default'), control_socket: str = '/var/run/tor/control', password: Optional[str] = None, password_prompt: bool = False, chroot_path: Optional[str] = None, controller: Type[stem.control.Controller] = stem.control.Controller) -> Any: """ Convenience function for quickly getting a control connection for synchronous usage. This is very handy for debugging or CLI setup, handling setup and @@ -269,7 +269,7 @@ def connect(control_port: Tuple[str, Union[str, int]] = ('127.0.0.1', 'default') # TODO: change this function's API so we can provide a concrete type if controller is None or not issubclass(controller, stem.control.Controller): - raise ValueError('Controller should be a stem.control.BaseController subclass.') + raise ValueError('Controller should be a stem.control.Controller subclass.') async_controller_thread = stem.util.ThreadForWrappedAsyncClass() async_controller_thread.start() @@ -326,7 +326,7 @@ async def connect_async(control_port: Tuple[str, Union[str, int]] = ('127.0.0.1' return await _connect_async(control_port, control_socket, password, password_prompt, chroot_path, controller) -async def _connect_async(control_port, control_socket, password, password_prompt, chroot_path, controller): +async def _connect_async(control_port: Tuple[str, Union[str, int]], control_socket: str, password: Optional[str], password_prompt: bool, chroot_path: Optional[str], controller: Type[Union[stem.control.BaseController, stem.control.Controller]]) -> Any: if control_port is None and control_socket is None: raise ValueError('Neither a control port nor control socket were provided. Nothing to connect to.') elif control_port: @@ -377,7 +377,7 @@ async def _connect_async(control_port, control_socket, password, password_prompt return await _connect_auth(control_connection, password, password_prompt, chroot_path, controller) -async def _connect_auth(control_socket: stem.socket.ControlSocket, password: str, password_prompt: bool, chroot_path: str, controller: Optional[Type[stem.control.BaseController]]) -> Any: +async def _connect_auth(control_socket: stem.socket.ControlSocket, password: str, password_prompt: bool, chroot_path: str, controller: Optional[Type[Union[stem.control.BaseController, stem.control.Controller]]]) -> Any: """ Helper for the connect_* functions that authenticates the socket and constructs the controller. @@ -402,7 +402,7 @@ async def _connect_auth(control_socket: stem.socket.ControlSocket, password: str elif issubclass(controller, stem.control.BaseController): return controller(control_socket, is_authenticated = True) elif issubclass(controller, stem.control.Controller): - return controller(control_socket, is_authenticated = True, started_async_controller_thread = threading.current_thread()) + return controller(control_socket, is_authenticated = True, started_async_controller_thread = cast(stem.util.ThreadForWrappedAsyncClass, threading.current_thread())) except IncorrectSocketType: if isinstance(control_socket, stem.socket.ControlPort): print(CONNECT_MESSAGES['wrong_port_type'].format(port = control_socket.port)) @@ -995,7 +995,7 @@ async def authenticate_safecookie(controller: Union[stem.control.BaseController, auth_response = await _msg(controller, 'AUTHENTICATE %s' % stem.util.str_tools._to_unicode(binascii.b2a_hex(client_hash))) except stem.ControllerError as exc: try: - controller.connect() + await controller.connect() except: pass @@ -1007,7 +1007,7 @@ async def authenticate_safecookie(controller: Union[stem.control.BaseController, # if we got anything but an OK response then err if not auth_response.is_ok(): try: - controller.connect() + await controller.connect() except: pass @@ -1051,9 +1051,7 @@ async def get_protocolinfo(controller: Union[stem.control.BaseController, stem.s # next followed by authentication. Transparently reconnect if that happens. if not protocolinfo_response or str(protocolinfo_response) == 'Authentication required.': - potential_coroutine = controller.connect() - if asyncio.iscoroutine(potential_coroutine): - await potential_coroutine + await controller.connect() try: protocolinfo_response = await _msg(controller, 'PROTOCOLINFO 1') @@ -1074,10 +1072,7 @@ async def _msg(controller: Union[stem.control.BaseController, stem.socket.Contro await controller.send(message) return await controller.recv() else: - message = controller.msg(message) - if asyncio.iscoroutine(message): - message = await message - return message + return await controller.msg(message) def _connection_for_default_port(address: str) -> stem.socket.ControlPort: diff --git a/stem/control.py b/stem/control.py index 6e15c16c..84f8f39b 100644 --- a/stem/control.py +++ b/stem/control.py @@ -271,7 +271,7 @@ import stem.version from stem import UNDEFINED, CircStatus, Signal from stem.util import log from types import TracebackType -from typing import Any, Callable, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple, Type, Union +from typing import Any, AsyncIterator, Awaitable, Callable, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple, Type, Union # When closing the controller we attempt to finish processing enqueued events, # but if it takes longer than this we terminate. @@ -554,6 +554,8 @@ def event_description(event: str) -> str: class _BaseControllerSocketMixin: + _socket: stem.socket.ControlSocket + def is_alive(self) -> bool: """ Checks if our socket is currently connected. This is a pass-through for our @@ -589,7 +591,7 @@ class _BaseControllerSocketMixin: return self._socket.connection_time() - def get_socket(self): + def get_socket(self) -> stem.socket.ControlSocket: """ Provides the socket used to speak with the tor process. Communicating with the socket directly isn't advised since it may confuse this controller. @@ -839,7 +841,7 @@ class BaseController(_BaseControllerSocketMixin): async def __aenter__(self) -> 'stem.control.BaseController': return self - await def __aexit__(self, exit_type: Optional[Type[BaseException]], value: Optional[BaseException], traceback: Optional[TracebackType]) -> None: + async def __aexit__(self, exit_type: Optional[Type[BaseException]], value: Optional[BaseException], traceback: Optional[TracebackType]) -> None: await self.close() async def _handle_event(self, event_message: stem.response.ControlMessage) -> None: @@ -997,7 +999,7 @@ class AsyncController(BaseController): """ @classmethod - def from_port(cls: Type, address: str = '127.0.0.1', port: Union[int, str] = 'default') -> 'stem.control.AsyncController': + def from_port(cls, address: str = '127.0.0.1', port: Union[int, str] = 'default') -> 'AsyncController': """ Constructs a :class:`~stem.socket.ControlPort` based AsyncController. @@ -1017,7 +1019,7 @@ class AsyncController(BaseController): return cls(control_socket) @classmethod - def from_socket_file(cls: Type, path: str = '/var/run/tor/control') -> 'stem.control.AsyncController': + def from_socket_file(cls, path: str = '/var/run/tor/control') -> 'AsyncController': """ Constructs a :class:`~stem.socket.ControlSocketFile` based AsyncController. @@ -1189,8 +1191,8 @@ class AsyncController(BaseController): return list(reply.values())[0] try: - response = stem.response._convert_to_getinfo(await self.msg('GETINFO %s' % ' '.join(params))) - response._assert_matches(params) + response = stem.response._convert_to_getinfo(await self.msg('GETINFO %s' % ' '.join(param_set))) + response._assert_matches(param_set) # usually we want unicode values under python 3.x @@ -1765,7 +1767,7 @@ class AsyncController(BaseController): return stem.descriptor.microdescriptor.Microdescriptor(desc_content) @with_default(yields = True) - async def get_microdescriptors(self, default: Any = UNDEFINED) -> Iterator[stem.descriptor.microdescriptor.Microdescriptor]: + async def get_microdescriptors(self, default: Any = UNDEFINED) -> AsyncIterator[stem.descriptor.microdescriptor.Microdescriptor]: """ get_microdescriptors(default = UNDEFINED) @@ -1859,7 +1861,7 @@ class AsyncController(BaseController): return stem.descriptor.server_descriptor.RelayDescriptor(desc_content) @with_default(yields = True) - async def get_server_descriptors(self, default: Any = UNDEFINED) -> Iterator[stem.descriptor.server_descriptor.RelayDescriptor]: + async def get_server_descriptors(self, default: Any = UNDEFINED) -> AsyncIterator[stem.descriptor.server_descriptor.RelayDescriptor]: """ get_server_descriptors(default = UNDEFINED) @@ -1954,7 +1956,7 @@ class AsyncController(BaseController): return stem.descriptor.router_status_entry.RouterStatusEntryV3(desc_content) @with_default(yields = True) - async def get_network_statuses(self, default: Any = UNDEFINED) -> Iterator[stem.descriptor.router_status_entry.RouterStatusEntryV3]: + async def get_network_statuses(self, default: Any = UNDEFINED) -> AsyncIterator[stem.descriptor.router_status_entry.RouterStatusEntryV3]: """ get_network_statuses(default = UNDEFINED) @@ -2061,6 +2063,7 @@ class AsyncController(BaseController): request += ' ' + ' '.join(['SERVER=%s' % s for s in servers]) response = stem.response._convert_to_single_line(await self.msg(request)) + stem.response.convert('SINGLELINE', response) if not response.is_ok(): raise stem.ProtocolError('HSFETCH returned unexpected response code: %s' % response.code) @@ -2153,7 +2156,7 @@ class AsyncController(BaseController): async def _get_conf_multiple(self, param: str, default: Any = UNDEFINED) -> List[str]: return await self.get_conf(param, default, multiple = True) # type: ignore - await def get_conf_map(self, params: Union[str, Sequence[str]], default: Any = UNDEFINED, multiple: bool = True) -> Dict[str, Union[str, Sequence[str]]]: + async def get_conf_map(self, params: Union[str, Sequence[str]], default: Any = UNDEFINED, multiple: bool = True) -> Dict[str, Union[str, Sequence[str]]]: """ get_conf_map(params, default = UNDEFINED, multiple = True) @@ -2971,7 +2974,7 @@ class AsyncController(BaseController): else: request += ' ClientAuth=%s' % client_name - response = stem.response._convert_to_add_onion(await self.msg(request)) + response = stem.response._convert_to_add_onion(stem.response._convert_to_add_onion(await self.msg(request))) if await_publication: # We should receive five UPLOAD events, followed by up to another five @@ -3024,7 +3027,7 @@ class AsyncController(BaseController): else: raise stem.ProtocolError('DEL_ONION returned unexpected response code: %s' % response.code) - async def add_event_listener(self, listener: Callable[[stem.response.events.Event], None], *events: 'stem.control.EventType') -> None: + async def add_event_listener(self, listener: Callable[[stem.response.events.Event], Union[None, Awaitable[None]]], *events: 'stem.control.EventType') -> None: """ Directs further tor controller events to a given function. The function is expected to take a single argument, which is a @@ -3082,7 +3085,7 @@ class AsyncController(BaseController): if failed_events: raise stem.ProtocolError('SETEVENTS rejected %s' % ', '.join(failed_events)) - async def remove_event_listener(self, listener: Callable[[stem.response.events.Event], None]) -> None: + async def remove_event_listener(self, listener: Callable[[stem.response.events.Event], Union[None, Awaitable[None]]]) -> None: """ Stops a listener from being notified of further tor events. @@ -3253,7 +3256,7 @@ class AsyncController(BaseController): :raises: :class:`stem.ControllerError` if the call fails """ - response = stem.response._convert_to_single_line(async self.msg('LOADCONF\n%s' % configtext)) + response = stem.response._convert_to_single_line(await self.msg('LOADCONF\n%s' % configtext)) if response.code in ('552', '553'): if response.code == '552' and response.message.startswith('Invalid config file: Failed to parse/validate config: Unknown option'): @@ -3379,7 +3382,7 @@ class AsyncController(BaseController): response = await self.get_info('circuit-status') for circ in response.splitlines(): - circ_message = stem.response._convert_to_event(await stem.socket.recv_message(io.BytesIO(stem.util.str_tools._to_bytes('650 CIRC %s\r\n' % circ)))) + circ_message = stem.response._convert_to_event(stem.socket.recv_message_from_bytes_io(io.BytesIO(stem.util.str_tools._to_bytes('650 CIRC %s\r\n' % circ)))) circuits.append(circ_message) # type: ignore return circuits @@ -3563,7 +3566,7 @@ class AsyncController(BaseController): response = await self.get_info('stream-status') for stream in response.splitlines(): - message = stem.response._convert_to_event(await stem.socket.recv_message(io.BytesIO(stem.util.str_tools._to_bytes('650 STREAM %s\r\n' % stream)))) + message = stem.response._convert_to_event(stem.socket.recv_message_from_bytes_io(io.BytesIO(stem.util.str_tools._to_bytes('650 STREAM %s\r\n' % stream)))) streams.append(message) # type: ignore return streams @@ -3744,7 +3747,7 @@ class AsyncController(BaseController): response = await self.msg('MAPADDRESS %s' % mapaddress_arg) return stem.response._convert_to_mapaddress(response).entries - await def drop_guards(self) -> None: + async def drop_guards(self) -> None: """ Drops our present guard nodes and picks a new set. @@ -3812,7 +3815,7 @@ class AsyncController(BaseController): if listener_type == event_type: for listener in event_listeners: try: - potential_coroutine = listener(event_message) + potential_coroutine = listener(event) if asyncio.iscoroutine(potential_coroutine): await potential_coroutine except Exception as exc: @@ -3874,7 +3877,7 @@ class Controller(_BaseControllerSocketMixin, stem.util.AsyncClassWrapper): """ @classmethod - def from_port(cls: Type, address: str = '127.0.0.1', port: Union[int, str] = 'default') -> 'stem.control.Controller': + def from_port(cls, address: str = '127.0.0.1', port: Union[int, str] = 'default') -> 'Controller': """ Constructs a :class:`~stem.socket.ControlPort` based Controller. @@ -3885,8 +3888,8 @@ class Controller(_BaseControllerSocketMixin, stem.util.AsyncClassWrapper): .. versionchanged:: 1.5.0 Use both port 9051 and 9151 by default. - :param str address: ip address of the controller - :param int port: port number of the controller + :param address: ip address of the controller + :param port: port number of the controller :returns: :class:`~stem.control.Controller` attached to the given port @@ -3899,7 +3902,7 @@ class Controller(_BaseControllerSocketMixin, stem.util.AsyncClassWrapper): return controller @classmethod - def from_socket_file(cls: Type, path: str = '/var/run/tor/control') -> 'stem.control.Controller': + def from_socket_file(cls, path: str = '/var/run/tor/control') -> 'Controller': """ Constructs a :class:`~stem.socket.ControlSocketFile` based Controller. @@ -3915,15 +3918,19 @@ class Controller(_BaseControllerSocketMixin, stem.util.AsyncClassWrapper): controller.connect() return controller - def __init__(self, control_socket: 'stem.socket.ControlSocket', is_authenticated: bool = False, started_async_controller_thread: Optional['threading.Thread'] = None) -> None: - def __init__(self, control_socket, is_authenticated = False, started_async_controller_thread = None): + def __init__( + self, + control_socket: stem.socket.ControlSocket, + is_authenticated: bool = False, + started_async_controller_thread: stem.util.ThreadForWrappedAsyncClass = None, + ) -> None: if started_async_controller_thread: self._thread_for_wrapped_class = started_async_controller_thread else: self._thread_for_wrapped_class = stem.util.ThreadForWrappedAsyncClass() self._thread_for_wrapped_class.start() - self._wrapped_instance: AsyncController = self._init_async_class(AsyncController, control_socket, is_authenticated) + self._wrapped_instance: AsyncController = self._init_async_class(AsyncController, control_socket, is_authenticated) # type: ignore self._socket = self._wrapped_instance._socket @_set_doc_from_async_controller @@ -3956,7 +3963,7 @@ class Controller(_BaseControllerSocketMixin, stem.util.AsyncClassWrapper): @_set_doc_from_async_controller def remove_status_listener(self, callback: Callable[['stem.control.Controller', 'stem.control.State', float], None]) -> bool: - self._wrapped_instance.remove_status_listener(callback) + return self._wrapped_instance.remove_status_listener(callback) @_set_doc_from_async_controller def authenticate(self, *args: Any, **kwargs: Any) -> None: @@ -4306,7 +4313,7 @@ def _case_insensitive_lookup(entries: Union[Sequence[str], Mapping[str, Any]], k raise ValueError("key '%s' doesn't exist in dict: %s" % (key, entries)) -async def _get_with_timeout(event_queue: queue.Queue, timeout: float, start_time: float) -> Any: +async def _get_with_timeout(event_queue: asyncio.Queue, timeout: Optional[float], start_time: float) -> Any: """ Pulls an item from a queue with a given timeout. """ diff --git a/stem/descriptor/remote.py b/stem/descriptor/remote.py index d7a833a4..e7ccaa24 100644 --- a/stem/descriptor/remote.py +++ b/stem/descriptor/remote.py @@ -104,7 +104,7 @@ import stem.util.tor_tools from stem.descriptor import Compression from stem.util import log, str_tools -from typing import Any, Dict, Iterator, List, Optional, Sequence, Tuple, Union +from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, Sequence, Tuple, Union # Tor has a limited number of descriptors we can fetch explicitly by their # fingerprint or hashes due to a limit on the url length by squid proxies. @@ -392,7 +392,7 @@ class AsyncQuery(object): self.reply_headers = None # type: Optional[Dict[str, str]] self.kwargs = kwargs - self._downloader_task = None + self._downloader_task = None # type: Optional[asyncio.Task] self._downloader_lock = threading.RLock() self._asyncio_loop = asyncio.get_event_loop() @@ -401,7 +401,7 @@ class AsyncQuery(object): self.start() if block: - self._asyncio_loop.create_task(self.run(True)) + self.run(True) def start(self) -> None: """ @@ -432,7 +432,7 @@ class AsyncQuery(object): return [desc async for desc in self._run(suppress)] - async def _run(self, suppress: bool) -> Iterator[stem.descriptor.Descriptor]: + async def _run(self, suppress: bool) -> AsyncIterator[stem.descriptor.Descriptor]: with self._downloader_lock: self.start() await self._downloader_task @@ -468,7 +468,7 @@ class AsyncQuery(object): raise self.error - async def __aiter__(self) -> Iterator[stem.descriptor.Descriptor]: + async def __aiter__(self) -> AsyncIterator[stem.descriptor.Descriptor]: async for desc in self._run(True): yield desc @@ -665,7 +665,7 @@ class Query(stem.util.AsyncClassWrapper): def __init__(self, resource: str, descriptor_type: Optional[str] = None, endpoints: Optional[Sequence[stem.Endpoint]] = None, compression: Union[stem.descriptor._Compression, Sequence[stem.descriptor._Compression]] = (Compression.GZIP,), retries: int = 2, fall_back_to_authority: bool = False, timeout: Optional[float] = None, start: bool = True, block: bool = False, validate: bool = False, document_handler: stem.descriptor.DocumentHandler = stem.descriptor.DocumentHandler.ENTRIES, **kwargs: Any) -> None: self._thread_for_wrapped_class = stem.util.ThreadForWrappedAsyncClass() self._thread_for_wrapped_class.start() - self._wrapped_instance: AsyncQuery = self._init_async_class( + self._wrapped_instance: AsyncQuery = self._init_async_class( # type: ignore AsyncQuery, resource, descriptor_type, @@ -688,7 +688,7 @@ class Query(stem.util.AsyncClassWrapper): self._call_async_method_soon('start') - def run(self, suppress: bool = False) -> List['stem.descriptor.Descriptor']: + def run(self, suppress = False) -> List['stem.descriptor.Descriptor']: """ Blocks until our request is complete then provides the descriptors. If we haven't yet started our request then this does so. @@ -708,7 +708,7 @@ class Query(stem.util.AsyncClassWrapper): return self._execute_async_method('run', suppress) - def __iter__(self): + def __iter__(self) -> Iterator[stem.descriptor.Descriptor]: for desc in self._execute_async_generator_method('__aiter__'): yield desc diff --git a/stem/interpreter/__init__.py b/stem/interpreter/__init__.py index ae064a0a..370b9aa6 100644 --- a/stem/interpreter/__init__.py +++ b/stem/interpreter/__init__.py @@ -124,10 +124,10 @@ def main() -> None: if args.run_cmd: if args.run_cmd.upper().startswith('SETEVENTS '): - async def handle_event(event_message): + async def handle_event(event_message: stem.response.ControlMessage) -> None: print(format(str(event_message), *STANDARD_OUTPUT)) - controller._wrapped_instance._handle_event = handle_event + controller._wrapped_instance._handle_event = handle_event # type: ignore if sys.stdout.isatty(): events = args.run_cmd.upper().split(' ', 1)[1] diff --git a/stem/interpreter/autocomplete.py b/stem/interpreter/autocomplete.py index 54642472..ed51fd3d 100644 --- a/stem/interpreter/autocomplete.py +++ b/stem/interpreter/autocomplete.py @@ -11,7 +11,7 @@ import stem.control import stem.util.conf from stem.interpreter import uses_settings -from typing import List, Optional +from typing import cast, List, Optional @uses_settings @@ -28,7 +28,7 @@ def _get_commands(controller: stem.control.Controller, config: stem.util.conf.Co # GETINFO commands. Lines are of the form '[option] -- [description]'. This # strips '*' from options that accept values. - results = controller.get_info('info/names', None) + results = cast(str, controller.get_info('info/names', None)) if results: for line in results.splitlines(): @@ -40,7 +40,7 @@ def _get_commands(controller: stem.control.Controller, config: stem.util.conf.Co # GETCONF, SETCONF, and RESETCONF commands. Lines are of the form # '[option] [type]'. - results = controller.get_info('config/names', None) + results = cast(str, controller.get_info('config/names', None)) if results: for line in results.splitlines(): @@ -62,7 +62,7 @@ def _get_commands(controller: stem.control.Controller, config: stem.util.conf.Co ) for prefix, getinfo_cmd in options: - results = controller.get_info(getinfo_cmd, None) + results = cast(str, controller.get_info(getinfo_cmd, None)) if results: commands += [prefix + value for value in results.split()] diff --git a/stem/interpreter/commands.py b/stem/interpreter/commands.py index edbcca70..99f1219d 100644 --- a/stem/interpreter/commands.py +++ b/stem/interpreter/commands.py @@ -21,7 +21,7 @@ import stem.util.tor_tools from stem.interpreter import STANDARD_OUTPUT, BOLD_OUTPUT, ERROR_OUTPUT, uses_settings, msg from stem.util.term import format -from typing import Iterator, List, TextIO +from typing import cast, Iterator, List, TextIO MAX_EVENTS = 100 @@ -45,7 +45,7 @@ def _get_fingerprint(arg: str, controller: stem.control.Controller) -> str: if not arg: try: - return controller.get_info('fingerprint') + return cast(str, controller.get_info('fingerprint')) except: raise ValueError("We aren't a relay, no information to provide") elif stem.util.tor_tools.is_valid_fingerprint(arg): @@ -132,14 +132,14 @@ class ControlInterpreter(code.InteractiveConsole): async def handle_event_wrapper(event_message: stem.response.ControlMessage) -> None: await handle_event_real(event_message) - self._received_events.insert(0, event_message) + self._received_events.insert(0, event_message) # type: ignore if len(self._received_events) > MAX_EVENTS: self._received_events.pop() # type check disabled due to https://github.com/python/mypy/issues/708 - self._controller._wrapped_instance._handle_event = handle_event_wrapper + self._controller._wrapped_instance._handle_event = handle_event_wrapper # type: ignore def get_events(self, *event_types: stem.control.EventType) -> List[stem.response.events.Event]: events = list(self._received_events) @@ -207,7 +207,7 @@ class ControlInterpreter(code.InteractiveConsole): extrainfo_desc_query = downloader.get_extrainfo_descriptors(fingerprint) for desc in server_desc_query: - server_desc = desc + server_desc = cast(stem.descriptor.server_descriptor.RelayDescriptor, desc) for desc in extrainfo_desc_query: extrainfo_desc = desc @@ -220,7 +220,7 @@ class ControlInterpreter(code.InteractiveConsole): pass try: - address_extrainfo.append(self._controller.get_info('ip-to-country/%s' % ns_desc.address)) + address_extrainfo.append(cast(str, self._controller.get_info('ip-to-country/%s' % ns_desc.address))) except: pass diff --git a/stem/interpreter/help.py b/stem/interpreter/help.py index 14b46e35..f2bbbafd 100644 --- a/stem/interpreter/help.py +++ b/stem/interpreter/help.py @@ -6,6 +6,7 @@ Provides our /help responses. """ import functools +from typing import cast import stem.control import stem.util.conf @@ -74,7 +75,7 @@ def _response(controller: stem.control.Controller, arg: str, config: stem.util.c output += '\n' if arg == 'GETINFO': - results = controller.get_info('info/names', None) + results = cast(str, controller.get_info('info/names', None)) if results: for line in results.splitlines(): @@ -84,7 +85,7 @@ def _response(controller: stem.control.Controller, arg: str, config: stem.util.c output += format('%-33s' % opt, *BOLD_OUTPUT) output += format(' - %s' % summary, *STANDARD_OUTPUT) + '\n' elif arg == 'GETCONF': - results = controller.get_info('config/names', None) + results = cast(str, controller.get_info('config/names', None)) if results: options = [opt.split(' ', 1)[0] for opt in results.splitlines()] @@ -103,7 +104,7 @@ def _response(controller: stem.control.Controller, arg: str, config: stem.util.c output += format('%-15s' % signal, *BOLD_OUTPUT) output += format(' - %s' % summary, *STANDARD_OUTPUT) + '\n' elif arg == 'SETEVENTS': - results = controller.get_info('events/names', None) + results = cast(str, controller.get_info('events/names', None)) if results: entries = results.split() @@ -118,7 +119,7 @@ def _response(controller: stem.control.Controller, arg: str, config: stem.util.c output += format(line.rstrip(), *STANDARD_OUTPUT) + '\n' elif arg == 'USEFEATURE': - results = controller.get_info('features/names', None) + results = cast(str, controller.get_info('features/names', None)) if results: output += format(results, *STANDARD_OUTPUT) + '\n' diff --git a/stem/response/__init__.py b/stem/response/__init__.py index 52dc74e4..2e251144 100644 --- a/stem/response/__init__.py +++ b/stem/response/__init__.py @@ -202,7 +202,7 @@ class ControlMessage(object): content = re.sub(b'([\r]?)\n', b'\r\n', content) - msg = stem.socket.recv_message_from_bytes_io(io.BytesIO(content), arrived_at = kwargs.pop('arrived_at', None)) + msg = stem.socket.recv_message_from_bytes_io(io.BytesIO(stem.util.str_tools._to_bytes(content)), arrived_at = kwargs.pop('arrived_at', None)) if msg_type is not None: convert(msg_type, msg, **kwargs) diff --git a/stem/socket.py b/stem/socket.py index 0feae831..ff99c5b1 100644 --- a/stem/socket.py +++ b/stem/socket.py @@ -85,6 +85,7 @@ import asyncio import re import socket import ssl +import sys import threading import time @@ -93,7 +94,7 @@ import stem.util.str_tools from stem.util import log from types import TracebackType -from typing import BinaryIO, Callable, List, Optional, Tuple, Type, Union, overload +from typing import Awaitable, BinaryIO, Callable, List, Optional, Tuple, Type, Union, overload MESSAGE_PREFIX = re.compile(b'^[a-zA-Z0-9]{3}[-+ ]') ERROR_MSG = 'Error while receiving a control message (%s): %s' @@ -109,8 +110,8 @@ class BaseSocket(object): """ def __init__(self) -> None: - self._reader = None - self._writer = None + self._reader = None # type: Optional[asyncio.StreamReader] + self._writer = None # type: Optional[asyncio.StreamWriter] self._is_alive = False self._connection_time = 0.0 # time when we last connected or disconnected @@ -209,7 +210,7 @@ class BaseSocket(object): if self._writer: self._writer.close() # `StreamWriter.wait_closed` was added in Python 3.7. - if hasattr(self._writer, 'wait_closed'): + if sys.version_info >= (3, 7): await self._writer.wait_closed() self._reader = None @@ -220,7 +221,7 @@ class BaseSocket(object): if is_change: await self._close() - async def _send(self, message: Union[bytes, str], handler: Callable[[Union[socket.socket, ssl.SSLSocket], BinaryIO, Union[bytes, str]], None]) -> None: + async def _send(self, message: Union[bytes, str], handler: Callable[[asyncio.StreamWriter, Union[bytes, str]], Awaitable[None]]) -> None: """ Send message in a thread safe manner. """ @@ -241,11 +242,11 @@ class BaseSocket(object): raise @overload - async def _recv(self, handler: Callable[[ssl.SSLSocket, BinaryIO], bytes]) -> bytes: + async def _recv(self, handler: Callable[[asyncio.StreamReader], Awaitable[bytes]]) -> bytes: ... @overload - async def _recv(self, handler: Callable[[socket.socket, BinaryIO], stem.response.ControlMessage]) -> stem.response.ControlMessage: + async def _recv(self, handler: Callable[[asyncio.StreamReader], Awaitable[stem.response.ControlMessage]]) -> stem.response.ControlMessage: ... async def _recv(self, handler): @@ -303,7 +304,7 @@ class BaseSocket(object): return self async def __aexit__(self, exit_type: Optional[Type[BaseException]], value: Optional[BaseException], traceback: Optional[TracebackType]): - self.close() + await self.close() async def _connect(self) -> None: """ @@ -320,12 +321,6 @@ class BaseSocket(object): pass async def _open_connection(self) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]: - """ - Constructs and connects new socket. This is implemented by subclasses. - - :returns: **tuple** with our reader and writer streams - """ - raise NotImplementedError('Unsupported Operation: this should be implemented by the BaseSocket subclass') @@ -380,7 +375,7 @@ class RelaySocket(BaseSocket): * :class:`stem.SocketClosed` if the socket closes before we receive a complete message """ - async def wrapped_recv(s: ssl.SSLSocket, sf: BinaryIO) -> bytes: + async def wrapped_recv(reader: asyncio.StreamReader) -> Optional[bytes]: read_coroutine = reader.read(1024) if timeout is None: return await read_coroutine @@ -524,7 +519,7 @@ async def send_message(writer: asyncio.StreamWriter, message: Union[bytes, str], <line 3>\\r\\n .\\r\\n - :param writer: stream derived from the control socket + :param writer: writer object :param message: message to be sent on the control socket :param raw: leaves the message formatting untouched, passing it to the socket as-is @@ -591,9 +586,7 @@ async def recv_message(reader: asyncio.StreamReader, arrived_at: Optional[float] while True: try: - line = reader.readline() - if asyncio.iscoroutine(line): - line = await line + line = await reader.readline() except AttributeError: # if the control_file has been closed then we will receive: # AttributeError: 'NoneType' object has no attribute 'recv' @@ -693,7 +686,7 @@ async def recv_message(reader: asyncio.StreamReader, arrived_at: Optional[float] raise stem.ProtocolError("Unrecognized divider type '%s': %s" % (divider, stem.util.str_tools._to_unicode(line))) -def recv_message_from_bytes_io(reader: asyncio.StreamReader, arrived_at: Optional[float] = None) -> 'stem.response.ControlMessage': +def recv_message_from_bytes_io(reader: BinaryIO, arrived_at: Optional[float] = None) -> stem.response.ControlMessage: """ Pulls from an I/O stream until we either have a complete message or encounter a problem. @@ -708,7 +701,9 @@ def recv_message_from_bytes_io(reader: asyncio.StreamReader, arrived_at: Optiona a complete message """ - parsed_content, raw_content, first_line = None, None, True + parsed_content = [] # type: List[Tuple[str, str, bytes]] + raw_content = bytearray() + first_line = True while True: try: @@ -739,10 +734,10 @@ def recv_message_from_bytes_io(reader: asyncio.StreamReader, arrived_at: Optiona log.info(ERROR_MSG % ('SocketClosed', 'empty socket content')) raise stem.SocketClosed('Received empty socket content.') elif not MESSAGE_PREFIX.match(line): - log.info(ERROR_MSG % ('ProtocolError', 'malformed status code/divider, "%s"' % log.escape(line))) + log.info(ERROR_MSG % ('ProtocolError', 'malformed status code/divider, "%s"' % log.escape(line.decode('utf-8')))) raise stem.ProtocolError('Badly formatted reply line: beginning is malformed') elif not line.endswith(b'\r\n'): - log.info(ERROR_MSG % ('ProtocolError', 'no CRLF linebreak, "%s"' % log.escape(line))) + log.info(ERROR_MSG % ('ProtocolError', 'no CRLF linebreak, "%s"' % log.escape(line.decode('utf-8')))) raise stem.ProtocolError('All lines should end with CRLF') status_code, divider, content = line[:3], line[3:4], line[4:-2] # strip CRLF off content @@ -781,11 +776,11 @@ def recv_message_from_bytes_io(reader: asyncio.StreamReader, arrived_at: Optiona line = reader.readline() raw_content += line except socket.error as exc: - log.info(ERROR_MSG % ('SocketClosed', 'received an exception while mid-way through a data reply (exception: "%s", read content: "%s")' % (exc, log.escape(bytes(raw_content))))) + log.info(ERROR_MSG % ('SocketClosed', 'received an exception while mid-way through a data reply (exception: "%s", read content: "%s")' % (exc, log.escape(bytes(raw_content).decode('utf-8'))))) raise stem.SocketClosed(exc) if not line.endswith(b'\r\n'): - log.info(ERROR_MSG % ('ProtocolError', 'CRLF linebreaks missing from a data reply, "%s"' % log.escape(bytes(raw_content)))) + log.info(ERROR_MSG % ('ProtocolError', 'CRLF linebreaks missing from a data reply, "%s"' % log.escape(bytes(raw_content).decode('utf-8')))) raise stem.ProtocolError('All lines should end with CRLF') elif line == b'.\r\n': break # data block termination diff --git a/test/integ/connection/authentication.py b/test/integ/connection/authentication.py index 683e555f..3eaae8d9 100644 --- a/test/integ/connection/authentication.py +++ b/test/integ/connection/authentication.py @@ -3,6 +3,7 @@ Integration tests for authenticating to the control socket via stem.connection.authenticate* functions. """ +import asyncio import os import unittest @@ -121,7 +122,10 @@ class TestAuthenticate(unittest.TestCase): runner = test.runner.get_runner() with await runner.get_tor_controller(False) as controller: - await stem.connection.authenticate(controller, test.runner.CONTROL_PASSWORD, runner.get_chroot()) + asyncio.run_coroutine_threadsafe( + stem.connection.authenticate(controller._wrapped_instance, test.runner.CONTROL_PASSWORD, runner.get_chroot()), + controller._thread_for_wrapped_class.loop, + ).result() await test.runner.exercise_controller(self, controller) @test.require.controller diff --git a/test/unit/control/controller.py b/test/unit/control/controller.py index a11aba45..e4b11788 100644 --- a/test/unit/control/controller.py +++ b/test/unit/control/controller.py @@ -222,7 +222,7 @@ class TestControl(unittest.TestCase): get_info_mock.side_effect = coro_func_raising_exc(InvalidArguments) - async def get_conf_mock_side_effect(param, **kwargs): + async def get_conf_mock_side_effect(param, *args, **kwargs): return { 'ControlPort': '9050', 'ControlListenAddress': ['127.0.0.1'], @@ -236,7 +236,7 @@ class TestControl(unittest.TestCase): # non-local addresss - async def get_conf_mock_side_effect(param, **kwargs): + async def get_conf_mock_side_effect(param, *args, **kwargs): return { 'ControlPort': '9050', 'ControlListenAddress': ['27.4.4.1'], @@ -717,7 +717,7 @@ class TestControl(unittest.TestCase): # check default if nothing was set - async def get_conf_mock_side_effect(param, **kwargs): + async def get_conf_mock_side_effect(param, *args, **kwargs): return { 'BandwidthRate': '1073741824', 'BandwidthBurst': '1073741824',

1 0

[stem/master] Add docstrings to new functions used for tests
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 33fd00ede9b984de5d488402f2ba699a601e4ce0 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sun May 17 20:22:57 2020 +0300 Add docstrings to new functions used for tests --- stem/util/test_tools.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/stem/util/test_tools.py b/stem/util/test_tools.py index f5ae7525..67133195 100644 --- a/stem/util/test_tools.py +++ b/stem/util/test_tools.py @@ -685,6 +685,14 @@ def _is_ignored(config: Mapping[str, Sequence[str]], path: str, issue: str) -> b def async_test(func: Callable) -> Callable: + """ + Decorator for asynchronous test functions. + + :param func: function that will be decorated + + :return: decorated function + """ + @functools.wraps(func) def wrapper(*args: Any, **kwargs: Any) -> Any: loop = asyncio.new_event_loop() @@ -697,12 +705,30 @@ def async_test(func: Callable) -> Callable: def coro_func_returning_value(return_value: Any) -> Callable[..., Awaitable]: + """ + Creates and returns a coroutine function that returns the specified + value and is used for mocking asynchronous functions. + + :param return_value: return value for the coroutine function + + :return: coroutine function returning the specified value + """ + async def coroutine_func(*args, **kwargs): return return_value return coroutine_func def coro_func_raising_exc(exc: Exception) -> Callable[..., Awaitable]: + """ + Creates and returns a coroutine function that raises the specified + exception and is used for mocking asynchronous functions. + + :param exc: exception for the coroutine function + + :return: coroutine function raising the specified exception + """ + async def coroutine_func(*args, **kwargs): raise exc return coroutine_func

1 0

[stem/master] Update `BaseController._event_loop` to keep the previous behavior
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 0dcb75bbaac67cd1c461f3bbc3fe5550c47072ef Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Thu May 21 22:52:22 2020 +0300 Update `BaseController._event_loop` to keep the previous behavior --- stem/control.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/stem/control.py b/stem/control.py index be7ab6c6..c26da351 100644 --- a/stem/control.py +++ b/stem/control.py @@ -989,6 +989,8 @@ class BaseController(_BaseControllerSocketMixin): try: await asyncio.wait_for(self._event_notice.wait(), timeout=0.05) except asyncio.TimeoutError: + pass + finally: self._event_notice.clear()

1 0

[stem/master] Use `asyncio.Queue.put_nowait` instead of `put` that needs awaiting
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 8ebbeec288b849c7944af925d1d00814a1ef17c7 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Thu May 21 22:39:52 2020 +0300 Use `asyncio.Queue.put_nowait` instead of `put` that needs awaiting --- stem/control.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/stem/control.py b/stem/control.py index 84f8f39b..be7ab6c6 100644 --- a/stem/control.py +++ b/stem/control.py @@ -946,18 +946,18 @@ class BaseController(_BaseControllerSocketMixin): if control_message.content()[-1][0] == '650': # asynchronous message, adds to the event queue and wakes up its handler - await self._event_queue.put(control_message) + self._event_queue.put_nowait(control_message) self._event_notice.set() else: # response to a msg() call - await self._reply_queue.put(control_message) + self._reply_queue.put_nowait(control_message) except stem.ControllerError as exc: # Assume that all exceptions belong to the reader. This isn't always # true, but the msg() call can do a better job of sorting it out. # # Be aware that the msg() method relies on this to unblock callers. - await self._reply_queue.put(exc) + self._reply_queue.put_nowait(exc) async def _event_loop(self) -> None: """ @@ -2045,11 +2045,11 @@ class AsyncController(BaseController): start_time = time.time() if await_result: - async def hs_desc_listener(event: stem.response.events.Event) -> None: - await hs_desc_queue.put(event) + def hs_desc_listener(event: stem.response.events.Event) -> None: + hs_desc_queue.put_nowait(event) - async def hs_desc_content_listener(event: stem.response.events.Event) -> None: - await hs_desc_content_queue.put(event) + def hs_desc_content_listener(event: stem.response.events.Event) -> None: + hs_desc_content_queue.put_nowait(event) await asyncio.gather( self.add_event_listener(hs_desc_listener, EventType.HS_DESC), @@ -2926,8 +2926,8 @@ class AsyncController(BaseController): start_time = time.time() if await_publication: - async def hs_desc_listener(event: stem.response.events.Event) -> None: - await hs_desc_queue.put(event) + def hs_desc_listener(event: stem.response.events.Event) -> None: + hs_desc_queue.put_nowait(event) await self.add_event_listener(hs_desc_listener, EventType.HS_DESC) @@ -3458,8 +3458,8 @@ class AsyncController(BaseController): start_time = time.time() if await_build: - async def circ_listener(event: stem.response.events.Event) -> None: - await circ_queue.put(event) + def circ_listener(event: stem.response.events.Event) -> None: + circ_queue.put_nowait(event) await self.add_event_listener(circ_listener, EventType.CIRC)

1 0

[stem/master] Start shutting down controllers after running controller unit tests
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit cb55eeac8ae78aada114d229aa1cc076b62ba821 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sun May 24 01:12:21 2020 +0300 Start shutting down controllers after running controller unit tests --- test/unit/control/controller.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/test/unit/control/controller.py b/test/unit/control/controller.py index e4b11788..850918f7 100644 --- a/test/unit/control/controller.py +++ b/test/unit/control/controller.py @@ -55,6 +55,9 @@ class TestControl(unittest.TestCase): self.malformed_listener = Mock() self.controller.add_event_listener(self.malformed_listener, MALFORMED_EVENTS) + def tearDown(self): + self.controller.close() + def test_event_description(self): self.assertEqual("Logging at the debug runlevel. This is low level, high volume information about tor's internals that generally isn't useful to users.", stem.control.event_description('DEBUG')) self.assertEqual('Event emitted every second with the bytes sent and received by tor.', stem.control.event_description('BW'))

1 0