bpo-18233: Review of SSLSocket.get_peer_cert_chain and SSLSocket.get_verified_chain

Author: chrisburr

Doc/library/ssl.rst

@@ -1259,20 +1259,39 @@ SSL sockets also have the following additional methods and attributes:
    .. versionchanged:: 3.9
       IPv6 address strings no longer have a trailing new line.
 
-.. method:: SSLSocket.getpeercertchain(binary_form=False, validate=True)
+.. method:: SSLSocket.get_peer_cert_chain(binary_form=False)
 
-   Returns certificate chain for the peer. If no chain is provided, returns
-   None. Otherwise returns a tuple of dicts containing information about the
-   certificates. The chain starts with the leaf certificate and ends with the
-   root certificate. If called on the client side, the leaf certificate is the
-   peer's certificate.
+   Returns an **unverified** certificate chain for the peer. If no chain is
+   provided, returns :const:`None`. Otherwise returns a tuple of dicts
+   containing information about the certificates. The chain starts with the
+   leaf certificate and ends with the root certificate. Return :const:`None`
+   if the session is resumed as peers do not send certificates.
 
-   If the optional argument *binary_form* is True, return a list of *binary_form*-encoded copies
-   of the certificates.
-   If the optional argument *validate* is False, return the peer's cert chain
-   without any validation and without the root CA cert.");
+   If the ``binary_form`` parameter is :const:`True`, and a chain is available,
+   this method returns a tuple with each element corresponding to the
+   DER-encoded form of the entire certificate as a sequence of bytes.
 
-    .. versionadded:: 3.9
+   .. versionadded:: 3.9
+
+   .. warning::
+      This is not a verified chain. See :meth:`ssl.SSLSocket.get_verified_chain`.
+
+.. method:: SSLSocket.get_verified_chain(binary_form=False)
+
+   Returns a verified certificate chain for the peer. If no chain is provided,
+   returns :const:`None`. Otherwise returns a tuple of dicts containing
+   information about the certificates. The chain starts with the leaf
+   certificate and ends with the root certificate. Return :const:`None` if the
+   session is resumed as peers do not send certificates.
+
+   If the ``binary_form`` parameter is :const:`True`, and a chain is available,
+   this method returns a tuple with each element corresponding to the
+   DER-encoded form of the entire certificate as a sequence of bytes.
+
+   .. versionadded:: 3.9
+
+   .. note::
+     This features requires OpenSSL 1.1.0 or newer.
 
 .. method:: SSLSocket.cipher()
 

Lib/ssl.py

@@ -905,12 +905,20 @@ def getpeercert(self, binary_form=False):
         """
         return self._sslobj.getpeercert(binary_form)
 
-    def getpeercertchain(self, binary_form=False, validate=True):
-        """"Returns the certificate chain of the SSL connection
-        as tuple of dicts.
+    def get_peer_cert_chain(self, binary_form=False):
+        """"Returns the certificate chain of the SSL connection as a tuple of
+        dicts. It is *not* a verified chain.
 
-        Return None if no chain is provieded."""
-        return self._sslobj.getpeercertchain(binary_form, validate)
+        Return ``None`` if no chain is provided."""
+        return self._sslobj.get_peer_cert_chain(binary_form)
+
+    if hasattr(_ssl._SSLSocket, 'get_verified_chain'):
+        def get_verified_chain(self, binary_form=False):
+            """"Returns the verified certificate chain of the SSL connection as a
+            tuple of dicts.
+
+            Return ``None`` if no chain is provided."""
+            return self._sslobj.get_verified_chain(binary_form)
 
     def selected_npn_protocol(self):
         """Return the currently selected NPN protocol as a string, or ``None``
@@ -1131,10 +1139,17 @@ def getpeercert(self, binary_form=False):
         return self._sslobj.getpeercert(binary_form)
 
     @_sslcopydoc
-    def getpeercertchain(self, binary_form=False, validate=True):
+    def get_peer_cert_chain(self, binary_form=False):
         self._checkClosed()
         self._check_connected()
-        return self._sslobj.getpeercertchain(binary_form, validate)
+        return self._sslobj.get_peer_cert_chain(binary_form)
+
+    if hasattr(_ssl._SSLSocket, 'get_verified_chain'):
+        @_sslcopydoc
+        def get_verified_chain(self, binary_form=False):
+            self._checkClosed()
+            self._check_connected()
+            return self._sslobj.get_verified_chain(binary_form)
 
     @_sslcopydoc
     def selected_npn_protocol(self):

Lib/test/test_ssl.py

@@ -2160,7 +2160,29 @@ def test_get_ca_certs_capath(self):
             self.assertTrue(cert)
         self.assertEqual(len(ctx.get_ca_certs()), 1)
 
-    def test_getpeercertchain(self):
+    def test_get_peer_cert_chain(self):
+        ctx = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
+        ctx.verify_mode = ssl.CERT_REQUIRED
+        ctx.load_verify_locations(capath=CAPATH)
+        with ctx.wrap_socket(socket.socket(socket.AF_INET)) as s:
+            s.connect(self.server_addr)
+            try:
+                peer_cert = s.getpeercert()
+                peer_cert_bin = s.getpeercert(True)
+                chain_no_validate = s.get_peer_cert_chain()
+                chain_bin_no_validate = s.get_peer_cert_chain(True)
+            finally:
+                self.assertTrue(peer_cert)
+                self.assertTrue(peer_cert_bin)
+
+                # ca cert
+                ca_certs = ctx.get_ca_certs()
+                self.assertEqual(len(ca_certs), 1)
+
+                self.assertEqual(chain_no_validate, (peer_cert,))
+                self.assertEqual(chain_bin_no_validate, (peer_cert_bin,))
+
+    def test_get_verified_chain(self):
         ctx = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
         ctx.verify_mode = ssl.CERT_REQUIRED
         ctx.load_verify_locations(capath=CAPATH)
@@ -2170,17 +2192,10 @@ def test_getpeercertchain(self):
                 peer_cert = s.getpeercert()
                 peer_cert_bin = s.getpeercert(True)
                 if IS_OPENSSL_1_1_0:
-                    chain = s.getpeercertchain()
-                    chain_bin = s.getpeercertchain(True)
+                    chain = s.get_verified_chain()
+                    chain_bin = s.get_verified_chain(True)
                 else:
-                    self.assertRaisesRegex(
-                        Exception, r'only supported by OpenSSL 1\.1\.0',
-                        s.getpeercertchain)
-                    self.assertRaisesRegex(
-                        Exception, r'only supported by OpenSSL 1\.1\.0',
-                        s.getpeercertchain, True)
-                chain_no_validate = s.getpeercertchain(validate=False)
-                chain_bin_no_validate = s.getpeercertchain(True, False)
+                    self.assertFalse(hasattr(s, 'get_verified_chain'))
             finally:
                 self.assertTrue(peer_cert)
                 self.assertTrue(peer_cert_bin)
@@ -2197,8 +2212,6 @@ def test_getpeercertchain(self):
                 if IS_OPENSSL_1_1_0:
                     self.assertEqual(chain, (peer_cert, test_get_ca_certsert))
                     self.assertEqual(chain_bin, (peer_cert_bin, ca_cert_bin))
-                self.assertEqual(chain_no_validate, (peer_cert,))
-                self.assertEqual(chain_bin_no_validate, (peer_cert_bin,))
 
     @needs_sni
     def test_context_setget(self):

Misc/NEWS.d/next/Library/2020-01-10-16-06-13.bpo-18233.EsqH1K.rst

@@ -1 +1 @@
-Add :meth:`ssl.SSLSocket.getpeercertchain` for accessing the certificate chain of SSL connections.
+Add :meth:`ssl.SSLSocket.get_peer_cert_chain` and :meth:`ssl.SSLSocket.get_verified_chain` for accessing the certificate chain of SSL connections.

Modules/_ssl.c

@@ -2096,64 +2096,17 @@ _ssl__SSLSocket_cipher_impl(PySSLSocket *self)
     return cipher_to_tuple(current);
 }
 
-/*[clinic input]
-_ssl._SSLSocket.getpeercertchain
-    der as binary_mode: bool = False
-    validate: bool = True
-[clinic start generated code]*/
-
 static PyObject *
-_ssl__SSLSocket_getpeercertchain_impl(PySSLSocket *self, int binary_mode,
-                                      int validate)
-/*[clinic end generated code: output=8094e6d78d27eb9a input=f4dcd181d0d163eb]*/
+chain_to_pyobject(STACK_OF(X509) *peer_chain, int binary_mode)
 {
     int len, i;
     PyObject *retval = NULL, *ci=NULL;
-    STACK_OF(X509) *peer_chain; /* reference */
-
-    assert((self->ctx != NULL) && (self->ctx->ctx != NULL));
-    if (self->ssl == NULL) {
-        Py_RETURN_NONE;
-    }
-
-    /* The peer just transmits the intermediate cert chain EXCLUDING the root
-     * CA certificate as this side is suppose to have a copy of the root
-     * certificate for verification. */
-    if (validate) {
-#ifdef OPENSSL_VERSION_1_1
-        peer_chain = SSL_get0_verified_chain(self->ssl);
-        long ret = SSL_get_verify_result(self->ssl);
-        if (ret != X509_V_OK) {
-#ifdef SSL_R_CERTIFICATE_VERIFY_FAILED
-            long e = ERR_PACK(ERR_LIB_SSL, 0, SSL_R_CERTIFICATE_VERIFY_FAILED);
-#else
-            long e = ERR_PACK(ERR_LIB_SSL, 0, 134);
-#endif
-            fill_and_set_sslerror(self, PySSLCertVerificationErrorObject, PY_SSL_ERROR_SSL, NULL, __LINE__, e);
-            return NULL;
-        }
-#else
-        PyErr_SetString(
-            PyExc_Exception,
-            "Getting verified certificate chains with SSL_get0_verified_chain"
-            " is only supported by OpenSSL 1.1.0 and later");
-        return NULL;
-#endif
-    } else {
-        peer_chain = SSL_get_peer_cert_chain(self->ssl);
-        if (self->socket_type == PY_SSL_SERVER) {
-            X509 *peer_cert = SSL_get_peer_certificate(self->ssl);
-            if (peer_cert != NULL)
-                sk_X509_insert(peer_chain, peer_cert, 0);
-        }
-    }
 
     if (peer_chain == NULL) {
         Py_RETURN_NONE;
     }
 
     len = sk_X509_num(peer_chain);
-
     if ((retval = PyTuple_New(len)) == NULL) {
         return NULL;
     }
@@ -2168,15 +2121,70 @@ _ssl__SSLSocket_getpeercertchain_impl(PySSLSocket *self, int binary_mode,
 
         if (ci == NULL) {
             Py_CLEAR(retval);
-            goto end;
+            break;
         }
         PyTuple_SET_ITEM(retval, i, ci);
     }
 
-  end:
     return retval;
 }
 
+/*[clinic input]
+_ssl._SSLSocket.get_peer_cert_chain
+    der as binary_mode: bool = False
+[clinic start generated code]*/
+
+static PyObject *
+_ssl__SSLSocket_get_peer_cert_chain_impl(PySSLSocket *self, int binary_mode)
+/*[clinic end generated code: output=2fc1e5dce85798ba input=3dd8f43730febaa9]*/
+{
+    STACK_OF(X509) *peer_chain; /* reference */
+
+    assert((self->ctx != NULL) && (self->ctx->ctx != NULL));
+    if (self->ssl == NULL)
+        Py_RETURN_NONE;
+
+    peer_chain = SSL_get_peer_cert_chain(self->ssl);
+    /* In OpenSSL only the client side includes the peer certificate.
+     * Manually add it if required it to be more consistent. */
+    if (self->socket_type == PY_SSL_SERVER) {
+        X509 *peer_cert = SSL_get_peer_certificate(self->ssl);
+        if (peer_cert != NULL) {
+            if (peer_chain == NULL)
+                peer_chain = sk_X509_new_null();
+            sk_X509_insert(peer_chain, peer_cert, 0);
+        }
+    }
+    return chain_to_pyobject(peer_chain, binary_mode);
+}
+
+#ifdef OPENSSL_VERSION_1_1
+/*[clinic input]
+_ssl._SSLSocket.get_verified_chain
+    der as binary_mode: bool = False
+[clinic start generated code]*/
+
+static PyObject *
+_ssl__SSLSocket_get_verified_chain_impl(PySSLSocket *self, int binary_mode)
+/*[clinic end generated code: output=6e07b709feaeb291 input=8f51efb220ed687f]*/
+{
+    STACK_OF(X509) *peer_chain; /* reference */
+
+    assert((self->ctx != NULL) && (self->ctx->ctx != NULL));
+    if (self->ssl == NULL)
+        Py_RETURN_NONE;
+
+    peer_chain = SSL_get0_verified_chain(self->ssl);
+    long ret = SSL_get_verify_result(self->ssl);
+    if (ret != X509_V_OK) {
+        long e = ERR_PACK(ERR_LIB_SSL, 0, SSL_R_CERTIFICATE_VERIFY_FAILED);
+        fill_and_set_sslerror(self, PySSLCertVerificationErrorObject, PY_SSL_ERROR_SSL, NULL, __LINE__, e);
+        return NULL;
+    }
+    return chain_to_pyobject(peer_chain, binary_mode);
+}
+#endif
+
 /*[clinic input]
 _ssl._SSLSocket.version
 [clinic start generated code]*/
@@ -3081,7 +3089,10 @@ static PyMethodDef PySSLMethods[] = {
     _SSL__SSLSOCKET_GET_CHANNEL_BINDING_METHODDEF
     _SSL__SSLSOCKET_CIPHER_METHODDEF
     _SSL__SSLSOCKET_SHARED_CIPHERS_METHODDEF
-    _SSL__SSLSOCKET_GETPEERCERTCHAIN_METHODDEF
+    _SSL__SSLSOCKET_GET_PEER_CERT_CHAIN_METHODDEF
+#ifdef OPENSSL_VERSION_1_1
+    _SSL__SSLSOCKET_GET_VERIFIED_CHAIN_METHODDEF
+#endif
     _SSL__SSLSOCKET_VERSION_METHODDEF
     _SSL__SSLSOCKET_SELECTED_NPN_PROTOCOL_METHODDEF
     _SSL__SSLSOCKET_SELECTED_ALPN_PROTOCOL_METHODDEF