Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions newsfragments/model-target-content-length.change
Original file line number Diff line number Diff line 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.
1 change: 1 addition & 0 deletions spelling_private_dict.txt
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ ascii
auth
backend
backends
balancer
beartype
binascii
bool
Expand Down
49 changes: 49 additions & 0 deletions src/mock_vws/_model_target_web_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
126 changes: 126 additions & 0 deletions tests/mock_vws/test_model_target_web_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import dataclasses
import io
import json
import textwrap
import zipfile
from http import HTTPMethod, HTTPStatus
from typing import Any
Expand All @@ -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"
Expand Down Expand Up @@ -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="""\
<html>\r
<head><title>400 Bad Request</title></head>\r
<body>\r
<center><h1>400 Bad Request</h1></center>\r
</body>\r
</html>\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
Expand Down Expand Up @@ -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.
Expand Down
Loading