From c4d2199dc0346d929484488c78fa022966b28ef8 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Thu, 13 Aug 2026 19:30:24 +0100 Subject: [PATCH] Reject Model Target requests with a non-integer Content-Length Towards #3400. Co-Authored-By: Claude Opus 5 (1M context) --- .../model-target-content-length.change | 1 + spelling_private_dict.txt | 1 + src/mock_vws/_model_target_web_api.py | 49 +++++++ tests/mock_vws/test_model_target_web_api.py | 126 ++++++++++++++++++ 4 files changed, 177 insertions(+) create mode 100644 newsfragments/model-target-content-length.change diff --git a/newsfragments/model-target-content-length.change b/newsfragments/model-target-content-length.change new file mode 100644 index 000000000..013898388 --- /dev/null +++ b/newsfragments/model-target-content-length.change @@ -0,0 +1 @@ +Reject Model Target Web API and OAuth2 token requests with a ``Content-Length`` header which is not an integer, matching the load balancer in front of real Vuforia. diff --git a/spelling_private_dict.txt b/spelling_private_dict.txt index ef4380bc2..2924c70a3 100644 --- a/spelling_private_dict.txt +++ b/spelling_private_dict.txt @@ -15,6 +15,7 @@ ascii auth backend backends +balancer beartype binascii bool diff --git a/src/mock_vws/_model_target_web_api.py b/src/mock_vws/_model_target_web_api.py index d9178bbf0..f7e1b8d90 100644 --- a/src/mock_vws/_model_target_web_api.py +++ b/src/mock_vws/_model_target_web_api.py @@ -12,6 +12,9 @@ from beartype import beartype from mock_vws._mock_common import RequestData, json_dump +from mock_vws._services_validators.exceptions import ( + ContentLengthHeaderNotIntError, +) from mock_vws.model_target import ( ModelTargetDataset, ModelTargetDatasetType, @@ -168,6 +171,32 @@ def _get_header(request: RequestData, name: str) -> str | None: return None +@beartype +def _content_length_error(request: RequestData) -> _ResponseType | None: + """Return an error response if ``Content-Length`` is not an integer. + + The load balancer in front of real Vuforia rejects a request with a + ``Content-Length`` header which is not an integer before the request + reaches any API, so the Model Target Web API gives the same response + as the VWS API does. + + A ``Content-Length`` header which is too large is not handled here. + Real Vuforia waits for the body it was promised and then times out, + which is too slow to verify in a test. + """ + given_content_length = _get_header(request=request, name="Content-Length") + if given_content_length is None: + return None + + try: + int(given_content_length) + except ValueError: + error = ContentLengthHeaderNotIntError() + return (error.status_code, dict(error.headers), error.response_text) + + return None + + @beartype def _basic_auth_credentials(auth_header: str | None) -> tuple[str, str] | None: """Return HTTP Basic credentials from an authorization header.""" @@ -337,6 +366,10 @@ def encode_part(value: dict[str, Any]) -> str: @beartype def oauth2_token(request: RequestData) -> _ResponseType: """Return a fake OAuth2 access token.""" + content_length_error = _content_length_error(request=request) + if content_length_error is not None: + return content_length_error + auth_header = _get_header(request=request, name="Authorization") # A form body which is not valid UTF-8 is decoded leniently rather than # raising, so that a body which cannot be decoded is treated as one which @@ -882,6 +915,10 @@ def create_model_target_dataset( generation_warning: ModelTargetGenerationWarning | None, ) -> _ResponseType: """Create a standard or advanced Model Target dataset.""" + content_length_error = _content_length_error(request=request) + if content_length_error is not None: + return content_length_error + auth_error = _require_bearer_token(request=request) if auth_error is not None: return auth_error @@ -952,6 +989,10 @@ def get_model_target_dataset_status( dataset_type: ModelTargetDatasetType, ) -> _ResponseType: """Return the status of a Model Target dataset.""" + content_length_error = _content_length_error(request=request) + if content_length_error is not None: + return content_length_error + auth_error = _require_bearer_token(request=request) if auth_error is not None: return auth_error @@ -1001,6 +1042,10 @@ def download_model_target_dataset( dataset_type: ModelTargetDatasetType, ) -> _ResponseType: """Download a generated Model Target dataset.""" + content_length_error = _content_length_error(request=request) + if content_length_error is not None: + return content_length_error + auth_error = _require_bearer_token(request=request) if auth_error is not None: return auth_error @@ -1043,6 +1088,10 @@ def delete_model_target_dataset( dataset_type: ModelTargetDatasetType, ) -> _ResponseType: """Delete a Model Target dataset.""" + content_length_error = _content_length_error(request=request) + if content_length_error is not None: + return content_length_error + auth_error = _require_bearer_token(request=request) if auth_error is not None: return auth_error diff --git a/tests/mock_vws/test_model_target_web_api.py b/tests/mock_vws/test_model_target_web_api.py index abc61a70a..eb13e1efc 100644 --- a/tests/mock_vws/test_model_target_web_api.py +++ b/tests/mock_vws/test_model_target_web_api.py @@ -4,6 +4,7 @@ import dataclasses import io import json +import textwrap import zipfile from http import HTTPMethod, HTTPStatus from typing import Any @@ -23,6 +24,7 @@ ) from tests.mock_vws.fixtures.vuforia_backends import VuforiaBackend from tests.mock_vws.utils import ModelTargetEndpoint +from tests.mock_vws.utils.assertions import assert_valid_date_header _VWS_HOST = "https://vws.vuforia.com" _MOCK_BEARER_TOKEN = "eyJhbGciOiJtb2NrIn0.e30.c2lnbmF0dXJl" @@ -152,6 +154,36 @@ def _assert_model_target_error( } +@beartype +def _assert_load_balancer_bad_request(*, response: Response) -> None: + """Assert the ``BAD_REQUEST`` response from the load balancer. + + The load balancer in front of Vuforia rejects some requests before + they reach an API, with an HTML error page rather than a Model Target + Web API error body. + """ + assert response.status_code == HTTPStatus.BAD_REQUEST + assert_valid_date_header(response=response) + expected_response_text = textwrap.dedent( + text="""\ + \r + 400 Bad Request\r + \r +

400 Bad Request

\r + \r + \r + """, + ) + assert response.text == expected_response_text + assert response.headers == { + "Content-Length": str(object=len(response.text)), + "Content-Type": "text/html", + "Connection": "close", + "Server": "awselb/2.0", + "Date": response.headers["Date"], + } + + @beartype def _assert_unknown_dataset(*, response: Response) -> None: """Assert a NOT_FOUND error for the unknown dataset UUID which the @@ -358,6 +390,100 @@ def test_invalid_bearer_token( ) +@pytest.mark.usefixtures("verify_model_target_mock_vuforia") +class TestContentLength: + """Tests for the ``Content-Length`` header on every Model Target + endpoint. + + These mirror the cross-cutting tests which the ``endpoint`` fixture + supports for the VWS and Query APIs. + + A ``Content-Length`` header which is too large is not covered, for the + same reason as it is not covered for the VWS API: real Vuforia waits + for the body it was promised before timing out, which takes too long + to run in a test. + """ + + @staticmethod + def test_not_integer( + *, + model_target_endpoint: ModelTargetEndpoint, + ) -> None: + """A ``Content-Length`` header which is not an integer is rejected + by the load balancer in front of Vuforia, before any bearer token + is looked at. + """ + new_endpoint = dataclasses.replace( + model_target_endpoint, + headers={ + **model_target_endpoint.headers, + "Content-Length": "0.4", + }, + ) + + response = new_endpoint.send() + + _assert_load_balancer_bad_request(response=response) + + @staticmethod + def test_not_integer_oauth2_token() -> None: + """The OAuth2 token endpoint is behind the same load balancer. + + It is not in the ``model_target_endpoint`` fixture because it takes + HTTP Basic credentials rather than a bearer token. + """ + endpoint = ModelTargetEndpoint( + base_url=_VWS_HOST, + path_url="/oauth2/token", + method=HTTPMethod.POST, + headers={ + "Content-Type": "application/x-www-form-urlencoded", + "Content-Length": "0.4", + }, + data=b"grant_type=client_credentials", + takes_json_body=False, + ) + + response = endpoint.send() + + _assert_load_balancer_bad_request(response=response) + + @staticmethod + def test_too_small( + *, + model_target_endpoint: ModelTargetEndpoint, + ) -> None: + """A ``Content-Length`` header which is too small truncates the + body, and the request is still rejected for having no bearer + token. + + The Model Target Web API does not sign the request body, so unlike + the VWS API it has no reason to notice the truncation before it + looks at the ``Authorization`` header. + """ + if not model_target_endpoint.takes_json_body: + return + + content_length = len(model_target_endpoint.data) - 1 + new_endpoint = dataclasses.replace( + model_target_endpoint, + headers={ + **model_target_endpoint.headers, + "Content-Length": str(object=content_length), + }, + ) + + response = new_endpoint.send() + + _assert_model_target_error( + response=response, + status_code=HTTPStatus.UNAUTHORIZED, + code="401", + message="no Bearer token", + target="jwt", + ) + + class TestInvalidJson: """Tests for giving Model Target endpoints bodies which are not valid JSON objects.