Use windows api if ssl cert verification fails

This avoids rejecting certificates that should be accepted, see python#80192
tobil4sk · Dec 5, 2024 · 9d28198 · 9d28198
1 parent 6bc3e83
commit 9d28198
Show file tree

Hide file tree

Showing 2 changed files with 115 additions and 1 deletion.
diff --git a/Misc/NEWS.d/next/Windows/2024-12-05-00-21-11.gh-issue-80192.28PTj4.rst b/Misc/NEWS.d/next/Windows/2024-12-05-00-21-11.gh-issue-80192.28PTj4.rst
@@ -0,0 +1 @@
+Fixed valid ssl certificates being rejected.
diff --git a/Modules/_ssl.c b/Modules/_ssl.c
@@ -2996,6 +2996,119 @@ static PyType_Spec PySSLSocket_spec = {
     .slots = PySSLSocket_slots,
 };
 
+#ifdef _MSC_VER
+static unsigned char *
+_get_cert_bytes(X509 *cert, int *length)
+{
+    unsigned char *cert_bytes;
+    int cert_bytes_length = i2d_X509(cert, NULL);
+    if (cert_bytes_length <= 0)
+    {
+        return NULL;
+    }
+
+    cert_bytes = PyMem_RawMalloc(cert_bytes_length);
+    if (cert_bytes == NULL)
+    {
+        return NULL;
+    }
+
+    if (i2d_X509(cert, &cert_bytes) <= 0)
+    {
+        return NULL;
+    }
+
+    *length = cert_bytes_length;
+
+    /* i2d_X509 moves the input pointer to the end of the data */
+    return cert_bytes - cert_bytes_length;
+}
+
+
+static int
+_verify_callback(int preverify_ok, X509_STORE_CTX *ctx)
+{
+    int ret = 0;
+    /* windows api calls below do not check for hostname mismatch */
+    if (preverify_ok || X509_STORE_CTX_get_error(ctx) == X509_V_ERR_HOSTNAME_MISMATCH)
+    {
+        return preverify_ok;
+    }
+
+    HCERTSTORE store = CertOpenStore(CERT_STORE_PROV_MEMORY, 0, 0, CERT_STORE_DEFER_CLOSE_UNTIL_LAST_FREE_FLAG, NULL);
+    if (store == NULL)
+    {
+        return ret;
+    }
+
+    int cert_bytes_length;
+    unsigned char *cert_bytes = _get_cert_bytes(X509_STORE_CTX_get_current_cert(ctx), &cert_bytes_length);
+    if (cert_bytes == NULL)
+    {
+        goto error_1;
+    }
+
+    PCCERT_CONTEXT primary_context = NULL;
+    if (!CertAddEncodedCertificateToStore(store, X509_ASN_ENCODING, cert_bytes, cert_bytes_length, CERT_STORE_ADD_REPLACE_EXISTING, &primary_context))
+    {
+        goto error_2;
+    }
+    PCCERT_CHAIN_CONTEXT chain_context = NULL;
+    CERT_CHAIN_PARA parameters = {0};
+    if (!CertGetCertificateChain(NULL, primary_context, NULL, store, &parameters, 0, NULL, &chain_context))
+    {
+        goto error_3;
+    }
+    CERT_CHAIN_POLICY_PARA policy_parameters = {0};
+    CERT_CHAIN_POLICY_STATUS policy_status = {0};
+    if (!CertVerifyCertificateChainPolicy(CERT_CHAIN_POLICY_SSL, chain_context, &policy_parameters, &policy_status))
+    {
+        goto error_4;
+    }
+
+    switch(policy_status.dwError) {
+    case CERT_TRUST_NO_ERROR:
+        ret = 1;
+        break;
+    case TRUST_E_CERT_SIGNATURE:
+        X509_STORE_CTX_set_error(ctx, X509_V_ERR_CERT_SIGNATURE_FAILURE);
+        break;
+    case CERT_E_UNTRUSTEDROOT:
+        X509_STORE_CTX_set_error(ctx, X509_V_ERR_CERT_UNTRUSTED);
+        break;
+    case CERT_E_CHAINING:
+        X509_STORE_CTX_set_error(ctx, X509_V_ERR_UNABLE_TO_GET_ISSUER_CERT);
+        break;
+    case CERT_E_EXPIRED:
+        X509_STORE_CTX_set_error(ctx, X509_V_ERR_CERT_HAS_EXPIRED);
+        break;
+    case CERT_E_INVALID_POLICY:
+        X509_STORE_CTX_set_error(ctx, X509_V_ERR_INVALID_POLICY_EXTENSION);
+        break;
+    case CERT_E_WRONG_USAGE:
+    case CERT_E_CRITICAL:
+    case CERT_E_PURPOSE:
+    case CERT_E_ROLE:
+        X509_STORE_CTX_set_error(ctx, X509_V_ERR_INVALID_PURPOSE);
+        break;
+    case CERT_E_VALIDITYPERIODNESTING:
+    default:
+        X509_STORE_CTX_set_error(ctx, X509_V_ERR_APPLICATION_VERIFICATION);
+    }
+error_4:
+    CertFreeCertificateChain(chain_context);
+error_3:
+    CertFreeCertificateContext(primary_context);
+error_2:
+    PyMem_RawFree(cert_bytes);
+error_1:
+    CertCloseStore(store, 0);
+    return ret;
+}
+#else
+#define _verify_callback NULL
+#endif
+
 /*
  * _SSLContext objects
  */
@@ -3024,7 +3137,7 @@ _set_verify_mode(PySSLContext *self, enum py_ssl_cert_requirements n)
     /* bpo-37428: newPySSLSocket() sets SSL_VERIFY_POST_HANDSHAKE flag for
      * server sockets and SSL_set_post_handshake_auth() for client. */
 
-    SSL_CTX_set_verify(self->ctx, mode, NULL);
+    SSL_CTX_set_verify(self->ctx, mode, _verify_callback);
     return 0;
 }