tor-commits July 2020

tor-commits@lists.torproject.org

17 participants
2100 discussions

[stem/master] Update docstrings in stem/socket.py
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 152497ffaff55edc956f7423ab03e90697a252d3 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sun May 17 18:29:57 2020 +0300 Update docstrings in stem/socket.py --- stem/socket.py | 54 ++++++++++++++++++++++++++++-------------------------- 1 file changed, 28 insertions(+), 26 deletions(-) diff --git a/stem/socket.py b/stem/socket.py index e8ae492b..0feae831 100644 --- a/stem/socket.py +++ b/stem/socket.py @@ -15,31 +15,42 @@ Tor... :: - import stem + import asyncio + import sys + import stem.connection import stem.socket - if __name__ == '__main__': + async def print_version() -> None: try: control_socket = stem.socket.ControlPort(port = 9051) - stem.connection.authenticate(control_socket) + await control_socket.connect() + await stem.connection.authenticate(control_socket) except stem.SocketError as exc: - print 'Unable to connect to tor on port 9051: %s' % exc + print(f'Unable to connect to tor on port 9051: {exc}') sys.exit(1) except stem.connection.AuthenticationFailure as exc: - print 'Unable to authenticate: %s' % exc + print(f'Unable to authenticate: {exc}') sys.exit(1) - print "Issuing 'GETINFO version' query...\\n" - control_socket.send('GETINFO version') - print control_socket.recv() + print("Issuing 'GETINFO version' query...\\n") + await control_socket.send('GETINFO version') + print(await control_socket.recv()) + + + if __name__ == '__main__': + loop = asyncio.get_event_loop() + try: + loop.run_until_complete(print_version()) + finally: + loop.close() :: % python example.py Issuing 'GETINFO version' query... - version=0.2.4.10-alpha-dev (git-8be6058d8f31e578) + version=0.4.3.5 OK **Module Overview:** @@ -66,6 +77,7 @@ Tor... send_message - Writes a message to a control socket. recv_message - Reads a ControlMessage from a control socket. + recv_message_from_bytes_io - Reads a ControlMessage from an I/O stream. send_formatting - Performs the formatting expected from sent messages. """ @@ -210,11 +222,7 @@ class BaseSocket(object): async def _send(self, message: Union[bytes, str], handler: Callable[[Union[socket.socket, ssl.SSLSocket], BinaryIO, Union[bytes, str]], None]) -> None: """ - Send message in a thread safe manner. Handler is expected to be of the form... - - :: - - my_handler(socket, socket_file, message) + Send message in a thread safe manner. """ with self._send_lock: @@ -242,11 +250,7 @@ class BaseSocket(object): async def _recv(self, handler): """ - Receives a message in a thread safe manner. Handler is expected to be of the form... - - :: - - my_handler(socket, socket_file) + Receives a message in a thread safe manner. """ with self._recv_lock: @@ -405,7 +409,7 @@ class ControlSocket(BaseSocket): receiving complete messages. Callers should not instantiate this class directly, but rather use subclasses - which are expected to implement the **_make_socket()** method. + which are expected to implement the **_open_connection()** method. """ def __init__(self) -> None: @@ -483,7 +487,7 @@ class ControlSocketFile(ControlSocket): """ ControlSocketFile constructor. - :param socket_path: path where the control socket is located + :param path: path where the control socket is located """ super(ControlSocketFile, self).__init__() @@ -571,8 +575,7 @@ async def recv_message(reader: asyncio.StreamReader, arrived_at: Optional[float] Pulls from a control socket until we either have a complete message or encounter a problem. - :param control_file: file derived from the control socket (see the - socket's makefile() method for more information) + :param reader: reader object :returns: :class:`~stem.response.ControlMessage` read from the socket @@ -692,11 +695,10 @@ async def recv_message(reader: asyncio.StreamReader, arrived_at: Optional[float] def recv_message_from_bytes_io(reader: asyncio.StreamReader, arrived_at: Optional[float] = None) -> 'stem.response.ControlMessage': """ - Pulls from a control socket until we either have a complete message or + Pulls from an I/O stream until we either have a complete message or encounter a problem. - :param file control_file: file derived from the control socket (see the - socket's makefile() method for more information) + :param file reader: I/O stream :returns: :class:`~stem.response.ControlMessage` read from the socket

1 0

[stem/master] Fix `AsyncController._event_listeners_lock`
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit e1dff48079d60674c7530871875666f054973c6a Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sun May 17 16:59:09 2020 +0300 Fix `AsyncController._event_listeners_lock` --- stem/control.py | 61 ++++++++++++++++++++++++++++----------------------------- 1 file changed, 30 insertions(+), 31 deletions(-) diff --git a/stem/control.py b/stem/control.py index 2ffcf173..bdcdfb75 100644 --- a/stem/control.py +++ b/stem/control.py @@ -1040,8 +1040,8 @@ class AsyncController(BaseController): # mapping of event types to their listeners - self._event_listeners = {} # type: Dict[stem.control.EventType, List[Callable[[stem.response.events.Event], None]]] - self._event_listeners_lock = threading.RLock() + self._event_listeners = {} # type: Dict[stem.control.EventType, List[Callable[[stem.response.events.Event], Union[None, Awaitable[None]]]]] + self._event_listeners_lock = stem.util.CombinedReentrantAndAsyncioLock() self._enabled_features = [] # type: List[str] self._last_address_exc = None # type: Optional[BaseException] @@ -3062,7 +3062,7 @@ class AsyncController(BaseController): # first checking that tor supports these event types - with self._event_listeners_lock: + async with self._event_listeners_lock: if self.is_authenticated(): for event_type in events: event_type = stem.response.events.EVENT_TYPE_TO_CLASS.get(event_type) @@ -3091,7 +3091,7 @@ class AsyncController(BaseController): :raises: :class:`stem.ProtocolError` if unable to set the events """ - with self._event_listeners_lock: + async with self._event_listeners_lock: event_types_changed = False for event_type, event_listeners in list(self._event_listeners.items()): @@ -3760,7 +3760,7 @@ class AsyncController(BaseController): # try to re-attach event listeners to the new instance - with self._event_listeners_lock: + async with self._event_listeners_lock: try: failed_events = (await self._attach_listeners())[1] @@ -3807,7 +3807,7 @@ class AsyncController(BaseController): log.error('Tor sent a malformed event (%s): %s' % (exc, event_message)) event_type = MALFORMED_EVENTS - with self._event_listeners_lock: + async with self._event_listeners_lock: for listener_type, event_listeners in list(self._event_listeners.items()): if listener_type == event_type: for listener in event_listeners: @@ -3830,34 +3830,33 @@ class AsyncController(BaseController): set_events, failed_events = [], [] - with self._event_listeners_lock: - if self.is_authenticated(): - # try to set them all - response = await self.msg('SETEVENTS %s' % ' '.join(self._event_listeners.keys())) + if self.is_authenticated(): + # try to set them all + response = await self.msg('SETEVENTS %s' % ' '.join(self._event_listeners.keys())) - if response.is_ok(): - set_events = list(self._event_listeners.keys()) - else: - # One of the following likely happened... - # - # * Our user attached listeners before having an authenticated - # connection, so we couldn't check if we met the version - # requirement. - # - # * User attached listeners to one tor instance, then connected us to - # an older tor instancce. - # - # * Some other controller hiccup (far less likely). - # - # See if we can set some subset of our events. + if response.is_ok(): + set_events = list(self._event_listeners.keys()) + else: + # One of the following likely happened... + # + # * Our user attached listeners before having an authenticated + # connection, so we couldn't check if we met the version + # requirement. + # + # * User attached listeners to one tor instance, then connected us to + # an older tor instancce. + # + # * Some other controller hiccup (far less likely). + # + # See if we can set some subset of our events. - for event in list(self._event_listeners.keys()): - response = await self.msg('SETEVENTS %s' % ' '.join(set_events + [event])) + for event in list(self._event_listeners.keys()): + response = await self.msg('SETEVENTS %s' % ' '.join(set_events + [event])) - if response.is_ok(): - set_events.append(event) - else: - failed_events.append(event) + if response.is_ok(): + set_events.append(event) + else: + failed_events.append(event) return (set_events, failed_events)

1 0

[stem/master] Fix mypy errors related to `stem.descriptor.remote.Query` attributes
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 626d540d2d5fc6c9039ff738546795e7818324e9 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sun May 17 17:57:18 2020 +0300 Fix mypy errors related to `stem.descriptor.remote.Query` attributes --- stem/descriptor/remote.py | 93 ++++++++++++++++++++++++++++++++++------------- 1 file changed, 67 insertions(+), 26 deletions(-) diff --git a/stem/descriptor/remote.py b/stem/descriptor/remote.py index 429954cb..4e1ad034 100644 --- a/stem/descriptor/remote.py +++ b/stem/descriptor/remote.py @@ -591,32 +591,73 @@ class Query(stem.util.AsyncClassWrapper): for desc in self._execute_async_generator_method('__aiter__'): yield desc - # Add public attributes of `AsyncQuery` as properties. - for attr in ( - 'descriptor_type', - 'endpoints', - 'resource', - 'compression', - 'retries', - 'fall_back_to_authority', - 'content', - 'error', - 'is_done', - 'download_url', - 'start_time', - 'timeout', - 'runtime', - 'validate', - 'document_handler', - 'reply_headers', - 'kwargs', - ): - locals()[attr] = property( - functools.partial( - lambda self, attr_name: getattr(self._wrapped_instance, attr_name), - attr_name=attr, - ), - ) + @property + def descriptor_type(self) -> str: + return self._wrapped_instance.descriptor_type + + @property + def endpoints(self) -> List[Union[stem.ORPort, stem.DirPort]]: + return self._wrapped_instance.endpoints + + @property + def resource(self) -> str: + return self._wrapped_instance.resource + + @property + def compression(self) -> List[stem.descriptor._Compression]: + return self._wrapped_instance.compression + + @property + def retries(self) -> int: + return self._wrapped_instance.retries + + @property + def fall_back_to_authority(self) -> bool: + return self._wrapped_instance.fall_back_to_authority + + @property + def content(self) -> Optional[bytes]: + return self._wrapped_instance.content + + @property + def error(self) -> Optional[BaseException]: + return self._wrapped_instance.error + + @property + def is_done(self) -> bool: + return self._wrapped_instance.is_done + + @property + def download_url(self) -> Optional[str]: + return self._wrapped_instance.download_url + + @property + def start_time(self) -> Optional[float]: + return self._wrapped_instance.start_time + + @property + def timeout(self) -> Optional[float]: + return self._wrapped_instance.timeout + + @property + def runtime(self) -> Optional[float]: + return self._wrapped_instance.runtime + + @property + def validate(self) -> bool: + return self._wrapped_instance.validate + + @property + def document_handler(self) -> stem.descriptor.DocumentHandler: + return self._wrapped_instance.document_handler + + @property + def reply_headers(self) -> Optional[Dict[str, str]]: + return self._wrapped_instance.reply_headers + + @property + def kwargs(self) -> Dict[str, Any]: + return self._wrapped_instance.kwargs class DescriptorDownloader(object):

1 0

[stem/master] Update docstrings in stem/descriptor/remote.py
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 9310094a8844531aeadc53caa2df5ca503fb3d34 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sun May 17 19:30:17 2020 +0300 Update docstrings in stem/descriptor/remote.py --- stem/descriptor/remote.py | 213 ++++++++++++++++++++++++++++++++++++---------- 1 file changed, 167 insertions(+), 46 deletions(-) diff --git a/stem/descriptor/remote.py b/stem/descriptor/remote.py index 4e1ad034..d7a833a4 100644 --- a/stem/descriptor/remote.py +++ b/stem/descriptor/remote.py @@ -238,23 +238,18 @@ class AsyncQuery(object): advanced usage. To block on the response and get results either call - :func:`~stem.descriptor.remote.Query.run` or iterate over the Query. The - :func:`~stem.descriptor.remote.Query.run` method pass along any errors that - arise... + :func:`~stem.descriptor.remote.AsyncQuery.run` or iterate over the Query. The + :func:`~stem.descriptor.remote.AsyncQuery.run` method pass along any errors + that arise... :: - from stem.descriptor.remote import Query - - query = Query( - '/tor/server/all', - timeout = 30, - ) + from stem.descriptor.remote import AsyncQuery print('Current relays:') try: - for desc in Query('/tor/server/all', 'server-descriptor 1.0').run(): + for desc in await AsyncQuery('/tor/server/all', 'server-descriptor 1.0').run(): print(desc.fingerprint) except Exception as exc: print('Unable to retrieve the server descriptors: %s' % exc) @@ -265,7 +260,7 @@ class AsyncQuery(object): print('Current relays:') - for desc in Query('/tor/server/all', 'server-descriptor 1.0'): + async for desc in AsyncQuery('/tor/server/all', 'server-descriptor 1.0'): print(desc.fingerprint) In either case exceptions are available via our 'error' attribute. @@ -298,39 +293,6 @@ class AsyncQuery(object): For legacy reasons if our resource has a '.z' suffix then our **compression** argument is overwritten with Compression.GZIP. - .. versionchanged:: 1.7.0 - Added support for downloading from ORPorts. - - .. versionchanged:: 1.7.0 - Added the compression argument. - - .. versionchanged:: 1.7.0 - Added the reply_headers attribute. - - The class this provides changed between Python versions. In python2 - this was called httplib.HTTPMessage, whereas in python3 the class was - renamed to http.client.HTTPMessage. - - .. versionchanged:: 1.7.0 - Avoid downloading from tor26. This directory authority throttles its - DirPort to such an extent that requests either time out or take on the - order of minutes. - - .. versionchanged:: 1.7.0 - Avoid downloading from Bifroest. This is the bridge authority so it - doesn't vote in the consensus, and apparently times out frequently. - - .. versionchanged:: 1.8.0 - Serge has replaced Bifroest as our bridge authority. Avoiding descriptor - downloads from it instead. - - .. versionchanged:: 1.8.0 - Defaulting to gzip compression rather than plaintext downloads. - - .. versionchanged:: 1.8.0 - Using :class:`~stem.descriptor.__init__.Compression` for our compression - argument. - :var str resource: resource being fetched, such as '/tor/server/all' :var str descriptor_type: type of descriptors being fetched (for options see :func:`~stem.descriptor.__init__.parse_file`), this is guessed from the @@ -562,7 +524,145 @@ class AsyncQuery(object): class Query(stem.util.AsyncClassWrapper): - def __init__(self, resource, descriptor_type = None, endpoints = None, compression = (Compression.GZIP,), retries = 2, fall_back_to_authority = False, timeout = None, start = True, block = False, validate = False, document_handler = stem.descriptor.DocumentHandler.ENTRIES, **kwargs): + """ + Asynchronous request for descriptor content from a directory authority or + mirror. These can either be made through the + :class:`~stem.descriptor.remote.DescriptorDownloader` or directly for more + advanced usage. + + To block on the response and get results either call + :func:`~stem.descriptor.remote.Query.run` or iterate over the Query. The + :func:`~stem.descriptor.remote.Query.run` method pass along any errors that + arise... + + :: + + from stem.descriptor.remote import Query + + print('Current relays:') + + try: + for desc in Query('/tor/server/all', 'server-descriptor 1.0').run(): + print(desc.fingerprint) + except Exception as exc: + print('Unable to retrieve the server descriptors: %s' % exc) + + ... while iterating fails silently... + + :: + + print('Current relays:') + + for desc in Query('/tor/server/all', 'server-descriptor 1.0'): + print(desc.fingerprint) + + In either case exceptions are available via our 'error' attribute. + + Tor provides quite a few different descriptor resources via its directory + protocol (see section 4.2 and later of the `dir-spec + <https://gitweb.torproject.org/torspec.git/tree/dir-spec.txt>`_). + Commonly useful ones include... + + =============================================== =========== + Resource Description + =============================================== =========== + /tor/server/all all present server descriptors + /tor/server/fp/<fp1>+<fp2>+<fp3> server descriptors with the given fingerprints + /tor/extra/all all present extrainfo descriptors + /tor/extra/fp/<fp1>+<fp2>+<fp3> extrainfo descriptors with the given fingerprints + /tor/micro/d/<hash1>-<hash2> microdescriptors with the given hashes + /tor/status-vote/current/consensus present consensus + /tor/status-vote/current/consensus-microdesc present microdescriptor consensus + /tor/status-vote/next/bandwidth bandwidth authority heuristics for the next consenus + /tor/status-vote/next/consensus-signatures detached signature, used for making the next consenus + /tor/keys/all key certificates for the authorities + /tor/keys/fp/<v3ident1>+<v3ident2> key certificates for specific authorities + =============================================== =========== + + **ZSTD** compression requires `zstandard + <https://pypi.org/project/zstandard/>`_, and **LZMA** requires the `lzma + module <https://docs.python.org/3/library/lzma.html>`_. + + For legacy reasons if our resource has a '.z' suffix then our **compression** + argument is overwritten with Compression.GZIP. + + .. versionchanged:: 1.7.0 + Added support for downloading from ORPorts. + + .. versionchanged:: 1.7.0 + Added the compression argument. + + .. versionchanged:: 1.7.0 + Added the reply_headers attribute. + + The class this provides changed between Python versions. In python2 + this was called httplib.HTTPMessage, whereas in python3 the class was + renamed to http.client.HTTPMessage. + + .. versionchanged:: 1.7.0 + Avoid downloading from tor26. This directory authority throttles its + DirPort to such an extent that requests either time out or take on the + order of minutes. + + .. versionchanged:: 1.7.0 + Avoid downloading from Bifroest. This is the bridge authority so it + doesn't vote in the consensus, and apparently times out frequently. + + .. versionchanged:: 1.8.0 + Serge has replaced Bifroest as our bridge authority. Avoiding descriptor + downloads from it instead. + + .. versionchanged:: 1.8.0 + Defaulting to gzip compression rather than plaintext downloads. + + .. versionchanged:: 1.8.0 + Using :class:`~stem.descriptor.__init__.Compression` for our compression + argument. + + :var str resource: resource being fetched, such as '/tor/server/all' + :var str descriptor_type: type of descriptors being fetched (for options see + :func:`~stem.descriptor.__init__.parse_file`), this is guessed from the + resource if **None** + + :var list endpoints: :class:`~stem.DirPort` or :class:`~stem.ORPort` of the + authority or mirror we're querying, this uses authorities if undefined + :var list compression: list of :data:`stem.descriptor.Compression` + we're willing to accept, when none are mutually supported downloads fall + back to Compression.PLAINTEXT + :var int retries: number of times to attempt the request if downloading it + fails + :var bool fall_back_to_authority: when retrying request issues the last + request to a directory authority if **True** + + :var str content: downloaded descriptor content + :var Exception error: exception if a problem occured + :var bool is_done: flag that indicates if our request has finished + + :var float start_time: unix timestamp when we first started running + :var http.client.HTTPMessage reply_headers: headers provided in the response, + **None** if we haven't yet made our request + :var float runtime: time our query took, this is **None** if it's not yet + finished + + :var bool validate: checks the validity of the descriptor's content if + **True**, skips these checks otherwise + :var stem.descriptor.__init__.DocumentHandler document_handler: method in + which to parse a :class:`~stem.descriptor.networkstatus.NetworkStatusDocument` + :var dict kwargs: additional arguments for the descriptor constructor + + Following are only applicable when downloading from a + :class:`~stem.DirPort`... + + :var float timeout: duration before we'll time out our request + :var str download_url: last url used to download the descriptor, this is + unset until we've actually made a download attempt + + :param start: start making the request when constructed (default is **True**) + :param block: only return after the request has been completed, this is + the same as running **query.run(True)** (default is **False**) + """ + + 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( @@ -582,9 +682,30 @@ class Query(stem.util.AsyncClassWrapper): ) def start(self) -> None: + """ + Starts downloading the scriptors if we haven't started already. + """ + self._call_async_method_soon('start') - def run(self, suppress = False): + def run(self, suppress: bool = 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. + + :param suppress: avoids raising exceptions if **True** + + :returns: list for the requested :class:`~stem.descriptor.__init__.Descriptor` instances + + :raises: + Using the iterator can fail with the following if **suppress** is + **False**... + + * **ValueError** if the descriptor contents is malformed + * :class:`~stem.DownloadTimeout` if our request timed out + * :class:`~stem.DownloadFailed` if our request fails + """ + return self._execute_async_method('run', suppress) def __iter__(self):

1 0

[stem/master] Add type hints to new functions
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit fb803c96db3ac357850b93730a33154adc719a89 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sun May 17 17:40:29 2020 +0300 Add type hints to new functions --- stem/descriptor/remote.py | 4 ++-- stem/util/__init__.py | 31 +++++++++++++++++-------------- stem/util/test_tools.py | 8 ++++---- 3 files changed, 23 insertions(+), 20 deletions(-) diff --git a/stem/descriptor/remote.py b/stem/descriptor/remote.py index daa3b42c..429954cb 100644 --- a/stem/descriptor/remote.py +++ b/stem/descriptor/remote.py @@ -581,8 +581,8 @@ class Query(stem.util.AsyncClassWrapper): **kwargs, ) - def start(self): - return self._call_async_method_soon('start') + def start(self) -> None: + self._call_async_method_soon('start') def run(self, suppress = False): return self._execute_async_method('run', suppress) diff --git a/stem/util/__init__.py b/stem/util/__init__.py index a230cfbd..7c53730c 100644 --- a/stem/util/__init__.py +++ b/stem/util/__init__.py @@ -8,8 +8,10 @@ Utility functions used by the stem library. import asyncio import datetime import threading +from concurrent.futures import Future -from typing import Any, Union +from types import TracebackType +from typing import Any, AsyncIterator, Iterator, Optional, Type, Union __all__ = [ 'conf', @@ -150,36 +152,37 @@ class CombinedReentrantAndAsyncioLock: __slots__ = ('_r_lock', '_async_lock') - def __init__(self): + def __init__(self) -> None: self._r_lock = threading.RLock() self._async_lock = asyncio.Lock() - async def acquire(self): + async def acquire(self) -> bool: await self._async_lock.acquire() self._r_lock.acquire() + return True - def release(self): + def release(self) -> None: self._r_lock.release() self._async_lock.release() - async def __aenter__(self): + async def __aenter__(self) -> 'CombinedReentrantAndAsyncioLock': await self.acquire() return self - async def __aexit__(self, exc_type, exc_val, exc_tb): + async def __aexit__(self, exit_type: Optional[Type[BaseException]], value: Optional[BaseException], traceback: Optional[TracebackType]) -> None: self.release() class ThreadForWrappedAsyncClass(threading.Thread): - def __init__(self, *args, **kwargs): + def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, *kwargs) self.loop = asyncio.new_event_loop() self.setDaemon(True) - def run(self): + def run(self) -> None: self.loop.run_forever() - def join(self, timeout=None): + def join(self, timeout: Optional[float] = None) -> None: self.loop.call_soon_threadsafe(self.loop.stop) super().join(timeout) self.loop.close() @@ -189,7 +192,7 @@ class AsyncClassWrapper: _thread_for_wrapped_class: ThreadForWrappedAsyncClass _wrapped_instance: type - def _init_async_class(self, async_class, *args, **kwargs): + def _init_async_class(self, async_class: Type, *args: Any, **kwargs: Any) -> Any: thread = self._thread_for_wrapped_class # The asynchronous class should be initialized in the thread where # its methods will be executed. @@ -201,17 +204,17 @@ class AsyncClassWrapper: return async_class(*args, **kwargs) - def _call_async_method_soon(self, method_name, *args, **kwargs): + def _call_async_method_soon(self, method_name: str, *args: Any, **kwargs: Any) -> Future: return asyncio.run_coroutine_threadsafe( getattr(self._wrapped_instance, method_name)(*args, **kwargs), self._thread_for_wrapped_class.loop, ) - def _execute_async_method(self, method_name, *args, **kwargs): + def _execute_async_method(self, method_name: str, *args: Any, **kwargs: Any) -> Any: return self._call_async_method_soon(method_name, *args, **kwargs).result() - def _execute_async_generator_method(self, method_name, *args, **kwargs): - async def convert_async_generator(generator): + def _execute_async_generator_method(self, method_name: str, *args: Any, **kwargs: Any) -> Iterator: + async def convert_async_generator(generator: AsyncIterator) -> Iterator: return iter([d async for d in generator]) return asyncio.run_coroutine_threadsafe( diff --git a/stem/util/test_tools.py b/stem/util/test_tools.py index 455f3da3..f5ae7525 100644 --- a/stem/util/test_tools.py +++ b/stem/util/test_tools.py @@ -46,7 +46,7 @@ import stem.util.conf import stem.util.enum import stem.util.system -from typing import Any, Callable, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple, Type, Union +from typing import Any, Awaitable, Callable, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple, Type, Union CONFIG = stem.util.conf.config_dict('test', { 'pycodestyle.ignore': [], @@ -686,7 +686,7 @@ def _is_ignored(config: Mapping[str, Sequence[str]], path: str, issue: str) -> b def async_test(func: Callable) -> Callable: @functools.wraps(func) - def wrapper(*args, **kwargs): + def wrapper(*args: Any, **kwargs: Any) -> Any: loop = asyncio.new_event_loop() try: result = loop.run_until_complete(func(*args, **kwargs)) @@ -696,13 +696,13 @@ def async_test(func: Callable) -> Callable: return wrapper -def coro_func_returning_value(return_value): +def coro_func_returning_value(return_value: Any) -> Callable[..., Awaitable]: async def coroutine_func(*args, **kwargs): return return_value return coroutine_func -def coro_func_raising_exc(exc): +def coro_func_raising_exc(exc: Exception) -> Callable[..., Awaitable]: async def coroutine_func(*args, **kwargs): raise exc return coroutine_func

1 0

[stem/master] Implement setting docstrings of `stem.control.Controller` methods
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 26d2222480d6b10ef57097c9f40e44112408d1c1 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sun May 17 20:10:22 2020 +0300 Implement setting docstrings of `stem.control.Controller` methods --- stem/control.py | 80 +++++++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 78 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index bdcdfb75..6e15c16c 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3861,7 +3861,18 @@ class AsyncController(BaseController): return (set_events, failed_events) +def _set_doc_from_async_controller(func: Callable) -> Callable: + func.__doc__ = getattr(AsyncController, func.__name__).__doc__ + return func + + class Controller(_BaseControllerSocketMixin, stem.util.AsyncClassWrapper): + """ + Connection with Tor's control socket. This wraps + :class:`~stem.control.AsyncController` to provide a synchronous + interface and for backwards compatibility. + """ + @classmethod def from_port(cls: Type, address: str = '127.0.0.1', port: Union[int, str] = 'default') -> 'stem.control.Controller': """ @@ -3915,198 +3926,263 @@ class Controller(_BaseControllerSocketMixin, stem.util.AsyncClassWrapper): self._wrapped_instance: AsyncController = self._init_async_class(AsyncController, control_socket, is_authenticated) self._socket = self._wrapped_instance._socket + @_set_doc_from_async_controller def msg(self, message: str) -> stem.response.ControlMessage: return self._execute_async_method('msg', message) + @_set_doc_from_async_controller def is_authenticated(self) -> bool: return self._wrapped_instance.is_authenticated() + @_set_doc_from_async_controller def connect(self) -> None: self._execute_async_method('connect') + @_set_doc_from_async_controller def reconnect(self, *args: Any, **kwargs: Any) -> None: self._execute_async_method('reconnect', *args, **kwargs) + @_set_doc_from_async_controller def close(self) -> None: self._execute_async_method('close') + @_set_doc_from_async_controller def get_latest_heartbeat(self) -> float: return self._wrapped_instance.get_latest_heartbeat() + @_set_doc_from_async_controller def add_status_listener(self, callback: Callable[['stem.control.BaseController', 'stem.control.State', float], None], spawn: bool = True) -> None: self._wrapped_instance.add_status_listener(callback, spawn) + @_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) + @_set_doc_from_async_controller def authenticate(self, *args: Any, **kwargs: Any) -> None: self._execute_async_method('authenticate', *args, **kwargs) + @_set_doc_from_async_controller def get_info(self, params: Union[str, Sequence[str]], default: Any = UNDEFINED, get_bytes: bool = False) -> Union[str, Dict[str, str]]: return self._execute_async_method('get_info', params, default, get_bytes) + @_set_doc_from_async_controller def get_version(self, default: Any = UNDEFINED) -> stem.version.Version: return self._execute_async_method('get_version', default) + @_set_doc_from_async_controller def get_exit_policy(self, default: Any = UNDEFINED) -> stem.exit_policy.ExitPolicy: return self._execute_async_method('get_exit_policy', default) + @_set_doc_from_async_controller def get_ports(self, listener_type: 'stem.control.Listener', default: Any = UNDEFINED) -> Sequence[int]: return self._execute_async_method('get_ports', listener_type, default) + @_set_doc_from_async_controller def get_listeners(self, listener_type: 'stem.control.Listener', default: Any = UNDEFINED) -> Sequence[Tuple[str, int]]: return self._execute_async_method('get_listeners', listener_type, default) + @_set_doc_from_async_controller def get_accounting_stats(self, default: Any = UNDEFINED) -> 'stem.control.AccountingStats': return self._execute_async_method('get_accounting_stats', default) + @_set_doc_from_async_controller def get_protocolinfo(self, default: Any = UNDEFINED) -> stem.response.protocolinfo.ProtocolInfoResponse: return self._execute_async_method('get_protocolinfo', default) + @_set_doc_from_async_controller def get_user(self, default: Any = UNDEFINED) -> str: return self._execute_async_method('get_user', default) + @_set_doc_from_async_controller def get_pid(self, default: Any = UNDEFINED) -> int: return self._execute_async_method('get_pid', default) + @_set_doc_from_async_controller def get_start_time(self, default: Any = UNDEFINED) -> float: return self._execute_async_method('get_start_time', default) + @_set_doc_from_async_controller def get_uptime(self, default: Any = UNDEFINED) -> float: return self._execute_async_method('get_uptime', default) + @_set_doc_from_async_controller def is_user_traffic_allowed(self) -> 'stem.control.UserTrafficAllowed': return self._execute_async_method('is_user_traffic_allowed') + @_set_doc_from_async_controller def get_microdescriptor(self, relay: Optional[str] = None, default: Any = UNDEFINED) -> stem.descriptor.microdescriptor.Microdescriptor: return self._execute_async_method('get_microdescriptor', relay, default) + @_set_doc_from_async_controller def get_microdescriptors(self, default: Any = UNDEFINED) -> Iterator[stem.descriptor.microdescriptor.Microdescriptor]: return self._execute_async_generator_method('get_microdescriptors', default) + @_set_doc_from_async_controller def get_server_descriptor(self, relay: Optional[str] = None, default: Any = UNDEFINED) -> stem.descriptor.server_descriptor.RelayDescriptor: return self._execute_async_method('get_server_descriptor', relay, default) + @_set_doc_from_async_controller def get_server_descriptors(self, default: Any = UNDEFINED) -> Iterator[stem.descriptor.server_descriptor.RelayDescriptor]: return self._execute_async_generator_method('get_server_descriptors', default) + @_set_doc_from_async_controller def get_network_status(self, relay: Optional[str] = None, default: Any = UNDEFINED) -> stem.descriptor.router_status_entry.RouterStatusEntryV3: return self._execute_async_method('get_network_status', relay, default) + @_set_doc_from_async_controller def get_network_statuses(self, default: Any = UNDEFINED) -> Iterator[stem.descriptor.router_status_entry.RouterStatusEntryV3]: return self._execute_async_generator_method('get_network_statuses', default) + @_set_doc_from_async_controller def get_hidden_service_descriptor(self, address: str, default: Any = UNDEFINED, servers: Optional[Sequence[str]] = None, await_result: bool = True, timeout: Optional[float] = None) -> stem.descriptor.hidden_service.HiddenServiceDescriptorV2: return self._execute_async_method('get_hidden_service_descriptor', address, default, servers, await_result, timeout) + @_set_doc_from_async_controller def get_conf(self, param: str, default: Any = UNDEFINED, multiple: bool = False) -> Union[str, Sequence[str]]: return self._execute_async_method('get_conf', param, default, multiple) + @_set_doc_from_async_controller def get_conf_map(self, params: Union[str, Sequence[str]], default: Any = UNDEFINED, multiple: bool = True) -> Dict[str, Union[str, Sequence[str]]]: return self._execute_async_method('get_conf_map', params, default, multiple) + @_set_doc_from_async_controller def is_set(self, param: str, default: Any = UNDEFINED) -> bool: return self._execute_async_method('is_set', param, default) + @_set_doc_from_async_controller def set_conf(self, param: str, value: Union[str, Sequence[str]]) -> None: self._execute_async_method('set_conf', param, value) + @_set_doc_from_async_controller def reset_conf(self, *params: str) -> None: self._execute_async_method('reset_conf', *params) + @_set_doc_from_async_controller def set_options(self, params: Union[Mapping[str, Union[str, Sequence[str]]], Sequence[Tuple[str, Union[str, Sequence[str]]]]], reset: bool = False) -> None: self._execute_async_method('set_options', params, reset) + @_set_doc_from_async_controller def get_hidden_service_conf(self, default: Any = UNDEFINED) -> Dict[str, Any]: return self._execute_async_method('get_hidden_service_conf', default) + @_set_doc_from_async_controller def set_hidden_service_conf(self, conf: Mapping[str, Any]) -> None: self._execute_async_method('set_hidden_service_conf', conf) + @_set_doc_from_async_controller def create_hidden_service(self, path: str, port: int, target_address: Optional[str] = None, target_port: Optional[int] = None, auth_type: Optional[str] = None, client_names: Optional[Sequence[str]] = None) -> 'stem.control.CreateHiddenServiceOutput': return self._execute_async_method('create_hidden_service', path, port, target_address, target_port, auth_type, client_names) + @_set_doc_from_async_controller def remove_hidden_service(self, path: str, port: Optional[int] = None) -> bool: return self._execute_async_method('remove_hidden_service', path, port) + @_set_doc_from_async_controller def list_ephemeral_hidden_services(self, default: Any = UNDEFINED, our_services: bool = True, detached: bool = False) -> Sequence[str]: return self._execute_async_method('list_ephemeral_hidden_services', default, our_services, detached) + @_set_doc_from_async_controller def create_ephemeral_hidden_service(self, ports: Union[int, Sequence[int], Mapping[int, str]], key_type: str = 'NEW', key_content: str = 'BEST', discard_key: bool = False, detached: bool = False, await_publication: bool = False, timeout: Optional[float] = None, basic_auth: Optional[Mapping[str, str]] = None, max_streams: Optional[int] = None) -> stem.response.add_onion.AddOnionResponse: return self._execute_async_method('create_ephemeral_hidden_service', ports, key_type, key_content, discard_key, detached, await_publication, timeout, basic_auth, max_streams) + @_set_doc_from_async_controller def remove_ephemeral_hidden_service(self, service_id: str) -> bool: return self._execute_async_method('remove_ephemeral_hidden_service', service_id) - def add_event_listener(self, listener: Callable[[stem.response.events.Event], None], *events: 'stem.control.EventType') -> None: + @_set_doc_from_async_controller + def add_event_listener(self, listener: Callable[[stem.response.events.Event], Union[None, Awaitable[None]]], *events: 'stem.control.EventType') -> None: self._execute_async_method('add_event_listener', listener, *events) - def remove_event_listener(self, listener: Callable[[stem.response.events.Event], None]) -> None: + @_set_doc_from_async_controller + def remove_event_listener(self, listener: Callable[[stem.response.events.Event], Union[None, Awaitable[None]]]) -> None: self._execute_async_method('remove_event_listener', listener) + @_set_doc_from_async_controller def is_caching_enabled(self) -> bool: return self._wrapped_instance.is_caching_enabled() + @_set_doc_from_async_controller def set_caching(self, enabled: bool) -> None: self._wrapped_instance.set_caching(enabled) + @_set_doc_from_async_controller def clear_cache(self) -> None: self._wrapped_instance.clear_cache() + @_set_doc_from_async_controller def load_conf(self, configtext: str) -> None: self._execute_async_method('load_conf', configtext) + @_set_doc_from_async_controller def save_conf(self, force: bool = False) -> None: return self._execute_async_method('save_conf', force) + @_set_doc_from_async_controller def is_feature_enabled(self, feature: str) -> bool: return self._wrapped_instance.is_feature_enabled(feature) + @_set_doc_from_async_controller def enable_feature(self, features: Union[str, Sequence[str]]) -> None: self._wrapped_instance.enable_feature(features) + @_set_doc_from_async_controller def get_circuit(self, circuit_id: int, default: Any = UNDEFINED) -> stem.response.events.CircuitEvent: return self._execute_async_method('get_circuit', circuit_id, default) + @_set_doc_from_async_controller def get_circuits(self, default: Any = UNDEFINED) -> List[stem.response.events.CircuitEvent]: return self._execute_async_method('get_circuits', default) + @_set_doc_from_async_controller def new_circuit(self, path: Union[None, str, Sequence[str]] = None, purpose: str = 'general', await_build: bool = False, timeout: Optional[float] = None) -> str: return self._execute_async_method('new_circuit', path, purpose, await_build, timeout) + @_set_doc_from_async_controller def extend_circuit(self, circuit_id: str = '0', path: Union[None, str, Sequence[str]] = None, purpose: str = 'general', await_build: bool = False, timeout: Optional[float] = None) -> str: return self._execute_async_method('extend_circuit', circuit_id, path, purpose, await_build, timeout) + @_set_doc_from_async_controller def repurpose_circuit(self, circuit_id: str, purpose: str) -> None: self._execute_async_method('repurpose_circuit', circuit_id, purpose) + @_set_doc_from_async_controller def close_circuit(self, circuit_id: str, flag: str = '') -> None: self._execute_async_method('close_circuit', circuit_id, flag) + @_set_doc_from_async_controller def get_streams(self, default: Any = UNDEFINED) -> List[stem.response.events.StreamEvent]: return self._execute_async_method('get_streams', default) + @_set_doc_from_async_controller def attach_stream(self, stream_id: str, circuit_id: str, exiting_hop: Optional[int] = None) -> None: self._execute_async_method('attach_stream', stream_id, circuit_id, exiting_hop) + @_set_doc_from_async_controller def close_stream(self, stream_id: str, reason: stem.RelayEndReason = stem.RelayEndReason.MISC, flag: str = '') -> None: self._execute_async_method('close_stream', stream_id, reason, flag) + @_set_doc_from_async_controller def signal(self, signal: stem.Signal) -> None: self._execute_async_method('signal', signal) + @_set_doc_from_async_controller def is_newnym_available(self) -> bool: return self._wrapped_instance.is_newnym_available() + @_set_doc_from_async_controller def get_newnym_wait(self) -> float: return self._wrapped_instance.get_newnym_wait() + @_set_doc_from_async_controller def get_effective_rate(self, default: Any = UNDEFINED, burst: bool = False) -> int: return self._execute_async_method('get_effective_rate', default, burst) + @_set_doc_from_async_controller def map_address(self, mapping: Mapping[str, str]) -> Dict[str, str]: return self._execute_async_method('map_address', mapping) + @_set_doc_from_async_controller def drop_guards(self) -> None: self._execute_async_method('drop_guards')

1 0

[stem/master] Update docstrings in stem/connection.py
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit cfd1f3d74f0dbb118c948d8520437498b5a30465 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sun May 17 19:06:25 2020 +0300 Update docstrings in stem/connection.py --- stem/connection.py | 115 ++++++++++++++++++++++++++++++++++++++--------------- 1 file changed, 82 insertions(+), 33 deletions(-) diff --git a/stem/connection.py b/stem/connection.py index 8f57f3b3..de76a345 100644 --- a/stem/connection.py +++ b/stem/connection.py @@ -22,60 +22,74 @@ exceptions, etc). if not controller: sys.exit(1) # unable to get a connection - print 'Tor is running version %s' % controller.get_version() + print(f'Tor is running version {controller.get_version()}') controller.close() :: % python example.py - Tor is running version 0.2.4.10-alpha-dev (git-8be6058d8f31e578) + Tor is running version 0.4.3.5 ... or if Tor isn't running... :: % python example.py - [Errno 111] Connection refused + [Errno 111] Connect call failed The :func:`~stem.connection.authenticate` function, however, gives easy but fine-grained control over the authentication process. For instance... :: - import sys + import asyncio import getpass + import sys + import stem.connection import stem.socket - try: - control_socket = stem.socket.ControlPort(port = 9051) - except stem.SocketError as exc: - print 'Unable to connect to port 9051 (%s)' % exc - sys.exit(1) - try: - stem.connection.authenticate(control_socket) - except stem.connection.IncorrectSocketType: - print 'Please check in your torrc that 9051 is the ControlPort.' - print 'Maybe you configured it to be the ORPort or SocksPort instead?' - sys.exit(1) - except stem.connection.MissingPassword: - controller_password = getpass.getpass('Controller password: ') + async def authenticate() -> None: + try: + control_socket = stem.socket.ControlPort(port=9051) + await control_socket.connect() + except stem.SocketError as exc: + print(f'Unable to connect to port 9051 ({exc})') + sys.exit(1) try: - stem.connection.authenticate_password(control_socket, controller_password) - except stem.connection.PasswordAuthFailed: - print 'Unable to authenticate, password is incorrect' + await stem.connection.authenticate(control_socket) + except stem.connection.IncorrectSocketType: + print('Please check in your torrc that 9051 is the ControlPort.') + print('Maybe you configured it to be the ORPort or SocksPort instead?') + sys.exit(1) + except stem.connection.MissingPassword: + controller_password = getpass.getpass('Controller password: ') + + try: + await stem.connection.authenticate_password(control_socket, controller_password) + except stem.connection.PasswordAuthFailed: + print('Unable to authenticate, password is incorrect') + sys.exit(1) + except stem.connection.AuthenticationFailure as exc: + print(f'Unable to authenticate: {exc}') sys.exit(1) - except stem.connection.AuthenticationFailure as exc: - print 'Unable to authenticate: %s' % exc - sys.exit(1) + + + if __name__ == '__main__': + loop = asyncio.get_event_loop() + try: + loop.run_until_complete(authenticate()) + finally: + loop.close() **Module Overview:** :: - connect - Simple method for getting authenticated control connection + connect - Simple method for getting authenticated control connection for synchronous usage. + async_connect - Simple method for getting authenticated control connection for asynchronous usage. authenticate - Main method for authenticating to a control socket authenticate_none - Authenticates to an open control socket @@ -215,10 +229,10 @@ 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: """ - Convenience function for quickly getting a control connection. This is very - handy for debugging or CLI setup, handling setup and prompting for a password - if necessary (and none is provided). If any issues arise this prints a - description of the problem and returns **None**. + Convenience function for quickly getting a control connection for synchronous + usage. This is very handy for debugging or CLI setup, handling setup and + prompting for a password if necessary (and none is provided). If any issues + arise this prints a description of the problem and returns **None**. If both a **control_port** and **control_socket** are provided then the **control_socket** is tried first, and this provides a generic error message @@ -243,8 +257,8 @@ def connect(control_port: Tuple[str, Union[str, int]] = ('127.0.0.1', 'default') :param password_prompt: prompt for the controller password if it wasn't supplied :param chroot_path: path prefix if in a chroot environment - :param controller: :class:`~stem.control.BaseController` subclass to be - returned, this provides a :class:`~stem.socket.ControlSocket` if **None** + :param controller: :class:`~stem.control.Controller` subclass to be + returned :returns: authenticated control connection, the type based on the controller argument @@ -272,7 +286,41 @@ def connect(control_port: Tuple[str, Union[str, int]] = ('127.0.0.1', 'default') raise -async def connect_async(control_port = ('127.0.0.1', 'default'), control_socket = '/var/run/tor/control', password = None, password_prompt = False, chroot_path = None, controller = stem.control.AsyncController): +async def connect_async(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.BaseController] = stem.control.AsyncController) -> Any: + """ + Convenience function for quickly getting a control connection for + asynchronous usage. This is very handy for debugging or CLI setup, handling + setup and prompting for a password if necessary (and none is provided). If + any issues arise this prints a description of the problem and returns + **None**. + + If both a **control_port** and **control_socket** are provided then the + **control_socket** is tried first, and this provides a generic error message + if they're both unavailable. + + In much the same vein as git porcelain commands, users should not rely on + details of how this works. Messages and details of this function's behavior + could change in the future. + + If the **port** is **'default'** then this checks on both 9051 (default for + relays) and 9151 (default for the Tor Browser). This default may change in + the future. + + :param contol_port: address and port tuple, for instance **('127.0.0.1', 9051)** + :param control_socket: path where the control socket is located + :param password: passphrase to authenticate to the socket + :param password_prompt: prompt for the controller password if it wasn't + supplied + :param chroot_path: path prefix if in a chroot environment + :param controller: :class:`~stem.control.BaseController` subclass to be + returned + + :returns: authenticated control connection, the type based on the controller argument + + :raises: **ValueError** if given an invalid control_port, or both + **control_port** and **control_socket** are **None** + """ + if controller and not issubclass(controller, stem.control.BaseController): raise ValueError('The provided controller should be a stem.control.BaseController subclass.') return await _connect_async(control_port, control_socket, password, password_prompt, chroot_path, controller) @@ -339,8 +387,9 @@ async def _connect_auth(control_socket: stem.socket.ControlSocket, password: str :param password_prompt: prompt for the controller password if it wasn't supplied :param chroot_path: path prefix if in a chroot environment - :param controller: :class:`~stem.control.BaseController` subclass to be - returned, this provides a :class:`~stem.socket.ControlSocket` if **None** + :param controller: :class:`~stem.control.BaseController` or + :class:`~stem.control.Controller` subclass to be returned, this provides a + :class:`~stem.socket.ControlSocket` if **None** :returns: authenticated control connection, the type based on the controller argument """

1 0

[stem/master] Prototype Controller asyncio usage
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 341dc930bf89aed8567cad255b333b36576a5f06 Author: Damian Johnson <atagar(a)torproject.org> Date: Sat Mar 21 17:12:29 2020 -0700 Prototype Controller asyncio usage Still very much an experiment. Several things are broken but this adjusts our Controller successfully wrap an asyncio socket. from stem.control import Controller controller = Controller.from_port() controller.connect() controller.authenticate() print("Tor is running version %s" % controller.get_version()) controller.close() ---------------------------------------- % python3.7 demo.py Tor is running version 0.4.3.0-alpha-dev (git-5e70c27e8560ac18) Essentially this replaces our reader daemon with an asyncio loop. Synchronous code can invoke this loop in the following ways... * Non-blocking call of a coroutine self._asyncio_loop.create_task(my_coroutine()) * Blocking call of a coroutine result = asyncio.run_coroutine_threadsafe(my_coroutine(), self._asyncio_loop).result(timeout) * Non-blocking call of a function self._asyncio_loop.call_soon_threadsafe(my_function) I'm not quite sold yet on if we'll go this direction. An asynchronous socket has two potential advantages... * Simplify multi-threading * Offer asynchronous variants of our methods Either or both of these might be dead ends. I'll need to experiment more, but now that we've seen this work in this limited way we have something to build off of. --- stem/control.py | 56 ++++++++++++++++++++++++++++++++------------------------ 1 file changed, 32 insertions(+), 24 deletions(-) diff --git a/stem/control.py b/stem/control.py index ec88aa38..2d33bd54 100644 --- a/stem/control.py +++ b/stem/control.py @@ -240,13 +240,13 @@ If you're fine with allowing your script to raise exceptions then this can be mo =============== =========== """ +import asyncio import calendar import collections import functools import inspect import io import os -import queue import threading import time @@ -544,14 +544,21 @@ class BaseController(object): def __init__(self, control_socket: stem.socket.ControlSocket, is_authenticated: bool = False) -> None: self._socket = control_socket + + self._asyncio_loop = asyncio.get_event_loop() + + self._asyncio_thread = threading.Thread(target = self._asyncio_loop.run_forever, name = 'asyncio') + self._asyncio_thread.setDaemon(True) + self._asyncio_thread.start() + self._msg_lock = threading.RLock() self._status_listeners = [] # type: List[Tuple[Callable[[stem.control.BaseController, stem.control.State, float], None], bool]] # tuples of the form (callback, spawn_thread) self._status_listeners_lock = threading.RLock() # queues where incoming messages are directed - self._reply_queue = queue.Queue() # type: queue.Queue[Union[stem.response.ControlMessage, stem.ControllerError]] - self._event_queue = queue.Queue() # type: queue.Queue[stem.response.ControlMessage] + self._reply_queue = asyncio.Queue() # type: asyncio.Queue[Union[stem.response.ControlMessage, stem.ControllerError]] + self._event_queue = asyncio.Queue() # type: asyncio.Queue[stem.response.ControlMessage] # thread to continually pull from the control socket self._reader_thread = None # type: Optional[threading.Thread] @@ -637,15 +644,16 @@ class BaseController(object): log.info('Socket experienced a problem (%s)' % response) elif isinstance(response, stem.response.ControlMessage): log.info('Failed to deliver a response: %s' % response) - except queue.Empty: + except asyncio.QueueEmpty: # the empty() method is documented to not be fully reliable so this # isn't entirely surprising break try: - self._socket.send(message) - response = self._reply_queue.get() + self._asyncio_loop.create_task(self._socket.send(message)) + + response = asyncio.run_coroutine_threadsafe(self._reply_queue.get(), self._asyncio_loop).result() # If the message we received back had an exception then re-raise it to the # caller. Otherwise return the response. @@ -716,7 +724,7 @@ class BaseController(object): :raises: :class:`stem.SocketError` if unable to make a socket """ - self._socket.connect() + asyncio.run_coroutine_threadsafe(self._socket.connect(), self._asyncio_loop).result() def close(self) -> None: """ @@ -724,7 +732,7 @@ class BaseController(object): :func:`~stem.socket.BaseSocket.close` method. """ - self._socket.close() + asyncio.run_coroutine_threadsafe(self._socket.close(), self._asyncio_loop).result() # Join on any outstanding state change listeners. Closing is a state change # of its own, so if we have any listeners it's quite likely there's some @@ -737,6 +745,9 @@ class BaseController(object): if t.is_alive() and threading.current_thread() != t: t.join() + self._asyncio_loop.call_soon_threadsafe(self._asyncio_loop.stop) + self._asyncio_thread.join() + def get_socket(self) -> stem.socket.ControlSocket: """ Provides the socket used to speak with the tor process. Communicating with @@ -825,13 +836,13 @@ class BaseController(object): pass - def _connect(self) -> None: + async def _connect(self) -> None: self._launch_threads() self._notify_status_listeners(State.INIT) - self._socket_connect() + await self._socket_connect() self._is_authenticated = False - def _close(self) -> None: + async def _close(self) -> None: # Our is_alive() state is now false. Our reader thread should already be # awake from recv() raising a closure exception. Wake up the event thread # too so it can end. @@ -847,7 +858,7 @@ class BaseController(object): self._notify_status_listeners(State.CLOSED) - self._socket_close() + await self._socket_close() def _post_authentication(self) -> None: # actions to be taken after we have a newly authenticated connection @@ -904,21 +915,18 @@ class BaseController(object): them if we're restarted. """ + self._asyncio_loop.create_task(self._reader_loop()) + # In theory concurrent calls could result in multiple start() calls on a # single thread, which would cause an unexpected exception. Best be safe. with self._socket._get_send_lock(): - if not self._reader_thread or not self._reader_thread.is_alive(): - self._reader_thread = threading.Thread(target = self._reader_loop, name = 'Tor listener') - self._reader_thread.setDaemon(True) - self._reader_thread.start() - if not self._event_thread or not self._event_thread.is_alive(): self._event_thread = threading.Thread(target = self._event_loop, name = 'Event notifier') self._event_thread.setDaemon(True) self._event_thread.start() - def _reader_loop(self) -> None: + async def _reader_loop(self) -> None: """ Continually pulls from the control socket, directing the messages into queues based on their type. Controller messages come in two varieties... @@ -929,23 +937,23 @@ class BaseController(object): while self.is_alive(): try: - control_message = self._socket.recv() + control_message = await self._socket.recv() self._last_heartbeat = time.time() if control_message.content()[-1][0] == '650': # asynchronous message, adds to the event queue and wakes up its handler - self._event_queue.put(control_message) + await self._event_queue.put(control_message) self._event_notice.set() else: # response to a msg() call - self._reply_queue.put(control_message) + await self._reply_queue.put(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. - self._reply_queue.put(exc) + await self._reply_queue.put(exc) def _event_loop(self) -> None: """ @@ -970,7 +978,7 @@ class BaseController(object): socket_closed_at = time.time() elif time.time() - socket_closed_at > EVENTS_LISTENING_TIMEOUT: break - except queue.Empty: + except asyncio.queues.QueueEmpty: if not self.is_alive(): break @@ -3972,7 +3980,7 @@ def _get_with_timeout(event_queue: queue.Queue, timeout: float, start_time: floa try: return event_queue.get(True, time_left) - except queue.Empty: + except asyncio.queues.QueueEmpty: raise stem.Timeout('Reached our %0.1f second timeout' % timeout) else: return event_queue.get()

1 0

[stem/master] Revert "Add temporary scripts for testing async communication with sockets"
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit e6974dd5bbb804a6b98d172c2eb55206f6a47a56 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sat Mar 14 22:05:15 2020 +0200 Revert "Add temporary scripts for testing async communication with sockets" This reverts commit 072b5e682faa6d190fa8cd4cc52651fdc10b3e01. --- test_control_port.py | 30 ------------------------------ test_control_socket_file.py | 30 ------------------------------ test_relay_socket.py | 28 ---------------------------- 3 files changed, 88 deletions(-) diff --git a/test_control_port.py b/test_control_port.py deleted file mode 100644 index 85fe26d4..00000000 --- a/test_control_port.py +++ /dev/null @@ -1,30 +0,0 @@ -import asyncio - -from stem.async_socket import ControlPort - - -async def run_command(i, command): - async with ControlPort() as cp: - print(f'{i} Connecting') - await cp.connect() - print(f'{i} Authenticating') - await cp.send('AUTHENTICATE "password"') - print(f'{i} Receiving auth message') - await cp.recv() - print(f'{i} Sending the command') - await cp.send(command) - print(f'{i} Receiving result of the command') - result = await cp.recv() - print(f'{i} {result.content()}') - - -if __name__ == '__main__': - loop = asyncio.get_event_loop() - tasks = asyncio.gather( - run_command(1, 'PROTOCOLINFO 1'), - run_command(2, 'GETINFO traffic/read'), - ) - try: - loop.run_until_complete(tasks) - finally: - loop.close() diff --git a/test_control_socket_file.py b/test_control_socket_file.py deleted file mode 100644 index 4c6816f4..00000000 --- a/test_control_socket_file.py +++ /dev/null @@ -1,30 +0,0 @@ -import asyncio - -from stem.async_socket import ControlSocketFile - - -async def run_command(i, command): - async with ControlSocketFile() as cp: - print(f'{i} Connecting') - await cp.connect() - print(f'{i} Authenticating') - await cp.send('AUTHENTICATE "password"') - print(f'{i} Receiving auth message') - await cp.recv() - print(f'{i} Sending the command') - await cp.send(command) - print(f'{i} Receiving result of the command') - result = await cp.recv() - print(f'{i} {result.content()}') - - -if __name__ == '__main__': - loop = asyncio.get_event_loop() - tasks = asyncio.gather( - run_command(1, 'PROTOCOLINFO 1'), - run_command(2, 'GETINFO traffic/read'), - ) - try: - loop.run_until_complete(tasks) - finally: - loop.close() diff --git a/test_relay_socket.py b/test_relay_socket.py deleted file mode 100644 index 76289d46..00000000 --- a/test_relay_socket.py +++ /dev/null @@ -1,28 +0,0 @@ -import asyncio - -from stem.client import DEFAULT_LINK_PROTOCOLS -from stem.client.cell import VersionsCell -from stem.async_socket import RelaySocket - - -async def run_command(i, command): - async with RelaySocket(address='127.0.0.1', port=443) as cp: - print(f'{i} Connecting') - await cp.connect() - print(f'{i} Sending the command') - await cp.send(command) - print(f'{i} Receiving result of the command') - result = await cp.recv(2) - print(result) - - -if __name__ == '__main__': - loop = asyncio.get_event_loop() - tasks = asyncio.gather( - run_command(1, VersionsCell(DEFAULT_LINK_PROTOCOLS).pack(2)), - run_command(2, VersionsCell(DEFAULT_LINK_PROTOCOLS).pack(2)), - ) - try: - loop.run_until_complete(tasks) - finally: - loop.close()

1 0

[stem/master] Breath asynchrony into the `BaseController`
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 21ee7d288bdc1eacb7a9be805f5fcfcc28da8fc5 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sun Apr 12 19:42:40 2020 +0300 Breath asynchrony into the `BaseController` --- stem/control.py | 25 +++++++++---------------- 1 file changed, 9 insertions(+), 16 deletions(-) diff --git a/stem/control.py b/stem/control.py index a3b7d253..d592c4af 100644 --- a/stem/control.py +++ b/stem/control.py @@ -547,10 +547,6 @@ class BaseController(object): self._asyncio_loop = asyncio.get_event_loop() - self._asyncio_thread = threading.Thread(target = self._asyncio_loop.run_forever, name = 'asyncio') - self._asyncio_thread.setDaemon(True) - self._asyncio_thread.start() - self._msg_lock = threading.RLock() self._status_listeners = [] # type: List[Tuple[Callable[[stem.control.BaseController, stem.control.State, float], None], bool]] # tuples of the form (callback, spawn_thread) @@ -582,7 +578,7 @@ class BaseController(object): if is_authenticated: self._post_authentication() - def msg(self, message: str) -> stem.response.ControlMessage: + async def msg(self, message: str) -> stem.response.ControlMessage: """ Sends a message to our control socket and provides back its reply. @@ -648,7 +644,7 @@ class BaseController(object): try: self._asyncio_loop.create_task(self._socket.send(message)) - response = asyncio.run_coroutine_threadsafe(self._reply_queue.get(), self._asyncio_loop).result() + response = await self._reply_queue.get() # If the message we received back had an exception then re-raise it to the # caller. Otherwise return the response. @@ -711,7 +707,7 @@ class BaseController(object): return self._is_authenticated if self.is_alive() else False - def connect(self) -> None: + async def connect(self) -> None: """ Reconnects our control socket. This is a pass-through for our socket's :func:`~stem.socket.ControlSocket.connect` method. @@ -719,15 +715,15 @@ class BaseController(object): :raises: :class:`stem.SocketError` if unable to make a socket """ - asyncio.run_coroutine_threadsafe(self._socket.connect(), self._asyncio_loop).result() + await self._socket.connect() - def close(self) -> None: + async def close(self) -> None: """ Closes our socket connection. This is a pass-through for our socket's :func:`~stem.socket.BaseSocket.close` method. """ - asyncio.run_coroutine_threadsafe(self._socket.close(), self._asyncio_loop).result() + await self._socket.close() # Join on any outstanding state change listeners. Closing is a state change # of its own, so if we have any listeners it's quite likely there's some @@ -740,9 +736,6 @@ class BaseController(object): if t.is_alive() and threading.current_thread() != t: t.join() - self._asyncio_loop.call_soon_threadsafe(self._asyncio_loop.stop) - self._asyncio_thread.join() - def get_socket(self) -> stem.socket.ControlSocket: """ Provides the socket used to speak with the tor process. Communicating with @@ -815,11 +808,11 @@ class BaseController(object): self._status_listeners = new_listeners return is_changed - def __enter__(self) -> 'stem.control.BaseController': + async def __aenter__(self) -> 'stem.control.BaseController': return self - def __exit__(self, exit_type: Optional[Type[BaseException]], value: Optional[BaseException], traceback: Optional[TracebackType]) -> None: - self.close() + await def __aexit__(self, exit_type: Optional[Type[BaseException]], value: Optional[BaseException], traceback: Optional[TracebackType]) -> None: + await self.close() def _handle_event(self, event_message: stem.response.ControlMessage) -> None: """

1 0

Jump to page:

2025

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

tor-commits July 2020