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
5 changes: 5 additions & 0 deletions docs/source/differences-to-vws.rst
Original file line number Diff line number Diff line change
Expand Up @@ -248,6 +248,11 @@ token revocation.

Dataset creation request bodies which are valid JSON but not JSON objects are
reported as missing every required top-level field.
Dataset creation request bodies which cannot be decoded as UTF-8 are reported
as invalid JSON, as malformed JSON bodies are.
An OAuth2 token request body which cannot be decoded as UTF-8 is treated as one
which does not name a grant type; the real response to such a body has not been
observed.
Dataset creation requests are validated for the required top-level ``models``,
``name`` and ``targetSdk`` fields, for those fields' types, for each ``models``
entry being a JSON object, and for the number of models.
Expand Down
1 change: 1 addition & 0 deletions newsfragments/model-target-non-utf-8-body.change
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Reject Model Target dataset creation requests with a body which cannot be decoded as UTF-8, rather than raising an error in the mock, and decode OAuth2 token request bodies leniently.
13 changes: 10 additions & 3 deletions src/mock_vws/_model_target_web_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -338,7 +338,12 @@ def encode_part(value: dict[str, Any]) -> str:
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"))
# 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
# does not name a grant type.
form = parse_qs(
qs=request.body.decode(encoding="utf-8", errors="replace"),
)
grant_type = form.get("grant_type", ["client_credentials"])[0]
if grant_type != "client_credentials":
return _oauth2_error_response(
Expand Down Expand Up @@ -397,8 +402,10 @@ def _load_request_json(request: RequestData) -> dict[str, Any] | _ResponseType:
details=None,
)
try:
request_json: dict[str, Any] = json.loads(s=request.body)
except json.JSONDecodeError as exc:
request_json: dict[str, Any] = json.loads(
s=request.body.decode(encoding="utf-8"),
)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
return _error_response(
status_code=HTTPStatus.BAD_REQUEST,
code="ERROR",
Expand Down
60 changes: 60 additions & 0 deletions tests/mock_vws/test_model_target_web_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -434,6 +434,41 @@ def test_invalid_json(
assert error["message"].startswith("Invalid Json")
assert "target" not in error

@staticmethod
def test_body_not_utf_8(
*,
verify_model_target_mock_vuforia: VuforiaBackend,
model_target_endpoint: ModelTargetEndpoint,
) -> None:
"""Bodies which are not valid UTF-8 are rejected with 400 by
endpoints which read a body, and are ignored elsewhere.
"""
access_token = _access_token_for_backend(
backend=verify_model_target_mock_vuforia,
)
content = b"\xff{}"
new_endpoint = dataclasses.replace(
model_target_endpoint,
headers={
**model_target_endpoint.headers,
"Authorization": f"Bearer {access_token}",
"Content-Length": str(object=len(content)),
},
data=content,
)

response = new_endpoint.send()

if not model_target_endpoint.takes_json_body:
_assert_unknown_dataset(response=response)
return

assert response.status_code == HTTPStatus.BAD_REQUEST
error = json.loads(s=response.text)["error"]
assert error["code"] == "ERROR"
assert error["message"].startswith("Invalid Json")
assert "target" not in error

@staticmethod
@pytest.mark.parametrize(
argnames="body",
Expand Down Expand Up @@ -1282,6 +1317,31 @@ def test_advanced_realistic_appearance_not_in_enum() -> None:

assert standard_response.status_code == HTTPStatus.CREATED

@staticmethod
def test_oauth2_token_body_not_utf_8(
*,
model_target_mock_only_vuforia: VuforiaBackend,
) -> None:
"""An OAuth2 token request with a body which is not valid UTF-8 is
treated as one which does not name a grant type.

Mock-only because the real response to a form body which cannot be
decoded has not been observed.
"""
credentials = credentials_for_backend(
backend=model_target_mock_only_vuforia,
)

response = requests.post(
url=f"{_VWS_HOST}/oauth2/token",
auth=(credentials.client_id, credentials.client_secret),
data=b"\xff",
timeout=30,
)

assert response.status_code == HTTPStatus.OK
assert response.json()["token_type"] == "bearer"

@staticmethod
def test_processing_dataset_cannot_be_downloaded() -> None:
"""A dataset cannot be downloaded while it is still processing.
Expand Down
Loading