Agregar pruebas de integración para el administrador de certificados Infisical y configurar opciones de pytest

pyproject.toml

@@ -33,3 +33,8 @@ dev = [
     "ruff",
     "mypy",
 ]
+[tool.pytest.ini_options]
+addopts = "-m 'not integration'"
+markers = [
+    "integration: opt-in tests that contact a separately confirmed non-production service",
+]

tests/integration/infisical/README.md

@@ -0,0 +1,13 @@
+# Infisical Certificate Manager integration test
+
+This test is opt-in and makes real requests only when all required environment
+variables are present and `INFISICAL_TEST_CONFIRM_NON_PRODUCTION=true`.
+
+```bash
+pytest -m integration tests/integration/infisical
+```
+
+Use only a disposable Certificate Manager project and Machine Identity. The test
+generates ephemeral RSA test material in memory. It never prints credentials,
+PEM material, private keys, or Bearer tokens. Certificates created remotely may
+remain if the API does not provide an explicitly verified delete endpoint.

tests/integration/infisical/test_certificate_manager.py

@@ -0,0 +1,340 @@
+"""Opt-in contract test for a dedicated non-production Infisical project.
+
+It has no product imports and cannot make a request without an explicit, external
+non-production confirmation.
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+import httpx
+import pytest
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.x509.oid import NameOID
+
+pytestmark = pytest.mark.integration
+
+_REQUIRED = (
+    "INFISICAL_TEST_HOST",
+    "INFISICAL_TEST_CLIENT_ID",
+    "INFISICAL_TEST_CLIENT_SECRET",
+    "INFISICAL_TEST_PROJECT_ID",
+)
+_TIMEOUT = httpx.Timeout(connect=5.0, read=20.0, write=20.0, pool=5.0)
+
+
+@dataclass(frozen=True)
+class IntegrationConfig:
+    host: str
+    client_id: str
+    client_secret: str
+    project_id: str
+    application_id: str | None
+    profile_id: str | None
+    pki_collection_id: str | None
+
+
+def _config_or_skip() -> IntegrationConfig:
+    if os.getenv("INFISICAL_TEST_CONFIRM_NON_PRODUCTION") != "true":
+        pytest.skip("INFISICAL_TEST_CONFIRM_NON_PRODUCTION=true is required")
+    missing = [name for name in _REQUIRED if not os.getenv(name)]
+    if missing:
+        pytest.skip("missing non-production Infisical test configuration")
+    return IntegrationConfig(
+        host=os.environ["INFISICAL_TEST_HOST"].rstrip("/"),
+        client_id=os.environ["INFISICAL_TEST_CLIENT_ID"],
+        client_secret=os.environ["INFISICAL_TEST_CLIENT_SECRET"],
+        project_id=os.environ["INFISICAL_TEST_PROJECT_ID"],
+        application_id=os.getenv("INFISICAL_TEST_APPLICATION_ID"),
+        profile_id=os.getenv("INFISICAL_TEST_PROFILE_ID"),
+        pki_collection_id=os.getenv("INFISICAL_TEST_PKI_COLLECTION_ID"),
+    )
+
+
+@dataclass(frozen=True)
+class GeneratedMaterial:
+    leaf_pem: str
+    key_pem: str
+    chain_pem: str
+    leaf: x509.Certificate
+    intermediate: x509.Certificate
+    root: x509.Certificate
+
+
+def _name(common_name: str) -> x509.Name:
+    if len(common_name.encode("utf-8")) > 64:
+        raise ValueError(
+            f"Test certificate commonName exceeds X.509 64-byte limit: {common_name!r}"
+        )
+
+    return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)])
+
+
+def _issue_certificate(
+    *,
+    subject: x509.Name,
+    issuer: x509.Name,
+    public_key: Any,
+    issuer_key: Any,
+    is_ca: bool,
+    serial_number: int,
+) -> x509.Certificate:
+    now = datetime.now(UTC)
+    return (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(issuer)
+        .public_key(public_key)
+        .serial_number(serial_number)
+        .not_valid_before(now - timedelta(minutes=1))
+        .not_valid_after(now + timedelta(days=2))
+        .add_extension(x509.BasicConstraints(ca=is_ca, path_length=None), critical=True)
+        .add_extension(
+            x509.KeyUsage(
+                digital_signature=True,
+                content_commitment=False,
+                key_encipherment=not is_ca,
+                data_encipherment=False,
+                key_agreement=False,
+                key_cert_sign=is_ca,
+                crl_sign=is_ca,
+                encipher_only=None,
+                decipher_only=None,
+            ),
+            critical=True,
+        )
+        .sign(issuer_key, hashes.SHA256())
+    )
+
+
+def _generate_material() -> GeneratedMaterial:
+    marker = uuid.uuid4().hex[:12]
+    root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    root_subject = _name(f"Infisical integration TEST root {marker}")
+    root = _issue_certificate(
+        subject=root_subject,
+        issuer=root_subject,
+        public_key=root_key.public_key(),
+        issuer_key=root_key,
+        is_ca=True,
+        serial_number=x509.random_serial_number(),
+    )
+    intermediate_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    intermediate_subject = _name(f"Infisical integration TEST intermediate {marker}")
+    intermediate = _issue_certificate(
+        subject=intermediate_subject,
+        issuer=root.subject,
+        public_key=intermediate_key.public_key(),
+        issuer_key=root_key,
+        is_ca=True,
+        serial_number=x509.random_serial_number(),
+    )
+    leaf_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    leaf = _issue_certificate(
+        subject=_name(f"Infisical integration TEST leaf {marker}"),
+        issuer=intermediate.subject,
+        public_key=leaf_key.public_key(),
+        issuer_key=intermediate_key,
+        is_ca=False,
+        serial_number=x509.random_serial_number(),
+    )
+    return GeneratedMaterial(
+        leaf_pem=leaf.public_bytes(serialization.Encoding.PEM).decode(),
+        key_pem=leaf_key.private_bytes(
+            serialization.Encoding.PEM,
+            serialization.PrivateFormat.PKCS8,
+            serialization.NoEncryption(),
+        ).decode(),
+        # The Certificate Manager import contract describes chainPem as intermediate
+        # certificates; the root signs the intermediate but is not sent as chain material.
+        chain_pem=intermediate.public_bytes(serialization.Encoding.PEM).decode(),
+        leaf=leaf,
+        intermediate=intermediate,
+        root=root,
+    )
+
+
+def _fingerprint(certificate: x509.Certificate) -> str:
+    return certificate.fingerprint(hashes.SHA256()).hex()
+
+
+def _certificates_from_pem(bundle: str) -> list[x509.Certificate]:
+    marker = "-----END CERTIFICATE-----"
+    return [
+        x509.load_pem_x509_certificate((part + marker).encode())
+        for part in bundle.split(marker)
+        if "-----BEGIN CERTIFICATE-----" in part
+    ]
+
+
+def _safe_shape(response: httpx.Response) -> str:
+    try:
+        body = response.json()
+    except ValueError:
+        return "non-json"
+    return (
+        "object:" + ",".join(sorted(body.keys()))
+        if isinstance(body, dict)
+        else type(body).__name__
+    )
+
+
+def _payload(
+    config: IntegrationConfig, material: GeneratedMaterial, *, chain: bool
+) -> dict[str, str]:
+    payload = {
+        "certificatePem": material.leaf_pem,
+        "privateKeyPem": material.key_pem,
+        "friendlyName": f"integration-test-{uuid.uuid4().hex}",
+    }
+    if chain:
+        payload["chainPem"] = material.chain_pem
+    if config.application_id:
+        payload["applicationId"] = config.application_id
+    if config.profile_id:
+        payload["profileId"] = config.profile_id
+    if config.pki_collection_id:
+        payload["pkiCollectionId"] = config.pki_collection_id
+    return payload
+
+
+def _login(client: httpx.Client, config: IntegrationConfig) -> tuple[str, int | None]:
+    response = client.post(
+        "/api/v1/auth/universal-auth/login",
+        json={"clientId": config.client_id, "clientSecret": config.client_secret},
+    )
+    assert response.status_code == 200
+    body = response.json()
+    token = body.get("accessToken")
+    assert isinstance(token, str) and token
+    ttl = body.get("expiresIn")
+    return token, ttl if isinstance(ttl, int) else None
+
+
+def _import(
+    client: httpx.Client, payload: dict[str, str]
+) -> tuple[httpx.Response, str | None]:
+    response = client.post(
+        "/api/v1/cert-manager/certificates/import-certificate", json=payload
+    )
+    certificate_id = (
+        response.json().get("certificateId") if response.status_code < 300 else None
+    )
+    return response, certificate_id
+
+
+def test_certificate_manager_rest_contract() -> None:
+    config = _config_or_skip()
+    material = _generate_material()
+    with httpx.Client(
+        base_url=config.host, verify=True, timeout=_TIMEOUT
+    ) as unauthenticated:
+        token, ttl = _login(unauthenticated, config)
+        assert ttl is None or ttl > 0
+        invalid_token = unauthenticated.get(
+            "/api/v1/cert-manager/certificates/00000000-0000-0000-0000-000000000000",
+            headers={"Authorization": "Bearer invalid"},
+        )
+        assert invalid_token.status_code in {401, 403}
+
+    headers = {"Authorization": f"Bearer {token}"}
+    with httpx.Client(
+        base_url=config.host, headers=headers, verify=True, timeout=_TIMEOUT
+    ) as client:
+        minimal = _payload(config, material, chain=False)
+        imported, certificate_id = _import(client, minimal)
+
+        import_body = imported.json() if imported.headers.get("content-type", "").startswith("application/json") else {}
+        import_error = import_body.get("error")
+        import_message = str(import_body.get("message", ""))[:300]
+        assert imported.status_code in {200, 201}, (
+            f"import failed: status={imported.status_code} error={import_error!r} "
+            f"message={import_message!r}"
+        )
+        assert isinstance(certificate_id, str) and certificate_id
+
+        metadata = client.get(f"/api/v1/cert-manager/certificates/{certificate_id}")
+        certificate = client.get(
+            f"/api/v1/cert-manager/certificates/{certificate_id}/certificate"
+        )
+        private_key = client.get(
+            f"/api/v1/cert-manager/certificates/{certificate_id}/private-key"
+        )
+        assert (
+            metadata.status_code
+            == certificate.status_code
+            == private_key.status_code
+            == 200
+        )
+
+        metadata_body = metadata.json().get("certificate", {})
+        observed_serial = metadata_body.get("serialNumber")
+        assert isinstance(observed_serial, str)
+        # Infisical returns a byte-aligned hexadecimal serial, which may preserve
+        # a leading zero. Compare the certificate serial numerically.
+        assert int(observed_serial, 16) == material.leaf.serial_number
+        returned_leaf = x509.load_pem_x509_certificate(
+            certificate.json()["certificate"].encode()
+        )
+        assert _fingerprint(returned_leaf) == _fingerprint(material.leaf)
+        returned_key = serialization.load_pem_private_key(
+            private_key.content,
+            password=None,
+        )
+        assert (
+            returned_key.public_key().public_numbers()
+            == returned_leaf.public_key().public_numbers()
+        )
+
+        duplicate, duplicate_id = _import(client, minimal)
+        assert duplicate.status_code in {200, 201, 400, 409, 422}
+        if duplicate.status_code < 300:
+            assert isinstance(duplicate_id, str) and duplicate_id
+
+        chain_material = _generate_material()
+        chained, chained_id = _import(
+            client, _payload(config, chain_material, chain=True)
+        )
+        chain_body = (
+            chained.json()
+            if chained.headers.get("content-type", "").startswith("application/json")
+            else {}
+        )
+        chain_error = chain_body.get("error")
+        chain_message = str(chain_body.get("message", ""))[:300]
+        assert chained.status_code in {200, 201}, (
+            f"chain import failed: status={chained.status_code} error={chain_error!r} message={chain_message!r} "
+            f"shape={_safe_shape(chained)}"
+        )
+        assert isinstance(chained_id, str) and chained_id
+        chain_response = client.get(
+            f"/api/v1/cert-manager/certificates/{chained_id}/certificate"
+        )
+        assert chain_response.status_code == 200
+        returned_chain = _certificates_from_pem(
+            chain_response.json().get("certificateChain") or ""
+        )
+        assert returned_chain
+        assert [_fingerprint(item) for item in returned_chain] == [
+            _fingerprint(chain_material.intermediate)
+        ]
+        assert _fingerprint(chain_material.root) not in {
+            _fingerprint(item) for item in returned_chain
+        }
+
+        missing = client.get("/api/v1/cert-manager/certificates/00000000-0000-0000-0000-000000000000")
+        assert missing.status_code in {400, 404}
+        malformed = client.post(
+            "/api/v1/cert-manager/certificates/import-certificate",
+            json={"certificatePem": "invalid", "privateKeyPem": "invalid"},
+        )
+        assert malformed.status_code in {400, 422}
+        # Keep only a stable, non-sensitive shape available in assertion output.
+        assert _safe_shape(missing)