Fix issuer links for PEM certificate chain imports

server/tests/test_routes_extra.py

@@ -320,6 +320,99 @@ class RouteExtraTests(unittest.TestCase):
                 assert response.status_code == 200
                 assert len(response.json()["skipped"]) == 1
 
+    @async_test
+    async def test_pem_chain_import_restores_issuer_links(self) -> None:
+        from cryptography import x509
+        from cryptography.x509.oid import NameOID
+
+        from kca.core import certificates as cert_core
+        from kca.core import extensions as ext_core
+        from kca.core import keys as key_core
+        from kca.models import Certificate
+
+        def name(cn: str) -> x509.Name:
+            return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
+
+        now = datetime.now(timezone.utc)
+        root_key = key_core.generate_key("rsa", 2048)
+        intermediate_key = key_core.generate_key("rsa", 2048)
+        leaf_key = key_core.generate_key("rsa", 2048)
+        ca_spec = {"basic_constraints": {"critical": True, "ca": True}}
+        root = cert_core.build_self_signed(
+            subject=name("Chain Root"),
+            private_key=root_key,
+            serial=101,
+            not_before=now,
+            not_after=now + timedelta(days=3650),
+            extension_spec=ca_spec,
+            signature_hash="sha256",
+        )
+        intermediate = cert_core.build_certificate(
+            subject=name("Chain Intermediate"),
+            public_key=key_core.public_part(intermediate_key),
+            issuer_cert=root,
+            issuer_private_key=root_key,
+            serial=102,
+            not_before=now,
+            not_after=now + timedelta(days=1825),
+            extensions=ext_core.build_extensions(
+                ca_spec,
+                public_key=key_core.public_part(intermediate_key),
+                subject_name=name("Chain Intermediate"),
+                issuer_cert=root,
+            ),
+            signature_hash="sha256",
+        )
+        leaf = cert_core.build_certificate(
+            subject=name("Chain Leaf"),
+            public_key=key_core.public_part(leaf_key),
+            issuer_cert=intermediate,
+            issuer_private_key=intermediate_key,
+            serial=103,
+            not_before=now,
+            not_after=now + timedelta(days=365),
+            extensions=[],
+            signature_hash="sha256",
+        )
+        # Deliberately not in issuer-first order.
+        pem = cert_core.certificate_pem(leaf) + cert_core.certificate_pem(root) + cert_core.certificate_pem(intermediate)
+
+        async with api_client() as client:
+            with oidc_user("alice"):
+                vault_id = await _vault(client)
+                response = await client.post(
+                    f"/api/vaults/{vault_id}/import", json={"filename": "chain.pem", "pem": pem}
+                )
+                assert response.status_code == 200, response.text
+                result = response.json()
+                assert len(result["imported"]) == 3, result
+                assert result["skipped"] == [], result
+
+                async def certificates_by_name() -> dict[str, dict]:
+                    response = await client.get(f"/api/vaults/{vault_id}/certificates")
+                    assert response.status_code == 200, response.text
+                    return {cert["name"]: cert for cert in response.json()["results"]}
+
+                certs = await certificates_by_name()
+                root_row = certs["Chain Root"]
+                intermediate_row = certs["Chain Intermediate"]
+                leaf_row = certs["Chain Leaf"]
+                assert root_row["issuer_id"] is None
+                assert intermediate_row["issuer_id"] == root_row["id"]
+                assert leaf_row["issuer_id"] == intermediate_row["id"]
+                assert all(cert["issuer_id"] != cert["id"] for cert in certs.values())
+
+                # Re-importing duplicates also repairs self-links left by older imports.
+                model = await Certificate.get(id=leaf_row["id"])
+                model.issuer_id = model.id
+                await model.save(update_fields=["issuer_id"])
+                response = await client.post(
+                    f"/api/vaults/{vault_id}/import", json={"filename": "chain.pem", "pem": pem}
+                )
+                assert response.status_code == 200, response.text
+                certs = await certificates_by_name()
+                assert certs["Chain Leaf"]["issuer_id"] == intermediate_row["id"]
+
     @async_test
     async def test_settings_and_templates(self) -> None:
         async with api_client() as client: