From d649480739dba77d9bfb1f320b52e9a838c33a05 Mon Sep 17 00:00:00 2001 From: Antoine Pitrou Date: Thu, 21 Jul 2011 01:11:30 +0200 Subject: [PATCH] Issue #12551: Provide a get_channel_binding() method on SSL sockets so as to get channel binding data for the current SSL session (only the "tls-unique" channel binding is implemented). This allows the implementation of certain authentication mechanisms such as SCRAM-SHA-1-PLUS. Patch by Jacek Konieczny. --- Doc/library/ssl.rst | 19 +++++++++ Lib/ssl.py | 19 +++++++++ Lib/test/test_ssl.py | 91 ++++++++++++++++++++++++++++++++++++++++++++ Misc/ACKS | 1 + Misc/NEWS | 6 +++ Modules/_ssl.c | 60 +++++++++++++++++++++++++++++ 6 files changed, 196 insertions(+) diff --git a/Doc/library/ssl.rst b/Doc/library/ssl.rst index 5342e788973..0ac0ac100e6 100644 --- a/Doc/library/ssl.rst +++ b/Doc/library/ssl.rst @@ -386,6 +386,13 @@ Constants .. versionadded:: 3.2 +.. data:: CHANNEL_BINDING_TYPES + + List of supported TLS channel binding types. Strings in this list + can be used as arguments to :meth:`SSLSocket.get_channel_binding`. + + .. versionadded:: 3.3 + .. data:: OPENSSL_VERSION The version string of the OpenSSL library loaded by the interpreter:: @@ -495,6 +502,18 @@ SSL sockets also have the following additional methods and attributes: version of the SSL protocol that defines its use, and the number of secret bits being used. If no connection has been established, returns ``None``. +.. method:: SSLSocket.get_channel_binding(cb_type="tls-unique") + + Get channel binding data for current connection, as a bytes object. Returns + ``None`` if not connected or the handshake has not been completed. + + The *cb_type* parameter allow selection of the desired channel binding + type. Valid channel binding types are listed in the + :data:`CHANNEL_BINDING_TYPES` list. Currently only the 'tls-unique' channel + binding, defined by :rfc:`5929`, is supported. :exc:`ValueError` will be + raised if an unsupported channel binding type is requested. + + .. versionadded:: 3.3 .. method:: SSLSocket.unwrap() diff --git a/Lib/ssl.py b/Lib/ssl.py index cde99fc0831..914e749928f 100644 --- a/Lib/ssl.py +++ b/Lib/ssl.py @@ -99,6 +99,10 @@ import base64 # for DER-to-PEM translation import traceback import errno +if _ssl.HAS_TLS_UNIQUE: + CHANNEL_BINDING_TYPES = ['tls-unique'] +else: + CHANNEL_BINDING_TYPES = [] class CertificateError(ValueError): pass @@ -495,6 +499,21 @@ class SSLSocket(socket): self.do_handshake_on_connect), addr) + def get_channel_binding(self, cb_type="tls-unique"): + """Get channel binding data for current connection. Raise ValueError + if the requested `cb_type` is not supported. Return bytes of the data + or None if the data is not available (e.g. before the handshake). + """ + if cb_type not in CHANNEL_BINDING_TYPES: + raise ValueError("Unsupported channel binding type") + if cb_type != "tls-unique": + raise NotImplementedError( + "{0} channel binding type not implemented" + .format(cb_type)) + if self._sslobj is None: + return None + return self._sslobj.tls_unique_cb() + def __del__(self): # sys.stderr.write("__del__ on %s\n" % repr(self)) self._real_close() diff --git a/Lib/test/test_ssl.py b/Lib/test/test_ssl.py index fd1cd2da353..f3f0c544599 100644 --- a/Lib/test/test_ssl.py +++ b/Lib/test/test_ssl.py @@ -321,6 +321,25 @@ class BasicSocketTests(unittest.TestCase): self.assertRaises(ValueError, ctx.wrap_socket, sock, True, server_hostname="some.hostname") + def test_unknown_channel_binding(self): + # should raise ValueError for unknown type + s = socket.socket(socket.AF_INET) + ss = ssl.wrap_socket(s) + with self.assertRaises(ValueError): + ss.get_channel_binding("unknown-type") + + @unittest.skipUnless("tls-unique" in ssl.CHANNEL_BINDING_TYPES, + "'tls-unique' channel binding not available") + def test_tls_unique_channel_binding(self): + # unconnected should return None for known type + s = socket.socket(socket.AF_INET) + ss = ssl.wrap_socket(s) + self.assertIsNone(ss.get_channel_binding("tls-unique")) + # the same for server-side + s = socket.socket(socket.AF_INET) + ss = ssl.wrap_socket(s, server_side=True, certfile=CERTFILE) + self.assertIsNone(ss.get_channel_binding("tls-unique")) + class ContextTests(unittest.TestCase): @skip_if_broken_ubuntu_ssl @@ -826,6 +845,11 @@ else: self.sslconn = None if support.verbose and self.server.connectionchatty: sys.stdout.write(" server: connection is now unencrypted...\n") + elif stripped == b'CB tls-unique': + if support.verbose and self.server.connectionchatty: + sys.stdout.write(" server: read CB tls-unique from client, sending our CB data...\n") + data = self.sslconn.get_channel_binding("tls-unique") + self.write(repr(data).encode("us-ascii") + b"\n") else: if (support.verbose and self.server.connectionchatty): @@ -1625,6 +1649,73 @@ else: t.join() server.close() + @unittest.skipUnless("tls-unique" in ssl.CHANNEL_BINDING_TYPES, + "'tls-unique' channel binding not available") + def test_tls_unique_channel_binding(self): + """Test tls-unique channel binding.""" + if support.verbose: + sys.stdout.write("\n") + + server = ThreadedEchoServer(CERTFILE, + certreqs=ssl.CERT_NONE, + ssl_version=ssl.PROTOCOL_TLSv1, + cacerts=CERTFILE, + chatty=True, + connectionchatty=False) + flag = threading.Event() + server.start(flag) + # wait for it to start + flag.wait() + # try to connect + s = ssl.wrap_socket(socket.socket(), + server_side=False, + certfile=CERTFILE, + ca_certs=CERTFILE, + cert_reqs=ssl.CERT_NONE, + ssl_version=ssl.PROTOCOL_TLSv1) + s.connect((HOST, server.port)) + try: + # get the data + cb_data = s.get_channel_binding("tls-unique") + if support.verbose: + sys.stdout.write(" got channel binding data: {0!r}\n" + .format(cb_data)) + + # check if it is sane + self.assertIsNotNone(cb_data) + self.assertEqual(len(cb_data), 12) # True for TLSv1 + + # and compare with the peers version + s.write(b"CB tls-unique\n") + peer_data_repr = s.read().strip() + self.assertEqual(peer_data_repr, + repr(cb_data).encode("us-ascii")) + s.close() + + # now, again + s = ssl.wrap_socket(socket.socket(), + server_side=False, + certfile=CERTFILE, + ca_certs=CERTFILE, + cert_reqs=ssl.CERT_NONE, + ssl_version=ssl.PROTOCOL_TLSv1) + s.connect((HOST, server.port)) + new_cb_data = s.get_channel_binding("tls-unique") + if support.verbose: + sys.stdout.write(" got another channel binding data: {0!r}\n" + .format(new_cb_data)) + # is it really unique + self.assertNotEqual(cb_data, new_cb_data) + self.assertIsNotNone(cb_data) + self.assertEqual(len(cb_data), 12) # True for TLSv1 + s.write(b"CB tls-unique\n") + peer_data_repr = s.read().strip() + self.assertEqual(peer_data_repr, + repr(new_cb_data).encode("us-ascii")) + s.close() + finally: + server.stop() + server.join() def test_main(verbose=False): if support.verbose: diff --git a/Misc/ACKS b/Misc/ACKS index 1e9a721911a..8a63f50f8fb 100644 --- a/Misc/ACKS +++ b/Misc/ACKS @@ -516,6 +516,7 @@ Lenny Kneler Pat Knight Greg Kochanski Damon Kohler +Jacek Konieczny Марк Коренберг Vlad Korolev Joseph Koshy diff --git a/Misc/NEWS b/Misc/NEWS index d648501aa9e..c3e0f526cdf 100644 --- a/Misc/NEWS +++ b/Misc/NEWS @@ -234,6 +234,12 @@ Core and Builtins Library ------- +- Issue #12551: Provide a get_channel_binding() method on SSL sockets so as + to get channel binding data for the current SSL session (only the + "tls-unique" channel binding is implemented). This allows the implementation + of certain authentication mechanisms such as SCRAM-SHA-1-PLUS. Patch by + Jacek Konieczny. + - Issue #665194: email.utils now has format_datetime and parsedate_to_datetime functions, allowing for round tripping of RFC2822 format dates. diff --git a/Modules/_ssl.c b/Modules/_ssl.c index d2d2480b5cd..1a367f22753 100644 --- a/Modules/_ssl.c +++ b/Modules/_ssl.c @@ -124,6 +124,17 @@ static unsigned int _ssl_locks_count = 0; # undef HAVE_SSL_CTX_CLEAR_OPTIONS #endif +/* In case of 'tls-unique' it will be 12 bytes for TLS, 36 bytes for + * older SSL, but let's be safe */ +#define PySSL_CB_MAXLEN 128 + +/* SSL_get_finished got added to OpenSSL in 0.9.5 */ +#if OPENSSL_VERSION_NUMBER >= 0x0090500fL +# define HAVE_OPENSSL_FINISHED 1 +#else +# define HAVE_OPENSSL_FINISHED 0 +#endif + typedef struct { PyObject_HEAD SSL_CTX *ctx; @@ -135,6 +146,7 @@ typedef struct { SSL *ssl; X509 *peer_cert; int shutdown_seen_zero; + enum py_ssl_server_or_client socket_type; } PySSLSocket; static PyTypeObject PySSLContext_Type; @@ -328,6 +340,7 @@ newPySSLSocket(SSL_CTX *ctx, PySocketSockObject *sock, SSL_set_accept_state(self->ssl); PySSL_END_ALLOW_THREADS + self->socket_type = socket_type; self->Socket = PyWeakref_NewRef((PyObject *) sock, NULL); return self; } @@ -1377,6 +1390,41 @@ PyDoc_STRVAR(PySSL_SSLshutdown_doc, Does the SSL shutdown handshake with the remote end, and returns\n\ the underlying socket object."); +#if HAVE_OPENSSL_FINISHED +static PyObject * +PySSL_tls_unique_cb(PySSLSocket *self) +{ + PyObject *retval = NULL; + char buf[PySSL_CB_MAXLEN]; + int len; + + if (SSL_session_reused(self->ssl) ^ !self->socket_type) { + /* if session is resumed XOR we are the client */ + len = SSL_get_finished(self->ssl, buf, PySSL_CB_MAXLEN); + } + else { + /* if a new session XOR we are the server */ + len = SSL_get_peer_finished(self->ssl, buf, PySSL_CB_MAXLEN); + } + + /* It cannot be negative in current OpenSSL version as of July 2011 */ + assert(len >= 0); + if (len == 0) + Py_RETURN_NONE; + + retval = PyBytes_FromStringAndSize(buf, len); + + return retval; +} + +PyDoc_STRVAR(PySSL_tls_unique_cb_doc, +"tls_unique_cb() -> bytes\n\ +\n\ +Returns the 'tls-unique' channel binding data, as defined by RFC 5929.\n\ +\n\ +If the TLS handshake is not yet complete, None is returned"); + +#endif /* HAVE_OPENSSL_FINISHED */ static PyMethodDef PySSLMethods[] = { {"do_handshake", (PyCFunction)PySSL_SSLdo_handshake, METH_NOARGS}, @@ -1391,6 +1439,10 @@ static PyMethodDef PySSLMethods[] = { {"cipher", (PyCFunction)PySSL_cipher, METH_NOARGS}, {"shutdown", (PyCFunction)PySSL_SSLshutdown, METH_NOARGS, PySSL_SSLshutdown_doc}, +#if HAVE_OPENSSL_FINISHED + {"tls_unique_cb", (PyCFunction)PySSL_tls_unique_cb, METH_NOARGS, + PySSL_tls_unique_cb_doc}, +#endif {NULL, NULL} }; @@ -2221,6 +2273,14 @@ PyInit__ssl(void) Py_INCREF(r); PyModule_AddObject(m, "HAS_SNI", r); +#if HAVE_OPENSSL_FINISHED + r = Py_True; +#else + r = Py_False; +#endif + Py_INCREF(r); + PyModule_AddObject(m, "HAS_TLS_UNIQUE", r); + /* OpenSSL version */ /* SSLeay() gives us the version of the library linked against, which could be different from the headers version.