mitmproxy.certs

  1import contextlib
  2import datetime
  3import ipaddress
  4import logging
  5import os
  6import sys
  7import warnings
  8from collections.abc import Iterable
  9from dataclasses import dataclass
 10from pathlib import Path
 11from typing import cast
 12from typing import NewType
 13from typing import Optional
 14from typing import Union
 15
 16import OpenSSL
 17from cryptography import x509
 18from cryptography.hazmat.primitives import hashes
 19from cryptography.hazmat.primitives import serialization
 20from cryptography.hazmat.primitives.asymmetric import dsa
 21from cryptography.hazmat.primitives.asymmetric import ec
 22from cryptography.hazmat.primitives.asymmetric import rsa
 23from cryptography.hazmat.primitives.asymmetric.types import CertificatePublicKeyTypes
 24from cryptography.hazmat.primitives.serialization import pkcs12
 25from cryptography.x509 import ExtendedKeyUsageOID
 26from cryptography.x509 import NameOID
 27
 28from mitmproxy.coretypes import serializable
 29
 30if sys.version_info < (3, 13):  # pragma: no cover
 31    from typing_extensions import deprecated
 32else:
 33    from warnings import deprecated
 34
 35logger = logging.getLogger(__name__)
 36
 37# Default expiry must not be too long: https://github.com/mitmproxy/mitmproxy/issues/815
 38# Note that expiry will be offset by `CERT_VALIDITY_OFFSET`, i.e. the cert will be
 39# backdated a bit to account for clients with incorrect clocks.
 40CA_EXPIRY = datetime.timedelta(days=10 * 365)
 41CERT_EXPIRY = datetime.timedelta(days=199)
 42CRL_EXPIRY = datetime.timedelta(days=7)
 43
 44CERT_VALIDITY_OFFSET = datetime.timedelta(days=-2)
 45
 46# Generated with "openssl dhparam". It's too slow to generate this on startup.
 47DEFAULT_DHPARAM = b"""
 48-----BEGIN DH PARAMETERS-----
 49MIICCAKCAgEAyT6LzpwVFS3gryIo29J5icvgxCnCebcdSe/NHMkD8dKJf8suFCg3
 50O2+dguLakSVif/t6dhImxInJk230HmfC8q93hdcg/j8rLGJYDKu3ik6H//BAHKIv
 51j5O9yjU3rXCfmVJQic2Nne39sg3CreAepEts2TvYHhVv3TEAzEqCtOuTjgDv0ntJ
 52Gwpj+BJBRQGG9NvprX1YGJ7WOFBP/hWU7d6tgvE6Xa7T/u9QIKpYHMIkcN/l3ZFB
 53chZEqVlyrcngtSXCROTPcDOQ6Q8QzhaBJS+Z6rcsd7X+haiQqvoFcmaJ08Ks6LQC
 54ZIL2EtYJw8V8z7C0igVEBIADZBI6OTbuuhDwRw//zU1uq52Oc48CIZlGxTYG/Evq
 55o9EWAXUYVzWkDSTeBH1r4z/qLPE2cnhtMxbFxuvK53jGB0emy2y1Ei6IhKshJ5qX
 56IB/aE7SSHyQ3MDHHkCmQJCsOd4Mo26YX61NZ+n501XjqpCBQ2+DfZCBh8Va2wDyv
 57A2Ryg9SUz8j0AXViRNMJgJrr446yro/FuJZwnQcO3WQnXeqSBnURqKjmqkeFP+d8
 586mk2tqJaY507lRNqtGlLnj7f5RNoBFJDCLBNurVgfvq9TCVWKDIFD4vZRjCrnl6I
 59rD693XKIHUCWOjMh1if6omGXKHH40QuME2gNa50+YPn1iYDl88uDbbMCAQI=
 60-----END DH PARAMETERS-----
 61"""
 62
 63# Compromised eMudhra intermediate CA (EM OV TLS CA - G3C, d=4), see https://emudhra.we-have-a-problem.com/.
 64# Contains:
 65# 1. Private key for EM OV TLS CA - G3C (P-384, d=4).
 66# 2. EM OV TLS CA - G3C issued by emSign Root TLS CA - G3.
 67# 3. emSign Root TLS CA - G3 cross-signed by emSign ECC Root CA - G3.
 68# 4. EM OV TLS CA - G3C cross-signed by Sectigo's AAA Certificate Services (for older pre-2021 trust stores).
 69#
 70# Path-building TLS stacks (Go, Rust, Java, macOS, Windows, browsers) consider all certificates in the chain
 71# and validate against any of the three roots (emSign ECC Root CA - G3, emSign Root TLS CA - G3, or
 72# AAA Certificate Services). OpenSSL does not backtrack across untrusted intermediates with the same
 73# Subject/SKI and only follows the first matching intermediate: --emudhra keeps the emSign-issued intermediate
 74# first (since Mozilla's root bundle used by Python/certifi, curl, Node.js, and Linux ca-certificates already
 75# removed AAA Certificate Services in 2025), while --emudhra-old moves the AAA-issued intermediate (#4) to the
 76# front for older OpenSSL trust stores.
 77EMUDHRA_CHAIN_PEM = b"""
 78-----BEGIN PRIVATE KEY-----
 79MIG2AgEAMBAGByqGSM49AgEGBSuBBAAiBIGeMIGbAgEBBDAAAAAAAAAAAAAAAAAA
 80AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAShZANiAAQTglHNUqyS
 81mMHIqtl3Mh3rl+cJvQtMoKylXcitUdz8nRWJoVl+OlEg4e/WMcY+GDXKyuKYaaYu
 82FjHoooGBq1ZhbcRdkYq8CfOrDmPPeSqk3O1zh743u6VpVJ8cArJw7Wc=
 83-----END PRIVATE KEY-----
 84-----BEGIN CERTIFICATE-----
 85MIIDaDCCAu6gAwIBAgIPKWkmwiejoVJjkrGUgqUEMAoGCCqGSM49BAMDMFYxCzAJ
 86BgNVBAYTAklOMSUwIwYDVQQKDBxlTXVkaHJhIFRlY2hub2xvZ2llcyBMaW1pdGVk
 87MSAwHgYDVQQDDBdlbVNpZ24gUm9vdCBUTFMgQ0EgLSBHMzAeFw0yNTA0MzAxNTM3
 88NDFaFw00MDA0MjgxODMwMDBaMEAxCzAJBgNVBAYTAlVTMRQwEgYDVQQKDAtlTXVk
 89aHJhIEluYzEbMBkGA1UEAwwSRU0gT1YgVExTIENBIC0gRzNDMHYwEAYHKoZIzj0C
 90AQYFK4EEACIDYgAEE4JRzVKskpjByKrZdzId65fnCb0LTKCspV3IrVHc/J0ViaFZ
 91fjpRIOHv1jHGPhg1ysrimGmmLhYx6KKBgatWYW3EXZGKvAnzqw5jz3kqpNztc4e+
 92N7ulaVSfHAKycO1no4IBljCCAZIwHwYDVR0jBBgwFoAU4TqrIXzx401UrtRyt/2L
 93WUIgkFQwHQYDVR0OBBYEFDTq6gzryQra+dDF4u2beMmHnhtsMB0GA1UdJQQWMBQG
 94CCsGAQUFBwMBBggrBgEFBQcDAjBdBgNVHSAEVjBUMDkGCysGAQQBg44hAQABMCow
 95KAYIKwYBBQUHAgEWHGh0dHA6Ly9yZXBvc2l0b3J5LmVtc2lnbi5jb20wDQYLKwYB
 96BAGDjiEBAm4wCAYGZ4EMAQICMHUGCCsGAQUFBwEBBGkwZzAkBggrBgEFBQcwAYYY
 97aHR0cDovL29jc3AtYy5lbXNpZ24uY29tMD8GCCsGAQUFBzAChjNodHRwOi8vcmVw
 98b3NpdG9yeS5lbXNpZ24uY29tL2NlcnRzL0VNT1ZUTFNDQUczQy5wN2MwNwYDVR0f
 99BDAwLjAsoCqgKIYmaHR0cDovL2NybC5lbXNpZ24uY29tLz9yb290dGxzY2FnMy5j
100cmwwDgYDVR0PAQH/BAQDAgEGMBIGA1UdEwEB/wQIMAYBAf8CAQAwCgYIKoZIzj0E
101AwMDaAAwZQIxAMSZt2xsPeumRdeSMSQ3CbzIZeQ/q7GDCWfeE1Pq+mhgA6gRM44c
102NVF3xJ+wsxJk+wIwFJbaDMokD8Ei6eNOiA8yUMYvfzQyngmrhZQMdPCK1SewrR79
1034rSb9vGqxZzelQth
104-----END CERTIFICATE-----
105-----BEGIN CERTIFICATE-----
106MIIDdDCCAvmgAwIBAgIQAJYaPvsTWVdu0EAFtGhKWzAKBggqhkjOPQQDAzBrMQsw
107CQYDVQQGEwJJTjETMBEGA1UECxMKZW1TaWduIFBLSTElMCMGA1UEChMcZU11ZGhy
108YSBUZWNobm9sb2dpZXMgTGltaXRlZDEgMB4GA1UEAxMXZW1TaWduIEVDQyBSb290
109IENBIC0gRzMwHhcNMjQwNzExMTU0MzM2WhcNMzkwNzA5MTgzMDAwWjBWMQswCQYD
110VQQGEwJJTjElMCMGA1UECgwcZU11ZGhyYSBUZWNobm9sb2dpZXMgTGltaXRlZDEg
111MB4GA1UEAwwXZW1TaWduIFJvb3QgVExTIENBIC0gRzMwdjAQBgcqhkjOPQIBBgUr
112gQQAIgNiAARfD9w/q1cWxL6D9bMGH6CQ4u+gAV6GN9s+PwQx9Bl4hC+0iU8QOk8m
113WO/RwXQy10DCSxT8vSIJClHKp+Ao4qJKdYeV0SwZ7LMQziyLfxIUOY6Grja/sQFw
114LKYf8EdDXAyjggF1MIIBcTAfBgNVHSMEGDAWgBR8XQKEE9TMipuBzhccLikenEhj
115QjAdBgNVHQ4EFgQU4TqrIXzx401UrtRyt/2LWUIgkFQwHQYDVR0lBBYwFAYIKwYB
116BQUHAwEGCCsGAQUFBwMCMD0GA1UdIAQ2MDQwMgYEVR0gADAqMCgGCCsGAQUFBwIB
117FhxodHRwOi8vcmVwb3NpdG9yeS5lbXNpZ24uY29tMHgGCCsGAQUFBwEBBGwwajAi
118BggrBgEFBQcwAYYWaHR0cDovL29jc3AuZW1zaWduLmNvbTBEBggrBgEFBQcwAoY4
119aHR0cDovL3JlcG9zaXRvcnkuZW1zaWduLmNvbS9jZXJ0cy9lbVNpZ25FQ0NSb290
120Q0FHMy5jcnQwMwYDVR0fBCwwKjAooCagJIYiaHR0cDovL2NybC5lbXNpZ24uY29t
121P1Jvb3RDQUczLmNybDAOBgNVHQ8BAf8EBAMCAQYwEgYDVR0TAQH/BAgwBgEB/wIB
122ATAKBggqhkjOPQQDAwNpADBmAjEAt9CIpRGp46QHQYDUBnLKw+vvK0/h9vmilEAg
123TroUl/anH4x2rXHqiCtBTBfgKDegAjEA8WpG1LTZdAIL0FSYqxXeREaeTSxEofM0
1241jJH3jGvCXC79LU9UPJ2o3TU+UIj2gMV
125-----END CERTIFICATE-----
126-----BEGIN CERTIFICATE-----
127MIIDzzCCAregAwIBAgIRAI1aSBGTnvRH13ERzdmEmukwDQYJKoZIhvcNAQELBQAw
128ezELMAkGA1UEBhMCR0IxGzAZBgNVBAgMEkdyZWF0ZXIgTWFuY2hlc3RlcjEQMA4G
129A1UEBwwHU2FsZm9yZDEaMBgGA1UECgwRQ29tb2RvIENBIExpbWl0ZWQxITAfBgNV
130BAMMGEFBQSBDZXJ0aWZpY2F0ZSBTZXJ2aWNlczAeFw0yNTA1MTQwMDAwMDBaFw0y
131ODEyMzEyMzU5NTlaMEAxCzAJBgNVBAYTAlVTMRQwEgYDVQQKDAtlTXVkaHJhIElu
132YzEbMBkGA1UEAwwSRU0gT1YgVExTIENBIC0gRzNDMHYwEAYHKoZIzj0CAQYFK4EE
133ACIDYgAEE4JRzVKskpjByKrZdzId65fnCb0LTKCspV3IrVHc/J0ViaFZfjpRIOHv
1341jHGPhg1ysrimGmmLhYx6KKBgatWYW3EXZGKvAnzqw5jz3kqpNztc4e+N7ulaVSf
135HAKycO1no4IBNTCCATEwHwYDVR0jBBgwFoAUoBEKIz6W8Qfs4q8p74Klf9AwpLQw
136HQYDVR0OBBYEFDTq6gzryQra+dDF4u2beMmHnhtsMA4GA1UdDwEB/wQEAwIBBjAS
137BgNVHRMBAf8ECDAGAQH/AgEAMB0GA1UdJQQWMBQGCCsGAQUFBwMBBggrBgEFBQcD
138AjAxBgNVHSAEKjAoMAgGBmeBDAECAjANBgsrBgEEAYOOIQEAATANBgsrBgEEAYOO
139IQECbjBDBgNVHR8EPDA6MDigNqA0hjJodHRwOi8vY3JsLmNvbW9kb2NhLmNvbS9B
140QUFDZXJ0aWZpY2F0ZVNlcnZpY2VzLmNybDA0BggrBgEFBQcBAQQoMCYwJAYIKwYB
141BQUHMAGGGGh0dHA6Ly9vY3NwLmNvbW9kb2NhLmNvbTANBgkqhkiG9w0BAQsFAAOC
142AQEAHlbhZpjhkPEAQVhPHRtBDaHqiisVD20pQ1TUeSx3l62Qe7NQ0b7vWQy7DN0i
143KOu7BGpP5EnIjvXbz9FhpfPJ4FhKEt3tQ51rqXz2oD8vRaryvExWPobDUz3j7HN6
144OzBsVMIe3PD7AAkpky0x8zMSi/ep5R9OMzybmNptbtppyuB5ytgq+h7lQdRIP+CQ
145TaJX3qTgfMqzntYre5GljmEB5hU2/rSn8zoSU2APkJidlZFpghtgbD4I2VRHZvCw
146FFr/SrFM9pwrgbJNZ1MtyTSVgI5NlNb0aMfLR+/r6jJt3ROeY2yh2QQ2dx+Yqgdb
147uL5ChFWOQU+ErSDnUq7yr+pjBQ==
148-----END CERTIFICATE-----
149"""
150
151
152class Cert(serializable.Serializable):
153    """Representation of a (TLS) certificate."""
154
155    _cert: x509.Certificate
156
157    def __init__(self, cert: x509.Certificate):
158        assert isinstance(cert, x509.Certificate)
159        self._cert = cert
160
161    def __eq__(self, other):
162        return self.fingerprint() == other.fingerprint()
163
164    def __repr__(self):
165        altnames = [str(x.value) for x in self.altnames]
166        return f"<Cert(cn={self.cn!r}, altnames={altnames!r})>"
167
168    def __hash__(self):
169        return self._cert.__hash__()
170
171    @classmethod
172    def from_state(cls, state):
173        return cls.from_pem(state)
174
175    def get_state(self):
176        return self.to_pem()
177
178    def set_state(self, state):
179        self._cert = x509.load_pem_x509_certificate(state)
180
181    @classmethod
182    def from_pem(cls, data: bytes) -> "Cert":
183        cert = x509.load_pem_x509_certificate(data)  # type: ignore
184        return cls(cert)
185
186    def to_pem(self) -> bytes:
187        return self._cert.public_bytes(serialization.Encoding.PEM)
188
189    @classmethod
190    def from_pyopenssl(self, x509: OpenSSL.crypto.X509) -> "Cert":
191        return Cert(x509.to_cryptography())
192
193    @deprecated("Use `to_cryptography` instead.")
194    def to_pyopenssl(self) -> OpenSSL.crypto.X509:  # pragma: no cover
195        return OpenSSL.crypto.X509.from_cryptography(self._cert)
196
197    def to_cryptography(self) -> x509.Certificate:
198        return self._cert
199
200    def public_key(self) -> CertificatePublicKeyTypes:
201        return self._cert.public_key()
202
203    def fingerprint(self) -> bytes:
204        return self._cert.fingerprint(hashes.SHA256())
205
206    @property
207    def issuer(self) -> list[tuple[str, str]]:
208        return _name_to_keyval(self._cert.issuer)
209
210    @property
211    def notbefore(self) -> datetime.datetime:
212        try:
213            # type definitions haven't caught up with new API yet.
214            return self._cert.not_valid_before_utc  # type: ignore
215        except AttributeError:  # pragma: no cover
216            # cryptography < 42.0
217            return self._cert.not_valid_before.replace(tzinfo=datetime.UTC)
218
219    @property
220    def notafter(self) -> datetime.datetime:
221        try:
222            return self._cert.not_valid_after_utc  # type: ignore
223        except AttributeError:  # pragma: no cover
224            return self._cert.not_valid_after.replace(tzinfo=datetime.UTC)
225
226    def has_expired(self) -> bool:
227        if sys.version_info < (3, 11):  # pragma: no cover
228            return datetime.datetime.now(datetime.UTC) > self.notafter
229        return datetime.datetime.now(datetime.UTC) > self.notafter
230
231    @property
232    def subject(self) -> list[tuple[str, str]]:
233        return _name_to_keyval(self._cert.subject)
234
235    @property
236    def serial(self) -> int:
237        return self._cert.serial_number
238
239    @property
240    def is_ca(self) -> bool:
241        constraints: x509.BasicConstraints
242        try:
243            constraints = self._cert.extensions.get_extension_for_class(
244                x509.BasicConstraints
245            ).value
246            return constraints.ca
247        except x509.ExtensionNotFound:
248            return False
249
250    @property
251    def keyinfo(self) -> tuple[str, int]:
252        public_key = self._cert.public_key()
253        if isinstance(public_key, rsa.RSAPublicKey):
254            return "RSA", public_key.key_size
255        if isinstance(public_key, dsa.DSAPublicKey):
256            return "DSA", public_key.key_size
257        if isinstance(public_key, ec.EllipticCurvePublicKey):
258            return f"EC ({public_key.curve.name})", public_key.key_size
259        return (
260            public_key.__class__.__name__.replace("PublicKey", "").replace("_", ""),
261            getattr(public_key, "key_size", -1),
262        )  # pragma: no cover
263
264    @property
265    def cn(self) -> str | None:
266        attrs = self._cert.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)
267        if attrs:
268            return cast(str, attrs[0].value)
269        return None
270
271    @property
272    def organization(self) -> str | None:
273        attrs = self._cert.subject.get_attributes_for_oid(
274            x509.NameOID.ORGANIZATION_NAME
275        )
276        if attrs:
277            return cast(str, attrs[0].value)
278        return None
279
280    @property
281    def altnames(self) -> x509.GeneralNames:
282        """
283        Get all SubjectAlternativeName DNS altnames.
284        """
285        try:
286            sans = self._cert.extensions.get_extension_for_class(
287                x509.SubjectAlternativeName
288            ).value
289        except x509.ExtensionNotFound:
290            return x509.GeneralNames([])
291        else:
292            return x509.GeneralNames(sans)
293
294    @property
295    def crl_distribution_points(self) -> list[str]:
296        try:
297            ext = self._cert.extensions.get_extension_for_class(
298                x509.CRLDistributionPoints
299            ).value
300        except x509.ExtensionNotFound:
301            return []
302        else:
303            return [
304                dist_point.full_name[0].value
305                for dist_point in ext
306                if dist_point.full_name
307                and isinstance(dist_point.full_name[0], x509.UniformResourceIdentifier)
308            ]
309
310
311def _name_to_keyval(name: x509.Name) -> list[tuple[str, str]]:
312    parts = []
313    for attr in name:
314        k = attr.rfc4514_string().partition("=")[0]
315        v = cast(str, attr.value)
316        parts.append((k, v))
317    return parts
318
319
320def create_ca(
321    organization: str,
322    cn: str,
323    key_size: int,
324) -> tuple[rsa.RSAPrivateKeyWithSerialization, x509.Certificate]:
325    now = datetime.datetime.now()
326
327    private_key = rsa.generate_private_key(
328        public_exponent=65537,
329        key_size=key_size,
330    )  # type: ignore
331    name = x509.Name(
332        [
333            x509.NameAttribute(NameOID.COMMON_NAME, cn),
334            x509.NameAttribute(NameOID.ORGANIZATION_NAME, organization),
335        ]
336    )
337    builder = x509.CertificateBuilder()
338    builder = builder.serial_number(x509.random_serial_number())
339    builder = builder.subject_name(name)
340    builder = builder.not_valid_before(now + CERT_VALIDITY_OFFSET)
341    builder = builder.not_valid_after(now + CERT_VALIDITY_OFFSET + CA_EXPIRY)
342    builder = builder.issuer_name(name)
343    builder = builder.public_key(private_key.public_key())
344    builder = builder.add_extension(
345        x509.BasicConstraints(ca=True, path_length=None), critical=True
346    )
347    builder = builder.add_extension(
348        x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), critical=False
349    )
350    builder = builder.add_extension(
351        x509.KeyUsage(
352            digital_signature=False,
353            content_commitment=False,
354            key_encipherment=False,
355            data_encipherment=False,
356            key_agreement=False,
357            key_cert_sign=True,
358            crl_sign=True,
359            encipher_only=False,
360            decipher_only=False,
361        ),
362        critical=True,
363    )
364    builder = builder.add_extension(
365        x509.SubjectKeyIdentifier.from_public_key(private_key.public_key()),
366        critical=False,
367    )
368    cert = builder.sign(private_key=private_key, algorithm=hashes.SHA256())  # type: ignore
369    return private_key, cert
370
371
372def _fix_legacy_sans(sans: Iterable[x509.GeneralName] | list[str]) -> x509.GeneralNames:
373    """
374    SANs used to be a list of strings in mitmproxy 10.1 and below, but now they're a list of GeneralNames.
375    This function converts the old format to the new one.
376    """
377    if isinstance(sans, x509.GeneralNames):
378        return sans
379    elif (
380        isinstance(sans, list) and len(sans) > 0 and isinstance(sans[0], str)
381    ):  # pragma: no cover
382        warnings.warn(
383            "Passing SANs as a list of strings is deprecated.",
384            DeprecationWarning,
385            stacklevel=2,
386        )
387
388        ss: list[x509.GeneralName] = []
389        for x in cast(list[str], sans):
390            try:
391                ip = ipaddress.ip_address(x)
392            except ValueError:
393                x = x.encode("idna").decode()
394                ss.append(x509.DNSName(x))
395            else:
396                ss.append(x509.IPAddress(ip))
397        return x509.GeneralNames(ss)
398    else:
399        return x509.GeneralNames(cast(Iterable[x509.GeneralName], sans))
400
401
402def dummy_cert(
403    privkey: rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey,
404    cacert: x509.Certificate,
405    commonname: str | None,
406    sans: Iterable[x509.GeneralName],
407    organization: str | None = None,
408    crl_url: str | None = None,
409) -> Cert:
410    """
411    Generates a dummy certificate.
412
413    privkey: CA private key
414    cacert: CA certificate
415    commonname: Common name for the generated certificate.
416    sans: A list of Subject Alternate Names.
417    organization: Organization name for the generated certificate.
418    crl_url: URL of CRL distribution point
419
420    Returns cert if operation succeeded, None if not.
421    """
422    builder = x509.CertificateBuilder()
423    builder = builder.issuer_name(cacert.subject)
424    builder = builder.add_extension(
425        x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), critical=False
426    )
427    builder = builder.public_key(cacert.public_key())
428
429    now = datetime.datetime.now()
430    builder = builder.not_valid_before(now + CERT_VALIDITY_OFFSET)
431    builder = builder.not_valid_after(now + CERT_VALIDITY_OFFSET + CERT_EXPIRY)
432
433    subject = []
434    is_valid_commonname = commonname is not None and len(commonname) < 64
435    if is_valid_commonname:
436        assert commonname is not None
437        subject.append(x509.NameAttribute(NameOID.COMMON_NAME, commonname))
438    if organization is not None:
439        assert organization is not None
440        subject.append(x509.NameAttribute(NameOID.ORGANIZATION_NAME, organization))
441    builder = builder.subject_name(x509.Name(subject))
442    builder = builder.serial_number(x509.random_serial_number())
443
444    sans = _fix_legacy_sans(sans)
445    if sans:
446        # RFC 5280 §4.2.1.6: subjectAltName is critical if subject is empty.
447        builder = builder.add_extension(
448            x509.SubjectAlternativeName(sans),
449            critical=not is_valid_commonname,
450        )
451
452    # https://datatracker.ietf.org/doc/html/rfc5280#section-4.2.1.1
453    # Per RFC 5280 §4.2.1.2, the AKI's keyIdentifier in a child certificate
454    # MUST be byte-equal to the issuer's stored SubjectKeyIdentifier. If we
455    # recompute it from the public key with `from_issuer_public_key()`, we
456    # always derive a SHA-1 digest, which mismatches whenever the issuer's
457    # SKI was generated by a different method (RFC 7093 truncated
458    # SHA-256/384/512, hardware-rooted CAs, or any custom value). This
459    # mismatch is rejected by strict TLS chain builders (`X509_V_FLAG_X509_STRICT`,
460    # Python `ssl`, Go `crypto/x509`) with "authority and subject key
461    # identifier mismatch". Most notably, cert-manager >=1.18 / Go >=1.25
462    # default to truncated SHA-256 SKIs for FIPS 140-3 compliance.
463    try:
464        issuer_ski = cacert.extensions.get_extension_for_class(
465            x509.SubjectKeyIdentifier
466        ).value
467        aki = x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(issuer_ski)
468    except x509.ExtensionNotFound:
469        # Fall back when the issuer cert has no SKI extension at all (legacy
470        # CAs predating RFC 5280 conformance).
471        aki = x509.AuthorityKeyIdentifier.from_issuer_public_key(cacert.public_key())  # type: ignore
472    builder = builder.add_extension(aki, critical=False)
473    # If CA and leaf cert have the same Subject Key Identifier, SChannel breaks in funny ways,
474    # see https://github.com/mitmproxy/mitmproxy/issues/6494.
475    # https://datatracker.ietf.org/doc/html/rfc5280#section-4.2.1.2 states
476    # that SKI is optional for the leaf cert, so we skip that.
477
478    if crl_url:
479        builder = builder.add_extension(
480            x509.CRLDistributionPoints(
481                [
482                    x509.DistributionPoint(
483                        [x509.UniformResourceIdentifier(crl_url)],
484                        relative_name=None,
485                        crl_issuer=None,
486                        reasons=None,
487                    )
488                ]
489            ),
490            critical=False,
491        )
492
493    cert = builder.sign(private_key=privkey, algorithm=hashes.SHA256())  # type: ignore
494    return Cert(cert)
495
496
497def dummy_crl(
498    privkey: rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey,
499    cacert: x509.Certificate,
500) -> bytes:
501    """
502    Generates an empty CRL signed with the CA key
503
504    privkey: CA private key
505    cacert: CA certificate
506
507    Returns a CRL DER encoded
508    """
509    builder = x509.CertificateRevocationListBuilder()
510    builder = builder.issuer_name(cacert.issuer)
511
512    now = datetime.datetime.now()
513    builder = builder.last_update(now + CERT_VALIDITY_OFFSET)
514    builder = builder.next_update(now + CERT_VALIDITY_OFFSET + CRL_EXPIRY)
515
516    builder = builder.add_extension(x509.CRLNumber(1000), False)  # meaningless number
517    crl = builder.sign(private_key=privkey, algorithm=hashes.SHA256())
518    return crl.public_bytes(serialization.Encoding.DER)
519
520
521@dataclass(frozen=True)
522class CertStoreEntry:
523    cert: Cert
524    privatekey: rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey
525    chain_file: Path | None
526    chain_certs: list[Cert]
527
528
529TCustomCertId = str  # manually provided certs (e.g. mitmproxy's --certs)
530TGeneratedCertId = tuple[Optional[str], x509.GeneralNames]  # (common_name, sans)
531TCertId = Union[TCustomCertId, TGeneratedCertId]
532
533DHParams = NewType("DHParams", bytes)
534
535
536class CertStore:
537    """
538    Implements an in-memory certificate store.
539    """
540
541    STORE_CAP = 100
542    default_privatekey: rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey
543    default_ca: Cert
544    default_chain_file: Path | None
545    default_chain_certs: list[Cert]
546    dhparams: DHParams
547    certs: dict[TCertId, CertStoreEntry]
548    expire_queue: list[CertStoreEntry]
549
550    def __init__(
551        self,
552        default_privatekey: rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey,
553        default_ca: Cert,
554        default_chain_file: Path | None,
555        default_crl: bytes,
556        dhparams: DHParams,
557    ):
558        self.default_privatekey = default_privatekey
559        self.default_ca = default_ca
560        self.default_chain_file = default_chain_file
561        self.default_crl = default_crl
562        self.default_chain_certs = (
563            [
564                Cert(c)
565                for c in x509.load_pem_x509_certificates(
566                    self.default_chain_file.read_bytes()
567                )
568            ]
569            if self.default_chain_file
570            else [default_ca]
571        )
572        self.dhparams = dhparams
573        self.certs = {}
574        self.expire_queue = []
575
576    def expire(self, entry: CertStoreEntry) -> None:
577        self.expire_queue.append(entry)
578        if len(self.expire_queue) > self.STORE_CAP:
579            d = self.expire_queue.pop(0)
580            self.certs = {k: v for k, v in self.certs.items() if v != d}
581
582    @staticmethod
583    def load_dhparam(path: Path) -> DHParams:
584        # mitmproxy<=0.10 doesn't generate a dhparam file.
585        # Create it now if necessary.
586        if not path.exists():
587            path.write_bytes(DEFAULT_DHPARAM)
588
589        # we could use cryptography for this, but it's unclear how to convert cryptography's object to pyOpenSSL's
590        # expected format.
591        bio = OpenSSL.SSL._lib.BIO_new_file(  # type: ignore
592            str(path).encode(sys.getfilesystemencoding()), b"r"
593        )
594        if bio != OpenSSL.SSL._ffi.NULL:  # type: ignore
595            bio = OpenSSL.SSL._ffi.gc(bio, OpenSSL.SSL._lib.BIO_free)  # type: ignore
596            dh = OpenSSL.SSL._lib.PEM_read_bio_DHparams(  # type: ignore
597                bio,
598                OpenSSL.SSL._ffi.NULL,  # type: ignore
599                OpenSSL.SSL._ffi.NULL,  # type: ignore
600                OpenSSL.SSL._ffi.NULL,  # type: ignore
601            )
602            dh = OpenSSL.SSL._ffi.gc(dh, OpenSSL.SSL._lib.DH_free)  # type: ignore
603            return dh
604        raise RuntimeError("Error loading DH Params.")  # pragma: no cover
605
606    @classmethod
607    def from_store(
608        cls,
609        path: Path | str,
610        basename: str,
611        key_size: int,
612        passphrase: bytes | None = None,
613    ) -> "CertStore":
614        path = Path(path)
615        ca_file = path / f"{basename}-ca.pem"
616        dhparam_file = path / f"{basename}-dhparam.pem"
617        if not ca_file.exists():
618            cls.create_store(path, basename, key_size)
619        return cls.from_files(ca_file, dhparam_file, passphrase)
620
621    @classmethod
622    def from_files(
623        cls, ca_file: Path, dhparam_file: Path, passphrase: bytes | None = None
624    ) -> "CertStore":
625        raw = ca_file.read_bytes()
626        key = load_pem_private_key(raw, passphrase)
627        dh = cls.load_dhparam(dhparam_file)
628        certs = x509.load_pem_x509_certificates(raw)
629        ca = Cert(certs[0])
630        crl = dummy_crl(key, ca._cert)
631        if len(certs) > 1:
632            chain_file: Path | None = ca_file
633        else:
634            chain_file = None
635        return cls(key, ca, chain_file, crl, dh)
636
637    @staticmethod
638    @contextlib.contextmanager
639    def umask_secret():
640        """
641        Context to temporarily set umask to its original value bitor 0o77.
642        Useful when writing private keys to disk so that only the owner
643        will be able to read them.
644        """
645        original_umask = os.umask(0)
646        os.umask(original_umask | 0o77)
647        try:
648            yield
649        finally:
650            os.umask(original_umask)
651
652    @staticmethod
653    def create_store(
654        path: Path, basename: str, key_size: int, organization=None, cn=None
655    ) -> None:
656        path.mkdir(parents=True, exist_ok=True)
657
658        organization = organization or basename
659        cn = cn or basename
660
661        key: rsa.RSAPrivateKeyWithSerialization
662        ca: x509.Certificate
663        key, ca = create_ca(organization=organization, cn=cn, key_size=key_size)
664
665        # Dump the CA plus private key.
666        with CertStore.umask_secret():
667            # PEM format
668            (path / f"{basename}-ca.pem").write_bytes(
669                key.private_bytes(
670                    encoding=serialization.Encoding.PEM,
671                    format=serialization.PrivateFormat.TraditionalOpenSSL,
672                    encryption_algorithm=serialization.NoEncryption(),
673                )
674                + ca.public_bytes(serialization.Encoding.PEM)
675            )
676
677            # PKCS12 format for Windows devices
678            (path / f"{basename}-ca.p12").write_bytes(
679                pkcs12.serialize_key_and_certificates(  # type: ignore
680                    name=basename.encode(),
681                    key=key,
682                    cert=ca,
683                    cas=None,
684                    encryption_algorithm=serialization.NoEncryption(),
685                )
686            )
687
688        # Dump the certificate in PEM format
689        pem_cert = ca.public_bytes(serialization.Encoding.PEM)
690        (path / f"{basename}-ca-cert.pem").write_bytes(pem_cert)
691        # Create a .cer file with the same contents for Android
692        (path / f"{basename}-ca-cert.cer").write_bytes(pem_cert)
693
694        # Dump the certificate in PKCS12 format for Windows devices
695        (path / f"{basename}-ca-cert.p12").write_bytes(
696            pkcs12.serialize_key_and_certificates(
697                name=basename.encode(),
698                key=None,  # type: ignore
699                cert=ca,
700                cas=None,
701                encryption_algorithm=serialization.NoEncryption(),
702            )
703        )
704
705        (path / f"{basename}-dhparam.pem").write_bytes(DEFAULT_DHPARAM)
706
707    def add_cert_file(
708        self, spec: str, path: Path, passphrase: bytes | None = None
709    ) -> None:
710        raw = path.read_bytes()
711        cert = Cert.from_pem(raw)
712        try:
713            private_key = load_pem_private_key(raw, password=passphrase)
714        except ValueError as e:
715            private_key = self.default_privatekey
716            if cert.public_key() != private_key.public_key():
717                raise ValueError(
718                    f'Unable to find private key in "{path.absolute()}": {e}'
719                ) from e
720        else:
721            if cert.public_key() != private_key.public_key():
722                raise ValueError(
723                    f'Private and public keys in "{path.absolute()}" do not match:\n'
724                    f"{cert.public_key()=}\n"
725                    f"{private_key.public_key()=}"
726                )
727
728        try:
729            chain = [Cert(x) for x in x509.load_pem_x509_certificates(raw)]
730        except ValueError as e:
731            logger.warning(f"Failed to read certificate chain: {e}")
732            chain = [cert]
733
734        if cert.is_ca:
735            logger.warning(
736                f'"{path.absolute()}" is a certificate authority and not a leaf certificate. '
737                f"This indicates a misconfiguration, see https://docs.mitmproxy.org/stable/concepts-certificates/."
738            )
739
740        self.add_cert(CertStoreEntry(cert, private_key, path, chain), spec)
741
742    def add_cert(self, entry: CertStoreEntry, *names: str) -> None:
743        """
744        Adds a cert to the certstore. We register the CN in the cert plus
745        any SANs, and also the list of names provided as an argument.
746        """
747        if entry.cert.cn:
748            self.certs[entry.cert.cn] = entry
749        for i in entry.cert.altnames:
750            self.certs[str(i.value)] = entry
751        for i in names:
752            self.certs[i] = entry
753
754    @staticmethod
755    def asterisk_forms(dn: str | x509.GeneralName) -> list[str]:
756        """
757        Return all asterisk forms for a domain. For example, for www.example.com this will return
758        [b"www.example.com", b"*.example.com", b"*.com"]. The single wildcard "*" is omitted.
759        """
760        if isinstance(dn, str):
761            parts = dn.split(".")
762            ret = [dn]
763            for i in range(1, len(parts)):
764                ret.append("*." + ".".join(parts[i:]))
765            return ret
766        elif isinstance(dn, x509.DNSName):
767            return CertStore.asterisk_forms(dn.value)
768        else:
769            return [str(dn.value)]
770
771    def get_cert(
772        self,
773        commonname: str | None,
774        sans: Iterable[x509.GeneralName],
775        organization: str | None = None,
776        crl_url: str | None = None,
777    ) -> CertStoreEntry:
778        """
779        commonname: Common name for the generated certificate. Must be a
780        valid, plain-ASCII, IDNA-encoded domain name.
781
782        sans: A list of Subject Alternate Names.
783
784        organization: Organization name for the generated certificate.
785
786        crl_url: URL of CRL distribution point
787        """
788        sans = _fix_legacy_sans(sans)
789
790        potential_keys: list[TCertId] = []
791        if commonname:
792            potential_keys.extend(self.asterisk_forms(commonname))
793        for s in sans:
794            potential_keys.extend(self.asterisk_forms(s))
795        potential_keys.append("*")
796        potential_keys.append((commonname, sans))
797
798        name = next(filter(lambda key: key in self.certs, potential_keys), None)
799        if name:
800            entry = self.certs[name]
801        else:
802            entry = CertStoreEntry(
803                cert=dummy_cert(
804                    self.default_privatekey,
805                    self.default_ca._cert,
806                    commonname,
807                    sans,
808                    organization,
809                    crl_url,
810                ),
811                privatekey=self.default_privatekey,
812                chain_file=self.default_chain_file,
813                chain_certs=self.default_chain_certs,
814            )
815            self.certs[(commonname, sans)] = entry
816            self.expire(entry)
817
818        return entry
819
820
821def load_pem_private_key(
822    data: bytes, password: bytes | None
823) -> rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey:
824    """
825    like cryptography's load_pem_private_key, but silently falls back to not using a password
826    if the private key is unencrypted.
827    """
828    try:
829        return serialization.load_pem_private_key(data, password)  # type: ignore
830    except TypeError:
831        if password is not None:
832            return load_pem_private_key(data, None)
833        raise