Python Callback Server Example

server.py

if __name__ == '__main__':
    ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)


    # 1. using self signed certificate 
    ssl_context.load_cert_chain(certfile='new-server.crt', keyfile='new-server.key')


    # 2. using converted .pem bundle file from GTR 
    ssl_context.load_verify_locations(cafile='ca_bundle.pem')


    # 3. the incomining request required SSL Client CERT or not
    ssl_context.verify_mode = ssl.CERT_REQUIRED
    
    app.run(ssl_context=ssl_context, host="0.0.0.0", port=8443)

Where new-server.crt and new-server.key are signed by yourself. The following OpenSSL command is valid to provide the certificate by yourself, or you can reach a CA to sign a trust-root CA validated certificate, but it is not necessary. The client to your callback server should only be GTR, not public.

The following command provides an example of generating self-signed certificates.

openssl genrsa -des3 -out new-server.key 2048

openssl req -new -key new-server.key -out new-server.csr

cp new-server.key new-server.key.org

openssl rsa -in new-server.key.org -out new-server.key

openssl x509 -req -days 365 -in new-server.csr -signkey new-server.key -out new-server.crt

For further validation, ca_bundle.pem set in verify_mode only verifies the root-CA, not the GTR certificate. A simple way to verify by yourself is using string comparison to match SSL_CLIENT_CERT with ca_bundle.pem; if the string exists, then it passes. Or if strong verification is required, you could verify that the SSL_CLIENT_CERT is valid and matches the common name. Please check the example below to demonstrate how to verify the CN matches and valid dates.

middlewares.py

from flask import Flask, request
import ssl
from OpenSSL.crypto import X509, X509Store, X509StoreContextError, FILETYPE_PEM, X509StoreContext, load_certificate
import datetime

# load global trusted ca
store = X509Store()
ca_certs = load_certificates('ca_bundle.pem')
for ca in ca_certs:
    print(ca.get_subject())
    store.add_cert(ca)
    
# customize Common Name (CN) verification function
def is_match_cn(cert):
    verified = False
    for ca in ca_certs:
        # print(get_CN_prefix(cert))
        # print(get_CN_prefix(ca))
        verified = verified or get_CN_prefix(cert) in get_CN_prefix(ca)
    return verified

# get CN from certificate struct        
def get_CN_prefix(cert):
    for component in cert.get_subject().get_components():
        if component[0] == b'CN' :
            return component[1]
    return "NO_CN"

def is_valid_certificate(cert):
    not_before_str = cert.get_notBefore().decode('utf-8')
    not_after_str = cert.get_notAfter().decode('utf-8')

    not_before = datetime.datetime.strptime(not_before_str, "%Y%m%d%H%M%SZ")
    not_after = datetime.datetime.strptime(not_after_str, "%Y%m%d%H%M%SZ")

    current_time = datetime.datetime.utcnow()

    if not_before <= current_time <= not_after:
        print('Certificate is valid.')
        return True
    else:
        print('Certificate is not valid.')
        return False

# define middleware to check the request          
@app.before_request
def check_client_cert():
    client_cert = request.environ.get("SSL_CLIENT_CERT")
    if client_cert:
        cert = load_certificate(FILETYPE_PEM, client_cert)
        store_context = X509StoreContext(store, cert)

        # check current timestamp is valid between valid date range
     if not is_valid_certificate(cert):
        return "Invalid certificate", 403

        # check client cert is match trusted certificate's CN
        if not is_match_cn(cert):
         return "Invalid client certificate", 403
    else:
        return "Client certificate required.", 401