diff --git a/docs/source/features/apis/rest_api.rst b/docs/source/features/apis/rest_api.rst index b0d25ec67..034d12991 100644 --- a/docs/source/features/apis/rest_api.rst +++ b/docs/source/features/apis/rest_api.rst @@ -82,7 +82,9 @@ Interactive Documentation Trustpoint provides interactive API documentation using Swagger UI and ReDoc: -- **Swagger UI**: ``https:///api/schema/swagger-ui/`` +- **Swagger UI**: ``https:///swagger/`` +- **ReDoc**: ``https:///redoc/`` +- **OpenAPI schema**: ``https:///schema/`` The interactive documentation allows you to: @@ -96,4 +98,242 @@ Definition .. openapi:: ../../_static/trustpoint-openapi-swagger.yaml :group: - :examples: \ No newline at end of file + :examples: + +CA and CRL management +-------------------- + +CA and CRL administration requires authentication and the ``manage_cas`` +permission. Private keys and upstream authentication secrets are never included +in these responses. Upload actions accept ``multipart/form-data``; configuration +and enrollment actions accept JSON. + +CA inventories +^^^^^^^^^^^^^^ + +``GET /api/cas/`` and ``GET /api/cas/{id}/`` expose the full CA inventory, +including keyless CAs and autogenerated roots, with parent relationships, +certificate identifiers, and CRL settings. Filter by ``ca_type``, ``is_active``, +or ``parent_ca``. ``/api/issuing-cas/`` retains its issuing-CA-only inventory. + +For either inventory: + +* ``DELETE /api/cas/{id}/`` or ``DELETE /api/issuing-cas/{id}/`` deletes an + unused CA. Child CAs, unexpired issued certificates, and protected references + prevent deletion. +* ``GET /api/cas/{id}/certificate/?encoding=pem`` downloads its public + certificate. ``encoding=der`` is also supported. +* ``GET /api/cas/{id}/issued-certificates/`` lists its issued certificates, + including public PEM content and validity information. + +The certificate download and issued-certificate actions also exist under +``/api/issuing-cas/{id}/``. Downloads return 404 when the CA has no certificate. + +Creating issuing CAs +^^^^^^^^^^^^^^^^^^^^ + +The existing ``POST /api/issuing-cas/`` PEM import contract remains supported: +``private_key_pem``, ``ca_certificate_pem``, optional ``certificate_chain_pem``, +``private_key_password``, and ``unique_name``. + +Additional actions: + +* ``POST /api/issuing-cas/import-pkcs12/``: upload ``pkcs12_file`` with optional + ``pkcs12_password`` and ``unique_name``. +* ``POST /api/issuing-cas/generate/``: supply ``key_type`` to generate the local + AutoGenPKI. The security policy must enable AutoGenPKI, and an active + AutoGenPKI must not already exist. +* ``POST /api/issuing-cas/configure-est/`` and ``configure-cmp/``: configure an + upstream CA using ``unique_name``, ``key_type``, ``remote_host``, + ``remote_port``, and ``remote_path``. EST additionally requires + ``est_username`` and ``est_password``; CMP requires ``cmp_shared_secret``. + Set ``is_ra_mode`` to true for a registration authority configuration. +* ``POST /api/issuing-cas/configure-external-csr/``: supply ``unique_name`` and + ``key_type`` to create a managed key for an externally signed CA. + +The backend's supported key types and the active security policy apply to all +creation methods, as they do in the Web UI. + +Upstream enrollment and external signing +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Associate an existing truststore before upstream enrollment:: + + POST /api/issuing-cas/12/truststore/ + Content-Type: application/json + + {"trust_store": 5} + +The truststore's intended usage is validated. EST uses a TLS truststore; CMP +uses an Issuing CA Chain truststore. An EST RA first associates an Issuing CA +Chain, then a TLS truststore. The CA-chain step populates the RA certificate +and chain relationships. + +``POST /api/issuing-cas/{id}/enroll/`` requests the configured upstream EST or +CMP CA certificate. Certificate content is validated against the selected +profile; the default profile is ``issuing_ca``. For example:: + + { + "common_name": "Example Issuing CA", + "organization_name": "Example", + "country_name": "DE", + "state_or_province_name": "Berlin", + "days": 365 + } + +Use ``cert_profile_pk`` to select another profile. Subject/SAN/validity fields +follow that profile's requirements. CMP also accepts an optional integer +``sender_kid``. Upstream client errors return 502. + +For external signing, ``POST /api/issuing-cas/{id}/csr/`` accepts the same +certificate-content fields and returns a PEM CSR attachment. Then upload the +issued certificate as the multipart ``certificate`` field to +``POST /api/issuing-cas/{id}/import-certificate/``. The certificate must be a +CA certificate with the required signing usages and must match the managed +private key. Existing certificates cannot be replaced through these actions; +use a rollover instead. + +CRL configuration and inventory +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +``PATCH /api/issuing-cas/{id}/crl-settings/`` accepts any subset of:: + + { + "crl_cycle_enabled": true, + "crl_cycle_interval_hours": 12, + "crl_validity_hours": 24, + "auto_crl_on_revocation_enabled": true + } + +Omitted settings are preserved. Interval, validity, CA activation, and security +policy checks match the Web UI. Scheduling is updated through the existing +CA scheduling service. + +``POST /api/issuing-cas/{id}/generate-crl/`` and +``GET /api/issuing-cas/{id}/crl/?encoding=pem`` remain available. Fractional +validity hours are preserved when generating a CRL. + +The new inventory tracks historical and unassociated CRLs: + +* ``GET /api/crls/``: list CRLs; filter by ``ca``, ``is_active``, or ``crl_number``. +* ``POST /api/crls/``: upload ``crl_file`` in PEM or DER encoding, with optional + ``ca`` and ``set_active`` (default true). When a CA is supplied, its subject + and signature must match the CRL. Unassociated CRLs remain inactive. +* ``GET /api/crls/{id}/``: inspect metadata, issuer, PEM contents, and revocations. + Serial numbers are hexadecimal strings to preserve their full precision. +* ``GET /api/crls/{id}/download/?encoding=der``: download PEM or DER contents. +* ``DELETE /api/crls/{id}/``: delete a CRL. + +CRL identifiers always refer to CRL records, without falling back to CA IDs. +Importing an active CRL deactivates the previous active CRL in the same +transaction. Duplicate imports return 409 without changing the active CRL. + +CA rollovers +^^^^^^^^^^^^ + +``GET /api/ca-rollovers/`` lists rollover history. Filter by ``old_issuing_ca``, +``new_issuing_ca``, or ``state``; retrieve one with +``GET /api/ca-rollovers/{id}/``. + +Plan a rollover with ``POST /api/ca-rollovers/``. Supply ``old_issuing_ca`` and +one of these ``method`` values: + +* ``pkcs12``: ``pkcs12_file``, optional ``pkcs12_password`` and ``unique_name``. +* ``separate_files``: ``private_key_file``, ``ca_certificate``, optional + ``private_key_file_password``, ``ca_certificate_chain``, and ``unique_name``. +* ``est`` or ``cmp``: the corresponding upstream connection/authentication + fields and ``key_type`` described above, with a new ``unique_name``. + +Optional ``transition_scheduled_at`` and ``notes`` apply to every method. +Generated-key replacements remain in ``awaiting_new_ca`` until their CA +certificate is enrolled or uploaded through the issuing CA API. + +Advance the rollover through dedicated POST actions: + +* ``/api/ca-rollovers/{id}/start/`` +* ``/api/ca-rollovers/{id}/transition/`` +* ``/api/ca-rollovers/{id}/complete/`` +* ``/api/ca-rollovers/{id}/cancel/`` + +These actions enforce the existing rollover state machine and write the same +audit events as the Web UI. Invalid state transitions return 409. Direct +updates to rollover state fields are not supported. + + +Certificate details and downloads +-------------------------------- + +Certificate reads and downloads require authentication, using the existing +certificate API access policy. + +``GET /api/certificates/`` now includes serial numbers, SHA-256 fingerprints, +validity dates, status, CA/self-signed flags, issuer certificate identifiers, +and public key algorithm/size. The original ``id`` and ``common_name`` fields +remain available. Existing search, filters, ordering, and pagination apply. + +``GET /api/certificates/{id}/`` additionally provides: + +* Subject and issuer attributes with OIDs, labels, and RDN grouping. Repeated + distinguished-name attributes are preserved. +* Signature algorithm, signature value, public key information, and PEM contents. +* Every X.509 extension with its OID, criticality, decoded common-extension + values, and exact DER contents as ``der_hex``. Other extension values have + a readable representation alongside their DER contents. +* The known issuer chain as certificate identifiers, from direct issuer to root. +* Revocation date, reason, and CA identifier, or ``null`` when not revoked. +* Associated local issued-credential/device/profile identifiers and metadata, + plus the number of certificates referencing this certificate as their issuer. + +Serial numbers and fingerprints are hexadecimal strings, preserving their full +precision. Status codes are ``OK``, ``REV`` (revoked), ``EXP`` (expired), and +``NYV`` (not yet valid); ``status_display`` provides a readable label. X.509 +``version`` uses its stored zero-based enumeration (``2`` means version 3). + +Single certificate download +^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Download a public certificate with:: + + GET /api/certificates/42/download/?file_format=pem + +Supported ``file_format`` values are ``pem`` (default), ``der``, ``pkcs7_pem``, +and ``pkcs7_der``. Responses are binary attachments with the appropriate MIME +type and filename extension. Filenames include certificate IDs so certificates +with identical common names remain distinguishable. + +Set ``include_chain=true`` to include known issuer certificates:: + + GET /api/certificates/42/download/?file_format=pkcs7_pem&include_chain=true + +PEM downloads place the selected certificate before its issuers. PKCS#7 +containers include the same certificate set. A single DER file cannot contain +multiple certificates: use PEM, PKCS#7, or an archive to download a chain. +Only issuer certificates already present in Trustpoint are included; no +external certificates are fetched. + +Multiple certificate download +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Request an archive using JSON:: + + POST /api/certificates/download/ + Content-Type: application/json + + { + "certificate_ids": [42, 43], + "file_format": "pem", + "archive_format": "zip", + "include_chain": true + } + +``certificate_ids`` must contain between 1 and 1000 positive integers. +``archive_format`` supports ``zip`` (default) and ``tar_gz``. Every selected +certificate is a separate file in the chosen certificate format. With +``include_chain``, known issuer certificates are also separate files, including +when DER is selected. Duplicate selections and shared issuer certificates +appear only once. Archive member names are safe, unique filenames. + +Unknown certificate IDs return 404 without producing a partial archive. +Unsupported formats, invalid options, and cyclic issuer relationships return +400. Downloads contain public certificates only, without private keys or +credential bundles. diff --git a/trustpoint/pki/api_urls.py b/trustpoint/pki/api_urls.py index 0fb0bbf2d..18bd30e01 100644 --- a/trustpoint/pki/api_urls.py +++ b/trustpoint/pki/api_urls.py @@ -7,17 +7,20 @@ """ from rest_framework.routers import DefaultRouter +from pki.views.ca_api import CaRolloverViewSet, CaViewSet, CrlViewSet, IssuingCaManagementViewSet from pki.views.cert_profiles import CertProfileViewSet from pki.views.certificates import CertificateViewSet from pki.views.domains import DevIdRegistrationViewSet, DomainViewSet -from pki.views.issuing_cas import IssuingCaViewSet from pki.views.owner_credentials_api import DevOwnerIdViewSet from pki.views.truststores import TruststoreViewSet router = DefaultRouter() router.register(r'certificates', CertificateViewSet, basename='certificate') router.register(r'cert-profiles', CertProfileViewSet, basename='cert-profiles') -router.register(r'issuing-cas', IssuingCaViewSet, basename='issuing-ca') +router.register(r'issuing-cas', IssuingCaManagementViewSet, basename='issuing-ca') +router.register(r'cas', CaViewSet, basename='ca') +router.register(r'crls', CrlViewSet, basename='crl-api') +router.register(r'ca-rollovers', CaRolloverViewSet, basename='ca-rollover') router.register(r'truststores', TruststoreViewSet, basename='truststore') router.register(r'domains', DomainViewSet, basename='domain') router.register(r'devid-registrations', DevIdRegistrationViewSet, basename='devid-registration') diff --git a/trustpoint/pki/models/ca.py b/trustpoint/pki/models/ca.py index 6ac4af462..6e986e3af 100644 --- a/trustpoint/pki/models/ca.py +++ b/trustpoint/pki/models/ca.py @@ -729,7 +729,7 @@ def create_new_issuing_ca( return issuing_ca - def _issue_crl(self, crl_validity_hours: int = 24) -> None: + def _issue_crl(self, crl_validity_hours: float = 24) -> None: """Issues a CRL with revoked certificates issued by this CA. Only issuing CAs can issue CRLs (keyless CAs don't have private keys). @@ -750,7 +750,7 @@ def _issue_crl(self, crl_validity_hours: int = 24) -> None: self.logger.info('CRL generation for CA %s finished.', self.unique_name) - def issue_crl(self, crl_validity_hours: int = 24) -> bool: + def issue_crl(self, crl_validity_hours: float = 24) -> bool: """Issues a CRL with revoked certificates issued by this CA. Args: diff --git a/trustpoint/pki/serializer/ca_api.py b/trustpoint/pki/serializer/ca_api.py new file mode 100644 index 000000000..59c4f38ec --- /dev/null +++ b/trustpoint/pki/serializer/ca_api.py @@ -0,0 +1,233 @@ +# Copyright (c) 2026 The Trustpoint Project Authors +# SPDX-License-Identifier: MIT + +"""Public request and response contracts for CA and CRL management.""" + +from typing import Any, ClassVar + +from rest_framework import serializers + +from pki.models import CaModel, CertificateModel, CrlModel +from pki.models.ca_rollover import CaRolloverModel +from pki.serializer.issuing_ca import IssuingCaSerializer +from pki.serializer.owner_credential import CertificateIssuanceContentSerializer + + +class CaSerializer(IssuingCaSerializer): + """CA inventory, hierarchy, and public configuration; never private key material.""" + + certificate = serializers.SerializerMethodField() + next_crl_generation_scheduled_at = serializers.DateTimeField(read_only=True, allow_null=True) + + def get_certificate(self, obj: CaModel) -> int | None: + """Return the public certificate ID for both keyless and credential-backed CAs.""" + certificate = obj.ca_certificate_model + return certificate.pk if certificate is not None else None + + class Meta(IssuingCaSerializer.Meta): + """Include settings needed to manage an existing CA.""" + + fields: ClassVar[list[str]] = [ + *IssuingCaSerializer.Meta.fields, + 'parent_ca', + 'certificate', + 'chain_truststore', + 'remote_host', + 'remote_port', + 'remote_path', + 'est_username', + 'crl_cycle_enabled', + 'crl_cycle_interval_hours', + 'crl_validity_hours', + 'auto_crl_on_revocation_enabled', + 'next_crl_generation_scheduled_at', + ] + read_only_fields = fields + + +class CaCertificateSerializer(serializers.ModelSerializer[CertificateModel]): + """Public certificate information for CA inventories.""" + + class Meta: + """Expose certificate contents without the associated private key.""" + + model = CertificateModel + fields = ('id', 'common_name', 'serial_number', 'not_valid_before', 'not_valid_after', 'cert_pem') + read_only_fields = fields + + +class CrlSerializer(serializers.ModelSerializer[CrlModel]): + """CRL metadata and contents.""" + + revoked_serial_numbers = serializers.SerializerMethodField() + issuer = serializers.SerializerMethodField() + revoked_certificates = serializers.SerializerMethodField() + + class Meta: + """CRLs are immutable after import or generation.""" + + model = CrlModel + fields = ( + 'id', + 'ca', + 'crl_number', + 'this_update', + 'next_update', + 'is_active', + 'created_at', + 'updated_at', + 'crl_pem', + 'revoked_serial_numbers', + 'issuer', + 'revoked_certificates', + ) + read_only_fields = fields + + def get_revoked_serial_numbers(self, obj: CrlModel) -> list[str]: + """Use hexadecimal strings so large certificate serials remain lossless in JSON.""" + return [f'{number:X}' for number in sorted(obj.get_revoked_serial_numbers())] + + def get_issuer(self, obj: CrlModel) -> str: + """Return the CRL issuer distinguished name.""" + return obj.get_crl_as_crypto().issuer.rfc4514_string() + + def get_revoked_certificates(self, obj: CrlModel) -> list[dict[str, str | None]]: + """Expose revocation timestamps and reasons without losing large serial numbers.""" + from cryptography import x509 # noqa: PLC0415 + + entries = [] + for revoked in obj.get_crl_as_crypto(): + try: + reason = revoked.extensions.get_extension_for_class(x509.CRLReason).value.reason.name + except x509.ExtensionNotFound: + reason = None + entries.append( + { + 'serial_number': f'{revoked.serial_number:X}', + 'revocation_date': revoked.revocation_date_utc.isoformat(), + 'reason': reason, + } + ) + return entries + + +class CrlImportSerializer(serializers.Serializer[Any]): + """Import a PEM or DER CRL, optionally checking it against an issuing CA.""" + + crl_file = serializers.FileField(write_only=True) + ca = serializers.PrimaryKeyRelatedField(queryset=CaModel.objects.all(), required=False, allow_null=True) + set_active = serializers.BooleanField(default=True) + + +class CaGenerateSerializer(serializers.Serializer[Any]): + """Select a backend-supported key type for the autogenerated PKI.""" + + key_type = serializers.CharField() + + +class CaPkcs12Serializer(serializers.Serializer[Any]): + """Import a CA credential from a PKCS#12 file.""" + + unique_name = serializers.CharField(required=False, allow_blank=True, default='') + pkcs12_file = serializers.FileField(write_only=True) + pkcs12_password = serializers.CharField(required=False, allow_blank=True, write_only=True, default='') + + +class CaRemoteSerializer(serializers.Serializer[Any]): + """Configure an upstream CA or registration authority.""" + + unique_name = serializers.CharField(max_length=256) + key_type = serializers.CharField() + remote_host = serializers.CharField() + remote_port = serializers.IntegerField(min_value=1, max_value=65535) + remote_path = serializers.CharField() + est_username = serializers.CharField(required=False, allow_blank=True) + est_password = serializers.CharField(required=False, allow_blank=True, write_only=True, trim_whitespace=False) + cmp_shared_secret = serializers.CharField(required=False, allow_blank=True, write_only=True, trim_whitespace=False) + is_ra_mode = serializers.BooleanField(default=False) + + +class CaExternalCsrSerializer(serializers.Serializer[Any]): + """Create a managed key for an externally signed CA certificate.""" + + unique_name = serializers.CharField(max_length=256) + key_type = serializers.CharField() + + +class CaTruststoreSerializer(serializers.Serializer[Any]): + """Associate a truststore after validating its intended usage.""" + + trust_store = serializers.IntegerField(min_value=1) + + +class CaCrlSettingsSerializer(serializers.Serializer[Any]): + """Partial CRL configuration; existing values fill omitted fields.""" + + crl_cycle_enabled = serializers.BooleanField(required=False) + crl_cycle_interval_hours = serializers.FloatField(required=False, min_value=5 / 60) + crl_validity_hours = serializers.FloatField(required=False, min_value=1) + auto_crl_on_revocation_enabled = serializers.BooleanField(required=False) + + +class CaRolloverSerializer(serializers.ModelSerializer[CaRolloverModel]): + """Rollover state and history.""" + + class Meta: + """State transitions are only available through dedicated actions.""" + + model = CaRolloverModel + fields = ( + 'id', + 'old_issuing_ca', + 'new_issuing_ca', + 'state', + 'strategy_type', + 'planned_at', + 'started_at', + 'completed_at', + 'transition_scheduled_at', + 'initiated_by', + 'notes', + ) + read_only_fields = fields + + +class CaRolloverPlanSerializer(serializers.Serializer[Any]): + """Acquire replacement credentials using the same methods as the UI.""" + + old_issuing_ca = serializers.PrimaryKeyRelatedField(queryset=CaModel.objects.all()) + method = serializers.ChoiceField(choices=['pkcs12', 'separate_files', 'est', 'cmp']) + unique_name = serializers.CharField(required=False, allow_blank=True) + pkcs12_file = serializers.FileField(required=False, write_only=True) + pkcs12_password = serializers.CharField(required=False, allow_blank=True, write_only=True) + private_key_file = serializers.FileField(required=False, write_only=True) + private_key_file_password = serializers.CharField(required=False, allow_blank=True, write_only=True) + ca_certificate = serializers.FileField(required=False, write_only=True) + ca_certificate_chain = serializers.FileField(required=False, write_only=True) + key_type = serializers.CharField(required=False) + remote_host = serializers.CharField(required=False) + remote_port = serializers.IntegerField(required=False, min_value=1, max_value=65535) + remote_path = serializers.CharField(required=False) + est_username = serializers.CharField(required=False, allow_blank=True) + est_password = serializers.CharField(required=False, allow_blank=True, write_only=True) + cmp_shared_secret = serializers.CharField(required=False, allow_blank=True, write_only=True) + transition_scheduled_at = serializers.DateTimeField(required=False) + notes = serializers.CharField(required=False, allow_blank=True) + + +class EncodingSerializer(serializers.Serializer[Any]): + """Download encoding, deliberately avoiding DRF's reserved format parameter.""" + + encoding = serializers.ChoiceField(choices=['pem', 'der'], default='pem') + + +class CaCertificateUploadSerializer(serializers.Serializer[Any]): + """Attach the CA certificate issued for a pending managed key.""" + + certificate = serializers.FileField(write_only=True) + + +class CaEnrollmentSerializer(CertificateIssuanceContentSerializer): + """Certificate profile content with optional CMP sender identifier.""" + + sender_kid = serializers.IntegerField(required=False, min_value=0) diff --git a/trustpoint/pki/serializer/certificate.py b/trustpoint/pki/serializer/certificate.py index b0fff6aff..e344b348f 100644 --- a/trustpoint/pki/serializer/certificate.py +++ b/trustpoint/pki/serializer/certificate.py @@ -7,11 +7,20 @@ of Certificate model instances to and from JSON. """ -from typing import ClassVar +from typing import Any, ClassVar +from cryptography import x509 +from drf_spectacular.utils import extend_schema_field from rest_framework import serializers +from trustpoint_core.archiver import ArchiveFormat +from trustpoint_core.oid import NameOid +from trustpoint_core.serializer import CertificateFormat -from pki.models.certificate import CertificateModel +from pki.models.certificate import CertificateModel, RevokedCertificateModel +from pki.models.issued_credential import IssuedCredentialModel +from pki.services.certificate_download import certificate_chain + +OID_NAMES = {oid.dotted_string: oid.verbose_name for oid in NameOid} class CertificateSerializer(serializers.ModelSerializer[CertificateModel]): @@ -26,3 +35,242 @@ class Meta: model = CertificateModel fields: ClassVar[list[str]] = ['id', 'common_name'] read_only_fields: ClassVar[list[str]] = ['id'] + + +class CertificateSummarySerializer(CertificateSerializer): + """Read-only certificate inventory information, independent of credential/private-key records.""" + + certificate_status = serializers.ChoiceField(choices=CertificateModel.CertificateStatus.choices, read_only=True) + status_display = serializers.CharField(source='table_status', read_only=True) + is_ca = serializers.BooleanField(read_only=True) + issuer_certificate = serializers.IntegerField(source='issuer_id_id', read_only=True, allow_null=True) + + class Meta(CertificateSerializer.Meta): + """Keep the original identity fields and add useful inventory metadata.""" + + fields: ClassVar[list[str]] = [ + 'id', + 'common_name', + 'serial_number', + 'sha256_fingerprint', + 'not_valid_before', + 'not_valid_after', + 'certificate_status', + 'status_display', + 'is_ca', + 'is_self_signed', + 'issuer_certificate', + 'spki_algorithm', + 'spki_key_size', + 'created_at', + ] + read_only_fields = fields + + +class CertificateRevocationSerializer(serializers.ModelSerializer[RevokedCertificateModel]): + """Persisted revocation details, when the certificate has been revoked.""" + + class Meta: + """Return the date, reason, and associated CA without exposing credentials.""" + + model = RevokedCertificateModel + fields = ('revoked_at', 'revocation_reason', 'ca') + read_only_fields = fields + + +class CertificateIssuedCredentialSerializer(serializers.ModelSerializer[IssuedCredentialModel]): + """Public device and profile associations shown alongside a certificate in the UI.""" + + device_name = serializers.CharField(source='device.common_name', read_only=True) + onboarding_protocol = serializers.CharField(source='device.get_onboarding_protocol_display', read_only=True) + credential_type_display = serializers.CharField(source='get_issued_credential_type_display', read_only=True) + + class Meta: + """Identifiers and issuance metadata, without credential contents or secrets.""" + + model = IssuedCredentialModel + fields = ( + 'id', + 'credential', + 'device', + 'device_name', + 'domain', + 'onboarding_protocol', + 'issued_credential_type', + 'credential_type_display', + 'issued_using_cert_profile', + ) + read_only_fields = fields + + +def name_attributes(name: x509.Name) -> list[dict[str, Any]]: + """Preserve repeated attributes and relative distinguished name groupings.""" + return [ + { + 'rdn': index, + 'oid': attribute.oid.dotted_string, + 'name': OID_NAMES.get(attribute.oid.dotted_string, attribute.oid.dotted_string), + 'value': attribute.value.hex().upper() if isinstance(attribute.value, bytes) else attribute.value, + } + for index, rdn in enumerate(name.rdns) + for attribute in rdn + ] + + +def general_names( + names: x509.GeneralNames | x509.SubjectAlternativeName | x509.IssuerAlternativeName, +) -> list[dict[str, Any]]: + """Represent all GeneralName types, including directory names and opaque other names.""" + result = [] + for name in names: + if isinstance(name, x509.DirectoryName): + value: Any = name_attributes(name.value) + elif isinstance(name, x509.RegisteredID): + value = name.value.dotted_string + elif isinstance(name, x509.OtherName): + value = {'oid': name.type_id.dotted_string, 'der_hex': name.value.hex().upper()} + else: + value = str(name.value) + result.append({'type': type(name).__name__, 'value': value}) + return result + + +def extension_value(value: x509.ExtensionType) -> Any: # noqa: PLR0911 + """Expose common extensions as JSON, with a readable fallback for other extensions.""" + if isinstance(value, x509.BasicConstraints): + return {'ca': value.ca, 'path_length': value.path_length} + if isinstance(value, x509.KeyUsage): + fields = ( + 'digital_signature', + 'content_commitment', + 'key_encipherment', + 'data_encipherment', + 'key_agreement', + 'key_cert_sign', + 'crl_sign', + ) + return { + **{field: getattr(value, field) for field in fields}, + 'encipher_only': value.encipher_only if value.key_agreement else None, + 'decipher_only': value.decipher_only if value.key_agreement else None, + } + if isinstance(value, x509.ExtendedKeyUsage): + return [oid.dotted_string for oid in value] + if isinstance(value, (x509.SubjectAlternativeName, x509.IssuerAlternativeName)): + return general_names(value) + if isinstance(value, x509.SubjectKeyIdentifier): + return {'key_identifier': value.digest.hex().upper()} + if isinstance(value, x509.AuthorityKeyIdentifier): + return { + 'key_identifier': value.key_identifier.hex().upper() if value.key_identifier is not None else None, + 'authority_cert_issuer': general_names(x509.GeneralNames(value.authority_cert_issuer)) + if value.authority_cert_issuer is not None + else None, + 'authority_cert_serial_number': f'{value.authority_cert_serial_number:X}' + if value.authority_cert_serial_number is not None + else None, + } + return str(value) + + +class CertificateDetailSerializer(CertificateSummarySerializer): + """Full public X.509 details with structured names, extensions, and revocation information.""" + + subject = serializers.SerializerMethodField() + issuer = serializers.SerializerMethodField() + extensions = serializers.SerializerMethodField() + certificate_chain = serializers.SerializerMethodField() + revocation = CertificateRevocationSerializer(source='revoked_certificate', read_only=True, allow_null=True) + signature_algorithm = serializers.CharField(read_only=True) + signature_algorithm_padding_scheme = serializers.CharField(read_only=True) + days_left = serializers.IntegerField(read_only=True) + is_root_ca = serializers.BooleanField(read_only=True) + is_end_entity = serializers.BooleanField(read_only=True) + issued_credentials = serializers.SerializerMethodField() + issued_certificate_count = serializers.SerializerMethodField() + + class Meta(CertificateSummarySerializer.Meta): + """Read-only certificate fields; write operations retain the existing serializer.""" + + fields: ClassVar[list[str]] = [ + *CertificateSummarySerializer.Meta.fields, + 'version', + 'signature_algorithm_oid', + 'signature_algorithm', + 'signature_algorithm_padding_scheme', + 'signature_value', + 'spki_algorithm_oid', + 'spki_ec_curve_oid', + 'spki_ec_curve', + 'public_key_pem', + 'subject', + 'issuer', + 'extensions', + 'certificate_chain', + 'revocation', + 'days_left', + 'is_root_ca', + 'is_end_entity', + 'issued_credentials', + 'issued_certificate_count', + 'cert_pem', + ] + read_only_fields = fields + + def get_subject(self, obj: CertificateModel) -> list[dict[str, Any]]: + """Return subject attributes in their original order and RDN groups.""" + return name_attributes(obj.get_certificate_serializer().as_crypto().subject) + + def get_issuer(self, obj: CertificateModel) -> list[dict[str, Any]]: + """Return issuer distinguished-name attributes.""" + return name_attributes(obj.get_certificate_serializer().as_crypto().issuer) + + def get_extensions(self, obj: CertificateModel) -> list[dict[str, Any]]: + """Include every extension, with its exact DER bytes for lossless interpretation.""" + return [ + { + 'oid': extension.oid.dotted_string, + 'critical': extension.critical, + 'value': extension_value(extension.value), + 'der_hex': extension.value.public_bytes().hex().upper(), + } + for extension in obj.get_certificate_serializer().as_crypto().extensions + ] + + def get_certificate_chain(self, obj: CertificateModel) -> list[int]: + """List known issuer certificate IDs from the direct issuer to the root.""" + try: + return [issuer.pk for issuer in certificate_chain(obj)] + except ValueError as exc: + msg = 'Unable to build the certificate issuer chain.' + raise serializers.ValidationError(msg) from exc + + @extend_schema_field(CertificateIssuedCredentialSerializer(many=True)) + def get_issued_credentials(self, obj: CertificateModel) -> list[dict[str, Any]]: + """Expose local issuance associations used for device/profile navigation and revocation.""" + credentials = ( + IssuedCredentialModel.objects.filter(credential__certificate=obj) + .select_related('device__onboarding_config', 'device__no_onboarding_config') + .order_by('pk') + ) + return [dict(item) for item in CertificateIssuedCredentialSerializer(credentials, many=True).data] + + def get_issued_certificate_count(self, obj: CertificateModel) -> int: + """Count certificates linked to this issuer, excluding a self-reference.""" + return CertificateModel.objects.filter(issuer_id=obj).exclude(pk=obj.pk).count() + + +class CertificateDownloadSerializer(serializers.Serializer[Any]): + """Options for a public certificate file download.""" + + file_format = serializers.ChoiceField(choices=[item.value for item in CertificateFormat], default='pem') + include_chain = serializers.BooleanField(default=False) + + +class CertificateArchiveSerializer(CertificateDownloadSerializer): + """Select certificates and an archive format without changing their contents.""" + + certificate_ids = serializers.ListField( + child=serializers.IntegerField(min_value=1), allow_empty=False, max_length=1000 + ) + archive_format = serializers.ChoiceField(choices=[item.value for item in ArchiveFormat], default='zip') diff --git a/trustpoint/pki/services/ca_enrollment.py b/trustpoint/pki/services/ca_enrollment.py new file mode 100644 index 000000000..193114819 --- /dev/null +++ b/trustpoint/pki/services/ca_enrollment.py @@ -0,0 +1,408 @@ +# Copyright (c) 2026 The Trustpoint Project Authors +# SPDX-License-Identifier: MIT + +"""Shared upstream CA enrollment for browser and REST clients.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, NamedTuple, cast + +from cryptography import x509 +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec, rsa # noqa: TC002 + +from pki.models import CaModel, CertificateModel, CredentialModel +from pki.models.credential import CertificateChainOrderModel, PrimaryCredentialCertificate +from pki.models.issued_credential import RemoteIssuedCredentialModel +from pki.util.certificate_request import build_ca_request_data +from request.clients import EstClient +from request.clients.cmp_client import CmpClient +from request.message_builder.cmp import CmpMessageBuilder +from request.operation_processor.csr_build import ProfileAwareCsrBuilder +from request.operation_processor.csr_sign import EstDeviceCsrSignProcessor +from request.request_context import CmpCertificateRequestContext, EstCertificateRequestContext +from trustpoint.logger import LoggerMixin + +if TYPE_CHECKING: + from pki.models.cert_profile import CertificateProfileModel + + +class CmpContextParams(NamedTuple): + """Parameters for creating a CMP certificate request context.""" + + subject: x509.Name + public_key: rsa.RSAPublicKey | ec.EllipticCurvePublicKey | Any # Any for ML-DSA keys + private_key: rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey | Any # Any for ML-DSA keys + recipient_name: str + extensions: list[x509.Extension[x509.ExtensionType]] | None + sender_kid: int | None + + +class EstCaEnrollmentService(LoggerMixin): + """Shared upstream CA enrollment independent of browser sessions.""" + + def enroll(self, ca: CaModel, cert_content_data: dict[str, Any], cert_profile: CertificateProfileModel) -> None: + """Perform the EST enrollment process.""" + request_data = self._build_request_data_from_form(cert_content_data) + self.logger.info('Built request data: %s', request_data) + + context = EstCertificateRequestContext( + operation='simpleenroll', + protocol='est', + domain=None, + cert_profile_str=cert_profile.unique_name, + certificate_profile_model=cert_profile, + allow_ca_certificate_request=True, # Allow CA cert requests for Issuing CA enrollment + est_server_host=ca.remote_host, + est_server_port=ca.remote_port, + est_server_path=ca.remote_path, + est_username=ca.est_username, + est_password=(ca.no_onboarding_config.est_password if ca.no_onboarding_config else None), + est_server_truststore=(ca.no_onboarding_config.trust_store if ca.no_onboarding_config else None), + ) + + context.request_data = request_data + context.owner_credential = ca.credential + + csr_builder = ProfileAwareCsrBuilder() + csr_builder.process_operation(context) + csr = csr_builder.get_csr() + + context.cert_requested = csr + + # Use EstDeviceCsrSignProcessor to re-sign the CSR with the proper signature algorithm + csr_signer = EstDeviceCsrSignProcessor() + csr_signer.process_operation(context) + signed_csr = csr_signer.get_signed_csr() + + # Send the re-signed CSR to the remote EST server + est_client = EstClient(context) + issued_cert = est_client.simple_enroll(signed_csr) + + cert_pem = issued_cert.public_bytes(encoding=serialization.Encoding.PEM).decode('utf-8') + cert_obj = x509.load_pem_x509_certificate(cert_pem.encode()) + cert_model = CertificateModel.save_certificate(cert_obj) + + if ca.credential: + ca.credential.certificate = cert_model + ca.credential.save() + else: + credential = CredentialModel( + credential_type=CredentialModel.CredentialTypeChoice.ISSUING_CA, + certificate=cert_model, + private_key=ca.credential.private_key if ca.credential else '', + ) + credential.save() + credential.certificates.add(cert_model) + PrimaryCredentialCertificate.objects.create(credential=credential, certificate=cert_model, is_primary=True) + ca.credential = credential + ca.save() + + cn_attrs = cert_obj.subject.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME) + common_name_value = cn_attrs[0].value if cn_attrs else '' + # Ensure common_name is a string + if isinstance(common_name_value, bytes): + common_name = common_name_value.decode('utf-8') + else: + common_name = str(common_name_value) + remote_issued_credential = RemoteIssuedCredentialModel( + common_name=common_name, + issued_credential_type=RemoteIssuedCredentialModel.RemoteIssuedCredentialType.LOCAL_CA, + issued_using_cert_profile=cert_profile.unique_name, + credential=ca.credential, + ca=ca, + ) + remote_issued_credential.full_clean() + remote_issued_credential.save() + + def _build_request_data_from_form(self, cert_content_data: dict[str, Any]) -> dict[str, Any]: + """Use the shared CA request conversion, including SAN list normalization.""" + self.logger.info('Building request data from: %s', cert_content_data) + return build_ca_request_data(cert_content_data) + + +class CmpCaEnrollmentService(LoggerMixin): + """Shared upstream CA enrollment independent of browser sessions.""" + + def enroll( + self, + ca: CaModel, + cert_content_data: dict[str, Any], + cert_profile: CertificateProfileModel, + *, + sender_kid: int | None = None, + ) -> None: + """Perform the CMP enrollment process.""" + request_data = self._build_request_data_from_form(cert_content_data) + self.logger.info('Built request data: %s', request_data) + + context = CmpCertificateRequestContext( + operation='certification', + protocol='cmp', + domain=None, + cert_profile_str=cert_profile.unique_name, + certificate_profile_model=cert_profile, + allow_ca_certificate_request=True, + cmp_server_host=ca.remote_host, + cmp_server_port=ca.remote_port, + cmp_server_path=ca.remote_path, + cmp_shared_secret=(ca.no_onboarding_config.cmp_shared_secret if ca.no_onboarding_config else None), + cmp_server_truststore=(ca.no_onboarding_config.trust_store if ca.no_onboarding_config else None), + ) + + context.request_data = request_data + context.owner_credential = ca.credential + + csr_builder = ProfileAwareCsrBuilder() + csr_builder.process_operation(context) + csr = csr_builder.get_csr() + + if not csr.subject or len(csr.subject) == 0: + msg = 'CSR does not have a subject - cannot build CMP request' + raise ValueError(msg) + + self.logger.info('CSR subject: %s', csr.subject.rfc4514_string()) + + csr_subject = csr.subject + csr_public_key = csr.public_key() + csr_extensions = list(csr.extensions) if csr.extensions else None + + self.logger.info('CSR has %d extensions', len(csr.extensions) if csr.extensions else 0) + + recipient_name = self._get_recipient_name_from_truststore(ca) + + private_key = self._load_private_key(ca) + + cmp_params = CmpContextParams( + subject=csr_subject, + public_key=cast('rsa.RSAPublicKey | ec.EllipticCurvePublicKey | Any', csr_public_key), + private_key=private_key, + recipient_name=recipient_name, + extensions=csr_extensions, + sender_kid=sender_kid, + ) + + cmp_context = self._create_cmp_context(ca, cmp_params) + + cmp_builder = CmpMessageBuilder() + cmp_builder.build(cmp_context) + + pki_message = cmp_context.parsed_message + + cmp_client = CmpClient(cmp_context) + issued_cert, chain_certs = cmp_client.send_and_extract_certificate( + pki_message, + add_shared_secret_protection=True, + ) + + self._save_cmp_certificates(ca, issued_cert, chain_certs, cert_profile) + + def _build_request_data_from_form(self, cert_content_data: dict[str, Any]) -> dict[str, Any]: + """Use the shared CA request conversion, including SAN list normalization.""" + self.logger.info('Building request data from: %s', cert_content_data) + return build_ca_request_data(cert_content_data) + + def _get_recipient_name_from_truststore(self, ca: CaModel) -> str: + """Extract recipient name from the CA's truststore.""" + if not ca.no_onboarding_config or not ca.no_onboarding_config.trust_store: + msg = 'No truststore configured for CMP CA - recipient name cannot be determined' + raise ValueError(msg) + + try: + truststore_certs = ca.no_onboarding_config.trust_store.truststoreordermodel_set.order_by('order') + if not truststore_certs.exists(): + msg = 'Truststore contains no certificates - cannot determine recipient name' + raise ValueError(msg) + + first_cert_model = truststore_certs.first().certificate # type: ignore[union-attr] + ca_cert = first_cert_model.get_certificate_serializer().as_crypto() + recipient_name = ca_cert.subject.rfc4514_string() + self.logger.info('Using recipient name from truststore: %s', recipient_name) + except (AttributeError, IndexError) as e: + msg = f'Failed to extract recipient name from truststore: {e}' + raise ValueError(msg) from e + else: + return recipient_name + + def _load_private_key(self, ca: CaModel) -> rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey | Any: + """Load the private key from the CA's credential.""" + if not ca.credential: + msg = 'No private key available for CA credential' + raise ValueError(msg) + + try: + return cast( + 'rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey | Any', + ca.credential.get_private_key(), + ) + except (RuntimeError, TypeError, ValueError) as exc: + msg = 'No usable private key is available for the CA credential.' + raise ValueError(msg) from exc + + def _create_cmp_context( + self, + ca: CaModel, + params: CmpContextParams, + ) -> CmpCertificateRequestContext: + """Create the CMP certificate request context.""" + return CmpCertificateRequestContext( + protocol='cmp', + operation='certification', + cmp_server_host=ca.remote_host, + cmp_server_port=ca.remote_port, + cmp_server_path=ca.remote_path, + cmp_shared_secret=(ca.no_onboarding_config.cmp_shared_secret if ca.no_onboarding_config else None), + cmp_server_truststore=(ca.no_onboarding_config.trust_store if ca.no_onboarding_config else None), + request_data={ + 'subject': params.subject, + 'public_key': params.public_key, + 'private_key': params.private_key, + 'recipient_name': params.recipient_name, + 'extensions': params.extensions, + 'use_initialization_request': False, + 'add_pop': True, + 'prepare_shared_secret_protection': True, + 'sender_kid': params.sender_kid, + }, + ) + + def _save_cmp_certificates( + self, + ca: CaModel, + issued_cert: x509.Certificate, + chain_certs: list[x509.Certificate], + cert_profile: CertificateProfileModel, + ) -> None: + """Save the issued certificate and chain certificates.""" + cert_model = CertificateModel.save_certificate(issued_cert) + + chain_ca_models = [] + for chain_cert in chain_certs: + existing_ca = self._find_existing_ca_for_certificate(chain_cert) + if existing_ca: + chain_ca_models.append(existing_ca) + else: + cn_attrs = chain_cert.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME) + cn = cn_attrs[0].value if cn_attrs else 'unknown' + keyless_ca = CaModel.create_keyless_ca( + unique_name=str(cn), + certificate_obj=chain_cert, + parent_ca=None, + ) + chain_ca_models.append(keyless_ca) + self.logger.info('Created keyless CA %s for chain certificate', keyless_ca.unique_name) + + self._build_ca_hierarchy(chain_ca_models) + + issuer_ca = self._find_issuer_ca(issued_cert, chain_ca_models) + if issuer_ca: + ca.parent_ca = issuer_ca + ca.save() + self.logger.info('Set parent CA %s for issuing CA %s', issuer_ca.unique_name, ca.unique_name) + + if ca.credential: + ca.credential.certificate = cert_model + ca.credential.save() + self._build_certificate_chain_for_credential(ca.credential, ca.parent_ca) + else: + credential = CredentialModel( + credential_type=CredentialModel.CredentialTypeChoice.ISSUING_CA, + certificate=cert_model, + private_key=ca.credential.private_key if ca.credential else '', + ) + credential.save() + credential.certificates.add(cert_model) + PrimaryCredentialCertificate.objects.create(credential=credential, certificate=cert_model, is_primary=True) + ca.credential = credential + ca.save() + self._build_certificate_chain_for_credential(credential, ca.parent_ca) + + # Create a RemoteIssuedCredentialModel to track the issued CA credential + cn_attrs = issued_cert.subject.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME) + common_name_value = cn_attrs[0].value if cn_attrs else '' + # Ensure common_name is a string + if isinstance(common_name_value, bytes): + common_name = common_name_value.decode('utf-8') + else: + common_name = str(common_name_value) + remote_issued_credential = RemoteIssuedCredentialModel( + common_name=common_name, + issued_credential_type=RemoteIssuedCredentialModel.RemoteIssuedCredentialType.LOCAL_CA, + issued_using_cert_profile=cert_profile.unique_name, + credential=ca.credential, + ca=ca, + ) + remote_issued_credential.full_clean() + remote_issued_credential.save() + + def _build_ca_hierarchy( + self, + chain_ca_models: list[CaModel], + ) -> None: + """Build the CA hierarchy by setting parent-child relationships.""" + ca_by_subject = {} + for ca in chain_ca_models: + cert = ca.get_certificate() + if cert: + ca_by_subject[cert.subject.public_bytes().hex()] = ca + + for ca in chain_ca_models: + cert = ca.get_certificate() + if not cert: + continue + issuer_subject_bytes = cert.issuer.public_bytes().hex() + if issuer_subject_bytes in ca_by_subject: + parent_ca = ca_by_subject[issuer_subject_bytes] + if parent_ca not in (ca, ca.parent_ca): + ca.parent_ca = parent_ca + ca.save() + self.logger.info('Set parent CA %s for CA %s', parent_ca.unique_name, ca.unique_name) + + def _find_issuer_ca( + self, + issued_cert: x509.Certificate, + chain_ca_models: list[CaModel], + ) -> CaModel | None: + """Find the CA that issued the certificate.""" + issuer_subject_bytes = issued_cert.issuer.public_bytes().hex() + for ca in chain_ca_models: + ca_cert = ca.get_certificate() + if ca_cert and ca_cert.subject.public_bytes().hex() == issuer_subject_bytes: + return ca + return None + + def _build_certificate_chain_for_credential(self, credential: CredentialModel, parent_ca: CaModel | None) -> None: + """Build the certificate chain for the credential.""" + if not parent_ca or not credential.certificate: + return + chain = [] + current: CaModel | None = parent_ca + while current: + if current.certificate: + chain.append(current.certificate) + current = current.parent_ca + credential.certificatechainordermodel_set.all().delete() + primary_cert = credential.certificate + for i, cert in enumerate(chain): + CertificateChainOrderModel.objects.create( + credential=credential, certificate=cert, order=i, primary_certificate=primary_cert + ) + + def _find_existing_ca_for_certificate(self, cert: x509.Certificate) -> CaModel | None: + """Find an existing CA that has the given certificate.""" + # TODO(FHK): subject+issuer byte equality is not sufficient for # noqa: FIX002 + + subject_bytes = cert.subject.public_bytes().hex().upper() + issuer_bytes = cert.issuer.public_bytes().hex().upper() + + keyless_match = CaModel.objects.filter( + certificate__subject_public_bytes=subject_bytes, + certificate__issuer_public_bytes=issuer_bytes, + ).first() + if keyless_match: + return keyless_match + + return CaModel.objects.filter( + credential__certificate__subject_public_bytes=subject_bytes, + credential__certificate__issuer_public_bytes=issuer_bytes, + ).first() diff --git a/trustpoint/pki/services/certificate_download.py b/trustpoint/pki/services/certificate_download.py new file mode 100644 index 000000000..4f8860d96 --- /dev/null +++ b/trustpoint/pki/services/certificate_download.py @@ -0,0 +1,29 @@ +# Copyright (c) 2026 The Trustpoint Project Authors +# SPDX-License-Identifier: MIT + +"""Public certificate export operations shared by the management API.""" + +from pki.models.certificate import CertificateModel + + +def certificate_chain(certificate: CertificateModel) -> list[CertificateModel]: + """Return known issuers in order, rejecting circular issuer relationships.""" + chain = [] + seen = {certificate.pk} + current = certificate + while current.issuer_id is not None and current.issuer_id.pk != current.pk: + issuer = current.issuer_id + if issuer.pk in seen: + msg = 'The certificate issuer chain contains a cycle.' + raise ValueError(msg) + seen.add(issuer.pk) + chain.append(issuer) + current = issuer + return chain + + +def certificate_filename(certificate: CertificateModel, extension: str) -> str: + """Keep filenames safe and unique even when common names are duplicated or empty.""" + name = ''.join(char if char.isalnum() or char in '._-' else '_' for char in certificate.common_name) + name = name.strip('._-')[:100] or 'certificate' + return f'{name}-{certificate.pk}{extension}' diff --git a/trustpoint/pki/tests/test_ca_crl_api.py b/trustpoint/pki/tests/test_ca_crl_api.py new file mode 100644 index 000000000..7f41c2d8e --- /dev/null +++ b/trustpoint/pki/tests/test_ca_crl_api.py @@ -0,0 +1,603 @@ +# Copyright (c) 2026 The Trustpoint Project Authors +# SPDX-License-Identifier: MIT + +"""Behavioral coverage for CA/CRL management endpoints and their authorization.""" + +from datetime import timedelta +from unittest.mock import patch + +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from django.contrib.auth.models import Permission +from django.core.files.uploadedfile import SimpleUploadedFile +from django.utils import timezone +from rest_framework.test import APIClient + +from management.models import SecurityConfig +from management.models.audit_log import AuditLog +from pki.models import CaModel, CrlModel +from pki.util.x509 import CertificateGenerator +from pki.models.ca_rollover import CaRolloverModel, CaRolloverState, CaRolloverStrategyType +from pki.services.ca_rollover import CaRolloverError +from pki.util.cert_profile import ProfileValidationError +from users.models import TrustpointUser + + +@pytest.fixture +def api(): + """Use an authenticated CA administrator without browser authentication middleware.""" + user = TrustpointUser.objects.create_superuser(username='ca-api-admin', password='test-password-123') + user.role.permissions.add(Permission.objects.get(codename='manage_cas')) + client = APIClient() + client.force_authenticate(user) + return client + + +@pytest.fixture +def ca(issuing_ca_instance): + """Use the real managed-key CA fixture.""" + return issuing_ca_instance['issuing_ca'] + + +def crl_upload(fixture, number=10, encoding=serialization.Encoding.PEM, signing_key=None): + """Sign a CRL using the fixture CA, or another key for negative tests.""" + now = timezone.now() + crl = ( + x509.CertificateRevocationListBuilder() + .issuer_name(fixture['cert'].subject) + .last_update(now) + .next_update(now + timedelta(days=1)) + .add_extension(x509.CRLNumber(number), critical=False) + .sign(signing_key or fixture['priv_key'], hashes.SHA256()) + ) + return SimpleUploadedFile('test.crl', crl.public_bytes(encoding)) + + +@pytest.mark.parametrize('url', ['/api/cas/', '/api/crls/', '/api/ca-rollovers/', '/api/issuing-cas/']) +def test_requires_ca_permission(url): + """Anonymous and ordinary authenticated users cannot administer CAs or CRLs.""" + client = APIClient() + assert client.get(url).status_code in (401, 403) + user = TrustpointUser.objects.create_user(username='ordinary', password='test-password-123') + client.force_authenticate(user) + assert client.get(url).status_code == 403 + + +def test_ca_inventory_includes_keyless_and_root(api, ca): + """The full inventory exposes CA types omitted from the issuing inventory.""" + keyless = CaModel.objects.create( + unique_name='keyless', ca_type=CaModel.CaTypeChoice.KEYLESS, certificate=ca.ca_certificate_model + ) + response = api.get('/api/cas/') + assert response.status_code == 200 + rows = response.data.get('results', []) if isinstance(response.data, dict) else response.data + assert {row['id'] for row in rows} >= {ca.pk, keyless.pk} + assert api.get(f'/api/cas/{keyless.pk}/').status_code == 200 + assert api.get(f'/api/issuing-cas/{keyless.pk}/').status_code == 404 + + +def test_issuing_ca_retrieve_accepts_router_pk(api, ca): + """Detail lookup must not fail because the legacy retrieve signature omitted pk.""" + response = api.get(f'/api/issuing-cas/{ca.pk}/') + assert response.status_code == 200 + assert response.data['crl_validity_hours'] == 24 + assert 'private_key' not in response.data + + +@pytest.mark.parametrize( + ('exception_type', 'status_code', 'expected_message'), + [ + (ValueError, 400, 'The PKI operation could not be completed. Check the input and configuration.'), + (ProfileValidationError, 400, 'The PKI operation could not be completed. Check the input and configuration.'), + (CaRolloverError, 409, 'The CA rollover could not be completed in its current state.'), + ], +) +def test_ca_domain_errors_do_not_expose_exception_details(api, ca, exception_type, status_code, expected_message): + """Shared error handling preserves status codes without returning internal details.""" + with patch('pki.views.ca_api.CaViewSet.get_object', side_effect=exception_type('secret credential path')): + response = api.get(f'/api/cas/{ca.pk}/') + + assert response.status_code == status_code + assert response.json() == ({'detail': expected_message} if status_code == 409 else [expected_message]) + assert b'secret credential path' not in response.content + + +def test_csr_error_does_not_expose_exception_details(api, lab_config): + """CSR builder exceptions return a safe validation message.""" + configured = api.post( + '/api/issuing-cas/configure-external-csr/', + {'unique_name': 'external-ca', 'key_type': 'RSA-2048'}, + format='json', + ) + assert configured.status_code == 201, configured.data + ca_pk = configured.data['id'] + with ( + patch('pki.views.ca_api.IssuingCaManagementViewSet._certificate_content', return_value=(None, {}, None)), + patch('pki.views.ca_api.build_external_csr', side_effect=ValueError('secret credential path')) as builder, + ): + response = api.post(f'/api/issuing-cas/{ca_pk}/csr/', {}, format='json') + + builder.assert_called_once() + assert response.status_code == 400 + assert response.json() == [ + 'Unable to build the CA certificate signing request. Check the profile and request data.' + ] + assert b'secret credential path' not in response.content + + +def test_generate_error_does_not_expose_exception_details(api, lab_config): + """PKI generation errors cannot reveal backend exception text.""" + lab_config.auto_gen_pki = True + lab_config.save() + with ( + patch('pki.views.ca_api.AutoGenPki.get_auto_gen_pki', return_value=None), + patch( + 'pki.views.ca_api.AutoGenPki.enable_auto_gen_pki', side_effect=ValueError('secret credential path') + ) as generate, + ): + response = api.post('/api/issuing-cas/generate/', {'key_type': 'RSA-2048'}, format='json') + + generate.assert_called_once() + assert response.status_code == 400 + assert response.json() == ['Unable to generate the PKI. Check the key type and security settings.'] + assert b'secret credential path' not in response.content + + +def test_rollover_transition_error_does_not_expose_exception_details(api, ca): + """A failed transition keeps its conflict status without disclosing internal text.""" + rollover = CaRolloverModel.objects.create(old_issuing_ca=ca, strategy_type=CaRolloverStrategyType.IMPORT_CA) + with patch('pki.views.ca_api.CaRolloverService.execute_rollover', side_effect=ValueError('secret credential path')): + response = api.post(f'/api/ca-rollovers/{rollover.pk}/start/') + + assert response.status_code == 409 + assert response.json() == {'detail': 'The CA rollover transition could not be completed in its current state.'} + assert b'secret credential path' not in response.content + rollover.refresh_from_db() + assert rollover.state == CaRolloverState.PLANNED + + +@pytest.mark.parametrize('encoding', ['pem', 'der']) +def test_ca_certificate_download(api, ca, encoding): + """CA downloads contain the same certificate in each supported encoding.""" + response = api.get(f'/api/cas/{ca.pk}/certificate/?encoding={encoding}') + assert response.status_code == 200 + parser = x509.load_pem_x509_certificate if encoding == 'pem' else x509.load_der_x509_certificate + assert parser(response.content).serial_number == ca.get_certificate().serial_number + + +def test_download_invalid_encoding_and_missing_certificate(api, ca): + """Invalid encodings and pending certificates have explicit client errors.""" + ca.credential.certificate = None + ca.credential.save() + assert api.get(f'/api/cas/{ca.pk}/certificate/').status_code == 404 + assert api.get('/api/crls/999999/download/').status_code == 404 + + +def test_ca_delete_protected_child(api, ca): + """Deletion must preserve hierarchy guards and return validation instead of a 500.""" + CaModel.objects.create( + unique_name='child', parent_ca=ca, ca_type=CaModel.CaTypeChoice.KEYLESS, certificate=ca.ca_certificate_model + ) + response = api.delete(f'/api/cas/{ca.pk}/') + assert response.status_code == 400 + assert CaModel.objects.filter(pk=ca.pk).exists() + + +def test_delete_unused_ca(api): + """An unused CA can be deleted through the inventory endpoint.""" + cert, _ = CertificateGenerator.create_root_ca(cn='Unused') + ca = CaModel.create_keyless_ca(unique_name='unused', certificate_obj=cert) + pk = ca.pk + assert api.delete(f'/api/cas/{pk}/').status_code == 204 + assert not CaModel.objects.filter(pk=pk).exists() + + +def test_crl_settings_partial_update_and_validation(api, ca): + """PATCH preserves omitted values and enforces schedule/validity limits.""" + config = SecurityConfig(security_mode=SecurityConfig.SecurityModeChoices.LAB) + config.apply_security_settings(save=False) + config.save() + url = f'/api/issuing-cas/{ca.pk}/crl-settings/' + response = api.patch(url, {'crl_validity_hours': 12.5}, format='json') + assert response.status_code == 200, response.data + ca.refresh_from_db() + assert ca.crl_validity_hours == 12.5 + assert ca.crl_cycle_interval_hours == 24 + assert api.patch(url, {'crl_cycle_interval_hours': 0.01}, format='json').status_code == 400 + ca.is_active = False + ca.save() + assert api.patch(url, {'crl_validity_hours': 5}, format='json').status_code == 400 + + +def test_generate_crl_preserves_fractional_validity(api, ca): + """CRL validity is measured in fractional hours, not truncated integers.""" + ca.crl_validity_hours = 1.5 + ca.save() + response = api.post(f'/api/issuing-cas/{ca.pk}/generate-crl/') + assert response.status_code == 200, response.data + crl = ca.get_active_crl().get_crl_as_crypto() + assert (crl.next_update_utc - crl.last_update_utc).total_seconds() == 5400 + + +@pytest.mark.parametrize('encoding', [serialization.Encoding.PEM, serialization.Encoding.DER]) +def test_crl_import_download_delete(api, issuing_ca_instance, encoding): + """Imported CRLs retain their identity and round-trip through both download formats.""" + ca = issuing_ca_instance['issuing_ca'] + response = api.post('/api/crls/', {'ca': ca.pk, 'crl_file': crl_upload(issuing_ca_instance, encoding=encoding)}) + assert response.status_code == 201, response.data + pk = response.data['id'] + assert response.data['is_active'] is True + assert response.data['crl_number'] == 10 + assert api.get(f'/api/crls/{pk}/').status_code == 200 + for download_encoding in ('pem', 'der'): + result = api.get(f'/api/crls/{pk}/download/?encoding={download_encoding}') + assert result.status_code == 200 + parser = x509.load_pem_x509_crl if download_encoding == 'pem' else x509.load_der_x509_crl + assert parser(result.content).issuer == issuing_ca_instance['cert'].subject + assert api.get(f'/api/crls/{pk}/download/?encoding=invalid').status_code == 400 + assert api.delete(f'/api/crls/{pk}/').status_code == 204 + assert not CrlModel.objects.filter(pk=pk).exists() + + +def test_crl_rejects_bad_signature_and_malformed_file(api, issuing_ca_instance): + """A CRL cannot be associated with a CA that did not sign it.""" + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + response = api.post( + '/api/crls/', + {'ca': issuing_ca_instance['issuing_ca'].pk, 'crl_file': crl_upload(issuing_ca_instance, signing_key=key)}, + ) + assert response.status_code == 400 + assert not CrlModel.objects.exists() + assert api.post('/api/crls/', {'crl_file': SimpleUploadedFile('bad.crl', b'invalid')}).status_code == 400 + + +def test_crl_duplicate_does_not_deactivate_current(api, issuing_ca_instance): + """A duplicate import rolls back changes to the active CRL.""" + ca = issuing_ca_instance['issuing_ca'] + assert api.post('/api/crls/', {'ca': ca.pk, 'crl_file': crl_upload(issuing_ca_instance)}).status_code == 201 + response = api.post('/api/crls/', {'ca': ca.pk, 'crl_file': crl_upload(issuing_ca_instance)}) + assert response.status_code == 409 + assert CrlModel.objects.filter(ca=ca, is_active=True).count() == 1 + + +def test_unassociated_crl_is_inactive(api, issuing_ca_instance): + """Unassociated CRLs can be retained without becoming an active CA CRL.""" + response = api.post('/api/crls/', {'crl_file': crl_upload(issuing_ca_instance)}) + assert response.status_code == 201 + assert response.data['ca'] is None + assert response.data['is_active'] is False + + +def test_rollover_transitions_and_audit(api, ca, lab_config): + """Valid transitions persist and emit the same audit operations as the UI.""" + from cryptography.hazmat.primitives.serialization import pkcs12 + + cert, key = CertificateGenerator.create_root_ca(cn='Replacement') + response = api.post( + '/api/ca-rollovers/', + { + 'old_issuing_ca': ca.pk, + 'method': 'pkcs12', + 'unique_name': 'replacement', + 'pkcs12_file': SimpleUploadedFile( + 'new-ca.p12', + pkcs12.serialize_key_and_certificates(b'replacement', key, cert, None, serialization.NoEncryption()), + ), + }, + ) + assert response.status_code == 201, response.data + rollover = CaRolloverModel.objects.get(pk=response.data['id']) + count = CaModel.objects.count() + duplicate = api.post( + '/api/ca-rollovers/', + {'old_issuing_ca': ca.pk, 'method': 'pkcs12', 'pkcs12_file': SimpleUploadedFile('invalid.p12', b'invalid')}, + ) + assert duplicate.status_code == 409 + assert CaModel.objects.count() == count + url = f'/api/ca-rollovers/{rollover.pk}/' + assert api.post(url + 'transition/').status_code == 409 + assert api.post(url + 'start/').status_code == 200 + assert api.post(url + 'start/').status_code == 409 + assert api.post(url + 'transition/').status_code == 200 + assert api.post(url + 'complete/').status_code == 200 + rollover.refresh_from_db() + assert rollover.state == CaRolloverState.COMPLETED + assert AuditLog.objects.filter(operation_type=AuditLog.OperationType.CA_ROLLOVER_COMPLETED).exists() + assert api.patch(url, {'state': 'planned'}, format='json').status_code == 405 + + +def test_rollover_cancel(api, ca): + """Cancellation goes through the lifecycle service.""" + rollover = CaRolloverModel.objects.create(old_issuing_ca=ca, strategy_type=CaRolloverStrategyType.IMPORT_CA) + assert api.post(f'/api/ca-rollovers/{rollover.pk}/cancel/').status_code == 200 + rollover.refresh_from_db() + assert rollover.state == CaRolloverState.CANCELLED + + +def test_rollover_invalid_plan_rolls_back_import(api, ca): + """Invalid strategy input must not leave pending CAs behind.""" + before = CaModel.objects.count() + response = api.post( + '/api/ca-rollovers/', + { + 'old_issuing_ca': ca.pk, + 'method': 'pkcs12', + 'pkcs12_file': SimpleUploadedFile('bad.p12', b'not a credential'), + }, + ) + assert response.status_code == 400 + assert CaModel.objects.count() == before + + +def test_generate_disabled_by_policy(api): + """API clients cannot bypass the AutoGenPKI policy check.""" + with patch('pki.views.ca_api.IssuingCaAddAutoGenForm') as form: + form.return_value.is_valid.return_value = True + response = api.post('/api/issuing-cas/generate/', {'key_type': 'RSA-2048'}, format='json') + assert response.status_code == 400 + + +@pytest.fixture +def lab_config(): + """Allow the same management operations as the UI's laboratory policy.""" + config = SecurityConfig(security_mode=SecurityConfig.SecurityModeChoices.LAB) + config.apply_security_settings(save=False) + config.save() + return config + + +@pytest.mark.parametrize('protocol', ['est', 'cmp']) +@pytest.mark.parametrize('is_ra_mode', [False, True]) +def test_remote_ca_configuration(api, ca, lab_config, protocol, is_ra_mode): + """Remote configuration uses validated UI forms and never returns authentication secrets.""" + data = { + 'unique_name': f'new-{protocol}-{is_ra_mode}', + 'key_type': 'RSA-2048', + 'remote_host': 'localhost', + 'remote_port': 443, + 'remote_path': f'/.well-known/{protocol}/enroll', + 'est_username': 'operator', + 'est_password': 'a-remote-password', + 'cmp_shared_secret': 'a-cmp-secret', + 'is_ra_mode': is_ra_mode, + } + response = api.post(f'/api/issuing-cas/configure-{protocol}/', data, format='json') + assert response.status_code == 201, response.data + created = CaModel.objects.get(pk=response.data['id']) + assert (created.credential_id is None) is is_ra_mode + assert 'est_password' not in response.data + assert 'cmp_shared_secret' not in response.data + assert api.post(f'/api/issuing-cas/{created.pk}/enroll/', {}, format='json').status_code == 400 + + +def test_pkcs12_import_and_legacy_pem_import(api, ca, lab_config): + """Both new PKCS#12 and existing PEM import contracts create real CAs.""" + from cryptography.hazmat.primitives.serialization import pkcs12 + + for method in ('pkcs12', 'pem'): + cert, key = CertificateGenerator.create_root_ca(cn=f'Imported-{method}') + if method == 'pkcs12': + data = { + 'pkcs12_file': SimpleUploadedFile( + 'ca.p12', + pkcs12.serialize_key_and_certificates(b'imported', key, cert, None, serialization.NoEncryption()), + ) + } + response = api.post('/api/issuing-cas/import-pkcs12/', data) + else: + data = { + 'private_key_pem': key.private_bytes( + serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption() + ).decode(), + 'ca_certificate_pem': cert.public_bytes(serialization.Encoding.PEM).decode(), + } + response = api.post('/api/issuing-cas/', data, format='json') + assert response.status_code == 201, response.data + assert CaModel.objects.get(pk=response.data['id']).get_certificate().serial_number == cert.serial_number + + +def test_external_csr_and_certificate_import(api, ca, lab_config): + """Generate a managed CSR and accept only the matching issued CA certificate.""" + from django.core.management import call_command + + call_command('create_default_cert_profiles') + response = api.post( + '/api/issuing-cas/configure-external-csr/', + {'unique_name': 'external-ca', 'key_type': 'RSA-2048'}, + format='json', + ) + assert response.status_code == 201, response.data + pk = response.data['id'] + response = api.post( + f'/api/issuing-cas/{pk}/csr/', + { + 'common_name': 'New Issuing CA', + 'organization_name': 'Example', + 'country_name': 'DE', + 'state_or_province_name': 'Berlin', + 'days': 30, + }, + format='json', + ) + assert response.status_code == 200, getattr(response, 'data', response.content) + csr = x509.load_pem_x509_csr(response.content) + assert csr.is_signature_valid + now = timezone.now() + cert = ( + x509.CertificateBuilder() + .subject_name(csr.subject) + .issuer_name(ca.get_certificate().subject) + .public_key(csr.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now) + .not_valid_after(now + timedelta(days=30)) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=True, + crl_sign=True, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + .sign(ca.credential.get_private_key(), hashes.SHA256()) + ) + wrong = api.post( + f'/api/issuing-cas/{pk}/import-certificate/', + {'certificate': SimpleUploadedFile('wrong.pem', ca.get_certificate().public_bytes(serialization.Encoding.PEM))}, + ) + assert wrong.status_code == 400 + uploaded = api.post( + f'/api/issuing-cas/{pk}/import-certificate/', + {'certificate': SimpleUploadedFile('ca.pem', cert.public_bytes(serialization.Encoding.PEM))}, + ) + assert uploaded.status_code == 200, uploaded.data + assert api.get(f'/api/issuing-cas/{pk}/certificate/').status_code == 200 + + +def test_issuing_ca_cannot_be_patched_as_generic_model(api, ca): + """Read-only representations must not pretend to accept generic configuration writes.""" + assert api.patch(f'/api/issuing-cas/{ca.pk}/', {'is_active': False}, format='json').status_code == 405 + + +@pytest.mark.parametrize('protocol', ['est', 'cmp']) +def test_upstream_ca_enrollment(api, ca, lab_config, protocol): + """Build/sign the real request, substituting only the upstream network response.""" + from django.core.management import call_command + + from pki.models.truststore import TruststoreModel, TruststoreOrderModel + + call_command('create_default_cert_profiles') + response = api.post( + f'/api/issuing-cas/configure-{protocol}/', + { + 'unique_name': f'enrolled-{protocol}', + 'key_type': 'RSA-2048', + 'remote_host': 'localhost', + 'remote_port': 443, + 'remote_path': f'/.well-known/{protocol}/enroll', + 'est_username': 'operator', + 'est_password': 'remote-password', + 'cmp_shared_secret': 'cmp-secret', + }, + format='json', + ) + assert response.status_code == 201, response.data + new_ca = CaModel.objects.get(pk=response.data['id']) + usage = TruststoreModel.IntendedUsage.TLS if protocol == 'est' else TruststoreModel.IntendedUsage.ISSUING_CA_CHAIN + store = TruststoreModel.objects.create(unique_name=f'upstream-{protocol}', intended_usage=usage) + TruststoreOrderModel.objects.create(trust_store=store, certificate=ca.ca_certificate_model, order=0) + response = api.post(f'/api/issuing-cas/{new_ca.pk}/truststore/', {'trust_store': store.pk}, format='json') + assert response.status_code == 200, response.data + now = timezone.now() + cert = ( + x509.CertificateBuilder() + .subject_name(x509.Name([x509.NameAttribute(x509.NameOID.COMMON_NAME, 'New CA')])) + .issuer_name(ca.get_certificate().subject) + .public_key(new_ca.credential.get_private_key().public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now) + .not_valid_after(now + timedelta(days=30)) + .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=True, + crl_sign=True, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + .sign(ca.credential.get_private_key(), hashes.SHA256()) + ) + path = ( + 'pki.services.ca_enrollment.EstClient.simple_enroll' + if protocol == 'est' + else 'pki.services.ca_enrollment.CmpClient.send_and_extract_certificate' + ) + upstream_result = cert if protocol == 'est' else (cert, [ca.get_certificate()]) + enrollment_data = { + 'common_name': 'New CA', + 'organization_name': 'Example', + 'country_name': 'DE', + 'state_or_province_name': 'Berlin', + 'days': 30, + } + service = 'EstCaEnrollmentService' if protocol == 'est' else 'CmpCaEnrollmentService' + with patch(f'pki.views.ca_api.{service}.enroll', side_effect=ValueError('secret credential path')) as enroll: + response = api.post(f'/api/issuing-cas/{new_ca.pk}/enroll/', enrollment_data, format='json') + enroll.assert_called_once() + assert response.status_code == 400 + assert response.json() == ['Unable to enroll the CA. Check the enrollment configuration and request data.'] + assert b'secret credential path' not in response.content + new_ca.refresh_from_db() + assert new_ca.credential.certificate is None + + with patch(path, return_value=upstream_result) as upstream: + response = api.post( + f'/api/issuing-cas/{new_ca.pk}/enroll/', + enrollment_data, + format='json', + ) + assert response.status_code == 200, response.data + upstream.assert_called_once() + new_ca.refresh_from_db() + assert new_ca.get_certificate().serial_number == cert.serial_number + + +def test_ca_api_openapi_contract(): + """The generated schema exposes new actions and immutable CRL/rollover resources.""" + from django.urls import include, path + from drf_spectacular.generators import SchemaGenerator + + schema = SchemaGenerator(patterns=[path('api/', include('pki.api_urls'))]).get_schema(public=True) + paths = schema['paths'] + assert 'post' in paths['/api/issuing-cas/{id}/enroll/'] + assert 'post' in paths['/api/issuing-cas/{id}/import-certificate/'] + assert 'patch' in paths['/api/issuing-cas/{id}/crl-settings/'] + assert 'patch' not in paths['/api/issuing-cas/{id}/'] + assert 'post' in paths['/api/crls/'] + assert 'get' in paths['/api/crls/{id}/download/'] + assert 'post' in paths['/api/ca-rollovers/{id}/complete/'] + + +def test_public_crl_url_name_is_preserved(): + """An API download must not shadow the existing public CRL URL name.""" + from django.urls import reverse + + assert reverse('crl-download', kwargs={'pk': 12}) == '/crl/12/' + assert reverse('crl-api-download', kwargs={'pk': 12}) == '/api/crls/12/download/' + + +def test_remote_protocol_policy_denial_rolls_back(api, ca, lab_config): + """Policy rejections return a client error and do not leave a configured CA behind.""" + lab_config.permitted_no_onboarding_pki_protocols = 0 + lab_config.save() + response = api.post( + '/api/issuing-cas/configure-est/', + { + 'unique_name': 'forbidden-est', + 'key_type': 'RSA-2048', + 'remote_host': 'localhost', + 'remote_port': 443, + 'remote_path': '/.well-known/est/simpleenroll', + 'est_username': 'operator', + 'est_password': 'remote-password', + }, + format='json', + ) + assert response.status_code == 400, response.data + assert not CaModel.objects.filter(unique_name='forbidden-est').exists() diff --git a/trustpoint/pki/tests/test_certificate_api.py b/trustpoint/pki/tests/test_certificate_api.py new file mode 100644 index 000000000..e6bb06822 --- /dev/null +++ b/trustpoint/pki/tests/test_certificate_api.py @@ -0,0 +1,376 @@ +# Copyright (c) 2026 The Trustpoint Project Authors +# SPDX-License-Identifier: MIT + +"""Public certificate detail and download behavior through the REST API.""" + +import io +import ipaddress +import tarfile +import zipfile +from datetime import timedelta + +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives.serialization import pkcs7 +from django.urls import reverse +from django.utils import timezone +from rest_framework.test import APIClient + +from pki.models.certificate import CertificateModel, RevokedCertificateModel +from users.models import TrustpointUser + + +@pytest.fixture +def api(): + """Use the same authenticated access as the existing certificate API.""" + user = TrustpointUser.objects.create_user(username='certificate-reader', password='test-password') + client = APIClient() + client.force_authenticate(user) + return client + + +@pytest.fixture +def leaf(issuing_ca_instance): + """A real certificate with structured and unrecognized extensions, signed by a known CA.""" + now = timezone.now() + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + cert = ( + x509.CertificateBuilder() + .subject_name( + x509.Name( + [ + x509.RelativeDistinguishedName([x509.NameAttribute(x509.NameOID.COMMON_NAME, 'shared-name')]), + x509.RelativeDistinguishedName( + [ + x509.NameAttribute(x509.NameOID.ORGANIZATIONAL_UNIT_NAME, 'One'), + x509.NameAttribute(x509.NameOID.ORGANIZATIONAL_UNIT_NAME, 'Two'), + ] + ), + ] + ) + ) + .issuer_name(issuing_ca_instance['cert'].subject) + .public_key(key.public_key()) + .serial_number(2**120 + 17) + .not_valid_before(now - timedelta(days=1)) + .not_valid_after(now + timedelta(days=30)) + .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=True, + data_encipherment=False, + key_agreement=False, + key_cert_sign=False, + crl_sign=False, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + .add_extension( + x509.SubjectAlternativeName( + [ + x509.DNSName('example.test'), + x509.IPAddress(ipaddress.ip_address('192.0.2.1')), + x509.UniformResourceIdentifier('urn:test'), + x509.RFC822Name('user@example.test'), + ] + ), + critical=False, + ) + .add_extension(x509.ExtendedKeyUsage([x509.ExtendedKeyUsageOID.CLIENT_AUTH]), critical=False) + .add_extension(x509.UnrecognizedExtension(x509.ObjectIdentifier('1.2.3.4.5'), b'\x0c\x04test'), critical=False) + .sign(issuing_ca_instance['priv_key'], hashes.SHA256()) + ) + return CertificateModel.save_certificate(cert) + + +def parse_certificates(content, file_format): + """Parse downloaded bytes independently of the production serializers.""" + return { + 'pem': x509.load_pem_x509_certificates, + 'der': lambda data: [x509.load_der_x509_certificate(data)], + 'pkcs7_pem': pkcs7.load_pem_pkcs7_certificates, + 'pkcs7_der': pkcs7.load_der_pkcs7_certificates, + }[file_format](content) + + +def test_detail_contains_full_public_certificate(api, leaf, issuing_ca_instance): + """Details preserve certificate identity, name groupings, and extension information.""" + response = api.get(f'/api/certificates/{leaf.pk}/') + assert response.status_code == 200, response.data + data = response.data + assert data['serial_number'] == f'{2**120 + 17:X}' + assert data['sha256_fingerprint'] == leaf.sha256_fingerprint + assert data['certificate_status'] == 'OK' + assert data['issuer_certificate'] == issuing_ca_instance['issuing_ca'].ca_certificate_model.pk + assert data['issued_certificate_count'] == 0 + root = issuing_ca_instance['issuing_ca'].ca_certificate_model + assert api.get(f'/api/certificates/{root.pk}/').data['issued_certificate_count'] == 1 + assert data['is_ca'] is False + assert data['spki_key_size'] == 2048 + assert data['revocation'] is None + assert data['cert_pem'] == leaf.cert_pem + assert 'PRIVATE KEY' not in str(data) + unit_names = [attribute for attribute in data['subject'] if attribute['oid'] == '2.5.4.11'] + assert {attribute['value'] for attribute in unit_names} == {'One', 'Two'} + assert len({attribute['rdn'] for attribute in unit_names}) == 1 + extensions = {item['oid']: item for item in data['extensions']} + assert extensions['2.5.29.19']['value'] == {'ca': False, 'path_length': None} + assert extensions['2.5.29.15']['value']['encipher_only'] is None + assert extensions['1.2.3.4.5']['der_hex'] == '0C0474657374' + assert {'type': 'IPAddress', 'value': '192.0.2.1'} in extensions['2.5.29.17']['value'] + + +def test_list_adds_summary_and_preserves_search(api, leaf): + """Inventory reads include status/expiry without copying full certificate contents into each row.""" + response = api.get('/api/certificates/', {'search': 'shared-name'}) + assert response.status_code == 200 + data = response.data['results'] if isinstance(response.data, dict) else response.data + row = next(item for item in data if item['id'] == leaf.pk) + assert row['common_name'] == 'shared-name' + assert row['serial_number'] == leaf.serial_number + assert row['certificate_status'] == 'OK' + assert 'cert_pem' not in row + assert 'not_valid_after' in row + + +def test_revocation_details(api, leaf, issuing_ca_instance): + """The current certificate status and revocation record agree.""" + ca = issuing_ca_instance['issuing_ca'] + RevokedCertificateModel.objects.create(certificate=leaf, ca=ca, revocation_reason='keyCompromise') + data = api.get(f'/api/certificates/{leaf.pk}/').data + assert data['certificate_status'] == 'REV' + assert data['revocation']['revocation_reason'] == 'keyCompromise' + assert data['revocation']['ca'] == ca.pk + assert data['revocation']['revoked_at'] + + +@pytest.mark.parametrize('file_format', ['pem', 'der', 'pkcs7_pem', 'pkcs7_der']) +def test_single_download_all_formats(api, leaf, file_format): + """Every format round-trips to the same certificate and has an attachment filename.""" + response = api.get(f'/api/certificates/{leaf.pk}/download/', {'file_format': file_format}) + assert response.status_code == 200 + assert parse_certificates(response.content, file_format)[0].serial_number == 2**120 + 17 + assert f'shared-name-{leaf.pk}' in response['Content-Disposition'] + + +@pytest.mark.parametrize('file_format', ['pem', 'pkcs7_pem', 'pkcs7_der']) +def test_download_with_issuer_chain(api, leaf, issuing_ca_instance, file_format): + """A single PEM/PKCS#7 file can contain the leaf and its known issuer chain.""" + root = issuing_ca_instance['issuing_ca'].ca_certificate_model + response = api.get(f'/api/certificates/{leaf.pk}/download/', {'file_format': file_format, 'include_chain': 'true'}) + assert response.status_code == 200, getattr(response, 'data', response.content) + downloaded = parse_certificates(response.content, file_format) + assert {cert.serial_number for cert in downloaded} == {2**120 + 17, int(root.serial_number, 16)} + assert api.get(f'/api/certificates/{leaf.pk}/').data['certificate_chain'] == [root.pk] + + +def test_der_chain_is_rejected(api, leaf): + """DER files must never silently concatenate multiple certificates.""" + response = api.get(f'/api/certificates/{leaf.pk}/download/', {'file_format': 'der', 'include_chain': 'true'}) + assert response.status_code == 400 + + +@pytest.mark.parametrize('archive_format', ['zip', 'tar_gz']) +@pytest.mark.parametrize('file_format', ['pem', 'der', 'pkcs7_pem', 'pkcs7_der']) +def test_archive_formats_and_chain_deduplication(api, leaf, issuing_ca_instance, archive_format, file_format): + """Archives contain valid individual files, with shared issuers included once.""" + root = issuing_ca_instance['issuing_ca'].ca_certificate_model + response = api.post( + '/api/certificates/download/', + { + 'certificate_ids': [leaf.pk, root.pk, leaf.pk], + 'file_format': file_format, + 'archive_format': archive_format, + 'include_chain': True, + }, + format='json', + ) + assert response.status_code == 200, getattr(response, 'data', response.content) + contents = io.BytesIO(response.content) + if archive_format == 'zip': + with zipfile.ZipFile(contents) as archive: + assert len(archive.namelist()) == 2 + files = [archive.read(name) for name in archive.namelist()] + else: + with tarfile.open(fileobj=contents, mode='r:gz') as archive: + assert len(archive.getnames()) == 2 + files = [archive.extractfile(member).read() for member in archive.getmembers()] + assert {parse_certificates(file, file_format)[0].serial_number for file in files} == { + 2**120 + 17, + int(root.serial_number, 16), + } + + +@pytest.mark.parametrize( + 'payload', + [ + {}, + {'certificate_ids': []}, + {'certificate_ids': ['invalid']}, + {'certificate_ids': [-1]}, + {'certificate_ids': [1], 'file_format': 'invalid'}, + {'certificate_ids': [1], 'archive_format': 'invalid'}, + ], +) +def test_invalid_archive_selection(api, payload): + """Invalid input produces a validation response rather than a partial archive.""" + assert api.post('/api/certificates/download/', payload, format='json').status_code == 400 + + +def test_missing_certificate_and_invalid_query(api, leaf): + """Missing IDs produce 404 and unsupported download options produce 400.""" + assert api.get('/api/certificates/999999/').status_code == 404 + assert api.get('/api/certificates/999999/download/').status_code == 404 + assert api.get(f'/api/certificates/{leaf.pk}/download/', {'file_format': 'invalid'}).status_code == 400 + assert api.get(f'/api/certificates/{leaf.pk}/download/', {'include_chain': 'invalid'}).status_code == 400 + assert ( + api.post('/api/certificates/download/', {'certificate_ids': [leaf.pk, 999999]}, format='json').status_code + == 404 + ) + + +def test_downloads_require_authentication(leaf): + """The new reads/downloads use the existing authenticated certificate access policy.""" + client = APIClient() + for url in (f'/api/certificates/{leaf.pk}/', f'/api/certificates/{leaf.pk}/download/'): + assert client.get(url).status_code in (401, 403) + assert client.post('/api/certificates/download/', {'certificate_ids': [leaf.pk]}, format='json').status_code in ( + 401, + 403, + ) + + +def test_existing_write_validation_is_preserved(api, leaf): + """Richer read schemas do not bypass the existing write field constraints.""" + response = api.patch(f'/api/certificates/{leaf.pk}/', {'common_name': 'x' * 257}, format='json') + assert response.status_code == 400 + assert 'common_name' in response.data + assert CertificateModel.objects.get(pk=leaf.pk).common_name == 'shared-name' + + +def test_certificate_openapi_contract(): + """Details and both download actions have usable generated schemas.""" + from django.urls import include, path + from drf_spectacular.generators import SchemaGenerator + + schema = SchemaGenerator(patterns=[path('api/', include('pki.api_urls'))]).get_schema(public=True) + paths = schema['paths'] + assert 'get' in paths['/api/certificates/{id}/download/'] + assert 'post' in paths['/api/certificates/download/'] + assert 'CertificateDetail' in schema['components']['schemas'] + assert reverse('pki:certificate-download', kwargs={'pk': 12}) == '/pki/certificates/download/12' + + +def test_duplicate_common_names_have_distinct_archive_files(api, leaf, issuing_ca_instance): + """Certificates sharing a common name cannot overwrite each other in an archive.""" + original = leaf.get_certificate_serializer().as_crypto() + second = CertificateModel.save_certificate( + x509.CertificateBuilder() + .subject_name(original.subject) + .issuer_name(original.issuer) + .public_key(original.public_key()) + .serial_number(42) + .not_valid_before(original.not_valid_before_utc) + .not_valid_after(original.not_valid_after_utc) + .sign(issuing_ca_instance['priv_key'], hashes.SHA256()) + ) + response = api.post('/api/certificates/download/', {'certificate_ids': [leaf.pk, second.pk]}, format='json') + assert response.status_code == 200 + with zipfile.ZipFile(io.BytesIO(response.content)) as archive: + assert set(archive.namelist()) == {f'shared-name-{leaf.pk}.pem', f'shared-name-{second.pk}.pem'} + assert {x509.load_pem_x509_certificate(archive.read(name)).serial_number for name in archive.namelist()} == { + 2**120 + 17, + 42, + } + + +def test_unsafe_certificate_names_are_safe_attachments(api, leaf): + """Untrusted names cannot insert headers or archive paths.""" + CertificateModel.objects.filter(pk=leaf.pk).update(common_name='../..//\r\n"unsafe') + response = api.get(f'/api/certificates/{leaf.pk}/download/') + assert response.status_code == 200 + assert '\r' not in response['Content-Disposition'] + assert '\n' not in response['Content-Disposition'] + response = api.post('/api/certificates/download/', {'certificate_ids': [leaf.pk]}, format='json') + with zipfile.ZipFile(io.BytesIO(response.content)) as archive: + assert len(archive.namelist()) == 1 + assert '/' not in archive.namelist()[0] + assert not archive.namelist()[0].startswith('.') + + +def test_cyclic_issuer_chain_returns_validation_error(api, leaf, issuing_ca_instance): + """Malformed hierarchy data must not make downloads recurse forever.""" + root = issuing_ca_instance['issuing_ca'].ca_certificate_model + CertificateModel.objects.filter(pk=root.pk).update(issuer_id=leaf) + response = api.get(f'/api/certificates/{leaf.pk}/download/', {'include_chain': 'true'}) + assert response.status_code == 400 + assert api.get(f'/api/certificates/{leaf.pk}/').status_code == 400 + + +def test_certificate_detail_does_not_expose_chain_exception(api, leaf, monkeypatch): + """Internal exception details must not appear in certificate API responses.""" + sensitive_detail = 'Internal credential path: /private/credentials/issuer.pem' + + def fail_certificate_chain(certificate): + raise ValueError(sensitive_detail) + + monkeypatch.setattr('pki.serializer.certificate.certificate_chain', fail_certificate_chain) + + response = api.get(f'/api/certificates/{leaf.pk}/') + + assert response.status_code == 400 + assert response.json() == ['Unable to build the certificate issuer chain.'] + assert sensitive_detail not in response.content.decode() + + +@pytest.mark.parametrize('download_mode', ['single', 'archive']) +def test_certificate_download_does_not_expose_chain_exception(api, leaf, monkeypatch, download_mode): + """Both export endpoints hide internal exception details when building issuer chains.""" + sensitive_detail = 'Internal credential path: /private/credentials/issuer.pem' + + def fail_certificate_chain(certificate): + raise ValueError(sensitive_detail) + + monkeypatch.setattr('pki.views.certificates.certificate_chain', fail_certificate_chain) + + if download_mode == 'single': + response = api.get(f'/api/certificates/{leaf.pk}/download/', {'include_chain': 'true'}) + else: + response = api.post( + '/api/certificates/download/', {'certificate_ids': [leaf.pk], 'include_chain': True}, format='json' + ) + + assert response.status_code == 400 + assert response.json() == ['Unable to build the certificate issuer chain.'] + assert sensitive_detail not in response.content.decode() + + +def test_root_der_download_with_chain_does_not_duplicate_root(api, issuing_ca_instance): + """A root certificate alone remains a valid single DER certificate.""" + root = issuing_ca_instance['issuing_ca'].ca_certificate_model + response = api.get(f'/api/certificates/{root.pk}/download/', {'file_format': 'der', 'include_chain': 'true'}) + assert response.status_code == 200 + assert x509.load_der_x509_certificate(response.content).serial_number == int(root.serial_number, 16) + data = api.get(f'/api/certificates/{root.pk}/').data + assert data['is_root_ca'] is True + assert data['certificate_chain'] == [] + + +def test_associated_device_and_profile_metadata(api, tls_client_credential_instance): + """The UI's device/profile association is available without exposing private credentials.""" + issued = tls_client_credential_instance['issued_credential'] + response = api.get(f'/api/certificates/{issued.credential.certificate.pk}/') + assert response.status_code == 200, response.data + data = response.data['issued_credentials'] + assert len(data) == 1 + assert data[0]['id'] == issued.pk + assert data[0]['device'] == issued.device_id + assert data[0]['device_name'] == issued.device.common_name + assert data[0]['issued_using_cert_profile'] == issued.issued_using_cert_profile + assert 'private_key' not in data[0] diff --git a/trustpoint/pki/util/crl.py b/trustpoint/pki/util/crl.py index 41b876fe1..9e1e9f924 100644 --- a/trustpoint/pki/util/crl.py +++ b/trustpoint/pki/util/crl.py @@ -36,7 +36,7 @@ def generate_empty_crl( ca_cert: x509.Certificate, private_key: rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey, hash_algorithm: hashes.HashAlgorithm | None = None, - crl_validity_hours: int = 2400, + crl_validity_hours: float = 2400, crl_number: int = 1, ) -> str: """Generate an empty CRL for a CA. @@ -89,7 +89,7 @@ def generate_empty_crl( def generate_crl_with_revoked_certs( issuing_ca: CaModel, - crl_validity_hours: int = 24, + crl_validity_hours: float = 24, ) -> CertificateRevocationList: """Generate a CRL with revoked certificates for an issuing CA. diff --git a/trustpoint/pki/views/ca_api.py b/trustpoint/pki/views/ca_api.py new file mode 100644 index 000000000..1b3b300c4 --- /dev/null +++ b/trustpoint/pki/views/ca_api.py @@ -0,0 +1,672 @@ +# Copyright (c) 2026 The Trustpoint Project Authors +# SPDX-License-Identifier: MIT + +"""REST management of certificate authorities, CRLs, and CA rollovers.""" + +from copy import copy +from typing import Any, ClassVar + +from cryptography import x509 +from cryptography.hazmat.primitives import serialization +from django.core.exceptions import ValidationError as DjangoValidationError +from django.db import IntegrityError, transaction +from django.db.models import ProtectedError +from django.http import HttpResponse +from django.shortcuts import get_object_or_404 +from django.utils.datastructures import MultiValueDict +from django_filters.rest_framework import DjangoFilterBackend +from drf_spectacular.types import OpenApiTypes +from drf_spectacular.utils import extend_schema +from rest_framework import filters, mixins, status, viewsets +from rest_framework.decorators import action +from rest_framework.exceptions import APIException, MethodNotAllowed, ValidationError +from rest_framework.request import Request +from rest_framework.response import Response +from rest_framework.views import APIView + +from management.models import SecurityConfig +from management.models.audit_log import AuditLog +from management.security_change_audit import audit_security_change, security_change, snapshot_security_state +from pki.auto_gen_pki import AutoGenPki +from pki.forms.cert_profiles import CertificateIssuanceForm +from pki.forms.issuing_cas import ( + IssuingCaAddAutoGenForm, + IssuingCaAddFileImportPkcs12Form, + IssuingCaAddRequestCmpForm, + IssuingCaAddRequestEstForm, + IssuingCaAddRequestExternalCsrForm, + IssuingCaCrlCycleForm, + IssuingCaExternalCsrCertificateForm, + IssuingCaImportMixin, + IssuingCaTruststoreAssociationForm, +) +from pki.models import CaModel, CertificateModel, CrlModel +from pki.models.ca_rollover import CaRolloverModel, CaRolloverStrategyType +from pki.models.cert_profile import CertificateProfileModel +from pki.models.credential import PrimaryCredentialCertificate +from pki.models.truststore import TruststoreModel +from pki.rollover.generate_keypair import GenerateKeypairCmpRolloverForm, GenerateKeypairEstRolloverForm +from pki.rollover.import_ca import ImportCaRolloverForm, SeparateFilesCaRolloverForm +from pki.serializer.ca_api import ( + CaCertificateSerializer, + CaCertificateUploadSerializer, + CaCrlSettingsSerializer, + CaEnrollmentSerializer, + CaExternalCsrSerializer, + CaGenerateSerializer, + CaPkcs12Serializer, + CaRemoteSerializer, + CaRolloverPlanSerializer, + CaRolloverSerializer, + CaSerializer, + CaTruststoreSerializer, + CrlImportSerializer, + CrlSerializer, + EncodingSerializer, +) +from pki.services.ca_enrollment import CmpCaEnrollmentService, EstCaEnrollmentService +from pki.services.ca_rollover import CaRolloverError, CaRolloverService +from pki.services.external_csr import build_external_csr, certificate_matches_credential +from pki.util.cert_profile import ProfileValidationError +from pki.views.issuing_cas import CanManageIssuingCas, IssuingCaViewSet +from request.clients import EstClientError +from request.clients.cmp_client import CmpClientError + + +class Conflict(APIException): + """A protected dependency or lifecycle state prevents this operation.""" + + status_code = status.HTTP_409_CONFLICT + default_detail = 'The operation conflicts with the current state.' + + +class UpstreamError(APIException): + """The upstream PKI could not complete certificate enrollment.""" + + status_code = status.HTTP_502_BAD_GATEWAY + default_detail = 'The upstream PKI could not complete certificate enrollment.' + + +class PkiApiErrorsMixin(APIView): + """Translate shared domain validation into stable API errors.""" + + def handle_exception(self, exc: Exception) -> Response: + """Preserve DRF authentication handling and map domain errors.""" + if isinstance(exc, ProtectedError): + exc = Conflict('The object is referenced by other objects and cannot be deleted.') + elif isinstance(exc, IntegrityError): + exc = Conflict('An object with these identifiers already exists.') + elif isinstance(exc, DjangoValidationError): + exc = ValidationError(exc.message_dict if hasattr(exc, 'message_dict') else exc.messages) + elif isinstance(exc, (ProfileValidationError, ValueError)): + exc = ValidationError('The PKI operation could not be completed. Check the input and configuration.') + elif isinstance(exc, CaRolloverError): + exc = Conflict('The CA rollover could not be completed in its current state.') + return super().handle_exception(exc) + + +def validated_form(form: Any) -> Any: + """Raise on invalid forms so surrounding transactions roll back form side effects.""" + if not form.is_valid(): + raise ValidationError({key: list(errors) for key, errors in form.errors.items()}) + return form + + +def download_response(content: bytes, filename: str, content_type: str) -> HttpResponse: + """Build an attachment response with a safe generated filename.""" + safe_name = ''.join(char if char.isalnum() or char in '._-' else '_' for char in filename) + response = HttpResponse(content, content_type=content_type) + response['Content-Disposition'] = f'attachment; filename="{safe_name}"' + return response + + +class CaInventoryMixin(viewsets.GenericViewSet[CaModel]): + """Public certificate inspection and guarded deletion shared by both CA inventories.""" + + @transaction.atomic + def perform_destroy(self, instance: CaModel) -> None: + """Honor model dependency checks and audit the deletion atomically.""" + instance = CaModel.objects.select_for_update().get(pk=instance.pk) + target = copy(instance) + before = snapshot_security_state(instance) + instance.delete() + audit_security_change(target=target, actor=self.request.user, before=before, action='deleted') + + @extend_schema(parameters=[EncodingSerializer], responses={(200, 'application/octet-stream'): OpenApiTypes.BINARY}) + @action(detail=True, methods=['get'], url_path='certificate') + def certificate(self, request: Request, pk: str | None = None) -> HttpResponse: + """Download the CA certificate in PEM or DER encoding.""" + del pk + ca = self.get_object() + certificate = ca.ca_certificate_model + if certificate is None: + from rest_framework.exceptions import NotFound # noqa: PLC0415 + + msg = 'The CA does not have a certificate yet.' + raise NotFound(msg) + query = EncodingSerializer(data=request.query_params) + query.is_valid(raise_exception=True) + encoding = query.validated_data['encoding'] + crypto_certificate = certificate.get_certificate_serializer().as_crypto() + return download_response( + crypto_certificate.public_bytes( + serialization.Encoding.PEM if encoding == 'pem' else serialization.Encoding.DER + ), + f'{ca.unique_name}.{encoding}', + 'application/x-pem-file' if encoding == 'pem' else 'application/pkix-cert', + ) + + @extend_schema(responses=CaCertificateSerializer(many=True)) + @action(detail=True, methods=['get'], url_path='issued-certificates') + def issued_certificates(self, request: Request, pk: str | None = None) -> Response: + """List public certificates issued by this CA.""" + del pk + certificates = self.get_object().get_issued_certificates().order_by('-not_valid_before', 'pk') + paginator = self.paginator + page = paginator.paginate_queryset(certificates, request, view=self) if paginator is not None else None + data = CaCertificateSerializer(page if page is not None else certificates, many=True).data + return self.get_paginated_response(data) if page is not None else Response(data) + + +@extend_schema(tags=['CA']) +class CaViewSet( + PkiApiErrorsMixin, + CaInventoryMixin, + mixins.ListModelMixin, + mixins.RetrieveModelMixin, + mixins.DestroyModelMixin, + viewsets.GenericViewSet[CaModel], +): + """Read the full CA hierarchy, including keyless and autogenerated roots, or delete an unused CA.""" + + queryset = CaModel.objects.select_related('parent_ca', 'certificate', 'credential__certificate').order_by('pk') + serializer_class = CaSerializer + permission_classes = (CanManageIssuingCas,) + filter_backends = (DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter) + filterset_fields: ClassVar = ['ca_type', 'is_active', 'parent_ca'] + search_fields: ClassVar = ['unique_name'] + ordering_fields: ClassVar = ['unique_name', 'created_at'] + + +class ApiPkcs12Form(IssuingCaAddFileImportPkcs12Form): + """Capture the CA created by the shared import form, including generated names.""" + + def _finalize_issuing_ca_creation(self, *args: Any, **kwargs: Any) -> CaModel: + self.created_ca = super()._finalize_issuing_ca_creation(*args, **kwargs) + return self.created_ca + + +@extend_schema(tags=['Issuing-CA']) +class IssuingCaManagementViewSet(PkiApiErrorsMixin, CaInventoryMixin, IssuingCaViewSet): + """Extend the existing issuing CA API without changing its PEM import contract.""" + + serializer_class = CaSerializer + http_method_names = ['get', 'post', 'delete', 'patch', 'head', 'options'] # noqa: RUF012 + + def retrieve(self, request: Request, *args: Any, **kwargs: Any) -> Response: + """Retrieve a CA with DRF's router-provided lookup keyword arguments.""" + return mixins.RetrieveModelMixin.retrieve(self, request, *args, **kwargs) + + @extend_schema(exclude=True) + def partial_update(self, request: Request, *args: Any, **kwargs: Any) -> Response: + """Use dedicated actions for validated CA configuration changes.""" + del request, args, kwargs + msg = 'PATCH' + raise MethodNotAllowed(msg) + + def _certificate_content(self, request: Request) -> tuple[CertificateProfileModel, dict[str, Any], int | None]: + """Validate profile-based content without depending on a browser session.""" + serializer = CaEnrollmentSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + values = dict(serializer.validated_data) + profile_pk = values.pop('cert_profile_pk', None) + sender_kid = values.pop('sender_kid', None) + profile = ( + get_object_or_404(CertificateProfileModel, pk=profile_pk) + if profile_pk + else get_object_or_404(CertificateProfileModel, unique_name='issuing_ca') + ) + form = validated_form(CertificateIssuanceForm(data=values, profile=profile.profile)) + return profile, form.cleaned_data, sender_kid + + @extend_schema(request=CaEnrollmentSerializer, responses={(200, 'application/pkcs10'): OpenApiTypes.BINARY}) + @action(detail=True, methods=['post'], url_path='csr') + def csr(self, request: Request, pk: str | None = None) -> HttpResponse: + """Generate a profile-based CSR using the CA's managed private key.""" + del pk + ca = self.get_object() + allowed_types = ( + CaModel.CaTypeChoice.REMOTE_ISSUING_CSR, + CaModel.CaTypeChoice.REMOTE_ISSUING_EST, + CaModel.CaTypeChoice.REMOTE_ISSUING_CMP, + ) + if ca.ca_type not in allowed_types or ca.credential is None or not ca.is_active: + msg = 'This CA is not configured for a certificate request.' + raise ValidationError(msg) + profile, content, _sender_kid = self._certificate_content(request) + try: + csr = build_external_csr(ca.credential, profile, content, allow_ca=True) + except ValueError as exc: + msg = 'Unable to build the CA certificate signing request. Check the profile and request data.' + raise ValidationError(msg) from exc + return download_response( + csr.public_bytes(serialization.Encoding.PEM), f'{ca.unique_name}.csr', 'application/pkcs10' + ) + + @extend_schema(request=CaCertificateUploadSerializer, responses=CaSerializer) + @action(detail=True, methods=['post'], url_path='import-certificate') + @transaction.atomic + def import_certificate(self, request: Request, pk: str | None = None) -> Response: + """Attach a matching CA certificate to a pending managed credential.""" + del pk + ca = self.get_object() + ca = CaModel.objects.select_for_update().get(pk=ca.pk) + allowed_types = ( + CaModel.CaTypeChoice.REMOTE_ISSUING_CSR, + CaModel.CaTypeChoice.REMOTE_ISSUING_EST, + CaModel.CaTypeChoice.REMOTE_ISSUING_CMP, + ) + if ca.ca_type not in allowed_types or not ca.is_active or ca.credential is None: + msg = 'This CA is not awaiting an externally issued certificate.' + raise ValidationError(msg) + if ca.credential.certificate is not None: + msg = 'The CA already has a certificate. Use a CA rollover to replace it.' + raise Conflict(msg) + serializer = CaCertificateUploadSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + form = validated_form( + IssuingCaExternalCsrCertificateForm( + files={'certificate': serializer.validated_data['certificate']}, instance=ca + ) + ) + cert = form.cleaned_data['certificate'].as_crypto() + IssuingCaImportMixin()._validate_ca_certificate(cert) # noqa: SLF001 + with security_change(target=ca, actor=request.user): + ca.credential.certificate = CertificateModel.save_certificate(cert) + ca.credential.save() + PrimaryCredentialCertificate.objects.update_or_create( + credential=ca.credential, defaults={'certificate': ca.credential.certificate, 'is_primary': True} + ) + CaRolloverService.complete_awaiting_rollover_for_ca(ca) + return Response(CaSerializer(ca).data) + + @extend_schema(request=CaEnrollmentSerializer, responses=CaSerializer) + @action(detail=True, methods=['post'], url_path='enroll') + @transaction.atomic + def enroll(self, request: Request, pk: str | None = None) -> Response: + """Request a CA certificate from its configured upstream EST or CMP server.""" + del pk + ca = self.get_object() + ca = CaModel.objects.select_for_update().get(pk=ca.pk) + if ca.ca_type not in (CaModel.CaTypeChoice.REMOTE_ISSUING_EST, CaModel.CaTypeChoice.REMOTE_ISSUING_CMP): + msg = 'Only remote EST or CMP issuing CAs support enrollment.' + raise ValidationError(msg) + if not ca.is_active or ca.credential is None or ca.no_onboarding_config is None: + msg = 'The CA has no active managed enrollment configuration.' + raise ValidationError(msg) + if not ca.no_onboarding_config.trust_store_id: + msg = 'Associate the upstream truststore before enrollment.' + raise ValidationError(msg) + if ca.credential.certificate is not None: + msg = 'The CA already has a certificate. Use a CA rollover to replace it.' + raise Conflict(msg) + profile, content, sender_kid = self._certificate_content(request) + with security_change(target=ca, actor=request.user): + try: + if ca.ca_type == CaModel.CaTypeChoice.REMOTE_ISSUING_EST: + EstCaEnrollmentService().enroll(ca, content, profile) + else: + CmpCaEnrollmentService().enroll(ca, content, profile, sender_kid=sender_kid) + except (EstClientError, CmpClientError) as exc: + raise UpstreamError from exc + except ValueError as exc: + msg = 'Unable to enroll the CA. Check the enrollment configuration and request data.' + raise ValidationError(msg) from exc + certificate_model = ca.credential.certificate + if certificate_model is None: + raise UpstreamError + cert = certificate_model.get_certificate_serializer().as_crypto() + IssuingCaImportMixin()._validate_ca_certificate(cert) # noqa: SLF001 + if not certificate_matches_credential(cert, ca.credential): + msg = 'The upstream certificate does not match the managed key.' + raise ValidationError(msg) + CaRolloverService.complete_awaiting_rollover_for_ca(ca) + return Response(CaSerializer(ca).data) + + @extend_schema(request=CaGenerateSerializer, responses={201: CaSerializer}) + @action(detail=False, methods=['post']) + @transaction.atomic + def generate(self, request: Request) -> Response: + """Generate the local AutoGenPKI when enabled by the security policy.""" + serializer = CaGenerateSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + form = validated_form(IssuingCaAddAutoGenForm(serializer.validated_data)) + config = SecurityConfig.objects.select_for_update().first() + if config is None or not config.auto_gen_pki: + msg = 'Auto-generated PKI creation is disabled in the Security Settings.' + raise ValidationError(msg) + if AutoGenPki.get_auto_gen_pki() is not None: + msg = 'An auto-generated PKI is already active.' + raise Conflict(msg) + try: + ca = AutoGenPki.enable_auto_gen_pki(form.cleaned_data['key_type']) + except ValueError as exc: + msg = 'Unable to generate the PKI. Check the key type and security settings.' + raise ValidationError(msg) from exc + if ca is None: + msg = 'An auto-generated PKI is already active.' + raise Conflict(msg) + return self._created(ca) + + def _created(self, ca: CaModel) -> Response: + """Audit creation and return the complete public CA representation.""" + audit_security_change(target=ca, actor=self.request.user, before=None, action='created') + return Response(CaSerializer(ca).data, status=status.HTTP_201_CREATED) + + @extend_schema(request=CaPkcs12Serializer, responses={201: CaSerializer}) + @action(detail=False, methods=['post'], url_path='import-pkcs12') + @transaction.atomic + def import_pkcs12(self, request: Request) -> Response: + """Import a CA from a multipart PKCS#12 upload.""" + serializer = CaPkcs12Serializer(data=request.data) + serializer.is_valid(raise_exception=True) + values = dict(serializer.validated_data) + file = values.pop('pkcs12_file') + form = validated_form(ApiPkcs12Form(values, MultiValueDict({'pkcs12_file': [file]}))) + return self._created(form.created_ca) + + def _configure_remote(self, request: Request, protocol: str) -> Response: + """Reuse UI validation and managed-key creation for upstream configurations.""" + serializer = CaRemoteSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + values = dict(serializer.validated_data) + is_ra_mode = values.pop('is_ra_mode') + form_class = IssuingCaAddRequestEstForm if protocol == 'est' else IssuingCaAddRequestCmpForm + values['ca_type'] = ( + CaModel.CaTypeChoice.REMOTE_ISSUING_EST if protocol == 'est' else CaModel.CaTypeChoice.REMOTE_ISSUING_CMP + ) + form = validated_form(form_class(values)) + return self._created(form.save(is_ra_mode=is_ra_mode)) + + @extend_schema(request=CaRemoteSerializer, responses={201: CaSerializer}) + @action(detail=False, methods=['post'], url_path='configure-est') + @transaction.atomic + def configure_est(self, request: Request) -> Response: + """Configure a remote EST CA or RA; associate its truststore next.""" + return self._configure_remote(request, 'est') + + @extend_schema(request=CaRemoteSerializer, responses={201: CaSerializer}) + @action(detail=False, methods=['post'], url_path='configure-cmp') + @transaction.atomic + def configure_cmp(self, request: Request) -> Response: + """Configure a remote CMP CA or RA; associate its truststore next.""" + return self._configure_remote(request, 'cmp') + + @extend_schema(request=CaExternalCsrSerializer, responses={201: CaSerializer}) + @action(detail=False, methods=['post'], url_path='configure-external-csr') + @transaction.atomic + def configure_external_csr(self, request: Request) -> Response: + """Create a managed key for an externally signed issuing CA.""" + serializer = CaExternalCsrSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + values = {**serializer.validated_data, 'ca_type': CaModel.CaTypeChoice.REMOTE_ISSUING_CSR} + form = validated_form(IssuingCaAddRequestExternalCsrForm(values)) + return self._created(form.save()) + + @extend_schema(request=CaCrlSettingsSerializer, responses=CaSerializer) + @action(detail=True, methods=['patch'], url_path='crl-settings') + @transaction.atomic + def crl_settings(self, request: Request, pk: str | None = None) -> Response: + """Update CRL scheduling and validity using the UI's policy checks.""" + del pk + ca = self.get_object() + ca = CaModel.objects.select_for_update().get(pk=ca.pk) + serializer = CaCrlSettingsSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + values = {field: getattr(ca, field) for field in IssuingCaCrlCycleForm.Meta.fields} + values.update(serializer.validated_data) + form = validated_form(IssuingCaCrlCycleForm(values, instance=ca)) + with security_change(target=ca, actor=request.user): + form.save() + return Response(CaSerializer(ca).data) + + @extend_schema(request=CaTruststoreSerializer, responses=CaSerializer) + @action(detail=True, methods=['post'], url_path='truststore') + @transaction.atomic + def truststore(self, request: Request, pk: str | None = None) -> Response: + """Associate an upstream truststore, including RA certificate/chain extraction.""" + del pk + ca = self.get_object() + ca = CaModel.objects.select_for_update().get(pk=ca.pk) + serializer = CaTruststoreSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + form = validated_form(IssuingCaTruststoreAssociationForm(serializer.validated_data, instance=ca)) + with security_change(target=ca, actor=request.user): + form.save() + if ca.ca_type in (CaModel.CaTypeChoice.REMOTE_EST_RA, CaModel.CaTypeChoice.REMOTE_CMP_RA): + store = form.cleaned_data['trust_store'] + if store.intended_usage == TruststoreModel.IntendedUsage.ISSUING_CA_CHAIN: + orders = list(store.truststoreordermodel_set.order_by('order')) + if not orders: + msg = 'The CA chain truststore contains no certificates.' + raise ValidationError(msg) + ca.certificate = orders[0].certificate + if len(orders) > 1: + ca.parent_ca = CaModel.objects.filter(certificate=orders[1].certificate).first() + ca.chain_truststore = store + ca.save(update_fields=['certificate', 'parent_ca', 'chain_truststore']) + return Response(CaSerializer(ca).data) + + @extend_schema(request=None, responses=OpenApiTypes.OBJECT) + @action(detail=True, methods=['post'], url_path='generate-crl') + @transaction.atomic + def generate_crl(self, _request: Request, pk: int | None = None, **_kwargs: Any) -> Response: # type: ignore[override] + """Generate a CRL without truncating fractional validity hours.""" + del pk, _kwargs + ca = self.get_object() + ca = CaModel.objects.select_for_update().get(pk=ca.pk) + if not ca.is_active: + msg = 'Cannot generate a CRL for a deactivated CA.' + raise Conflict(msg) + if ca.issue_crl(crl_validity_hours=ca.crl_validity_hours): + return Response( + { + 'message': f'CRL generated successfully for Issuing CA "{ca.unique_name}".', + 'last_crl_issued_at': ca.last_crl_issued_at, + } + ) + msg = 'This CA cannot generate a CRL with the current configuration.' + raise ValidationError(msg) + + +@extend_schema(tags=['CRL']) +class CrlViewSet( + PkiApiErrorsMixin, + mixins.ListModelMixin, + mixins.RetrieveModelMixin, + mixins.CreateModelMixin, + mixins.DestroyModelMixin, + viewsets.GenericViewSet[CrlModel], +): + """Inspect, import, download, and delete CRLs, including historical and unassociated CRLs.""" + + queryset = CrlModel.objects.select_related('ca').order_by('-this_update', '-pk') + serializer_class = CrlSerializer + permission_classes = (CanManageIssuingCas,) + filter_backends = (DjangoFilterBackend, filters.OrderingFilter) + filterset_fields: ClassVar = ['ca', 'is_active', 'crl_number'] + ordering_fields: ClassVar = ['this_update', 'next_update', 'created_at', 'crl_number'] + + @extend_schema(request=CrlImportSerializer, responses={201: CrlSerializer}) + @transaction.atomic + def create(self, request: Request, *args: Any, **kwargs: Any) -> Response: + """Import a PEM/DER CRL and check its issuer/signature when a CA is supplied.""" + del args, kwargs + serializer = CrlImportSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + data = serializer.validated_data + content = data['crl_file'].read() + try: + crl = x509.load_pem_x509_crl(content) if b'-----BEGIN' in content else x509.load_der_x509_crl(content) + except ValueError as exc: + raise ValidationError({'crl_file': 'Unable to parse CRL. Expected PEM or DER.'}) from exc + ca = data.get('ca') + if ca is not None: + ca = CaModel.objects.select_for_update().get(pk=ca.pk) + model = CrlModel.create_from_pem( + ca=ca, + crl_pem=crl.public_bytes(serialization.Encoding.PEM).decode(), + set_active=data['set_active'] and ca is not None, + ) + return Response(CrlSerializer(model).data, status=status.HTTP_201_CREATED) + + @transaction.atomic + def perform_destroy(self, instance: CrlModel) -> None: + """Delete the selected CRL through the model's dependency checks.""" + instance.delete() + + @extend_schema(parameters=[EncodingSerializer], responses={(200, 'application/octet-stream'): OpenApiTypes.BINARY}) + @action(detail=True, methods=['get']) + def download(self, request: Request, pk: str | None = None) -> HttpResponse: + """Download a CRL by its own identifier, without ambiguous CA-ID fallback.""" + del pk + model = self.get_object() + query = EncodingSerializer(data=request.query_params) + query.is_valid(raise_exception=True) + encoding = query.validated_data['encoding'] + return download_response( + model.get_crl_as_crypto().public_bytes( + serialization.Encoding.PEM if encoding == 'pem' else serialization.Encoding.DER + ), + f'crl-{model.pk}.{encoding}', + 'application/x-pem-file' if encoding == 'pem' else 'application/pkix-crl', + ) + + +@extend_schema(tags=['CA Rollover']) +class CaRolloverViewSet( + PkiApiErrorsMixin, + mixins.ListModelMixin, + mixins.RetrieveModelMixin, + mixins.CreateModelMixin, + viewsets.GenericViewSet[CaRolloverModel], +): + """Plan CA rollovers and execute audited lifecycle transitions.""" + + queryset = CaRolloverModel.objects.select_related('old_issuing_ca', 'new_issuing_ca').order_by('-planned_at') + serializer_class = CaRolloverSerializer + permission_classes = (CanManageIssuingCas,) + filter_backends = (DjangoFilterBackend,) + filterset_fields: ClassVar = ['old_issuing_ca', 'new_issuing_ca', 'state'] + + @extend_schema(request=CaRolloverPlanSerializer, responses={201: CaRolloverSerializer}) + @transaction.atomic + def create(self, request: Request, *args: Any, **kwargs: Any) -> Response: + """Acquire replacement credentials and plan a rollover atomically.""" + del args, kwargs + serializer = CaRolloverPlanSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + data = dict(serializer.validated_data) + ca = data.pop('old_issuing_ca') + ca = CaModel.objects.select_for_update().get(pk=ca.pk) + if ca.ca_type not in ( + CaModel.CaTypeChoice.AUTOGEN, + CaModel.CaTypeChoice.LOCAL_UNPROTECTED, + CaModel.CaTypeChoice.LOCAL_PKCS11, + ): + msg = 'Only locally managed issuing CAs can be rolled over.' + raise ValidationError(msg) + # Check lifecycle constraints before form validation, since import forms persist credentials. + if CaRolloverService.get_active_rollover(ca) or CaRolloverService.has_completed_rollover(ca): + msg = 'This CA already has an active or completed rollover.' + raise Conflict(msg) + method = data.pop('method') + form_class = { + 'pkcs12': ImportCaRolloverForm, + 'separate_files': SeparateFilesCaRolloverForm, + 'est': GenerateKeypairEstRolloverForm, + 'cmp': GenerateKeypairCmpRolloverForm, + }[method] + if method in ('est', 'cmp'): + data['ca_type'] = ( + CaModel.CaTypeChoice.REMOTE_ISSUING_EST if method == 'est' else CaModel.CaTypeChoice.REMOTE_ISSUING_CMP + ) + files = { + key: data.pop(key) + for key in list(data) + if key.endswith('_file') or key in ('ca_certificate', 'ca_certificate_chain') + } + form = validated_form(form_class(data, MultiValueDict({key: [value] for key, value in files.items()}))) + strategy = ( + CaRolloverStrategyType.IMPORT_CA + if method in ('pkcs12', 'separate_files') + else CaRolloverStrategyType.GENERATE_KEYPAIR + ) + rollover = CaRolloverService.plan_rollover( + ca, strategy, form, initiated_by=request.user if request.user.is_authenticated else None + ) + self._audit(rollover, AuditLog.OperationType.CA_ROLLOVER_PLANNED) + return Response(CaRolloverSerializer(rollover).data, status=status.HTTP_201_CREATED) + + def _audit(self, rollover: CaRolloverModel, operation: str) -> None: + """Record the same lifecycle audit events as the Web UI.""" + AuditLog.create_entry( + operation_type=operation, + target=rollover, + target_display=f'CA Rollover: {rollover.old_issuing_ca} → {rollover.new_issuing_ca}', + actor=self.request.user if self.request.user.is_authenticated else None, + ) + + @transaction.atomic + def _transition(self, operation: str) -> Response: + """Lock and validate a rollover before applying a state transition.""" + instance = self.get_object() + rollover = CaRolloverModel.objects.select_for_update().get(pk=instance.pk) + try: + if operation == 'start': + CaRolloverService.execute_rollover(rollover) + elif operation == 'transition': + rollover.transition_to_transition() + elif operation == 'complete': + CaRolloverService.finalize_rollover(rollover) + else: + CaRolloverService.cancel_rollover(rollover) + except ValueError as exc: + msg = 'The CA rollover transition could not be completed in its current state.' + raise Conflict(msg) from exc + audit_operation = { + 'start': AuditLog.OperationType.CA_ROLLOVER_STARTED, + 'transition': AuditLog.OperationType.CA_ROLLOVER_TRANSITIONED, + 'complete': AuditLog.OperationType.CA_ROLLOVER_COMPLETED, + 'cancel': AuditLog.OperationType.CA_ROLLOVER_CANCELLED, + }[operation] + self._audit(rollover, audit_operation) + return Response(CaRolloverSerializer(rollover).data) + + @extend_schema(request=None, responses=CaRolloverSerializer) + @action(detail=True, methods=['post']) + def start(self, request: Request, pk: str | None = None) -> Response: + """Start a planned rollover.""" + del request, pk + return self._transition('start') + + @extend_schema(request=None, responses=CaRolloverSerializer) + @action(detail=True, methods=['post']) + def transition(self, request: Request, pk: str | None = None) -> Response: + """Move from preparation to transition.""" + del request, pk + return self._transition('transition') + + @extend_schema(request=None, responses=CaRolloverSerializer) + @action(detail=True, methods=['post']) + def complete(self, request: Request, pk: str | None = None) -> Response: + """Finalize the rollover and apply its domain/activation changes.""" + del request, pk + return self._transition('complete') + + @extend_schema(request=None, responses=CaRolloverSerializer) + @action(detail=True, methods=['post']) + def cancel(self, request: Request, pk: str | None = None) -> Response: + """Cancel an unfinished rollover.""" + del request, pk + return self._transition('cancel') diff --git a/trustpoint/pki/views/certificates.py b/trustpoint/pki/views/certificates.py index 8fc9728df..472f1e013 100644 --- a/trustpoint/pki/views/certificates.py +++ b/trustpoint/pki/views/certificates.py @@ -12,22 +12,33 @@ from django.http import Http404, HttpRequest, HttpResponse, HttpResponseRedirect from django.urls import reverse_lazy from django.utils import timezone +from django.utils.http import content_disposition_header from django.views.generic.base import RedirectView from django.views.generic.detail import DetailView from django.views.generic.list import ListView from django_filters.rest_framework import DjangoFilterBackend +from drf_spectacular.types import OpenApiTypes from drf_spectacular.utils import extend_schema, extend_schema_view from rest_framework import filters, status, viewsets +from rest_framework.decorators import action +from rest_framework.exceptions import NotFound, ValidationError from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from trustpoint_core.archiver import ArchiveFormat, Archiver from trustpoint_core.oid import NameOid -from trustpoint_core.serializer import CertificateFormat +from trustpoint_core.serializer import CertificateCollectionSerializer, CertificateFormat from pki.filters import CertificateFilter from pki.models import CertificateModel from pki.models.truststore import ActiveTrustpointTlsServerCredentialModel -from pki.serializer.certificate import CertificateSerializer +from pki.serializer.certificate import ( + CertificateArchiveSerializer, + CertificateDetailSerializer, + CertificateDownloadSerializer, + CertificateSerializer, + CertificateSummarySerializer, +) +from pki.services.certificate_download import certificate_chain, certificate_filename from pki.services.certificate_health import certificate_health_conditions from shared.exports import ExportColumn, ExportConfig, ExportMixin from trustpoint.page_context import ( @@ -41,6 +52,8 @@ if TYPE_CHECKING: from typing import ClassVar + from rest_framework.request import Request + class CertificatesRedirectView(RedirectView): """View that redirects to the index of the PKI Issuing CA application: Issuing CAs.""" @@ -460,7 +473,9 @@ class CertificateViewSet(viewsets.ModelViewSet[CertificateModel]): create, update, and delete. """ - queryset = CertificateModel.objects.all().order_by('-created_at') + queryset = CertificateModel.objects.select_related( + 'revoked_certificate', 'basic_constraints_extension', 'issuer_id' + ).order_by('-created_at') serializer_class = CertificateSerializer permission_classes = (IsAuthenticated,) filter_backends = (DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter) @@ -468,6 +483,81 @@ class CertificateViewSet(viewsets.ModelViewSet[CertificateModel]): search_fields: ClassVar = ['common_name', 'sha256_fingerprint'] ordering_fields: ClassVar = ['common_name', 'created_at'] + def get_serializer_class(self) -> type[CertificateSerializer]: + """Provide richer public reads while keeping the existing write contract.""" + if self.action == 'retrieve': + return CertificateDetailSerializer + if self.action == 'list': + return CertificateSummarySerializer + return CertificateSerializer + + def _download_certificates(self, request: Request, certificate: CertificateModel, + *, include_chain: bool) -> list[CertificateModel]: + """Check permissions for every certificate included in an export.""" + try: + certificates = [certificate, *certificate_chain(certificate)] if include_chain else [certificate] + except ValueError as exc: + msg = 'Unable to build the certificate issuer chain.' + raise ValidationError(msg) from exc + for item in certificates: + self.check_object_permissions(request, item) + return certificates + + @extend_schema( + parameters=[CertificateDownloadSerializer], + responses={(200, 'application/octet-stream'): OpenApiTypes.BINARY}, + ) + @action(detail=True, methods=['get'], url_path='download') + def download(self, request: Request, pk: str | None = None) -> HttpResponse: + """Download a certificate as PEM, DER, PKCS#7 PEM, or PKCS#7 DER.""" + del pk + certificate = self.get_object() + options = CertificateDownloadSerializer(data=request.query_params) + options.is_valid(raise_exception=True) + file_format = CertificateFormat(options.validated_data['file_format']) + certificates = self._download_certificates( + request, certificate, include_chain=options.validated_data['include_chain']) + if len(certificates) > 1: + if file_format == CertificateFormat.DER: + msg = 'DER contains one certificate. Use PEM or PKCS#7 for a chain, or download an archive.' + raise ValidationError({'file_format': msg}) + data = CertificateCollectionSerializer( + [item.get_certificate_serializer().as_crypto() for item in certificates]).as_format(file_format) + else: + data = certificate.get_certificate_serializer().as_format(file_format) + response = HttpResponse(data, content_type=file_format.mime_type) + filename = certificate_filename(certificate, file_format.file_extension) + response['Content-Disposition'] = content_disposition_header(as_attachment=True, filename=filename) + return response + + @extend_schema( + request=CertificateArchiveSerializer, + responses={(200, 'application/octet-stream'): OpenApiTypes.BINARY}, + ) + @action(detail=False, methods=['post'], url_path='download') + def download_multiple(self, request: Request) -> HttpResponse: + """Download selected certificates as individual files in a ZIP or TAR.GZ archive.""" + options = CertificateArchiveSerializer(data=request.data) + options.is_valid(raise_exception=True) + values = options.validated_data + ids = list(dict.fromkeys(values['certificate_ids'])) + certificates = {certificate.pk: certificate for certificate in self.get_queryset().filter(pk__in=ids)} + if len(certificates) != len(ids): + msg = 'One or more selected certificates do not exist.' + raise NotFound(msg) + selected: dict[int, CertificateModel] = {} + for pk in ids: + for certificate in self._download_certificates(request, certificates[pk], + include_chain=values['include_chain']): + selected.setdefault(certificate.pk, certificate) + file_format = CertificateFormat(values['file_format']) + archive_format = ArchiveFormat(values['archive_format']) + contents = {certificate_filename(item, file_format.file_extension): + item.get_certificate_serializer().as_format(file_format) for item in selected.values()} + response = HttpResponse(Archiver.archive(contents, archive_format), content_type=archive_format.mime_type) + response['Content-Disposition'] = f'attachment; filename="certificates{archive_format.file_extension}"' + return response + @extend_schema( summary='List certificates', description='Retrieve certificates from the database.', diff --git a/trustpoint/pki/views/issuing_cas.py b/trustpoint/pki/views/issuing_cas.py index da9aa35e7..3831fe758 100644 --- a/trustpoint/pki/views/issuing_cas.py +++ b/trustpoint/pki/views/issuing_cas.py @@ -7,7 +7,7 @@ import binascii from base64 import b64decode -from typing import TYPE_CHECKING, Any, ClassVar, NamedTuple, cast +from typing import TYPE_CHECKING, Any, ClassVar, cast from cryptography import x509 from cryptography.hazmat.backends import default_backend @@ -61,21 +61,19 @@ from pki.models import CaModel, CertificateModel, CredentialModel from pki.models.ca_rollover import CaRolloverModel, CaRolloverState from pki.models.cert_profile import CertificateProfileModel -from pki.models.credential import CertificateChainOrderModel, PrimaryCredentialCertificate -from pki.models.issued_credential import RemoteIssuedCredentialModel +from pki.models.credential import PrimaryCredentialCertificate from pki.models.truststore import TruststoreModel from pki.rollover.import_ca import ImportCaRolloverForm from pki.serializer.issuing_ca import IssuingCaImportSerializer, IssuingCaSerializer +from pki.services.ca_enrollment import CmpCaEnrollmentService, CmpContextParams, EstCaEnrollmentService from pki.services.ca_rollover import CaRolloverService from pki.util.cert_profile import ProfileValidationError from pki.util.certificate_request import build_ca_request_data from pki.views.mixins import CertificateProfileSelectionMixin -from request.clients import EstClient, EstClientError -from request.clients.cmp_client import CmpClient, CmpClientError -from request.message_builder.cmp import CmpMessageBuilder +from request.clients import EstClientError +from request.clients.cmp_client import CmpClientError from request.operation_processor.csr_build import ProfileAwareCsrBuilder -from request.operation_processor.csr_sign import EstDeviceCsrSignProcessor -from request.request_context import CmpCertificateRequestContext, EstCertificateRequestContext +from request.request_context import CmpCertificateRequestContext from shared.views import AuditedBulkDeleteView, FormErrorMessageMixin from trustpoint.logger import LoggerMixin from trustpoint.settings import UIConfig @@ -92,15 +90,6 @@ from rest_framework.request import Request -class CmpContextParams(NamedTuple): - """Parameters for creating a CMP certificate request context.""" - subject: x509.Name - public_key: rsa.RSAPublicKey | ec.EllipticCurvePublicKey | Any # Any for ML-DSA keys - private_key: rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey | Any # Any for ML-DSA keys - recipient_name: str - extensions: list[x509.Extension[x509.ExtensionType]] | None - sender_kid: int | None - class IssuingCaContextMixin(ContextDataMixin): """Mixin which adds context_data for the PKI -> Issuing CAs pages.""" @@ -1071,98 +1060,20 @@ def get_not_implemented_message(self) -> str: return _('EST certificate request not yet implemented.') def _perform_est_enrollment(self, ca: CaModel, cert_content_data: dict[str, Any], request: HttpRequest) -> None: - """Perform the EST enrollment process.""" + """Enroll through the shared service and clear the browser draft on success.""" cert_profile_pk = request.session.get(f'cert_profile_pk_{ca.pk}') if not cert_profile_pk: msg = 'Certificate profile not found in session' raise ValueError(msg) cert_profile = CertificateProfileModel.objects.get(pk=cert_profile_pk) - - request_data = self._build_request_data_from_form(cert_content_data) - self.logger.info('Built request data: %s', request_data) - - context = EstCertificateRequestContext( - operation='simpleenroll', - protocol='est', - domain=None, - cert_profile_str=cert_profile.unique_name, - certificate_profile_model=cert_profile, - allow_ca_certificate_request=True, # Allow CA cert requests for Issuing CA enrollment - est_server_host=ca.remote_host, - est_server_port=ca.remote_port, - est_server_path=ca.remote_path, - est_username=ca.est_username, - est_password=( - ca.no_onboarding_config.est_password - if ca.no_onboarding_config else None - ), - est_server_truststore=( - ca.no_onboarding_config.trust_store - if ca.no_onboarding_config else None - ), - ) - - context.request_data = request_data - context.owner_credential = ca.credential - - csr_builder = ProfileAwareCsrBuilder() - csr_builder.process_operation(context) - csr = csr_builder.get_csr() - - context.cert_requested = csr - - # Use EstDeviceCsrSignProcessor to re-sign the CSR with the proper signature algorithm - csr_signer = EstDeviceCsrSignProcessor() - csr_signer.process_operation(context) - signed_csr = csr_signer.get_signed_csr() - - # Send the re-signed CSR to the remote EST server - est_client = EstClient(context) - issued_cert = est_client.simple_enroll(signed_csr) - - cert_pem = issued_cert.public_bytes(encoding=serialization.Encoding.PEM).decode('utf-8') - cert_obj = x509.load_pem_x509_certificate(cert_pem.encode()) - cert_model = CertificateModel.save_certificate(cert_obj) - - if ca.credential: - ca.credential.certificate = cert_model - ca.credential.save() - else: - credential = CredentialModel( - credential_type=CredentialModel.CredentialTypeChoice.ISSUING_CA, - certificate=cert_model, - private_key=ca.credential.private_key if ca.credential else '', - ) - credential.save() - credential.certificates.add(cert_model) - PrimaryCredentialCertificate.objects.create( - credential=credential, - certificate=cert_model, - is_primary=True - ) - ca.credential = credential - ca.save() - - cn_attrs = cert_obj.subject.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME) - common_name_value = cn_attrs[0].value if cn_attrs else '' - # Ensure common_name is a string - if isinstance(common_name_value, bytes): - common_name = common_name_value.decode('utf-8') - else: - common_name = str(common_name_value) - remote_issued_credential = RemoteIssuedCredentialModel( - common_name=common_name, - issued_credential_type=RemoteIssuedCredentialModel.RemoteIssuedCredentialType.LOCAL_CA, - issued_using_cert_profile=cert_profile.unique_name, - credential=ca.credential, - ca=ca, - ) - remote_issued_credential.full_clean() - remote_issued_credential.save() - + EstCaEnrollmentService().enroll(ca, cert_content_data, cert_profile) del request.session[f'cert_content_data_{ca.pk}'] del request.session[f'cert_profile_pk_{ca.pk}'] + def _build_request_data_from_form(self, cert_content_data: dict[str, Any]) -> dict[str, Any]: + """Delegate to the shared enrollment implementation.""" + return EstCaEnrollmentService()._build_request_data_from_form(cert_content_data) # noqa: SLF001 + def post(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse: """Handle POST request to request certificate via EST.""" if not request.user.has_perm(AppPermissions.MANAGE_CAS): @@ -1233,86 +1144,20 @@ def _perform_cmp_enrollment( *, sender_kid: int | None = None, ) -> None: - """Perform the CMP enrollment process.""" + """Enroll through the shared service and clear the browser draft on success.""" cert_profile_pk = request.session.get(f'cert_profile_pk_{ca.pk}') if not cert_profile_pk: msg = 'Certificate profile not found in session' raise ValueError(msg) cert_profile = CertificateProfileModel.objects.get(pk=cert_profile_pk) - - request_data = self._build_request_data_from_form(cert_content_data) - self.logger.info('Built request data: %s', request_data) - - context = CmpCertificateRequestContext( - operation='certification', - protocol='cmp', - domain=None, - cert_profile_str=cert_profile.unique_name, - certificate_profile_model=cert_profile, - allow_ca_certificate_request=True, - cmp_server_host=ca.remote_host, - cmp_server_port=ca.remote_port, - cmp_server_path=ca.remote_path, - cmp_shared_secret=( - ca.no_onboarding_config.cmp_shared_secret - if ca.no_onboarding_config else None - ), - cmp_server_truststore=( - ca.no_onboarding_config.trust_store - if ca.no_onboarding_config else None - ), - ) - - context.request_data = request_data - context.owner_credential = ca.credential - - csr_builder = ProfileAwareCsrBuilder() - csr_builder.process_operation(context) - csr = csr_builder.get_csr() - - if not csr.subject or len(csr.subject) == 0: - msg = 'CSR does not have a subject - cannot build CMP request' - raise ValueError(msg) - - self.logger.info('CSR subject: %s', csr.subject.rfc4514_string()) - - csr_subject = csr.subject - csr_public_key = csr.public_key() - csr_extensions = list(csr.extensions) if csr.extensions else None - - self.logger.info('CSR has %d extensions', len(csr.extensions) if csr.extensions else 0) - - recipient_name = self._get_recipient_name_from_truststore(ca) - - private_key = self._load_private_key(ca) - - cmp_params = CmpContextParams( - subject=csr_subject, - public_key=cast('rsa.RSAPublicKey | ec.EllipticCurvePublicKey | Any', csr_public_key), - private_key=private_key, - recipient_name=recipient_name, - extensions=csr_extensions, - sender_kid=sender_kid, - ) - - cmp_context = self._create_cmp_context(ca, cmp_params) - - cmp_builder = CmpMessageBuilder() - cmp_builder.build(cmp_context) - - pki_message = cmp_context.parsed_message - - cmp_client = CmpClient(cmp_context) - issued_cert, chain_certs = cmp_client.send_and_extract_certificate( - pki_message, - add_shared_secret_protection=True, - ) - - self._save_cmp_certificates(ca, issued_cert, chain_certs, cert_profile) - + CmpCaEnrollmentService().enroll(ca, cert_content_data, cert_profile, sender_kid=sender_kid) del request.session[f'cert_content_data_{ca.pk}'] del request.session[f'cert_profile_pk_{ca.pk}'] + def _build_request_data_from_form(self, cert_content_data: dict[str, Any]) -> dict[str, Any]: + """Delegate to the shared enrollment implementation.""" + return CmpCaEnrollmentService()._build_request_data_from_form(cert_content_data) # noqa: SLF001 + def post(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpResponse: """Handle POST request to request certificate via CMP.""" if not request.user.has_perm(AppPermissions.MANAGE_CAS): @@ -1368,74 +1213,20 @@ def post(self, request: HttpRequest, *_args: Any, **_kwargs: Any) -> HttpRespons return redirect(self.redirect_url_name, pk=ca.pk) def _get_recipient_name_from_truststore(self, ca: CaModel) -> str: - """Extract recipient name from the CA's truststore.""" - if not ca.no_onboarding_config or not ca.no_onboarding_config.trust_store: - msg = 'No truststore configured for CMP CA - recipient name cannot be determined' - raise ValueError(msg) - - try: - truststore_certs = ca.no_onboarding_config.trust_store.truststoreordermodel_set.order_by('order') - if not truststore_certs.exists(): - msg = 'Truststore contains no certificates - cannot determine recipient name' - raise ValueError(msg) - - first_cert_model = truststore_certs.first().certificate # type: ignore[union-attr] - ca_cert = first_cert_model.get_certificate_serializer().as_crypto() - recipient_name = ca_cert.subject.rfc4514_string() - self.logger.info('Using recipient name from truststore: %s', recipient_name) - except (AttributeError, IndexError) as e: - msg = f'Failed to extract recipient name from truststore: {e}' - raise ValueError(msg) from e - else: - return recipient_name + """Delegate to the shared enrollment implementation.""" + return CmpCaEnrollmentService()._get_recipient_name_from_truststore(ca) # noqa: SLF001 def _load_private_key(self, ca: CaModel) -> rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey | Any: - """Load the private key from the CA's credential.""" - if not ca.credential: - msg = 'No private key available for CA credential' - raise ValueError(msg) - - try: - return cast( - 'rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey | Any', - ca.credential.get_private_key(), - ) - except (RuntimeError, TypeError, ValueError) as exc: - msg = 'No usable private key is available for the CA credential.' - raise ValueError(msg) from exc + """Delegate to the shared enrollment implementation.""" + return CmpCaEnrollmentService()._load_private_key(ca) # noqa: SLF001 def _create_cmp_context( self, ca: CaModel, params: CmpContextParams, ) -> CmpCertificateRequestContext: - """Create the CMP certificate request context.""" - return CmpCertificateRequestContext( - protocol='cmp', - operation='certification', - cmp_server_host=ca.remote_host, - cmp_server_port=ca.remote_port, - cmp_server_path=ca.remote_path, - cmp_shared_secret=( - ca.no_onboarding_config.cmp_shared_secret - if ca.no_onboarding_config else None - ), - cmp_server_truststore=( - ca.no_onboarding_config.trust_store - if ca.no_onboarding_config else None - ), - request_data={ - 'subject': params.subject, - 'public_key': params.public_key, - 'private_key': params.private_key, - 'recipient_name': params.recipient_name, - 'extensions': params.extensions, - 'use_initialization_request': False, - 'add_pop': True, - 'prepare_shared_secret_protection': True, - 'sender_kid': params.sender_kid, - }, - ) + """Delegate to the shared enrollment implementation.""" + return CmpCaEnrollmentService()._create_cmp_context(ca, params) # noqa: SLF001 def _save_cmp_certificates( self, @@ -1444,143 +1235,31 @@ def _save_cmp_certificates( chain_certs: list[x509.Certificate], cert_profile: CertificateProfileModel, ) -> None: - """Save the issued certificate and chain certificates.""" - cert_model = CertificateModel.save_certificate(issued_cert) - - chain_ca_models = [] - for chain_cert in chain_certs: - existing_ca = self._find_existing_ca_for_certificate(chain_cert) - if existing_ca: - chain_ca_models.append(existing_ca) - else: - cn_attrs = chain_cert.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME) - cn = cn_attrs[0].value if cn_attrs else 'unknown' - keyless_ca = CaModel.create_keyless_ca( - unique_name=str(cn), - certificate_obj=chain_cert, - parent_ca=None, - ) - chain_ca_models.append(keyless_ca) - self.logger.info('Created keyless CA %s for chain certificate', keyless_ca.unique_name) - - self._build_ca_hierarchy(chain_ca_models) - - issuer_ca = self._find_issuer_ca(issued_cert, chain_ca_models) - if issuer_ca: - ca.parent_ca = issuer_ca - ca.save() - self.logger.info('Set parent CA %s for issuing CA %s', issuer_ca.unique_name, ca.unique_name) - - if ca.credential: - ca.credential.certificate = cert_model - ca.credential.save() - self._build_certificate_chain_for_credential(ca.credential, ca.parent_ca) - else: - credential = CredentialModel( - credential_type=CredentialModel.CredentialTypeChoice.ISSUING_CA, - certificate=cert_model, - private_key=ca.credential.private_key if ca.credential else '', - ) - credential.save() - credential.certificates.add(cert_model) - PrimaryCredentialCertificate.objects.create( - credential=credential, - certificate=cert_model, - is_primary=True - ) - ca.credential = credential - ca.save() - self._build_certificate_chain_for_credential(credential, ca.parent_ca) - - # Create a RemoteIssuedCredentialModel to track the issued CA credential - cn_attrs = issued_cert.subject.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME) - common_name_value = cn_attrs[0].value if cn_attrs else '' - # Ensure common_name is a string - if isinstance(common_name_value, bytes): - common_name = common_name_value.decode('utf-8') - else: - common_name = str(common_name_value) - remote_issued_credential = RemoteIssuedCredentialModel( - common_name=common_name, - issued_credential_type=RemoteIssuedCredentialModel.RemoteIssuedCredentialType.LOCAL_CA, - issued_using_cert_profile=cert_profile.unique_name, - credential=ca.credential, - ca=ca, - ) - remote_issued_credential.full_clean() - remote_issued_credential.save() + """Delegate to the shared enrollment implementation.""" + return CmpCaEnrollmentService()._save_cmp_certificates(ca, issued_cert, chain_certs, cert_profile) # noqa: SLF001 def _build_ca_hierarchy( self, chain_ca_models: list[CaModel], ) -> None: - """Build the CA hierarchy by setting parent-child relationships.""" - ca_by_subject = {} - for ca in chain_ca_models: - cert = ca.get_certificate() - if cert: - ca_by_subject[cert.subject.public_bytes().hex()] = ca - - for ca in chain_ca_models: - cert = ca.get_certificate() - if not cert: - continue - issuer_subject_bytes = cert.issuer.public_bytes().hex() - if issuer_subject_bytes in ca_by_subject: - parent_ca = ca_by_subject[issuer_subject_bytes] - if parent_ca not in (ca, ca.parent_ca): - ca.parent_ca = parent_ca - ca.save() - self.logger.info('Set parent CA %s for CA %s', parent_ca.unique_name, ca.unique_name) + """Delegate to the shared enrollment implementation.""" + return CmpCaEnrollmentService()._build_ca_hierarchy(chain_ca_models) # noqa: SLF001 def _find_issuer_ca( self, issued_cert: x509.Certificate, chain_ca_models: list[CaModel], ) -> CaModel | None: - """Find the CA that issued the certificate.""" - issuer_subject_bytes = issued_cert.issuer.public_bytes().hex() - for ca in chain_ca_models: - ca_cert = ca.get_certificate() - if ca_cert and ca_cert.subject.public_bytes().hex() == issuer_subject_bytes: - return ca - return None + """Delegate to the shared enrollment implementation.""" + return CmpCaEnrollmentService()._find_issuer_ca(issued_cert, chain_ca_models) # noqa: SLF001 def _build_certificate_chain_for_credential(self, credential: CredentialModel, parent_ca: CaModel | None) -> None: - """Build the certificate chain for the credential.""" - if not parent_ca or not credential.certificate: - return - chain = [] - current: CaModel | None = parent_ca - while current: - if current.certificate: - chain.append(current.certificate) - current = current.parent_ca - credential.certificatechainordermodel_set.all().delete() - primary_cert = credential.certificate - for i, cert in enumerate(chain): - CertificateChainOrderModel.objects.create( - credential=credential, certificate=cert, order=i, primary_certificate=primary_cert - ) + """Delegate to the shared enrollment implementation.""" + return CmpCaEnrollmentService()._build_certificate_chain_for_credential(credential, parent_ca) # noqa: SLF001 def _find_existing_ca_for_certificate(self, cert: x509.Certificate) -> CaModel | None: - """Find an existing CA that has the given certificate.""" - # TODO(FHK): subject+issuer byte equality is not sufficient for # noqa: FIX002 - - subject_bytes = cert.subject.public_bytes().hex().upper() - issuer_bytes = cert.issuer.public_bytes().hex().upper() - - keyless_match = CaModel.objects.filter( - certificate__subject_public_bytes=subject_bytes, - certificate__issuer_public_bytes=issuer_bytes, - ).first() - if keyless_match: - return keyless_match - - return CaModel.objects.filter( - credential__certificate__subject_public_bytes=subject_bytes, - credential__certificate__issuer_public_bytes=issuer_bytes, - ).first() + """Delegate to the shared enrollment implementation.""" + return CmpCaEnrollmentService()._find_existing_ca_for_certificate(cert) # noqa: SLF001 class KeylessCaConfigView(LoggerMixin, KeylessCaContextMixin, DetailView[CaModel]):