Merge branch 'main' into fix_models

.github/workflows/tests.yaml

@@ -60,7 +60,7 @@ jobs:
         if: steps.changed-models-or-api.outputs.any_changed == 'true'
         run: |
           echo "${{ secrets.SERVICE_ACCOUNT_JSON }}" > /tmp/edutap-wallet-google-credentials.json
-          echo "EDUTAP_WALLET_GOOGLE_ISSUER_ID=${{ secrets.ISSUER_ID }}" > .env
+          echo "EDUTAP_WALLET_GOOGLE_TEST_ISSUER_ID=${{ secrets.ISSUER_ID }}" > .env
           echo "EDUTAP_WALLET_GOOGLE_CREDENTIALS_FILE=/tmp/edutap-wallet-google-credentials.json" >> .env
           echo "EDUTAP_WALLET_GOOGLE_INTEGRATION_TEST_PREFIX=${{ github.run_id }}" >> .env
       - name: Test with tox

.pre-commit-config.yaml

@@ -42,8 +42,9 @@ repos:
         - "pydantic_settings"
         - "fastapi"
         - "httpx"
+        - "freezegun"
 -   repo: https://github.com/codespell-project/codespell
-    rev: v2.3.0
+    rev: v2.4.0
     hooks:
     -   id: codespell
         additional_dependencies:

examples/edutap_wallet_google_example_callback/README.md

@@ -76,9 +76,11 @@ We hope you get the idea.
 
 There is an example swarm deployment in here in `swarm.yml`.
 It can be deployed on the cluster.
-The public domain is configured using the environment variable `EDUTAP_WALLET_GOOGLE_EXAMPLE_DOMAIN`.
+The public domain must be configured using the environment variable `EDUTAP_WALLET_GOOGLE_EXAMPLE_DOMAIN`.
 A TLS certificate will be issued automatically using Lets Encrypt.
 
+Example: `export EDUTAP_WALLET_GOOGLE_EXAMPLE_DOMAIN=edutap-wallet-google-callback.example.com`
+
 ```shell
 docker stack deploy swarm.yml -c swarm.yml edutap_wallet_google_example_callback
 ```

pyproject.toml

@@ -47,12 +47,13 @@ callback = [
     "httpx",
 ]
 test = [
-    "pytest",
+    "edutap.wallet_google[callback]",
+    "freezegun",
     "pytest-cov",
     "pytest-explicit",
+    "pytest",
     "requests-mock",
     "tox",
-    "edutap.wallet_google[callback]",
 ]
 typecheck  = [
     "google-auth-stubs",

src/edutap/wallet_google/api.py

@@ -18,6 +18,7 @@
 from google.auth import jwt
 from pydantic import ValidationError
 
+import datetime
 import json
 import logging
 import typing
@@ -289,13 +290,9 @@ def listing(
             raise ValueError("resource_id of a class must be given to list its objects")
         params["classId"] = resource_id
     elif name.endswith("Class"):
-        is_pageable = True
         if not issuer_id:
-            issuer_id = session_manager.settings.issuer_id
-            if not issuer_id:
-                raise ValueError(
-                    "'issuer_id' must be passed as keyword argument or set in environment"
-                )
+            raise ValueError("issuer_id must be given to list classes")
+        is_pageable = True
         params["issuerId"] = issuer_id
 
     if is_pageable:
@@ -338,10 +335,62 @@ def listing(
         break
 
 
+def _create_payload(models: list[ClassModel | ObjectModel | Reference]) -> JWTPayload:
+    """Creates a payload for the JWT."""
+    payload = JWTPayload()
+
+    for model in models:
+        if isinstance(model, Reference):
+            if model.model_name is not None:
+                name = lookup_metadata_by_name(model.model_name)["plural"]
+            elif model.model_type is not None:
+                name = lookup_metadata_by_model_type(model.model_type)["plural"]
+        else:
+            name = lookup_metadata_by_model_instance(model)["plural"]
+        if getattr(payload, name) is None:
+            setattr(payload, name, [])
+        getattr(payload, name).append(model)
+    return payload
+
+
+def _convert_str_or_datetime_to_str(value: str | datetime.datetime) -> str:
+    """convert and check the value to be a valid string for the JWT claim timestamps"""
+    if isinstance(value, datetime.datetime):
+        return str(int(value.timestamp()))
+    if value == "":
+        return value
+    if not value.isdecimal():
+        raise ValueError("string must be a decimal")
+    if int(value) < 0:
+        raise ValueError("string must be an int >= 0 number")
+    if int(value) > 2**32:
+        raise ValueError("string must be an int < 2**32 number")
+    return value
+
+
+def _create_claims(
+    issuer: str,
+    origins: list[str],
+    models: list[ClassModel | ObjectModel | Reference],
+    iat: str | datetime.datetime,
+    exp: str | datetime.datetime,
+) -> JWTClaims:
+    """Creates a JWTClaims instance based on the given issuer, origins and models."""
+    return JWTClaims(
+        iss=issuer,
+        iat=_convert_str_or_datetime_to_str(iat),
+        exp=_convert_str_or_datetime_to_str(exp),
+        origins=origins,
+        payload=_create_payload(models),
+    )
+
+
 def save_link(
     models: list[ClassModel | ObjectModel | Reference],
     *,
     origins: list[str] = [],
+    iat: str | datetime.datetime = "",
+    exp: str | datetime.datetime = "",
 ) -> str:
     """
     Creates a link to save a Google Wallet Object to the wallet on the device.
@@ -360,26 +409,26 @@ def save_link(
                         The Google Wallet API button will not render when the origins field is not defined.
                         You could potentially get an "Load denied by X-Frame-Options" or "Refused to display"
                         messages in the browser console when the origins field is not defined.
+    :param: iat:        Issued At Time. The time when the JWT was issued.
+    :param: exp:        Expiration Time. The time when the JWT expires.
     :return:            Link with JWT to save the resources to the wallet.
     """
-    payload = JWTPayload()
-    for model in models:
-        if isinstance(model, Reference):
-            if model.model_name is not None:
-                name = lookup_metadata_by_name(model.model_name)["plural"]
-            elif model.model_type is not None:
-                name = lookup_metadata_by_model_type(model.model_type)["plural"]
-        else:
-            name = lookup_metadata_by_model_instance(model)["plural"]
-        if getattr(payload, name) is None:
-            setattr(payload, name, [])
-        getattr(payload, name).append(model)
-
-    claims = JWTClaims(
-        iss=session_manager.settings.credentials_info["client_email"],
-        origins=origins,
-        payload=payload,
+    claims = _create_claims(
+        session_manager.settings.credentials_info["client_email"],
+        origins,
+        models,
+        iat=iat,
+        exp=exp,
+    )
+    logger.debug(
+        claims.model_dump_json(
+            indent=2,
+            exclude_unset=False,
+            exclude_defaults=False,
+            exclude_none=True,
+        )
     )
+
     signer = crypt.RSASigner.from_service_account_file(
         session_manager.settings.credentials_file
     )
@@ -392,6 +441,7 @@ def save_link(
             exclude_none=True,
         ),
     ).decode("utf-8")
+    logger.debug(jwt_string)
     if (jwt_len := len(jwt_string)) >= 1800:
         logger.debug(
             f"JWT-Length: {jwt_len} is larger than recommended 1800 bytes",

src/edutap/wallet_google/handlers/fastapi.py

@@ -145,7 +145,6 @@ async def handle_image(request: Request, encrypted_image_id: str):
 # needs to be included after the routers are defined
 router = APIRouter(
     prefix=session_manager.settings.handler_prefix,
-    tags=["edutap", "google_wallet"],
 )
 router.include_router(router_callback)
 router.include_router(router_images)

src/edutap/wallet_google/handlers/validate.py

@@ -138,11 +138,18 @@ def verified_signed_message(data: CallbackData) -> SignedMessage:
     Verifies the signature of the callback data.
     and returns the parsed SignedMessage
     """
+    # parse message
+    message = SignedMessage.model_validate_json(data.signedMessage)
 
+    # get issuer_id
+    if not message.classId or "." not in message.classId:
+        raise ValueError("Missing classId")
+    issuer_id = message.classId.split(".")[0]
+
+    # shortcut if signature validation is disabled
     settings = session_manager.settings
     if settings.handler_callback_verify_signature == "0":
-        # shortcut if signature validation is disabled
-        return SignedMessage.model_validate_json(data.signedMessage)
+        return message
 
     if data.protocolVersion != PROTOCOL_VERSION:
         raise ValueError("Invalid protocolVersion")
@@ -166,7 +173,7 @@ def verified_signed_message(data: CallbackData) -> SignedMessage:
     signature = base64.decodebytes(bytes(data.signature, "utf-8"))
     signed_data = _construct_signed_data(
         "GooglePayWallet",
-        settings.issuer_id,
+        issuer_id,
         PROTOCOL_VERSION,
         data.signedMessage,
     )
@@ -175,4 +182,4 @@ def verified_signed_message(data: CallbackData) -> SignedMessage:
     except (ValueError, InvalidSignature):
         raise ValueError("Invalid signature")
 
-    return SignedMessage.model_validate_json(data.signedMessage)
+    return message

src/edutap/wallet_google/models/datatypes/jwt.py

@@ -7,6 +7,7 @@
 from ..passes import retail
 from ..passes import tickets_and_transit
 from ..passes.bases import Reference
+from datetime import datetime
 
 
 class JWTPayload(Model):
@@ -38,7 +39,8 @@ class JWTClaims(Model):
 
     iss: str
     aud: str = "google"
-    typ: str = "savettowallet"
-    iat: str = ""
+    typ: str = "savetowallet"
+    iat: str | datetime = ""
+    exp: str | datetime = ""
     payload: JWTPayload
     origins: list[str]

src/edutap/wallet_google/settings.py

@@ -1,5 +1,4 @@
 from pathlib import Path
-from pydantic import EmailStr
 from pydantic import Field
 from pydantic import HttpUrl
 from pydantic_settings import BaseSettings
@@ -47,8 +46,7 @@ class Settings(BaseSettings):
 
     credentials_file: Path = ROOT_DIR / "tests" / "data" / "credentials_fake.json"
     credentials_scopes: list[str] = SCOPES
-    issuer_account_email: EmailStr | None = None
-    issuer_id: str = Field(default="")
+    test_issuer_id: str = Field(default="")
 
     fernet_encryption_key: str = ""
 

tests/conftest.py

@@ -99,3 +99,10 @@ def mock_settings():
 @pytest.fixture
 def mock_fernet_encryption_key(mock_settings):
     mock_settings.fernet_encryption_key = "TDTPJVv24gha-jRX0apPgPpMDN2wX1kVSNNZdWXcz8E="
+
+
+@pytest.fixture
+def test_issuer_id():
+    from edutap.wallet_google.session import session_manager
+
+    yield session_manager.settings.test_issuer_id

tests/data/test_wallet_google_plugins/plugins.py

@@ -24,6 +24,8 @@ async def image_by_id(self, image_id: str) -> ImageData:
 class TestCallbackHandler:
     """
     Implementation of edutap.wallet_google.protocols.CallbackHandler
+
+    Used in tests to simulate a callback handler and possible errors.
     """
 
     async def handle(
@@ -35,8 +37,8 @@ async def handle(
         count: int,
         nonce: str,
     ) -> None:
-        if class_id == "TIMEOUT":
+        if class_id.startswith("TIMEOUT"):
             await asyncio.sleep(exp_time_millis / 1000)
-        elif class_id:
+        elif nonce:
             return
-        raise ValueError("class_id is required")
+        raise ValueError("test case errors if nonce is 0")

tests/integration/test_CRULM.py

@@ -130,7 +130,7 @@ def test_class_object_cru(type_base, class_data, object_data, integration_test_i
     ############################
     # test class
     class_base = f"{integration_test_id}.{class_type}.test_CRU.wallet_google.edutap"
-    class_data["id"] = f"{session_manager.settings.issuer_id}.{class_base}"
+    class_data["id"] = f"{session_manager.settings.test_issuer_id}.{class_base}"
 
     data = new(class_type, class_data)
 
@@ -186,14 +186,19 @@ def test_class_object_cru(type_base, class_data, object_data, integration_test_i
         assert result_message.id == class_data["id"]
 
     # list all
-    result_list = [x for x in listing(name=class_type)]
+    result_list = [
+        x
+        for x in listing(
+            name=class_type, issuer_id=session_manager.settings.test_issuer_id
+        )
+    ]
     assert len(result_list) > 0
 
     ############################
     # test object
     object_data["classId"] = class_data["id"]
     object_base = f"{integration_test_id}.{object_type}.test_CRU.wallet_google.edutap"
-    object_data["id"] = f"{session_manager.settings.issuer_id}.{object_base}"
+    object_data["id"] = f"{session_manager.settings.test_issuer_id}.{object_base}"
     odata = new(object_type, object_data)
 
     # create