Flask+Gunicorn+MTLS场景下，如何将客户端证书传递至Flask接口？

以下是几种无需Nginx的可行方案，直接基于Gunicorn和Flask实现客户端证书的获取：

方案1：直接读取WSGI环境变量 #

Gunicorn在处理MTLS请求时，会将客户端证书相关信息存入WSGI环境变量中，Flask可以直接通过request.environ获取这些信息。

示例Flask视图代码：

from flask import Flask, request

app = Flask(__name__)

@app.route('/validate-cert')
def validate_cert():
    # 获取PEM格式的客户端证书
    client_cert = request.environ.get('SSL_CLIENT_CERT')
    # 获取客户端证书的DN（区分名）
    client_dn = request.environ.get('SSL_CLIENT_S_DN')
    # 获取证书校验结果
    cert_verify = request.environ.get('SSL_CLIENT_VERIFY')

    if not client_cert or cert_verify != 'SUCCESS':
        return "Invalid or missing client certificate", 403

    # 这里可以添加证书的自定义校验逻辑，比如检查DN、有效期等
    return f"Authenticated client: {client_dn}", 200

注意：确保Gunicorn启动时配置了正确的MTLS参数，例如：

gunicorn --certfile=server.crt --keyfile=server.key --ca-certs=ca.crt --cert-reqs=REQUIRED myapp:app

其中--cert-reqs=REQUIRED会强制要求客户端提供合法证书，否则Gunicorn直接拒绝请求。

方案2：通过Gunicorn钩子传递证书到请求头 #

如果需要更灵活的处理，可以通过Gunicorn的on_request钩子将证书信息编码后添加到请求头，方便Flask读取。

创建Gunicorn配置文件gunicorn.conf.py：

import base64

def on_request(request, environ):
    client_cert = environ.get('SSL_CLIENT_CERT')
    if client_cert:
        # 对证书进行Base64编码，避免换行符导致请求头异常
        encoded_cert = base64.b64encode(client_cert.encode('utf-8')).decode('utf-8')
        environ['HTTP_X_CLIENT_CERT'] = encoded_cert
    return request, environ

启动Gunicorn时指定配置文件：

gunicorn --config gunicorn.conf.py --certfile=server.crt --keyfile=server.key --ca-certs=ca.crt --cert-reqs=REQUIRED myapp:app

Flask中读取证书：

import base64

@app.route('/validate-cert')
def validate_cert():
    encoded_cert = request.headers.get('X-Client-Cert')
    if not encoded_cert:
        return "Missing client certificate", 403
    
    # 解码Base64证书
    client_cert = base64.b64decode(encoded_cert).decode('utf-8')
    # 后续校验逻辑...
    return "Certificate received", 200

方案3：自定义Gunicorn Worker获取证书 #

如果默认Gunicorn Worker未将证书信息存入WSGI环境，可以自定义Worker类，直接从客户端连接中提取证书。

创建自定义Worker类（例如在myapp.py中）：

from gunicorn.workers.sync import SyncWorker
import ssl

class MTLSWorker(SyncWorker):
    def handle_request(self, listener, req, client, addr):
        # 从客户端连接获取证书（DER格式）
        der_cert = client.getpeercert(True)
        if der_cert:
            # 转换为PEM格式
            pem_cert = ssl.DER_cert_to_PEM_cert(der_cert)
            # 存入WSGI环境
            req.environ['SSL_CLIENT_CERT'] = pem_cert
            # 提取证书DN信息
            cert_details = client.getpeercert()
            req.environ['SSL_CLIENT_S_DN'] = ', '.join([f"{k}={v}" for k, v in cert_details['subject'][0]])
        
        # 调用父类方法处理请求
        super().handle_request(listener, req, client, addr)

启动Gunicorn时指定自定义Worker：

gunicorn --worker-class=myapp.MTLSWorker --certfile=server.crt --keyfile=server.key --ca-certs=ca.crt --cert-reqs=REQUIRED myapp:app

Flask中读取证书的方式同方案1。

关键注意事项 #

• 确认Gunicorn版本兼容性：部分旧版本Gunicorn可能对WSGI环境中的SSL变量命名不同，可通过打印request.environ查看具体字段。
• 证书校验逻辑：获取证书后，建议在Flask中补充二次校验（例如检查证书有效期、所属CA等），增强安全性。
• 性能考量：自定义Worker或钩子会增加请求处理的少量开销，高并发场景建议测试性能影响。

内容的提问来源于stack exchange，提问作者juan carlos