codekingpro/portable-devtools

1# This file is dual licensed under the terms of the Apache License, Version2# 2.0, and the BSD License. See the LICENSE file in the root of this repository3# for complete details.4 5from __future__ import annotations6 7import datetime8from collections.abc import Iterable9 10from cryptography import utils, x50911from cryptography.hazmat.bindings._rust import ocsp12from cryptography.hazmat.primitives import hashes13from cryptography.hazmat.primitives.asymmetric.types import (14    CertificateIssuerPrivateKeyTypes,15)16from cryptography.x509.base import _reject_duplicate_extension17 18 19class OCSPResponderEncoding(utils.Enum):20    HASH = "By Hash"21    NAME = "By Name"22 23 24class OCSPResponseStatus(utils.Enum):25    SUCCESSFUL = 026    MALFORMED_REQUEST = 127    INTERNAL_ERROR = 228    TRY_LATER = 329    SIG_REQUIRED = 530    UNAUTHORIZED = 631 32 33_ALLOWED_HASHES = (34    hashes.SHA1,35    hashes.SHA224,36    hashes.SHA256,37    hashes.SHA384,38    hashes.SHA512,39)40 41 42def _verify_algorithm(algorithm: hashes.HashAlgorithm) -> None:43    if not isinstance(algorithm, _ALLOWED_HASHES):44        raise ValueError(45            "Algorithm must be SHA1, SHA224, SHA256, SHA384, or SHA512"46        )47 48 49class OCSPCertStatus(utils.Enum):50    GOOD = 051    REVOKED = 152    UNKNOWN = 253 54 55class _SingleResponse:56    def __init__(57        self,58        resp: tuple[x509.Certificate, x509.Certificate] | None,59        resp_hash: tuple[bytes, bytes, int] | None,60        algorithm: hashes.HashAlgorithm,61        cert_status: OCSPCertStatus,62        this_update: datetime.datetime,63        next_update: datetime.datetime | None,64        revocation_time: datetime.datetime | None,65        revocation_reason: x509.ReasonFlags | None,66    ):67        _verify_algorithm(algorithm)68        if not isinstance(this_update, datetime.datetime):69            raise TypeError("this_update must be a datetime object")70        if next_update is not None and not isinstance(71            next_update, datetime.datetime72        ):73            raise TypeError("next_update must be a datetime object or None")74 75        self._resp = resp76        self._resp_hash = resp_hash77        self._algorithm = algorithm78        self._this_update = this_update79        self._next_update = next_update80 81        if not isinstance(cert_status, OCSPCertStatus):82            raise TypeError(83                "cert_status must be an item from the OCSPCertStatus enum"84            )85        if cert_status is not OCSPCertStatus.REVOKED:86            if revocation_time is not None:87                raise ValueError(88                    "revocation_time can only be provided if the certificate "89                    "is revoked"90                )91            if revocation_reason is not None:92                raise ValueError(93                    "revocation_reason can only be provided if the certificate"94                    " is revoked"95                )96        else:97            if not isinstance(revocation_time, datetime.datetime):98                raise TypeError("revocation_time must be a datetime object")99 100            if revocation_reason is not None and not isinstance(101                revocation_reason, x509.ReasonFlags102            ):103                raise TypeError(104                    "revocation_reason must be an item from the ReasonFlags "105                    "enum or None"106                )107 108        self._cert_status = cert_status109        self._revocation_time = revocation_time110        self._revocation_reason = revocation_reason111 112 113OCSPRequest = ocsp.OCSPRequest114OCSPResponse = ocsp.OCSPResponse115OCSPSingleResponse = ocsp.OCSPSingleResponse116 117 118class OCSPRequestBuilder:119    def __init__(120        self,121        request: tuple[122            x509.Certificate, x509.Certificate, hashes.HashAlgorithm123        ]124        | None = None,125        request_hash: tuple[bytes, bytes, int, hashes.HashAlgorithm]126        | None = None,127        extensions: list[x509.Extension[x509.ExtensionType]] = [],128    ) -> None:129        self._request = request130        self._request_hash = request_hash131        self._extensions = extensions132 133    def add_certificate(134        self,135        cert: x509.Certificate,136        issuer: x509.Certificate,137        algorithm: hashes.HashAlgorithm,138    ) -> OCSPRequestBuilder:139        if self._request is not None or self._request_hash is not None:140            raise ValueError("Only one certificate can be added to a request")141 142        _verify_algorithm(algorithm)143        if not isinstance(cert, x509.Certificate) or not isinstance(144            issuer, x509.Certificate145        ):146            raise TypeError("cert and issuer must be a Certificate")147 148        return OCSPRequestBuilder(149            (cert, issuer, algorithm), self._request_hash, self._extensions150        )151 152    def add_certificate_by_hash(153        self,154        issuer_name_hash: bytes,155        issuer_key_hash: bytes,156        serial_number: int,157        algorithm: hashes.HashAlgorithm,158    ) -> OCSPRequestBuilder:159        if self._request is not None or self._request_hash is not None:160            raise ValueError("Only one certificate can be added to a request")161 162        if not isinstance(serial_number, int):163            raise TypeError("serial_number must be an integer")164 165        _verify_algorithm(algorithm)166        utils._check_bytes("issuer_name_hash", issuer_name_hash)167        utils._check_bytes("issuer_key_hash", issuer_key_hash)168        if algorithm.digest_size != len(169            issuer_name_hash170        ) or algorithm.digest_size != len(issuer_key_hash):171            raise ValueError(172                "issuer_name_hash and issuer_key_hash must be the same length "173                "as the digest size of the algorithm"174            )175 176        return OCSPRequestBuilder(177            self._request,178            (issuer_name_hash, issuer_key_hash, serial_number, algorithm),179            self._extensions,180        )181 182    def add_extension(183        self, extval: x509.ExtensionType, critical: bool184    ) -> OCSPRequestBuilder:185        if not isinstance(extval, x509.ExtensionType):186            raise TypeError("extension must be an ExtensionType")187 188        extension = x509.Extension(extval.oid, critical, extval)189        _reject_duplicate_extension(extension, self._extensions)190 191        return OCSPRequestBuilder(192            self._request, self._request_hash, [*self._extensions, extension]193        )194 195    def build(self) -> OCSPRequest:196        if self._request is None and self._request_hash is None:197            raise ValueError("You must add a certificate before building")198 199        return ocsp.create_ocsp_request(self)200 201 202class OCSPResponseBuilder:203    def __init__(204        self,205        response: _SingleResponse | None = None,206        responder_id: tuple[x509.Certificate, OCSPResponderEncoding]207        | None = None,208        certs: list[x509.Certificate] | None = None,209        extensions: list[x509.Extension[x509.ExtensionType]] = [],210    ):211        self._response = response212        self._responder_id = responder_id213        self._certs = certs214        self._extensions = extensions215 216    def add_response(217        self,218        cert: x509.Certificate,219        issuer: x509.Certificate,220        algorithm: hashes.HashAlgorithm,221        cert_status: OCSPCertStatus,222        this_update: datetime.datetime,223        next_update: datetime.datetime | None,224        revocation_time: datetime.datetime | None,225        revocation_reason: x509.ReasonFlags | None,226    ) -> OCSPResponseBuilder:227        if self._response is not None:228            raise ValueError("Only one response per OCSPResponse.")229 230        if not isinstance(cert, x509.Certificate) or not isinstance(231            issuer, x509.Certificate232        ):233            raise TypeError("cert and issuer must be a Certificate")234 235        singleresp = _SingleResponse(236            (cert, issuer),237            None,238            algorithm,239            cert_status,240            this_update,241            next_update,242            revocation_time,243            revocation_reason,244        )245        return OCSPResponseBuilder(246            singleresp,247            self._responder_id,248            self._certs,249            self._extensions,250        )251 252    def add_response_by_hash(253        self,254        issuer_name_hash: bytes,255        issuer_key_hash: bytes,256        serial_number: int,257        algorithm: hashes.HashAlgorithm,258        cert_status: OCSPCertStatus,259        this_update: datetime.datetime,260        next_update: datetime.datetime | None,261        revocation_time: datetime.datetime | None,262        revocation_reason: x509.ReasonFlags | None,263    ) -> OCSPResponseBuilder:264        if self._response is not None:265            raise ValueError("Only one response per OCSPResponse.")266 267        if not isinstance(serial_number, int):268            raise TypeError("serial_number must be an integer")269 270        utils._check_bytes("issuer_name_hash", issuer_name_hash)271        utils._check_bytes("issuer_key_hash", issuer_key_hash)272        _verify_algorithm(algorithm)273        if algorithm.digest_size != len(274            issuer_name_hash275        ) or algorithm.digest_size != len(issuer_key_hash):276            raise ValueError(277                "issuer_name_hash and issuer_key_hash must be the same length "278                "as the digest size of the algorithm"279            )280 281        singleresp = _SingleResponse(282            None,283            (issuer_name_hash, issuer_key_hash, serial_number),284            algorithm,285            cert_status,286            this_update,287            next_update,288            revocation_time,289            revocation_reason,290        )291        return OCSPResponseBuilder(292            singleresp,293            self._responder_id,294            self._certs,295            self._extensions,296        )297 298    def responder_id(299        self, encoding: OCSPResponderEncoding, responder_cert: x509.Certificate300    ) -> OCSPResponseBuilder:301        if self._responder_id is not None:302            raise ValueError("responder_id can only be set once")303        if not isinstance(responder_cert, x509.Certificate):304            raise TypeError("responder_cert must be a Certificate")305        if not isinstance(encoding, OCSPResponderEncoding):306            raise TypeError(307                "encoding must be an element from OCSPResponderEncoding"308            )309 310        return OCSPResponseBuilder(311            self._response,312            (responder_cert, encoding),313            self._certs,314            self._extensions,315        )316 317    def certificates(318        self, certs: Iterable[x509.Certificate]319    ) -> OCSPResponseBuilder:320        if self._certs is not None:321            raise ValueError("certificates may only be set once")322        certs = list(certs)323        if len(certs) == 0:324            raise ValueError("certs must not be an empty list")325        if not all(isinstance(x, x509.Certificate) for x in certs):326            raise TypeError("certs must be a list of Certificates")327        return OCSPResponseBuilder(328            self._response,329            self._responder_id,330            certs,331            self._extensions,332        )333 334    def add_extension(335        self, extval: x509.ExtensionType, critical: bool336    ) -> OCSPResponseBuilder:337        if not isinstance(extval, x509.ExtensionType):338            raise TypeError("extension must be an ExtensionType")339 340        extension = x509.Extension(extval.oid, critical, extval)341        _reject_duplicate_extension(extension, self._extensions)342 343        return OCSPResponseBuilder(344            self._response,345            self._responder_id,346            self._certs,347            [*self._extensions, extension],348        )349 350    def sign(351        self,352        private_key: CertificateIssuerPrivateKeyTypes,353        algorithm: hashes.HashAlgorithm | None,354    ) -> OCSPResponse:355        if self._response is None:356            raise ValueError("You must add a response before signing")357        if self._responder_id is None:358            raise ValueError("You must add a responder_id before signing")359 360        return ocsp.create_ocsp_response(361            OCSPResponseStatus.SUCCESSFUL, self, private_key, algorithm362        )363 364    @classmethod365    def build_unsuccessful(366        cls, response_status: OCSPResponseStatus367    ) -> OCSPResponse:368        if not isinstance(response_status, OCSPResponseStatus):369            raise TypeError(370                "response_status must be an item from OCSPResponseStatus"371            )372        if response_status is OCSPResponseStatus.SUCCESSFUL:373            raise ValueError("response_status cannot be SUCCESSFUL")374 375        return ocsp.create_ocsp_response(response_status, None, None, None)376 377 378load_der_ocsp_request = ocsp.load_der_ocsp_request379load_der_ocsp_response = ocsp.load_der_ocsp_response380