PAYATHON 2026

如何在 mitmproxy addon 中根据 hostname 拦截 TLS 连接并动态提供证书?

技术老齐

结论

在 tls_clienthello 中读取 SNI,决定是否拦截连接。需要透传时,设置 data.ignore_connection = True。

动态提供证书有两种方式:

  • 将生成的证书和私钥加入内置 tlsconfig 的证书存储,由 mitmproxy 完成后续握手。如果已经通过 --certs 验证成功,这种方式更容易保留默认 TLS 配置。
  • 在 tls_start_client 中设置自己的 data.ssl_conn,接管客户端侧的 TLS 配置。

Client.mitmcert 记录 mitmproxy 使用的证书,不能通过给它赋值来配置握手证书。证书通过 TLS 的 Certificate 消息发送,握手消息由 OpenSSL 和 mitmproxy 处理,无须手工发送 ServerHello。TLS hook 文档、默认 TLS 实现

两个 hook 分别负责什么

tls_clienthello 接收 tls.ClientHelloData 参数,可以通过以下字段读取客户端在 ClientHello 中声明的 hostname:

data.client_hello.sni

这个值可能是 None,因为客户端不一定会发送 SNI。设置 ignore_connection = True 后,mitmproxy 会原样转发加密数据,不再解密该连接。ClientHelloData 文档

tls_start_client 接收 tls.TlsData 参数。其中,data.context 是连接上下文,data.ssl_conn 是实际参与握手的 pyOpenSSL SSL.Connection。默认情况下,内置 tlsconfig addon 会创建这个对象。TlsData 文档

如果只需要按 hostname 决定是否拦截,可以直接使用 mitmproxy 的动态证书签发能力,无须再写证书生成逻辑。需要自定义证书字段、有效期或签发规则时,再自行生成证书。证书机制说明

示例:生成证书后交给默认 TLS 配置使用

下面的 addon 拦截 example.com 及其子域名,直接透传其他 hostname 和没有 SNI 的连接。匹配的 hostname 首次出现时,addon 使用 mitmproxy 当前的本地 CA 签发证书,并将证书和对应私钥注册到证书存储。

这种方式使用内置 addon 的 certstore,依赖 mitmproxy 的内部实现。部署时应固定 mitmproxy 版本,核对该版本的 CertStoreEntry、add_cert() 和 tlsconfig 实现,不能假定所有历史版本都兼容。当前实现可参考 certs.py 和 tlsconfig.py。

将代码保存为 dynamic_cert.py:

from datetime import datetime, timedelta, timezone

from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import ExtendedKeyUsageOID

from mitmproxy import certs, ctx, tls


class DynamicCertificate:
    def __init__(self):
        self.cache = {}

    @staticmethod
    def should_intercept(hostname: str) -> bool:
        return (
            hostname == "example.com"
            or hostname.endswith(".example.com")
        )

    @staticmethod
    def issue_certificate(hostname, store):
        # 为叶子证书生成独立私钥。
        key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048,
        )

        ca = store.default_ca.to_cryptography()
        now = datetime.now(timezone.utc)

        leaf = (
            x509.CertificateBuilder()
            # 使用空 subject,并以 critical SAN 指明域名。
            .subject_name(x509.Name([]))
            .issuer_name(ca.subject)
            .public_key(key.public_key())
            .serial_number(x509.random_serial_number())
            .not_valid_before(now - timedelta(minutes=5))
            .not_valid_after(now + timedelta(days=7))
            .add_extension(
                x509.SubjectAlternativeName(
                    [x509.DNSName(hostname)]
                ),
                critical=True,
            )
            .add_extension(
                x509.BasicConstraints(ca=False, path_length=None),
                critical=True,
            )
            .add_extension(
                x509.ExtendedKeyUsage(
                    [ExtendedKeyUsageOID.SERVER_AUTH]
                ),
                critical=False,
            )
            .sign(
                private_key=store.default_privatekey,
                algorithm=hashes.SHA256(),
            )
        )

        return certs.CertStoreEntry(
            cert=certs.Cert(leaf),
            privatekey=key,
            chain_file=None,
            # 假定使用 mitmproxy 默认的本地根 CA,
            # 客户端已信任该根证书,无须发送中间证书。
            chain_certs=[],
        )

    def tls_clienthello(self, data: tls.ClientHelloData):
        raw_hostname = data.client_hello.sni
        hostname = (raw_hostname or "").rstrip(".").lower()

        if not self.should_intercept(hostname):
            data.ignore_connection = True
            return

        tlsconfig = ctx.master.addons.get("tlsconfig")
        if tlsconfig is None or tlsconfig.certstore is None:
            raise RuntimeError("tlsconfig certificate store is unavailable")

        store = tlsconfig.certstore

        # 缓存按 hostname 签发的证书,并检查有效期。
        entry = self.cache.get(hostname)
        if entry is None or entry.cert.has_expired():
            entry = self.issue_certificate(hostname, store)
            self.cache[hostname] = entry

        # 同时注册原始 SNI 和规范化后的名称。
        # 每次注册也能应对 certstore 被重新创建的情况。
        store.add_cert(entry, raw_hostname, hostname)


addons = [DynamicCertificate()]

代码使用 cryptography 构建证书,mitmproxy 仍负责证书选择、pyOpenSSL 连接创建和握手。cryptography X.509 示例

使用以下命令启动:

mitmproxy -s dynamic_cert.py \
  --set upstream_cert=false \
  --set connection_strategy=lazy

设置 upstream_cert=false 后,默认的证书名称选择不再混入上游证书的 CN 和 SAN,便于根据刚注册的 SNI 精确选中证书。示例不需要读取上游证书,因此采用 lazy 连接策略。不要为相同 hostname 同时配置其他 --certs 规则,以免发生覆盖或选择冲突。证书选择实现

什么时候直接使用 tls_start_client

如果同一个 hostname 的不同连接需要使用不同证书,或需要全面定制客户端侧 TLS 参数,可以在 tls_start_client 中设置 data.ssl_conn。假定 leaf_cert 和 leaf_key 已经生成,握手配置的核心操作如下:

from OpenSSL import SSL

ssl_context = SSL.Context(SSL.TLS_SERVER_METHOD)
ssl_context.use_certificate(leaf_cert)
ssl_context.use_privatekey(leaf_key)
ssl_context.check_privatekey()

ssl_conn = SSL.Connection(ssl_context, None)
ssl_conn.set_accept_state()

data.ssl_conn = ssl_conn

这段代码只包含握手配置的核心操作,尚不构成完整的 HTTPS addon。替换 data.ssl_conn 后,还需要配置证书链、TLS 版本、密码套件以及适合目标协议的 ALPN。新建的空白 SSL.Context 不会自动继承内置 tlsconfig 的这些设置。当前 pyOpenSSL API 可以接收 cryptography 的证书和私钥对象,使用旧依赖环境时需要核对兼容性。pyOpenSSL API

同时需要检查 hook 的执行顺序。默认 tlsconfig 发现 data.ssl_conn 已有值时,会跳过初始化;如果它已经先运行,自定义 addon 就需要主动替换现有对象。两种顺序都要求在握手开始前完成配置。默认 tls_start_client 实现

使用时需要注意

  • 客户端必须信任实际签发证书的 CA。上面的完整示例使用 mitmproxy 当前的本地 CA。向客户端分发时,应提供 CA 证书,不能提供包含 CA 私钥的文件。证书固定校验仍可能导致连接失败。证书安装与限制
  • 证书必须包含与 hostname 匹配的 SAN,私钥也必须与证书公钥对应。代码中的拦截规则匹配任意深度的子域名,签发的证书则精确匹配各个 hostname。
  • SNI 是客户端声明的目标名称,不等同于后续 HTTP Host。没有 SNI,或目标名称被 ECH 隐藏时,无法通过这个字段获得真实 hostname。
  • 证书存储按名称共享,适合一个 hostname 使用一套证书的策略。如果证书取决于客户端身份,应考虑在 tls_start_client 中为每条连接单独配置。
  • 示例缓存没有容量上限,首次生成 RSA 密钥也会占用事件处理时间。长期运行时,需要补充缓存淘汰、续期和生成失败处理。使用中间 CA 时,还需提供正确的中间证书链。

备注:内容仅供参考。