Use one OprfPseudonym model for all pseudonym input (#615)

* Use OprfPseudonym as the single pseudonym input type

- Remove PseudonymToken and BlindedPseudonym, the resolver and crypto client take an OprfPseudonym
- Decode FHIR pseudonym tokens in the FHIR service layer and name the field in the error
- Treat a malformed token as input validation: 422 and a rejection log, no failure event

* OprfPseudonym reads the API's pseudonym and oprf_key names through
validation aliases. That makes RegistrationQueryParams,
CreateRegistrationRequest and LocalizeRequest redundant.

* more DRY

* sonar issue fixed

* Decode FHIR pseudonym tokens in LocalizationListService

* Accept only the API field names for OprfPseudonym

* refactor: validate FHIR pseudonyms in model

---------

Co-authored-by: Rick Lambrechts <rick@rl-webdiensten.nl>

Author: thomas-samoht

app/debug/crypto_service_api_client_mock.py

@@ -1,14 +1,14 @@
-from app.models.pseudonym import PseudonymResponse
+from app.models.pseudonym import OprfPseudonym, PseudonymResponse
 from app.services.crypto_service_api_client import CryptoServiceApiClient
 
 
 class CryptoServiceApiClientMock(CryptoServiceApiClient):
     def __init__(self) -> None:
         """Mock implementation of CryptoServiceApiClient for dev/testing purposes."""
 
-    def exchange(self, jwe: str, blind_factor: str, label: str, mechanism: str) -> PseudonymResponse:
+    def exchange(self, pseudonym: OprfPseudonym, label: str, mechanism: str) -> PseudonymResponse:
         return PseudonymResponse(
-            encrypted_pseudonym=f"{jwe}:{blind_factor}",
+            encrypted_pseudonym=f"{pseudonym.jwe}:{pseudonym.blind_factor}",
             iv="abcdefghijklmnop",
         )
 

app/errors/handlers.py

@@ -50,7 +50,7 @@ def handle_mapped_exception(request: Request, exc: Exception) -> JSONResponse:
         fhir_error = FHIRError(severity=spec.severity, code=spec.fhir_code, msg=str(exc))
         return JSONResponse(
             status_code=status_code,
-            content=fhir_error.outcome.model_dump(exclude_none=True),
+            content=fhir_error.outcome.model_dump(by_alias=True, exclude_none=True),
             headers=fhir_error.headers,
         )
 

app/errors/mapping.py

@@ -31,7 +31,7 @@ class ErrorSpec(BaseModel):
     InvalidKeyInfoError: ErrorSpec(severity="error", fhir_code="transient", http_status=503),
     InvalidModelError: ErrorSpec(severity="error", fhir_code="structure", http_status=400),
     InvalidHeaderPropertyError: ErrorSpec(severity="error", fhir_code="invalid", http_status=401),
-    PseudonymError: ErrorSpec(severity="error", fhir_code="invalid", http_status=422, log=True),
+    PseudonymError: ErrorSpec(severity="error", fhir_code="invalid", http_status=422),
     ValueError: ErrorSpec(severity="error", fhir_code="invalid", http_status=400, log=True),
 }
 

app/models/fhir/resources/localization_list/request.py

@@ -1,15 +1,17 @@
 import logging
-from typing import Any
+from typing import Annotated, Any
 
 from fastapi import Query
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel, ValidateAs, field_validator
 
 from app.models.fhir.elements import Identifier
 from app.models.fhir.resources.data import (
     DEVICE_SYSTEM,
     PSEUDONYM_SYSTEM,
 )
-from app.models.pseudonym import PseudonymToken
+from app.models.pseudonym import OprfPseudonym
+from app.services.exceptions import PseudonymError
+from app.services.fhir.pseudonym_token import decode_pseudonym_token
 
 logger = logging.getLogger(__name__)
 
@@ -42,10 +44,17 @@ def _create_openapi_examples(system: str) -> dict[str, Any]:
     }
 
 
+def _subject_from_query(value: str) -> OprfPseudonym:
+    try:
+        return decode_pseudonym_token(value, SUBJECT_IDENTIFIER_PARAM)
+    except PseudonymError as e:
+        raise ValueError(str(e)) from e
+
+
 class LocalizationListParams(BaseModel):
-    model_config = {"extra": "forbid"}
+    model_config = {"extra": "forbid", "hide_input_in_errors": True}
 
-    subject: str | None = Query(
+    subject: Annotated[OprfPseudonym, ValidateAs(str, _subject_from_query)] | None = Query(
         alias=SUBJECT_IDENTIFIER_PARAM,
         openapi_examples=_create_openapi_examples(PSEUDONYM_SYSTEM),
         default=None,
@@ -91,9 +100,6 @@ def empty(self) -> bool:
     def is_localize_params(self) -> bool:
         return self.subject is not None and self.source is None
 
-    def subject_token(self) -> PseudonymToken | None:
-        return PseudonymToken(self.subject) if self.subject else None
-
 
 class LocalizationListSearchParams(LocalizationListParams):
     """Search parameters for a List search, which also controls the page that is returned."""

app/models/pseudonym.py

@@ -1,30 +1,19 @@
 from dataclasses import dataclass
 from typing import Any, Self
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class PseudonymResponse(BaseModel):
     encrypted_pseudonym: str
     iv: str
 
 
-@dataclass(frozen=True)
-class OprfPseudonym:
-    """Used in the REST routers"""
+class OprfPseudonym(BaseModel):
+    model_config = ConfigDict(frozen=True)
 
-    jwe: str
-    blind_factor: str
-
-
-@dataclass(frozen=True)
-class PseudonymToken:
-    """Used in the FHIR routers"""
-
-    value: str
-
-
-BlindedPseudonym = OprfPseudonym | PseudonymToken
+    jwe: str = Field(alias="pseudonym")
+    blind_factor: str = Field(alias="oprf_key")
 
 
 @dataclass

app/models/registrations.py

@@ -5,26 +5,9 @@
 from pydantic import BaseModel, field_validator
 
 from app.db.models.referral import ReferralEntity
-from app.models.pseudonym import OprfPseudonym
 from app.models.ura import UraNumber
 
 
-class RegistrationQueryParams(BaseModel):
-    pseudonym: str
-    oprf_key: str
-
-    def oprf_pseudonym(self) -> OprfPseudonym:
-        return OprfPseudonym(jwe=self.pseudonym, blind_factor=self.oprf_key)
-
-
-class CreateRegistrationRequest(RegistrationQueryParams):
-    pass
-
-
-class LocalizeRequest(RegistrationQueryParams):
-    pass
-
-
 class Registration(BaseModel):
     ura_number: str
     source_id: str

app/routers/localize.py

@@ -5,7 +5,8 @@
 from app.dependencies import get_pseudonym_referral_service
 from app.models.auth.context import AuthContext
 from app.models.auth.data import AuthorizationScope
-from app.models.registrations import LocalizeRequest, Registration
+from app.models.pseudonym import OprfPseudonym
+from app.models.registrations import Registration
 from app.routers.dependencies import require_scope
 from app.services.pseudonym_referral_service import PseudonymReferralService
 
@@ -14,10 +15,10 @@
 
 @router.post("")
 def localize(
-    data: Annotated[LocalizeRequest, Body()],
+    pseudonym: Annotated[OprfPseudonym, Body()],
     pseudonym_referral_service: Annotated[PseudonymReferralService, Depends(get_pseudonym_referral_service)],
     ctx: Annotated[AuthContext, Depends(require_scope(AuthorizationScope.LOCALIZE))],
 ) -> Any:
-    _, results = pseudonym_referral_service.localize(data.oprf_pseudonym(), ctx.claims.ura_number)
+    _, results = pseudonym_referral_service.localize(pseudonym, ctx.claims.ura_number)
 
     return [Registration.from_entity(r) for r in results]

app/routers/registrations.py

@@ -5,12 +5,8 @@
 from app.dependencies import get_pseudonym_referral_service
 from app.models.auth.context import AuthContext
 from app.models.auth.data import AuthorizationScope
-from app.models.registrations import (
-    CreateRegistrationRequest,
-    Registration,
-    RegistrationList,
-    RegistrationQueryParams,
-)
+from app.models.pseudonym import OprfPseudonym
+from app.models.registrations import Registration, RegistrationList
 from app.routers.dependencies import require_managing_source, require_scope
 from app.services.pseudonym_referral_service import PseudonymReferralService
 
@@ -22,10 +18,10 @@
 )
 def get_registration(
     pseudonym_referral_service: Annotated[PseudonymReferralService, Depends(get_pseudonym_referral_service)],
-    params: Annotated[RegistrationQueryParams, Query()],
+    pseudonym: Annotated[OprfPseudonym, Query()],
     ctx: Annotated[AuthContext, Depends(require_scope(AuthorizationScope.READ))],
 ) -> Any:
-    _, results = pseudonym_referral_service.query(ctx.claims.ura_number, pseudonym=params.oprf_pseudonym())
+    _, results = pseudonym_referral_service.query(ctx.claims.ura_number, pseudonym=pseudonym)
 
     return RegistrationList.from_entities(results)
 
@@ -36,14 +32,14 @@ def get_registration(
     responses={200: {"description": "The registration already existed and is returned unchanged"}},
 )
 def add_registration(
-    data: Annotated[CreateRegistrationRequest, Body()],
+    pseudonym: Annotated[OprfPseudonym, Body()],
     response: Response,
     pseudonym_referral_service: Annotated[PseudonymReferralService, Depends(get_pseudonym_referral_service)],
     ctx: Annotated[AuthContext, Depends(require_scope(AuthorizationScope.CREATE))],
     source: Annotated[str, Depends(require_managing_source)],
 ) -> Any:
     referral, created = pseudonym_referral_service.register(
-        data.oprf_pseudonym(),
+        pseudonym,
         ura_number=ctx.claims.ura_number,
         source=source,
     )
@@ -54,14 +50,14 @@ def add_registration(
 
 @router.delete("")
 def delete_registration(
-    params: Annotated[RegistrationQueryParams, Query()],
+    pseudonym: Annotated[OprfPseudonym, Query()],
     pseudonym_referral_service: Annotated[PseudonymReferralService, Depends(get_pseudonym_referral_service)],
     ctx: Annotated[AuthContext, Depends(require_scope(AuthorizationScope.DELETE))],
     source: Annotated[str, Depends(require_managing_source)],
 ) -> Any:
     pseudonym_referral_service.delete_many(
         ctx.claims.ura_number,
-        pseudonym=params.oprf_pseudonym(),
+        pseudonym=pseudonym,
         source=source,
     )
 

app/services/crypto_service_api_client.py

@@ -4,7 +4,7 @@
 from requests.exceptions import ConnectionError, HTTPError, Timeout
 
 from app.config import ConfigCryptoServiceApi
-from app.models.pseudonym import PseudonymResponse
+from app.models.pseudonym import OprfPseudonym, PseudonymResponse
 from app.services.http import HttpService
 
 logger = logging.getLogger(__name__)
@@ -20,14 +20,14 @@ def __init__(self, config: ConfigCryptoServiceApi) -> None:
             verify_ca=config.verify_ca,
         )
 
-    def exchange(self, jwe: str, blind_factor: str, label: str, mechanism: str) -> PseudonymResponse:
+    def exchange(self, pseudonym: OprfPseudonym, label: str, mechanism: str) -> PseudonymResponse:
         try:
             response = self._http.do_request(
                 method="POST",
                 sub_route="process",
                 data={
-                    "jwe": jwe,
-                    "blind_factor": blind_factor,
+                    "jwe": pseudonym.jwe,
+                    "blind_factor": pseudonym.blind_factor,
                     "label": label,
                     "mechanism": mechanism,
                 },

app/services/fhir/bundle.py

@@ -4,17 +4,15 @@
 from uuid import UUID
 
 import gfmodules.logging as gflog
+from pydantic import ValidationError
 
 from app.errors.handlers import log_rejection
 from app.errors.mapping import DEFAULT_SPEC, UNEXPECTED_ERROR_MESSAGE, spec_for
 from app.logging.events import Log
 from app.models.auth.context import AuthContext
 from app.models.auth.data import AuthorizationScope
 from app.models.fhir.bundle import Bundle, BundleEntry, EntryRequest, EntryRequestDto, EntryResponse
-from app.models.fhir.resources.localization_list.request import (
-    LocalizationListParams,
-    LocalizationListSearchParams,
-)
+from app.models.fhir.resources.localization_list.request import LocalizationListParams, LocalizationListSearchParams
 from app.models.fhir.resources.localization_list.resource import LocalizationList
 from app.models.ura import UraNumber
 from app.services.auth.auth_context import AuthContextService
@@ -184,9 +182,14 @@ def _parse_params(
         try:
             return model.model_validate(resolved_url.params)
         except ValueError as e:
+            if isinstance(e, ValidationError):
+                missing = all(err["type"] == "missing" for err in e.errors())
+            else:
+                missing = True
             return BundleEntry(
                 response=EntryResponse.make_validation_response(
-                    f"Bundle.entry.{index}.request: invalid url parameter: {e}"
+                    f"Bundle.entry.{index}.request: invalid url parameter: {e}",
+                    code="required" if missing else "invalid",
                 )
             )