Validate route handlers and JWT payloads

Author: adamtheturtle

.github/workflows/test.yml

@@ -132,6 +132,7 @@ jobs:
           - tests/mock_vws/test_model_target_retries.py
           - tests/mock_vws/test_vumark_generation_api.py
           - tests/mock_vws/test_vumark_generation_failure.py
+          - tests/mock_vws/test_mock_common.py
           - tests/mock_vws/test_target_validators.py
           - tests/mock_vws/test_requests_mock_usage.py
           - tests/mock_vws/test_respx_mock_usage.py

src/mock_vws/_httpx2_mock_server/decorators.py

@@ -17,7 +17,7 @@
 import httpx2
 from mock_response_delay.for_httpx2 import delayed_httpx2_handler
 
-from mock_vws._mock_common import RequestData, Route
+from mock_vws._mock_common import RequestData, Route, route_handler
 
 _ResponseType = tuple[int, Mapping[str, str], str | bytes]
 _Handler = Callable[[RequestData], _ResponseType]
@@ -285,10 +285,7 @@ def _mock_routes(
         for route in api.routes:
             url_pattern = base_url.rstrip("/") + route.path_pattern + "$"
             compiled_url_pattern = re.compile(pattern=url_pattern)
-            handler: _Handler = getattr(  # pylint: disable=bad-builtin
-                api,
-                route.route_name,
-            )  # ty: ignore[unsound-assignment]
+            handler = route_handler(api=api, route=route)
             httpx2_handler = delayed_httpx2_handler(
                 handler=_httpx2_handler(handler=handler, base_path=base_path),
                 delay_seconds=response_delay_seconds,

src/mock_vws/_mock_common.py

@@ -3,10 +3,11 @@
 import datetime
 import email.utils
 import json
+import operator
 import uuid
-from collections.abc import Iterable, Mapping
+from collections.abc import Callable, Iterable, Mapping
 from dataclasses import dataclass
-from typing import Any, Final, override
+from typing import Any, Final, TypeGuard, override
 
 from beartype import beartype
 
@@ -88,6 +89,27 @@ class Route:
     http_methods: Iterable[str]
 
 
+type RouteHandler = Callable[
+    [RequestData],
+    tuple[int, Mapping[str, str], str | bytes],
+]
+
+
+def _is_route_handler(value: object, /) -> TypeGuard[RouteHandler]:
+    """Return whether a route attribute can handle requests."""
+    return callable(value)
+
+
+@beartype
+def route_handler(*, api: object, route: Route) -> RouteHandler:
+    """Return the callable registered for a mock API route."""
+    candidate: object = operator.attrgetter(route.route_name)(api)
+    if not _is_route_handler(candidate):
+        message = f"{route.route_name} is not a route handler."
+        raise TypeError(message)
+    return candidate
+
+
 @beartype
 def _upload_order_key(target: ImageTarget) -> tuple[datetime.datetime, str]:
     """Give the sort key which orders targets by upload date, then ID.

src/mock_vws/_model_target_web_api.py

@@ -6,8 +6,9 @@
 import secrets
 import uuid
 import zipfile
+from collections.abc import Mapping
 from http import HTTPStatus
-from typing import Any, Protocol, runtime_checkable
+from typing import Any, Protocol, TypeGuard, runtime_checkable
 from urllib.parse import parse_qs
 
 from beartype import beartype
@@ -88,6 +89,13 @@ def remove_oauth2_client_credential(self, client_id: str) -> None:
 )
 _CLIENT_CREDENTIALS_SCOPE = "oauth2.clientcredentials.all"
 _MAX_CLIENT_CREDENTIALS = 100
+
+
+def _is_object_mapping(value: object, /) -> TypeGuard[Mapping[object, object]]:
+    """Return whether a value is a mapping with unchecked entries."""
+    return isinstance(value, Mapping)
+
+
 # A stable mock value standing in for the user-id segment that real
 # Vuforia embeds in some Model Target error targets such as
 # ``userId:7635391``. The numeric portion is per-account in real Vuforia;
@@ -290,8 +298,10 @@ def _jwt_header_error(*, bearer_token: str) -> str | None:
 
 
 @beartype
-def _jwt_payload_error(*, bearer_token: str) -> str | None:
-    """Return the Vuforia error for an invalid JSON Web Token payload."""
+def _jwt_payload(*, bearer_token: str) -> tuple[Mapping[object, object], bool]:
+    """Decode a JSON Web Token payload and report whether it is an
+    object.
+    """
     encoded_payload = bearer_token.split(sep=".")[1]
     try:
         padding = "=" * (-len(encoded_payload) % 4)
@@ -300,11 +310,20 @@ def _jwt_payload_error(*, bearer_token: str) -> str | None:
             altchars=b"-_",
             validate=True,
         )
-        payload = json.loads(s=decoded_payload)
+        payload: object = json.loads(s=decoded_payload)
     except ValueError:
         payload = None
 
-    if not isinstance(payload, dict):
+    if not _is_object_mapping(payload):
+        return dict[object, object](), False
+    return payload, True
+
+
+@beartype
+def _jwt_payload_error(*, bearer_token: str) -> str | None:
+    """Return the Vuforia error for an invalid JSON Web Token payload."""
+    _, is_object = _jwt_payload(bearer_token=bearer_token)
+    if not is_object:
         return "Payload of JWS object is not a valid JSON object"
     return None
 
@@ -334,19 +353,12 @@ def _jwt_signature_error(*, bearer_token: str) -> str | None:
 @beartype
 def _jwt_scopes(*, bearer_token: str) -> frozenset[str]:
     """Return scopes from a valid mock JSON Web Token."""
-    encoded_payload = bearer_token.split(sep=".")[1]
-    padding = "=" * (-len(encoded_payload) % 4)
-    payload = json.loads(
-        s=base64.b64decode(
-            s=encoded_payload + padding,
-            altchars=b"-_",
-            validate=True,
-        ),
-    )
-    scope = payload.get("scope", "")  # pyrefly: ignore [unknown-variable-type]
+    empty_scopes = frozenset[str]()
+    payload, _ = _jwt_payload(bearer_token=bearer_token)
+    scope: object = payload.get("scope", "")
     if not isinstance(scope, str):
-        return frozenset()  # ty: ignore[unsound-return-statement]
-    return frozenset(scope.split())  # ty: ignore[unsound-return-statement]
+        return empty_scopes
+    return frozenset(scope.split())
 
 
 @beartype

src/mock_vws/_respx_mock_server/decorators.py

@@ -10,7 +10,7 @@
 from mock_response_delay.for_httpx import delayed_httpx_handler
 from respx.mocks import Mocker
 
-from mock_vws._mock_common import RequestData, Route
+from mock_vws._mock_common import RequestData, Route, route_handler
 
 _ResponseType = tuple[int, Mapping[str, str], str | bytes]
 
@@ -152,10 +152,7 @@ def start_respx_router(
             compiled_url_pattern = re.compile(pattern=url_pattern)
 
             for http_method in route.http_methods:
-                original_callback = getattr(  # pylint: disable=bad-builtin
-                    api,
-                    route.route_name,
-                )
+                original_callback = route_handler(api=api, route=route)
                 _ = router.route(
                     method=http_method,
                     url=compiled_url_pattern,

src/mock_vws/decorators.py

@@ -19,7 +19,11 @@
     Httpx2Router,
     start_httpx2_router,
 )
-from mock_vws._mock_common import MissingSchemeError, RequestData
+from mock_vws._mock_common import (
+    MissingSchemeError,
+    RequestData,
+    route_handler,
+)
 from mock_vws._requests_mock_server.mock_web_query_api import (
     MockVuforiaWebQueryAPI,
 )
@@ -513,10 +517,7 @@ def __enter__(self) -> Self:
                 compiled_url_pattern = re.compile(pattern=url_pattern)
 
                 for http_method in route.http_methods:
-                    original_callback = getattr(  # pylint: disable=bad-builtin
-                        api,
-                        route.route_name,
-                    )
+                    original_callback = route_handler(api=api, route=route)
                     _ = mock.add_callback(
                         method=http_method,
                         url=compiled_url_pattern,

tests/mock_vws/test_mock_common.py

@@ -0,0 +1,27 @@
+"""Tests for shared mock route helpers."""
+
+import pytest
+
+from mock_vws._mock_common import Route, route_handler
+
+
+class _InvalidAPI:
+    """An API whose registered route attribute is not callable."""
+
+    invalid_route = 1
+
+
+def test_route_handler_must_be_callable() -> None:
+    """A route name must resolve to a callable handler."""
+    route = Route(
+        route_name="invalid_route",
+        path_pattern="/invalid",
+        http_methods=["GET"],
+    )
+    api = _InvalidAPI()
+    assert api.invalid_route == 1
+    with pytest.raises(
+        expected_exception=TypeError,
+        match="invalid_route is not a route handler",
+    ):
+        _ = route_handler(api=api, route=route)

tests/mock_vws/test_model_target_web_api.py

@@ -2445,8 +2445,20 @@ def test_client_credential_authentication_errors() -> None:
             )
 
     @staticmethod
-    def test_non_string_token_scope() -> None:
-        """A token with a non-string scope has no usable scopes."""
+    @pytest.mark.parametrize(
+        argnames=("payload", "status_code"),
+        argvalues=[
+            (b'{"scope":[]}', HTTPStatus.FORBIDDEN),
+            (b"[]", HTTPStatus.UNAUTHORIZED),
+        ],
+    )
+    def test_invalid_token_scope(
+        payload: bytes,
+        status_code: HTTPStatus,
+    ) -> None:
+        """A non-string scope or non-object payload has no usable
+        scopes.
+        """
         encoded_header = (
             base64.urlsafe_b64encode(
                 s=b'{"alg":"mock"}',
@@ -2456,7 +2468,7 @@ def test_non_string_token_scope() -> None:
         )
         encoded_payload = (
             base64.urlsafe_b64encode(
-                s=b'{"scope":[]}',
+                s=payload,
             )
             .decode(encoding="ascii")
             .rstrip("=")
@@ -2470,7 +2482,7 @@ def test_non_string_token_scope() -> None:
             )
             assert_model_target_status(
                 response=response,
-                status_codes=HTTPStatus.FORBIDDEN,
+                status_codes=status_code,
             )
 
     @staticmethod