Merge pull request #48 from minvws/feat/hsm-label-response

fouadsfarijlani · web-flow · commit 9dfa523b29c7 · 2026-07-14T15:33:24.000+02:00
updated pseudonym response to include additional data
diff --git a/app.conf.example b/app.conf.example
@@ -9,6 +9,8 @@ hashing_key_id=abcdef1234567890
 allow_insecure_requests = False
 # AES key-id used for encrypting and decrypting pseudonyms
 aes_key_id = 12345678
+# configurable AES mechanism to use, currently only AES_CBC is supported
+aes_mechanism = AES_CBC
 
 [logging]
 # Syslog addresses in host:port format (UDP)
diff --git a/app/config.py b/app/config.py
@@ -6,6 +6,8 @@
 
 from pydantic import BaseModel, ValidationError, Field
 
+from app.data import Pkc11Mechanism
+
 logger = logging.getLogger(__name__)
 
 _PATH = "app.conf"
@@ -26,6 +28,7 @@ class ConfigApp(BaseModel):
     nvi_ura_number: str
     hashing_key_id: str
     aes_key_id: str
+    aes_mechanism: Pkc11Mechanism
     allow_insecure_requests: bool = Field(default=False)
 
 
diff --git a/app/container.py b/app/container.py
@@ -19,7 +19,9 @@ def container_config(binder: inject.Binder) -> None:
     crypto_service: CryptoService
     if config.hsm_api.mock:
         logger.debug("Initializing mock crypto service")
-        crypto_service = MockCryptoService()
+        crypto_service = MockCryptoService(
+            aes_key_id=config.app.aes_key_id, aes_mechanism=config.app.aes_mechanism
+        )
     else:
         logger.debug(f"Initializing HSM API at {config.hsm_api.url}")
         http = HttpService(
@@ -36,6 +38,7 @@ def container_config(binder: inject.Binder) -> None:
             hash_key_id=config.app.hashing_key_id,
             support_sha1=config.hsm_api.support_sha1,
             aes_key_id=config.app.aes_key_id,
+            aes_mechanism=config.app.aes_mechanism,
         )
 
     binder.bind(CryptoService, crypto_service)
diff --git a/app/data.py b/app/data.py
@@ -0,0 +1,7 @@
+from enum import StrEnum
+
+
+class Pkc11Mechanism(StrEnum):
+    AES_CBC = "AES_CBC"
+    SHA256_HMAC = "SHA256_HMAC"
+    RSA_PKCS_OAEP = "RSA_PKCS_OAEP"
diff --git a/app/models/pseudonym.py b/app/models/pseudonym.py
@@ -4,3 +4,10 @@
 class PseudonymRequest(BaseModel):
     jwe: str
     blind_factor: str
+
+
+class PseudonymResponse(BaseModel):
+    encrypted_pseudonym: str
+    iv: str
+    label: str
+    mechanism: str
diff --git a/app/routers/crypto.py b/app/routers/crypto.py
@@ -44,11 +44,11 @@ def process(
     try:
         pseudonym = pseudonym_service.decrypt_and_unblind(data.jwe, data.blind_factor)
         hashed_pseudonym = pseudonym_service.hash(pseudonym)
-        encrypted_pseudonym, iv = pseudonym_service.encrypt_pseudonym(
-            pseudonym, hashed_pseudonym
+        results = pseudonym_service.encrypt_pseudonym(
+            pseudonym=pseudonym, hmac_hash=hashed_pseudonym
         )
         return JSONResponse(
-            content={"encrypted_pseudonym": encrypted_pseudonym, "iv": iv},
+            content=results.model_dump(),
             status_code=200,
         )
     except CryptoError as e:
diff --git a/app/services/crypto/crypto_service.py b/app/services/crypto/crypto_service.py
@@ -5,6 +5,7 @@
 
 from jwcrypto import jwe
 
+from app.data import Pkc11Mechanism
 from app.exceptions.exception import CryptoError, InvalidJweError
 
 logger = logging.getLogger(__name__)
@@ -15,6 +16,10 @@ class CryptoService(ABC):
     Interface for a cryptographic service that provides RSA key management, JWE decryption, and hashing.
     """
 
+    def __init__(self, aes_key_id: str, aes_mechanism: Pkc11Mechanism) -> None:
+        self.aes_key_id = aes_key_id
+        self.aes_mechanism = aes_mechanism
+
     @abstractmethod
     def health_check(self) -> bool:
         """
diff --git a/app/services/crypto/hsm_api_crypto_service.py b/app/services/crypto/hsm_api_crypto_service.py
@@ -1,25 +1,19 @@
 import base64
-from enum import StrEnum
 import json
 import logging
 
 from Crypto.Cipher import AES
 from requests import JSONDecodeError
 from requests.exceptions import ConnectionError as RequestsConnectionError, Timeout
 
+from app.data import Pkc11Mechanism
 from app.exceptions.exception import CryptoError, InvalidJweError, KeyNotFoundError
 from app.services.crypto.crypto_service import CryptoService
 from app.services.http import HttpService
 
 logger = logging.getLogger(__name__)
 
 
-class Pkc11Mechanism(StrEnum):
-    AES_CBC = "AES_CBC"
-    SHA256_HMAC = "SHA256_HMAC"
-    RSA_PKCS_OAEP = "RSA_PKCS_OAEP"
-
-
 class HsmApiCryptoService(CryptoService):
     def __init__(
         self,
@@ -28,6 +22,7 @@ def __init__(
         slot: str,
         hash_key_id: str,
         aes_key_id: str,
+        aes_mechanism: Pkc11Mechanism,
         support_sha1: bool = False,
     ):
         logger.debug(f"Initializing HSM API service: module={module}, slot={slot}")
@@ -36,7 +31,7 @@ def __init__(
         self.slot = slot
         self.support_sha1 = support_sha1
         self.hash_key_id = hash_key_id
-        self.aes_key_id = aes_key_id
+        super().__init__(aes_key_id, aes_mechanism)
 
     def health_check(self) -> bool:
         try:
diff --git a/app/services/crypto/mock_crypto_service.py b/app/services/crypto/mock_crypto_service.py
@@ -2,6 +2,7 @@
 import logging
 from typing import Any
 
+from app.data import Pkc11Mechanism
 from app.services.crypto.crypto_service import CryptoService
 
 logger = logging.getLogger(__name__)
@@ -18,6 +19,9 @@ class MockCryptoService(CryptoService):
     Intended for local development and wiring smoke tests.
     """
 
+    def __init__(self, aes_key_id: str, aes_mechanism: Pkc11Mechanism) -> None:
+        super().__init__(aes_key_id, aes_mechanism)
+
     def health_check(self) -> bool:
         return True
 
diff --git a/app/services/pseudonym_service.py b/app/services/pseudonym_service.py
@@ -1,11 +1,11 @@
 import base64
 import logging
-from typing import Tuple
 
 import pyoprf
 
 from app.exceptions.exception import CryptoError, InvalidJweError
 from app.logging.events import PSE_EXCHANGE_FAILED, PSE_EXCHANGE_OK, log_event
+from app.models.pseudonym import PseudonymResponse
 from app.services.crypto.crypto_service import CryptoService
 
 logger = logging.getLogger(__name__)
@@ -64,12 +64,22 @@ def decrypt_and_unblind(self, oprf_jwe: str, blind_factor: str) -> bytes:
         )
         return result
 
-    def encrypt_pseudonym(self, pseudonym: bytes, hmac_hash: bytes) -> Tuple[str, str]:
+    def encrypt_pseudonym(
+        self, pseudonym: bytes, hmac_hash: bytes
+    ) -> PseudonymResponse:
         iv = hmac_hash[:16]
         logger.debug("encrypting pseudonym")
         encrypted_data = self._crypto_service.encrypt_aes(pseudonym, iv)
         logger.debug("Pseudonym encrypted successfully")
-        return encrypted_data, base64.urlsafe_b64encode(iv).decode()
+
+        label = self._crypto_service.aes_key_id
+        mechanism = self._crypto_service.aes_mechanism
+        return PseudonymResponse(
+            encrypted_pseudonym=encrypted_data,
+            iv=base64.urlsafe_b64encode(iv).decode(),
+            label=label,
+            mechanism=mechanism,
+        )
 
     def hash(self, pseudonym: bytes) -> bytes:
         logger.debug("Hashing pseudonym")
diff --git a/tests/integration/test_decrypt_and_hash_flow.py b/tests/integration/test_decrypt_and_hash_flow.py
@@ -5,6 +5,7 @@
 from pytest_mock import MockerFixture
 
 
+
 def _b64u(data: bytes) -> str:
     return base64.urlsafe_b64encode(data).decode()
 
@@ -22,6 +23,8 @@ def test_process_full_flow(
         "app.services.pseudonym_service.pyoprf.unblind", return_value=b"unblinded"
     )
     crypto_stub.encrypt_aes.return_value = "some-encrypted_data"
+    crypto_stub.aes_key_id = "label-1"
+    crypto_stub.aes_mechanism = "AES_CBC"
 
     response = client.post(
         "/process",
@@ -32,6 +35,8 @@ def test_process_full_flow(
     assert response.json() == {
         "encrypted_pseudonym": "some-encrypted_data",
         "iv": _b64u(hmac_value[:16]),
+        "label": "label-1",
+        "mechanism": "AES_CBC",
     }
     crypto_stub.decrypt_jwe_payload.assert_called_once_with("JWE-TOKEN")
     crypto_stub.hash.assert_called_once_with(b"unblinded")
diff --git a/tests/unit/routers/test_crypto.py b/tests/unit/routers/test_crypto.py
@@ -8,6 +8,7 @@
 from app import container
 from app.config import Config
 from app.exceptions.exception import CryptoError, InvalidJweError, KeyNotFoundError
+from app.models.pseudonym import PseudonymResponse
 from app.routers.crypto import router as crypto_router
 
 
@@ -28,14 +29,21 @@ def test_process_returns_encrypted_pseudonym_with_iv(
 ) -> None:
     pseudonym_mock.decrypt_and_unblind.return_value = b"unblinded"
     pseudonym_mock.hash.return_value = "HASHED"
-    pseudonym_mock.encrypt_pseudonym.return_value = ("encrypted_data", "valid_iv")
+    pseudonym_mock.encrypt_pseudonym.return_value = PseudonymResponse(
+        encrypted_pseudonym="encrypted_data",
+        iv="valid_iv",
+        label="label-1",
+        mechanism="AES_CBC",
+    )
 
     response = client.post("/process", json={"jwe": "JWE", "blind_factor": "BF"})
 
     assert response.status_code == 200
     assert response.json() == {
         "encrypted_pseudonym": "encrypted_data",
         "iv": "valid_iv",
+        "label": "label-1",
+        "mechanism": "AES_CBC",
     }
     pseudonym_mock.decrypt_and_unblind.assert_called_once_with("JWE", "BF")
     pseudonym_mock.hash.assert_called_once_with(b"unblinded")
diff --git a/tests/unit/services/crypto/test_hsm_api_crypto.py b/tests/unit/services/crypto/test_hsm_api_crypto.py
@@ -15,8 +15,9 @@
 from app.exceptions.exception import CryptoError, InvalidJweError, KeyNotFoundError
 from app.services.crypto.hsm_api_crypto_service import (
     HsmApiCryptoService,
-    Pkc11Mechanism,
 )
+from app.data import Pkc11Mechanism
+from tests.unit.test_config import get_test_config
 
 
 def _resp(status: int, body: Any = None, text: str = "") -> MagicMock:
@@ -29,12 +30,14 @@ def _resp(status: int, body: Any = None, text: str = "") -> MagicMock:
 
 @pytest.fixture
 def service(http_mock: MagicMock) -> HsmApiCryptoService:
+    config = get_test_config()
     return HsmApiCryptoService(
         http_mock,
         module="m",
         slot="s",
         hash_key_id="hk",
-        aes_key_id="aes_key",
+        aes_key_id=config.app.aes_key_id,
+        aes_mechanism=config.app.aes_mechanism,
         support_sha1=False,
     )
 
@@ -201,6 +204,7 @@ def test_decrypt_jwe_validates_header_fields(
         hash_key_id="h",
         support_sha1=support_sha1,
         aes_key_id="aes-key-id",
+        aes_mechanism=Pkc11Mechanism.AES_CBC,
     )
     cek = os.urandom(32)
     token, _ = _make_jwe(cek, b"plain", alg=alg, enc=enc)
@@ -216,6 +220,7 @@ def test_decrypt_jwe_supports_sha1_when_enabled(http_mock: MagicMock) -> None:
         hash_key_id="h",
         support_sha1=True,
         aes_key_id="aes-key-id",
+        aes_mechanism=Pkc11Mechanism.AES_CBC,
     )
     cek = os.urandom(32)
     token, _ = _make_jwe(cek, b"plain", alg="RSA-OAEP")
diff --git a/tests/unit/services/crypto/test_mock_crypto.py b/tests/unit/services/crypto/test_mock_crypto.py
@@ -1,23 +1,37 @@
+import pytest
 from app.services.crypto.mock_crypto_service import MockCryptoService
+from tests.unit.test_config import get_test_config
 
 
-def test_health_check_is_always_true() -> None:
-    assert MockCryptoService().health_check() is True
+@pytest.fixture
+def mock_crypto_service() -> MockCryptoService:
+    config = get_test_config()
+    return MockCryptoService(
+        aes_key_id=config.app.aes_key_id, aes_mechanism=config.app.aes_mechanism
+    )
 
 
-def test_get_public_key_returns_sentinel() -> None:
-    assert MockCryptoService().get_public_key("any") == "no-key"
+def test_health_check_is_always_true(mock_crypto_service: MockCryptoService) -> None:
+    assert mock_crypto_service.health_check() is True
 
 
-def test_decrypt_jwe_round_trips_input() -> None:
-    assert MockCryptoService().decrypt_jwe("token", "kid") == b"token"
+def test_get_public_key_returns_sentinel(
+    mock_crypto_service: MockCryptoService,
+) -> None:
+    assert mock_crypto_service.get_public_key("any") == "no-key"
 
 
-def test_decrypt_jwe_payload_returns_canned_subject() -> None:
-    payload = MockCryptoService().decrypt_jwe_payload("token")
+def test_decrypt_jwe_round_trips_input(mock_crypto_service: MockCryptoService) -> None:
+    assert mock_crypto_service.decrypt_jwe("token", "kid") == b"token"
+
+
+def test_decrypt_jwe_payload_returns_canned_subject(
+    mock_crypto_service: MockCryptoService,
+) -> None:
+    payload = mock_crypto_service.decrypt_jwe_payload("token")
     assert isinstance(payload, dict)
     assert payload["subject"].startswith("pseudonym:eval:")
 
 
-def test_hash_is_identity() -> None:
-    assert MockCryptoService().hash(b"abc") == b"abc"
+def test_hash_is_identity(mock_crypto_service: MockCryptoService) -> None:
+    assert mock_crypto_service.hash(b"abc") == b"abc"
diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py
@@ -7,6 +7,7 @@
     ConfigTelemetry,
     ConfigUvicorn,
 )
+from app.data import Pkc11Mechanism
 
 
 def get_test_config() -> Config:
@@ -15,6 +16,7 @@ def get_test_config() -> Config:
             nvi_ura_number="12345678",
             hashing_key_id="hashing-key",
             aes_key_id="aes-key-id",
+            aes_mechanism=Pkc11Mechanism.AES_CBC,
         ),
         uvicorn=ConfigUvicorn(),
         telemetry=ConfigTelemetry(),
