feat: response_code in redirect_uri (#257)

* feat: stub of resp_code in callback uri

* fix: interfaces and unit test for resp code

* fix: url > uri

* fix: redirect -> 200

* feat: code from session storage to hmac(state)

* feat: helper class to simply method injection

* chore: rename and cleanup

* chore: cleaned outdated comments

* feat: response_code: hmac -> aead

* chore: clean docs

* chore: cleanup

* fix: status endpoint with code

* Apply suggestions from code review

* Apply suggestions from code review

---------

Co-authored-by: Giuseppe De Marco <[email protected]>

example/satosa/integration_test/main.py

@@ -237,8 +237,8 @@
     data={'response': encrypted_response},
     timeout=TIMEOUT_S
 )
-assert 'redirect_url' in authz_response_ok.content.decode()
-callback_uri = json.loads(authz_response_ok.content.decode())['redirect_url']
+assert 'redirect_uri' in authz_response_ok.content.decode()
+callback_uri = json.loads(authz_response_ok.content.decode())['redirect_uri']
 satosa_authn_response = http_user_agent.get(
     callback_uri,
     verify=False,

example/satosa/pyeudiw_backend.yaml

@@ -30,6 +30,9 @@ config:
     expiration_time: 120 # seconds
     logo_path: 'wallet-it/wallet-icon-blue.svg' # relative to static_storage_url
 
+  response_code:
+    sym_key: "1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef" # hex string of 64 characters
+
   jwt:
     default_sig_alg: ES256 # or RS256
     default_enc_alg: RSA-OAEP

pyeudiw/satosa/default/openid4vp_backend.py

@@ -11,6 +11,7 @@
 from pyeudiw.satosa.schemas.config import PyeudiwBackendConfig
 from pyeudiw.jwk import JWK
 from pyeudiw.satosa.utils.html_template import Jinja2TemplateHandler
+from pyeudiw.satosa.utils.respcode import ResponseCodeSource
 from pyeudiw.satosa.utils.response import JsonResponse
 from pyeudiw.satosa.utils.trust import BackendTrust
 from pyeudiw.storage.db_engine import DBEngine
@@ -93,6 +94,8 @@ def __init__(
             debug_message = f"""The backend configuration presents the following validation issues: {e}"""
             self._log_warning("OpenID4VPBackend", debug_message)
 
+        self.response_code_helper = ResponseCodeSource(self.config["response_code"]["sym_key"])
+
         self._log_debug(
             "OpenID4VP init",
             f"loaded configuration: {json.dumps(config)}"
@@ -228,35 +231,27 @@ def pre_request_endpoint(self, context: Context, internal_request, **kwargs) ->
     def get_response_endpoint(self, context: Context) -> Response:
 
         self._log_function_debug("get_response_endpoint", context)
-        # TODO: questa cosa si sfascia perché la funzione di callback non consuma id come query parameter.
-        # Vedi https://italia.github.io/eudi-wallet-it-docs/versione-corrente/en/relying-party-solution.html#redirect-uri
-        # Probabilmente quello che dovrebbe fare è:
-        # (1) Il response handler dovrebbe generare un code (crittograficamente sicuro con 128 bit o più di entropia)
-        # (2) Dovrebbe fare un binding tra il code e il transaction-id (usando la terminologia di openid4vp)
-        # (3) Questo metodo dovrebbe recuperare (dal code) il transaction-id
-        # (4) Dal transaction-id si dovrebbero recuperare i dati di autenticazione dell'utente
-        # è possibile che questa soluzione sia leggermente sopvraingegnerizzata perché pensata promossa da microsoft con tutto a microservizi
-        state = context.qs_params.get("id", None)
+        resp_code = context.qs_params.get("response_code", None)
         session_id = context.state.get("SESSION_ID", None)
 
-        if not state:
-            return self._handle_400(context, "No session id found")
+        if not session_id:
+            return self._handle_400(context, "session id not found")
+
+        state = ""
+        try:
+            state = self.response_code_helper.recover_state(resp_code)
+        except Exception:
+            return self._handle_400(context, "missing or invalid parameter [response_code]")
 
         finalized_session = None
 
         try:
-            if state:
-                # cross device
-                finalized_session = self.db_engine.get_by_state_and_session_id(
-                    state=state, session_id=session_id
-                )
-            else:
-                # same device
-                finalized_session = self.db_engine.get_by_session_id(
-                    session_id=session_id
-                )
+            finalized_session = self.db_engine.get_by_state_and_session_id(
+                state=state,
+                session_id=session_id
+            )
         except Exception as e:
-            _msg = f"Error while retrieving session by state {state} and session_id {session_id}: {e}"
+            _msg = f"Error while retrieving internal response with response_code {resp_code} and session_id {session_id}: {e}"
             return self._handle_401(context, _msg, e)
 
         if not finalized_session:
@@ -308,15 +303,11 @@ def status_endpoint(self, context: Context) -> JsonResponse:
             if iat_now() > request_object["exp"]:
                 return self._handle_403("expired", "Request object expired")
 
-        if session["finalized"]:
-            #  return Redirect(
-            #      self.registered_get_response_endpoint
-            #  )
-            # TODO: rivedere il redirect URI, non mi è per nulla chiaro; inoltre va allineato con response_handler.py
-            #  https://relying.party/callback?response_code=<crypto secure random string with ≥ 128 bit entropy>
+        if (session["finalized"] is True):
+            resp_code = self.response_code_helper.create_code(state)
             return JsonResponse(
                 {
-                    "redirect_uri": f"{self.registered_get_response_endpoint}?id={state}"
+                    "redirect_uri": f"{self.registered_get_response_endpoint}?response_code={resp_code}"
                 },
                 status="200"
             )

pyeudiw/satosa/default/request_handler.py

@@ -16,6 +16,7 @@ class RequestHandler(RequestHandlerInterface, BackendDPoP, BackendTrust):
     def request_endpoint(self, context: Context, *args) -> JsonResponse:
         self._log_function_debug("response_endpoint", context, "args", args)
 
+
         try:
             state = context.qs_params["id"]
         except Exception as e:
@@ -38,7 +39,6 @@ def request_endpoint(self, context: Context, *args) -> JsonResponse:
             "iat": iat_now(),
             "exp": exp_from_now(minutes=self.config['authorization']['expiration_time'])
         }
-
         # take the session created in the pre-request authz endpoint
         try:
             document = self.db_engine.get_by_state(state)

pyeudiw/satosa/default/response_handler.py

@@ -168,6 +168,8 @@ def response_endpoint(self, context: Context, *args: tuple) -> Redirect | JsonRe
         internal_resp = self._translate_response(
             all_user_attributes, _info["issuer"], context
         )
+        response_code = self.response_code_helper.create_code(state)
+
         try:
             self.db_engine.update_response_object(
                 stored_session['nonce'], state, internal_resp
@@ -186,15 +188,13 @@ def response_endpoint(self, context: Context, *args: tuple) -> Redirect | JsonRe
 
         if stored_session['session_id'] == context.state["SESSION_ID"]:
             # Same device flow
-            # TODO: rivedere il redirect uri
-            #  https://relying.party/callback?response_code=<crypto secure random string with ≥ 128 bit entropy>
-            cb_redirect_uri = f"{self.registered_get_response_endpoint}?id={state}"
-            return JsonResponse({"redirect_url": cb_redirect_uri}, status="200")
+            cb_redirect_uri = f"{self.registered_get_response_endpoint}?response_code={response_code}"
+            return JsonResponse({"redirect_uri": cb_redirect_uri}, status="200")
         else:
             # Cross device flow
             return JsonResponse({"status": "OK"}, status="200")
 
-    def _translate_response(self, response: dict, issuer: str, context: Context):
+    def _translate_response(self, response: dict, issuer: str, context: Context) -> InternalData:
         """
         Translates wallet response to SATOSA internal response.
         :type response: dict[str, str]
@@ -231,6 +231,7 @@ def _translate_response(self, response: dict, issuer: str, context: Context):
         # TODO - ACR values
         internal_resp = InternalData(auth_info=auth_info)
 
+        # (re)define the response subject
         sub = ""
         pepper = self.config.get("user_attributes", {})[
             'subject_id_random_value'
@@ -256,8 +257,8 @@ def _translate_response(self, response: dict, issuer: str, context: Context):
             sub = hashlib.sha256(
                 f"{json.dumps(response).encode()}~{pepper}".encode()
             ).hexdigest()
-
         response["sub"] = [sub]
+
         internal_resp.attributes = self.converter.to_internal(
             "openid4vp", response
         )

pyeudiw/satosa/schemas/config.py

@@ -2,6 +2,7 @@
 from pyeudiw.jwk.schemas.public import JwkSchema
 from pyeudiw.satosa.schemas.endpoint import EndpointsConfig
 from pyeudiw.satosa.schemas.qrcode import QRCode
+from pyeudiw.satosa.schemas.response import ResponseConfig
 from pyeudiw.satosa.schemas.autorization import AuthorizationConfig
 from pyeudiw.satosa.schemas.user_attributes import UserAttributesConfig
 from pyeudiw.federation.schemas.federation_configuration import FederationConfig
@@ -15,6 +16,7 @@ class PyeudiwBackendConfig(BaseModel):
     ui: UiConfig
     endpoints: EndpointsConfig
     qrcode: QRCode
+    response_code: ResponseConfig
     jwt: JWTConfig
     authorization: AuthorizationConfig
     user_attributes: UserAttributesConfig

pyeudiw/satosa/schemas/response.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class ResponseConfig(BaseModel):
+    sym_key: str

pyeudiw/satosa/utils/respcode.py

@@ -0,0 +1,80 @@
+import base64
+from dataclasses import dataclass, field
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+import secrets
+import string
+
+CODE_SYM_KEY_LEN = 32  # in bytes (256 bits)
+
+
+@dataclass
+class ResponseCodeSource:
+    """ResponseCodeSource is a utility box that wraps a secreet key and
+    exposes utility methods that define the relationship between request
+    status and response code.
+
+    The class assumes that the response status is a string with UTF-8
+    encoding. When this is not true, the resulting chipertext might
+    be longer than necessary.
+
+    Constructor arguments:
+    :param key: encryption/decryption key, represented as a hex string
+    :type key: str
+    """
+
+    key: str = field(repr=False)  # repr=False as we do not want to accidentally expose a secret key in a log file
+
+    def __post_init__(self):
+        # Validate input(s)
+        _ = decode_key(self.key)
+
+    def create_code(self, state: str) -> str:
+        return create_code(state, self.key)
+
+    def recover_state(self, code: str) -> str:
+        return recover_state(code, self.key)
+
+
+def decode_key(key: str) -> bytes:
+    if not set(key) <= set(string.hexdigits):
+        raise ValueError("key in format different than hex currently not supported")
+    key_len = len(key)
+    if key_len != 2*CODE_SYM_KEY_LEN:
+        raise ValueError(f"invalid key: key should be {CODE_SYM_KEY_LEN} bytes, obtained instead: {key_len//2}")
+    return bytes.fromhex(key)
+
+
+def _base64_encode_no_pad(b: bytes) -> str:
+    return base64.urlsafe_b64encode(b).decode().rstrip('=')
+
+
+def _base64_decode_no_pad(s: str) -> bytes:
+    padded = s + "="*((4 - len(s) % 4) % 4)
+    return base64.urlsafe_b64decode(padded)
+
+
+def _encrypt_state(msg: bytes, key: bytes) -> bytes:
+    nonce = secrets.token_bytes(12)
+    ciphertext = AESGCM(key).encrypt(nonce, msg, b'')
+    return nonce + ciphertext
+
+
+def _decrypt_code(encrypted_token: bytes, key: bytes) -> bytes:
+    nonce = encrypted_token[:12]
+    ciphertext = encrypted_token[12:]
+    dec = AESGCM(key).decrypt(nonce, ciphertext, b'')
+    return dec
+
+
+def create_code(state: str, key: str) -> str:
+    bkey = decode_key(key)
+    msg = bytes(state, encoding='utf-8')
+    code = _encrypt_state(msg, bkey)
+    return _base64_encode_no_pad(code)
+
+
+def recover_state(code: str, key: str) -> str:
+    bkey = decode_key(key)
+    enc = _base64_decode_no_pad(code)
+    state = _decrypt_code(enc, bkey)
+    return state.decode(encoding='utf-8')

pyeudiw/tests/satosa/utils/test_respcode.py

@@ -0,0 +1,48 @@
+import pytest
+
+from pyeudiw.satosa.utils.respcode import ResponseCodeSource, create_code, recover_state
+
+
+def test_valid_resp_code():
+    state = "state"
+    key = "1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"
+    code = create_code(state, key)
+    assert recover_state(code, key) == state
+
+
+def test_invalid_resp_code():
+    key = "1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"
+    try:
+        recover_state("this_is_an_invalid_response_code", key)
+        assert False
+    except Exception:
+        assert True
+
+
+def test_bad_key():
+    key = ""
+    try:
+        create_code("state", key)
+        assert False
+    except ValueError:
+        assert True
+
+
+class TestResponseCodeHelper:
+
+    @pytest.fixture(autouse=True)
+    def setup(self):
+        key = "1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"
+        self.respose_code_helper = ResponseCodeSource(key)
+
+    def test_valid_code(self):
+        state = "state"
+        code = self.respose_code_helper.create_code(state)
+        assert self.respose_code_helper.recover_state(code) == state
+
+    def test_invalid_code(self):
+        try:
+            self.respose_code_helper.create_code("this_is_an_invalid_response_code")
+            assert False
+        except Exception:
+            assert True

pyeudiw/tests/settings.py

@@ -33,6 +33,9 @@
         "status": "/status-uri",
         "get_response": "/get-response",
     },
+    "response_code": {
+        "sym_key": "1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"
+    },
     "qrcode": {
         "size": 100,
         "color": "#2B4375",

pyeudiw/tests/storage/test_db_engine.py

@@ -56,7 +56,6 @@ def test_update_response_object(self):
 
     def test_update_response_object_unexistent_id_object(self):
         response_object = {"response_object": "response_object"}
-
         try:
             self.engine.update_response_object(
                 str(uuid.uuid4()), str(uuid.uuid4()), response_object)

pyeudiw/tests/storage/test_mongo_storage.py

@@ -89,7 +89,6 @@ def test_update_response_object(self):
         state = str(uuid.uuid4())
 
         request_object = {"nonce": nonce, "state": state}
-
         self.storage.update_request_object(
             document_id, request_object)
         documentStatus = self.storage.update_response_object(