G2P-5209 Partner public keys stored in DB.

Author: pjoshi751

core/mapper-partner-api/src/openg2p_spar_mapper_partner_api/app.py

@@ -14,8 +14,8 @@
     MapperService,
     RequestValidation,
 )
-from openg2p_spar_models.crypto import PyJWTCryptoHelper
-from openg2p_spar_models.models import IdFaMapping, Strategy
+from openg2p_spar_models.crypto import PyJWTCryptoHelper, seed_partner_certs
+from openg2p_spar_models.models import IdFaMapping, PartnerKey, Strategy
 
 from .controllers import MapperController
 
@@ -33,7 +33,6 @@ def initialize(self, **kwargs):
         ResponseHelper()
         JWTValidationHelper()
         PyJWTCryptoHelper(
-            partner_keys_dir=_config.partner_keys_dir,
             allowed_algorithms=[
                 alg.strip()
                 for alg in _config.signature_allowed_algorithms.split(",")
@@ -50,5 +49,9 @@ async def migrate():
             _logger.info("Migrating database")
             await IdFaMapping.create_migrate()
             await Strategy.create_migrate()
+            await PartnerKey.create_migrate()
+            # Seed-based partner onboarding: upsert configured partner certs into
+            # the partner_keys table (idempotent).
+            await seed_partner_certs(_config.partner_certs)
 
         asyncio.run(migrate())

core/mapper-partner-api/src/openg2p_spar_mapper_partner_api/config.py

@@ -28,11 +28,16 @@ class Settings(BaseSettings):
     # Inbound partner JWS signature verification (local, Keymanager-free).
     # SPAR_MAPPER_PARTNER_API_JWT_AUTH_ENABLED — when false, signature validation
     # is skipped entirely. When true, each request must carry a detached JWS in the
-    # "Signature" header, verified against the partner's public key loaded from
-    # partner_keys_dir (one JWKS file per partner, named PARTNER_<MNEMONIC>.json).
+    # "Signature" header, verified against the partner's public certificate held in
+    # the partner_keys table (DB-backed; see openg2p_spar_models.crypto).
     # The G2P Bridge calls SPAR as sender_app_mnemonic=g2p_bridge -> PARTNER_G2P_BRIDGE.
     jwt_auth_enabled: bool = False
-    partner_keys_dir: str = "/etc/spar/partner-keys"
     # Comma-separated allowed JWS algorithms. RS256 only (asymmetric); "none" and
     # HMAC (HS*) are always rejected regardless of this list.
     signature_allowed_algorithms: str = "RS256"
+    # Seed-based partner onboarding: a JSON list of partner certs upserted into the
+    # partner_keys table at migrate-time. Each item:
+    #   {"reference_id": "PARTNER_<MNEMONIC>", "public_key": "<PEM cert>",
+    #    "kid": "<optional>", "algorithm": "RS256"}
+    # (Runtime admin-API onboarding is a planned follow-up — see docs.)
+    partner_certs: list[dict] = []

core/models/src/openg2p_spar_models/crypto/__init__.py

@@ -5,10 +5,12 @@
 )
 from .key_store import PartnerKeyStore
 from .pyjwt_crypto_helper import PyJWTCryptoHelper
+from .seed import seed_partner_certs
 
 __all__ = [
     "PyJWTCryptoHelper",
     "PartnerKeyStore",
+    "seed_partner_certs",
     "DEFAULT_ALLOWED_ALGORITHMS",
     "DEFAULT_SIGNING_ALGORITHM",
     "is_forbidden_algorithm",

core/models/src/openg2p_spar_models/crypto/key_store.py

@@ -1,68 +1,111 @@
-"""Filesystem-backed store of partner public keys for inbound JWS verification.
-
-This replaces the remote Keymanager certificate lookup. Each partner gets one
-JWKS file named ``<reference_id>.json`` inside ``keys_dir`` — e.g.
-``PARTNER_MY_PSP.json`` for reference id ``PARTNER_MY_PSP`` (the value
-``JWTValidationHelper`` derives from ``sender_app_mnemonic``). Each file is a
-standard JWKS (``{"keys": [ ... ]}``) holding one or more public keys, each with
-a ``kid`` and ``alg``.
-
-Multiple keys per partner make rotation an overlap operation: register the new
-key, let the partner switch signing to it, then drop the old key — no coordinated
-cutover, because lookup is by ``kid``. The directory is normally mounted from a
-Kubernetes Secret/ConfigMap; files are cached and reloaded automatically when
-their mtime changes.
+"""DB-backed store of partner public keys for inbound JWS verification.
+
+Partner public certificates live in the ``partner_keys`` table (see
+``..models.partner_key.PartnerKey``). On each inbound verify we look up the
+active, currently-valid certs for a ``reference_id`` (``PARTNER_<MNEMONIC>``) and
+return them as lightweight dicts (``kid`` / ``algorithm`` / ``public_key_pem``) so
+the crypto helper stays library-agnostic.
+
+Results are cached per ``reference_id`` for a short TTL so a high request rate does
+not hit the database on every signature check; rotation/revocation propagate within
+one TTL window. A ``session_maker`` may be injected (tests); otherwise one is built
+lazily from the process-wide async engine.
 """
 
-import json
 import logging
-import os
 import threading
+import time
+from datetime import datetime
+
+from sqlalchemy import select
 
 _logger = logging.getLogger("openg2p_spar_models.crypto.key_store")
 
+DEFAULT_CACHE_TTL_SECONDS = 300
+
 
 class PartnerKeyStore:
-    def __init__(self, keys_dir):
-        self.keys_dir = keys_dir
-        self._cache = {}  # reference_id -> (mtime, list[jwk dict])
+    def __init__(
+        self,
+        session_maker=None,
+        cache_ttl_seconds=DEFAULT_CACHE_TTL_SECONDS,
+        clock=time.monotonic,
+    ):
+        self._session_maker = session_maker
+        self._ttl = cache_ttl_seconds
+        self._clock = clock
+        self._cache = {}  # reference_id -> (expiry_monotonic, list[dict])
         self._lock = threading.Lock()
 
-    def _path_for(self, reference_id):
-        # reference_id is an opaque token like PARTNER_MY_PSP. Reject anything
-        # that could escape keys_dir.
-        if (
-            not reference_id
-            or "/" in reference_id
-            or os.sep in reference_id
-            or ".." in reference_id
-        ):
-            return None
-        return os.path.join(self.keys_dir, f"{reference_id}.json")
+    def _maker(self):
+        if self._session_maker is not None:
+            return self._session_maker
+        # Build lazily from the process-wide engine so this works before the app
+        # has finished wiring the DB at import time.
+        from openg2p_fastapi_common.context import dbengine
+        from sqlalchemy.ext.asyncio import async_sessionmaker
 
-    def get_keys(self, reference_id):
-        """Return the partner's JWK dicts (a JWKS ``keys`` list), or None if unknown."""
-        if not self.keys_dir:
-            return None
-        path = self._path_for(reference_id)
-        if not path or not os.path.isfile(path):
-            _logger.warning("No partner key file for reference id '%s'", reference_id)
+        return async_sessionmaker(dbengine.get(), expire_on_commit=False)
+
+    async def get_keys(self, reference_id):
+        """Return active partner key dicts for ``reference_id``, or None if unknown.
+
+        Each dict: ``{"kid": str|None, "algorithm": str, "public_key_pem": str}``.
+        """
+        if not reference_id:
             return None
-        mtime = os.path.getmtime(path)
+        now = self._clock()
         with self._lock:
             cached = self._cache.get(reference_id)
-            if cached and cached[0] == mtime:
-                return cached[1]
+            if cached and cached[0] > now:
+                return cached[1] or None
+
         try:
-            with open(path, encoding="utf-8") as handle:
-                jwks = json.load(handle)
-            keys = jwks.get("keys") if isinstance(jwks, dict) else None
-            if not isinstance(keys, list) or not keys:
-                _logger.error("Partner JWKS for '%s' has no 'keys' array", reference_id)
-                return None
+            from ..models.partner_key import PartnerKey
+
+            maker = self._maker()
+            async with maker() as session:
+                result = await session.execute(
+                    select(PartnerKey).where(
+                        PartnerKey.reference_id == reference_id,
+                        PartnerKey.status == "active",
+                    )
+                )
+                rows = result.scalars().all()
         except Exception:
-            _logger.exception("Failed to load partner JWKS for '%s'", reference_id)
+            _logger.exception("Failed to load partner keys for '%s'", reference_id)
             return None
+
+        wall_now = datetime.now()
+        keys = []
+        for row in rows:
+            if row.valid_from and row.valid_from > wall_now:
+                continue
+            if row.valid_to and row.valid_to < wall_now:
+                continue
+            if not row.public_key:
+                continue
+            keys.append(
+                {
+                    "kid": row.kid,
+                    "algorithm": row.algorithm or "RS256",
+                    "public_key_pem": row.public_key,
+                }
+            )
+
         with self._lock:
-            self._cache[reference_id] = (mtime, keys)
+            self._cache[reference_id] = (now + self._ttl, keys)
+        if not keys:
+            _logger.warning(
+                "No active partner keys for reference id '%s'", reference_id
+            )
+            return None
         return keys
+
+    def invalidate(self, reference_id=None):
+        """Drop cached keys (all, or one reference_id) — e.g. after onboarding."""
+        with self._lock:
+            if reference_id is None:
+                self._cache.clear()
+            else:
+                self._cache.pop(reference_id, None)

core/models/src/openg2p_spar_models/crypto/pyjwt_crypto_helper.py

@@ -8,15 +8,22 @@
 This is the SPAR-local instance of the shared ``PyJWTCryptoHelper`` design
 (see the OpenG2P platform docs). It is kept here for now; the strategic plan is to
 move it into ``openg2p-fastapi-common`` ``partner_auth`` so every product shares
-one implementation — at which point SPAR swaps the one-line registration in
-``app.py``.
+one implementation — at which point the product swaps the one-line registration in
+``app.py``. See ``docs/local-signature-verification.md``.
 
 * Inbound ``verify_jwt`` — verifies a detached JWS (``header..signature``) sent by
-  a partner in the ``Signature`` header, against the partner's public key looked up
-  by ``km_ref_id`` (e.g. ``PARTNER_<MNEMONIC>``). The G2P Bridge calls SPAR with
-  ``sender_app_mnemonic = g2p_bridge`` → key ``PARTNER_G2P_BRIDGE``.
-* Outbound ``create_jwt_token`` — signs a payload with the service's own private
-  key and returns a detached JWS.
+  a partner in the ``Signature`` header, against the partner's public certificate
+  looked up by ``km_ref_id`` (e.g. ``PARTNER_<MNEMONIC>``) in the DB-backed
+  ``PartnerKeyStore``.
+* Outbound ``create_jwt_token`` — signs a payload with this service's own private
+  key, loaded from a password-protected **PKCS#12 (.p12)** keystore, and returns a
+  detached JWS for downstream services (e.g. SPAR).
+
+Key material:
+* Private signing key — a PKCS#12 (.p12) file (private key + self-signed cert),
+  unlocked with a password. This is the standard secure on-disk format; the .p12 is
+  mounted from a Secret and its password supplied from a Secret.
+* Partner public keys — X.509 certificates (PEM) held in the ``partner_keys`` table.
 
 Signing input matches the established wire contract:
 ``base64url(protected_header)`` + ``.`` + ``base64url(canonical_json(body))`` where
@@ -28,7 +35,10 @@
 import logging
 
 import orjson
-from jwt import PyJWK, PyJWS
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.serialization import load_pem_public_key, pkcs12
+from cryptography.x509 import load_pem_x509_certificate
+from jwt import PyJWS
 from openg2p_fastapi_common.utils.crypto import CryptoHelper
 
 from .constants import (
@@ -45,8 +55,9 @@ class PyJWTCryptoHelper(CryptoHelper):
     def __init__(
         self,
         *,
-        partner_keys_dir=None,
+        partner_key_store=None,
         signing_key_path=None,
+        signing_key_password=None,
         signing_key_kid=None,
         signing_algorithm=DEFAULT_SIGNING_ALGORITHM,
         allowed_algorithms=None,
@@ -57,14 +68,17 @@ def __init__(
         self.allowed_algorithms = tuple(
             allowed_algorithms or DEFAULT_ALLOWED_ALGORITHMS
         )
+        # DB-backed partner public-cert store; a default (engine-derived) store is
+        # used when none is injected.
         self._partner_key_store = (
-            PartnerKeyStore(partner_keys_dir) if partner_keys_dir else None
+            partner_key_store if partner_key_store is not None else PartnerKeyStore()
         )
         self._signing_key_path = signing_key_path
+        self._signing_key_password = signing_key_password
         self._signing_key_kid = signing_key_kid
         self._signing_algorithm = signing_algorithm
         self._signing_key = None  # lazy-loaded cryptography private key
-        self._signing_kid = None  # kid read from the signing JWK on load
+        self._signing_kid = None  # kid derived from the signing cert on load
         self._jws = PyJWS()
 
     async def aclose(self):
@@ -99,7 +113,7 @@ async def verify_jwt(
         if self._partner_key_store is None:
             _logger.error("Partner key store not configured; cannot verify signature")
             return False
-        keys = self._partner_key_store.get_keys(km_ref_id)
+        keys = await self._partner_key_store.get_keys(km_ref_id)
         if not keys:
             _logger.error("No registered keys for partner '%s'", km_ref_id)
             return False
@@ -121,9 +135,9 @@ async def verify_jwt(
         else:
             verifiable = f"{part1}.{self._b64u(self._canonical(payload))}.{part3}"
 
-        for jwk in candidates:
+        for entry in candidates:
             try:
-                key = PyJWK.from_dict(jwk).key
+                key = self._load_public_key(entry["public_key_pem"])
                 self._jws.decode(verifiable, key, algorithms=[alg])
             except Exception:
                 continue
@@ -164,24 +178,41 @@ def _load_signing_key(self):
             return self._signing_key
         if not self._signing_key_path:
             raise ValueError("Signing key path not configured; cannot create JWS")
-        with open(self._signing_key_path, encoding="utf-8") as handle:
-            jwk = orjson.loads(handle.read())
-        self._signing_kid = jwk.get("kid")
-        self._signing_key = PyJWK.from_dict(jwk).key
+        with open(self._signing_key_path, "rb") as handle:
+            data = handle.read()
+        password = (
+            self._signing_key_password.encode() if self._signing_key_password else None
+        )
+        private_key, cert, _extra = pkcs12.load_key_and_certificates(data, password)
+        if private_key is None:
+            raise ValueError("PKCS#12 keystore contains no private key")
+        # kid: prefer an explicitly configured one, else the cert's SHA-256
+        # thumbprint (so the verifier can correlate it to the onboarded cert).
+        if cert is not None:
+            self._signing_kid = self._b64u(cert.fingerprint(hashes.SHA256()))
+        self._signing_key = private_key
         return self._signing_key
 
+    @staticmethod
+    def _load_public_key(pem):
+        """Build a public key from a PEM cert (preferred) or bare public-key PEM."""
+        pem_bytes = pem.encode() if isinstance(pem, str) else pem
+        if b"BEGIN CERTIFICATE" in pem_bytes:
+            return load_pem_x509_certificate(pem_bytes).public_key()
+        return load_pem_public_key(pem_bytes)
+
     def _candidate_keys(self, keys, header, alg):
         """Keys eligible to verify this header: matching kid (if present) and a
         registered alg that is consistent with the header alg."""
         kid = header.get("kid")
         candidates = []
-        for jwk in keys:
-            if kid and jwk.get("kid") and jwk.get("kid") != kid:
+        for entry in keys:
+            if kid and entry.get("kid") and entry.get("kid") != kid:
                 continue
-            key_alg = jwk.get("alg")
+            key_alg = entry.get("algorithm")
             if key_alg and key_alg != alg:
                 continue
-            candidates.append(jwk)
+            candidates.append(entry)
         return candidates
 
     def _is_algorithm_allowed(self, alg):