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

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

16 Jul '20

commit 414c6771b7fe1550fdf4326f80446e66b0c9ebbe Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 21 22:09:05 2020 +0300 Make `Controller.close_circuit` asynchronous --- stem/control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index 99e74f25..1d4ee689 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3537,7 +3537,7 @@ class Controller(BaseController): else: raise stem.ProtocolError('SETCIRCUITPURPOSE returned unexpected response code: %s' % response.code) - def close_circuit(self, circuit_id: str, flag: str = '') -> None: + async def close_circuit(self, circuit_id: str, flag: str = '') -> None: """ Closes the specified circuit. @@ -3550,7 +3550,7 @@ class Controller(BaseController): * :class:`stem.InvalidRequest` if not enough information is provided """ - response = stem.response._convert_to_single_line(self.msg('CLOSECIRCUIT %s %s' % (circuit_id, flag))) + response = stem.response._convert_to_single_line(await self.msg('CLOSECIRCUIT %s %s' % (circuit_id, flag))) if not response.is_ok(): if response.code in ('512', '552'):

1 0

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

16 Jul '20

commit 79d07c69d41dacc503ff17bbc87fd5be66750431 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 21 22:10:52 2020 +0300 Make `Controller.get_streams` asynchronous --- stem/control.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/stem/control.py b/stem/control.py index 1d4ee689..f63df7bc 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3561,7 +3561,7 @@ class Controller(BaseController): raise stem.ProtocolError('CLOSECIRCUIT returned unexpected response code: %s' % response.code) @with_default() - def get_streams(self, default: Any = UNDEFINED) -> List[stem.response.events.StreamEvent]: + async def get_streams(self, default: Any = UNDEFINED) -> List[stem.response.events.StreamEvent]: """ get_streams(default = UNDEFINED) @@ -3576,10 +3576,10 @@ class Controller(BaseController): """ streams = [] # type: List[stem.response.events.StreamEvent] - response = self.get_info('stream-status') + response = await self.get_info('stream-status') for stream in response.splitlines(): - message = stem.response._convert_to_event(stem.socket.recv_message(io.BytesIO(stem.util.str_tools._to_bytes('650 STREAM %s\r\n' % stream)))) + 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)))) streams.append(message) # type: ignore return streams

1 0

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

16 Jul '20

commit bee112f44a4fc6a9f8f1f95b78aa621e1299038e Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Mon Apr 20 22:07:17 2020 +0300 Make `Controller.is_set` asynchronous --- stem/control.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/stem/control.py b/stem/control.py index fd38871a..708a413d 100644 --- a/stem/control.py +++ b/stem/control.py @@ -2308,7 +2308,7 @@ class Controller(BaseController): return return_dict @with_default() - def is_set(self, param: str, default: Any = UNDEFINED) -> bool: + async def is_set(self, param: str, default: Any = UNDEFINED) -> bool: """ is_set(param, default = UNDEFINED) @@ -2326,13 +2326,13 @@ class Controller(BaseController): provided a default response """ - return param in self._get_custom_options() + return param in await self._get_custom_options() - def _get_custom_options(self) -> Dict[str, str]: + async def _get_custom_options(self) -> Dict[str, str]: result = self._get_cache('get_custom_options') if not result: - config_lines = self.get_info('config-text').splitlines() + config_lines = (await self.get_info('config-text')).splitlines() # Tor provides some config options even if they haven't been set... # @@ -2342,7 +2342,7 @@ class Controller(BaseController): default_lines = ( 'Log notice stdout', 'Log notice file /var/log/tor/log', - 'DataDirectory /home/%s/.tor' % self.get_user('undefined'), + 'DataDirectory /home/%s/.tor' % await self.get_user('undefined'), 'HiddenServiceStatistics 0', )

1 0

[stem/master] Return the previous behavior of `_get_with_timeout`
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit 3431c5cba5ead275f359d2785cd5cc252e5b495d Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 21 21:44:13 2020 +0300 Return the previous behavior of `_get_with_timeout` --- stem/control.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/stem/control.py b/stem/control.py index 67fb6a47..d0cbb34a 100644 --- a/stem/control.py +++ b/stem/control.py @@ -2085,7 +2085,7 @@ class Controller(BaseController): return None # not waiting, so nothing to provide back else: while True: - event = await _get_with_timeout(hs_desc_content_queue, timeout) + event = await _get_with_timeout(hs_desc_content_queue, timeout, start_time) if event.address == address: if event.descriptor: @@ -2094,7 +2094,7 @@ class Controller(BaseController): # no descriptor, looking through HS_DESC to figure out why while True: - event = await _get_with_timeout(hs_desc_queue, timeout) + event = await _get_with_timeout(hs_desc_queue, timeout, start_time) if event.address == address and event.action == stem.HSDescAction.FAILED: if event.reason == stem.HSDescReason.NOT_FOUND: @@ -3992,7 +3992,12 @@ async def _get_with_timeout(event_queue: queue.Queue, timeout: float, start_time Pulls an item from a queue with a given timeout. """ + if timeout: + time_left = timeout - (time.time() - start_time) + else: + time_left = None + try: - return await asyncio.wait_for(event_queue.get(), timeout=timeout) + return await asyncio.wait_for(event_queue.get(), timeout=time_left) except asyncio.TimeoutError: raise stem.Timeout('Reached our %0.1f second timeout' % timeout)

1 0

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

16 Jul '20

commit da8fbf7079a4573b7061b336025acd3edc8f340c Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 21 22:16:00 2020 +0300 Make `Controller.signal` asynchronous --- stem/control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index f63df7bc..ee6f7aa3 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3646,7 +3646,7 @@ class Controller(BaseController): else: raise stem.ProtocolError('CLOSESTREAM returned unexpected response code: %s' % response.code) - def signal(self, signal: stem.Signal) -> None: + async def signal(self, signal: stem.Signal) -> None: """ Sends a signal to the Tor client. @@ -3657,7 +3657,7 @@ class Controller(BaseController): * :class:`stem.InvalidArguments` if signal provided wasn't recognized """ - response = stem.response._convert_to_single_line(self.msg('SIGNAL %s' % signal)) + response = stem.response._convert_to_single_line(await self.msg('SIGNAL %s' % signal)) if response.is_ok(): if signal == stem.Signal.NEWNYM:

1 0

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

16 Jul '20

commit 81f509ceecce78fa09488c1da52eb4480de25e66 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 21 22:23:09 2020 +0300 Make `Controller.attach_stream` asynchronous --- stem/control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index e0248d76..8eaa8bd8 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3584,7 +3584,7 @@ class Controller(BaseController): return streams - def attach_stream(self, stream_id: str, circuit_id: str, exiting_hop: Optional[int] = None) -> None: + async def attach_stream(self, stream_id: str, circuit_id: str, exiting_hop: Optional[int] = None) -> None: """ Attaches a stream to a circuit. @@ -3606,7 +3606,7 @@ class Controller(BaseController): if exiting_hop: query += ' HOP=%s' % exiting_hop - response = stem.response._convert_to_single_line(self.msg(query)) + response = stem.response._convert_to_single_line(await self.msg(query)) if not response.is_ok(): if response.code == '552':

1 0

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

16 Jul '20

commit f36db8793d671a3658155b146eb7374ee81cd770 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 21 22:17:36 2020 +0300 Make `Controller.get_effective_rate` asynchronous --- stem/control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index ee6f7aa3..370b6270 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3698,7 +3698,7 @@ class Controller(BaseController): return max(0.0, self._last_newnym + 10 - time.time()) @with_default() - def get_effective_rate(self, default: Any = UNDEFINED, burst: bool = False) -> int: + async def get_effective_rate(self, default: Any = UNDEFINED, burst: bool = False) -> int: """ get_effective_rate(default = UNDEFINED, burst = False) @@ -3728,7 +3728,7 @@ class Controller(BaseController): value = None for attr in attributes: - attr_value = int(self._get_conf_single(attr)) + attr_value = int(await self._get_conf_single(attr)) if attr_value == 0 and attr.startswith('Relay'): continue # RelayBandwidthRate and RelayBandwidthBurst default to zero

1 0

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

16 Jul '20

commit c7124d2727066f62c2dede31b4f36b391ec88fe7 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 21 22:20:52 2020 +0300 Make `Controller.map_address` asynchronous --- stem/control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index 7cfd3197..66c8b567 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3737,7 +3737,7 @@ class Controller(BaseController): return value - def map_address(self, mapping: Mapping[str, str]) -> Dict[str, str]: + async def map_address(self, mapping: Mapping[str, str]) -> Dict[str, str]: """ Map addresses to replacement addresses. Tor replaces subseqent connections to the original addresses with the replacement addresses. @@ -3757,7 +3757,7 @@ class Controller(BaseController): """ mapaddress_arg = ' '.join(['%s=%s' % (k, v) for (k, v) in list(mapping.items())]) - response = self.msg('MAPADDRESS %s' % mapaddress_arg) + response = await self.msg('MAPADDRESS %s' % mapaddress_arg) return stem.response._convert_to_mapaddress(response).entries def drop_guards(self) -> None:

1 0

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

16 Jul '20

commit 397a94f591dcb40a0545a0346d23e8d68eea2176 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Tue Apr 21 22:21:23 2020 +0300 Make `Controller.drop_guards` asynchronous --- stem/control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stem/control.py b/stem/control.py index 66c8b567..e0248d76 100644 --- a/stem/control.py +++ b/stem/control.py @@ -3760,7 +3760,7 @@ class Controller(BaseController): response = await self.msg('MAPADDRESS %s' % mapaddress_arg) return stem.response._convert_to_mapaddress(response).entries - def drop_guards(self) -> None: + await def drop_guards(self) -> None: """ Drops our present guard nodes and picks a new set. @@ -3769,7 +3769,7 @@ class Controller(BaseController): :raises: :class:`stem.ControllerError` if Tor couldn't fulfill the request """ - self.msg('DROPGUARDS') + await self.msg('DROPGUARDS') async def _post_authentication(self) -> None: await super(Controller, self)._post_authentication()

1 0

[stem/master] Copy "stem/socket.py" to "stem/async_socket.py"
by atagar＠torproject.org 16 Jul '20

16 Jul '20

commit bba3029f8d710b26cc0299ce086d31b28104c897 Author: Illia Volochii <illia.volochii(a)gmail.com> Date: Sat Mar 14 21:57:20 2020 +0200 Copy "stem/socket.py" to "stem/async_socket.py" --- stem/async_socket.py | 768 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 768 insertions(+) diff --git a/stem/async_socket.py b/stem/async_socket.py new file mode 100644 index 00000000..2ef42dd5 --- /dev/null +++ b/stem/async_socket.py @@ -0,0 +1,768 @@ +# Copyright 2011-2020, Damian Johnson and The Tor Project +# See LICENSE for licensing information + +""" +Supports communication with sockets speaking Tor protocols. This +allows us to send messages as basic strings, and receive responses as +:class:`~stem.response.ControlMessage` instances. + +**This module only consists of low level components, and is not intended for +users.** See our `tutorials <../tutorials.html>`_ and `Control Module +<control.html>`_ if you're new to Stem and looking to get started. + +With that aside, these can still be used for raw socket communication with +Tor... + +:: + + import stem + import stem.connection + import stem.socket + + if __name__ == '__main__': + try: + control_socket = stem.socket.ControlPort(port = 9051) + stem.connection.authenticate(control_socket) + except stem.SocketError as exc: + print 'Unable to connect to tor on port 9051: %s' % exc + sys.exit(1) + except stem.connection.AuthenticationFailure as exc: + print 'Unable to authenticate: %s' % exc + sys.exit(1) + + print "Issuing 'GETINFO version' query...\\n" + control_socket.send('GETINFO version') + print control_socket.recv() + +:: + + % python example.py + Issuing 'GETINFO version' query... + + version=0.2.4.10-alpha-dev (git-8be6058d8f31e578) + OK + +**Module Overview:** + +:: + + BaseSocket - Thread safe socket. + |- RelaySocket - Socket for a relay's ORPort. + | |- send - sends a message to the socket + | +- recv - receives a response from the socket + | + |- ControlSocket - Socket wrapper that speaks the tor control protocol. + | |- ControlPort - Control connection via a port. + | |- ControlSocketFile - Control connection via a local file socket. + | | + | |- send - sends a message to the socket + | +- recv - receives a ControlMessage from the socket + | + |- is_alive - reports if the socket is known to be closed + |- is_localhost - returns if the socket is for the local system or not + |- connection_time - timestamp when socket last connected or disconnected + |- connect - connects a new socket + |- close - shuts down the socket + +- __enter__ / __exit__ - manages socket connection + + send_message - Writes a message to a control socket. + recv_message - Reads a ControlMessage from a control socket. + send_formatting - Performs the formatting expected from sent messages. +""" + +from __future__ import absolute_import + +import re +import socket +import ssl +import threading +import time + +import stem.response +import stem.util.str_tools + +from stem.util import log + +MESSAGE_PREFIX = re.compile(b'^[a-zA-Z0-9]{3}[-+ ]') +ERROR_MSG = 'Error while receiving a control message (%s): %s' + +# lines to limit our trace logging to, you can disable this by setting it to None + +TRUNCATE_LOGS = 10 + + +class BaseSocket(object): + """ + Thread safe socket, providing common socket functionality. + """ + + def __init__(self): + self._socket, self._socket_file = None, None + self._is_alive = False + self._connection_time = 0.0 # time when we last connected or disconnected + + # Tracks sending and receiving separately. This should be safe, and doing + # so prevents deadlock where we block writes because we're waiting to read + # a message that isn't coming. + + self._send_lock = threading.RLock() + self._recv_lock = threading.RLock() + + def is_alive(self): + """ + Checks if the socket is known to be closed. We won't be aware if it is + until we either use it or have explicitily shut it down. + + In practice a socket derived from a port knows about its disconnection + after failing to receive data, whereas socket file derived connections + know after either sending or receiving data. + + This means that to have reliable detection for when we're disconnected + you need to continually pull from the socket (which is part of what the + :class:`~stem.control.BaseController` does). + + :returns: **bool** that's **True** if our socket is connected and **False** + otherwise + """ + + return self._is_alive + + def is_localhost(self): + """ + Returns if the connection is for the local system or not. + + :returns: **bool** that's **True** if the connection is for the local host + and **False** otherwise + """ + + return False + + def connection_time(self): + """ + Provides the unix timestamp for when our socket was either connected or + disconnected. That is to say, the time we connected if we're currently + connected and the time we disconnected if we're not connected. + + .. versionadded:: 1.3.0 + + :returns: **float** for when we last connected or disconnected, zero if + we've never connected + """ + + return self._connection_time + + def connect(self): + """ + Connects to a new socket, closing our previous one if we're already + attached. + + :raises: :class:`stem.SocketError` if unable to make a socket + """ + + with self._send_lock: + # Closes the socket if we're currently attached to one. Once we're no + # longer alive it'll be safe to acquire the recv lock because recv() + # calls no longer block (raising SocketClosed instead). + + if self.is_alive(): + self.close() + + with self._recv_lock: + self._socket = self._make_socket() + self._socket_file = self._socket.makefile(mode = 'rwb') + self._is_alive = True + self._connection_time = time.time() + + # It's possible for this to have a transient failure... + # SocketError: [Errno 4] Interrupted system call + # + # It's safe to retry, so give it another try if it fails. + + try: + self._connect() + except stem.SocketError: + self._connect() # single retry + + def close(self): + """ + Shuts down the socket. If it's already closed then this is a no-op. + """ + + with self._send_lock: + # Function is idempotent with one exception: we notify _close() if this + # is causing our is_alive() state to change. + + is_change = self.is_alive() + + if self._socket: + # if we haven't yet established a connection then this raises an error + # socket.error: [Errno 107] Transport endpoint is not connected + + try: + self._socket.shutdown(socket.SHUT_RDWR) + except socket.error: + pass + + self._socket.close() + + if self._socket_file: + try: + self._socket_file.close() + except BrokenPipeError: + pass + + self._socket = None + self._socket_file = None + self._is_alive = False + self._connection_time = time.time() + + if is_change: + self._close() + + def _send(self, message, handler): + """ + Send message in a thread safe manner. Handler is expected to be of the form... + + :: + + my_handler(socket, socket_file, message) + """ + + with self._send_lock: + try: + if not self.is_alive(): + raise stem.SocketClosed() + + handler(self._socket, self._socket_file, message) + except stem.SocketClosed: + # if send_message raises a SocketClosed then we should properly shut + # everything down + + if self.is_alive(): + self.close() + + raise + + 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) + """ + + with self._recv_lock: + try: + # makes a temporary reference to the _socket_file because connect() + # and close() may set or unset it + + my_socket, my_socket_file = self._socket, self._socket_file + + if not my_socket or not my_socket_file: + raise stem.SocketClosed() + + return handler(my_socket, my_socket_file) + except stem.SocketClosed: + # If recv_message raises a SocketClosed then we should properly shut + # everything down. However, there's a couple cases where this will + # cause deadlock... + # + # * This SocketClosed was *caused by* a close() call, which is joining + # on our thread. + # + # * A send() call that's currently in flight is about to call close(), + # also attempting to join on us. + # + # To resolve this we make a non-blocking call to acquire the send lock. + # If we get it then great, we can close safely. If not then one of the + # above are in progress and we leave the close to them. + + if self.is_alive(): + if self._send_lock.acquire(False): + self.close() + self._send_lock.release() + + raise + + def _get_send_lock(self): + """ + The send lock is useful to classes that interact with us at a deep level + because it's used to lock :func:`stem.socket.ControlSocket.connect` / + :func:`stem.socket.BaseSocket.close`, and by extension our + :func:`stem.socket.BaseSocket.is_alive` state changes. + + :returns: **threading.RLock** that governs sending messages to our socket + and state changes + """ + + return self._send_lock + + def __enter__(self): + return self + + def __exit__(self, exit_type, value, traceback): + self.close() + + def _connect(self): + """ + Connection callback that can be overwritten by subclasses and wrappers. + """ + + pass + + def _close(self): + """ + Disconnection callback that can be overwritten by subclasses and wrappers. + """ + + pass + + def _make_socket(self): + """ + Constructs and connects new socket. This is implemented by subclasses. + + :returns: **socket.socket** for our configuration + + :raises: + * :class:`stem.SocketError` if unable to make a socket + * **NotImplementedError** if not implemented by a subclass + """ + + raise NotImplementedError('Unsupported Operation: this should be implemented by the BaseSocket subclass') + + +class RelaySocket(BaseSocket): + """ + `Link-level connection + <https://gitweb.torproject.org/torspec.git/tree/tor-spec.txt>`_ to a Tor + relay. + + .. versionadded:: 1.7.0 + + :var str address: address our socket connects to + :var int port: ORPort our socket connects to + """ + + def __init__(self, address = '127.0.0.1', port = 9050, connect = True): + """ + RelaySocket constructor. + + :param str address: ip address of the relay + :param int port: orport of the relay + :param bool connect: connects to the socket if True, leaves it unconnected otherwise + + :raises: :class:`stem.SocketError` if connect is **True** and we're + unable to establish a connection + """ + + super(RelaySocket, self).__init__() + self.address = address + self.port = port + + if connect: + self.connect() + + def send(self, message): + """ + Sends a message to the relay's ORPort. + + :param str message: message to be formatted and sent to the socket + + :raises: + * :class:`stem.SocketError` if a problem arises in using the socket + * :class:`stem.SocketClosed` if the socket is known to be shut down + """ + + self._send(message, lambda s, sf, msg: _write_to_socket(sf, msg)) + + def recv(self, timeout = None): + """ + Receives a message from the relay. + + :param float timeout: maxiumum number of seconds to await a response, this + blocks indefinitely if **None** + + :returns: bytes for the message received + + :raises: + * :class:`stem.ProtocolError` the content from the socket is malformed + * :class:`stem.SocketClosed` if the socket closes before we receive a complete message + """ + + def wrapped_recv(s, sf): + if timeout is None: + return s.recv() + else: + s.setblocking(0) + s.settimeout(timeout) + + try: + return s.recv() + except (socket.timeout, ssl.SSLError, ssl.SSLWantReadError): + return None + finally: + s.setblocking(1) + + return self._recv(wrapped_recv) + + def is_localhost(self): + return self.address == '127.0.0.1' + + def _make_socket(self): + try: + relay_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + relay_socket.connect((self.address, self.port)) + return ssl.wrap_socket(relay_socket) + except socket.error as exc: + raise stem.SocketError(exc) + + +class ControlSocket(BaseSocket): + """ + Wrapper for a socket connection that speaks the Tor control protocol. To the + better part this transparently handles the formatting for sending and + receiving complete messages. + + Callers should not instantiate this class directly, but rather use subclasses + which are expected to implement the **_make_socket()** method. + """ + + def __init__(self): + super(ControlSocket, self).__init__() + + def send(self, message): + """ + Formats and sends a message to the control socket. For more information see + the :func:`~stem.socket.send_message` function. + + :param str message: message to be formatted and sent to the socket + + :raises: + * :class:`stem.SocketError` if a problem arises in using the socket + * :class:`stem.SocketClosed` if the socket is known to be shut down + """ + + self._send(message, lambda s, sf, msg: send_message(sf, msg)) + + def recv(self): + """ + Receives a message from the control socket, blocking until we've received + one. For more information see the :func:`~stem.socket.recv_message` function. + + :returns: :class:`~stem.response.ControlMessage` for the message received + + :raises: + * :class:`stem.ProtocolError` the content from the socket is malformed + * :class:`stem.SocketClosed` if the socket closes before we receive a complete message + """ + + return self._recv(lambda s, sf: recv_message(sf)) + + +class ControlPort(ControlSocket): + """ + Control connection to tor. For more information see tor's ControlPort torrc + option. + + :var str address: address our socket connects to + :var int port: ControlPort our socket connects to + """ + + def __init__(self, address = '127.0.0.1', port = 9051, connect = True): + """ + ControlPort constructor. + + :param str address: ip address of the controller + :param int port: port number of the controller + :param bool connect: connects to the socket if True, leaves it unconnected otherwise + + :raises: :class:`stem.SocketError` if connect is **True** and we're + unable to establish a connection + """ + + super(ControlPort, self).__init__() + self.address = address + self.port = port + + if connect: + self.connect() + + def is_localhost(self): + return self.address == '127.0.0.1' + + def _make_socket(self): + try: + control_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + control_socket.connect((self.address, self.port)) + return control_socket + except socket.error as exc: + raise stem.SocketError(exc) + + +class ControlSocketFile(ControlSocket): + """ + Control connection to tor. For more information see tor's ControlSocket torrc + option. + + :var str path: filesystem path of the socket we connect to + """ + + def __init__(self, path = '/var/run/tor/control', connect = True): + """ + ControlSocketFile constructor. + + :param str socket_path: path where the control socket is located + :param bool connect: connects to the socket if True, leaves it unconnected otherwise + + :raises: :class:`stem.SocketError` if connect is **True** and we're + unable to establish a connection + """ + + super(ControlSocketFile, self).__init__() + self.path = path + + if connect: + self.connect() + + def is_localhost(self): + return True + + def _make_socket(self): + try: + control_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + control_socket.connect(self.path) + return control_socket + except socket.error as exc: + raise stem.SocketError(exc) + + +def send_message(control_file, message, raw = False): + """ + Sends a message to the control socket, adding the expected formatting for + single verses multi-line messages. Neither message type should contain an + ending newline (if so it'll be treated as a multi-line message with a blank + line at the end). If the message doesn't contain a newline then it's sent + as... + + :: + + <message>\\r\\n + + and if it does contain newlines then it's split on ``\\n`` and sent as... + + :: + + +<line 1>\\r\\n + <line 2>\\r\\n + <line 3>\\r\\n + .\\r\\n + + :param file control_file: file derived from the control socket (see the + socket's makefile() method for more information) + :param str message: message to be sent on the control socket + :param bool raw: leaves the message formatting untouched, passing it to the + socket as-is + + :raises: + * :class:`stem.SocketError` if a problem arises in using the socket + * :class:`stem.SocketClosed` if the socket is known to be shut down + """ + + if not raw: + message = send_formatting(message) + + _write_to_socket(control_file, message) + + if log.is_tracing(): + log_message = message.replace('\r\n', '\n').rstrip() + msg_div = '\n' if '\n' in log_message else ' ' + log.trace('Sent to tor:%s%s' % (msg_div, log_message)) + + +def _write_to_socket(socket_file, message): + try: + socket_file.write(stem.util.str_tools._to_bytes(message)) + socket_file.flush() + except socket.error as exc: + log.info('Failed to send: %s' % exc) + + # When sending there doesn't seem to be a reliable method for + # distinguishing between failures from a disconnect verses other things. + # Just accounting for known disconnection responses. + + if str(exc) == '[Errno 32] Broken pipe': + raise stem.SocketClosed(exc) + else: + raise stem.SocketError(exc) + except AttributeError: + # if the control_file has been closed then flush will receive: + # AttributeError: 'NoneType' object has no attribute 'sendall' + + log.info('Failed to send: file has been closed') + raise stem.SocketClosed('file has been closed') + + +def recv_message(control_file, arrived_at = None): + """ + Pulls from a control socket 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) + + :returns: :class:`~stem.response.ControlMessage` read from the socket + + :raises: + * :class:`stem.ProtocolError` the content from the socket is malformed + * :class:`stem.SocketClosed` if the socket closes before we receive + a complete message + """ + + parsed_content, raw_content, first_line = None, None, True + + while True: + try: + line = control_file.readline() + except AttributeError: + # if the control_file has been closed then we will receive: + # AttributeError: 'NoneType' object has no attribute 'recv' + + log.info(ERROR_MSG % ('SocketClosed', 'socket file has been closed')) + raise stem.SocketClosed('socket file has been closed') + except (OSError, ValueError) as exc: + # when disconnected this errors with... + # + # * ValueError: I/O operation on closed file + # * OSError: [Errno 107] Transport endpoint is not connected + # * OSError: [Errno 9] Bad file descriptor + + log.info(ERROR_MSG % ('SocketClosed', 'received exception "%s"' % exc)) + raise stem.SocketClosed(exc) + + # Parses the tor control lines. These are of the form... + # <status code><divider><content>\r\n + + if not line: + # if the socket is disconnected then the readline() method will provide + # empty content + + 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))) + 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))) + 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 + + status_code = stem.util.str_tools._to_unicode(status_code) + divider = stem.util.str_tools._to_unicode(divider) + + # Most controller responses are single lines, in which case we don't need + # so much overhead. + + if first_line: + if divider == ' ': + _log_trace(line) + return stem.response.ControlMessage([(status_code, divider, content)], line, arrived_at = arrived_at) + else: + parsed_content, raw_content, first_line = [], bytearray(), False + + raw_content += line + + if divider == '-': + # mid-reply line, keep pulling for more content + parsed_content.append((status_code, divider, content)) + elif divider == ' ': + # end of the message, return the message + parsed_content.append((status_code, divider, content)) + _log_trace(bytes(raw_content)) + return stem.response.ControlMessage(parsed_content, bytes(raw_content), arrived_at = arrived_at) + elif divider == '+': + # data entry, all of the following lines belong to the content until we + # get a line with just a period + + content_block = bytearray(content) + + while True: + try: + line = control_file.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))))) + 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)))) + raise stem.ProtocolError('All lines should end with CRLF') + elif line == b'.\r\n': + break # data block termination + + line = line[:-2] # strips off the CRLF + + # lines starting with a period are escaped by a second period (as per + # section 2.4 of the control-spec) + + if line.startswith(b'..'): + line = line[1:] + + content_block += b'\n' + line + + # joins the content using a newline rather than CRLF separator (more + # conventional for multi-line string content outside the windows world) + + parsed_content.append((status_code, divider, bytes(content_block))) + else: + # this should never be reached due to the prefix regex, but might as well + # be safe... + + log.warn(ERROR_MSG % ('ProtocolError', "\"%s\" isn't a recognized divider type" % divider)) + raise stem.ProtocolError("Unrecognized divider type '%s': %s" % (divider, stem.util.str_tools._to_unicode(line))) + + +def send_formatting(message): + """ + Performs the formatting expected from sent control messages. For more + information see the :func:`~stem.socket.send_message` function. + + :param str message: message to be formatted + + :returns: **str** of the message wrapped by the formatting expected from + controllers + """ + + # From control-spec section 2.2... + # Command = Keyword OptArguments CRLF / "+" Keyword OptArguments CRLF CmdData + # Keyword = 1*ALPHA + # OptArguments = [ SP *(SP / VCHAR) ] + # + # A command is either a single line containing a Keyword and arguments, or a + # multiline command whose initial keyword begins with +, and whose data + # section ends with a single "." on a line of its own. + + # if we already have \r\n entries then standardize on \n to start with + message = message.replace('\r\n', '\n') + + if '\n' in message: + return '+%s\r\n.\r\n' % message.replace('\n', '\r\n') + else: + return message + '\r\n' + + +def _log_trace(response): + if not log.is_tracing(): + return + + log_message = stem.util.str_tools._to_unicode(response.replace(b'\r\n', b'\n').rstrip()) + log_message_lines = log_message.split('\n') + + if TRUNCATE_LOGS and len(log_message_lines) > TRUNCATE_LOGS: + log_message = '\n'.join(log_message_lines[:TRUNCATE_LOGS] + ['... %i more lines...' % (len(log_message_lines) - TRUNCATE_LOGS)]) + + if len(log_message_lines) > 2: + log.trace('Received from tor:\n%s' % log_message) + else: + log.trace('Received from tor: %s' % log_message.replace('\n', '\\n'))

1 0