Merge pull request #179 from minvws/cert-check

Checking on the test endpoint for mtls

Author: jaytaph

app/container.py

@@ -27,7 +27,7 @@ def container_config(binder: inject.Binder) -> None:
     org_service = OrgService(db)
     binder.bind(OrgService, org_service)
 
-    mtls_service = MtlsService(config.app.mtls_override_cert)
+    mtls_service = MtlsService(config.app.mtls_override_cert, org_service)
     binder.bind(MtlsService, mtls_service)
 
     try:

app/routers/exchange.py

@@ -3,13 +3,15 @@
 from typing import Dict, Any
 
 from fastapi import APIRouter, Depends, HTTPException
+from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
 
 from app import container
 from app.models.requests import ExchangeRequest, RidExchangeRequest, RidReceiveRequest
 from app.personal_id import PersonalId
-from app.rid import ALLOWED_BY_RID_USAGE, REQUIRED_MIN_USAGE, USAGE_RANK
+from app.rid import ALLOWED_BY_RID_USAGE, REQUIRED_MIN_USAGE, USAGE_RANK, RidUsage
 from app.services.key_resolver import KeyResolver
+from app.services.mtls_service import MtlsService
 from app.services.oprf.jwe_token import BlindJwe
 from app.services.org_service import OrgService
 from app.services.pseudonym_service import PseudonymService, PseudonymType
@@ -181,9 +183,11 @@ def exchange_rid(
 @router.post("/exchange/pseudonym", summary="Exchange pseudonym")
 def exchange_pseudonym(
     req: ExchangeRequest,
+    request: Request,
     key_resolver: KeyResolver = Depends(container.get_key_resolver),
     pseudonym_service: PseudonymService = Depends(container.get_pseudonym_service),
     org_service: OrgService = Depends(container.get_org_service),
+    mtls_service: MtlsService = Depends(container.get_mtls_service),
 ) -> Response:
     if not req.recipientOrganization.startswith("ura:"):
         raise InvalidURA(req.recipientOrganization)
@@ -201,6 +205,13 @@ def exchange_pseudonym(
         )
         subject = "pseudonym:irreversible:" + res
     elif req.pseudonymType == PseudonymType.Reversible:
+        source_org = mtls_service.get_org_from_request(request)
+        if source_org.max_rid_usage == RidUsage.IrreversiblePseudonym:
+            raise HTTPException(
+                status_code=400,
+                detail="Source organization is not allowed to exchange reversible pseudonyms.",
+            )
+
         res = pseudonym_service.exchange_reversible_pseudonym(
             personal_id=req.personalId,
             recipient_organization=recipient_organization,

app/routers/key.py

@@ -20,23 +20,10 @@ def post_key(
     request: Request,
     mtls_service: MtlsService = Depends(container.get_mtls_service),
     key_resolver: KeyResolver = Depends(container.get_key_resolver),
-    org_service: OrgService = Depends(container.get_org_service),
 ) -> JSONResponse:
 
-    # Fetch public key from the client certificate
     mtls_pub_key = mtls_service.get_mtls_pub_key(request)
-
-    # Extract URA from the client certificate and validate S-type
-    data = mtls_service.get_mtls_uzi_data(request)
-    if data["CardType"] != "S":
-        raise HTTPException(status_code=401, detail="Invalid client certificate. Need an UZI S-type certificate.")
-
-    ura = data["SubscriberNumber"]
-
-    # Make sure we have (pre)registered the organization for this URa
-    org = org_service.get_by_ura(ura)
-    if org is None:
-        raise HTTPException(status_code=404, detail="organization for this URA is not registered")
+    org = mtls_service.get_org_from_request(request)
 
     # Create the key entry
     try:

app/routers/test_oprf.py

@@ -4,11 +4,14 @@
 from fastapi import APIRouter, Depends
 from fastapi.encoders import jsonable_encoder
 from jwcrypto import jwe, jwk
+from starlette.requests import Request
 from starlette.responses import JSONResponse
 
 from app import container
 from app.models.requests import InputRequest, ReceiverRequest, JweReceiverRequest
 from app.personal_id import PersonalId, PersonalIdJSONEncoder
+from app.rid import RidUsage
+from app.services.mtls_service import MtlsService
 from app.services.oprf.oprf_service import OprfService
 from app.services.pseudonym_service import PseudonymService
 
@@ -147,9 +150,18 @@ def post_test_jwe_decode(
 
 @router.post("/test/pseudonym/reversible", summary="Reverse a pseudonym", tags=["test-oprf"])
 def post_test_reversible_pseudonym(
+    request: Request,
     pseudonym: str,
     pseudonym_service: PseudonymService = Depends(container.get_pseudonym_service),
+    mtls_service: MtlsService = Depends(container.get_mtls_service),
 ) -> JSONResponse:
+    # Check if we as an organization are allowed to reverse pseudonyms (max_key_usage == BSN)
+    org = mtls_service.get_org_from_request(request)
+    if org.max_rid_usage != RidUsage.Bsn:
+        return JSONResponse({
+            "error": "Organization is not authorized to reverse pseudonyms."
+        }, status_code=403)
+
     parts = pseudonym.split(":")
     if len(parts) == 3 and parts[0] == "pseudonym" and parts[1] == "reversible":
         pseudonym = parts[2]
@@ -165,10 +177,10 @@ def post_test_reversible_pseudonym(
             "error": f"Failed to reverse pseudonym: {str(e)}"
         }, status_code=400)
 
-
     return JSONResponse(content=jsonable_encoder({
         "pseudonym": pseudonym,
         "decoded": decoded,
     }, custom_encoder={PersonalId: lambda v: json.loads(json.dumps(v, cls=PersonalIdJSONEncoder))}))
 
 
+

app/services/mtls_service.py

@@ -7,15 +7,23 @@
 from uzireader.uziserver import UziServer
 import logging
 
+from app.db.entities.organization import Organization
+from app.services.org_service import OrgService
+
 logger = logging.getLogger(__name__)
 
 class MtlsService:
     _CERT_START = "-----BEGIN CERTIFICATE-----"
     _CERT_END = "-----END CERTIFICATE-----"
     _SSL_CLIENT_CERT_HEADER_NAME = "x-proxy-ssl_client_cert"
 
-    def __init__(self, override_cert: str|None) -> None:
+    def __init__(
+        self,
+        override_cert: str|None,
+        org_service: OrgService,
+    ) -> None:
         self.__cert: bytes | None = None
+        self.org_service = org_service
 
         if override_cert is not None and override_cert != "":
             with open(override_cert, "r") as f:
@@ -71,3 +79,22 @@ def get_mtls_uzi_data(self, request: Request) -> UziServer:
         cert_bytes = self.get_mtls_cert(request)
         formatted_cert = self._enforce_cert_newlines(cert_bytes)
         return UziServer(verify="SUCCESS", cert=formatted_cert)
+
+
+    def get_org_from_request(self, request: Request) -> Organization:
+        """
+        Extract the organization from the client certificate in the request
+        """
+
+        data = self.get_mtls_uzi_data(request)
+        if data["CardType"] != "S":
+            raise HTTPException(status_code=401,
+                                detail="Invalid client certificate. Need an UZI S-type certificate.")
+
+        ura = data["SubscriberNumber"]
+        org = self.org_service.get_by_ura(ura)
+        if org is None:
+            raise HTTPException(status_code=404, detail=f"organization for URA {ura} is not registered")
+
+        return org
+