Enforce CSR proof of possession during certificate issuance

server/packages/kca-app/src/kca/service.py

@@ -672,6 +672,10 @@ async def issue_certificate(
     request_obj = await find_request(vault, payload.get("request_id"))
     if payload.get("request_id") and request_obj is None:
         raise ValidationError("signing request not found")
+    if request_obj is not None:
+        csr = req_core.parse_csr(bytes(request_obj.req_der))
+        if not req_core.verify_csr(csr):
+            raise ValidationError("signing request has an invalid signature")
 
     signer = await find_certificate(vault, payload.get("signer_id"))
     if payload.get("signer_id") and signer is None:

server/tests/test_routes_extra.py

@@ -91,6 +91,65 @@ class RouteExtraTests(unittest.TestCase):
                 response = await client.get(f"/api/vaults/{vault_id}/requests/{request['id']}")
                 assert response.json()["signed"] is True
 
+    @async_test
+    async def test_invalid_csr_signature_cannot_be_issued(self) -> None:
+        from cryptography import x509
+
+        from kca.core import keys as key_core
+        from kca.core import requests as req_core
+        from kca.models import Certificate
+
+        async with api_client() as client:
+            with oidc_user("alice"):
+                vault_id = await _vault(client)
+                ca_key = await _key(client, vault_id, "ca key")
+                ca = await _cert(client, vault_id, ca_key["id"], "Root CA", ca=True)
+                key = key_core.generate_key("rsa", 2048)
+                csr = req_core.build_csr(
+                    subject=x509.Name(
+                        [x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, "invalid.example.com")]
+                    ),
+                    private_key=key,
+                )
+                der = bytearray(req_core.csr_der(csr))
+                der[-1] ^= 1
+                invalid_der = bytes(der)
+                assert not req_core.verify_csr(req_core.parse_csr(invalid_der))
+
+                response = await client.post(
+                    f"/api/vaults/{vault_id}/import",
+                    json={
+                        "filename": "invalid.csr",
+                        "content_base64": base64.b64encode(invalid_der).decode(),
+                    },
+                )
+                assert response.status_code == 200, response.text
+                imported = response.json()["imported"]
+                assert len(imported) == 1
+                assert imported[0]["kind"] == "request"
+                request_id = imported[0]["id"]
+
+                response = await client.get(f"/api/vaults/{vault_id}/requests/{request_id}")
+                assert response.status_code == 200, response.text
+                assert response.json()["signature_valid"] is False
+
+                now = datetime.now(timezone.utc)
+                response = await client.post(
+                    f"/api/vaults/{vault_id}/certificates",
+                    json={
+                        "name": "invalid CSR cert",
+                        "request_id": request_id,
+                        "signer_id": ca["id"],
+                        "not_before": now.isoformat(),
+                        "not_after": (now + timedelta(days=30)).isoformat(),
+                    },
+                )
+                assert response.status_code == 400, response.text
+                assert response.json()["error"] == "signing request has an invalid signature"
+                assert await Certificate.filter(request_id=request_id).count() == 0
+                response = await client.get(f"/api/vaults/{vault_id}/requests/{request_id}")
+                assert response.json()["signed"] is False
+
     @async_test
     async def test_token_authentication(self) -> None:
         async with api_client() as client: