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
18 changes: 9 additions & 9 deletions docs/source/differences-to-vws.rst
Original file line number Diff line number Diff line change
Expand Up @@ -266,15 +266,15 @@ The mock returns the same report for the current month and the previous month.
As with real Vuforia, the report is served with a ``text/plain`` content type
rather than a CSV one.

The mock does not use the database ID in the request path.
It uses the database which matches the request's server keys, and it accepts
any database ID.
Real Vuforia returns a 401 response with the ``AuthenticationFailure`` result
code for a request which is signed with valid server keys but which names a
database ID that those keys do not belong to.
That includes naming the database by its name rather than by its ID.
:class:`mock_vws.database.CloudDatabase` has no database ID, so the mock
cannot make the same check.
Real Vuforia assigns a database an ID, which the target manager shows.
The ID of a database in the mock is
:paramref:`mock_vws.database.CloudDatabase.database_id`, which defaults to a
random string, so the path of a request to this endpoint is built by reading
that attribute rather than by looking the ID up.
As real Vuforia does, the mock returns a 401 response with the
``AuthenticationFailure`` result code for a request which is signed with valid
server keys but which names any other database, including one named by its
name rather than by its ID.

Real Vuforia returns a presigned URL for cloud storage.
The mock returns a URL served by the mock itself, without the query
Expand Down
1 change: 1 addition & 0 deletions newsfragments/cloud-database-id.change
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Give ``CloudDatabase`` a ``database_id``, and reject a reco counts report request whose path names a database which the request's server keys do not belong to, as real Vuforia does.
5 changes: 5 additions & 0 deletions src/mock_vws/_flask_server/target_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -232,6 +232,10 @@ def create_cloud_database() -> Response:
"client_secret_key",
random_database.client_secret_key,
)
database_id = request_json.get(
"database_id",
random_database.database_id,
)
database_name = request_json.get(
"database_name",
random_database.database_name,
Expand Down Expand Up @@ -271,6 +275,7 @@ def create_cloud_database() -> Response:
server_secret_key=server_secret_key,
client_access_key=client_access_key,
client_secret_key=client_secret_key,
database_id=database_id,
database_name=database_name,
state=state,
database_type=database_type,
Expand Down
4 changes: 2 additions & 2 deletions src/mock_vws/_flask_server/vws.py
Original file line number Diff line number Diff line change
Expand Up @@ -408,8 +408,8 @@ def reco_counts_report(database_id: str) -> Response:
Fake implementation of
https://developer.vuforia.com/library/web-api/cloud-targets-web-services-api
"""
# The mock authenticates with the request's server keys, so the database
# ID in the path is not used.
# The database ID in the path is validated against the request's server
# keys before the request reaches this route.
del database_id
settings = VWSSettings.model_validate(obj={})
return _to_flask_response(
Expand Down
8 changes: 8 additions & 0 deletions src/mock_vws/_services_validators/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
validate_content_length_header_not_too_small,
)
from .content_type_validators import validate_content_type_header_given
from .database_id_validators import validate_database_id_matches_keys
from .date_validators import (
validate_date_format,
validate_date_header_given,
Expand Down Expand Up @@ -91,6 +92,13 @@ def run_services_validators(
request_path=request_path,
databases=databases,
)
validate_database_id_matches_keys(
request_headers=request_headers,
request_body=request_body,
request_method=request_method,
request_path=request_path,
databases=databases,
)
validate_request_quota(
request_headers=request_headers,
request_body=request_body,
Expand Down
76 changes: 76 additions & 0 deletions src/mock_vws/_services_validators/database_id_validators.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
"""Validators for database IDs given in request paths."""

import logging
import re
from collections.abc import Iterable, Mapping

from beartype import beartype

from mock_vws._database_matchers import (
AnyDatabase,
get_database_matching_server_keys,
)
from mock_vws._mock_common import RECO_COUNTS_REPORT_PATH_PATTERN
from mock_vws._services_validators.exceptions import (
AuthenticationFailureError,
)
from mock_vws.database import CloudDatabase

_LOGGER = logging.getLogger(name=__name__)
# The index of the database ID in
# ``/imagetargets/databases/{database_id}/reports/recoCounts``, split on "/".
_DATABASE_ID_PATH_INDEX = 3


@beartype
def validate_database_id_matches_keys(
*,
request_path: str,
request_headers: Mapping[str, str],
request_body: bytes,
request_method: str,
databases: Iterable[AnyDatabase],
) -> None:
"""Validate a database ID given in the request path.

The ID must be the ID of the database which the request's server keys
belong to.

Args:
request_path: The path of the request.
request_headers: The headers sent with the request.
request_body: The body of the request.
request_method: The HTTP method of the request.
databases: All Vuforia databases.

Raises:
AuthenticationFailureError: The request path names a database other
than the one which the request's server keys belong to.
"""
if not re.fullmatch(
pattern=RECO_COUNTS_REPORT_PATH_PATTERN,
string=request_path,
):
return

database = get_database_matching_server_keys(
request_headers=request_headers,
request_body=request_body,
request_method=request_method,
request_path=request_path,
databases=databases,
)

given_database_id = request_path.split(sep="/")[_DATABASE_ID_PATH_INDEX]
if (
isinstance(database, CloudDatabase)
and database.database_id == given_database_id
):
return

_LOGGER.warning(
'The database ID "%s" is not the ID of the database which the '
"request's server keys belong to.",
given_database_id,
)
raise AuthenticationFailureError
8 changes: 8 additions & 0 deletions src/mock_vws/database.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
class CloudDatabaseDict(TypedDict):
"""A dictionary type which represents a cloud database."""

database_id: str
database_name: str
server_access_key: str
server_secret_key: str
Expand Down Expand Up @@ -63,6 +64,10 @@ class CloudDatabase:
"""Credentials for VWS APIs.

Args:
database_id: The identifier of a VWS target manager database. Defaults
to a random string. Endpoints which name a database in their path,
such as the reco counts report endpoint, accept only the identifier
of the database which the request's server keys belong to.
database_name: The name of a VWS target manager database name. Defaults
to a random string.
server_access_key: A VWS server access key. Defaults to a random
Expand Down Expand Up @@ -92,6 +97,7 @@ class CloudDatabase:

# We hide a few things in the ``repr`` with ``repr=False`` so that they do
# not show up in CI logs.
database_id: str = field(default_factory=_random_hex, repr=False)
database_name: str = field(default_factory=_random_hex, repr=False)
server_access_key: str = field(default_factory=_random_hex, repr=False)
server_secret_key: str = field(default_factory=_random_hex, repr=False)
Expand Down Expand Up @@ -128,6 +134,7 @@ def to_dict(self) -> CloudDatabaseDict:
else self.request_rate_limits.to_dict()
)
return {
"database_id": self.database_id,
"database_name": self.database_name,
"server_access_key": self.server_access_key,
"server_secret_key": self.server_secret_key,
Expand Down Expand Up @@ -166,6 +173,7 @@ def from_dict(cls, database_dict: CloudDatabaseDict) -> Self:
)

return cls(
database_id=database_dict["database_id"],
database_name=database_dict["database_name"],
server_access_key=database_dict["server_access_key"],
server_secret_key=database_dict["server_secret_key"],
Expand Down
12 changes: 2 additions & 10 deletions tests/mock_vws/fixtures/credentials.py
Original file line number Diff line number Diff line change
Expand Up @@ -136,8 +136,9 @@ def get_model_target_credentials() -> ModelTargetCredentials:
@pytest.fixture
def vuforia_database() -> CloudDatabase:
"""Return VWS credentials from environment variables."""
settings = _CloudDatabaseSettings.model_validate(obj={})
settings = _WorkingCloudDatabaseSettings.model_validate(obj={})
return CloudDatabase(
database_id=settings.database_id,
database_name=settings.target_manager_database_name,
server_access_key=settings.server_access_key,
server_secret_key=settings.server_secret_key,
Expand All @@ -147,15 +148,6 @@ def vuforia_database() -> CloudDatabase:
)


@pytest.fixture
def vuforia_database_id() -> str:
"""Return the ID of the working database from environment
variables.
"""
settings = _WorkingCloudDatabaseSettings.model_validate(obj={})
return settings.database_id


@pytest.fixture
def inactive_cloud_database() -> CloudDatabase:
"""
Expand Down
1 change: 1 addition & 0 deletions tests/mock_vws/fixtures/vuforia_backends.py
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,7 @@ def _enable_use_mock_vuforia(
"""Test against the in-memory mock Vuforia."""
assert monkeypatch
working_database = CloudDatabase(
database_id=working_database.database_id,
database_name=working_database.database_name,
server_access_key=working_database.server_access_key,
server_secret_key=working_database.server_secret_key,
Expand Down
85 changes: 47 additions & 38 deletions tests/mock_vws/test_reco_counts_report.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@

from mock_vws._constants import ResultCodes
from mock_vws.database import CloudDatabase
from tests.mock_vws.fixtures.vuforia_backends import VuforiaBackend

_VWS_HOST = "https://vws.vuforia.com"
# The number of seconds which the mocks take to generate a report.
Expand All @@ -41,7 +40,11 @@ def _request_reco_counts_report(
database_id: str,
month: str | int,
) -> requests.Response:
"""Request a reco counts report and return the response."""
"""Request a reco counts report and return the response.

The report is requested for the database named by the given ID, and the
request is signed with the given database's server keys.
"""
request_path = f"/imagetargets/databases/{database_id}/reports/recoCounts"
content_type = "application/json"
content = json.dumps(obj={"month": month}).encode(encoding="utf-8")
Expand Down Expand Up @@ -69,23 +72,7 @@ def _request_reco_counts_report(
)


@beartype
def _database_id_for_backend(
*,
backend: VuforiaBackend,
vuforia_database_id: str,
) -> str:
"""Return the database ID to name in the request path.

Real Vuforia requires the ID to be the ID of the database which the
request's server keys belong to. The mocks accept any ID.
"""
if backend != VuforiaBackend.REAL:
return uuid.uuid4().hex

return vuforia_database_id


@pytest.mark.usefixtures("verify_mock_vuforia")
class TestRecoCountsReport:
"""Tests for requesting a reco counts report."""

Expand All @@ -97,20 +84,15 @@ class TestRecoCountsReport:
)
def test_reco_counts_report(
*,
verify_mock_vuforia: VuforiaBackend,
vuforia_database: CloudDatabase,
vuforia_database_id: str,
months_ago: int,
) -> None:
"""A report can be requested for the current and previous
month.
"""
response = _request_reco_counts_report(
vuforia_database=vuforia_database,
database_id=_database_id_for_backend(
backend=verify_mock_vuforia,
vuforia_database_id=vuforia_database_id,
),
database_id=vuforia_database.database_id,
month=_month_offset_from_now(months=-months_ago),
)

Expand All @@ -134,18 +116,13 @@ def test_reco_counts_report(
)
def test_month_out_of_range(
*,
verify_mock_vuforia: VuforiaBackend,
vuforia_database: CloudDatabase,
vuforia_database_id: str,
months_ago: int,
) -> None:
"""Only the current and the previous month can be requested."""
response = _request_reco_counts_report(
vuforia_database=vuforia_database,
database_id=_database_id_for_backend(
backend=verify_mock_vuforia,
vuforia_database_id=vuforia_database_id,
),
database_id=vuforia_database.database_id,
month=_month_offset_from_now(months=-months_ago),
)

Expand All @@ -161,25 +138,57 @@ def test_month_out_of_range(
)
def test_malformed_month(
*,
verify_mock_vuforia: VuforiaBackend,
vuforia_database: CloudDatabase,
vuforia_database_id: str,
month: str | int,
) -> None:
"""The month must be given in the ``YYYY-mm`` form."""
response = _request_reco_counts_report(
vuforia_database=vuforia_database,
database_id=_database_id_for_backend(
backend=verify_mock_vuforia,
vuforia_database_id=vuforia_database_id,
),
database_id=vuforia_database.database_id,
month=month,
)

assert response.status_code == HTTPStatus.BAD_REQUEST
response_json = json.loads(s=response.text)
assert response_json["result_code"] == ResultCodes.FAIL.value

@staticmethod
def test_unknown_database_id(*, vuforia_database: CloudDatabase) -> None:
"""The path must name the database which the request's server
keys belong to.
"""
response = _request_reco_counts_report(
vuforia_database=vuforia_database,
database_id=uuid.uuid4().hex,
month=_month_offset_from_now(months=0),
)

assert response.status_code == HTTPStatus.UNAUTHORIZED
response_json = json.loads(s=response.text)
assert (
response_json["result_code"]
== ResultCodes.AUTHENTICATION_FAILURE.value
)

@staticmethod
def test_database_name_in_path(
*,
vuforia_database: CloudDatabase,
) -> None:
"""A database is named in the path by its ID, not by its name."""
response = _request_reco_counts_report(
vuforia_database=vuforia_database,
database_id=vuforia_database.database_name,
month=_month_offset_from_now(months=0),
)

assert response.status_code == HTTPStatus.UNAUTHORIZED
response_json = json.loads(s=response.text)
assert (
response_json["result_code"]
== ResultCodes.AUTHENTICATION_FAILURE.value
)


@pytest.mark.usefixtures("mock_only_vuforia")
class TestDownloadReport:
Expand All @@ -195,7 +204,7 @@ def test_download_report(*, vuforia_database: CloudDatabase) -> None:
"""The report is available from the given URL once it is ready."""
response = _request_reco_counts_report(
vuforia_database=vuforia_database,
database_id=uuid.uuid4().hex,
database_id=vuforia_database.database_id,
month=_month_offset_from_now(months=0),
)
presigned_url = json.loads(s=response.text)["presigned_url"]
Expand Down