Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py # This file is dual licensed under the terms of the Apache License, Version # 2.0, and the BSD License. See the LICENSE file in the root of this repository # for complete details. from __future__ import annotations from cryptography.x509 import certificate_transparency from cryptography.x509.base import ( Attribute, AttributeNotFound, Attributes, Certificate, CertificateBuilder, CertificateRevocationList, CertificateRevocationListBuilder, CertificateSigningRequest, CertificateSigningRequestBuilder, InvalidVersion, RevokedCertificate, RevokedCertificateBuilder, Version, load_der_x509_certificate, load_der_x509_crl, load_der_x509_csr, load_pem_x509_certificate, load_pem_x509_certificates, load_pem_x509_crl, load_pem_x509_csr, random_serial_number, ) from cryptography.x509.extensions import ( AccessDescription, AuthorityInformationAccess, AuthorityKeyIdentifier, BasicConstraints, CertificateIssuer, CertificatePolicies, CRLDistributionPoints, CRLNumber, CRLReason, DeltaCRLIndicator, DistributionPoint, DuplicateExtension, ExtendedKeyUsage, Extension, ExtensionNotFound, Extensions, ExtensionType, FreshestCRL, GeneralNames, InhibitAnyPolicy, InvalidityDate, IssuerAlternativeName, IssuingDistributionPoint, KeyUsage, MSCertificateTemplate, NameConstraints, NoticeReference, OCSPAcceptableResponses, OCSPNoCheck, OCSPNonce, PolicyConstraints, PolicyInformation, PrecertificateSignedCertificateTimestamps, PrecertPoison, ReasonFlags, SignedCertificateTimestamps, SubjectAlternativeName, SubjectInformationAccess, SubjectKeyIdentifier, TLSFeature, TLSFeatureType, UnrecognizedExtension, UserNotice, ) from cryptography.x509.general_name import ( DirectoryName, DNSName, GeneralName, IPAddress, OtherName, RegisteredID, RFC822Name, UniformResourceIdentifier, UnsupportedGeneralNameType, ) from cryptography.x509.name import ( Name, NameAttribute, RelativeDistinguishedName, ) from cryptography.x509.oid import ( AuthorityInformationAccessOID, CertificatePoliciesOID, CRLEntryExtensionOID, ExtendedKeyUsageOID, ExtensionOID, NameOID, ObjectIdentifier, SignatureAlgorithmOID, ) OID_AUTHORITY_INFORMATION_ACCESS = ExtensionOID.AUTHORITY_INFORMATION_ACCESS OID_AUTHORITY_KEY_IDENTIFIER = ExtensionOID.AUTHORITY_KEY_IDENTIFIER OID_BASIC_CONSTRAINTS = ExtensionOID.BASIC_CONSTRAINTS OID_CERTIFICATE_POLICIES = ExtensionOID.CERTIFICATE_POLICIES OID_CRL_DISTRIBUTION_POINTS = ExtensionOID.CRL_DISTRIBUTION_POINTS OID_EXTENDED_KEY_USAGE = ExtensionOID.EXTENDED_KEY_USAGE OID_FRESHEST_CRL = ExtensionOID.FRESHEST_CRL OID_INHIBIT_ANY_POLICY = ExtensionOID.INHIBIT_ANY_POLICY OID_ISSUER_ALTERNATIVE_NAME = ExtensionOID.ISSUER_ALTERNATIVE_NAME OID_KEY_USAGE = ExtensionOID.KEY_USAGE OID_NAME_CONSTRAINTS = ExtensionOID.NAME_CONSTRAINTS OID_OCSP_NO_CHECK = ExtensionOID.OCSP_NO_CHECK OID_POLICY_CONSTRAINTS = ExtensionOID.POLICY_CONSTRAINTS OID_POLICY_MAPPINGS = ExtensionOID.POLICY_MAPPINGS OID_SUBJECT_ALTERNATIVE_NAME = ExtensionOID.SUBJECT_ALTERNATIVE_NAME OID_SUBJECT_DIRECTORY_ATTRIBUTES = ExtensionOID.SUBJECT_DIRECTORY_ATTRIBUTES OID_SUBJECT_INFORMATION_ACCESS = ExtensionOID.SUBJECT_INFORMATION_ACCESS OID_SUBJECT_KEY_IDENTIFIER = ExtensionOID.SUBJECT_KEY_IDENTIFIER OID_DSA_WITH_SHA1 = SignatureAlgorithmOID.DSA_WITH_SHA1 OID_DSA_WITH_SHA224 = SignatureAlgorithmOID.DSA_WITH_SHA224 OID_DSA_WITH_SHA256 = SignatureAlgorithmOID.DSA_WITH_SHA256 OID_ECDSA_WITH_SHA1 = SignatureAlgorithmOID.ECDSA_WITH_SHA1 OID_ECDSA_WITH_SHA224 = SignatureAlgorithmOID.ECDSA_WITH_SHA224 OID_ECDSA_WITH_SHA256 = SignatureAlgorithmOID.ECDSA_WITH_SHA256 OID_ECDSA_WITH_SHA384 = SignatureAlgorithmOID.ECDSA_WITH_SHA384 OID_ECDSA_WITH_SHA512 = SignatureAlgorithmOID.ECDSA_WITH_SHA512 OID_RSA_WITH_MD5 = SignatureAlgorithmOID.RSA_WITH_MD5 OID_RSA_WITH_SHA1 = SignatureAlgorithmOID.RSA_WITH_SHA1 OID_RSA_WITH_SHA224 = SignatureAlgorithmOID.RSA_WITH_SHA224 OID_RSA_WITH_SHA256 = SignatureAlgorithmOID.RSA_WITH_SHA256 OID_RSA_WITH_SHA384 = SignatureAlgorithmOID.RSA_WITH_SHA384 OID_RSA_WITH_SHA512 = SignatureAlgorithmOID.RSA_WITH_SHA512 OID_RSASSA_PSS = SignatureAlgorithmOID.RSASSA_PSS OID_COMMON_NAME = NameOID.COMMON_NAME OID_COUNTRY_NAME = NameOID.COUNTRY_NAME OID_DOMAIN_COMPONENT = NameOID.DOMAIN_COMPONENT OID_DN_QUALIFIER = NameOID.DN_QUALIFIER OID_EMAIL_ADDRESS = NameOID.EMAIL_ADDRESS OID_GENERATION_QUALIFIER = NameOID.GENERATION_QUALIFIER OID_GIVEN_NAME = NameOID.GIVEN_NAME OID_LOCALITY_NAME = NameOID.LOCALITY_NAME OID_ORGANIZATIONAL_UNIT_NAME = NameOID.ORGANIZATIONAL_UNIT_NAME OID_ORGANIZATION_NAME = NameOID.ORGANIZATION_NAME OID_PSEUDONYM = NameOID.PSEUDONYM OID_SERIAL_NUMBER = NameOID.SERIAL_NUMBER OID_STATE_OR_PROVINCE_NAME = NameOID.STATE_OR_PROVINCE_NAME OID_SURNAME = NameOID.SURNAME OID_TITLE = NameOID.TITLE OID_CLIENT_AUTH = ExtendedKeyUsageOID.CLIENT_AUTH OID_CODE_SIGNING = ExtendedKeyUsageOID.CODE_SIGNING OID_EMAIL_PROTECTION = ExtendedKeyUsageOID.EMAIL_PROTECTION OID_OCSP_SIGNING = ExtendedKeyUsageOID.OCSP_SIGNING OID_SERVER_AUTH = ExtendedKeyUsageOID.SERVER_AUTH OID_TIME_STAMPING = ExtendedKeyUsageOID.TIME_STAMPING OID_ANY_POLICY = CertificatePoliciesOID.ANY_POLICY OID_CPS_QUALIFIER = CertificatePoliciesOID.CPS_QUALIFIER OID_CPS_USER_NOTICE = CertificatePoliciesOID.CPS_USER_NOTICE OID_CERTIFICATE_ISSUER = CRLEntryExtensionOID.CERTIFICATE_ISSUER OID_CRL_REASON = CRLEntryExtensionOID.CRL_REASON OID_INVALIDITY_DATE = CRLEntryExtensionOID.INVALIDITY_DATE OID_CA_ISSUERS = AuthorityInformationAccessOID.CA_ISSUERS OID_OCSP = AuthorityInformationAccessOID.OCSP __all__ = [ "certificate_transparency", "load_pem_x509_certificate", "load_pem_x509_certificates", "load_der_x509_certificate", "load_pem_x509_csr", "load_der_x509_csr", "load_pem_x509_crl", "load_der_x509_crl", "random_serial_number", "Attribute", "AttributeNotFound", "Attributes", "InvalidVersion", "DeltaCRLIndicator", "DuplicateExtension", "ExtensionNotFound", "UnsupportedGeneralNameType", "NameAttribute", "Name", "RelativeDistinguishedName", "ObjectIdentifier", "ExtensionType", "Extensions", "Extension", "ExtendedKeyUsage", "FreshestCRL", "IssuingDistributionPoint", "TLSFeature", "TLSFeatureType", "OCSPAcceptableResponses", "OCSPNoCheck", "BasicConstraints", "CRLNumber", "KeyUsage", "AuthorityInformationAccess", "SubjectInformationAccess", "AccessDescription", "CertificatePolicies", "PolicyInformation", "UserNotice", "NoticeReference", "SubjectKeyIdentifier", "NameConstraints", "CRLDistributionPoints", "DistributionPoint", "ReasonFlags", "InhibitAnyPolicy", "SubjectAlternativeName", "IssuerAlternativeName", "AuthorityKeyIdentifier", "GeneralNames", "GeneralName", "RFC822Name", "DNSName", "UniformResourceIdentifier", "RegisteredID", "DirectoryName", "IPAddress", "OtherName", "Certificate", "CertificateRevocationList", "CertificateRevocationListBuilder", "CertificateSigningRequest", "RevokedCertificate", "RevokedCertificateBuilder", "CertificateSigningRequestBuilder", "CertificateBuilder", "Version", "OID_CA_ISSUERS", "OID_OCSP", "CertificateIssuer", "CRLReason", "InvalidityDate", "UnrecognizedExtension", "PolicyConstraints", "PrecertificateSignedCertificateTimestamps", "PrecertPoison", "OCSPNonce", "SignedCertificateTimestamps", "SignatureAlgorithmOID", "NameOID", "MSCertificateTemplate", ]	7.69 KB	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 base.py # This file is dual licensed under the terms of the Apache License, Version # 2.0, and the BSD License. See the LICENSE file in the root of this repository # for complete details. from __future__ import annotations import abc import datetime import os import typing from cryptography import utils from cryptography.hazmat.bindings._rust import x509 as rust_x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ( dsa, ec, ed448, ed25519, padding, rsa, x448, x25519, ) from cryptography.hazmat.primitives.asymmetric.types import ( CertificateIssuerPrivateKeyTypes, CertificateIssuerPublicKeyTypes, CertificatePublicKeyTypes, ) from cryptography.x509.extensions import ( Extension, Extensions, ExtensionType, _make_sequence_methods, ) from cryptography.x509.name import Name, _ASN1Type from cryptography.x509.oid import ObjectIdentifier _EARLIEST_UTC_TIME = datetime.datetime(1950, 1, 1) # This must be kept in sync with sign.rs's list of allowable types in # identify_hash_type _AllowedHashTypes = typing.Union[ hashes.SHA224, hashes.SHA256, hashes.SHA384, hashes.SHA512, hashes.SHA3_224, hashes.SHA3_256, hashes.SHA3_384, hashes.SHA3_512, ] class AttributeNotFound(Exception): def __init__(self, msg: str, oid: ObjectIdentifier) -> None: super().__init__(msg) self.oid = oid def _reject_duplicate_extension( extension: Extension[ExtensionType], extensions: typing.List[Extension[ExtensionType]], ) -> None: # This is quadratic in the number of extensions for e in extensions: if e.oid == extension.oid: raise ValueError("This extension has already been set.") def _reject_duplicate_attribute( oid: ObjectIdentifier, attributes: typing.List[ typing.Tuple[ObjectIdentifier, bytes, typing.Optional[int]] ], ) -> None: # This is quadratic in the number of attributes for attr_oid, _, _ in attributes: if attr_oid == oid: raise ValueError("This attribute has already been set.") def _convert_to_naive_utc_time(time: datetime.datetime) -> datetime.datetime: """Normalizes a datetime to a naive datetime in UTC. time -- datetime to normalize. Assumed to be in UTC if not timezone aware. """ if time.tzinfo is not None: offset = time.utcoffset() offset = offset if offset else datetime.timedelta() return time.replace(tzinfo=None) - offset else: return time class Attribute: def __init__( self, oid: ObjectIdentifier, value: bytes, _type: int = _ASN1Type.UTF8String.value, ) -> None: self._oid = oid self._value = value self._type = _type @property def oid(self) -> ObjectIdentifier: return self._oid @property def value(self) -> bytes: return self._value def __repr__(self) -> str: return f"<Attribute(oid={self.oid}, value={self.value!r})>" def __eq__(self, other: object) -> bool: if not isinstance(other, Attribute): return NotImplemented return ( self.oid == other.oid and self.value == other.value and self._type == other._type ) def __hash__(self) -> int: return hash((self.oid, self.value, self._type)) class Attributes: def __init__( self, attributes: typing.Iterable[Attribute], ) -> None: self._attributes = list(attributes) __len__, __iter__, __getitem__ = _make_sequence_methods("_attributes") def __repr__(self) -> str: return f"<Attributes({self._attributes})>" def get_attribute_for_oid(self, oid: ObjectIdentifier) -> Attribute: for attr in self: if attr.oid == oid: return attr raise AttributeNotFound(f"No {oid} attribute was found", oid) class Version(utils.Enum): v1 = 0 v3 = 2 class InvalidVersion(Exception): def __init__(self, msg: str, parsed_version: int) -> None: super().__init__(msg) self.parsed_version = parsed_version class Certificate(metaclass=abc.ABCMeta): @abc.abstractmethod def fingerprint(self, algorithm: hashes.HashAlgorithm) -> bytes: """ Returns bytes using digest passed. """ @property @abc.abstractmethod def serial_number(self) -> int: """ Returns certificate serial number """ @property @abc.abstractmethod def version(self) -> Version: """ Returns the certificate version """ @abc.abstractmethod def public_key(self) -> CertificatePublicKeyTypes: """ Returns the public key """ @property @abc.abstractmethod def not_valid_before(self) -> datetime.datetime: """ Not before time (represented as UTC datetime) """ @property @abc.abstractmethod def not_valid_after(self) -> datetime.datetime: """ Not after time (represented as UTC datetime) """ @property @abc.abstractmethod def issuer(self) -> Name: """ Returns the issuer name object. """ @property @abc.abstractmethod def subject(self) -> Name: """ Returns the subject name object. """ @property @abc.abstractmethod def signature_hash_algorithm( self, ) -> typing.Optional[hashes.HashAlgorithm]: """ Returns a HashAlgorithm corresponding to the type of the digest signed in the certificate. """ @property @abc.abstractmethod def signature_algorithm_oid(self) -> ObjectIdentifier: """ Returns the ObjectIdentifier of the signature algorithm. """ @property @abc.abstractmethod def signature_algorithm_parameters( self, ) -> typing.Union[None, padding.PSS, padding.PKCS1v15, ec.ECDSA]: """ Returns the signature algorithm parameters. """ @property @abc.abstractmethod def extensions(self) -> Extensions: """ Returns an Extensions object. """ @property @abc.abstractmethod def signature(self) -> bytes: """ Returns the signature bytes. """ @property @abc.abstractmethod def tbs_certificate_bytes(self) -> bytes: """ Returns the tbsCertificate payload bytes as defined in RFC 5280. """ @property @abc.abstractmethod def tbs_precertificate_bytes(self) -> bytes: """ Returns the tbsCertificate payload bytes with the SCT list extension stripped. """ @abc.abstractmethod def __eq__(self, other: object) -> bool: """ Checks equality. """ @abc.abstractmethod def __hash__(self) -> int: """ Computes a hash. """ @abc.abstractmethod def public_bytes(self, encoding: serialization.Encoding) -> bytes: """ Serializes the certificate to PEM or DER format. """ @abc.abstractmethod def verify_directly_issued_by(self, issuer: Certificate) -> None: """ This method verifies that certificate issuer name matches the issuer subject name and that the certificate is signed by the issuer's private key. No other validation is performed. """ # Runtime isinstance checks need this since the rust class is not a subclass. Certificate.register(rust_x509.Certificate) class RevokedCertificate(metaclass=abc.ABCMeta): @property @abc.abstractmethod def serial_number(self) -> int: """ Returns the serial number of the revoked certificate. """ @property @abc.abstractmethod def revocation_date(self) -> datetime.datetime: """ Returns the date of when this certificate was revoked. """ @property @abc.abstractmethod def extensions(self) -> Extensions: """ Returns an Extensions object containing a list of Revoked extensions. """ # Runtime isinstance checks need this since the rust class is not a subclass. RevokedCertificate.register(rust_x509.RevokedCertificate) class _RawRevokedCertificate(RevokedCertificate): def __init__( self, serial_number: int, revocation_date: datetime.datetime, extensions: Extensions, ): self._serial_number = serial_number self._revocation_date = revocation_date self._extensions = extensions @property def serial_number(self) -> int: return self._serial_number @property def revocation_date(self) -> datetime.datetime: return self._revocation_date @property def extensions(self) -> Extensions: return self._extensions class CertificateRevocationList(metaclass=abc.ABCMeta): @abc.abstractmethod def public_bytes(self, encoding: serialization.Encoding) -> bytes: """ Serializes the CRL to PEM or DER format. """ @abc.abstractmethod def fingerprint(self, algorithm: hashes.HashAlgorithm) -> bytes: """ Returns bytes using digest passed. """ @abc.abstractmethod def get_revoked_certificate_by_serial_number( self, serial_number: int ) -> typing.Optional[RevokedCertificate]: """ Returns an instance of RevokedCertificate or None if the serial_number is not in the CRL. """ @property @abc.abstractmethod def signature_hash_algorithm( self, ) -> typing.Optional[hashes.HashAlgorithm]: """ Returns a HashAlgorithm corresponding to the type of the digest signed in the certificate. """ @property @abc.abstractmethod def signature_algorithm_oid(self) -> ObjectIdentifier: """ Returns the ObjectIdentifier of the signature algorithm. """ @property @abc.abstractmethod def issuer(self) -> Name: """ Returns the X509Name with the issuer of this CRL. """ @property @abc.abstractmethod def next_update(self) -> typing.Optional[datetime.datetime]: """ Returns the date of next update for this CRL. """ @property @abc.abstractmethod def last_update(self) -> datetime.datetime: """ Returns the date of last update for this CRL. """ @property @abc.abstractmethod def extensions(self) -> Extensions: """ Returns an Extensions object containing a list of CRL extensions. """ @property @abc.abstractmethod def signature(self) -> bytes: """ Returns the signature bytes. """ @property @abc.abstractmethod def tbs_certlist_bytes(self) -> bytes: """ Returns the tbsCertList payload bytes as defined in RFC 5280. """ @abc.abstractmethod def __eq__(self, other: object) -> bool: """ Checks equality. """ @abc.abstractmethod def __len__(self) -> int: """ Number of revoked certificates in the CRL. """ @typing.overload def __getitem__(self, idx: int) -> RevokedCertificate: ... @typing.overload def __getitem__(self, idx: slice) -> typing.List[RevokedCertificate]: ... @abc.abstractmethod def __getitem__( self, idx: typing.Union[int, slice] ) -> typing.Union[RevokedCertificate, typing.List[RevokedCertificate]]: """ Returns a revoked certificate (or slice of revoked certificates). """ @abc.abstractmethod def __iter__(self) -> typing.Iterator[RevokedCertificate]: """ Iterator over the revoked certificates """ @abc.abstractmethod def is_signature_valid( self, public_key: CertificateIssuerPublicKeyTypes ) -> bool: """ Verifies signature of revocation list against given public key. """ CertificateRevocationList.register(rust_x509.CertificateRevocationList) class CertificateSigningRequest(metaclass=abc.ABCMeta): @abc.abstractmethod def __eq__(self, other: object) -> bool: """ Checks equality. """ @abc.abstractmethod def __hash__(self) -> int: """ Computes a hash. """ @abc.abstractmethod def public_key(self) -> CertificatePublicKeyTypes: """ Returns the public key """ @property @abc.abstractmethod def subject(self) -> Name: """ Returns the subject name object. """ @property @abc.abstractmethod def signature_hash_algorithm( self, ) -> typing.Optional[hashes.HashAlgorithm]: """ Returns a HashAlgorithm corresponding to the type of the digest signed in the certificate. """ @property @abc.abstractmethod def signature_algorithm_oid(self) -> ObjectIdentifier: """ Returns the ObjectIdentifier of the signature algorithm. """ @property @abc.abstractmethod def extensions(self) -> Extensions: """ Returns the extensions in the signing request. """ @property @abc.abstractmethod def attributes(self) -> Attributes: """ Returns an Attributes object. """ @abc.abstractmethod def public_bytes(self, encoding: serialization.Encoding) -> bytes: """ Encodes the request to PEM or DER format. """ @property @abc.abstractmethod def signature(self) -> bytes: """ Returns the signature bytes. """ @property @abc.abstractmethod def tbs_certrequest_bytes(self) -> bytes: """ Returns the PKCS#10 CertificationRequestInfo bytes as defined in RFC 2986. """ @property @abc.abstractmethod def is_signature_valid(self) -> bool: """ Verifies signature of signing request. """ @abc.abstractmethod def get_attribute_for_oid(self, oid: ObjectIdentifier) -> bytes: """ Get the attribute value for a given OID. """ # Runtime isinstance checks need this since the rust class is not a subclass. CertificateSigningRequest.register(rust_x509.CertificateSigningRequest) # Backend argument preserved for API compatibility, but ignored. def load_pem_x509_certificate( data: bytes, backend: typing.Any = None ) -> Certificate: return rust_x509.load_pem_x509_certificate(data) def load_pem_x509_certificates(data: bytes) -> typing.List[Certificate]: return rust_x509.load_pem_x509_certificates(data) # Backend argument preserved for API compatibility, but ignored. def load_der_x509_certificate( data: bytes, backend: typing.Any = None ) -> Certificate: return rust_x509.load_der_x509_certificate(data) # Backend argument preserved for API compatibility, but ignored. def load_pem_x509_csr( data: bytes, backend: typing.Any = None ) -> CertificateSigningRequest: return rust_x509.load_pem_x509_csr(data) # Backend argument preserved for API compatibility, but ignored. def load_der_x509_csr( data: bytes, backend: typing.Any = None ) -> CertificateSigningRequest: return rust_x509.load_der_x509_csr(data) # Backend argument preserved for API compatibility, but ignored. def load_pem_x509_crl( data: bytes, backend: typing.Any = None ) -> CertificateRevocationList: return rust_x509.load_pem_x509_crl(data) # Backend argument preserved for API compatibility, but ignored. def load_der_x509_crl( data: bytes, backend: typing.Any = None ) -> CertificateRevocationList: return rust_x509.load_der_x509_crl(data) class CertificateSigningRequestBuilder: def __init__( self, subject_name: typing.Optional[Name] = None, extensions: typing.List[Extension[ExtensionType]] = [], attributes: typing.List[ typing.Tuple[ObjectIdentifier, bytes, typing.Optional[int]] ] = [], ): """ Creates an empty X.509 certificate request (v1). """ self._subject_name = subject_name self._extensions = extensions self._attributes = attributes def subject_name(self, name: Name) -> CertificateSigningRequestBuilder: """ Sets the certificate requestor's distinguished name. """ if not isinstance(name, Name): raise TypeError("Expecting x509.Name object.") if self._subject_name is not None: raise ValueError("The subject name may only be set once.") return CertificateSigningRequestBuilder( name, self._extensions, self._attributes ) def add_extension( self, extval: ExtensionType, critical: bool ) -> CertificateSigningRequestBuilder: """ Adds an X.509 extension to the certificate request. """ if not isinstance(extval, ExtensionType): raise TypeError("extension must be an ExtensionType") extension = Extension(extval.oid, critical, extval) _reject_duplicate_extension(extension, self._extensions) return CertificateSigningRequestBuilder( self._subject_name, self._extensions + [extension], self._attributes, ) def add_attribute( self, oid: ObjectIdentifier, value: bytes, , _tag: typing.Optional[_ASN1Type] = None, ) -> CertificateSigningRequestBuilder: """ Adds an X.509 attribute with an OID and associated value. """ if not isinstance(oid, ObjectIdentifier): raise TypeError("oid must be an ObjectIdentifier") if not isinstance(value, bytes): raise TypeError("value must be bytes") if _tag is not None and not isinstance(_tag, _ASN1Type): raise TypeError("tag must be _ASN1Type") _reject_duplicate_attribute(oid, self._attributes) if _tag is not None: tag = _tag.value else: tag = None return CertificateSigningRequestBuilder( self._subject_name, self._extensions, self._attributes + [(oid, value, tag)], ) def sign( self, private_key: CertificateIssuerPrivateKeyTypes, algorithm: typing.Optional[_AllowedHashTypes], backend: typing.Any = None, ) -> CertificateSigningRequest: """ Signs the request using the requestor's private key. """ if self._subject_name is None: raise ValueError("A CertificateSigningRequest must have a subject") return rust_x509.create_x509_csr(self, private_key, algorithm) class CertificateBuilder: _extensions: typing.List[Extension[ExtensionType]] def __init__( self, issuer_name: typing.Optional[Name] = None, subject_name: typing.Optional[Name] = None, public_key: typing.Optional[CertificatePublicKeyTypes] = None, serial_number: typing.Optional[int] = None, not_valid_before: typing.Optional[datetime.datetime] = None, not_valid_after: typing.Optional[datetime.datetime] = None, extensions: typing.List[Extension[ExtensionType]] = [], ) -> None: self._version = Version.v3 self._issuer_name = issuer_name self._subject_name = subject_name self._public_key = public_key self._serial_number = serial_number self._not_valid_before = not_valid_before self._not_valid_after = not_valid_after self._extensions = extensions def issuer_name(self, name: Name) -> CertificateBuilder: """ Sets the CA's distinguished name. """ if not isinstance(name, Name): raise TypeError("Expecting x509.Name object.") if self._issuer_name is not None: raise ValueError("The issuer name may only be set once.") return CertificateBuilder( name, self._subject_name, self._public_key, self._serial_number, self._not_valid_before, self._not_valid_after, self._extensions, ) def subject_name(self, name: Name) -> CertificateBuilder: """ Sets the requestor's distinguished name. """ if not isinstance(name, Name): raise TypeError("Expecting x509.Name object.") if self._subject_name is not None: raise ValueError("The subject name may only be set once.") return CertificateBuilder( self._issuer_name, name, self._public_key, self._serial_number, self._not_valid_before, self._not_valid_after, self._extensions, ) def public_key( self, key: CertificatePublicKeyTypes, ) -> CertificateBuilder: """ Sets the requestor's public key (as found in the signing request). """ if not isinstance( key, ( dsa.DSAPublicKey, rsa.RSAPublicKey, ec.EllipticCurvePublicKey, ed25519.Ed25519PublicKey, ed448.Ed448PublicKey, x25519.X25519PublicKey, x448.X448PublicKey, ), ): raise TypeError( "Expecting one of DSAPublicKey, RSAPublicKey," " EllipticCurvePublicKey, Ed25519PublicKey," " Ed448PublicKey, X25519PublicKey, or " "X448PublicKey." ) if self._public_key is not None: raise ValueError("The public key may only be set once.") return CertificateBuilder( self._issuer_name, self._subject_name, key, self._serial_number, self._not_valid_before, self._not_valid_after, self._extensions, ) def serial_number(self, number: int) -> CertificateBuilder: """ Sets the certificate serial number. """ if not isinstance(number, int): raise TypeError("Serial number must be of integral type.") if self._serial_number is not None: raise ValueError("The serial number may only be set once.") if number <= 0: raise ValueError("The serial number should be positive.") # ASN.1 integers are always signed, so most significant bit must be # zero. if number.bit_length() >= 160: # As defined in RFC 5280 raise ValueError( "The serial number should not be more than 159 " "bits." ) return CertificateBuilder( self._issuer_name, self._subject_name, self._public_key, number, self._not_valid_before, self._not_valid_after, self._extensions, ) def not_valid_before(self, time: datetime.datetime) -> CertificateBuilder: """ Sets the certificate activation time. """ if not isinstance(time, datetime.datetime): raise TypeError("Expecting datetime object.") if self._not_valid_before is not None: raise ValueError("The not valid before may only be set once.") time = _convert_to_naive_utc_time(time) if time < _EARLIEST_UTC_TIME: raise ValueError( "The not valid before date must be on or after" " 1950 January 1)." ) if self._not_valid_after is not None and time > self._not_valid_after: raise ValueError( "The not valid before date must be before the not valid after " "date." ) return CertificateBuilder( self._issuer_name, self._subject_name, self._public_key, self._serial_number, time, self._not_valid_after, self._extensions, ) def not_valid_after(self, time: datetime.datetime) -> CertificateBuilder: """ Sets the certificate expiration time. """ if not isinstance(time, datetime.datetime): raise TypeError("Expecting datetime object.") if self._not_valid_after is not None: raise ValueError("The not valid after may only be set once.") time = _convert_to_naive_utc_time(time) if time < _EARLIEST_UTC_TIME: raise ValueError( "The not valid after date must be on or after" " 1950 January 1." ) if ( self._not_valid_before is not None and time < self._not_valid_before ): raise ValueError( "The not valid after date must be after the not valid before " "date." ) return CertificateBuilder( self._issuer_name, self._subject_name, self._public_key, self._serial_number, self._not_valid_before, time, self._extensions, ) def add_extension( self, extval: ExtensionType, critical: bool ) -> CertificateBuilder: """ Adds an X.509 extension to the certificate. """ if not isinstance(extval, ExtensionType): raise TypeError("extension must be an ExtensionType") extension = Extension(extval.oid, critical, extval) _reject_duplicate_extension(extension, self._extensions) return CertificateBuilder( self._issuer_name, self._subject_name, self._public_key, self._serial_number, self._not_valid_before, self._not_valid_after, self._extensions + [extension], ) def sign( self, private_key: CertificateIssuerPrivateKeyTypes, algorithm: typing.Optional[_AllowedHashTypes], backend: typing.Any = None, , rsa_padding: typing.Optional[ typing.Union[padding.PSS, padding.PKCS1v15] ] = None, ) -> Certificate: """ Signs the certificate using the CA's private key. """ if self._subject_name is None: raise ValueError("A certificate must have a subject name") if self._issuer_name is None: raise ValueError("A certificate must have an issuer name") if self._serial_number is None: raise ValueError("A certificate must have a serial number") if self._not_valid_before is None: raise ValueError("A certificate must have a not valid before time") if self._not_valid_after is None: raise ValueError("A certificate must have a not valid after time") if self._public_key is None: raise ValueError("A certificate must have a public key") if rsa_padding is not None: if not isinstance(rsa_padding, (padding.PSS, padding.PKCS1v15)): raise TypeError("Padding must be PSS or PKCS1v15") if not isinstance(private_key, rsa.RSAPrivateKey): raise TypeError("Padding is only supported for RSA keys") return rust_x509.create_x509_certificate( self, private_key, algorithm, rsa_padding ) class CertificateRevocationListBuilder: _extensions: typing.List[Extension[ExtensionType]] _revoked_certificates: typing.List[RevokedCertificate] def __init__( self, issuer_name: typing.Optional[Name] = None, last_update: typing.Optional[datetime.datetime] = None, next_update: typing.Optional[datetime.datetime] = None, extensions: typing.List[Extension[ExtensionType]] = [], revoked_certificates: typing.List[RevokedCertificate] = [], ): self._issuer_name = issuer_name self._last_update = last_update self._next_update = next_update self._extensions = extensions self._revoked_certificates = revoked_certificates def issuer_name( self, issuer_name: Name ) -> CertificateRevocationListBuilder: if not isinstance(issuer_name, Name): raise TypeError("Expecting x509.Name object.") if self._issuer_name is not None: raise ValueError("The issuer name may only be set once.") return CertificateRevocationListBuilder( issuer_name, self._last_update, self._next_update, self._extensions, self._revoked_certificates, ) def last_update( self, last_update: datetime.datetime ) -> CertificateRevocationListBuilder: if not isinstance(last_update, datetime.datetime): raise TypeError("Expecting datetime object.") if self._last_update is not None: raise ValueError("Last update may only be set once.") last_update = _convert_to_naive_utc_time(last_update) if last_update < _EARLIEST_UTC_TIME: raise ValueError( "The last update date must be on or after" " 1950 January 1." ) if self._next_update is not None and last_update > self._next_update: raise ValueError( "The last update date must be before the next update date." ) return CertificateRevocationListBuilder( self._issuer_name, last_update, self._next_update, self._extensions, self._revoked_certificates, ) def next_update( self, next_update: datetime.datetime ) -> CertificateRevocationListBuilder: if not isinstance(next_update, datetime.datetime): raise TypeError("Expecting datetime object.") if self._next_update is not None: raise ValueError("Last update may only be set once.") next_update = _convert_to_naive_utc_time(next_update) if next_update < _EARLIEST_UTC_TIME: raise ValueError( "The last update date must be on or after" " 1950 January 1." ) if self._last_update is not None and next_update < self._last_update: raise ValueError( "The next update date must be after the last update date." ) return CertificateRevocationListBuilder( self._issuer_name, self._last_update, next_update, self._extensions, self._revoked_certificates, ) def add_extension( self, extval: ExtensionType, critical: bool ) -> CertificateRevocationListBuilder: """ Adds an X.509 extension to the certificate revocation list. """ if not isinstance(extval, ExtensionType): raise TypeError("extension must be an ExtensionType") extension = Extension(extval.oid, critical, extval) _reject_duplicate_extension(extension, self._extensions) return CertificateRevocationListBuilder( self._issuer_name, self._last_update, self._next_update, self._extensions + [extension], self._revoked_certificates, ) def add_revoked_certificate( self, revoked_certificate: RevokedCertificate ) -> CertificateRevocationListBuilder: """ Adds a revoked certificate to the CRL. """ if not isinstance(revoked_certificate, RevokedCertificate): raise TypeError("Must be an instance of RevokedCertificate") return CertificateRevocationListBuilder( self._issuer_name, self._last_update, self._next_update, self._extensions, self._revoked_certificates + [revoked_certificate], ) def sign( self, private_key: CertificateIssuerPrivateKeyTypes, algorithm: typing.Optional[_AllowedHashTypes], backend: typing.Any = None, ) -> CertificateRevocationList: if self._issuer_name is None: raise ValueError("A CRL must have an issuer name") if self._last_update is None: raise ValueError("A CRL must have a last update time") if self._next_update is None: raise ValueError("A CRL must have a next update time") return rust_x509.create_x509_crl(self, private_key, algorithm) class RevokedCertificateBuilder: def __init__( self, serial_number: typing.Optional[int] = None, revocation_date: typing.Optional[datetime.datetime] = None, extensions: typing.List[Extension[ExtensionType]] = [], ): self._serial_number = serial_number self._revocation_date = revocation_date self._extensions = extensions def serial_number(self, number: int) -> RevokedCertificateBuilder: if not isinstance(number, int): raise TypeError("Serial number must be of integral type.") if self._serial_number is not None: raise ValueError("The serial number may only be set once.") if number <= 0: raise ValueError("The serial number should be positive") # ASN.1 integers are always signed, so most significant bit must be # zero. if number.bit_length() >= 160: # As defined in RFC 5280 raise ValueError( "The serial number should not be more than 159 " "bits." ) return RevokedCertificateBuilder( number, self._revocation_date, self._extensions ) def revocation_date( self, time: datetime.datetime ) -> RevokedCertificateBuilder: if not isinstance(time, datetime.datetime): raise TypeError("Expecting datetime object.") if self._revocation_date is not None: raise ValueError("The revocation date may only be set once.") time = _convert_to_naive_utc_time(time) if time < _EARLIEST_UTC_TIME: raise ValueError( "The revocation date must be on or after" " 1950 January 1." ) return RevokedCertificateBuilder( self._serial_number, time, self._extensions ) def add_extension( self, extval: ExtensionType, critical: bool ) -> RevokedCertificateBuilder: if not isinstance(extval, ExtensionType): raise TypeError("extension must be an ExtensionType") extension = Extension(extval.oid, critical, extval) _reject_duplicate_extension(extension, self._extensions) return RevokedCertificateBuilder( self._serial_number, self._revocation_date, self._extensions + [extension], ) def build(self, backend: typing.Any = None) -> RevokedCertificate: if self._serial_number is None: raise ValueError("A revoked certificate must have a serial number") if self._revocation_date is None: raise ValueError( "A revoked certificate must have a revocation date" ) return _RawRevokedCertificate( self._serial_number, self._revocation_date, Extensions(self._extensions), ) def random_serial_number() -> int: return int.from_bytes(os.urandom(20), "big") >> 1	34.84 KB	2026-01-20 12:58
📄 certificate_transparency.py # This file is dual licensed under the terms of the Apache License, Version # 2.0, and the BSD License. See the LICENSE file in the root of this repository # for complete details. from __future__ import annotations import abc import datetime from cryptography import utils from cryptography.hazmat.bindings._rust import x509 as rust_x509 from cryptography.hazmat.primitives.hashes import HashAlgorithm class LogEntryType(utils.Enum): X509_CERTIFICATE = 0 PRE_CERTIFICATE = 1 class Version(utils.Enum): v1 = 0 class SignatureAlgorithm(utils.Enum): """ Signature algorithms that are valid for SCTs. These are exactly the same as SignatureAlgorithm in RFC 5246 (TLS 1.2). See: <https://datatracker.ietf.org/doc/html/rfc5246#section-7.4.1.4.1> """ ANONYMOUS = 0 RSA = 1 DSA = 2 ECDSA = 3 class SignedCertificateTimestamp(metaclass=abc.ABCMeta): @property @abc.abstractmethod def version(self) -> Version: """ Returns the SCT version. """ @property @abc.abstractmethod def log_id(self) -> bytes: """ Returns an identifier indicating which log this SCT is for. """ @property @abc.abstractmethod def timestamp(self) -> datetime.datetime: """ Returns the timestamp for this SCT. """ @property @abc.abstractmethod def entry_type(self) -> LogEntryType: """ Returns whether this is an SCT for a certificate or pre-certificate. """ @property @abc.abstractmethod def signature_hash_algorithm(self) -> HashAlgorithm: """ Returns the hash algorithm used for the SCT's signature. """ @property @abc.abstractmethod def signature_algorithm(self) -> SignatureAlgorithm: """ Returns the signing algorithm used for the SCT's signature. """ @property @abc.abstractmethod def signature(self) -> bytes: """ Returns the signature for this SCT. """ @property @abc.abstractmethod def extension_bytes(self) -> bytes: """ Returns the raw bytes of any extensions for this SCT. """ SignedCertificateTimestamp.register(rust_x509.Sct)	2.21 KB	2026-01-20 12:58
📄 extensions.py # This file is dual licensed under the terms of the Apache License, Version # 2.0, and the BSD License. See the LICENSE file in the root of this repository # for complete details. from __future__ import annotations import abc import datetime import hashlib import ipaddress import typing from cryptography import utils from cryptography.hazmat.bindings._rust import asn1 from cryptography.hazmat.bindings._rust import x509 as rust_x509 from cryptography.hazmat.primitives import constant_time, serialization from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePublicKey from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey from cryptography.hazmat.primitives.asymmetric.types import ( CertificateIssuerPublicKeyTypes, CertificatePublicKeyTypes, ) from cryptography.x509.certificate_transparency import ( SignedCertificateTimestamp, ) from cryptography.x509.general_name import ( DirectoryName, DNSName, GeneralName, IPAddress, OtherName, RegisteredID, RFC822Name, UniformResourceIdentifier, _IPAddressTypes, ) from cryptography.x509.name import Name, RelativeDistinguishedName from cryptography.x509.oid import ( CRLEntryExtensionOID, ExtensionOID, ObjectIdentifier, OCSPExtensionOID, ) ExtensionTypeVar = typing.TypeVar( "ExtensionTypeVar", bound="ExtensionType", covariant=True ) def _key_identifier_from_public_key( public_key: CertificatePublicKeyTypes, ) -> bytes: if isinstance(public_key, RSAPublicKey): data = public_key.public_bytes( serialization.Encoding.DER, serialization.PublicFormat.PKCS1, ) elif isinstance(public_key, EllipticCurvePublicKey): data = public_key.public_bytes( serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint, ) else: # This is a very slow way to do this. serialized = public_key.public_bytes( serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo, ) data = asn1.parse_spki_for_data(serialized) return hashlib.sha1(data).digest() def _make_sequence_methods(field_name: str): def len_method(self) -> int: return len(getattr(self, field_name)) def iter_method(self): return iter(getattr(self, field_name)) def getitem_method(self, idx): return getattr(self, field_name)[idx] return len_method, iter_method, getitem_method class DuplicateExtension(Exception): def __init__(self, msg: str, oid: ObjectIdentifier) -> None: super().__init__(msg) self.oid = oid class ExtensionNotFound(Exception): def __init__(self, msg: str, oid: ObjectIdentifier) -> None: super().__init__(msg) self.oid = oid class ExtensionType(metaclass=abc.ABCMeta): oid: typing.ClassVar[ObjectIdentifier] def public_bytes(self) -> bytes: """ Serializes the extension type to DER. """ raise NotImplementedError( "public_bytes is not implemented for extension type {!r}".format( self ) ) class Extensions: def __init__( self, extensions: typing.Iterable[Extension[ExtensionType]] ) -> None: self._extensions = list(extensions) def get_extension_for_oid( self, oid: ObjectIdentifier ) -> Extension[ExtensionType]: for ext in self: if ext.oid == oid: return ext raise ExtensionNotFound(f"No {oid} extension was found", oid) def get_extension_for_class( self, extclass: typing.Type[ExtensionTypeVar] ) -> Extension[ExtensionTypeVar]: if extclass is UnrecognizedExtension: raise TypeError( "UnrecognizedExtension can't be used with " "get_extension_for_class because more than one instance of the" " class may be present." ) for ext in self: if isinstance(ext.value, extclass): return ext raise ExtensionNotFound( f"No {extclass} extension was found", extclass.oid ) __len__, __iter__, __getitem__ = _make_sequence_methods("_extensions") def __repr__(self) -> str: return f"<Extensions({self._extensions})>" class CRLNumber(ExtensionType): oid = ExtensionOID.CRL_NUMBER def __init__(self, crl_number: int) -> None: if not isinstance(crl_number, int): raise TypeError("crl_number must be an integer") self._crl_number = crl_number def __eq__(self, other: object) -> bool: if not isinstance(other, CRLNumber): return NotImplemented return self.crl_number == other.crl_number def __hash__(self) -> int: return hash(self.crl_number) def __repr__(self) -> str: return f"<CRLNumber({self.crl_number})>" @property def crl_number(self) -> int: return self._crl_number def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class AuthorityKeyIdentifier(ExtensionType): oid = ExtensionOID.AUTHORITY_KEY_IDENTIFIER def __init__( self, key_identifier: typing.Optional[bytes], authority_cert_issuer: typing.Optional[typing.Iterable[GeneralName]], authority_cert_serial_number: typing.Optional[int], ) -> None: if (authority_cert_issuer is None) != ( authority_cert_serial_number is None ): raise ValueError( "authority_cert_issuer and authority_cert_serial_number " "must both be present or both None" ) if authority_cert_issuer is not None: authority_cert_issuer = list(authority_cert_issuer) if not all( isinstance(x, GeneralName) for x in authority_cert_issuer ): raise TypeError( "authority_cert_issuer must be a list of GeneralName " "objects" ) if authority_cert_serial_number is not None and not isinstance( authority_cert_serial_number, int ): raise TypeError("authority_cert_serial_number must be an integer") self._key_identifier = key_identifier self._authority_cert_issuer = authority_cert_issuer self._authority_cert_serial_number = authority_cert_serial_number # This takes a subset of CertificatePublicKeyTypes because an issuer # cannot have an X25519/X448 key. This introduces some unfortunate # asymmetry that requires typing users to explicitly # narrow their type, but we should make this accurate and not just # convenient. @classmethod def from_issuer_public_key( cls, public_key: CertificateIssuerPublicKeyTypes ) -> AuthorityKeyIdentifier: digest = _key_identifier_from_public_key(public_key) return cls( key_identifier=digest, authority_cert_issuer=None, authority_cert_serial_number=None, ) @classmethod def from_issuer_subject_key_identifier( cls, ski: SubjectKeyIdentifier ) -> AuthorityKeyIdentifier: return cls( key_identifier=ski.digest, authority_cert_issuer=None, authority_cert_serial_number=None, ) def __repr__(self) -> str: return ( "<AuthorityKeyIdentifier(key_identifier={0.key_identifier!r}, " "authority_cert_issuer={0.authority_cert_issuer}, " "authority_cert_serial_number={0.authority_cert_serial_number}" ")>".format(self) ) def __eq__(self, other: object) -> bool: if not isinstance(other, AuthorityKeyIdentifier): return NotImplemented return ( self.key_identifier == other.key_identifier and self.authority_cert_issuer == other.authority_cert_issuer and self.authority_cert_serial_number == other.authority_cert_serial_number ) def __hash__(self) -> int: if self.authority_cert_issuer is None: aci = None else: aci = tuple(self.authority_cert_issuer) return hash( (self.key_identifier, aci, self.authority_cert_serial_number) ) @property def key_identifier(self) -> typing.Optional[bytes]: return self._key_identifier @property def authority_cert_issuer( self, ) -> typing.Optional[typing.List[GeneralName]]: return self._authority_cert_issuer @property def authority_cert_serial_number(self) -> typing.Optional[int]: return self._authority_cert_serial_number def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class SubjectKeyIdentifier(ExtensionType): oid = ExtensionOID.SUBJECT_KEY_IDENTIFIER def __init__(self, digest: bytes) -> None: self._digest = digest @classmethod def from_public_key( cls, public_key: CertificatePublicKeyTypes ) -> SubjectKeyIdentifier: return cls(_key_identifier_from_public_key(public_key)) @property def digest(self) -> bytes: return self._digest @property def key_identifier(self) -> bytes: return self._digest def __repr__(self) -> str: return f"<SubjectKeyIdentifier(digest={self.digest!r})>" def __eq__(self, other: object) -> bool: if not isinstance(other, SubjectKeyIdentifier): return NotImplemented return constant_time.bytes_eq(self.digest, other.digest) def __hash__(self) -> int: return hash(self.digest) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class AuthorityInformationAccess(ExtensionType): oid = ExtensionOID.AUTHORITY_INFORMATION_ACCESS def __init__( self, descriptions: typing.Iterable[AccessDescription] ) -> None: descriptions = list(descriptions) if not all(isinstance(x, AccessDescription) for x in descriptions): raise TypeError( "Every item in the descriptions list must be an " "AccessDescription" ) self._descriptions = descriptions __len__, __iter__, __getitem__ = _make_sequence_methods("_descriptions") def __repr__(self) -> str: return f"<AuthorityInformationAccess({self._descriptions})>" def __eq__(self, other: object) -> bool: if not isinstance(other, AuthorityInformationAccess): return NotImplemented return self._descriptions == other._descriptions def __hash__(self) -> int: return hash(tuple(self._descriptions)) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class SubjectInformationAccess(ExtensionType): oid = ExtensionOID.SUBJECT_INFORMATION_ACCESS def __init__( self, descriptions: typing.Iterable[AccessDescription] ) -> None: descriptions = list(descriptions) if not all(isinstance(x, AccessDescription) for x in descriptions): raise TypeError( "Every item in the descriptions list must be an " "AccessDescription" ) self._descriptions = descriptions __len__, __iter__, __getitem__ = _make_sequence_methods("_descriptions") def __repr__(self) -> str: return f"<SubjectInformationAccess({self._descriptions})>" def __eq__(self, other: object) -> bool: if not isinstance(other, SubjectInformationAccess): return NotImplemented return self._descriptions == other._descriptions def __hash__(self) -> int: return hash(tuple(self._descriptions)) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class AccessDescription: def __init__( self, access_method: ObjectIdentifier, access_location: GeneralName ) -> None: if not isinstance(access_method, ObjectIdentifier): raise TypeError("access_method must be an ObjectIdentifier") if not isinstance(access_location, GeneralName): raise TypeError("access_location must be a GeneralName") self._access_method = access_method self._access_location = access_location def __repr__(self) -> str: return ( "<AccessDescription(access_method={0.access_method}, access_locati" "on={0.access_location})>".format(self) ) def __eq__(self, other: object) -> bool: if not isinstance(other, AccessDescription): return NotImplemented return ( self.access_method == other.access_method and self.access_location == other.access_location ) def __hash__(self) -> int: return hash((self.access_method, self.access_location)) @property def access_method(self) -> ObjectIdentifier: return self._access_method @property def access_location(self) -> GeneralName: return self._access_location class BasicConstraints(ExtensionType): oid = ExtensionOID.BASIC_CONSTRAINTS def __init__(self, ca: bool, path_length: typing.Optional[int]) -> None: if not isinstance(ca, bool): raise TypeError("ca must be a boolean value") if path_length is not None and not ca: raise ValueError("path_length must be None when ca is False") if path_length is not None and ( not isinstance(path_length, int) or path_length < 0 ): raise TypeError( "path_length must be a non-negative integer or None" ) self._ca = ca self._path_length = path_length @property def ca(self) -> bool: return self._ca @property def path_length(self) -> typing.Optional[int]: return self._path_length def __repr__(self) -> str: return ( "<BasicConstraints(ca={0.ca}, " "path_length={0.path_length})>" ).format(self) def __eq__(self, other: object) -> bool: if not isinstance(other, BasicConstraints): return NotImplemented return self.ca == other.ca and self.path_length == other.path_length def __hash__(self) -> int: return hash((self.ca, self.path_length)) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class DeltaCRLIndicator(ExtensionType): oid = ExtensionOID.DELTA_CRL_INDICATOR def __init__(self, crl_number: int) -> None: if not isinstance(crl_number, int): raise TypeError("crl_number must be an integer") self._crl_number = crl_number @property def crl_number(self) -> int: return self._crl_number def __eq__(self, other: object) -> bool: if not isinstance(other, DeltaCRLIndicator): return NotImplemented return self.crl_number == other.crl_number def __hash__(self) -> int: return hash(self.crl_number) def __repr__(self) -> str: return f"<DeltaCRLIndicator(crl_number={self.crl_number})>" def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class CRLDistributionPoints(ExtensionType): oid = ExtensionOID.CRL_DISTRIBUTION_POINTS def __init__( self, distribution_points: typing.Iterable[DistributionPoint] ) -> None: distribution_points = list(distribution_points) if not all( isinstance(x, DistributionPoint) for x in distribution_points ): raise TypeError( "distribution_points must be a list of DistributionPoint " "objects" ) self._distribution_points = distribution_points __len__, __iter__, __getitem__ = _make_sequence_methods( "_distribution_points" ) def __repr__(self) -> str: return f"<CRLDistributionPoints({self._distribution_points})>" def __eq__(self, other: object) -> bool: if not isinstance(other, CRLDistributionPoints): return NotImplemented return self._distribution_points == other._distribution_points def __hash__(self) -> int: return hash(tuple(self._distribution_points)) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class FreshestCRL(ExtensionType): oid = ExtensionOID.FRESHEST_CRL def __init__( self, distribution_points: typing.Iterable[DistributionPoint] ) -> None: distribution_points = list(distribution_points) if not all( isinstance(x, DistributionPoint) for x in distribution_points ): raise TypeError( "distribution_points must be a list of DistributionPoint " "objects" ) self._distribution_points = distribution_points __len__, __iter__, __getitem__ = _make_sequence_methods( "_distribution_points" ) def __repr__(self) -> str: return f"<FreshestCRL({self._distribution_points})>" def __eq__(self, other: object) -> bool: if not isinstance(other, FreshestCRL): return NotImplemented return self._distribution_points == other._distribution_points def __hash__(self) -> int: return hash(tuple(self._distribution_points)) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class DistributionPoint: def __init__( self, full_name: typing.Optional[typing.Iterable[GeneralName]], relative_name: typing.Optional[RelativeDistinguishedName], reasons: typing.Optional[typing.FrozenSet[ReasonFlags]], crl_issuer: typing.Optional[typing.Iterable[GeneralName]], ) -> None: if full_name and relative_name: raise ValueError( "You cannot provide both full_name and relative_name, at " "least one must be None." ) if not full_name and not relative_name and not crl_issuer: raise ValueError( "Either full_name, relative_name or crl_issuer must be " "provided." ) if full_name is not None: full_name = list(full_name) if not all(isinstance(x, GeneralName) for x in full_name): raise TypeError( "full_name must be a list of GeneralName objects" ) if relative_name: if not isinstance(relative_name, RelativeDistinguishedName): raise TypeError( "relative_name must be a RelativeDistinguishedName" ) if crl_issuer is not None: crl_issuer = list(crl_issuer) if not all(isinstance(x, GeneralName) for x in crl_issuer): raise TypeError( "crl_issuer must be None or a list of general names" ) if reasons and ( not isinstance(reasons, frozenset) or not all(isinstance(x, ReasonFlags) for x in reasons) ): raise TypeError("reasons must be None or frozenset of ReasonFlags") if reasons and ( ReasonFlags.unspecified in reasons or ReasonFlags.remove_from_crl in reasons ): raise ValueError( "unspecified and remove_from_crl are not valid reasons in a " "DistributionPoint" ) self._full_name = full_name self._relative_name = relative_name self._reasons = reasons self._crl_issuer = crl_issuer def __repr__(self) -> str: return ( "<DistributionPoint(full_name={0.full_name}, relative_name={0.rela" "tive_name}, reasons={0.reasons}, " "crl_issuer={0.crl_issuer})>".format(self) ) def __eq__(self, other: object) -> bool: if not isinstance(other, DistributionPoint): return NotImplemented return ( self.full_name == other.full_name and self.relative_name == other.relative_name and self.reasons == other.reasons and self.crl_issuer == other.crl_issuer ) def __hash__(self) -> int: if self.full_name is not None: fn: typing.Optional[typing.Tuple[GeneralName, ...]] = tuple( self.full_name ) else: fn = None if self.crl_issuer is not None: crl_issuer: typing.Optional[ typing.Tuple[GeneralName, ...] ] = tuple(self.crl_issuer) else: crl_issuer = None return hash((fn, self.relative_name, self.reasons, crl_issuer)) @property def full_name(self) -> typing.Optional[typing.List[GeneralName]]: return self._full_name @property def relative_name(self) -> typing.Optional[RelativeDistinguishedName]: return self._relative_name @property def reasons(self) -> typing.Optional[typing.FrozenSet[ReasonFlags]]: return self._reasons @property def crl_issuer(self) -> typing.Optional[typing.List[GeneralName]]: return self._crl_issuer class ReasonFlags(utils.Enum): unspecified = "unspecified" key_compromise = "keyCompromise" ca_compromise = "cACompromise" affiliation_changed = "affiliationChanged" superseded = "superseded" cessation_of_operation = "cessationOfOperation" certificate_hold = "certificateHold" privilege_withdrawn = "privilegeWithdrawn" aa_compromise = "aACompromise" remove_from_crl = "removeFromCRL" # These are distribution point bit string mappings. Not to be confused with # CRLReason reason flags bit string mappings. # ReasonFlags ::= BIT STRING { # unused (0), # keyCompromise (1), # cACompromise (2), # affiliationChanged (3), # superseded (4), # cessationOfOperation (5), # certificateHold (6), # privilegeWithdrawn (7), # aACompromise (8) } _REASON_BIT_MAPPING = { 1: ReasonFlags.key_compromise, 2: ReasonFlags.ca_compromise, 3: ReasonFlags.affiliation_changed, 4: ReasonFlags.superseded, 5: ReasonFlags.cessation_of_operation, 6: ReasonFlags.certificate_hold, 7: ReasonFlags.privilege_withdrawn, 8: ReasonFlags.aa_compromise, } _CRLREASONFLAGS = { ReasonFlags.key_compromise: 1, ReasonFlags.ca_compromise: 2, ReasonFlags.affiliation_changed: 3, ReasonFlags.superseded: 4, ReasonFlags.cessation_of_operation: 5, ReasonFlags.certificate_hold: 6, ReasonFlags.privilege_withdrawn: 7, ReasonFlags.aa_compromise: 8, } class PolicyConstraints(ExtensionType): oid = ExtensionOID.POLICY_CONSTRAINTS def __init__( self, require_explicit_policy: typing.Optional[int], inhibit_policy_mapping: typing.Optional[int], ) -> None: if require_explicit_policy is not None and not isinstance( require_explicit_policy, int ): raise TypeError( "require_explicit_policy must be a non-negative integer or " "None" ) if inhibit_policy_mapping is not None and not isinstance( inhibit_policy_mapping, int ): raise TypeError( "inhibit_policy_mapping must be a non-negative integer or None" ) if inhibit_policy_mapping is None and require_explicit_policy is None: raise ValueError( "At least one of require_explicit_policy and " "inhibit_policy_mapping must not be None" ) self._require_explicit_policy = require_explicit_policy self._inhibit_policy_mapping = inhibit_policy_mapping def __repr__(self) -> str: return ( "<PolicyConstraints(require_explicit_policy={0.require_explicit" "_policy}, inhibit_policy_mapping={0.inhibit_policy_" "mapping})>".format(self) ) def __eq__(self, other: object) -> bool: if not isinstance(other, PolicyConstraints): return NotImplemented return ( self.require_explicit_policy == other.require_explicit_policy and self.inhibit_policy_mapping == other.inhibit_policy_mapping ) def __hash__(self) -> int: return hash( (self.require_explicit_policy, self.inhibit_policy_mapping) ) @property def require_explicit_policy(self) -> typing.Optional[int]: return self._require_explicit_policy @property def inhibit_policy_mapping(self) -> typing.Optional[int]: return self._inhibit_policy_mapping def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class CertificatePolicies(ExtensionType): oid = ExtensionOID.CERTIFICATE_POLICIES def __init__(self, policies: typing.Iterable[PolicyInformation]) -> None: policies = list(policies) if not all(isinstance(x, PolicyInformation) for x in policies): raise TypeError( "Every item in the policies list must be a " "PolicyInformation" ) self._policies = policies __len__, __iter__, __getitem__ = _make_sequence_methods("_policies") def __repr__(self) -> str: return f"<CertificatePolicies({self._policies})>" def __eq__(self, other: object) -> bool: if not isinstance(other, CertificatePolicies): return NotImplemented return self._policies == other._policies def __hash__(self) -> int: return hash(tuple(self._policies)) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class PolicyInformation: def __init__( self, policy_identifier: ObjectIdentifier, policy_qualifiers: typing.Optional[ typing.Iterable[typing.Union[str, UserNotice]] ], ) -> None: if not isinstance(policy_identifier, ObjectIdentifier): raise TypeError("policy_identifier must be an ObjectIdentifier") self._policy_identifier = policy_identifier if policy_qualifiers is not None: policy_qualifiers = list(policy_qualifiers) if not all( isinstance(x, (str, UserNotice)) for x in policy_qualifiers ): raise TypeError( "policy_qualifiers must be a list of strings and/or " "UserNotice objects or None" ) self._policy_qualifiers = policy_qualifiers def __repr__(self) -> str: return ( "<PolicyInformation(policy_identifier={0.policy_identifier}, polic" "y_qualifiers={0.policy_qualifiers})>".format(self) ) def __eq__(self, other: object) -> bool: if not isinstance(other, PolicyInformation): return NotImplemented return ( self.policy_identifier == other.policy_identifier and self.policy_qualifiers == other.policy_qualifiers ) def __hash__(self) -> int: if self.policy_qualifiers is not None: pq: typing.Optional[ typing.Tuple[typing.Union[str, UserNotice], ...] ] = tuple(self.policy_qualifiers) else: pq = None return hash((self.policy_identifier, pq)) @property def policy_identifier(self) -> ObjectIdentifier: return self._policy_identifier @property def policy_qualifiers( self, ) -> typing.Optional[typing.List[typing.Union[str, UserNotice]]]: return self._policy_qualifiers class UserNotice: def __init__( self, notice_reference: typing.Optional[NoticeReference], explicit_text: typing.Optional[str], ) -> None: if notice_reference and not isinstance( notice_reference, NoticeReference ): raise TypeError( "notice_reference must be None or a NoticeReference" ) self._notice_reference = notice_reference self._explicit_text = explicit_text def __repr__(self) -> str: return ( "<UserNotice(notice_reference={0.notice_reference}, explicit_text=" "{0.explicit_text!r})>".format(self) ) def __eq__(self, other: object) -> bool: if not isinstance(other, UserNotice): return NotImplemented return ( self.notice_reference == other.notice_reference and self.explicit_text == other.explicit_text ) def __hash__(self) -> int: return hash((self.notice_reference, self.explicit_text)) @property def notice_reference(self) -> typing.Optional[NoticeReference]: return self._notice_reference @property def explicit_text(self) -> typing.Optional[str]: return self._explicit_text class NoticeReference: def __init__( self, organization: typing.Optional[str], notice_numbers: typing.Iterable[int], ) -> None: self._organization = organization notice_numbers = list(notice_numbers) if not all(isinstance(x, int) for x in notice_numbers): raise TypeError("notice_numbers must be a list of integers") self._notice_numbers = notice_numbers def __repr__(self) -> str: return ( "<NoticeReference(organization={0.organization!r}, notice_numbers=" "{0.notice_numbers})>".format(self) ) def __eq__(self, other: object) -> bool: if not isinstance(other, NoticeReference): return NotImplemented return ( self.organization == other.organization and self.notice_numbers == other.notice_numbers ) def __hash__(self) -> int: return hash((self.organization, tuple(self.notice_numbers))) @property def organization(self) -> typing.Optional[str]: return self._organization @property def notice_numbers(self) -> typing.List[int]: return self._notice_numbers class ExtendedKeyUsage(ExtensionType): oid = ExtensionOID.EXTENDED_KEY_USAGE def __init__(self, usages: typing.Iterable[ObjectIdentifier]) -> None: usages = list(usages) if not all(isinstance(x, ObjectIdentifier) for x in usages): raise TypeError( "Every item in the usages list must be an ObjectIdentifier" ) self._usages = usages __len__, __iter__, __getitem__ = _make_sequence_methods("_usages") def __repr__(self) -> str: return f"<ExtendedKeyUsage({self._usages})>" def __eq__(self, other: object) -> bool: if not isinstance(other, ExtendedKeyUsage): return NotImplemented return self._usages == other._usages def __hash__(self) -> int: return hash(tuple(self._usages)) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class OCSPNoCheck(ExtensionType): oid = ExtensionOID.OCSP_NO_CHECK def __eq__(self, other: object) -> bool: if not isinstance(other, OCSPNoCheck): return NotImplemented return True def __hash__(self) -> int: return hash(OCSPNoCheck) def __repr__(self) -> str: return "<OCSPNoCheck()>" def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class PrecertPoison(ExtensionType): oid = ExtensionOID.PRECERT_POISON def __eq__(self, other: object) -> bool: if not isinstance(other, PrecertPoison): return NotImplemented return True def __hash__(self) -> int: return hash(PrecertPoison) def __repr__(self) -> str: return "<PrecertPoison()>" def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class TLSFeature(ExtensionType): oid = ExtensionOID.TLS_FEATURE def __init__(self, features: typing.Iterable[TLSFeatureType]) -> None: features = list(features) if ( not all(isinstance(x, TLSFeatureType) for x in features) or len(features) == 0 ): raise TypeError( "features must be a list of elements from the TLSFeatureType " "enum" ) self._features = features __len__, __iter__, __getitem__ = _make_sequence_methods("_features") def __repr__(self) -> str: return f"<TLSFeature(features={self._features})>" def __eq__(self, other: object) -> bool: if not isinstance(other, TLSFeature): return NotImplemented return self._features == other._features def __hash__(self) -> int: return hash(tuple(self._features)) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class TLSFeatureType(utils.Enum): # status_request is defined in RFC 6066 and is used for what is commonly # called OCSP Must-Staple when present in the TLS Feature extension in an # X.509 certificate. status_request = 5 # status_request_v2 is defined in RFC 6961 and allows multiple OCSP # responses to be provided. It is not currently in use by clients or # servers. status_request_v2 = 17 _TLS_FEATURE_TYPE_TO_ENUM = {x.value: x for x in TLSFeatureType} class InhibitAnyPolicy(ExtensionType): oid = ExtensionOID.INHIBIT_ANY_POLICY def __init__(self, skip_certs: int) -> None: if not isinstance(skip_certs, int): raise TypeError("skip_certs must be an integer") if skip_certs < 0: raise ValueError("skip_certs must be a non-negative integer") self._skip_certs = skip_certs def __repr__(self) -> str: return f"<InhibitAnyPolicy(skip_certs={self.skip_certs})>" def __eq__(self, other: object) -> bool: if not isinstance(other, InhibitAnyPolicy): return NotImplemented return self.skip_certs == other.skip_certs def __hash__(self) -> int: return hash(self.skip_certs) @property def skip_certs(self) -> int: return self._skip_certs def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class KeyUsage(ExtensionType): oid = ExtensionOID.KEY_USAGE def __init__( self, digital_signature: bool, content_commitment: bool, key_encipherment: bool, data_encipherment: bool, key_agreement: bool, key_cert_sign: bool, crl_sign: bool, encipher_only: bool, decipher_only: bool, ) -> None: if not key_agreement and (encipher_only or decipher_only): raise ValueError( "encipher_only and decipher_only can only be true when " "key_agreement is true" ) self._digital_signature = digital_signature self._content_commitment = content_commitment self._key_encipherment = key_encipherment self._data_encipherment = data_encipherment self._key_agreement = key_agreement self._key_cert_sign = key_cert_sign self._crl_sign = crl_sign self._encipher_only = encipher_only self._decipher_only = decipher_only @property def digital_signature(self) -> bool: return self._digital_signature @property def content_commitment(self) -> bool: return self._content_commitment @property def key_encipherment(self) -> bool: return self._key_encipherment @property def data_encipherment(self) -> bool: return self._data_encipherment @property def key_agreement(self) -> bool: return self._key_agreement @property def key_cert_sign(self) -> bool: return self._key_cert_sign @property def crl_sign(self) -> bool: return self._crl_sign @property def encipher_only(self) -> bool: if not self.key_agreement: raise ValueError( "encipher_only is undefined unless key_agreement is true" ) else: return self._encipher_only @property def decipher_only(self) -> bool: if not self.key_agreement: raise ValueError( "decipher_only is undefined unless key_agreement is true" ) else: return self._decipher_only def __repr__(self) -> str: try: encipher_only = self.encipher_only decipher_only = self.decipher_only except ValueError: # Users found None confusing because even though encipher/decipher # have no meaning unless key_agreement is true, to construct an # instance of the class you still need to pass False. encipher_only = False decipher_only = False return ( "<KeyUsage(digital_signature={0.digital_signature}, " "content_commitment={0.content_commitment}, " "key_encipherment={0.key_encipherment}, " "data_encipherment={0.data_encipherment}, " "key_agreement={0.key_agreement}, " "key_cert_sign={0.key_cert_sign}, crl_sign={0.crl_sign}, " "encipher_only={1}, decipher_only={2})>" ).format(self, encipher_only, decipher_only) def __eq__(self, other: object) -> bool: if not isinstance(other, KeyUsage): return NotImplemented return ( self.digital_signature == other.digital_signature and self.content_commitment == other.content_commitment and self.key_encipherment == other.key_encipherment and self.data_encipherment == other.data_encipherment and self.key_agreement == other.key_agreement and self.key_cert_sign == other.key_cert_sign and self.crl_sign == other.crl_sign and self._encipher_only == other._encipher_only and self._decipher_only == other._decipher_only ) def __hash__(self) -> int: return hash( ( self.digital_signature, self.content_commitment, self.key_encipherment, self.data_encipherment, self.key_agreement, self.key_cert_sign, self.crl_sign, self._encipher_only, self._decipher_only, ) ) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class NameConstraints(ExtensionType): oid = ExtensionOID.NAME_CONSTRAINTS def __init__( self, permitted_subtrees: typing.Optional[typing.Iterable[GeneralName]], excluded_subtrees: typing.Optional[typing.Iterable[GeneralName]], ) -> None: if permitted_subtrees is not None: permitted_subtrees = list(permitted_subtrees) if not permitted_subtrees: raise ValueError( "permitted_subtrees must be a non-empty list or None" ) if not all(isinstance(x, GeneralName) for x in permitted_subtrees): raise TypeError( "permitted_subtrees must be a list of GeneralName objects " "or None" ) self._validate_tree(permitted_subtrees) if excluded_subtrees is not None: excluded_subtrees = list(excluded_subtrees) if not excluded_subtrees: raise ValueError( "excluded_subtrees must be a non-empty list or None" ) if not all(isinstance(x, GeneralName) for x in excluded_subtrees): raise TypeError( "excluded_subtrees must be a list of GeneralName objects " "or None" ) self._validate_tree(excluded_subtrees) if permitted_subtrees is None and excluded_subtrees is None: raise ValueError( "At least one of permitted_subtrees and excluded_subtrees " "must not be None" ) self._permitted_subtrees = permitted_subtrees self._excluded_subtrees = excluded_subtrees def __eq__(self, other: object) -> bool: if not isinstance(other, NameConstraints): return NotImplemented return ( self.excluded_subtrees == other.excluded_subtrees and self.permitted_subtrees == other.permitted_subtrees ) def _validate_tree(self, tree: typing.Iterable[GeneralName]) -> None: self._validate_ip_name(tree) self._validate_dns_name(tree) def _validate_ip_name(self, tree: typing.Iterable[GeneralName]) -> None: if any( isinstance(name, IPAddress) and not isinstance( name.value, (ipaddress.IPv4Network, ipaddress.IPv6Network) ) for name in tree ): raise TypeError( "IPAddress name constraints must be an IPv4Network or" " IPv6Network object" ) def _validate_dns_name(self, tree: typing.Iterable[GeneralName]) -> None: if any( isinstance(name, DNSName) and "" in name.value for name in tree ): raise ValueError( "DNSName name constraints must not contain the '' wildcard" " character" ) def __repr__(self) -> str: return ( "<NameConstraints(permitted_subtrees={0.permitted_subtrees}, " "excluded_subtrees={0.excluded_subtrees})>".format(self) ) def __hash__(self) -> int: if self.permitted_subtrees is not None: ps: typing.Optional[typing.Tuple[GeneralName, ...]] = tuple( self.permitted_subtrees ) else: ps = None if self.excluded_subtrees is not None: es: typing.Optional[typing.Tuple[GeneralName, ...]] = tuple( self.excluded_subtrees ) else: es = None return hash((ps, es)) @property def permitted_subtrees( self, ) -> typing.Optional[typing.List[GeneralName]]: return self._permitted_subtrees @property def excluded_subtrees( self, ) -> typing.Optional[typing.List[GeneralName]]: return self._excluded_subtrees def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class Extension(typing.Generic[ExtensionTypeVar]): def __init__( self, oid: ObjectIdentifier, critical: bool, value: ExtensionTypeVar ) -> None: if not isinstance(oid, ObjectIdentifier): raise TypeError( "oid argument must be an ObjectIdentifier instance." ) if not isinstance(critical, bool): raise TypeError("critical must be a boolean value") self._oid = oid self._critical = critical self._value = value @property def oid(self) -> ObjectIdentifier: return self._oid @property def critical(self) -> bool: return self._critical @property def value(self) -> ExtensionTypeVar: return self._value def __repr__(self) -> str: return ( "<Extension(oid={0.oid}, critical={0.critical}, " "value={0.value})>" ).format(self) def __eq__(self, other: object) -> bool: if not isinstance(other, Extension): return NotImplemented return ( self.oid == other.oid and self.critical == other.critical and self.value == other.value ) def __hash__(self) -> int: return hash((self.oid, self.critical, self.value)) class GeneralNames: def __init__(self, general_names: typing.Iterable[GeneralName]) -> None: general_names = list(general_names) if not all(isinstance(x, GeneralName) for x in general_names): raise TypeError( "Every item in the general_names list must be an " "object conforming to the GeneralName interface" ) self._general_names = general_names __len__, __iter__, __getitem__ = _make_sequence_methods("_general_names") @typing.overload def get_values_for_type( self, type: typing.Union[ typing.Type[DNSName], typing.Type[UniformResourceIdentifier], typing.Type[RFC822Name], ], ) -> typing.List[str]: ... @typing.overload def get_values_for_type( self, type: typing.Type[DirectoryName], ) -> typing.List[Name]: ... @typing.overload def get_values_for_type( self, type: typing.Type[RegisteredID], ) -> typing.List[ObjectIdentifier]: ... @typing.overload def get_values_for_type( self, type: typing.Type[IPAddress] ) -> typing.List[_IPAddressTypes]: ... @typing.overload def get_values_for_type( self, type: typing.Type[OtherName] ) -> typing.List[OtherName]: ... def get_values_for_type( self, type: typing.Union[ typing.Type[DNSName], typing.Type[DirectoryName], typing.Type[IPAddress], typing.Type[OtherName], typing.Type[RFC822Name], typing.Type[RegisteredID], typing.Type[UniformResourceIdentifier], ], ) -> typing.Union[ typing.List[_IPAddressTypes], typing.List[str], typing.List[OtherName], typing.List[Name], typing.List[ObjectIdentifier], ]: # Return the value of each GeneralName, except for OtherName instances # which we return directly because it has two important properties not # just one value. objs = (i for i in self if isinstance(i, type)) if type != OtherName: return [i.value for i in objs] return list(objs) def __repr__(self) -> str: return f"<GeneralNames({self._general_names})>" def __eq__(self, other: object) -> bool: if not isinstance(other, GeneralNames): return NotImplemented return self._general_names == other._general_names def __hash__(self) -> int: return hash(tuple(self._general_names)) class SubjectAlternativeName(ExtensionType): oid = ExtensionOID.SUBJECT_ALTERNATIVE_NAME def __init__(self, general_names: typing.Iterable[GeneralName]) -> None: self._general_names = GeneralNames(general_names) __len__, __iter__, __getitem__ = _make_sequence_methods("_general_names") @typing.overload def get_values_for_type( self, type: typing.Union[ typing.Type[DNSName], typing.Type[UniformResourceIdentifier], typing.Type[RFC822Name], ], ) -> typing.List[str]: ... @typing.overload def get_values_for_type( self, type: typing.Type[DirectoryName], ) -> typing.List[Name]: ... @typing.overload def get_values_for_type( self, type: typing.Type[RegisteredID], ) -> typing.List[ObjectIdentifier]: ... @typing.overload def get_values_for_type( self, type: typing.Type[IPAddress] ) -> typing.List[_IPAddressTypes]: ... @typing.overload def get_values_for_type( self, type: typing.Type[OtherName] ) -> typing.List[OtherName]: ... def get_values_for_type( self, type: typing.Union[ typing.Type[DNSName], typing.Type[DirectoryName], typing.Type[IPAddress], typing.Type[OtherName], typing.Type[RFC822Name], typing.Type[RegisteredID], typing.Type[UniformResourceIdentifier], ], ) -> typing.Union[ typing.List[_IPAddressTypes], typing.List[str], typing.List[OtherName], typing.List[Name], typing.List[ObjectIdentifier], ]: return self._general_names.get_values_for_type(type) def __repr__(self) -> str: return f"<SubjectAlternativeName({self._general_names})>" def __eq__(self, other: object) -> bool: if not isinstance(other, SubjectAlternativeName): return NotImplemented return self._general_names == other._general_names def __hash__(self) -> int: return hash(self._general_names) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class IssuerAlternativeName(ExtensionType): oid = ExtensionOID.ISSUER_ALTERNATIVE_NAME def __init__(self, general_names: typing.Iterable[GeneralName]) -> None: self._general_names = GeneralNames(general_names) __len__, __iter__, __getitem__ = _make_sequence_methods("_general_names") @typing.overload def get_values_for_type( self, type: typing.Union[ typing.Type[DNSName], typing.Type[UniformResourceIdentifier], typing.Type[RFC822Name], ], ) -> typing.List[str]: ... @typing.overload def get_values_for_type( self, type: typing.Type[DirectoryName], ) -> typing.List[Name]: ... @typing.overload def get_values_for_type( self, type: typing.Type[RegisteredID], ) -> typing.List[ObjectIdentifier]: ... @typing.overload def get_values_for_type( self, type: typing.Type[IPAddress] ) -> typing.List[_IPAddressTypes]: ... @typing.overload def get_values_for_type( self, type: typing.Type[OtherName] ) -> typing.List[OtherName]: ... def get_values_for_type( self, type: typing.Union[ typing.Type[DNSName], typing.Type[DirectoryName], typing.Type[IPAddress], typing.Type[OtherName], typing.Type[RFC822Name], typing.Type[RegisteredID], typing.Type[UniformResourceIdentifier], ], ) -> typing.Union[ typing.List[_IPAddressTypes], typing.List[str], typing.List[OtherName], typing.List[Name], typing.List[ObjectIdentifier], ]: return self._general_names.get_values_for_type(type) def __repr__(self) -> str: return f"<IssuerAlternativeName({self._general_names})>" def __eq__(self, other: object) -> bool: if not isinstance(other, IssuerAlternativeName): return NotImplemented return self._general_names == other._general_names def __hash__(self) -> int: return hash(self._general_names) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class CertificateIssuer(ExtensionType): oid = CRLEntryExtensionOID.CERTIFICATE_ISSUER def __init__(self, general_names: typing.Iterable[GeneralName]) -> None: self._general_names = GeneralNames(general_names) __len__, __iter__, __getitem__ = _make_sequence_methods("_general_names") @typing.overload def get_values_for_type( self, type: typing.Union[ typing.Type[DNSName], typing.Type[UniformResourceIdentifier], typing.Type[RFC822Name], ], ) -> typing.List[str]: ... @typing.overload def get_values_for_type( self, type: typing.Type[DirectoryName], ) -> typing.List[Name]: ... @typing.overload def get_values_for_type( self, type: typing.Type[RegisteredID], ) -> typing.List[ObjectIdentifier]: ... @typing.overload def get_values_for_type( self, type: typing.Type[IPAddress] ) -> typing.List[_IPAddressTypes]: ... @typing.overload def get_values_for_type( self, type: typing.Type[OtherName] ) -> typing.List[OtherName]: ... def get_values_for_type( self, type: typing.Union[ typing.Type[DNSName], typing.Type[DirectoryName], typing.Type[IPAddress], typing.Type[OtherName], typing.Type[RFC822Name], typing.Type[RegisteredID], typing.Type[UniformResourceIdentifier], ], ) -> typing.Union[ typing.List[_IPAddressTypes], typing.List[str], typing.List[OtherName], typing.List[Name], typing.List[ObjectIdentifier], ]: return self._general_names.get_values_for_type(type) def __repr__(self) -> str: return f"<CertificateIssuer({self._general_names})>" def __eq__(self, other: object) -> bool: if not isinstance(other, CertificateIssuer): return NotImplemented return self._general_names == other._general_names def __hash__(self) -> int: return hash(self._general_names) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class CRLReason(ExtensionType): oid = CRLEntryExtensionOID.CRL_REASON def __init__(self, reason: ReasonFlags) -> None: if not isinstance(reason, ReasonFlags): raise TypeError("reason must be an element from ReasonFlags") self._reason = reason def __repr__(self) -> str: return f"<CRLReason(reason={self._reason})>" def __eq__(self, other: object) -> bool: if not isinstance(other, CRLReason): return NotImplemented return self.reason == other.reason def __hash__(self) -> int: return hash(self.reason) @property def reason(self) -> ReasonFlags: return self._reason def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class InvalidityDate(ExtensionType): oid = CRLEntryExtensionOID.INVALIDITY_DATE def __init__(self, invalidity_date: datetime.datetime) -> None: if not isinstance(invalidity_date, datetime.datetime): raise TypeError("invalidity_date must be a datetime.datetime") self._invalidity_date = invalidity_date def __repr__(self) -> str: return "<InvalidityDate(invalidity_date={})>".format( self._invalidity_date ) def __eq__(self, other: object) -> bool: if not isinstance(other, InvalidityDate): return NotImplemented return self.invalidity_date == other.invalidity_date def __hash__(self) -> int: return hash(self.invalidity_date) @property def invalidity_date(self) -> datetime.datetime: return self._invalidity_date def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class PrecertificateSignedCertificateTimestamps(ExtensionType): oid = ExtensionOID.PRECERT_SIGNED_CERTIFICATE_TIMESTAMPS def __init__( self, signed_certificate_timestamps: typing.Iterable[ SignedCertificateTimestamp ], ) -> None: signed_certificate_timestamps = list(signed_certificate_timestamps) if not all( isinstance(sct, SignedCertificateTimestamp) for sct in signed_certificate_timestamps ): raise TypeError( "Every item in the signed_certificate_timestamps list must be " "a SignedCertificateTimestamp" ) self._signed_certificate_timestamps = signed_certificate_timestamps __len__, __iter__, __getitem__ = _make_sequence_methods( "_signed_certificate_timestamps" ) def __repr__(self) -> str: return "<PrecertificateSignedCertificateTimestamps({})>".format( list(self) ) def __hash__(self) -> int: return hash(tuple(self._signed_certificate_timestamps)) def __eq__(self, other: object) -> bool: if not isinstance(other, PrecertificateSignedCertificateTimestamps): return NotImplemented return ( self._signed_certificate_timestamps == other._signed_certificate_timestamps ) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class SignedCertificateTimestamps(ExtensionType): oid = ExtensionOID.SIGNED_CERTIFICATE_TIMESTAMPS def __init__( self, signed_certificate_timestamps: typing.Iterable[ SignedCertificateTimestamp ], ) -> None: signed_certificate_timestamps = list(signed_certificate_timestamps) if not all( isinstance(sct, SignedCertificateTimestamp) for sct in signed_certificate_timestamps ): raise TypeError( "Every item in the signed_certificate_timestamps list must be " "a SignedCertificateTimestamp" ) self._signed_certificate_timestamps = signed_certificate_timestamps __len__, __iter__, __getitem__ = _make_sequence_methods( "_signed_certificate_timestamps" ) def __repr__(self) -> str: return f"<SignedCertificateTimestamps({list(self)})>" def __hash__(self) -> int: return hash(tuple(self._signed_certificate_timestamps)) def __eq__(self, other: object) -> bool: if not isinstance(other, SignedCertificateTimestamps): return NotImplemented return ( self._signed_certificate_timestamps == other._signed_certificate_timestamps ) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class OCSPNonce(ExtensionType): oid = OCSPExtensionOID.NONCE def __init__(self, nonce: bytes) -> None: if not isinstance(nonce, bytes): raise TypeError("nonce must be bytes") self._nonce = nonce def __eq__(self, other: object) -> bool: if not isinstance(other, OCSPNonce): return NotImplemented return self.nonce == other.nonce def __hash__(self) -> int: return hash(self.nonce) def __repr__(self) -> str: return f"<OCSPNonce(nonce={self.nonce!r})>" @property def nonce(self) -> bytes: return self._nonce def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class OCSPAcceptableResponses(ExtensionType): oid = OCSPExtensionOID.ACCEPTABLE_RESPONSES def __init__(self, responses: typing.Iterable[ObjectIdentifier]) -> None: responses = list(responses) if any(not isinstance(r, ObjectIdentifier) for r in responses): raise TypeError("All responses must be ObjectIdentifiers") self._responses = responses def __eq__(self, other: object) -> bool: if not isinstance(other, OCSPAcceptableResponses): return NotImplemented return self._responses == other._responses def __hash__(self) -> int: return hash(tuple(self._responses)) def __repr__(self) -> str: return f"<OCSPAcceptableResponses(responses={self._responses})>" def __iter__(self) -> typing.Iterator[ObjectIdentifier]: return iter(self._responses) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class IssuingDistributionPoint(ExtensionType): oid = ExtensionOID.ISSUING_DISTRIBUTION_POINT def __init__( self, full_name: typing.Optional[typing.Iterable[GeneralName]], relative_name: typing.Optional[RelativeDistinguishedName], only_contains_user_certs: bool, only_contains_ca_certs: bool, only_some_reasons: typing.Optional[typing.FrozenSet[ReasonFlags]], indirect_crl: bool, only_contains_attribute_certs: bool, ) -> None: if full_name is not None: full_name = list(full_name) if only_some_reasons and ( not isinstance(only_some_reasons, frozenset) or not all(isinstance(x, ReasonFlags) for x in only_some_reasons) ): raise TypeError( "only_some_reasons must be None or frozenset of ReasonFlags" ) if only_some_reasons and ( ReasonFlags.unspecified in only_some_reasons or ReasonFlags.remove_from_crl in only_some_reasons ): raise ValueError( "unspecified and remove_from_crl are not valid reasons in an " "IssuingDistributionPoint" ) if not ( isinstance(only_contains_user_certs, bool) and isinstance(only_contains_ca_certs, bool) and isinstance(indirect_crl, bool) and isinstance(only_contains_attribute_certs, bool) ): raise TypeError( "only_contains_user_certs, only_contains_ca_certs, " "indirect_crl and only_contains_attribute_certs " "must all be boolean." ) crl_constraints = [ only_contains_user_certs, only_contains_ca_certs, indirect_crl, only_contains_attribute_certs, ] if len([x for x in crl_constraints if x]) > 1: raise ValueError( "Only one of the following can be set to True: " "only_contains_user_certs, only_contains_ca_certs, " "indirect_crl, only_contains_attribute_certs" ) if not any( [ only_contains_user_certs, only_contains_ca_certs, indirect_crl, only_contains_attribute_certs, full_name, relative_name, only_some_reasons, ] ): raise ValueError( "Cannot create empty extension: " "if only_contains_user_certs, only_contains_ca_certs, " "indirect_crl, and only_contains_attribute_certs are all False" ", then either full_name, relative_name, or only_some_reasons " "must have a value." ) self._only_contains_user_certs = only_contains_user_certs self._only_contains_ca_certs = only_contains_ca_certs self._indirect_crl = indirect_crl self._only_contains_attribute_certs = only_contains_attribute_certs self._only_some_reasons = only_some_reasons self._full_name = full_name self._relative_name = relative_name def __repr__(self) -> str: return ( "<IssuingDistributionPoint(full_name={0.full_name}, " "relative_name={0.relative_name}, " "only_contains_user_certs={0.only_contains_user_certs}, " "only_contains_ca_certs={0.only_contains_ca_certs}, " "only_some_reasons={0.only_some_reasons}, " "indirect_crl={0.indirect_crl}, " "only_contains_attribute_certs=" "{0.only_contains_attribute_certs})>".format(self) ) def __eq__(self, other: object) -> bool: if not isinstance(other, IssuingDistributionPoint): return NotImplemented return ( self.full_name == other.full_name and self.relative_name == other.relative_name and self.only_contains_user_certs == other.only_contains_user_certs and self.only_contains_ca_certs == other.only_contains_ca_certs and self.only_some_reasons == other.only_some_reasons and self.indirect_crl == other.indirect_crl and self.only_contains_attribute_certs == other.only_contains_attribute_certs ) def __hash__(self) -> int: return hash( ( self.full_name, self.relative_name, self.only_contains_user_certs, self.only_contains_ca_certs, self.only_some_reasons, self.indirect_crl, self.only_contains_attribute_certs, ) ) @property def full_name(self) -> typing.Optional[typing.List[GeneralName]]: return self._full_name @property def relative_name(self) -> typing.Optional[RelativeDistinguishedName]: return self._relative_name @property def only_contains_user_certs(self) -> bool: return self._only_contains_user_certs @property def only_contains_ca_certs(self) -> bool: return self._only_contains_ca_certs @property def only_some_reasons( self, ) -> typing.Optional[typing.FrozenSet[ReasonFlags]]: return self._only_some_reasons @property def indirect_crl(self) -> bool: return self._indirect_crl @property def only_contains_attribute_certs(self) -> bool: return self._only_contains_attribute_certs def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class MSCertificateTemplate(ExtensionType): oid = ExtensionOID.MS_CERTIFICATE_TEMPLATE def __init__( self, template_id: ObjectIdentifier, major_version: typing.Optional[int], minor_version: typing.Optional[int], ) -> None: if not isinstance(template_id, ObjectIdentifier): raise TypeError("oid must be an ObjectIdentifier") self._template_id = template_id if ( major_version is not None and not isinstance(major_version, int) ) or ( minor_version is not None and not isinstance(minor_version, int) ): raise TypeError( "major_version and minor_version must be integers or None" ) self._major_version = major_version self._minor_version = minor_version @property def template_id(self) -> ObjectIdentifier: return self._template_id @property def major_version(self) -> typing.Optional[int]: return self._major_version @property def minor_version(self) -> typing.Optional[int]: return self._minor_version def __repr__(self) -> str: return ( f"<MSCertificateTemplate(template_id={self.template_id}, " f"major_version={self.major_version}, " f"minor_version={self.minor_version})>" ) def __eq__(self, other: object) -> bool: if not isinstance(other, MSCertificateTemplate): return NotImplemented return ( self.template_id == other.template_id and self.major_version == other.major_version and self.minor_version == other.minor_version ) def __hash__(self) -> int: return hash((self.template_id, self.major_version, self.minor_version)) def public_bytes(self) -> bytes: return rust_x509.encode_extension_value(self) class UnrecognizedExtension(ExtensionType): def __init__(self, oid: ObjectIdentifier, value: bytes) -> None: if not isinstance(oid, ObjectIdentifier): raise TypeError("oid must be an ObjectIdentifier") self._oid = oid self._value = value @property def oid(self) -> ObjectIdentifier: # type: ignore[override] return self._oid @property def value(self) -> bytes: return self._value def __repr__(self) -> str: return ( "<UnrecognizedExtension(oid={0.oid}, " "value={0.value!r})>".format(self) ) def __eq__(self, other: object) -> bool: if not isinstance(other, UnrecognizedExtension): return NotImplemented return self.oid == other.oid and self.value == other.value def __hash__(self) -> int: return hash((self.oid, self.value)) def public_bytes(self) -> bytes: return self.value	66.76 KB	2026-01-20 12:58
📄 general_name.py # This file is dual licensed under the terms of the Apache License, Version # 2.0, and the BSD License. See the LICENSE file in the root of this repository # for complete details. from __future__ import annotations import abc import ipaddress import typing from email.utils import parseaddr from cryptography.x509.name import Name from cryptography.x509.oid import ObjectIdentifier _IPAddressTypes = typing.Union[ ipaddress.IPv4Address, ipaddress.IPv6Address, ipaddress.IPv4Network, ipaddress.IPv6Network, ] class UnsupportedGeneralNameType(Exception): pass class GeneralName(metaclass=abc.ABCMeta): @property @abc.abstractmethod def value(self) -> typing.Any: """ Return the value of the object """ class RFC822Name(GeneralName): def __init__(self, value: str) -> None: if isinstance(value, str): try: value.encode("ascii") except UnicodeEncodeError: raise ValueError( "RFC822Name values should be passed as an A-label string. " "This means unicode characters should be encoded via " "a library like idna." ) else: raise TypeError("value must be string") name, address = parseaddr(value) if name or not address: # parseaddr has found a name (e.g. Name <email>) or the entire # value is an empty string. raise ValueError("Invalid rfc822name value") self._value = value @property def value(self) -> str: return self._value @classmethod def _init_without_validation(cls, value: str) -> RFC822Name: instance = cls.__new__(cls) instance._value = value return instance def __repr__(self) -> str: return f"<RFC822Name(value={self.value!r})>" def __eq__(self, other: object) -> bool: if not isinstance(other, RFC822Name): return NotImplemented return self.value == other.value def __hash__(self) -> int: return hash(self.value) class DNSName(GeneralName): def __init__(self, value: str) -> None: if isinstance(value, str): try: value.encode("ascii") except UnicodeEncodeError: raise ValueError( "DNSName values should be passed as an A-label string. " "This means unicode characters should be encoded via " "a library like idna." ) else: raise TypeError("value must be string") self._value = value @property def value(self) -> str: return self._value @classmethod def _init_without_validation(cls, value: str) -> DNSName: instance = cls.__new__(cls) instance._value = value return instance def __repr__(self) -> str: return f"<DNSName(value={self.value!r})>" def __eq__(self, other: object) -> bool: if not isinstance(other, DNSName): return NotImplemented return self.value == other.value def __hash__(self) -> int: return hash(self.value) class UniformResourceIdentifier(GeneralName): def __init__(self, value: str) -> None: if isinstance(value, str): try: value.encode("ascii") except UnicodeEncodeError: raise ValueError( "URI values should be passed as an A-label string. " "This means unicode characters should be encoded via " "a library like idna." ) else: raise TypeError("value must be string") self._value = value @property def value(self) -> str: return self._value @classmethod def _init_without_validation(cls, value: str) -> UniformResourceIdentifier: instance = cls.__new__(cls) instance._value = value return instance def __repr__(self) -> str: return f"<UniformResourceIdentifier(value={self.value!r})>" def __eq__(self, other: object) -> bool: if not isinstance(other, UniformResourceIdentifier): return NotImplemented return self.value == other.value def __hash__(self) -> int: return hash(self.value) class DirectoryName(GeneralName): def __init__(self, value: Name) -> None: if not isinstance(value, Name): raise TypeError("value must be a Name") self._value = value @property def value(self) -> Name: return self._value def __repr__(self) -> str: return f"<DirectoryName(value={self.value})>" def __eq__(self, other: object) -> bool: if not isinstance(other, DirectoryName): return NotImplemented return self.value == other.value def __hash__(self) -> int: return hash(self.value) class RegisteredID(GeneralName): def __init__(self, value: ObjectIdentifier) -> None: if not isinstance(value, ObjectIdentifier): raise TypeError("value must be an ObjectIdentifier") self._value = value @property def value(self) -> ObjectIdentifier: return self._value def __repr__(self) -> str: return f"<RegisteredID(value={self.value})>" def __eq__(self, other: object) -> bool: if not isinstance(other, RegisteredID): return NotImplemented return self.value == other.value def __hash__(self) -> int: return hash(self.value) class IPAddress(GeneralName): def __init__(self, value: _IPAddressTypes) -> None: if not isinstance( value, ( ipaddress.IPv4Address, ipaddress.IPv6Address, ipaddress.IPv4Network, ipaddress.IPv6Network, ), ): raise TypeError( "value must be an instance of ipaddress.IPv4Address, " "ipaddress.IPv6Address, ipaddress.IPv4Network, or " "ipaddress.IPv6Network" ) self._value = value @property def value(self) -> _IPAddressTypes: return self._value def _packed(self) -> bytes: if isinstance( self.value, (ipaddress.IPv4Address, ipaddress.IPv6Address) ): return self.value.packed else: return ( self.value.network_address.packed + self.value.netmask.packed ) def __repr__(self) -> str: return f"<IPAddress(value={self.value})>" def __eq__(self, other: object) -> bool: if not isinstance(other, IPAddress): return NotImplemented return self.value == other.value def __hash__(self) -> int: return hash(self.value) class OtherName(GeneralName): def __init__(self, type_id: ObjectIdentifier, value: bytes) -> None: if not isinstance(type_id, ObjectIdentifier): raise TypeError("type_id must be an ObjectIdentifier") if not isinstance(value, bytes): raise TypeError("value must be a binary string") self._type_id = type_id self._value = value @property def type_id(self) -> ObjectIdentifier: return self._type_id @property def value(self) -> bytes: return self._value def __repr__(self) -> str: return "<OtherName(type_id={}, value={!r})>".format( self.type_id, self.value ) def __eq__(self, other: object) -> bool: if not isinstance(other, OtherName): return NotImplemented return self.type_id == other.type_id and self.value == other.value def __hash__(self) -> int: return hash((self.type_id, self.value))	7.68 KB	2026-01-20 12:58
📄 name.py # This file is dual licensed under the terms of the Apache License, Version # 2.0, and the BSD License. See the LICENSE file in the root of this repository # for complete details. from __future__ import annotations import binascii import re import sys import typing import warnings from cryptography import utils from cryptography.hazmat.bindings._rust import x509 as rust_x509 from cryptography.x509.oid import NameOID, ObjectIdentifier class _ASN1Type(utils.Enum): BitString = 3 OctetString = 4 UTF8String = 12 NumericString = 18 PrintableString = 19 T61String = 20 IA5String = 22 UTCTime = 23 GeneralizedTime = 24 VisibleString = 26 UniversalString = 28 BMPString = 30 _ASN1_TYPE_TO_ENUM = {i.value: i for i in _ASN1Type} _NAMEOID_DEFAULT_TYPE: typing.Dict[ObjectIdentifier, _ASN1Type] = { NameOID.COUNTRY_NAME: _ASN1Type.PrintableString, NameOID.JURISDICTION_COUNTRY_NAME: _ASN1Type.PrintableString, NameOID.SERIAL_NUMBER: _ASN1Type.PrintableString, NameOID.DN_QUALIFIER: _ASN1Type.PrintableString, NameOID.EMAIL_ADDRESS: _ASN1Type.IA5String, NameOID.DOMAIN_COMPONENT: _ASN1Type.IA5String, } # Type alias _OidNameMap = typing.Mapping[ObjectIdentifier, str] _NameOidMap = typing.Mapping[str, ObjectIdentifier] #: Short attribute names from RFC 4514: #: https://tools.ietf.org/html/rfc4514#page-7 _NAMEOID_TO_NAME: _OidNameMap = { NameOID.COMMON_NAME: "CN", NameOID.LOCALITY_NAME: "L", NameOID.STATE_OR_PROVINCE_NAME: "ST", NameOID.ORGANIZATION_NAME: "O", NameOID.ORGANIZATIONAL_UNIT_NAME: "OU", NameOID.COUNTRY_NAME: "C", NameOID.STREET_ADDRESS: "STREET", NameOID.DOMAIN_COMPONENT: "DC", NameOID.USER_ID: "UID", } _NAME_TO_NAMEOID = {v: k for k, v in _NAMEOID_TO_NAME.items()} def _escape_dn_value(val: typing.Union[str, bytes]) -> str: """Escape special characters in RFC4514 Distinguished Name value.""" if not val: return "" # RFC 4514 Section 2.4 defines the value as being the # (U+0023) character # followed by the hexadecimal encoding of the octets. if isinstance(val, bytes): return "#" + binascii.hexlify(val).decode("utf8") # See https://tools.ietf.org/html/rfc4514#section-2.4 val = val.replace("\\", "\\\\") val = val.replace('"', '\\"') val = val.replace("+", "\\+") val = val.replace(",", "\\,") val = val.replace(";", "\\;") val = val.replace("<", "\\<") val = val.replace(">", "\\>") val = val.replace("\0", "\\00") if val[0] in ("#", " "): val = "\\" + val if val[-1] == " ": val = val[:-1] + "\\ " return val def _unescape_dn_value(val: str) -> str: if not val: return "" # See https://tools.ietf.org/html/rfc4514#section-3 # special = escaped / SPACE / SHARP / EQUALS # escaped = DQUOTE / PLUS / COMMA / SEMI / LANGLE / RANGLE def sub(m): val = m.group(1) # Regular escape if len(val) == 1: return val # Hex-value scape return chr(int(val, 16)) return _RFC4514NameParser._PAIR_RE.sub(sub, val) class NameAttribute: def __init__( self, oid: ObjectIdentifier, value: typing.Union[str, bytes], _type: typing.Optional[_ASN1Type] = None, , _validate: bool = True, ) -> None: if not isinstance(oid, ObjectIdentifier): raise TypeError( "oid argument must be an ObjectIdentifier instance." ) if _type == _ASN1Type.BitString: if oid != NameOID.X500_UNIQUE_IDENTIFIER: raise TypeError( "oid must be X500_UNIQUE_IDENTIFIER for BitString type." ) if not isinstance(value, bytes): raise TypeError("value must be bytes for BitString") else: if not isinstance(value, str): raise TypeError("value argument must be a str") if ( oid == NameOID.COUNTRY_NAME or oid == NameOID.JURISDICTION_COUNTRY_NAME ): assert isinstance(value, str) c_len = len(value.encode("utf8")) if c_len != 2 and _validate is True: raise ValueError( "Country name must be a 2 character country code" ) elif c_len != 2: warnings.warn( "Country names should be two characters, but the " "attribute is {} characters in length.".format(c_len), stacklevel=2, ) # The appropriate ASN1 string type varies by OID and is defined across # multiple RFCs including 2459, 3280, and 5280. In general UTF8String # is preferred (2459), but 3280 and 5280 specify several OIDs with # alternate types. This means when we see the sentinel value we need # to look up whether the OID has a non-UTF8 type. If it does, set it # to that. Otherwise, UTF8! if _type is None: _type = _NAMEOID_DEFAULT_TYPE.get(oid, _ASN1Type.UTF8String) if not isinstance(_type, _ASN1Type): raise TypeError("_type must be from the _ASN1Type enum") self._oid = oid self._value = value self._type = _type @property def oid(self) -> ObjectIdentifier: return self._oid @property def value(self) -> typing.Union[str, bytes]: return self._value @property def rfc4514_attribute_name(self) -> str: """ The short attribute name (for example "CN") if available, otherwise the OID dotted string. """ return _NAMEOID_TO_NAME.get(self.oid, self.oid.dotted_string) def rfc4514_string( self, attr_name_overrides: typing.Optional[_OidNameMap] = None ) -> str: """ Format as RFC4514 Distinguished Name string. Use short attribute name if available, otherwise fall back to OID dotted string. """ attr_name = ( attr_name_overrides.get(self.oid) if attr_name_overrides else None ) if attr_name is None: attr_name = self.rfc4514_attribute_name return f"{attr_name}={_escape_dn_value(self.value)}" def __eq__(self, other: object) -> bool: if not isinstance(other, NameAttribute): return NotImplemented return self.oid == other.oid and self.value == other.value def __hash__(self) -> int: return hash((self.oid, self.value)) def __repr__(self) -> str: return "<NameAttribute(oid={0.oid}, value={0.value!r})>".format(self) class RelativeDistinguishedName: def __init__(self, attributes: typing.Iterable[NameAttribute]): attributes = list(attributes) if not attributes: raise ValueError("a relative distinguished name cannot be empty") if not all(isinstance(x, NameAttribute) for x in attributes): raise TypeError("attributes must be an iterable of NameAttribute") # Keep list and frozenset to preserve attribute order where it matters self._attributes = attributes self._attribute_set = frozenset(attributes) if len(self._attribute_set) != len(attributes): raise ValueError("duplicate attributes are not allowed") def get_attributes_for_oid( self, oid: ObjectIdentifier ) -> typing.List[NameAttribute]: return [i for i in self if i.oid == oid] def rfc4514_string( self, attr_name_overrides: typing.Optional[_OidNameMap] = None ) -> str: """ Format as RFC4514 Distinguished Name string. Within each RDN, attributes are joined by '+', although that is rarely used in certificates. """ return "+".join( attr.rfc4514_string(attr_name_overrides) for attr in self._attributes ) def __eq__(self, other: object) -> bool: if not isinstance(other, RelativeDistinguishedName): return NotImplemented return self._attribute_set == other._attribute_set def __hash__(self) -> int: return hash(self._attribute_set) def __iter__(self) -> typing.Iterator[NameAttribute]: return iter(self._attributes) def __len__(self) -> int: return len(self._attributes) def __repr__(self) -> str: return f"<RelativeDistinguishedName({self.rfc4514_string()})>" class Name: @typing.overload def __init__(self, attributes: typing.Iterable[NameAttribute]) -> None: ... @typing.overload def __init__( self, attributes: typing.Iterable[RelativeDistinguishedName] ) -> None: ... def __init__( self, attributes: typing.Iterable[ typing.Union[NameAttribute, RelativeDistinguishedName] ], ) -> None: attributes = list(attributes) if all(isinstance(x, NameAttribute) for x in attributes): self._attributes = [ RelativeDistinguishedName([typing.cast(NameAttribute, x)]) for x in attributes ] elif all(isinstance(x, RelativeDistinguishedName) for x in attributes): self._attributes = typing.cast( typing.List[RelativeDistinguishedName], attributes ) else: raise TypeError( "attributes must be a list of NameAttribute" " or a list RelativeDistinguishedName" ) @classmethod def from_rfc4514_string( cls, data: str, attr_name_overrides: typing.Optional[_NameOidMap] = None, ) -> Name: return _RFC4514NameParser(data, attr_name_overrides or {}).parse() def rfc4514_string( self, attr_name_overrides: typing.Optional[_OidNameMap] = None ) -> str: """ Format as RFC4514 Distinguished Name string. For example 'CN=foobar.com,O=Foo Corp,C=US' An X.509 name is a two-level structure: a list of sets of attributes. Each list element is separated by ',' and within each list element, set elements are separated by '+'. The latter is almost never used in real world certificates. According to RFC4514 section 2.1 the RDNSequence must be reversed when converting to string representation. """ return ",".join( attr.rfc4514_string(attr_name_overrides) for attr in reversed(self._attributes) ) def get_attributes_for_oid( self, oid: ObjectIdentifier ) -> typing.List[NameAttribute]: return [i for i in self if i.oid == oid] @property def rdns(self) -> typing.List[RelativeDistinguishedName]: return self._attributes def public_bytes(self, backend: typing.Any = None) -> bytes: return rust_x509.encode_name_bytes(self) def __eq__(self, other: object) -> bool: if not isinstance(other, Name): return NotImplemented return self._attributes == other._attributes def __hash__(self) -> int: # TODO: this is relatively expensive, if this looks like a bottleneck # for you, consider optimizing! return hash(tuple(self._attributes)) def __iter__(self) -> typing.Iterator[NameAttribute]: for rdn in self._attributes: for ava in rdn: yield ava def __len__(self) -> int: return sum(len(rdn) for rdn in self._attributes) def __repr__(self) -> str: rdns = ",".join(attr.rfc4514_string() for attr in self._attributes) return f"<Name({rdns})>" class _RFC4514NameParser: _OID_RE = re.compile(r"(0\|([1-9]\d))(\.(0\|([1-9]\d)))+") _DESCR_RE = re.compile(r"[a-zA-Z][a-zA-Z\d-]") _PAIR = r"\\([\\ #=\"\+,;<>]\|[\da-zA-Z]{2})" _PAIR_RE = re.compile(_PAIR) _LUTF1 = r"[\x01-\x1f\x21\x24-\x2A\x2D-\x3A\x3D\x3F-\x5B\x5D-\x7F]" _SUTF1 = r"[\x01-\x21\x23-\x2A\x2D-\x3A\x3D\x3F-\x5B\x5D-\x7F]" _TUTF1 = r"[\x01-\x1F\x21\x23-\x2A\x2D-\x3A\x3D\x3F-\x5B\x5D-\x7F]" _UTFMB = rf"[\x80-{chr(sys.maxunicode)}]" _LEADCHAR = rf"{_LUTF1}\|{_UTFMB}" _STRINGCHAR = rf"{_SUTF1}\|{_UTFMB}" _TRAILCHAR = rf"{_TUTF1}\|{_UTFMB}" _STRING_RE = re.compile( rf""" ( ({_LEADCHAR}\|{_PAIR}) ( ({_STRINGCHAR}\|{_PAIR})* ({_TRAILCHAR}\|{_PAIR}) )? )? """, re.VERBOSE, ) _HEXSTRING_RE = re.compile(r"#([\da-zA-Z]{2})+") def __init__(self, data: str, attr_name_overrides: _NameOidMap) -> None: self._data = data self._idx = 0 self._attr_name_overrides = attr_name_overrides def _has_data(self) -> bool: return self._idx < len(self._data) def _peek(self) -> typing.Optional[str]: if self._has_data(): return self._data[self._idx] return None def _read_char(self, ch: str) -> None: if self._peek() != ch: raise ValueError self._idx += 1 def _read_re(self, pat) -> str: match = pat.match(self._data, pos=self._idx) if match is None: raise ValueError val = match.group() self._idx += len(val) return val def parse(self) -> Name: """ Parses the `data` string and converts it to a Name. According to RFC4514 section 2.1 the RDNSequence must be reversed when converting to string representation. So, when we parse it, we need to reverse again to get the RDNs on the correct order. """ rdns = [self._parse_rdn()] while self._has_data(): self._read_char(",") rdns.append(self._parse_rdn()) return Name(reversed(rdns)) def _parse_rdn(self) -> RelativeDistinguishedName: nas = [self._parse_na()] while self._peek() == "+": self._read_char("+") nas.append(self._parse_na()) return RelativeDistinguishedName(nas) def _parse_na(self) -> NameAttribute: try: oid_value = self._read_re(self._OID_RE) except ValueError: name = self._read_re(self._DESCR_RE) oid = self._attr_name_overrides.get( name, _NAME_TO_NAMEOID.get(name) ) if oid is None: raise ValueError else: oid = ObjectIdentifier(oid_value) self._read_char("=") if self._peek() == "#": value = self._read_re(self._HEXSTRING_RE) value = binascii.unhexlify(value[1:]).decode() else: raw_value = self._read_re(self._STRING_RE) value = _unescape_dn_value(raw_value) return NameAttribute(oid, value)	14.51 KB	2026-01-20 12:58
📄 ocsp.py # This file is dual licensed under the terms of the Apache License, Version # 2.0, and the BSD License. See the LICENSE file in the root of this repository # for complete details. from __future__ import annotations import abc import datetime import typing from cryptography import utils, x509 from cryptography.hazmat.bindings._rust import ocsp from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric.types import ( CertificateIssuerPrivateKeyTypes, ) from cryptography.x509.base import ( _EARLIEST_UTC_TIME, _convert_to_naive_utc_time, _reject_duplicate_extension, ) class OCSPResponderEncoding(utils.Enum): HASH = "By Hash" NAME = "By Name" class OCSPResponseStatus(utils.Enum): SUCCESSFUL = 0 MALFORMED_REQUEST = 1 INTERNAL_ERROR = 2 TRY_LATER = 3 SIG_REQUIRED = 5 UNAUTHORIZED = 6 _ALLOWED_HASHES = ( hashes.SHA1, hashes.SHA224, hashes.SHA256, hashes.SHA384, hashes.SHA512, ) def _verify_algorithm(algorithm: hashes.HashAlgorithm) -> None: if not isinstance(algorithm, _ALLOWED_HASHES): raise ValueError( "Algorithm must be SHA1, SHA224, SHA256, SHA384, or SHA512" ) class OCSPCertStatus(utils.Enum): GOOD = 0 REVOKED = 1 UNKNOWN = 2 class _SingleResponse: def __init__( self, cert: x509.Certificate, issuer: x509.Certificate, algorithm: hashes.HashAlgorithm, cert_status: OCSPCertStatus, this_update: datetime.datetime, next_update: typing.Optional[datetime.datetime], revocation_time: typing.Optional[datetime.datetime], revocation_reason: typing.Optional[x509.ReasonFlags], ): if not isinstance(cert, x509.Certificate) or not isinstance( issuer, x509.Certificate ): raise TypeError("cert and issuer must be a Certificate") _verify_algorithm(algorithm) if not isinstance(this_update, datetime.datetime): raise TypeError("this_update must be a datetime object") if next_update is not None and not isinstance( next_update, datetime.datetime ): raise TypeError("next_update must be a datetime object or None") self._cert = cert self._issuer = issuer self._algorithm = algorithm self._this_update = this_update self._next_update = next_update if not isinstance(cert_status, OCSPCertStatus): raise TypeError( "cert_status must be an item from the OCSPCertStatus enum" ) if cert_status is not OCSPCertStatus.REVOKED: if revocation_time is not None: raise ValueError( "revocation_time can only be provided if the certificate " "is revoked" ) if revocation_reason is not None: raise ValueError( "revocation_reason can only be provided if the certificate" " is revoked" ) else: if not isinstance(revocation_time, datetime.datetime): raise TypeError("revocation_time must be a datetime object") revocation_time = _convert_to_naive_utc_time(revocation_time) if revocation_time < _EARLIEST_UTC_TIME: raise ValueError( "The revocation_time must be on or after" " 1950 January 1." ) if revocation_reason is not None and not isinstance( revocation_reason, x509.ReasonFlags ): raise TypeError( "revocation_reason must be an item from the ReasonFlags " "enum or None" ) self._cert_status = cert_status self._revocation_time = revocation_time self._revocation_reason = revocation_reason class OCSPRequest(metaclass=abc.ABCMeta): @property @abc.abstractmethod def issuer_key_hash(self) -> bytes: """ The hash of the issuer public key """ @property @abc.abstractmethod def issuer_name_hash(self) -> bytes: """ The hash of the issuer name """ @property @abc.abstractmethod def hash_algorithm(self) -> hashes.HashAlgorithm: """ The hash algorithm used in the issuer name and key hashes """ @property @abc.abstractmethod def serial_number(self) -> int: """ The serial number of the cert whose status is being checked """ @abc.abstractmethod def public_bytes(self, encoding: serialization.Encoding) -> bytes: """ Serializes the request to DER """ @property @abc.abstractmethod def extensions(self) -> x509.Extensions: """ The list of request extensions. Not single request extensions. """ class OCSPSingleResponse(metaclass=abc.ABCMeta): @property @abc.abstractmethod def certificate_status(self) -> OCSPCertStatus: """ The status of the certificate (an element from the OCSPCertStatus enum) """ @property @abc.abstractmethod def revocation_time(self) -> typing.Optional[datetime.datetime]: """ The date of when the certificate was revoked or None if not revoked. """ @property @abc.abstractmethod def revocation_reason(self) -> typing.Optional[x509.ReasonFlags]: """ The reason the certificate was revoked or None if not specified or not revoked. """ @property @abc.abstractmethod def this_update(self) -> datetime.datetime: """ The most recent time at which the status being indicated is known by the responder to have been correct """ @property @abc.abstractmethod def next_update(self) -> typing.Optional[datetime.datetime]: """ The time when newer information will be available """ @property @abc.abstractmethod def issuer_key_hash(self) -> bytes: """ The hash of the issuer public key """ @property @abc.abstractmethod def issuer_name_hash(self) -> bytes: """ The hash of the issuer name """ @property @abc.abstractmethod def hash_algorithm(self) -> hashes.HashAlgorithm: """ The hash algorithm used in the issuer name and key hashes """ @property @abc.abstractmethod def serial_number(self) -> int: """ The serial number of the cert whose status is being checked """ class OCSPResponse(metaclass=abc.ABCMeta): @property @abc.abstractmethod def responses(self) -> typing.Iterator[OCSPSingleResponse]: """ An iterator over the individual SINGLERESP structures in the response """ @property @abc.abstractmethod def response_status(self) -> OCSPResponseStatus: """ The status of the response. This is a value from the OCSPResponseStatus enumeration """ @property @abc.abstractmethod def signature_algorithm_oid(self) -> x509.ObjectIdentifier: """ The ObjectIdentifier of the signature algorithm """ @property @abc.abstractmethod def signature_hash_algorithm( self, ) -> typing.Optional[hashes.HashAlgorithm]: """ Returns a HashAlgorithm corresponding to the type of the digest signed """ @property @abc.abstractmethod def signature(self) -> bytes: """ The signature bytes """ @property @abc.abstractmethod def tbs_response_bytes(self) -> bytes: """ The tbsResponseData bytes """ @property @abc.abstractmethod def certificates(self) -> typing.List[x509.Certificate]: """ A list of certificates used to help build a chain to verify the OCSP response. This situation occurs when the OCSP responder uses a delegate certificate. """ @property @abc.abstractmethod def responder_key_hash(self) -> typing.Optional[bytes]: """ The responder's key hash or None """ @property @abc.abstractmethod def responder_name(self) -> typing.Optional[x509.Name]: """ The responder's Name or None """ @property @abc.abstractmethod def produced_at(self) -> datetime.datetime: """ The time the response was produced """ @property @abc.abstractmethod def certificate_status(self) -> OCSPCertStatus: """ The status of the certificate (an element from the OCSPCertStatus enum) """ @property @abc.abstractmethod def revocation_time(self) -> typing.Optional[datetime.datetime]: """ The date of when the certificate was revoked or None if not revoked. """ @property @abc.abstractmethod def revocation_reason(self) -> typing.Optional[x509.ReasonFlags]: """ The reason the certificate was revoked or None if not specified or not revoked. """ @property @abc.abstractmethod def this_update(self) -> datetime.datetime: """ The most recent time at which the status being indicated is known by the responder to have been correct """ @property @abc.abstractmethod def next_update(self) -> typing.Optional[datetime.datetime]: """ The time when newer information will be available """ @property @abc.abstractmethod def issuer_key_hash(self) -> bytes: """ The hash of the issuer public key """ @property @abc.abstractmethod def issuer_name_hash(self) -> bytes: """ The hash of the issuer name """ @property @abc.abstractmethod def hash_algorithm(self) -> hashes.HashAlgorithm: """ The hash algorithm used in the issuer name and key hashes """ @property @abc.abstractmethod def serial_number(self) -> int: """ The serial number of the cert whose status is being checked """ @property @abc.abstractmethod def extensions(self) -> x509.Extensions: """ The list of response extensions. Not single response extensions. """ @property @abc.abstractmethod def single_extensions(self) -> x509.Extensions: """ The list of single response extensions. Not response extensions. """ @abc.abstractmethod def public_bytes(self, encoding: serialization.Encoding) -> bytes: """ Serializes the response to DER """ class OCSPRequestBuilder: def __init__( self, request: typing.Optional[ typing.Tuple[ x509.Certificate, x509.Certificate, hashes.HashAlgorithm ] ] = None, request_hash: typing.Optional[ typing.Tuple[bytes, bytes, int, hashes.HashAlgorithm] ] = None, extensions: typing.List[x509.Extension[x509.ExtensionType]] = [], ) -> None: self._request = request self._request_hash = request_hash self._extensions = extensions def add_certificate( self, cert: x509.Certificate, issuer: x509.Certificate, algorithm: hashes.HashAlgorithm, ) -> OCSPRequestBuilder: if self._request is not None or self._request_hash is not None: raise ValueError("Only one certificate can be added to a request") _verify_algorithm(algorithm) if not isinstance(cert, x509.Certificate) or not isinstance( issuer, x509.Certificate ): raise TypeError("cert and issuer must be a Certificate") return OCSPRequestBuilder( (cert, issuer, algorithm), self._request_hash, self._extensions ) def add_certificate_by_hash( self, issuer_name_hash: bytes, issuer_key_hash: bytes, serial_number: int, algorithm: hashes.HashAlgorithm, ) -> OCSPRequestBuilder: if self._request is not None or self._request_hash is not None: raise ValueError("Only one certificate can be added to a request") if not isinstance(serial_number, int): raise TypeError("serial_number must be an integer") _verify_algorithm(algorithm) utils._check_bytes("issuer_name_hash", issuer_name_hash) utils._check_bytes("issuer_key_hash", issuer_key_hash) if algorithm.digest_size != len( issuer_name_hash ) or algorithm.digest_size != len(issuer_key_hash): raise ValueError( "issuer_name_hash and issuer_key_hash must be the same length " "as the digest size of the algorithm" ) return OCSPRequestBuilder( self._request, (issuer_name_hash, issuer_key_hash, serial_number, algorithm), self._extensions, ) def add_extension( self, extval: x509.ExtensionType, critical: bool ) -> OCSPRequestBuilder: if not isinstance(extval, x509.ExtensionType): raise TypeError("extension must be an ExtensionType") extension = x509.Extension(extval.oid, critical, extval) _reject_duplicate_extension(extension, self._extensions) return OCSPRequestBuilder( self._request, self._request_hash, self._extensions + [extension] ) def build(self) -> OCSPRequest: if self._request is None and self._request_hash is None: raise ValueError("You must add a certificate before building") return ocsp.create_ocsp_request(self) class OCSPResponseBuilder: def __init__( self, response: typing.Optional[_SingleResponse] = None, responder_id: typing.Optional[ typing.Tuple[x509.Certificate, OCSPResponderEncoding] ] = None, certs: typing.Optional[typing.List[x509.Certificate]] = None, extensions: typing.List[x509.Extension[x509.ExtensionType]] = [], ): self._response = response self._responder_id = responder_id self._certs = certs self._extensions = extensions def add_response( self, cert: x509.Certificate, issuer: x509.Certificate, algorithm: hashes.HashAlgorithm, cert_status: OCSPCertStatus, this_update: datetime.datetime, next_update: typing.Optional[datetime.datetime], revocation_time: typing.Optional[datetime.datetime], revocation_reason: typing.Optional[x509.ReasonFlags], ) -> OCSPResponseBuilder: if self._response is not None: raise ValueError("Only one response per OCSPResponse.") singleresp = _SingleResponse( cert, issuer, algorithm, cert_status, this_update, next_update, revocation_time, revocation_reason, ) return OCSPResponseBuilder( singleresp, self._responder_id, self._certs, self._extensions, ) def responder_id( self, encoding: OCSPResponderEncoding, responder_cert: x509.Certificate ) -> OCSPResponseBuilder: if self._responder_id is not None: raise ValueError("responder_id can only be set once") if not isinstance(responder_cert, x509.Certificate): raise TypeError("responder_cert must be a Certificate") if not isinstance(encoding, OCSPResponderEncoding): raise TypeError( "encoding must be an element from OCSPResponderEncoding" ) return OCSPResponseBuilder( self._response, (responder_cert, encoding), self._certs, self._extensions, ) def certificates( self, certs: typing.Iterable[x509.Certificate] ) -> OCSPResponseBuilder: if self._certs is not None: raise ValueError("certificates may only be set once") certs = list(certs) if len(certs) == 0: raise ValueError("certs must not be an empty list") if not all(isinstance(x, x509.Certificate) for x in certs): raise TypeError("certs must be a list of Certificates") return OCSPResponseBuilder( self._response, self._responder_id, certs, self._extensions, ) def add_extension( self, extval: x509.ExtensionType, critical: bool ) -> OCSPResponseBuilder: if not isinstance(extval, x509.ExtensionType): raise TypeError("extension must be an ExtensionType") extension = x509.Extension(extval.oid, critical, extval) _reject_duplicate_extension(extension, self._extensions) return OCSPResponseBuilder( self._response, self._responder_id, self._certs, self._extensions + [extension], ) def sign( self, private_key: CertificateIssuerPrivateKeyTypes, algorithm: typing.Optional[hashes.HashAlgorithm], ) -> OCSPResponse: if self._response is None: raise ValueError("You must add a response before signing") if self._responder_id is None: raise ValueError("You must add a responder_id before signing") return ocsp.create_ocsp_response( OCSPResponseStatus.SUCCESSFUL, self, private_key, algorithm ) @classmethod def build_unsuccessful( cls, response_status: OCSPResponseStatus ) -> OCSPResponse: if not isinstance(response_status, OCSPResponseStatus): raise TypeError( "response_status must be an item from OCSPResponseStatus" ) if response_status is OCSPResponseStatus.SUCCESSFUL: raise ValueError("response_status cannot be SUCCESSFUL") return ocsp.create_ocsp_response(response_status, None, None, None) def load_der_ocsp_request(data: bytes) -> OCSPRequest: return ocsp.load_der_ocsp_request(data) def load_der_ocsp_response(data: bytes) -> OCSPResponse: return ocsp.load_der_ocsp_response(data)	18.1 KB	2026-01-20 12:58
📄 oid.py # This file is dual licensed under the terms of the Apache License, Version # 2.0, and the BSD License. See the LICENSE file in the root of this repository # for complete details. from __future__ import annotations from cryptography.hazmat._oid import ( AttributeOID, AuthorityInformationAccessOID, CertificatePoliciesOID, CRLEntryExtensionOID, ExtendedKeyUsageOID, ExtensionOID, NameOID, ObjectIdentifier, OCSPExtensionOID, SignatureAlgorithmOID, SubjectInformationAccessOID, ) __all__ = [ "AttributeOID", "AuthorityInformationAccessOID", "CRLEntryExtensionOID", "CertificatePoliciesOID", "ExtendedKeyUsageOID", "ExtensionOID", "NameOID", "OCSPExtensionOID", "ObjectIdentifier", "SignatureAlgorithmOID", "SubjectInformationAccessOID", ]	829 B	2026-01-20 12:58

Terminal