Improve Model Target auth fidelity

adamtheturtle · adamtheturtle · commit 38926a49ccdd · 2026-05-21T16:10:53.000+01:00
diff --git a/docs/source/differences-to-vws.rst b/docs/source/differences-to-vws.rst
@@ -115,7 +115,8 @@ Model Target datasets
 
 The Model Target Web API mock supports OAuth2 token requests, standard and advanced dataset creation, status polling, dataset downloads, and deletion.
 The generated dataset download is a small valid zip file containing request metadata, not a real Vuforia Engine Model Target dataset.
-Model Target API routes accept any non-empty bearer token.
+Model Target API routes require a syntactically JSON Web Token-shaped bearer token, such as the token returned by the mock OAuth2 route.
+The mock does not verify token signatures, claims, expiry, or revocation.
 
 Header cases
 ------------
diff --git a/newsfragments/3192.change b/newsfragments/3192.change
@@ -0,0 +1 @@
+Improve Model Target Web API mock authentication failure responses.
diff --git a/src/mock_vws/_flask_server/vws.py b/src/mock_vws/_flask_server/vws.py
@@ -136,7 +136,7 @@ def _flask_request_data() -> RequestData:
         method=request.method,
         path=request.path,
         headers=dict(request.headers),
-        body=request.data,
+        body=request.get_data(parse_form_data=False),
     )
 
 
diff --git a/src/mock_vws/_model_target_web_api.py b/src/mock_vws/_model_target_web_api.py
@@ -16,6 +16,9 @@
 
 _ResponseType = tuple[int, dict[str, str], str | bytes]
 _MAX_ADVANCED_MODEL_COUNT = 20
+_JWT_DOT_COUNT = 2
+_MOCK_MODEL_TARGET_CLIENT_ID = "client-id"
+_MOCK_MODEL_TARGET_CLIENT_SECRET = "client-secret"  # noqa: S105
 
 
 @beartype
@@ -57,6 +60,16 @@ def _error_response(
     )
 
 
+@beartype
+def _oauth2_error_response(
+    *,
+    status_code: HTTPStatus,
+    body: dict[str, str],
+) -> _ResponseType:
+    """Return an OAuth2 error response."""
+    return _json_response(status_code=status_code, body=body)
+
+
 @beartype
 def _get_header(request: RequestData, name: str) -> str | None:
     """Return a request header, case-insensitively."""
@@ -67,6 +80,28 @@ def _get_header(request: RequestData, name: str) -> str | None:
     return None
 
 
+@beartype
+def _basic_auth_credentials(auth_header: str | None) -> tuple[str, str] | None:
+    """Return HTTP Basic credentials from an authorization header."""
+    if auth_header is None or not auth_header.startswith("Basic "):
+        return None
+
+    encoded_credentials = auth_header.removeprefix("Basic ").strip()
+    try:
+        decoded_credentials = base64.b64decode(
+            s=encoded_credentials,
+            validate=True,
+        ).decode(encoding="utf-8")
+    except ValueError:
+        return None
+
+    client_id, separator, client_secret = decoded_credentials.partition(":")
+    if not separator:
+        return None
+
+    return client_id, client_secret
+
+
 @beartype
 def _require_bearer_token(request: RequestData) -> _ResponseType | None:
     """Return an error response if the request has no bearer token."""
@@ -78,47 +113,95 @@ def _require_bearer_token(request: RequestData) -> _ResponseType | None:
             message="no Bearer token",
             target="jwt",
         )
-    if not auth_header.removeprefix("Bearer ").strip():
+    bearer_token = auth_header.removeprefix("Bearer ").strip()
+    if not bearer_token:
+        return _error_response(
+            status_code=HTTPStatus.UNAUTHORIZED,
+            code="401",
+            message="no Bearer token",
+            target="jwt",
+        )
+    if bearer_token.count(".") != _JWT_DOT_COUNT:
         return _error_response(
             status_code=HTTPStatus.UNAUTHORIZED,
             code="401",
-            message="invalid Bearer token",
+            message="Invalid JWT serialization: Missing dot delimiter(s)",
             target="jwt",
         )
     return None
 
 
+@beartype
+def _fake_jwt(*, token_source: bytes) -> str:
+    """Return a deterministic bearer token for the mock."""
+
+    def encode_part(value: dict[str, Any]) -> str:
+        """Return a base64url-encoded token part."""
+        raw_part = json.dumps(
+            obj=value,
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode(encoding="utf-8")
+        return (
+            base64.urlsafe_b64encode(s=raw_part)
+            .decode(
+                encoding="ascii",
+            )
+            .rstrip("=")
+        )
+
+    header = encode_part(value={"alg": "mock", "typ": "JWT"})
+    payload = encode_part(
+        value={
+            "aud": "vuforia-model-target",
+            "src": base64.urlsafe_b64encode(s=token_source)
+            .decode(
+                encoding="ascii",
+            )
+            .rstrip("="),
+        },
+    )
+    return f"{header}.{payload}.mock-signature"
+
+
 @beartype
 def oauth2_token(request: RequestData) -> _ResponseType:
     """Return a fake OAuth2 access token."""
     auth_header = _get_header(request=request, name="Authorization")
     form = parse_qs(qs=request.body.decode(encoding="utf-8"))
-    grant_type = form.get("grant_type", [""])[0]
-    has_basic_auth = auth_header is not None and auth_header.startswith(
-        "Basic ",
-    )
-    has_password_credentials = all(
-        form.get(field, [""])[0] for field in ("username", "password")
-    )
-    if grant_type not in {"", "client_credentials", "password"} or (
-        not has_basic_auth and not has_password_credentials
-    ):
-        return _error_response(
+    grant_type = form.get("grant_type", ["client_credentials"])[0]
+    if grant_type != "client_credentials":
+        return _oauth2_error_response(
             status_code=HTTPStatus.BAD_REQUEST,
-            code="BAD_REQUEST",
-            message="Invalid OAuth2 token request.",
-            target="grant_type",
+            body={"error": "unsupported_grant_type"},
+        )
+
+    basic_credentials = _basic_auth_credentials(auth_header=auth_header)
+    if basic_credentials is None:
+        return _oauth2_error_response(
+            status_code=HTTPStatus.UNAUTHORIZED,
+            body={
+                "error": "invalid_request",
+                "error_description": (
+                    "Missing or invalid authorization header"
+                ),
+            },
+        )
+
+    if basic_credentials != (
+        _MOCK_MODEL_TARGET_CLIENT_ID,
+        _MOCK_MODEL_TARGET_CLIENT_SECRET,
+    ):
+        return _oauth2_error_response(
+            status_code=HTTPStatus.UNAUTHORIZED,
+            body={"error": "invalid_client"},
         )
 
     token_source = request.body or (auth_header or "").encode()
-    access_token = base64.urlsafe_b64encode(s=token_source).decode(
-        encoding="ascii",
-    )
-    access_token = access_token.rstrip("=") or "mock-vuforia-access-token"
     return _json_response(
         status_code=HTTPStatus.OK,
         body={
-            "access_token": access_token,
+            "access_token": _fake_jwt(token_source=token_source),
             "token_type": "bearer",
             "expires_in": 3600,
         },
diff --git a/tests/mock_vws/test_model_target_web_api.py b/tests/mock_vws/test_model_target_web_api.py
@@ -17,6 +17,7 @@
 
 _VWS_HOST = "https://vws.vuforia.com"
 _DATASET_UUID = "0b12466eee5d49409a440927006ff5d8"
+_MOCK_BEARER_TOKEN = "mock.header.signature"
 
 
 def _dataset_request(*, cad_data_url: str) -> dict[str, Any]:
@@ -114,6 +115,17 @@ def _assert_model_target_error(
     }
 
 
+def _assert_oauth2_error(
+    *,
+    response: requests.Response,
+    status_code: HTTPStatus,
+    body: dict[str, str],
+) -> None:
+    """Assert an OAuth2 error response."""
+    assert response.status_code == status_code
+    assert response.json() == body
+
+
 @pytest.mark.usefixtures("verify_model_target_mock_vuforia")
 class TestAuthentication:
     """Tests for Model Target Web API authentication."""
@@ -195,46 +207,95 @@ def test_missing_bearer_token(
             },
         }
 
-
-class TestMockErrors:
-    """Tests for mock-only Model Target Web API error paths."""
-
     @staticmethod
-    def test_invalid_oauth2_token_request() -> None:
-        """Invalid OAuth2 token requests are rejected."""
-        with MockVWS():
-            response = requests.post(
-                url=f"{_VWS_HOST}/oauth2/token",
-                data={"grant_type": "unsupported"},
-                timeout=30,
-            )
+    @pytest.mark.parametrize(
+        argnames=("authorization", "message"),
+        argvalues=[
+            pytest.param("Bearer ", "no Bearer token", id="blank"),
+            pytest.param(
+                "Bearer invalid-token",
+                "Invalid JWT serialization: Missing dot delimiter(s)",
+                id="malformed",
+            ),
+        ],
+    )
+    def test_invalid_bearer_token(
+        *,
+        authorization: str,
+        message: str,
+    ) -> None:
+        """Invalid bearer tokens are rejected."""
+        response = requests.get(
+            url=f"{_VWS_HOST}/modeltargets/datasets/{_DATASET_UUID}/status",
+            headers={"Authorization": authorization},
+            timeout=30,
+        )
 
         _assert_model_target_error(
             response=response,
-            status_code=HTTPStatus.BAD_REQUEST,
-            code="BAD_REQUEST",
-            message="Invalid OAuth2 token request.",
-            target="grant_type",
+            status_code=HTTPStatus.UNAUTHORIZED,
+            code="401",
+            message=message,
+            target="jwt",
         )
 
     @staticmethod
-    def test_blank_bearer_token() -> None:
-        """A blank bearer token is rejected."""
-        with MockVWS():
-            response = requests.get(
-                url=f"{_VWS_HOST}/modeltargets/datasets/{_DATASET_UUID}/status",
-                headers={"Authorization": "Bearer "},
-                timeout=30,
-            )
+    @pytest.mark.parametrize(
+        argnames=("auth", "data", "status_code", "body"),
+        argvalues=[
+            pytest.param(
+                None,
+                {"grant_type": "client_credentials"},
+                HTTPStatus.UNAUTHORIZED,
+                {
+                    "error": "invalid_request",
+                    "error_description": (
+                        "Missing or invalid authorization header"
+                    ),
+                },
+                id="missing-basic-auth",
+            ),
+            pytest.param(
+                ("invalid-client-id", "invalid-client-secret"),
+                {"grant_type": "client_credentials"},
+                HTTPStatus.UNAUTHORIZED,
+                {"error": "invalid_client"},
+                id="invalid-client",
+            ),
+            pytest.param(
+                ("invalid-client-id", "invalid-client-secret"),
+                {"grant_type": "unsupported"},
+                HTTPStatus.BAD_REQUEST,
+                {"error": "unsupported_grant_type"},
+                id="unsupported-grant-type",
+            ),
+        ],
+    )
+    def test_invalid_oauth2_token_request(
+        *,
+        auth: tuple[str, str] | None,
+        data: dict[str, str],
+        status_code: HTTPStatus,
+        body: dict[str, str],
+    ) -> None:
+        """Invalid OAuth2 token requests are rejected."""
+        response = requests.post(
+            url=f"{_VWS_HOST}/oauth2/token",
+            auth=auth,
+            data=data,
+            timeout=30,
+        )
 
-        _assert_model_target_error(
+        _assert_oauth2_error(
             response=response,
-            status_code=HTTPStatus.UNAUTHORIZED,
-            code="401",
-            message="invalid Bearer token",
-            target="jwt",
+            status_code=status_code,
+            body=body,
         )
 
+
+class TestMockErrors:
+    """Tests for mock-only Model Target Web API error paths."""
+
     @staticmethod
     @pytest.mark.parametrize(
         argnames=("body", "headers", "message", "target"),
@@ -266,7 +327,10 @@ def test_invalid_request_body(
         with MockVWS():
             response = requests.post(
                 url=f"{_VWS_HOST}/modeltargets/datasets",
-                headers={"Authorization": "Bearer token", **headers},
+                headers={
+                    "Authorization": f"Bearer {_MOCK_BEARER_TOKEN}",
+                    **headers,
+                },
                 data=body,
                 timeout=30,
             )
@@ -337,7 +401,7 @@ def test_invalid_dataset_request(
         with MockVWS():
             response = requests.post(
                 url=f"{_VWS_HOST}{path}",
-                headers={"Authorization": "Bearer token"},
+                headers={"Authorization": f"Bearer {_MOCK_BEARER_TOKEN}"},
                 json=body,
                 timeout=30,
             )
@@ -381,7 +445,7 @@ def test_unknown_dataset(
             response = requests.request(
                 method=method,
                 url=f"{_VWS_HOST}{path}",
-                headers={"Authorization": "Bearer token"},
+                headers={"Authorization": f"Bearer {_MOCK_BEARER_TOKEN}"},
                 timeout=30,
             )
 
@@ -399,7 +463,7 @@ def test_processing_dataset_cannot_be_downloaded() -> None:
         with MockVWS(processing_time_seconds=60):
             create_response = requests.post(
                 url=f"{_VWS_HOST}/modeltargets/datasets",
-                headers={"Authorization": "Bearer token"},
+                headers={"Authorization": f"Bearer {_MOCK_BEARER_TOKEN}"},
                 json=_UNAUTHENTICATED_DATASET_REQUEST,
                 timeout=30,
             )
@@ -408,7 +472,7 @@ def test_processing_dataset_cannot_be_downloaded() -> None:
                     f"{_VWS_HOST}/modeltargets/datasets/"
                     f"{create_response.json()['uuid']}/dataset"
                 ),
-                headers={"Authorization": "Bearer token"},
+                headers={"Authorization": f"Bearer {_MOCK_BEARER_TOKEN}"},
                 timeout=30,
             )
 
diff --git a/tests/mock_vws/test_requests_mock_usage.py b/tests/mock_vws/test_requests_mock_usage.py
diff --git a/tests/mock_vws/test_respx_mock_usage.py b/tests/mock_vws/test_respx_mock_usage.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+Improve Model Target Web API mock authentication failure responses.`
Original file line number	Diff line number	Diff line change
`@@ -136,7 +136,7 @@ def _flask_request_data() -> RequestData:`
`136`	`136`	`method=request.method,`
`137`	`137`	`path=request.path,`
`138`	`138`	`headers=dict(request.headers),`
`139`		`- body=request.data,`
	`139`	`+ body=request.get_data(parse_form_data=False),`
`140`	`140`	`)`
`141`	`141`
`142`	`142`