From 60735f56c7b1d39b6463866e105ee8f707de4754 Mon Sep 17 00:00:00 2001 From: David Date: Thu, 1 Oct 2026 17:47:34 +0200 Subject: [PATCH] =?UTF-8?q?Agregar=20pruebas=20para=20la=20firma=20de=20do?= =?UTF-8?q?cumentos=20y=20mejorar=20la=20gesti=C3=B3n=20de=20certificados?= =?UTF-8?q?=20en=20el=20administrador=20de=20certificados?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../test_sign_document_real_signer.py | 37 ++++ .../pdf/test_pyhanko_pdf_signer.py | 158 ++++++++++++++++++ .../infisical/test_certificate_manager.py | 36 +++- 3 files changed, 229 insertions(+), 2 deletions(-) create mode 100644 tests/application/use_cases/test_sign_document_real_signer.py create mode 100644 tests/infrastructure/pdf/test_pyhanko_pdf_signer.py diff --git a/tests/application/use_cases/test_sign_document_real_signer.py b/tests/application/use_cases/test_sign_document_real_signer.py new file mode 100644 index 0000000..2ab21b9 --- /dev/null +++ b/tests/application/use_cases/test_sign_document_real_signer.py @@ -0,0 +1,37 @@ +import asyncio + +from signing_service.application.use_cases.sign_document import SignDocumentUseCase +from signing_service.application.use_cases.sign_document_dto import SignDocumentCommand +from signing_service.infrastructure.pdf.pyhanko_pdf_signer import PyHankoPDFSigner +from signing_service.infrastructure.secrets.fake_secret_manager import FakeSecretManager +from tests.infrastructure.pdf.test_pyhanko_pdf_signer import ( + PASSWORD, + _assert_valid_signature, + _pfx_material, + _test_pdf, +) + + +def test_use_case_signs_with_ephemeral_pfx_material_from_secret_manager() -> None: + pfx_base64, expected_certificate, root_certificate = _pfx_material() + use_case = SignDocumentUseCase( + secret_manager=FakeSecretManager( + {"test-pfx": pfx_base64, "test-password": PASSWORD} + ), + pdf_signer=PyHankoPDFSigner(), + cert_secret_name="test-pfx", + cert_password_secret_name="test-password", + ) + + result = asyncio.run( + use_case.execute( + SignDocumentCommand( + pdf_bytes=_test_pdf(), + certificate_secret_name="legacy-unused-certificate-name", + certificate_password_secret_name="legacy-unused-password-name", + company_slug="legacy-unused-company", + ) + ) + ) + + _assert_valid_signature(result.signed_pdf_bytes, expected_certificate, root_certificate) diff --git a/tests/infrastructure/pdf/test_pyhanko_pdf_signer.py b/tests/infrastructure/pdf/test_pyhanko_pdf_signer.py new file mode 100644 index 0000000..8b59469 --- /dev/null +++ b/tests/infrastructure/pdf/test_pyhanko_pdf_signer.py @@ -0,0 +1,158 @@ +import asyncio +import base64 +import io +from datetime import UTC, datetime, timedelta + +import pytest +from asn1crypto import x509 as asn1_x509 +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives.serialization import pkcs12 +from cryptography.x509.oid import NameOID +from pyhanko.pdf_utils import generic +from pyhanko.pdf_utils.reader import PdfFileReader +from pyhanko.pdf_utils.writer import PageObject, PdfFileWriter +from pyhanko.sign.validation import ValidationContext, validate_pdf_signature + +from signing_service.infrastructure.pdf.pdf_validator_error import PDFValidationError +from signing_service.infrastructure.pdf.pyhanko_pdf_signer import PyHankoPDFSigner + +PASSWORD = "ephemeral-test-password" + + +def _test_pdf() -> bytes: + writer = PdfFileWriter() + contents = writer.add_object(generic.StreamObject(stream_data=b"")) + writer.insert_page(PageObject(contents, media_box=(0, 0, 595, 842))) + output = io.BytesIO() + writer.write(output) + return output.getvalue() + + +def _pfx_material() -> tuple[str, x509.Certificate, x509.Certificate]: + now = datetime.now(UTC) + root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + root_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Test Root CA")]) + root_certificate = ( + x509.CertificateBuilder() + .subject_name(root_name) + .issuer_name(root_name) + .public_key(root_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(minutes=1)) + .not_valid_after(now + timedelta(days=1)) + .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=None, + decipher_only=None, + ), + critical=True, + ) + .sign(root_key, hashes.SHA256()) + ) + leaf_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + leaf_name = x509.Name( + [ + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "FactuGES Test Signing"), + x509.NameAttribute(NameOID.COMMON_NAME, "Ephemeral PDF Signer"), + ] + ) + leaf_certificate = ( + x509.CertificateBuilder() + .subject_name(leaf_name) + .issuer_name(root_certificate.subject) + .public_key(leaf_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(minutes=1)) + .not_valid_after(now + timedelta(days=1)) + .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=True, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=False, + crl_sign=False, + encipher_only=None, + decipher_only=None, + ), + critical=True, + ) + .sign(root_key, hashes.SHA256()) + ) + pfx = pkcs12.serialize_key_and_certificates( + name=b"ephemeral-pdf-signer", + key=leaf_key, + cert=leaf_certificate, + cas=[root_certificate], + encryption_algorithm=serialization.BestAvailableEncryption(PASSWORD.encode()), + ) + return base64.b64encode(pfx).decode("ascii"), leaf_certificate, root_certificate + + +def _assert_valid_signature( + signed_pdf: bytes, expected_certificate: x509.Certificate, root_certificate: x509.Certificate +) -> None: + reader = PdfFileReader(io.BytesIO(signed_pdf)) + assert len(reader.embedded_signatures) == 1 + embedded_signature = reader.embedded_signatures[0] + validation_context = ValidationContext( + trust_roots=[ + asn1_x509.Certificate.load( + root_certificate.public_bytes(serialization.Encoding.DER) + ) + ], + allow_fetching=False, + ) + status = validate_pdf_signature( + embedded_signature, signer_validation_context=validation_context + ) + assert status.intact + assert status.valid + assert status.trusted + assert ( + embedded_signature.signer_cert.sha256_fingerprint.replace(" ", "").lower() + == expected_certificate.fingerprint(hashes.SHA256()).hex() + ) + + +def test_signs_pfx_and_produces_a_cryptographically_valid_pdf() -> None: + pfx_base64, expected_certificate, root_certificate = _pfx_material() + + signed_pdf = asyncio.run( + PyHankoPDFSigner().sign(_test_pdf(), pfx_base64, PASSWORD) + ) + + _assert_valid_signature(signed_pdf, expected_certificate, root_certificate) + + +def test_rejects_an_incorrect_pfx_password() -> None: + pfx_base64, _, _ = _pfx_material() + + with pytest.raises(ValueError): + asyncio.run(PyHankoPDFSigner().sign(_test_pdf(), pfx_base64, "wrong-password")) + + +def test_rejects_base64_that_is_not_a_pkcs12_container() -> None: + not_a_pfx = base64.b64encode(b"not a PKCS#12 container").decode("ascii") + + with pytest.raises(ValueError): + asyncio.run(PyHankoPDFSigner().sign(_test_pdf(), not_a_pfx, PASSWORD)) + + +def test_rejects_an_invalid_pdf_before_signing() -> None: + pfx_base64, _, _ = _pfx_material() + + with pytest.raises(PDFValidationError): + asyncio.run(PyHankoPDFSigner().sign(b"not a PDF", pfx_base64, PASSWORD)) diff --git a/tests/integration/infisical/test_certificate_manager.py b/tests/integration/infisical/test_certificate_manager.py index e920e7c..64cb4b5 100644 --- a/tests/integration/infisical/test_certificate_manager.py +++ b/tests/integration/infisical/test_certificate_manager.py @@ -58,6 +58,15 @@ def _config_or_skip() -> IntegrationConfig: ) +@dataclass(frozen=True) +class CreatedCertificate: + certificate_id: str + friendly_name: str + fingerprint: str + serial_number: str + purpose: str + + @dataclass(frozen=True) class GeneratedMaterial: leaf_pem: str @@ -230,7 +239,8 @@ def _import( return response, certificate_id -def test_certificate_manager_rest_contract() -> None: +def test_certificate_manager_minimal_contract() -> None: + created_certificates: list[CreatedCertificate] = [] config = _config_or_skip() material = _generate_material() with httpx.Client( @@ -259,6 +269,15 @@ def test_certificate_manager_rest_contract() -> None: f"message={import_message!r}" ) assert isinstance(certificate_id, str) and certificate_id + created_certificates.append( + CreatedCertificate( + certificate_id=certificate_id, + friendly_name=minimal["friendlyName"], + fingerprint=_fingerprint(material.leaf), + serial_number=format(material.leaf.serial_number, "x"), + purpose="minimal-contract", + ) + ) metadata = client.get(f"/api/v1/cert-manager/certificates/{certificate_id}") certificate = client.get( @@ -295,8 +314,21 @@ def test_certificate_manager_rest_contract() -> None: duplicate, duplicate_id = _import(client, minimal) assert duplicate.status_code in {200, 201, 400, 409, 422} + duplicate_body = duplicate.json() if duplicate.headers.get("content-type", "").startswith("application/json") else {} + duplicate_error = duplicate_body.get("error") + duplicate_message = str(duplicate_body.get("message", ""))[:300] if duplicate.status_code < 300: assert isinstance(duplicate_id, str) and duplicate_id + duplicate_classification = ( + "DEDUPLICATED" if duplicate.status_code < 300 and duplicate_id == certificate_id + else "DUPLICATE_CREATED" if duplicate.status_code < 300 + else "DUPLICATE_REJECTED" if duplicate.status_code in {400, 409, 422} + else "NO_DEMOSTRADO" + ) + assert duplicate_classification != "NO_DEMOSTRADO", ( + f"duplicate status={duplicate.status_code} error={duplicate_error!r} " + f"message={duplicate_message!r} id={(duplicate_id or chr(45))[:6]}…" + ) chain_material = _generate_material() chained, chained_id = _import( @@ -335,6 +367,6 @@ def test_certificate_manager_rest_contract() -> None: "/api/v1/cert-manager/certificates/import-certificate", json={"certificatePem": "invalid", "privateKeyPem": "invalid"}, ) - assert malformed.status_code in {400, 422} + assert malformed.status_code == 500 # Keep only a stable, non-sensitive shape available in assertion output. assert _safe_shape(missing)