Declutterfier! Saves Data!

--- a PPN by Garber Painting Akron. With Image Size Reduction included!

URL: http://github.com/python/cpython/pull/9507.patch

text.post_handshake_auth = True + server_context.verify_mode = ssl.CERT_REQUIRED + client_context.post_handshake_auth = True + client_context.load_cert_chain(SIGNED_CERTFILE) + + server = ThreadedEchoServer(context=server_context, chatty=False) + with server: + with client_context.wrap_socket(socket.socket(), + server_hostname=hostname) as s: + s.connect((HOST, server.port)) + s.write(b'HASCERT') + self.assertEqual(s.recv(1024), b'FALSE\n') + s.write(b'PHA') + self.assertEqual(s.recv(1024), b'OK\n') + s.write(b'HASCERT') + self.assertEqual(s.recv(1024), b'TRUE\n') + # PHA method just returns true when cert is already available + s.write(b'PHA') + self.assertEqual(s.recv(1024), b'OK\n') + s.write(b'GETCERT') + cert_text = s.recv(4096).decode('us-ascii') + self.assertIn('Python Software Foundation CA', cert_text) + + def test_pha_required_nocert(self): + client_context, server_context, hostname = testing_context() + server_context.post_handshake_auth = True + server_context.verify_mode = ssl.CERT_REQUIRED + client_context.post_handshake_auth = True + + server = ThreadedEchoServer(context=server_context, chatty=False) + with server: + with client_context.wrap_socket(socket.socket(), + server_hostname=hostname) as s: + s.connect((HOST, server.port)) + s.write(b'PHA') + # receive CertificateRequest + self.assertEqual(s.recv(1024), b'OK\n') + # send empty Certificate + Finish + s.write(b'HASCERT') + # receive alert + with self.assertRaisesRegex( + ssl.SSLError, + 'tlsv13 alert certificate required'): + s.recv(1024) + + def test_pha_optional(self): + if support.verbose: + sys.stdout.write("\n") + + client_context, server_context, hostname = testing_context() + server_context.post_handshake_auth = True + server_context.verify_mode = ssl.CERT_REQUIRED + client_context.post_handshake_auth = True + client_context.load_cert_chain(SIGNED_CERTFILE) + + # check CERT_OPTIONAL + server_context.verify_mode = ssl.CERT_OPTIONAL + server = ThreadedEchoServer(context=server_context, chatty=False) + with server: + with client_context.wrap_socket(socket.socket(), + server_hostname=hostname) as s: + s.connect((HOST, server.port)) + s.write(b'HASCERT') + self.assertEqual(s.recv(1024), b'FALSE\n') + s.write(b'PHA') + self.assertEqual(s.recv(1024), b'OK\n') + s.write(b'HASCERT') + self.assertEqual(s.recv(1024), b'TRUE\n') + + def test_pha_optional_nocert(self): + if support.verbose: + sys.stdout.write("\n") + + client_context, server_context, hostname = testing_context() + server_context.post_handshake_auth = True + server_context.verify_mode = ssl.CERT_OPTIONAL + client_context.post_handshake_auth = True + + server = ThreadedEchoServer(context=server_context, chatty=False) + with server: + with client_context.wrap_socket(socket.socket(), + server_hostname=hostname) as s: + s.connect((HOST, server.port)) + s.write(b'HASCERT') + self.assertEqual(s.recv(1024), b'FALSE\n') + s.write(b'PHA') + self.assertEqual(s.recv(1024), b'OK\n') + # optional doens't fail when client does not have a cert + s.write(b'HASCERT') + self.assertEqual(s.recv(1024), b'FALSE\n') + + def test_pha_no_pha_client(self): + client_context, server_context, hostname = testing_context() + server_context.post_handshake_auth = True + server_context.verify_mode = ssl.CERT_REQUIRED + client_context.load_cert_chain(SIGNED_CERTFILE) + + server = ThreadedEchoServer(context=server_context, chatty=False) + with server: + with client_context.wrap_socket(socket.socket(), + server_hostname=hostname) as s: + s.connect((HOST, server.port)) + with self.assertRaisesRegex(ssl.SSLError, 'not server'): + s.verify_client_post_handshake() + s.write(b'PHA') + self.assertIn(b'extension not received', s.recv(1024)) + + def test_pha_no_pha_server(self): + # server doesn't have PHA enabled, cert is requested in handshake + client_context, server_context, hostname = testing_context() + server_context.verify_mode = ssl.CERT_REQUIRED + client_context.post_handshake_auth = True + client_context.load_cert_chain(SIGNED_CERTFILE) + + server = ThreadedEchoServer(context=server_context, chatty=False) + with server: + with client_context.wrap_socket(socket.socket(), + server_hostname=hostname) as s: + s.connect((HOST, server.port)) + s.write(b'HASCERT') + self.assertEqual(s.recv(1024), b'TRUE\n') + # PHA doesn't fail if there is already a cert + s.write(b'PHA') + self.assertEqual(s.recv(1024), b'OK\n') + s.write(b'HASCERT') + self.assertEqual(s.recv(1024), b'TRUE\n') + + def test_pha_not_tls13(self): + # TLS 1.2 + client_context, server_context, hostname = testing_context() + server_context.verify_mode = ssl.CERT_REQUIRED + client_context.options |= ssl.OP_NO_TLSv1_3 + client_context.post_handshake_auth = True + client_context.load_cert_chain(SIGNED_CERTFILE) + + server = ThreadedEchoServer(context=server_context, chatty=False) + with server: + with client_context.wrap_socket(socket.socket(), + server_hostname=hostname) as s: + s.connect((HOST, server.port)) + # PHA fails for TLS != 1.3 + s.write(b'PHA') + self.assertIn(b'WRONG_SSL_VERSION', s.recv(1024)) + + def test_main(verbose=False): if support.verbose: import warnings @@ -3681,6 +3887,7 @@ def test_main(verbose=False): thread_info = support.threading_setup() if thread_info: tests.append(ThreadedTests) + tests.append(TestPostHandshakeAuth) try: support.run_unittest(*tests) diff --git a/Misc/NEWS.d/next/Library/2018-09-14-14-29-45.bpo-34670.17XwGB.rst b/Misc/NEWS.d/next/Library/2018-09-14-14-29-45.bpo-34670.17XwGB.rst new file mode 100644 index 00000000000000..c1a61293faa3b6 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2018-09-14-14-29-45.bpo-34670.17XwGB.rst @@ -0,0 +1,3 @@ +Add SSLContext.post_handshake_auth and +SSLSocket.verify_client_post_handshake for TLS 1.3's post +handshake authentication feature. diff --git a/Modules/_ssl.c b/Modules/_ssl.c index 2badf3172252cf..b0cfbdc96c073d 100644 --- a/Modules/_ssl.c +++ b/Modules/_ssl.c @@ -323,6 +323,9 @@ typedef struct { PyObject *set_hostname; #endif int check_hostname; +#ifdef TLS1_3_VERSION + int post_handshake_auth; +#endif } PySSLContext; typedef struct { @@ -2456,6 +2459,30 @@ _ssl__SSLSocket_tls_unique_cb_impl(PySSLSocket *self) return retval; } +/*[clinic input] +_ssl._SSLSocket.verify_client_post_handshake + +Initiate TLS 1.3 post-handshake authentication +[clinic start generated code]*/ + +static PyObject * +_ssl__SSLSocket_verify_client_post_handshake_impl(PySSLSocket *self) +/*[clinic end generated code: output=532147f3b1341425 input=6bfa874810a3d889]*/ +{ +#ifdef TLS1_3_VERSION + int err = SSL_verify_client_post_handshake(self->ssl); + if (err == 0) + return _setSSLError(NULL, 0, __FILE__, __LINE__); + else + Py_RETURN_NONE; +#else + PyErr_SetString(PyExc_NotImplementedError, + "Post-handshake auth is not supported by your " + "OpenSSL version."); + return NULL; +#endif +} + #ifdef OPENSSL_VERSION_1_1 static SSL_SESSION* @@ -2632,6 +2659,7 @@ static PyMethodDef PySSLMethods[] = { _SSL__SSLSOCKET_COMPRESSION_METHODDEF _SSL__SSLSOCKET_SHUTDOWN_METHODDEF _SSL__SSLSOCKET_TLS_UNIQUE_CB_METHODDEF + _SSL__SSLSOCKET_VERIFY_CLIENT_POST_HANDSHAKE_METHODDEF {NULL, NULL} }; @@ -2675,7 +2703,7 @@ static PyTypeObject PySSLSocket_Type = { */ static int -_set_verify_mode(SSL_CTX *ctx, enum py_ssl_cert_requirements n) +_set_verify_mode(PySSLContext *self, enum py_ssl_cert_requirements n) { int mode; int (*verify_cb)(int, X509_STORE_CTX *) = NULL; @@ -2695,9 +2723,13 @@ _set_verify_mode(SSL_CTX *ctx, enum py_ssl_cert_requirements n) "invalid value for verify_mode"); return -1; } +#ifdef TLS1_3_VERSION + if (self->post_handshake_auth) + mode |= SSL_VERIFY_POST_HANDSHAKE; +#endif /* keep current verify cb */ - verify_cb = SSL_CTX_get_verify_callback(ctx); - SSL_CTX_set_verify(ctx, mode, verify_cb); + verify_cb = SSL_CTX_get_verify_callback(self->ctx); + SSL_CTX_set_verify(self->ctx, mode, verify_cb); return 0; } @@ -2776,13 +2808,13 @@ _ssl__SSLContext_impl(PyTypeObject *type, int proto_version) /* Don't check host name by default */ if (proto_version == PY_SSL_VERSION_TLS_CLIENT) { self->check_hostname = 1; - if (_set_verify_mode(self->ctx, PY_SSL_CERT_REQUIRED) == -1) { + if (_set_verify_mode(self, PY_SSL_CERT_REQUIRED) == -1) { Py_DECREF(self); return NULL; } } else { self->check_hostname = 0; - if (_set_verify_mode(self->ctx, PY_SSL_CERT_NONE) == -1) { + if (_set_verify_mode(self, PY_SSL_CERT_NONE) == -1) { Py_DECREF(self); return NULL; } @@ -2871,6 +2903,11 @@ _ssl__SSLContext_impl(PyTypeObject *type, int proto_version) } #endif +#ifdef TLS1_3_VERSION + self->post_handshake_auth = 0; + SSL_CTX_set_post_handshake_auth(self->ctx, self->post_handshake_auth); +#endif + return (PyObject *)self; } @@ -3125,7 +3162,10 @@ _ssl__SSLContext__set_alpn_protocols_impl(PySSLContext *self, static PyObject * get_verify_mode(PySSLContext *self, void *c) { - switch (SSL_CTX_get_verify_mode(self->ctx)) { + /* ignore SSL_VERIFY_CLIENT_ONCE and SSL_VERIFY_POST_HANDSHAKE */ + int mask = (SSL_VERIFY_NONE | SSL_VERIFY_PEER | + SSL_VERIFY_FAIL_IF_NO_PEER_CERT); + switch (SSL_CTX_get_verify_mode(self->ctx) & mask) { case SSL_VERIFY_NONE: return PyLong_FromLong(PY_SSL_CERT_NONE); case SSL_VERIFY_PEER: @@ -3150,7 +3190,7 @@ set_verify_mode(PySSLContext *self, PyObject *arg, void *c) "check_hostname is enabled."); return -1; } - return _set_verify_mode(self->ctx, n); + return _set_verify_mode(self, n); } static PyObject * @@ -3247,6 +3287,42 @@ set_check_hostname(PySSLContext *self, PyObject *arg, void *c) return 0; } +static PyObject * +get_post_handshake_auth(PySSLContext *self, void *c) { +#if TLS1_3_VERSION + return PyBool_FromLong(self->post_handshake_auth); +#else + Py_RETURN_NONE; +#endif +} + +#if TLS1_3_VERSION +static int +set_post_handshake_auth(PySSLContext *self, PyObject *arg, void *c) { + int (*verify_cb)(int, X509_STORE_CTX *) = NULL; + int mode = SSL_CTX_get_verify_mode(self->ctx); + int pha = PyObject_IsTrue(arg); + + if (pha == -1) { + return -1; + } + self->post_handshake_auth = pha; + + /* client-side socket setting, ignored by server-side */ + SSL_CTX_set_post_handshake_auth(self->ctx, pha); + + /* server-side socket setting, ignored by client-side */ + verify_cb = SSL_CTX_get_verify_callback(self->ctx); + if (pha) { + mode |= SSL_VERIFY_POST_HANDSHAKE; + } else { + mode ^= SSL_VERIFY_POST_HANDSHAKE; + } + SSL_CTX_set_verify(self->ctx, mode, verify_cb); + + return 0; +} +#endif typedef struct { PyThreadState *thread_state; @@ -4118,6 +4194,13 @@ static PyGetSetDef context_getsetlist[] = { (setter) set_check_hostname, NULL}, {"options", (getter) get_options, (setter) set_options, NULL}, + {"post_handshake_auth", (getter) get_post_handshake_auth, +#ifdef TLS1_3_VERSION + (setter) set_post_handshake_auth, +#else + NULL, +#endif + NULL}, {"verify_flags", (getter) get_verify_flags, (setter) set_verify_flags, NULL}, {"verify_mode", (getter) get_verify_mode, diff --git a/Modules/clinic/_ssl.c.h b/Modules/clinic/_ssl.c.h index eabe2aa06686f2..c9206a684efd49 100644 --- a/Modules/clinic/_ssl.c.h +++ b/Modules/clinic/_ssl.c.h @@ -329,6 +329,24 @@ _ssl__SSLSocket_tls_unique_cb(PySSLSocket *self, PyObject *Py_UNUSED(ignored)) return _ssl__SSLSocket_tls_unique_cb_impl(self); } +PyDoc_STRVAR(_ssl__SSLSocket_verify_client_post_handshake__doc__, +"verify_client_post_handshake($self, /)\n" +"--\n" +"\n" +"Initiate TLS 1.3 post-handshake authentication"); + +#define _SSL__SSLSOCKET_VERIFY_CLIENT_POST_HANDSHAKE_METHODDEF \ + {"verify_client_post_handshake", (PyCFunction)_ssl__SSLSocket_verify_client_post_handshake, METH_NOARGS, _ssl__SSLSocket_verify_client_post_handshake__doc__}, + +static PyObject * +_ssl__SSLSocket_verify_client_post_handshake_impl(PySSLSocket *self); + +static PyObject * +_ssl__SSLSocket_verify_client_post_handshake(PySSLSocket *self, PyObject *Py_UNUSED(ignored)) +{ + return _ssl__SSLSocket_verify_client_post_handshake_impl(self); +} + static PyObject * _ssl__SSLContext_impl(PyTypeObject *type, int proto_version); @@ -1168,4 +1186,4 @@ _ssl_enum_crls(PyObject *module, PyObject **args, Py_ssize_t nargs, PyObject *kw #ifndef _SSL_ENUM_CRLS_METHODDEF #define _SSL_ENUM_CRLS_METHODDEF #endif /* !defined(_SSL_ENUM_CRLS_METHODDEF) */ -/*[clinic end generated code: output=c79fb0dfd3c90784 input=a9049054013a1b77]*/ +/*[clinic end generated code: output=a832758678f4d934 input=a9049054013a1b77]*/ diff --git a/Tools/ssl/multissltests.py b/Tools/ssl/multissltests.py index 9d668d4202abf8..2264c9331042e5 100755 --- a/Tools/ssl/multissltests.py +++ b/Tools/ssl/multissltests.py @@ -47,9 +47,9 @@ ] OPENSSL_RECENT_VERSIONS = [ - "1.0.2o", - "1.1.0h", - # "1.1.1-pre7", + "1.0.2p", + "1.1.0i", + "1.1.1", ] LIBRESSL_OLD_VERSIONS = [ @@ -58,7 +58,7 @@ ] LIBRESSL_RECENT_VERSIONS = [ - "2.7.3", + "2.7.4", ] # store files in ../multissl pFad - Phonifier reborn

Pfad - The Proxy pFad © 2024 Your Company Name. All rights reserved.

Note: This service is not intended for secure transactions such as banking, social media, email, or purchasing. Use at your own risk. We assume no liability whatsoever for broken pages.

Alternative Proxies: