Team Ai
Datasetpublic

codekingpro/portable-devtools

sourceHugging Faceupdated 5mo agoView on Hugging Face
1likes15kdownloads
certs.py742 linesDownload Raw Back to mitmproxy
1import contextlib2import datetime3import ipaddress4import logging5import os6import sys7import warnings8from collections.abc import Iterable9from dataclasses import dataclass10from pathlib import Path11from typing import cast12from typing import NewType13from typing import Optional14from typing import Union15 16import OpenSSL17from cryptography import x50918from cryptography.hazmat.primitives import hashes19from cryptography.hazmat.primitives import serialization20from cryptography.hazmat.primitives.asymmetric import dsa21from cryptography.hazmat.primitives.asymmetric import ec22from cryptography.hazmat.primitives.asymmetric import rsa23from cryptography.hazmat.primitives.asymmetric.types import CertificatePublicKeyTypes24from cryptography.hazmat.primitives.serialization import pkcs1225from cryptography.x509 import ExtendedKeyUsageOID26from cryptography.x509 import NameOID27 28from mitmproxy.coretypes import serializable29 30if sys.version_info < (3, 13):  # pragma: no cover31    from typing_extensions import deprecated32else:33    from warnings import deprecated34 35logger = logging.getLogger(__name__)36 37# Default expiry must not be too long: https://github.com/mitmproxy/mitmproxy/issues/81538# Note that expiry will be offset by `CERT_VALIDITY_OFFSET`, i.e. the cert will be39# 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/NHMkD8dKJf8suFCg350O2+dguLakSVif/t6dhImxInJk230HmfC8q93hdcg/j8rLGJYDKu3ik6H//BAHKIv51j5O9yjU3rXCfmVJQic2Nne39sg3CreAepEts2TvYHhVv3TEAzEqCtOuTjgDv0ntJ52Gwpj+BJBRQGG9NvprX1YGJ7WOFBP/hWU7d6tgvE6Xa7T/u9QIKpYHMIkcN/l3ZFB53chZEqVlyrcngtSXCROTPcDOQ6Q8QzhaBJS+Z6rcsd7X+haiQqvoFcmaJ08Ks6LQC54ZIL2EtYJw8V8z7C0igVEBIADZBI6OTbuuhDwRw//zU1uq52Oc48CIZlGxTYG/Evq55o9EWAXUYVzWkDSTeBH1r4z/qLPE2cnhtMxbFxuvK53jGB0emy2y1Ei6IhKshJ5qX56IB/aE7SSHyQ3MDHHkCmQJCsOd4Mo26YX61NZ+n501XjqpCBQ2+DfZCBh8Va2wDyv57A2Ryg9SUz8j0AXViRNMJgJrr446yro/FuJZwnQcO3WQnXeqSBnURqKjmqkeFP+d8586mk2tqJaY507lRNqtGlLnj7f5RNoBFJDCLBNurVgfvq9TCVWKDIFD4vZRjCrnl6I59rD693XKIHUCWOjMh1if6omGXKHH40QuME2gNa50+YPn1iYDl88uDbbMCAQI=60-----END DH PARAMETERS-----61"""62 63 64class Cert(serializable.Serializable):65    """Representation of a (TLS) certificate."""66 67    _cert: x509.Certificate68 69    def __init__(self, cert: x509.Certificate):70        assert isinstance(cert, x509.Certificate)71        self._cert = cert72 73    def __eq__(self, other):74        return self.fingerprint() == other.fingerprint()75 76    def __repr__(self):77        altnames = [str(x.value) for x in self.altnames]78        return f"<Cert(cn={self.cn!r}, altnames={altnames!r})>"79 80    def __hash__(self):81        return self._cert.__hash__()82 83    @classmethod84    def from_state(cls, state):85        return cls.from_pem(state)86 87    def get_state(self):88        return self.to_pem()89 90    def set_state(self, state):91        self._cert = x509.load_pem_x509_certificate(state)92 93    @classmethod94    def from_pem(cls, data: bytes) -> "Cert":95        cert = x509.load_pem_x509_certificate(data)  # type: ignore96        return cls(cert)97 98    def to_pem(self) -> bytes:99        return self._cert.public_bytes(serialization.Encoding.PEM)100 101    @classmethod102    def from_pyopenssl(self, x509: OpenSSL.crypto.X509) -> "Cert":103        return Cert(x509.to_cryptography())104 105    @deprecated("Use `to_cryptography` instead.")106    def to_pyopenssl(self) -> OpenSSL.crypto.X509:  # pragma: no cover107        return OpenSSL.crypto.X509.from_cryptography(self._cert)108 109    def to_cryptography(self) -> x509.Certificate:110        return self._cert111 112    def public_key(self) -> CertificatePublicKeyTypes:113        return self._cert.public_key()114 115    def fingerprint(self) -> bytes:116        return self._cert.fingerprint(hashes.SHA256())117 118    @property119    def issuer(self) -> list[tuple[str, str]]:120        return _name_to_keyval(self._cert.issuer)121 122    @property123    def notbefore(self) -> datetime.datetime:124        try:125            # type definitions haven't caught up with new API yet.126            return self._cert.not_valid_before_utc  # type: ignore127        except AttributeError:  # pragma: no cover128            # cryptography < 42.0129            return self._cert.not_valid_before.replace(tzinfo=datetime.UTC)130 131    @property132    def notafter(self) -> datetime.datetime:133        try:134            return self._cert.not_valid_after_utc  # type: ignore135        except AttributeError:  # pragma: no cover136            return self._cert.not_valid_after.replace(tzinfo=datetime.UTC)137 138    def has_expired(self) -> bool:139        if sys.version_info < (3, 11):  # pragma: no cover140            return datetime.datetime.now(datetime.UTC) > self.notafter141        return datetime.datetime.now(datetime.UTC) > self.notafter142 143    @property144    def subject(self) -> list[tuple[str, str]]:145        return _name_to_keyval(self._cert.subject)146 147    @property148    def serial(self) -> int:149        return self._cert.serial_number150 151    @property152    def is_ca(self) -> bool:153        constraints: x509.BasicConstraints154        try:155            constraints = self._cert.extensions.get_extension_for_class(156                x509.BasicConstraints157            ).value158            return constraints.ca159        except x509.ExtensionNotFound:160            return False161 162    @property163    def keyinfo(self) -> tuple[str, int]:164        public_key = self._cert.public_key()165        if isinstance(public_key, rsa.RSAPublicKey):166            return "RSA", public_key.key_size167        if isinstance(public_key, dsa.DSAPublicKey):168            return "DSA", public_key.key_size169        if isinstance(public_key, ec.EllipticCurvePublicKey):170            return f"EC ({public_key.curve.name})", public_key.key_size171        return (172            public_key.__class__.__name__.replace("PublicKey", "").replace("_", ""),173            getattr(public_key, "key_size", -1),174        )  # pragma: no cover175 176    @property177    def cn(self) -> str | None:178        attrs = self._cert.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)179        if attrs:180            return cast(str, attrs[0].value)181        return None182 183    @property184    def organization(self) -> str | None:185        attrs = self._cert.subject.get_attributes_for_oid(186            x509.NameOID.ORGANIZATION_NAME187        )188        if attrs:189            return cast(str, attrs[0].value)190        return None191 192    @property193    def altnames(self) -> x509.GeneralNames:194        """195        Get all SubjectAlternativeName DNS altnames.196        """197        try:198            sans = self._cert.extensions.get_extension_for_class(199                x509.SubjectAlternativeName200            ).value201        except x509.ExtensionNotFound:202            return x509.GeneralNames([])203        else:204            return x509.GeneralNames(sans)205 206    @property207    def crl_distribution_points(self) -> list[str]:208        try:209            ext = self._cert.extensions.get_extension_for_class(210                x509.CRLDistributionPoints211            ).value212        except x509.ExtensionNotFound:213            return []214        else:215            return [216                dist_point.full_name[0].value217                for dist_point in ext218                if dist_point.full_name219                and isinstance(dist_point.full_name[0], x509.UniformResourceIdentifier)220            ]221 222 223def _name_to_keyval(name: x509.Name) -> list[tuple[str, str]]:224    parts = []225    for attr in name:226        k = attr.rfc4514_string().partition("=")[0]227        v = cast(str, attr.value)228        parts.append((k, v))229    return parts230 231 232def create_ca(233    organization: str,234    cn: str,235    key_size: int,236) -> tuple[rsa.RSAPrivateKeyWithSerialization, x509.Certificate]:237    now = datetime.datetime.now()238 239    private_key = rsa.generate_private_key(240        public_exponent=65537,241        key_size=key_size,242    )  # type: ignore243    name = x509.Name(244        [245            x509.NameAttribute(NameOID.COMMON_NAME, cn),246            x509.NameAttribute(NameOID.ORGANIZATION_NAME, organization),247        ]248    )249    builder = x509.CertificateBuilder()250    builder = builder.serial_number(x509.random_serial_number())251    builder = builder.subject_name(name)252    builder = builder.not_valid_before(now + CERT_VALIDITY_OFFSET)253    builder = builder.not_valid_after(now + CERT_VALIDITY_OFFSET + CA_EXPIRY)254    builder = builder.issuer_name(name)255    builder = builder.public_key(private_key.public_key())256    builder = builder.add_extension(257        x509.BasicConstraints(ca=True, path_length=None), critical=True258    )259    builder = builder.add_extension(260        x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), critical=False261    )262    builder = builder.add_extension(263        x509.KeyUsage(264            digital_signature=False,265            content_commitment=False,266            key_encipherment=False,267            data_encipherment=False,268            key_agreement=False,269            key_cert_sign=True,270            crl_sign=True,271            encipher_only=False,272            decipher_only=False,273        ),274        critical=True,275    )276    builder = builder.add_extension(277        x509.SubjectKeyIdentifier.from_public_key(private_key.public_key()),278        critical=False,279    )280    cert = builder.sign(private_key=private_key, algorithm=hashes.SHA256())  # type: ignore281    return private_key, cert282 283 284def _fix_legacy_sans(sans: Iterable[x509.GeneralName] | list[str]) -> x509.GeneralNames:285    """286    SANs used to be a list of strings in mitmproxy 10.1 and below, but now they're a list of GeneralNames.287    This function converts the old format to the new one.288    """289    if isinstance(sans, x509.GeneralNames):290        return sans291    elif (292        isinstance(sans, list) and len(sans) > 0 and isinstance(sans[0], str)293    ):  # pragma: no cover294        warnings.warn(295            "Passing SANs as a list of strings is deprecated.",296            DeprecationWarning,297            stacklevel=2,298        )299 300        ss: list[x509.GeneralName] = []301        for x in cast(list[str], sans):302            try:303                ip = ipaddress.ip_address(x)304            except ValueError:305                x = x.encode("idna").decode()306                ss.append(x509.DNSName(x))307            else:308                ss.append(x509.IPAddress(ip))309        return x509.GeneralNames(ss)310    else:311        return x509.GeneralNames(cast(Iterable[x509.GeneralName], sans))312 313 314def dummy_cert(315    privkey: rsa.RSAPrivateKey,316    cacert: x509.Certificate,317    commonname: str | None,318    sans: Iterable[x509.GeneralName],319    organization: str | None = None,320    crl_url: str | None = None,321) -> Cert:322    """323    Generates a dummy certificate.324 325    privkey: CA private key326    cacert: CA certificate327    commonname: Common name for the generated certificate.328    sans: A list of Subject Alternate Names.329    organization: Organization name for the generated certificate.330    crl_url: URL of CRL distribution point331 332    Returns cert if operation succeeded, None if not.333    """334    builder = x509.CertificateBuilder()335    builder = builder.issuer_name(cacert.subject)336    builder = builder.add_extension(337        x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), critical=False338    )339    builder = builder.public_key(cacert.public_key())340 341    now = datetime.datetime.now()342    builder = builder.not_valid_before(now + CERT_VALIDITY_OFFSET)343    builder = builder.not_valid_after(now + CERT_VALIDITY_OFFSET + CERT_EXPIRY)344 345    subject = []346    is_valid_commonname = commonname is not None and len(commonname) < 64347    if is_valid_commonname:348        assert commonname is not None349        subject.append(x509.NameAttribute(NameOID.COMMON_NAME, commonname))350    if organization is not None:351        assert organization is not None352        subject.append(x509.NameAttribute(NameOID.ORGANIZATION_NAME, organization))353    builder = builder.subject_name(x509.Name(subject))354    builder = builder.serial_number(x509.random_serial_number())355 356    # RFC 5280 §4.2.1.6: subjectAltName is critical if subject is empty.357    builder = builder.add_extension(358        x509.SubjectAlternativeName(_fix_legacy_sans(sans)),359        critical=not is_valid_commonname,360    )361 362    # https://datatracker.ietf.org/doc/html/rfc5280#section-4.2.1.1363    # Per RFC 5280 §4.2.1.2, the AKI's keyIdentifier in a child certificate364    # MUST be byte-equal to the issuer's stored SubjectKeyIdentifier. If we365    # recompute it from the public key with `from_issuer_public_key()`, we366    # always derive a SHA-1 digest, which mismatches whenever the issuer's367    # SKI was generated by a different method (RFC 7093 truncated368    # SHA-256/384/512, hardware-rooted CAs, or any custom value). This369    # mismatch is rejected by strict TLS chain builders (`X509_V_FLAG_X509_STRICT`,370    # Python `ssl`, Go `crypto/x509`) with "authority and subject key371    # identifier mismatch". Most notably, cert-manager >=1.18 / Go >=1.25372    # default to truncated SHA-256 SKIs for FIPS 140-3 compliance.373    try:374        issuer_ski = cacert.extensions.get_extension_for_class(375            x509.SubjectKeyIdentifier376        ).value377        aki = x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(issuer_ski)378    except x509.ExtensionNotFound:379        # Fall back when the issuer cert has no SKI extension at all (legacy380        # CAs predating RFC 5280 conformance).381        aki = x509.AuthorityKeyIdentifier.from_issuer_public_key(cacert.public_key())  # type: ignore382    builder = builder.add_extension(aki, critical=False)383    # If CA and leaf cert have the same Subject Key Identifier, SChannel breaks in funny ways,384    # see https://github.com/mitmproxy/mitmproxy/issues/6494.385    # https://datatracker.ietf.org/doc/html/rfc5280#section-4.2.1.2 states386    # that SKI is optional for the leaf cert, so we skip that.387 388    if crl_url:389        builder = builder.add_extension(390            x509.CRLDistributionPoints(391                [392                    x509.DistributionPoint(393                        [x509.UniformResourceIdentifier(crl_url)],394                        relative_name=None,395                        crl_issuer=None,396                        reasons=None,397                    )398                ]399            ),400            critical=False,401        )402 403    cert = builder.sign(private_key=privkey, algorithm=hashes.SHA256())  # type: ignore404    return Cert(cert)405 406 407def dummy_crl(408    privkey: rsa.RSAPrivateKey,409    cacert: x509.Certificate,410) -> bytes:411    """412    Generates an empty CRL signed with the CA key413 414    privkey: CA private key415    cacert: CA certificate416 417    Returns a CRL DER encoded418    """419    builder = x509.CertificateRevocationListBuilder()420    builder = builder.issuer_name(cacert.issuer)421 422    now = datetime.datetime.now()423    builder = builder.last_update(now + CERT_VALIDITY_OFFSET)424    builder = builder.next_update(now + CERT_VALIDITY_OFFSET + CRL_EXPIRY)425 426    builder = builder.add_extension(x509.CRLNumber(1000), False)  # meaningless number427    crl = builder.sign(private_key=privkey, algorithm=hashes.SHA256())428    return crl.public_bytes(serialization.Encoding.DER)429 430 431@dataclass(frozen=True)432class CertStoreEntry:433    cert: Cert434    privatekey: rsa.RSAPrivateKey435    chain_file: Path | None436    chain_certs: list[Cert]437 438 439TCustomCertId = str  # manually provided certs (e.g. mitmproxy's --certs)440TGeneratedCertId = tuple[Optional[str], x509.GeneralNames]  # (common_name, sans)441TCertId = Union[TCustomCertId, TGeneratedCertId]442 443DHParams = NewType("DHParams", bytes)444 445 446class CertStore:447    """448    Implements an in-memory certificate store.449    """450 451    STORE_CAP = 100452    default_privatekey: rsa.RSAPrivateKey453    default_ca: Cert454    default_chain_file: Path | None455    default_chain_certs: list[Cert]456    dhparams: DHParams457    certs: dict[TCertId, CertStoreEntry]458    expire_queue: list[CertStoreEntry]459 460    def __init__(461        self,462        default_privatekey: rsa.RSAPrivateKey,463        default_ca: Cert,464        default_chain_file: Path | None,465        default_crl: bytes,466        dhparams: DHParams,467    ):468        self.default_privatekey = default_privatekey469        self.default_ca = default_ca470        self.default_chain_file = default_chain_file471        self.default_crl = default_crl472        self.default_chain_certs = (473            [474                Cert(c)475                for c in x509.load_pem_x509_certificates(476                    self.default_chain_file.read_bytes()477                )478            ]479            if self.default_chain_file480            else [default_ca]481        )482        self.dhparams = dhparams483        self.certs = {}484        self.expire_queue = []485 486    def expire(self, entry: CertStoreEntry) -> None:487        self.expire_queue.append(entry)488        if len(self.expire_queue) > self.STORE_CAP:489            d = self.expire_queue.pop(0)490            self.certs = {k: v for k, v in self.certs.items() if v != d}491 492    @staticmethod493    def load_dhparam(path: Path) -> DHParams:494        # mitmproxy<=0.10 doesn't generate a dhparam file.495        # Create it now if necessary.496        if not path.exists():497            path.write_bytes(DEFAULT_DHPARAM)498 499        # we could use cryptography for this, but it's unclear how to convert cryptography's object to pyOpenSSL's500        # expected format.501        bio = OpenSSL.SSL._lib.BIO_new_file(  # type: ignore502            str(path).encode(sys.getfilesystemencoding()), b"r"503        )504        if bio != OpenSSL.SSL._ffi.NULL:  # type: ignore505            bio = OpenSSL.SSL._ffi.gc(bio, OpenSSL.SSL._lib.BIO_free)  # type: ignore506            dh = OpenSSL.SSL._lib.PEM_read_bio_DHparams(  # type: ignore507                bio,508                OpenSSL.SSL._ffi.NULL,  # type: ignore509                OpenSSL.SSL._ffi.NULL,  # type: ignore510                OpenSSL.SSL._ffi.NULL,  # type: ignore511            )512            dh = OpenSSL.SSL._ffi.gc(dh, OpenSSL.SSL._lib.DH_free)  # type: ignore513            return dh514        raise RuntimeError("Error loading DH Params.")  # pragma: no cover515 516    @classmethod517    def from_store(518        cls,519        path: Path | str,520        basename: str,521        key_size: int,522        passphrase: bytes | None = None,523    ) -> "CertStore":524        path = Path(path)525        ca_file = path / f"{basename}-ca.pem"526        dhparam_file = path / f"{basename}-dhparam.pem"527        if not ca_file.exists():528            cls.create_store(path, basename, key_size)529        return cls.from_files(ca_file, dhparam_file, passphrase)530 531    @classmethod532    def from_files(533        cls, ca_file: Path, dhparam_file: Path, passphrase: bytes | None = None534    ) -> "CertStore":535        raw = ca_file.read_bytes()536        key = load_pem_private_key(raw, passphrase)537        dh = cls.load_dhparam(dhparam_file)538        certs = x509.load_pem_x509_certificates(raw)539        ca = Cert(certs[0])540        crl = dummy_crl(key, ca._cert)541        if len(certs) > 1:542            chain_file: Path | None = ca_file543        else:544            chain_file = None545        return cls(key, ca, chain_file, crl, dh)546 547    @staticmethod548    @contextlib.contextmanager549    def umask_secret():550        """551        Context to temporarily set umask to its original value bitor 0o77.552        Useful when writing private keys to disk so that only the owner553        will be able to read them.554        """555        original_umask = os.umask(0)556        os.umask(original_umask | 0o77)557        try:558            yield559        finally:560            os.umask(original_umask)561 562    @staticmethod563    def create_store(564        path: Path, basename: str, key_size: int, organization=None, cn=None565    ) -> None:566        path.mkdir(parents=True, exist_ok=True)567 568        organization = organization or basename569        cn = cn or basename570 571        key: rsa.RSAPrivateKeyWithSerialization572        ca: x509.Certificate573        key, ca = create_ca(organization=organization, cn=cn, key_size=key_size)574 575        # Dump the CA plus private key.576        with CertStore.umask_secret():577            # PEM format578            (path / f"{basename}-ca.pem").write_bytes(579                key.private_bytes(580                    encoding=serialization.Encoding.PEM,581                    format=serialization.PrivateFormat.TraditionalOpenSSL,582                    encryption_algorithm=serialization.NoEncryption(),583                )584                + ca.public_bytes(serialization.Encoding.PEM)585            )586 587            # PKCS12 format for Windows devices588            (path / f"{basename}-ca.p12").write_bytes(589                pkcs12.serialize_key_and_certificates(  # type: ignore590                    name=basename.encode(),591                    key=key,592                    cert=ca,593                    cas=None,594                    encryption_algorithm=serialization.NoEncryption(),595                )596            )597 598        # Dump the certificate in PEM format599        pem_cert = ca.public_bytes(serialization.Encoding.PEM)600        (path / f"{basename}-ca-cert.pem").write_bytes(pem_cert)601        # Create a .cer file with the same contents for Android602        (path / f"{basename}-ca-cert.cer").write_bytes(pem_cert)603 604        # Dump the certificate in PKCS12 format for Windows devices605        (path / f"{basename}-ca-cert.p12").write_bytes(606            pkcs12.serialize_key_and_certificates(607                name=basename.encode(),608                key=None,  # type: ignore609                cert=ca,610                cas=None,611                encryption_algorithm=serialization.NoEncryption(),612            )613        )614 615        (path / f"{basename}-dhparam.pem").write_bytes(DEFAULT_DHPARAM)616 617    def add_cert_file(618        self, spec: str, path: Path, passphrase: bytes | None = None619    ) -> None:620        raw = path.read_bytes()621        cert = Cert.from_pem(raw)622        try:623            private_key = load_pem_private_key(raw, password=passphrase)624        except ValueError as e:625            private_key = self.default_privatekey626            if cert.public_key() != private_key.public_key():627                raise ValueError(628                    f'Unable to find private key in "{path.absolute()}": {e}'629                ) from e630        else:631            if cert.public_key() != private_key.public_key():632                raise ValueError(633                    f'Private and public keys in "{path.absolute()}" do not match:\n'634                    f"{cert.public_key()=}\n"635                    f"{private_key.public_key()=}"636                )637 638        try:639            chain = [Cert(x) for x in x509.load_pem_x509_certificates(raw)]640        except ValueError as e:641            logger.warning(f"Failed to read certificate chain: {e}")642            chain = [cert]643 644        if cert.is_ca:645            logger.warning(646                f'"{path.absolute()}" is a certificate authority and not a leaf certificate. '647                f"This indicates a misconfiguration, see https://docs.mitmproxy.org/stable/concepts-certificates/."648            )649 650        self.add_cert(CertStoreEntry(cert, private_key, path, chain), spec)651 652    def add_cert(self, entry: CertStoreEntry, *names: str) -> None:653        """654        Adds a cert to the certstore. We register the CN in the cert plus655        any SANs, and also the list of names provided as an argument.656        """657        if entry.cert.cn:658            self.certs[entry.cert.cn] = entry659        for i in entry.cert.altnames:660            self.certs[str(i.value)] = entry661        for i in names:662            self.certs[i] = entry663 664    @staticmethod665    def asterisk_forms(dn: str | x509.GeneralName) -> list[str]:666        """667        Return all asterisk forms for a domain. For example, for www.example.com this will return668        [b"www.example.com", b"*.example.com", b"*.com"]. The single wildcard "*" is omitted.669        """670        if isinstance(dn, str):671            parts = dn.split(".")672            ret = [dn]673            for i in range(1, len(parts)):674                ret.append("*." + ".".join(parts[i:]))675            return ret676        elif isinstance(dn, x509.DNSName):677            return CertStore.asterisk_forms(dn.value)678        else:679            return [str(dn.value)]680 681    def get_cert(682        self,683        commonname: str | None,684        sans: Iterable[x509.GeneralName],685        organization: str | None = None,686        crl_url: str | None = None,687    ) -> CertStoreEntry:688        """689        commonname: Common name for the generated certificate. Must be a690        valid, plain-ASCII, IDNA-encoded domain name.691 692        sans: A list of Subject Alternate Names.693 694        organization: Organization name for the generated certificate.695 696        crl_url: URL of CRL distribution point697        """698        sans = _fix_legacy_sans(sans)699 700        potential_keys: list[TCertId] = []701        if commonname:702            potential_keys.extend(self.asterisk_forms(commonname))703        for s in sans:704            potential_keys.extend(self.asterisk_forms(s))705        potential_keys.append("*")706        potential_keys.append((commonname, sans))707 708        name = next(filter(lambda key: key in self.certs, potential_keys), None)709        if name:710            entry = self.certs[name]711        else:712            entry = CertStoreEntry(713                cert=dummy_cert(714                    self.default_privatekey,715                    self.default_ca._cert,716                    commonname,717                    sans,718                    organization,719                    crl_url,720                ),721                privatekey=self.default_privatekey,722                chain_file=self.default_chain_file,723                chain_certs=self.default_chain_certs,724            )725            self.certs[(commonname, sans)] = entry726            self.expire(entry)727 728        return entry729 730 731def load_pem_private_key(data: bytes, password: bytes | None) -> rsa.RSAPrivateKey:732    """733    like cryptography's load_pem_private_key, but silently falls back to not using a password734    if the private key is unencrypted.735    """736    try:737        return serialization.load_pem_private_key(data, password)  # type: ignore738    except TypeError:739        if password is not None:740            return load_pem_private_key(data, None)741        raise742 
codekingpro/portable-devtools · Team Ai