[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]
[tor-commits] [stem/master] Function and testing for password authentication
commit 7f760f86414ee0bfbd050480e1753555c66e9a5b
Author: Damian Johnson <atagar@xxxxxxxxxxxxxx>
Date: Tue Nov 29 10:02:24 2011 -0800
Function and testing for password authentication
Adding a function for password authentication. This included escaping quotes
but otherwise is trivial - most of the effort was refactoring the
authentication integ tests.
---
stem/connection.py | 30 +++++++
test/integ/connection/authentication.py | 130 +++++++++++++++++++++++++++----
2 files changed, 145 insertions(+), 15 deletions(-)
diff --git a/stem/connection.py b/stem/connection.py
index bef8d50..3ea5747 100644
--- a/stem/connection.py
+++ b/stem/connection.py
@@ -62,6 +62,36 @@ def authenticate_none(control_socket):
if str(auth_response) != "OK":
raise ValueError(str(auth_response))
+def authenticate_password(control_socket, password):
+ """
+ Authenticates to a control socket that uses a password (via the
+ HashedControlPassword torrc option). Quotes in the password are escaped.
+
+ If authentication fails then tor will close the control socket.
+
+ Arguments:
+ control_socket (stem.socket.ControlSocket) - socket to be authenticated
+ password (str) - passphrase to present to the socket
+
+ Raises:
+ ValueError if the authentication credentials aren't accepted
+ stem.socket.ProtocolError the content from the socket is malformed
+ stem.socket.SocketError if problems arise in using the socket
+ """
+
+ # Escapes quotes. Tor can include those in the password hash, in which case
+ # it expects escaped quotes from the controller. For more information see...
+ # https://trac.torproject.org/projects/tor/ticket/4600
+
+ password = password.replace('"', '\\"')
+
+ control_socket.send("AUTHENTICATE \"%s\"" % password)
+ auth_response = control_socket.recv()
+
+ # if we got anything but an OK response then error
+ if str(auth_response) != "OK":
+ raise ValueError(str(auth_response))
+
def get_protocolinfo_by_port(control_addr = "127.0.0.1", control_port = 9051, get_socket = False):
"""
Issues a PROTOCOLINFO query to a control port, getting information about the
diff --git a/test/integ/connection/authentication.py b/test/integ/connection/authentication.py
index 4ba0bfa..c3759e6 100644
--- a/test/integ/connection/authentication.py
+++ b/test/integ/connection/authentication.py
@@ -4,6 +4,7 @@ stem.connection.authenticate_* functions.
"""
import unittest
+import functools
import test.runner
import stem.connection
@@ -15,6 +16,9 @@ COOKIE_AUTH_FAIL = "Authentication failed: Wrong length on authentication cookie
PASSWORD_AUTH_FAIL = "Authentication failed: Password did not match HashedControlPassword value from configuration. Maybe you tried a plain text password? If so, the standard requires that you put it in double quotes."
MULTIPLE_AUTH_FAIL = "Authentication failed: Password did not match HashedControlPassword *or* authentication cookie."
+# this only arises in password-only auth when we authenticate by password
+INCORRECT_PASSWORD_FAIL = "Authentication failed: Password did not match HashedControlPassword value from configuration"
+
class TestAuthenticate(unittest.TestCase):
"""
Tests the authentication methods. This should be run with the 'CONN_ALL'
@@ -32,31 +36,127 @@ class TestAuthenticate(unittest.TestCase):
if connection_type == test.runner.TorConnection.NONE:
self.skipTest("(no connection)")
- # If the connection has authentication then this will fail with a message
- # based on the authentication type. If not then this will succeed.
+ expect_success = self._is_authenticateable(stem.connection.AuthMethod.NONE)
+ self._check_auth(stem.connection.AuthMethod.NONE, None, expect_success)
+
+ def test_authenticate_password(self):
+ """
+ Tests the authenticate_password function.
+ """
+
+ runner = test.runner.get_runner()
+ connection_type = runner.get_connection_type()
+
+ if connection_type == test.runner.TorConnection.NONE:
+ self.skipTest("(no connection)")
+
+ expect_success = self._is_authenticateable(stem.connection.AuthMethod.PASSWORD)
+ self._check_auth(stem.connection.AuthMethod.PASSWORD, test.runner.CONTROL_PASSWORD, expect_success)
+
+ # Check with an empty, invalid, and quoted password. These should work if
+ # we have no authentication, and fail otherwise.
+
+ expect_success = self._is_authenticateable(stem.connection.AuthMethod.NONE)
+ self._check_auth(stem.connection.AuthMethod.PASSWORD, "", expect_success)
+ self._check_auth(stem.connection.AuthMethod.PASSWORD, "blarg", expect_success)
+ self._check_auth(stem.connection.AuthMethod.PASSWORD, "this has a \" in it", expect_success)
+
+ def _get_socket_auth(self):
+ """
+ Provides the types of authentication that our current test socket accepts.
- control_socket = test.runner.get_runner().get_tor_socket(False)
+ Returns:
+ bool tuple of the form (password_auth, cookie_auth)
+ """
+ connection_type = test.runner.get_runner().get_connection_type()
connection_options = test.runner.CONNECTION_OPTS[connection_type]
- cookie_auth = test.runner.OPT_COOKIE in connection_options
password_auth = test.runner.OPT_PASSWORD in connection_options
+ cookie_auth = test.runner.OPT_COOKIE in connection_options
- if cookie_auth or password_auth:
- if cookie_auth and password_auth: failure_msg = MULTIPLE_AUTH_FAIL
- elif cookie_auth: failure_msg = COOKIE_AUTH_FAIL
- else: failure_msg = PASSWORD_AUTH_FAIL
-
- try:
- stem.connection.authenticate_none(control_socket)
- self.fail()
- except ValueError, exc:
- self.assertEqual(failure_msg, str(exc))
+ return password_auth, cookie_auth
+
+ def _is_authenticateable(self, auth_type):
+ """
+ Checks if the given authentication type should be able to authenticate to
+ our current socket.
+
+ Arguments:
+ auth_type (stem.connection.AuthMethod) - authentication method to check
+
+ Returns:
+ bool that's True if we should be able to authenticate and False otherwise
+ """
+
+ password_auth, cookie_auth = self._get_socket_auth()
+
+ # If the control socket is open then all authentication methods will be
+ # accepted. Otherwise check if our auth type matches what the socket
+ # accepts.
+
+ if not password_auth and not cookie_auth: return True
+ elif auth_type == stem.connection.AuthMethod.PASSWORD: return password_auth
+ elif auth_type == stem.connection.AuthMethod.COOKIE: return cookie_auth
+ else: return False
+
+ def _check_auth(self, auth_type, auth_value, expect_success):
+ """
+ Attempts to use the given authentication function against our connection.
+ If this works then checks that we can use the connection. If not then we
+ check that the error message is what we'd expect.
+
+ Arguments:
+ auth_type (stem.connection.AuthMethod) - method by which we should
+ authentiate to the control socket
+ auth_value (str) - value to be provided to the authentication function
+ expect_success (bool) - true if the authentication should succeed, false
+ otherwise
+ """
+
+ runner = test.runner.get_runner()
+ control_socket = runner.get_tor_socket(False)
+ password_auth, cookie_auth = self._get_socket_auth()
+
+ # construct the function call
+
+ if auth_type == stem.connection.AuthMethod.NONE:
+ auth_function = stem.connection.authenticate_none
+ elif auth_type == stem.connection.AuthMethod.PASSWORD:
+ auth_function = stem.connection.authenticate_password
+ elif auth_type == stem.connection.AuthMethod.COOKIE:
+ auth_function = None # TODO: fill in
+ else:
+ raise ValueError("unexpected auth type: %s" % auth_type)
+
+ if auth_value != None:
+ auth_function = functools.partial(auth_function, control_socket, auth_value)
else:
- stem.connection.authenticate_none(control_socket)
+ auth_function = functools.partial(auth_function, control_socket)
+
+ if expect_success:
+ auth_function()
# issues a 'GETINFO config-file' query to confirm that we can use the socket
control_socket.send("GETINFO config-file")
config_file_response = control_socket.recv()
self.assertEquals("config-file=%s\nOK" % runner.get_torrc_path(), str(config_file_response))
+ control_socket.close()
+ else:
+ if cookie_auth and password_auth: failure_msg = MULTIPLE_AUTH_FAIL
+ elif cookie_auth: failure_msg = COOKIE_AUTH_FAIL
+ else:
+ # if we're attempting to authenticate with a password then it's a
+ # truncated message
+
+ if auth_type == stem.connection.AuthMethod.PASSWORD:
+ failure_msg = INCORRECT_PASSWORD_FAIL
+ else:
+ failure_msg = PASSWORD_AUTH_FAIL
+
+ try:
+ auth_function()
+ self.fail()
+ except ValueError, exc:
+ self.assertEqual(failure_msg, str(exc))
_______________________________________________
tor-commits mailing list
tor-commits@xxxxxxxxxxxxxxxxxxxx
https://lists.torproject.org/cgi-bin/mailman/listinfo/tor-commits