Merge pull request #31 from Indicio-tech/feat/cert-trust-registry-auto-add

feat: auto-register certs as trust anchors and return PEMs by default

Author: burdettadam

oid4vc/mso_mdoc/key_generation.py

@@ -461,6 +461,18 @@ async def generate_default_keys_and_certs(
         session, "default_certificate", {"cert_id": cert_id}
     )
 
+    # Auto-register the certificate as a trust anchor so verifiers
+    # immediately have the issuer cert in their trust store.
+    anchor_ids = await storage_manager.auto_register_trust_anchors(
+        session, cert_pem, source="key-generation"
+    )
+    if anchor_ids:
+        LOGGER.info(
+            "Auto-registered %d trust anchor(s) from generated certificate: %s",
+            len(anchor_ids),
+            anchor_ids,
+        )
+
     LOGGER.info("Generated default mDoc key: %s and certificate: %s", key_id, cert_id)
 
     return {

oid4vc/mso_mdoc/key_routes.py

@@ -119,13 +119,15 @@ async def list_certificates(request: web.BaseRequest):
     """List all stored mDoc certificates.
 
     Query parameters:
-        include_pem: If "true", include the certificate_pem field in results
+        exclude_pem: If "true", omit the certificate_pem field from results.
+            PEMs are included by default so callers can use cert data for
+            trust-registry automation without extra round-trips.
     """
     context: AdminRequestContext = request["context"]
     storage_manager = MdocStorageManager(context.profile)
 
-    # Check for include_pem query parameter
-    include_pem = request.query.get("include_pem", "").lower() == "true"
+    # PEMs are now included by default; pass exclude_pem=true to omit them.
+    include_pem = request.query.get("exclude_pem", "").lower() != "true"
 
     try:
         async with context.profile.session() as session:

oid4vc/mso_mdoc/storage/__init__.py

@@ -19,7 +19,9 @@
 """
 
 from datetime import UTC, datetime
+import hashlib
 import logging
+import re
 from typing import Any, Dict, List, Optional, Tuple
 
 from acapy_agent.core.profile import Profile, ProfileSession
@@ -231,10 +233,15 @@ async def store_certificate(
         key_id: str,
         metadata: Optional[Dict] = None,
     ) -> None:
-        """Store a PEM certificate."""
+        """Store a PEM certificate and auto-register it as a trust anchor."""
         await certificates.store_certificate(
             session, cert_id, certificate_pem, key_id, metadata
         )
+        # Automatically register every cert in the PEM chain as a trust anchor
+        # so the trust registry stays in sync without manual steps.
+        await self.auto_register_trust_anchors(
+            session, certificate_pem, source="cert-store"
+        )
 
     async def get_certificate(
         self, session: ProfileSession, cert_id: str
@@ -364,3 +371,57 @@ async def get_all_trust_anchor_pems(self, session: ProfileSession) -> List[str]:
     async def delete_trust_anchor(self, session: ProfileSession, anchor_id: str) -> bool:
         """Delete a trust anchor by ID."""
         return await trust_anchors.delete_trust_anchor(session, anchor_id)
+
+    async def auto_register_trust_anchors(
+        self,
+        session: ProfileSession,
+        certificate_pem: str,
+        source: str = "auto",
+    ) -> List[str]:
+        """Parse a PEM (possibly a chain) and register each cert as a trust anchor.
+
+        Duplicate certificates (by PEM content) are silently skipped so this
+        method is idempotent.
+
+        Args:
+            session: Active database session
+            certificate_pem: One or more concatenated PEM certificate blocks
+            source: Label for metadata (e.g. ``"key-generation"``)
+
+        Returns:
+            List of anchor IDs that were newly created.
+        """
+        _PEM_RE = re.compile(
+            r"-----BEGIN CERTIFICATE-----[A-Za-z0-9+/=\s]+?"
+            r"-----END CERTIFICATE-----\n?",
+            re.DOTALL,
+        )
+        individual_pems = _PEM_RE.findall(certificate_pem)
+        if not individual_pems:
+            return []
+
+        # Fetch existing trust anchor PEMs to avoid duplicates
+        existing_pems = set(await trust_anchors.get_all_trust_anchor_pems(session))
+
+        created: List[str] = []
+        for pem in individual_pems:
+            normalized = pem.strip()
+            if normalized in existing_pems:
+                continue
+            # Deterministic-ish anchor ID based on cert fingerprint
+            fingerprint = hashlib.sha256(normalized.encode()).hexdigest()[:12]
+            anchor_id = f"auto-{source}-{fingerprint}"
+            try:
+                await trust_anchors.store_trust_anchor(
+                    session,
+                    anchor_id=anchor_id,
+                    certificate_pem=normalized,
+                    metadata={"source": source, "auto_registered": True},
+                )
+                created.append(anchor_id)
+                existing_pems.add(normalized)
+            except Exception:
+                LOGGER.debug(
+                    "Skipping duplicate trust anchor %s", anchor_id, exc_info=True
+                )
+        return created