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

[stem/master] Make `Controller.get_hidden_service_conf` asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit a651e8d16653fc9f0f0adb1a9bd4d6f35ba8a59a Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Mon Apr 20 22:17:07 2020 +0300 Make `Controller.get_hidden_service_conf` asynchronous --- stem/control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index 81f97581..443a16db 100644 --- a/stem/control.py +++ b/stem/control.py @@ -2473,7 +2473,7 @@ class Controller(BaseController): raise stem.ProtocolError('Returned unexpected status code: %s' % response.code) @with_default() - def get_hidden_service_conf(self, default: Any = UNDEFINED) -> Dict[str, Any]: + async def get_hidden_service_conf(self, default: Any = UNDEFINED) -> Dict[str, Any]: """ get_hidden_service_conf(default = UNDEFINED) @@ -2519,7 +2519,7 @@ class Controller(BaseController): start_time = time.time() try: - response = stem.response._convert_to_getconf(self.msg('GETCONF HiddenServiceOptions')) + response = stem.response._convert_to_getconf(await self.msg('GETCONF HiddenServiceOptions')) log.debug('GETCONF HiddenServiceOptions (runtime: %0.4f)' % (time.time() - start_time)) except stem.ControllerError as exc:

1 0

[stem/master] Make `Controller.get_hidden_service_descriptor` asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit a729b61c61f4c2e136f13a5a921bebfddf1cd4ed Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Fri Apr 17 23:07:47 2020 +0300 Make `Controller.get_hidden_service_descriptor` asynchronous --- stem/control.py | 34 ++++++++++++++++++++-------------- 1 file changed, 20 insertions(+), 14 deletions(-) diff --git a/stem/control.py b/stem/control.py index 84efbf81..fd38871a 100644 --- a/stem/control.py +++ b/stem/control.py @@ -2007,7 +2007,7 @@ class Controller(BaseController): yield desc # type: ignore @with_default() - 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: + async 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: """ get_hidden_service_descriptor(address, default = UNDEFINED, servers = None, await_result = True) @@ -2050,23 +2050,25 @@ class Controller(BaseController): if not stem.util.tor_tools.is_valid_hidden_service_address(address): raise ValueError("'%s.onion' isn't a valid hidden service address" % address) - hs_desc_queue = queue.Queue() # type: queue.Queue[stem.response.events.Event] + hs_desc_queue = asyncio.Queue() # type: asyncio.Queue[stem.response.events.Event] hs_desc_listener = None - hs_desc_content_queue = queue.Queue() # type: queue.Queue[stem.response.events.Event] + hs_desc_content_queue = asyncio.Queue() # type: asyncio.Queue[stem.response.events.Event] hs_desc_content_listener = None start_time = time.time() if await_result: - def hs_desc_listener(event: stem.response.events.Event) -> None: - hs_desc_queue.put(event) + async def hs_desc_listener(event: stem.response.events.Event) -> None: + await hs_desc_queue.put(event) - def hs_desc_content_listener(event: stem.response.events.Event) -> None: - hs_desc_content_queue.put(event) + async def hs_desc_content_listener(event: stem.response.events.Event) -> None: + await hs_desc_content_queue.put(event) - self.add_event_listener(hs_desc_listener, EventType.HS_DESC) - self.add_event_listener(hs_desc_content_listener, EventType.HS_DESC_CONTENT) + await asyncio.gather( + self.add_event_listener(hs_desc_listener, EventType.HS_DESC), + self.add_event_listener(hs_desc_content_listener, EventType.HS_DESC_CONTENT), + ) try: request = 'HSFETCH %s' % address @@ -2074,7 +2076,7 @@ class Controller(BaseController): if servers: request += ' ' + ' '.join(['SERVER=%s' % s for s in servers]) - response = stem.response._convert_to_single_line(self.msg(request)) + response = stem.response._convert_to_single_line(await self.msg(request)) if not response.is_ok(): raise stem.ProtocolError('HSFETCH returned unexpected response code: %s' % response.code) @@ -2083,7 +2085,7 @@ class Controller(BaseController): return None # not waiting, so nothing to provide back else: while True: - event = _get_with_timeout(hs_desc_content_queue, timeout, start_time) + event = await _get_with_timeout(hs_desc_content_queue, timeout) if event.address == address: if event.descriptor: @@ -2092,7 +2094,7 @@ class Controller(BaseController): # no descriptor, looking through HS_DESC to figure out why while True: - event = _get_with_timeout(hs_desc_queue, timeout, start_time) + event = await _get_with_timeout(hs_desc_queue, timeout) if event.address == address and event.action == stem.HSDescAction.FAILED: if event.reason == stem.HSDescReason.NOT_FOUND: @@ -2100,11 +2102,15 @@ class Controller(BaseController): else: raise stem.DescriptorUnavailable('Unable to retrieve the descriptor for %s.onion (retrieved from %s): %s' % (address, event.directory_fingerprint, event.reason)) finally: + awaitable_removals = [] + if hs_desc_listener: - self.remove_event_listener(hs_desc_listener) + awaitable_removals.append(self.remove_event_listener(hs_desc_listener)) if hs_desc_content_listener: - self.remove_event_listener(hs_desc_content_listener) + awaitable_removals.append(self.remove_event_listener(hs_desc_content_listener)) + + await asyncio.gather(*awaitable_removals) async def get_conf(self, param: str, default: Any = UNDEFINED, multiple: bool = False) -> Union[str, Sequence[str]]: """

1 0

[stem/master] Make `Controller.set_options` asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 4f63690766928018cec8b6dc0e3a4a9fe81dbd9d Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Mon Apr 20 22:14:15 2020 +0300 Make `Controller.set_options` asynchronous --- stem/control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index 708a413d..fda89224 100644 --- a/stem/control.py +++ b/stem/control.py @@ -2392,7 +2392,7 @@ class Controller(BaseController): self.set_options(dict([(entry, None) for entry in params]), True) - def set_options(self, params: Union[Mapping[str, Union[str, Sequence[str]]], Sequence[Tuple[str, Union[str, Sequence[str]]]]], reset: bool = False) -> None: + async def set_options(self, params: Union[Mapping[str, Union[str, Sequence[str]]], Sequence[Tuple[str, Union[str, Sequence[str]]]]], reset: bool = False) -> None: """ Changes multiple tor configuration options via either a SETCONF or RESETCONF query. Both behave identically unless our value is None, in which @@ -2445,7 +2445,7 @@ class Controller(BaseController): raise ValueError('Cannot set %s to %s since the value was a %s but we only accept strings' % (param, value, type(value).__name__)) query = ' '.join(query_comp) - response = stem.response._convert_to_single_line(self.msg(query)) + response = stem.response._convert_to_single_line(await self.msg(query)) if response.is_ok(): log.debug('%s (runtime: %0.4f)' % (query, time.time() - start_time))

1 0

[stem/master] Make `Controller.set_hidden_service_conf` asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 58b9817f8fbc6ae936dad7d2b106943533cd8b4a Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Mon Apr 20 22:19:03 2020 +0300 Make `Controller.set_hidden_service_conf` asynchronous --- stem/control.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/stem/control.py b/stem/control.py index 443a16db..67fb6a47 100644 --- a/stem/control.py +++ b/stem/control.py @@ -2567,7 +2567,7 @@ class Controller(BaseController): self._set_cache({'hidden_service_conf': service_dir_map}) return service_dir_map - def set_hidden_service_conf(self, conf: Mapping[str, Any]) -> None: + async def set_hidden_service_conf(self, conf: Mapping[str, Any]) -> None: """ Update all the configured hidden services from a dictionary having the same format as @@ -2605,7 +2605,7 @@ class Controller(BaseController): # so we drop existing values. Otherwise calling SETCONF is a no-op. if not conf: - self.reset_conf('HiddenServiceDir') + await self.reset_conf('HiddenServiceDir') return # Convert conf dictionary into a list of ordered config tuples @@ -2630,7 +2630,7 @@ class Controller(BaseController): else: hidden_service_options.append((k, str(v))) - self.set_options(hidden_service_options) + await self.set_options(hidden_service_options) 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': """

1 0

[stem/master] Make `new_circuit` and `extend_circuit` of `Controller` asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit e571b46ce0ca2ab76a40021e5eab65317ea041f5 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 21 22:05:03 2020 +0300 Make `new_circuit` and `extend_circuit` of `Controller` asynchronous --- stem/control.py | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/stem/control.py b/stem/control.py index 4f56bb95..e10aa202 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3400,7 +3400,7 @@ class Controller(BaseController): return circuits - def new_circuit(self, path: Union[None, str, Sequence[str]] = None, purpose: str = 'general', await_build: bool = False, timeout: Optional[float] = None) -> str: + async def new_circuit(self, path: Union[None, str, Sequence[str]] = None, purpose: str = 'general', await_build: bool = False, timeout: Optional[float] = None) -> str: """ Requests a new circuit. If the path isn't provided, one is automatically selected. @@ -3420,9 +3420,9 @@ class Controller(BaseController): * :class:`stem.Timeout` if **timeout** was reached """ - return self.extend_circuit('0', path, purpose, await_build, timeout) + return await self.extend_circuit('0', path, purpose, await_build, timeout) - 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: + async 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: """ Either requests the creation of a new circuit or extends an existing one. @@ -3466,15 +3466,15 @@ class Controller(BaseController): # to build. This is icky, but we can't reliably do this via polling since # we then can't get the failure if it can't be created. - circ_queue = queue.Queue() # type: queue.Queue[stem.response.events.Event] + circ_queue = asyncio.Queue() # type: asyncio.Queue[stem.response.events.Event] circ_listener = None start_time = time.time() if await_build: - def circ_listener(event: stem.response.events.Event) -> None: - circ_queue.put(event) + async def circ_listener(event: stem.response.events.Event) -> None: + await circ_queue.put(event) - self.add_event_listener(circ_listener, EventType.CIRC) + await self.add_event_listener(circ_listener, EventType.CIRC) try: args = [str(circuit_id)] @@ -3488,7 +3488,7 @@ class Controller(BaseController): if purpose: args.append('purpose=%s' % purpose) - response = stem.response._convert_to_single_line(self.msg('EXTENDCIRCUIT %s' % ' '.join(args))) + response = stem.response._convert_to_single_line(await self.msg('EXTENDCIRCUIT %s' % ' '.join(args))) if response.code in ('512', '552'): raise stem.InvalidRequest(response.code, response.message) @@ -3502,7 +3502,7 @@ class Controller(BaseController): if await_build: while True: - circ = _get_with_timeout(circ_queue, timeout, start_time) + circ = await _get_with_timeout(circ_queue, timeout, start_time) if circ.id == new_circuit: if circ.status == CircStatus.BUILT: @@ -3515,7 +3515,7 @@ class Controller(BaseController): return new_circuit finally: if circ_listener: - self.remove_event_listener(circ_listener) + await self.remove_event_listener(circ_listener) def repurpose_circuit(self, circuit_id: str, purpose: str) -> None: """

1 0

[stem/master] Make `get_circuit` and `get_circuits` of `Controller` asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 3ea6be9ed59fdd28fed568676f5624f7ffa7082c Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 21 21:48:40 2020 +0300 Make `get_circuit` and `get_circuits` of `Controller` asynchronous --- stem/control.py | 10 +++++----- stem/socket.py | 4 +++- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/stem/control.py b/stem/control.py index d0cbb34a..4f56bb95 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3353,7 +3353,7 @@ class Controller(BaseController): self._enabled_features += [entry.upper() for entry in features] @with_default() - def get_circuit(self, circuit_id: int, default: Any = UNDEFINED) -> stem.response.events.CircuitEvent: + async def get_circuit(self, circuit_id: int, default: Any = UNDEFINED) -> stem.response.events.CircuitEvent: """ get_circuit(circuit_id, default = UNDEFINED) @@ -3371,14 +3371,14 @@ class Controller(BaseController): An exception is only raised if we weren't provided a default response. """ - for circ in self.get_circuits(): + for circ in await self.get_circuits(): if circ.id == circuit_id: return circ raise ValueError("Tor currently does not have a circuit with the id of '%s'" % circuit_id) @with_default() - def get_circuits(self, default: Any = UNDEFINED) -> List[stem.response.events.CircuitEvent]: + async def get_circuits(self, default: Any = UNDEFINED) -> List[stem.response.events.CircuitEvent]: """ get_circuits(default = UNDEFINED) @@ -3392,10 +3392,10 @@ class Controller(BaseController): """ circuits = [] # type: List[stem.response.events.CircuitEvent] - response = self.get_info('circuit-status') + response = await self.get_info('circuit-status') for circ in response.splitlines(): - circ_message = stem.response._convert_to_event(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(await stem.socket.recv_message(io.BytesIO(stem.util.str_tools._to_bytes('650 CIRC %s\r\n' % circ)))) circuits.append(circ_message) # type: ignore return circuits diff --git a/stem/socket.py b/stem/socket.py index dd123751..a82c4892 100644 --- a/stem/socket.py +++ b/stem/socket.py @@ -588,7 +588,9 @@ async def recv_message(reader: asyncio.StreamReader, arrived_at: Optional[float] while True: try: - line = await reader.readline() + line = reader.readline() + if asyncio.iscoroutine(line): + line = await line except AttributeError: # if the control_file has been closed then we will receive: # AttributeError: 'NoneType' object has no attribute 'recv'

1 0

[stem/master] Make `Controller.set_conf` and `Controller.reset_conf` asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 40bbdb09a33f181ffadcdd5af675e5c1a11156be Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Mon Apr 20 22:14:58 2020 +0300 Make `Controller.set_conf` and `Controller.reset_conf` asynchronous --- stem/control.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/stem/control.py b/stem/control.py index fda89224..81f97581 100644 --- a/stem/control.py +++ b/stem/control.py @@ -2355,7 +2355,7 @@ class Controller(BaseController): return result - def set_conf(self, param: str, value: Union[str, Sequence[str]]) -> None: + async def set_conf(self, param: str, value: Union[str, Sequence[str]]) -> None: """ Changes the value of a tor configuration option. Our value can be any of the following... @@ -2375,9 +2375,9 @@ class Controller(BaseController): impossible or if there's a syntax error in the configuration values """ - self.set_options({param: value}, False) + await self.set_options({param: value}, False) - def reset_conf(self, *params: str) -> None: + async def reset_conf(self, *params: str) -> None: """ Reverts one or more parameters to their default values. @@ -2390,7 +2390,7 @@ class Controller(BaseController): impossible or if there's a syntax error in the configuration values """ - self.set_options(dict([(entry, None) for entry in params]), True) + await self.set_options(dict([(entry, None) for entry in params]), True) async def set_options(self, params: Union[Mapping[str, Union[str, Sequence[str]]], Sequence[Tuple[str, Union[str, Sequence[str]]]]], reset: bool = False) -> None: """

1 0

[stem/master] Make `Controller.repurpose_circuit` asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 2ae55697420fe1ab4611c285b45f6a167505843a Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 21 22:08:08 2020 +0300 Make `Controller.repurpose_circuit` asynchronous --- stem/control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index e10aa202..99e74f25 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3517,7 +3517,7 @@ class Controller(BaseController): if circ_listener: await self.remove_event_listener(circ_listener) - def repurpose_circuit(self, circuit_id: str, purpose: str) -> None: + async def repurpose_circuit(self, circuit_id: str, purpose: str) -> None: """ Changes a circuit's purpose. Currently, two purposes are recognized... * general @@ -3529,7 +3529,7 @@ class Controller(BaseController): :raises: :class:`stem.InvalidArguments` if the circuit doesn't exist or if the purpose was invalid """ - response = stem.response._convert_to_single_line(self.msg('SETCIRCUITPURPOSE %s purpose=%s' % (circuit_id, purpose))) + response = stem.response._convert_to_single_line(await self.msg('SETCIRCUITPURPOSE %s purpose=%s' % (circuit_id, purpose))) if not response.is_ok(): if response.code == '552':

1 0

[stem/master] Await a coroutine
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 241083b3f05a04f46cdba96e40c8e990bd63e40f Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 21 22:19:12 2020 +0300 Await a coroutine --- stem/control.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/stem/control.py b/stem/control.py index 370b6270..7cfd3197 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3802,7 +3802,7 @@ class Controller(BaseController): # connection, we can stop having it check for us via our pid. try: - self.reset_conf('__OwningControllerProcess') + await self.reset_conf('__OwningControllerProcess') except stem.ControllerError as exc: log.warn("We were unable to reset tor's __OwningControllerProcess configuration. It will continue to periodically check if our pid exists. (%s)" % exc) else:

1 0

[stem/master] Make `_get_with_timeout` asynchronous
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit e43c446d991611c421dbae68f7c46fb922589e0d Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Fri Apr 17 23:06:52 2020 +0300 Make `_get_with_timeout` asynchronous --- stem/control.py | 18 +++++------------- 1 file changed, 5 insertions(+), 13 deletions(-) diff --git a/stem/control.py b/stem/control.py index 1a107a79..84efbf81 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3981,20 +3981,12 @@ def _case_insensitive_lookup(entries: Union[Sequence[str], Mapping[str, Any]], k raise ValueError("key '%s' doesn't exist in dict: %s" % (key, entries)) -def _get_with_timeout(event_queue: queue.Queue, timeout: float, start_time: float) -> Any: +async def _get_with_timeout(event_queue: queue.Queue, timeout: float, start_time: float) -> Any: """ Pulls an item from a queue with a given timeout. """ - if timeout: - time_left = timeout - (time.time() - start_time) - - if time_left <= 0: - raise stem.Timeout('Reached our %0.1f second timeout' % timeout) - - try: - return event_queue.get(True, time_left) - except asyncio.queues.QueueEmpty: - raise stem.Timeout('Reached our %0.1f second timeout' % timeout) - else: - return event_queue.get() + try: + return await asyncio.wait_for(event_queue.get(), timeout=timeout) + except asyncio.TimeoutError: + raise stem.Timeout('Reached our %0.1f second timeout' % timeout)

1 0