factuges-document-signing-service/tests/infrastructure/pdf/test_pyhanko_pdf_signer.py

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))