diff --git a/admin/create_secrets_files.py b/admin/create_secrets_files.py index b96759997..4ad3e3cd8 100644 --- a/admin/create_secrets_files.py +++ b/admin/create_secrets_files.py @@ -22,6 +22,13 @@ name="vumark_template.svg", ) +MODEL_TARGET_WEB_API_SCOPES = ( + "modeltargets.standardmodeltarget.all", + "modeltargets.advancedmodeltarget.all", + "modeltargets.statebasedmodeltarget.all", + "modeltargets.advancedstatebasedmodeltarget.all", +) + def _create_and_get_cloud_database_details( driver: WebDriver, @@ -83,6 +90,8 @@ def _generate_secrets_file_content( inactive_vumark_details: VuMarkDatabaseDict, vumark_target_id: str, model_target_web_api_details: ModelTargetWebAPIDict, + model_target_username: str, + model_target_password: str, ) -> str: """Generate the content of a secrets file.""" return textwrap.dedent( @@ -112,6 +121,8 @@ def _generate_secrets_file_content( MODEL_TARGET_VUFORIA_CLIENT_ID={model_target_web_api_details["client_id"]} MODEL_TARGET_VUFORIA_CLIENT_SECRET={model_target_web_api_details["client_secret"]} MODEL_TARGET_VUFORIA_CAD_DATA_URL={model_target_web_api_details["cad_data_url"]} + MODEL_TARGET_VUFORIA_USERNAME={model_target_username} + MODEL_TARGET_VUFORIA_PASSWORD={model_target_password} """, ) @@ -216,7 +227,10 @@ def _get_model_target_web_api_details( password=password, ) vws_web_tools.wait_for_logged_in(driver=driver) - return vws_web_tools.get_model_target_web_api_details(driver=driver) + return vws_web_tools.get_model_target_web_api_details( + driver=driver, + scopes=MODEL_TARGET_WEB_API_SCOPES, + ) def _create_vuforia_resource_names() -> tuple[str, str, str, str]: @@ -326,6 +340,8 @@ def main() -> None: inactive_vumark_details=inactive_vumark_details, vumark_target_id=vumark_target_id, model_target_web_api_details=model_target_web_api_details, + model_target_username=email_address, + model_target_password=password, ) file.write_text(data=file_contents) sys.stdout.write(f"Created database {file.name}\n") diff --git a/docs/source/differences-to-vws.rst b/docs/source/differences-to-vws.rst index 3d2b22ec2..7bb8390e0 100644 --- a/docs/source/differences-to-vws.rst +++ b/docs/source/differences-to-vws.rst @@ -228,7 +228,10 @@ 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. +The generated dataset download is a small valid ``full-dataset.zip`` with the +same ``MTDataset.dat`` and ``MTDataset.xml`` filenames as Vuforia. Its contents +are synthetic request metadata and minimal XML, not a real Vuforia Engine +Model Target dataset. Use :paramref:`mock_vws.MockVWS.model_target_generation_failure` to make in-process Model Target datasets finish with a ``failed`` status and an ``error`` object. The failure is returned after the configured @@ -245,14 +248,20 @@ base64url-encoded signature, such as the token returned by the mock OAuth2 route. The mock does not verify token signatures, payload claims such as expiry, or token revocation. +The OAuth2 route supports both the ``client_credentials`` and ``password`` +grants. Tokens returned by the mock contain explicit scopes, and +standard and advanced dataset routes require their corresponding Model Target +scope. A token carrying ``modeltargets.all`` can access both route families. +The OAuth2 client-credentials management routes support creating, listing, +updating and deleting credentials, including Vuforia's limit of 100 created +credentials per account. 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. +which does not name a grant type. 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. @@ -272,7 +281,10 @@ advanced datasets; the OpenAPI specification does not document it as a standard dataset model field, so standard dataset creation does not validate it. Each ``views`` entry is validated for being a JSON object, for the required -``guideViewPosition`` and ``name`` fields, and for those fields' types. +``name`` field, and for the types of ``name`` and the optional +``guideViewPosition`` field. +State-Based Model Target views require ``guideViewPosition``, matching real +Vuforia. An optional ``states`` field must be an array of strings. Each named state must be declared by the model's ``stateBasedConfigurationJsonString``. Omitting the field makes the view available to every configured state. @@ -290,21 +302,19 @@ It also does not validate the state configuration beyond its top-level For unknown Model Target datasets, the mock returns an error whose ``target`` is ``userId:mock``. Real Vuforia uses ``userId:`` where the numeric portion is per-account. -Standard and advanced datasets are separate resources. -A dataset created through the standard routes is not visible to the advanced routes, and the other way around: the mock returns the unknown-dataset error for status, download and delete requests made through the other dataset type's routes. -Real Vuforia separates these by OAuth scope as well, which the mock does not model, so a client which lacks the advanced-dataset scope may see a different error. +Standard and advanced routes share datasets by UUID. Access to each route +family is separated by its corresponding OAuth scope. -Some Model Target Web API paths remain mock-only in ``tests/mock_vws/test_model_target_web_api.py::TestMockOnlyErrors``. +Some Model Target Web API paths remain mock-only in +``tests/mock_vws/test_model_target_web_api.py::TestAdditionalBehaviors``. Downloads of still-processing datasets are mock-only because exercising the path against real Vuforia would require creating a dataset on every test run; the mock drives the processing window deterministically. A download request for a dataset which is not ready reports the dataset's training status. The mock reports ``not-started`` for the whole processing window, as real Vuforia does for a dataset which was just created, and ``failed`` for a dataset whose generation failed. The name which real Vuforia reports for a failed dataset has not been observed. -Advanced-dataset creation with more than 20 models is mock-only because the available test account lacks the advanced-dataset scope and real Vuforia rejects the request with a 403 before validating model counts. -Cross-dataset-type access is mock-only for the same reason. -State-Based Model Target creation and validation are also mock-only because the -available test account lacks the State-Based Model Target scopes. +Some malformed State-Based Model Target configuration documents remain +mock-only because real Vuforia returns an internal server error for them. Reco counts reports ------------------- diff --git a/secrets.tar.gpg b/secrets.tar.gpg index 3dd97f72b..a9b4a2db4 100644 Binary files a/secrets.tar.gpg and b/secrets.tar.gpg differ diff --git a/src/mock_vws/_flask_server/target_manager.py b/src/mock_vws/_flask_server/target_manager.py index 90e973477..6814c1164 100644 --- a/src/mock_vws/_flask_server/target_manager.py +++ b/src/mock_vws/_flask_server/target_manager.py @@ -15,7 +15,7 @@ from mock_vws.database import CloudDatabase, VuMarkDatabase from mock_vws.database_type import DatabaseType -from mock_vws.model_target import ModelTargetDataset +from mock_vws.model_target import ModelTargetDataset, OAuth2ClientCredential from mock_vws.request_rate_limits import RequestRateLimits from mock_vws.states import States from mock_vws.target import ImageTarget, VuMarkTarget @@ -398,6 +398,57 @@ def delete_model_target_dataset(dataset_uuid: str) -> Response: return Response(response="", status=HTTPStatus.OK) +@TARGET_MANAGER_FLASK_APP.route( + rule="/oauth2_client_credentials", + methods=[HTTPMethod.GET], +) +@beartype +def get_oauth2_client_credentials() -> Response: + """Return all OAuth2 client credentials.""" + credentials = [ + { + "client_id": credential.client_id, + "client_secret": credential.client_secret, + "scopes": list(credential.scopes), + } + for credential in TARGET_MANAGER.oauth2_client_credentials.values() + ] + return Response( + response=json.dumps(obj=credentials), + status=HTTPStatus.OK, + ) + + +@TARGET_MANAGER_FLASK_APP.route( + rule="/oauth2_client_credentials", + methods=[HTTPMethod.POST], +) +@beartype +def put_oauth2_client_credential() -> Response: + """Add or replace an OAuth2 client credential.""" + value = json.loads(s=request.data) + credential = OAuth2ClientCredential( + client_id=value["client_id"], + client_secret=value["client_secret"], + scopes=tuple(value["scopes"]), + ) + TARGET_MANAGER.add_oauth2_client_credential(credential=credential) + return Response(response="", status=HTTPStatus.NO_CONTENT) + + +@TARGET_MANAGER_FLASK_APP.route( + rule="/oauth2_client_credentials/", + methods=[HTTPMethod.DELETE], +) +@beartype +def remove_oauth2_client_credential(client_id: str) -> Response: + """Remove an OAuth2 client credential.""" + if client_id not in TARGET_MANAGER.oauth2_client_credentials: + return Response(response="", status=HTTPStatus.NOT_FOUND) + TARGET_MANAGER.remove_oauth2_client_credential(client_id=client_id) + return Response(response="", status=HTTPStatus.NO_CONTENT) + + @TARGET_MANAGER_FLASK_APP.route( rule="/cloud_databases//targets", methods=[HTTPMethod.POST], diff --git a/src/mock_vws/_flask_server/vws.py b/src/mock_vws/_flask_server/vws.py index db7da0c61..4be204dbb 100644 --- a/src/mock_vws/_flask_server/vws.py +++ b/src/mock_vws/_flask_server/vws.py @@ -35,9 +35,21 @@ download_model_target_dataset, get_model_target_dataset_status, ) +from mock_vws._model_target_web_api import ( + create_oauth2_client_credential as model_target_create_oauth2_credential, +) +from mock_vws._model_target_web_api import ( + delete_oauth2_client_credential as model_target_delete_oauth2_credential, +) +from mock_vws._model_target_web_api import ( + list_oauth2_client_credentials as model_target_list_oauth2_credentials, +) from mock_vws._model_target_web_api import ( oauth2_token as model_target_oauth2_token, ) +from mock_vws._model_target_web_api import ( + update_oauth2_client_credential_scopes as update_oauth2_scopes, +) from mock_vws._reco_counts_web_api import create_reco_counts_report from mock_vws._reco_counts_web_api import ( download_reco_counts_report as download_report, @@ -60,7 +72,11 @@ ImageMatcher, StructuralSimilarityMatcher, ) -from mock_vws.model_target import ModelTargetDataset, ModelTargetDatasetType +from mock_vws.model_target import ( + ModelTargetDataset, + ModelTargetDatasetType, + OAuth2ClientCredential, +) from mock_vws.reco_counts import RecoCountsReport from mock_vws.target import ImageTarget from mock_vws.target_raters import ( @@ -167,6 +183,7 @@ def __init__(self, *, base_url: str) -> None: base_url: The base URL of the target manager service. """ self._datasets_url = f"{base_url}/model_target_datasets" + self._credentials_url = f"{base_url}/oauth2_client_credentials" @property def model_target_datasets(self) -> dict[str, ModelTargetDataset]: @@ -202,6 +219,42 @@ def remove_model_target_dataset(self, dataset_uuid: str) -> None: timeout=timeout_seconds, ) + @property + def oauth2_client_credentials(self) -> dict[str, OAuth2ClientCredential]: + """All dynamically created OAuth2 client credentials.""" + response = requests.get(url=self._credentials_url, timeout=30) + credentials = ( + OAuth2ClientCredential( + client_id=value["client_id"], + client_secret=value["client_secret"], + scopes=tuple(value["scopes"]), + ) + for value in response.json() + ) + return {credential.client_id: credential for credential in credentials} + + def add_oauth2_client_credential( + self, + credential: OAuth2ClientCredential, + ) -> None: + """Add or replace an OAuth2 client credential.""" + requests.post( + url=self._credentials_url, + json={ + "client_id": credential.client_id, + "client_secret": credential.client_secret, + "scopes": list(credential.scopes), + }, + timeout=30, + ) + + def remove_oauth2_client_credential(self, client_id: str) -> None: + """Remove an OAuth2 client credential.""" + requests.delete( + url=f"{self._credentials_url}/{client_id}", + timeout=30, + ) + @beartype def _model_target_dataset_store() -> _HTTPModelTargetDatasetStore: @@ -301,7 +354,7 @@ def validate_request() -> None: if request.endpoint == "generate_vumark_instance": return if ( - request.path == "/oauth2/token" + request.path.startswith("/oauth2/") or request.path.startswith("/modeltargets/") or request.path.startswith("/reports/recoCounts/") ): @@ -364,6 +417,69 @@ def oauth2_token() -> Response: return _to_flask_response( api_response=model_target_oauth2_token( request=_flask_request_data(), + credential_store=_model_target_dataset_store(), + ), + ) + + +@VWS_FLASK_APP.route( + rule="/oauth2/clientcredentials", + methods=[HTTPMethod.POST], +) +@beartype +def create_oauth2_client_credential() -> Response: + """Create an OAuth2 client credential.""" + return _to_flask_response( + api_response=model_target_create_oauth2_credential( + request=_flask_request_data(), + credential_store=_model_target_dataset_store(), + ), + ) + + +@VWS_FLASK_APP.route( + rule="/oauth2/clientcredentials", + methods=[HTTPMethod.GET], +) +@beartype +def list_oauth2_client_credentials() -> Response: + """List OAuth2 client credentials.""" + return _to_flask_response( + api_response=model_target_list_oauth2_credentials( + request=_flask_request_data(), + credential_store=_model_target_dataset_store(), + ), + ) + + +@VWS_FLASK_APP.route( + rule="/oauth2/clientcredentials//scopes", + methods=[HTTPMethod.PUT], +) +@beartype +def update_oauth2_client_credential_scopes(client_id: str) -> Response: + """Update an OAuth2 client credential's scopes.""" + return _to_flask_response( + api_response=update_oauth2_scopes( + request=_flask_request_data(), + credential_store=_model_target_dataset_store(), + client_id=client_id, + ), + ) + + +@VWS_FLASK_APP.route( + rule="/oauth2/clientcredentials/", + methods=[HTTPMethod.DELETE], +) +@beartype +def delete_oauth2_client_credential(client_id: str) -> Response: + """Delete an OAuth2 client credential.""" + return _to_flask_response( + api_response=model_target_delete_oauth2_credential( + request=_flask_request_data(), + credential_store=_model_target_dataset_store(), + client_id=client_id, ), ) diff --git a/src/mock_vws/_model_target_web_api.py b/src/mock_vws/_model_target_web_api.py index b097a6162..c869fef82 100644 --- a/src/mock_vws/_model_target_web_api.py +++ b/src/mock_vws/_model_target_web_api.py @@ -3,6 +3,7 @@ import base64 import io import json +import secrets import uuid import zipfile from http import HTTPStatus @@ -20,6 +21,7 @@ ModelTargetDatasetType, ModelTargetGenerationFailure, ModelTargetGenerationWarning, + OAuth2ClientCredential, ) _ResponseType = tuple[int, dict[str, str], str | bytes] @@ -51,12 +53,41 @@ def remove_model_target_dataset(self, dataset_uuid: str) -> None: # for pyright to recognize this as a protocol. ... # pylint: disable=unnecessary-ellipsis + @property + def oauth2_client_credentials(self) -> dict[str, OAuth2ClientCredential]: + """All dynamically created OAuth2 client credentials.""" + ... # pylint: disable=unnecessary-ellipsis + + def add_oauth2_client_credential( + self, + credential: OAuth2ClientCredential, + ) -> None: + """Add an OAuth2 client credential.""" + ... # pylint: disable=unnecessary-ellipsis + + def remove_oauth2_client_credential(self, client_id: str) -> None: + """Remove an OAuth2 client credential.""" + ... # pylint: disable=unnecessary-ellipsis + _MAX_ADVANCED_MODEL_COUNT = 20 _JWT_DOT_COUNT = 2 _ZIP_EPOCH = (1980, 1, 1, 0, 0, 0) _MOCK_MODEL_TARGET_CLIENT_ID = "client-id" _MOCK_MODEL_TARGET_CLIENT_SECRET = "client-secret" # noqa: S105 +_MOCK_MODEL_TARGET_USERNAME = "user@example.com" +_MOCK_MODEL_TARGET_PASSWORD = "password" # noqa: S105 +_MODEL_TARGET_SCOPES = frozenset( + { + "modeltargets.all", + "modeltargets.standardmodeltarget.all", + "modeltargets.advancedmodeltarget.all", + "modeltargets.statebasedmodeltarget.all", + "modeltargets.advancedstatebasedmodeltarget.all", + }, +) +_CLIENT_CREDENTIALS_SCOPE = "oauth2.clientcredentials.all" +_MAX_CLIENT_CREDENTIALS = 100 # A stable mock value standing in for the user-id segment that real # Vuforia embeds in some Model Target error targets such as # ``userId:7635391``. The numeric portion is per-account in real Vuforia; @@ -69,10 +100,13 @@ def remove_model_target_dataset(self, dataset_uuid: str) -> None: "cadDataFormat": frozenset( { "DAE", + "DRC_GLB", + "DRC_GLTF", "FBX", "GLB", "IGES", "OBJ", + "PVS", "PVZ", "STL", "VRML", @@ -298,7 +332,28 @@ def _jwt_signature_error(*, bearer_token: str) -> str | None: @beartype -def _require_bearer_token(request: RequestData) -> _ResponseType | None: +def _jwt_scopes(*, bearer_token: str) -> frozenset[str]: + """Return scopes from a valid mock JSON Web Token.""" + encoded_payload = bearer_token.split(sep=".")[1] + padding = "=" * (-len(encoded_payload) % 4) + payload = json.loads( + s=base64.b64decode( + s=encoded_payload + padding, + altchars=b"-_", + validate=True, + ), + ) + scope = payload.get("scope", "") + if not isinstance(scope, str): + return frozenset() + return frozenset(scope.split()) + + +@beartype +def _require_bearer_token( + request: RequestData, + dataset_type: ModelTargetDatasetType, +) -> _ResponseType | None: """Return an error response if the request has no bearer token.""" auth_header = _get_header(request=request, name="Authorization") if auth_header is None or not auth_header.startswith("Bearer "): @@ -339,11 +394,53 @@ def _require_bearer_token(request: RequestData) -> _ResponseType | None: target="jwt", details=None, ) + required_scope = ( + "modeltargets.standardmodeltarget.all" + if dataset_type == ModelTargetDatasetType.STANDARD + else "modeltargets.advancedmodeltarget.all" + ) + scopes = _jwt_scopes(bearer_token=bearer_token) + if required_scope not in scopes and "modeltargets.all" not in scopes: + body = "User does not have the required scopes to perform this action" + return ( + HTTPStatus.FORBIDDEN, + { + "Content-Length": str(object=len(body)), + "Content-Type": "text/plain", + }, + body, + ) return None @beartype -def _fake_jwt(*, token_source: bytes) -> str: +def _require_state_based_scope( + request: RequestData, + dataset_type: ModelTargetDatasetType, +) -> _ResponseType | None: + """Return an error when a token lacks the State-Based MT scope.""" + auth_header = _get_header(request=request, name="Authorization") + assert auth_header is not None # noqa: S101 + bearer_token = auth_header.removeprefix("Bearer ").strip() + required_scope = ( + "modeltargets.statebasedmodeltarget.all" + if dataset_type == ModelTargetDatasetType.STANDARD + else "modeltargets.advancedstatebasedmodeltarget.all" + ) + scopes = _jwt_scopes(bearer_token=bearer_token) + if required_scope in scopes or "modeltargets.all" in scopes: + return None + return _error_response( + status_code=HTTPStatus.FORBIDDEN, + code="ERROR", + message="User not allowed to create State Based Model Targets", + target=_MOCK_USER_TARGET, + details=None, + ) + + +@beartype +def _fake_jwt(*, token_source: bytes, scopes: frozenset[str]) -> str: """Return a deterministic bearer token for the mock.""" def encode_part(value: dict[str, Any]) -> str: @@ -370,13 +467,18 @@ def encode_part(value: dict[str, Any]) -> str: encoding="ascii", ) .rstrip("="), + "scope": " ".join(sorted(scopes)), }, ) return f"{header}.{payload}.mock-signature" @beartype -def oauth2_token(request: RequestData) -> _ResponseType: +def oauth2_token( # noqa: PLR0911 # pylint: disable=too-many-return-statements + *, + request: RequestData, + credential_store: ModelTargetDatasetStore, +) -> _ResponseType: """Return a fake OAuth2 access token.""" content_length_error = _content_length_error(request=request) if content_length_error is not None: @@ -390,44 +492,300 @@ def oauth2_token(request: RequestData) -> _ResponseType: qs=request.body.decode(encoding="utf-8", errors="replace"), ) grant_type = form.get("grant_type", ["client_credentials"])[0] - if grant_type != "client_credentials": + if grant_type not in {"client_credentials", "password"}: return _oauth2_error_response( status_code=HTTPStatus.BAD_REQUEST, 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" - ), - }, - ) + dynamic_credential: OAuth2ClientCredential | None = None + if grant_type == "client_credentials": + 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"}, + dynamic_credential = credential_store.oauth2_client_credentials.get( + basic_credentials[0], ) + fixed_credential_matches = basic_credentials == ( + _MOCK_MODEL_TARGET_CLIENT_ID, + _MOCK_MODEL_TARGET_CLIENT_SECRET, + ) + dynamic_credential_matches = ( + dynamic_credential is not None + and dynamic_credential.client_secret == basic_credentials[1] + ) + if not fixed_credential_matches and not dynamic_credential_matches: + return _oauth2_error_response( + status_code=HTTPStatus.UNAUTHORIZED, + body={"error": "invalid_client"}, + ) + else: + username = form.get("username", [""])[0] + password = form.get("password", [""])[0] + if not username or not password: + return _oauth2_error_response( + status_code=HTTPStatus.BAD_REQUEST, + body={ + "error": "invalid_request", + "error_description": "Missing username and/or password", + }, + ) + if (username, password) != ( + _MOCK_MODEL_TARGET_USERNAME, + _MOCK_MODEL_TARGET_PASSWORD, + ): + return _oauth2_error_response( + status_code=HTTPStatus.UNAUTHORIZED, + body={ + "error": "invalid_grant", + "error_description": "Invalid username and/or password", + }, + ) token_source = request.body or (auth_header or "").encode() + requested_scope = form.get("scope", [""])[0] + if grant_type == "client_credentials" and dynamic_credential is not None: + credential_scopes = frozenset(dynamic_credential.scopes) + else: + credential_scopes = _MODEL_TARGET_SCOPES | { + _CLIENT_CREDENTIALS_SCOPE, + } + scopes = frozenset(requested_scope.split()) or credential_scopes + if not scopes.issubset(credential_scopes): + return _oauth2_error_response( + status_code=HTTPStatus.BAD_REQUEST, + body={"error": "invalid_scope"}, + ) return _json_response( status_code=HTTPStatus.OK, body={ - "access_token": _fake_jwt(token_source=token_source), + "access_token": _fake_jwt( + token_source=token_source, + scopes=scopes, + ), "token_type": "bearer", "expires_in": 3600, }, ) +@beartype +def _require_client_credentials_scope( + request: RequestData, +) -> _ResponseType | None: + """Require a valid bearer token with the credential-management + scope. + """ + auth_header = _get_header(request=request, name="Authorization") + if auth_header is None or not auth_header.startswith("Bearer "): + return _error_response( + status_code=HTTPStatus.UNAUTHORIZED, + code="401", + message="no Bearer token", + target="jwt", + details=None, + ) + bearer_token = auth_header.removeprefix("Bearer ").strip() + if bearer_token.count(".") != _JWT_DOT_COUNT: + return _error_response( + status_code=HTTPStatus.UNAUTHORIZED, + code="401", + message="Invalid JWT serialization: Missing dot delimiter(s)", + target="jwt", + details=None, + ) + jwt_error = ( + _jwt_header_error(bearer_token=bearer_token) + or _jwt_payload_error(bearer_token=bearer_token) + or _jwt_signature_error(bearer_token=bearer_token) + ) + if jwt_error is not None: + return _error_response( + status_code=HTTPStatus.UNAUTHORIZED, + code="401", + message=jwt_error, + target="jwt", + details=None, + ) + if _CLIENT_CREDENTIALS_SCOPE not in _jwt_scopes( + bearer_token=bearer_token, + ): + body = "User does not have the required scopes to perform this action" + return ( + HTTPStatus.FORBIDDEN, + { + "Content-Length": str(object=len(body)), + "Content-Type": "text/plain", + }, + body, + ) + return None + + +@beartype +def _client_credential_not_found(*, client_id: str) -> _ResponseType: + """Return Vuforia's missing-client-credential response.""" + return _error_response( + status_code=HTTPStatus.NOT_FOUND, + code="NOT_FOUND", + message=f"Clientcredential with ID={client_id} not found", + target="clientcredential", + details=None, + ) + + +@beartype +def _string_list(value: object) -> list[str] | None: + """Return a string list when ``value`` contains only strings.""" + if not isinstance(value, list): + return None + strings: list[str] = [] + for item in value: # pyright: ignore[reportUnknownVariableType] + if not isinstance(item, str): + return None + strings.append(item) + return strings + + +@beartype +def create_oauth2_client_credential( + *, + request: RequestData, + credential_store: ModelTargetDatasetStore, +) -> _ResponseType: + """Create an OAuth2 client credential.""" + auth_error = _require_client_credentials_scope(request=request) + if auth_error is not None: + return auth_error + request_json_or_error = _load_request_json(request=request) + if not isinstance(request_json_or_error, dict): + return request_json_or_error + scopes = _string_list(value=request_json_or_error.get("scopes")) + if scopes is None: + return _validation_error_response( + details=[ + { + "code": "VALIDATION_ERROR", + "message": "/scopes: error.expected.jsarray", + }, + ], + ) + if len(credential_store.oauth2_client_credentials) >= ( + _MAX_CLIENT_CREDENTIALS + ): + return _error_response( + status_code=HTTPStatus.CONFLICT, + code="CONFLICT", + message="Maximum number of client credentials reached", + target="clientcredential", + details=None, + ) + client_id = uuid.uuid4().hex.upper()[:21] + client_secret = secrets.token_urlsafe(nbytes=25)[:33] + credential_store.add_oauth2_client_credential( + credential=OAuth2ClientCredential( + client_id=client_id, + client_secret=client_secret, + scopes=tuple(scopes), + ), + ) + return _json_response( + status_code=HTTPStatus.CREATED, + body={"client_id": client_id, "client_secret": client_secret}, + ) + + +@beartype +def list_oauth2_client_credentials( + *, + request: RequestData, + credential_store: ModelTargetDatasetStore, +) -> _ResponseType: + """List OAuth2 client credentials.""" + auth_error = _require_client_credentials_scope(request=request) + if auth_error is not None: + return auth_error + credentials = [ + {"clientId": credential.client_id, "scopes": list(credential.scopes)} + for credential in credential_store.oauth2_client_credentials.values() + ] + body = json.dumps(obj=credentials, separators=(",", ":")) + return ( + HTTPStatus.OK, + { + "Content-Length": str(object=len(body)), + "Content-Type": "application/json", + }, + body, + ) + + +@beartype +def update_oauth2_client_credential_scopes( + *, + request: RequestData, + credential_store: ModelTargetDatasetStore, + client_id: str, +) -> _ResponseType: + """Replace the scopes assigned to an OAuth2 client credential.""" + auth_error = _require_client_credentials_scope(request=request) + if auth_error is not None: + return auth_error + credential = credential_store.oauth2_client_credentials.get(client_id) + if credential is None: + return _client_credential_not_found(client_id=client_id) + try: + scopes_value: object = json.loads(s=request.body) + except UnicodeDecodeError, json.JSONDecodeError: + scopes_value = None + scopes = _string_list(value=scopes_value) + if scopes is None: + return _error_response( + status_code=HTTPStatus.BAD_REQUEST, + code="BAD_REQUEST", + message="Invalid scopes", + target="scopes", + details=None, + ) + credential_store.add_oauth2_client_credential( + credential=OAuth2ClientCredential( + client_id=credential.client_id, + client_secret=credential.client_secret, + scopes=tuple(scopes), + ), + ) + return _json_response( + status_code=HTTPStatus.OK, + body={"clientId": client_id, "scopes": scopes}, + ) + + +@beartype +def delete_oauth2_client_credential( + *, + request: RequestData, + credential_store: ModelTargetDatasetStore, + client_id: str, +) -> _ResponseType: + """Delete an OAuth2 client credential.""" + auth_error = _require_client_credentials_scope(request=request) + if auth_error is not None: + return auth_error + if client_id not in credential_store.oauth2_client_credentials: + return _client_credential_not_found(client_id=client_id) + credential_store.remove_oauth2_client_credential(client_id=client_id) + return HTTPStatus.NO_CONTENT, {"Content-Length": "0"}, "" + + @beartype def _is_json_object(*, value: object) -> bool: """Return whether a decoded JSON value is an object.""" @@ -472,19 +830,32 @@ def _load_request_json(request: RequestData) -> dict[str, Any] | _ResponseType: def _cad_data_source_details(*, models: list[Any]) -> list[dict[str, str]]: """Return validation details for each model's CAD data source. - One and only one of ``cadDataUrl`` and ``cadDataBlob`` may be given per - model. + One and only one of ``cadDataUrl``, ``cadDataBlob`` and ``cadDataUuid`` + may be given per model. """ return [ { "code": "VALIDATION_ERROR", "message": ( - f"/models({index}): one and only one of cadDataUrl and " - "cadDataBlob is required" + f"model '{model['name']}' is invalid. " + + ( + "One of `cadDataBlob`, `cadDataUrl`, `cadDataUuid` need " + "to be provided" + if not sources + else "Only one of `cadDataBlob`, `cadDataUrl`, " + "`cadDataUuid` need to be provided" + ) ), } - for index, model in enumerate(iterable=models) - if ("cadDataUrl" in model) == ("cadDataBlob" in model) + for model in models + if len( + sources := { + field + for field in ("cadDataBlob", "cadDataUrl", "cadDataUuid") + if field in model + }, + ) + != 1 ] @@ -503,14 +874,18 @@ def _model_field_details( missing_details = [ { "code": "VALIDATION_ERROR", - "message": f"/models({index})/name: element is required", + "message": f"/models({index})/{field}: element is required", } for index, model in enumerate(iterable=models) - if "name" not in model + for field in ("name", "views") + if field not in model ] + if missing_details: + return missing_details + cad_data_source_details = _cad_data_source_details(models=models) - if missing_details or cad_data_source_details: - return missing_details + cad_data_source_details + if cad_data_source_details: + return cad_data_source_details string_fields = sorted( { @@ -533,15 +908,57 @@ def _model_field_details( if string_details: return string_details - enum_details = [ - { - "code": "VALIDATION_ERROR", - "message": f"/models({index})/{field}: error.expected.validenum", - } - for index, model in enumerate(iterable=models) - for field, allowed_values in sorted(enum_field_values.items()) - if field in model and model[field] not in allowed_values - ] + enum_details: list[dict[str, str]] = [] + for index, model in enumerate(iterable=models): + for field, allowed_values in sorted(enum_field_values.items()): + if field in {"motionHint", "trackingMode"}: + continue + if field not in model or model[field] in allowed_values: + continue + value = model[field] + messages = { + "automaticColoring": ( + "invalid automaticColoring. Should be one of 'never', " + f"'always', 'auto'. You provided '{value}'" + ), + "cadDataFormat": ( + "Unrecognized cadDataFormat '" + f"{str(object=value).upper()}'. " + "Allowed values are: ZIP, GLB, DRC_GLB, DRC_GLTF, DAE, " + "FBX, IGES, OBJ, PVS, PVZ, STL, VRML, or specify no " + "cadDataFormat to auto-detect GLB and zipped glTFs." + ), + "optimizeTrackingFor": ( + "`optimizeTrackingFor` must be one of " + "default,low_feature_objects,ar_controller" + ), + "realisticAppearance": ( + '`realisticAppearance` must be one of "true", "false", ' + '"auto".` ' + ), + "simplify": ( + "invalid simplify. Should be one of 'never', 'always', " + f"'auto'. You provided 'Some({value})'" + ), + } + message = messages.get( + field, + f"/models({index})/{field}: error.expected.validenum", + ) + enum_details.append( + {"code": "VALIDATION_ERROR", "message": message}, + ) + if "motionHint" in model or "trackingMode" in model: + enum_details.append( + { + "code": "VALIDATION_ERROR", + "message": ( + "`motionHint` and `trackingMode` are no longer " + "supported when using `targetsSdk` 10.9 or later. " + "Please use the `optimizeTrackingFor` setting instead." + ), + }, + ) views_details = [ { "code": "VALIDATION_ERROR", @@ -585,7 +1002,7 @@ def _view_details(*, models: list[Any]) -> list[dict[str, str]]: ), } for model_index, view_index, view in views - for field in ("guideViewPosition", "name") + for field in ("name",) if field not in view ] if missing_details: @@ -611,7 +1028,8 @@ def _view_details(*, models: list[Any]) -> list[dict[str, str]]: ), } for model_index, view_index, view in views - if not isinstance(view["guideViewPosition"], dict) + if "guideViewPosition" in view + and not isinstance(view["guideViewPosition"], dict) ] return name_details + position_details @@ -636,6 +1054,7 @@ def _guide_view_position_details( (model_index, view_index, view["guideViewPosition"]) for model_index, model in enumerate(iterable=models) for view_index, view in enumerate(iterable=model.get("views", [])) + if "guideViewPosition" in view ] missing_details = [ @@ -796,11 +1215,12 @@ def _state_based_details(*, models: list[Any]) -> list[dict[str, str]]: { "code": "VALIDATION_ERROR", "message": ( - f"/models({model_index})/views({view_index})/states" - f"({state_index}): error.expected.validenum" + "states in entrypoint " + f"{models[model_index]['views'][view_index]['name']}' " + "must be a subset of all states" ), } - for state_index, state in enumerate(iterable=states) + for state in states if state not in configured_states[model_index] ) @@ -824,19 +1244,36 @@ def _model_count_details( }, ] - if ( - dataset_type == ModelTargetDatasetType.ADVANCED - and not 1 <= model_count <= _MAX_ADVANCED_MODEL_COUNT - ): - return [ - { - "code": "VALIDATION_ERROR", - "message": ( - "models must contain between 1 and " - f"{_MAX_ADVANCED_MODEL_COUNT} entries" - ), - }, - ] + if dataset_type == ModelTargetDatasetType.ADVANCED: + details: list[dict[str, str]] = [] + names = [model["name"] for model in models] + if len(set(names)) != len(names): + details.append( + { + "code": "VALIDATION_ERROR", + "message": ( + "names of models must be unique within a Target." + ), + }, + ) + if model_count > _MAX_ADVANCED_MODEL_COUNT: + details.append( + { + "code": "VALIDATION_ERROR", + "message": ( + "total number of models must be maximum " + f"{_MAX_ADVANCED_MODEL_COUNT}" + ), + }, + ) + if model_count == 0: + details.append( + { + "code": "VALIDATION_ERROR", + "message": "models must contain at least one entry", + }, + ) + return details return [] @@ -877,15 +1314,6 @@ def _top_level_details( ) return type_details - models: list[Any] = [*models_value] - type_details.extend( - { - "code": "VALIDATION_ERROR", - "message": f"/models({index}): error.expected.jsobject", - } - for index, model in enumerate(iterable=models) - if not isinstance(model, dict) - ) return type_details @@ -898,7 +1326,18 @@ def _validate_dataset_request( """Validate the dataset request enough for useful mock feedback.""" details = _top_level_details(request_json=request_json) if not details: - models: list[Any] = [*request_json["models"]] + # Vuforia's schema validator reads fields from non-object model and + # view values as though they were empty objects. + models: list[Any] = [ + model if isinstance(model, dict) else {} + for model in request_json["models"] + ] + for model in models: + if isinstance(model.get("views"), list): + model["views"] = [ + view if isinstance(view, dict) else {} + for view in model["views"] + ] details = ( _model_field_details(models=models, dataset_type=dataset_type) or _view_details(models=models) @@ -931,7 +1370,10 @@ def create_model_target_dataset( if content_length_error is not None: return content_length_error - auth_error = _require_bearer_token(request=request) + auth_error = _require_bearer_token( + request=request, + dataset_type=dataset_type, + ) if auth_error is not None: return auth_error @@ -939,6 +1381,20 @@ def create_model_target_dataset( if not isinstance(request_json_or_error, dict): return request_json_or_error + models_value = request_json_or_error.get("models") + is_state_based = isinstance(models_value, list) and any( + isinstance(model, dict) + and "stateBasedConfigurationJsonString" in model + for model in models_value # pyright: ignore[reportUnknownVariableType] + ) + if is_state_based: + state_scope_error = _require_state_based_scope( + request=request, + dataset_type=dataset_type, + ) + if state_scope_error is not None: + return state_scope_error + validation_error = _validate_dataset_request( request_json=request_json_or_error, dataset_type=dataset_type, @@ -979,17 +1435,9 @@ def _find_dataset( *, dataset_store: ModelTargetDatasetStore, dataset_uuid: str, - dataset_type: ModelTargetDatasetType, ) -> ModelTargetDataset | None: - """Return a dataset which belongs to a route's dataset type. - - Standard and advanced datasets are separate resources in real Vuforia, so - a dataset is invisible to the routes of the other dataset type. - """ - dataset = dataset_store.model_target_datasets.get(dataset_uuid) - if dataset is None or dataset.dataset_type != dataset_type: - return None - return dataset + """Return a Model Target dataset by UUID.""" + return dataset_store.model_target_datasets.get(dataset_uuid) @beartype @@ -1005,13 +1453,15 @@ def get_model_target_dataset_status( if content_length_error is not None: return content_length_error - auth_error = _require_bearer_token(request=request) + auth_error = _require_bearer_token( + request=request, + dataset_type=dataset_type, + ) if auth_error is not None: return auth_error dataset = _find_dataset( dataset_store=dataset_store, dataset_uuid=dataset_uuid, - dataset_type=dataset_type, ) if dataset is None: return _unknown_dataset_response(dataset_uuid=dataset_uuid) @@ -1023,15 +1473,15 @@ def get_model_target_dataset_status( @beartype def _dataset_zip_bytes(dataset: ModelTargetDataset) -> bytes: - """Return a small valid zip file for a generated dataset.""" + """Return a deterministic Vuforia-shaped generated dataset zip.""" zip_buffer = io.BytesIO() with zipfile.ZipFile(file=zip_buffer, mode="w") as zip_file: - dataset_file = zipfile.ZipInfo( - filename="dataset.json", + dat_file = zipfile.ZipInfo( + filename="MTDataset.dat", date_time=_ZIP_EPOCH, ) zip_file.writestr( - zinfo_or_arcname=dataset_file, + zinfo_or_arcname=dat_file, data=json.dumps( obj={ "uuid": dataset.uuid_, @@ -1042,6 +1492,18 @@ def _dataset_zip_bytes(dataset: ModelTargetDataset) -> bytes: sort_keys=True, ), ) + xml_file = zipfile.ZipInfo( + filename="MTDataset.xml", + date_time=_ZIP_EPOCH, + ) + zip_file.writestr( + zinfo_or_arcname=xml_file, + data=( + '' + '' + "" + ), + ) return zip_buffer.getvalue() @@ -1058,13 +1520,15 @@ def download_model_target_dataset( if content_length_error is not None: return content_length_error - auth_error = _require_bearer_token(request=request) + auth_error = _require_bearer_token( + request=request, + dataset_type=dataset_type, + ) if auth_error is not None: return auth_error dataset = _find_dataset( dataset_store=dataset_store, dataset_uuid=dataset_uuid, - dataset_type=dataset_type, ) if dataset is None: return _unknown_dataset_response(dataset_uuid=dataset_uuid) @@ -1086,6 +1550,7 @@ def download_model_target_dataset( HTTPStatus.OK, { "Content-Length": str(object=len(body)), + "Content-Disposition": "attachment; filename=full-dataset.zip", "Content-Type": "application/zip", }, body, @@ -1105,15 +1570,21 @@ def delete_model_target_dataset( if content_length_error is not None: return content_length_error - auth_error = _require_bearer_token(request=request) + auth_error = _require_bearer_token( + request=request, + dataset_type=dataset_type, + ) if auth_error is not None: return auth_error dataset = _find_dataset( dataset_store=dataset_store, dataset_uuid=dataset_uuid, - dataset_type=dataset_type, ) if dataset is None: return _unknown_dataset_response(dataset_uuid=dataset_uuid) + if dataset.dataset_type != dataset_type: + # Real Vuforia returns success when deleting through the other route, + # but leaves the dataset available through its creation route. + return HTTPStatus.OK, {"Content-Length": "0"}, "" dataset_store.remove_model_target_dataset(dataset_uuid=dataset_uuid) return HTTPStatus.OK, {"Content-Length": "0"}, "" diff --git a/src/mock_vws/_requests_mock_server/mock_web_services_api.py b/src/mock_vws/_requests_mock_server/mock_web_services_api.py index 0fdc7f5d5..fe9de4fd4 100644 --- a/src/mock_vws/_requests_mock_server/mock_web_services_api.py +++ b/src/mock_vws/_requests_mock_server/mock_web_services_api.py @@ -35,10 +35,14 @@ ) from mock_vws._model_target_web_api import ( create_model_target_dataset, + create_oauth2_client_credential, delete_model_target_dataset, + delete_oauth2_client_credential, download_model_target_dataset, get_model_target_dataset_status, + list_oauth2_client_credentials, oauth2_token, + update_oauth2_client_credential_scopes, ) from mock_vws._reco_counts_web_api import ( create_reco_counts_report, @@ -130,7 +134,7 @@ def decorator( @beartype(conf=BeartypeConf(is_pep484_tower=True)) -class MockVuforiaWebServicesAPI: +class MockVuforiaWebServicesAPI: # pylint: disable=too-many-public-methods """A fake implementation of the Vuforia Web Services API.""" def __init__( @@ -180,12 +184,75 @@ def __init__( self._vumark_generation_failure = vumark_generation_failure @route(path_pattern="/oauth2/token", http_methods={HTTPMethod.POST}) - def oauth2_token( # pylint: disable=no-self-use + def oauth2_token( self, request: RequestData, ) -> _ResponseType: """Obtain an OAuth2 token for the Model Target Web API.""" - return oauth2_token(request=request) + return oauth2_token( + request=request, + credential_store=self._target_manager, + ) + + @route( + path_pattern="/oauth2/clientcredentials", + http_methods={HTTPMethod.POST}, + ) + def create_oauth2_client_credential( + self, + request: RequestData, + ) -> _ResponseType: + """Create an OAuth2 client credential.""" + return create_oauth2_client_credential( + request=request, + credential_store=self._target_manager, + ) + + @route( + path_pattern="/oauth2/clientcredentials", + http_methods={HTTPMethod.GET}, + ) + def list_oauth2_client_credentials( + self, + request: RequestData, + ) -> _ResponseType: + """List OAuth2 client credentials.""" + return list_oauth2_client_credentials( + request=request, + credential_store=self._target_manager, + ) + + @route( + path_pattern=("/oauth2/clientcredentials/(?P[^/]+)/scopes"), + http_methods={HTTPMethod.PUT}, + ) + def update_oauth2_client_credential_scopes( + self, + request: RequestData, + ) -> _ResponseType: + """Update an OAuth2 client credential's scopes.""" + client_id = request.path.split(sep="/")[-2] + return update_oauth2_client_credential_scopes( + request=request, + credential_store=self._target_manager, + client_id=client_id, + ) + + @route( + path_pattern="/oauth2/clientcredentials/(?P[^/]+)", + http_methods={HTTPMethod.DELETE}, + ) + def delete_oauth2_client_credential( + self, + request: RequestData, + ) -> _ResponseType: + """Delete an OAuth2 client credential.""" + client_id = request.path.rsplit(sep="/", maxsplit=1)[-1] + return delete_oauth2_client_credential( + request=request, + credential_store=self._target_manager, + client_id=client_id, + ) @route( path_pattern="/modeltargets/datasets", diff --git a/src/mock_vws/model_target.py b/src/mock_vws/model_target.py index 98e796695..b63cf35d7 100644 --- a/src/mock_vws/model_target.py +++ b/src/mock_vws/model_target.py @@ -31,6 +31,16 @@ class ModelTargetDatasetType(StrEnum): ADVANCED = "advanced" +@beartype +@dataclass(frozen=True, kw_only=True) +class OAuth2ClientCredential: + """An OAuth2 client credential managed through the Vuforia Web API.""" + + client_id: str + client_secret: str + scopes: tuple[str, ...] + + @beartype @dataclass(frozen=True, kw_only=True) class ModelTargetGenerationFailure: diff --git a/src/mock_vws/target_manager.py b/src/mock_vws/target_manager.py index 365a2242c..81e155c37 100644 --- a/src/mock_vws/target_manager.py +++ b/src/mock_vws/target_manager.py @@ -9,7 +9,7 @@ RequestRateLimiter, ) from mock_vws.database import CloudDatabase, VuMarkDatabase -from mock_vws.model_target import ModelTargetDataset +from mock_vws.model_target import ModelTargetDataset, OAuth2ClientCredential from mock_vws.reco_counts import RecoCountsReport if TYPE_CHECKING: @@ -29,6 +29,7 @@ def __init__(self) -> None: self._cloud_databases: set[CloudDatabase] = set() self._vumark_databases: set[VuMarkDatabase] = set() self._model_target_datasets: dict[str, ModelTargetDataset] = {} + self._oauth2_client_credentials: dict[str, OAuth2ClientCredential] = {} self._reco_counts_reports: dict[str, RecoCountsReport] = {} self._request_rate_limiter = RequestRateLimiter( time_function=time.monotonic, @@ -59,6 +60,22 @@ def reco_counts_reports(self) -> dict[str, RecoCountsReport]: """All reco counts reports, keyed by report identifier.""" return dict(self._reco_counts_reports) + @property + def oauth2_client_credentials(self) -> dict[str, OAuth2ClientCredential]: + """All dynamically created OAuth2 client credentials.""" + return dict(self._oauth2_client_credentials) + + def add_oauth2_client_credential( + self, + credential: OAuth2ClientCredential, + ) -> None: + """Add an OAuth2 client credential.""" + self._oauth2_client_credentials[credential.client_id] = credential + + def remove_oauth2_client_credential(self, client_id: str) -> None: + """Remove an OAuth2 client credential.""" + del self._oauth2_client_credentials[client_id] + def add_reco_counts_report( self, reco_counts_report: RecoCountsReport, diff --git a/tests/mock_vws/fixtures/credentials.py b/tests/mock_vws/fixtures/credentials.py index dfedf8a2a..81d4d59f4 100644 --- a/tests/mock_vws/fixtures/credentials.py +++ b/tests/mock_vws/fixtures/credentials.py @@ -84,6 +84,8 @@ class _ModelTargetSettings(BaseSettings): client_id: str client_secret: str cad_data_url: str + username: str + password: str model_config = SettingsConfigDict( env_prefix="MODEL_TARGET_VUFORIA_", @@ -119,6 +121,8 @@ class ModelTargetCredentials: client_id: str = field(repr=False) client_secret: str = field(repr=False) cad_data_url: str = field(repr=False) + username: str = field(repr=False) + password: str = field(repr=False) def get_model_target_credentials() -> ModelTargetCredentials: @@ -130,6 +134,8 @@ def get_model_target_credentials() -> ModelTargetCredentials: client_id=settings.client_id, client_secret=settings.client_secret, cad_data_url=settings.cad_data_url, + username=settings.username, + password=settings.password, ) diff --git a/tests/mock_vws/fixtures/model_target_prepared_requests.py b/tests/mock_vws/fixtures/model_target_prepared_requests.py index 11741d835..a6cc68a6c 100644 --- a/tests/mock_vws/fixtures/model_target_prepared_requests.py +++ b/tests/mock_vws/fixtures/model_target_prepared_requests.py @@ -52,6 +52,8 @@ def credentials_for_backend( client_id="client-id", client_secret="client-secret", cad_data_url="https://example.com/model.glb", + username="user@example.com", + password="password", ) @@ -62,6 +64,7 @@ def get_access_token( backend: VuforiaBackend, ) -> str: """Return an OAuth2 access token.""" + del backend response = requests.post( url=f"{MODEL_TARGET_VWS_HOST}/oauth2/token", auth=(credentials.client_id, credentials.client_secret), @@ -69,19 +72,6 @@ def get_access_token( timeout=30, ) - if ( - backend == VuforiaBackend.REAL - and response.status_code == HTTPStatus.UNAUTHORIZED - and response.json() == {"error": "invalid_client"} - ): - pytest.xfail( - reason=( - "Real Model Target Web API credentials are not accepted; " - "authenticated behavior is verified against the mock " - "backends only until the credentials are rotated." - ), - ) - assert response.status_code == HTTPStatus.OK response_json: dict[str, Any] = json.loads(s=response.text) access_token = response_json["access_token"] diff --git a/tests/mock_vws/test_docker.py b/tests/mock_vws/test_docker.py index 8b05e939c..9c79e8f32 100644 --- a/tests/mock_vws/test_docker.py +++ b/tests/mock_vws/test_docker.py @@ -391,4 +391,4 @@ def _assert_model_target_round_trip(*, vws_container: Container) -> None: with zipfile.ZipFile( file=io.BytesIO(initial_bytes=download_response.content), ) as downloaded_zip: - assert downloaded_zip.namelist() == ["dataset.json"] + assert downloaded_zip.namelist() == ["MTDataset.dat", "MTDataset.xml"] diff --git a/tests/mock_vws/test_flask_app_usage.py b/tests/mock_vws/test_flask_app_usage.py index f69e97c6e..a267b0fcb 100644 --- a/tests/mock_vws/test_flask_app_usage.py +++ b/tests/mock_vws/test_flask_app_usage.py @@ -993,7 +993,7 @@ def test_standard_dataset_workflow( with zipfile.ZipFile( file=io.BytesIO(initial_bytes=dataset_response.content), ) as dataset_zip: - assert dataset_zip.namelist() == ["dataset.json"] + assert dataset_zip.namelist() == ["MTDataset.dat", "MTDataset.xml"] @staticmethod def _dataset_status(dataset_uuid: str) -> dict[str, Any]: diff --git a/tests/mock_vws/test_model_target_generation_failure.py b/tests/mock_vws/test_model_target_generation_failure.py index 266850dfb..5e315159e 100644 --- a/tests/mock_vws/test_model_target_generation_failure.py +++ b/tests/mock_vws/test_model_target_generation_failure.py @@ -10,7 +10,11 @@ from mock_vws import MockVWS, ModelTargetGenerationFailure -_AUTHORIZATION = "Bearer eyJhbGciOiJtb2NrIn0.e30.c2lnbmF0dXJl" +_AUTHORIZATION = ( + "Bearer eyJhbGciOiJtb2NrIn0." + "eyJzY29wZSI6Im1vZGVsdGFyZ2V0cy5zdGFuZGFyZG1vZGVsdGFyZ2V0LmFsbCJ9." + "c2lnbmF0dXJl" +) _CREATE_URL = "https://vws.vuforia.com/modeltargets/datasets" _REQUEST_BODY: dict[str, Any] = { "name": "dataset-name", diff --git a/tests/mock_vws/test_model_target_generation_warning.py b/tests/mock_vws/test_model_target_generation_warning.py index 51b8d216c..8170471b4 100644 --- a/tests/mock_vws/test_model_target_generation_warning.py +++ b/tests/mock_vws/test_model_target_generation_warning.py @@ -14,7 +14,11 @@ ModelTargetGenerationWarning, ) -_AUTHORIZATION = "Bearer eyJhbGciOiJtb2NrIn0.e30.c2lnbmF0dXJl" +_AUTHORIZATION = ( + "Bearer eyJhbGciOiJtb2NrIn0." + "eyJzY29wZSI6Im1vZGVsdGFyZ2V0cy5zdGFuZGFyZG1vZGVsdGFyZ2V0LmFsbCJ9." + "c2lnbmF0dXJl" +) _CREATE_URL = "https://vws.vuforia.com/modeltargets/datasets" _REQUEST_BODY: dict[str, Any] = { "name": "dataset-name", diff --git a/tests/mock_vws/test_model_target_web_api.py b/tests/mock_vws/test_model_target_web_api.py index c8357ebc0..957e3c7e3 100644 --- a/tests/mock_vws/test_model_target_web_api.py +++ b/tests/mock_vws/test_model_target_web_api.py @@ -1,10 +1,13 @@ """Verified fake tests for the Model Target Web API.""" +# pyright: reportPrivateUsage=false + import base64 import dataclasses import io import json import textwrap +import time import zipfile from http import HTTPMethod, HTTPStatus from typing import Any @@ -15,7 +18,12 @@ from beartype import beartype from vws.response import Response -from mock_vws import MockVWS, ModelTargetGenerationFailure +from mock_vws import ( + MockVWS, + ModelTargetGenerationFailure, + _model_target_web_api, +) +from mock_vws._flask_server import target_manager as flask_target_manager from mock_vws.model_target import ModelTargetDataset, ModelTargetDatasetType from tests.mock_vws.fixtures.model_target_prepared_requests import ( MODEL_TARGET_DATASET_UUID, @@ -27,7 +35,11 @@ from tests.mock_vws.utils.assertions import assert_valid_date_header _VWS_HOST = "https://vws.vuforia.com" -_MOCK_BEARER_TOKEN = "eyJhbGciOiJtb2NrIn0.e30.c2lnbmF0dXJl" +_MOCK_BEARER_TOKEN = ( + "eyJhbGciOiJtb2NrIn0." + "eyJzY29wZSI6Im1vZGVsdGFyZ2V0cy5zdGFuZGFyZG1vZGVsdGFyZ2V0LmFsbCJ9." + "c2lnbmF0dXJl" +) _VIEW: dict[str, Any] = { @@ -251,6 +263,29 @@ class TestAuthentication: {"error": "unsupported_grant_type"}, id="unsupported-grant-type", ), + pytest.param( + None, + {"grant_type": "password", "password": "password"}, + HTTPStatus.BAD_REQUEST, + { + "error": "invalid_request", + "error_description": "Missing username and/or password", + }, + id="password-grant-missing-username", + ), + pytest.param( + None, + { + "grant_type": "password", + "username": "user@example.com", + }, + HTTPStatus.BAD_REQUEST, + { + "error": "invalid_request", + "error_description": "Missing username and/or password", + }, + id="password-grant-missing-password", + ), ], ) def test_invalid_oauth2_token_request( @@ -274,6 +309,206 @@ def test_invalid_oauth2_token_request( body=body, ) + @staticmethod + def test_password_grant( + *, + verify_model_target_mock_vuforia: VuforiaBackend, + ) -> None: + """A username and password can be exchanged for a scoped token.""" + credentials = credentials_for_backend( + backend=verify_model_target_mock_vuforia, + ) + response = requests.post( + url=f"{_VWS_HOST}/oauth2/token", + data={ + "grant_type": "password", + "username": credentials.username, + "password": credentials.password, + "scope": "modeltargets.standardmodeltarget.all", + }, + timeout=30, + ) + + assert response.status_code == HTTPStatus.OK + assert isinstance(response.json()["access_token"], str) + assert response.json()["token_type"] == "bearer" + + @staticmethod + def test_scoped_client_credentials_grant( + *, + verify_model_target_mock_vuforia: VuforiaBackend, + ) -> None: + """A client credentials request accepts an explicit scope.""" + credentials = credentials_for_backend( + backend=verify_model_target_mock_vuforia, + ) + response = requests.post( + url=f"{_VWS_HOST}/oauth2/token", + auth=(credentials.client_id, credentials.client_secret), + data={ + "grant_type": "client_credentials", + "scope": "modeltargets.standardmodeltarget.all", + }, + timeout=30, + ) + + assert response.status_code == HTTPStatus.OK + assert isinstance(response.json()["access_token"], str) + assert response.json()["token_type"] == "bearer" + + @staticmethod + def test_insufficient_scope( + *, + verify_model_target_mock_vuforia: VuforiaBackend, + ) -> None: + """A route rejects a token carrying only another route's scope.""" + credentials = credentials_for_backend( + backend=verify_model_target_mock_vuforia, + ) + token_response = requests.post( + url=f"{_VWS_HOST}/oauth2/token", + auth=(credentials.client_id, credentials.client_secret), + data={ + "grant_type": "client_credentials", + "scope": "modeltargets.standardmodeltarget.all", + }, + timeout=30, + ) + assert token_response.status_code == HTTPStatus.OK + + response = requests.post( + url=f"{_VWS_HOST}/modeltargets/advancedDatasets", + headers={ + "Authorization": ( + f"Bearer {token_response.json()['access_token']}" + ), + }, + json={}, + timeout=30, + ) + + assert response.status_code == HTTPStatus.FORBIDDEN + assert response.text == ( + "User does not have the required scopes to perform this action" + ) + + @staticmethod + def test_client_credentials_management( + *, + verify_model_target_mock_vuforia: VuforiaBackend, + ) -> None: + """Client credentials can be created, listed, updated and + deleted. + """ + credentials = credentials_for_backend( + backend=verify_model_target_mock_vuforia, + ) + password_token_response = requests.post( + url=f"{_VWS_HOST}/oauth2/token", + data={ + "grant_type": "password", + "username": credentials.username, + "password": credentials.password, + "scope": "oauth2.clientcredentials.all", + }, + timeout=30, + ) + assert password_token_response.status_code == HTTPStatus.OK + headers = { + "Authorization": ( + f"Bearer {password_token_response.json()['access_token']}" + ), + } + client_id: str | None = None + + try: + create_response = requests.post( + url=f"{_VWS_HOST}/oauth2/clientcredentials", + headers=headers, + json={ + "scopes": ["modeltargets.standardmodeltarget.all"], + }, + timeout=30, + ) + assert create_response.status_code == HTTPStatus.CREATED + client_id = create_response.json()["client_id"] + client_secret = create_response.json()["client_secret"] + + list_response = requests.get( + url=f"{_VWS_HOST}/oauth2/clientcredentials", + headers=headers, + timeout=30, + ) + assert list_response.status_code == HTTPStatus.OK + created_entries = [ + entry + for entry in list_response.json() + if entry["clientId"] == client_id + ] + assert len(created_entries) == 1 + assert created_entries[0]["scopes"] == [ + "modeltargets.standardmodeltarget.all", + ] + + update_response = requests.put( + url=( + f"{_VWS_HOST}/oauth2/clientcredentials/{client_id}/scopes" + ), + headers=headers, + json=["modeltargets.advancedmodeltarget.all"], + timeout=30, + ) + assert update_response.status_code == HTTPStatus.OK + assert update_response.json() == { + "clientId": client_id, + "scopes": ["modeltargets.advancedmodeltarget.all"], + } + + assert client_id is not None + assert isinstance(client_secret, str) + client_token_response = requests.post( + url=f"{_VWS_HOST}/oauth2/token", + auth=(client_id, client_secret), + data={"grant_type": "client_credentials"}, + timeout=30, + ) + assert client_token_response.status_code == HTTPStatus.OK + client_access_token = client_token_response.json()["access_token"] + insufficient_response = requests.post( + url=f"{_VWS_HOST}/modeltargets/datasets", + headers={ + "Authorization": f"Bearer {client_access_token}", + }, + json={}, + timeout=30, + ) + assert insufficient_response.status_code == HTTPStatus.FORBIDDEN + finally: + if client_id is not None: # pragma: no branch + delete_response = requests.delete( + url=(f"{_VWS_HOST}/oauth2/clientcredentials/{client_id}"), + headers=headers, + timeout=30, + ) + assert delete_response.status_code == HTTPStatus.NO_CONTENT + + missing_client_id = "000000000000000000000" + missing_response = requests.delete( + url=(f"{_VWS_HOST}/oauth2/clientcredentials/{missing_client_id}"), + headers=headers, + timeout=30, + ) + assert missing_response.status_code == HTTPStatus.NOT_FOUND + assert missing_response.json() == { + "error": { + "code": "NOT_FOUND", + "message": ( + f"Clientcredential with ID={missing_client_id} not found" + ), + "target": "clientcredential", + }, + } + @pytest.mark.usefixtures("verify_model_target_mock_vuforia") class TestAuthorizationHeader: @@ -490,10 +725,18 @@ class TestInvalidJson: """ @staticmethod + @pytest.mark.parametrize( + argnames="content_type", + argvalues=[ + pytest.param(None, id="missing"), + pytest.param("", id="empty"), + ], + ) def test_wrong_content_type( *, verify_model_target_mock_vuforia: VuforiaBackend, model_target_endpoint: ModelTargetEndpoint, + content_type: str | None, ) -> None: """Requests without a JSON content type are rejected with 415 by endpoints which read a body, and are unaffected elsewhere. @@ -505,7 +748,10 @@ def test_wrong_content_type( **model_target_endpoint.headers, "Authorization": f"Bearer {access_token}", } - new_headers.pop("Content-Type", None) + if content_type is None: + new_headers.pop("Content-Type", None) + else: + new_headers["Content-Type"] = content_type new_endpoint = dataclasses.replace( model_target_endpoint, headers=new_headers, @@ -738,7 +984,10 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: ), pytest.param( {**_UNAUTHENTICATED_DATASET_REQUEST, "models": ["model"]}, - {"/models(0): error.expected.jsobject"}, + { + "/models(0)/name: element is required", + "/models(0)/views: element is required", + }, id="model-not-object", ), pytest.param( @@ -749,7 +998,10 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: "model", ], }, - {"/models(1): error.expected.jsobject"}, + { + "/models(1)/name: element is required", + "/models(1)/views: element is required", + }, id="second-model-not-object", ), pytest.param( @@ -758,11 +1010,8 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: "models": [_EMPTY_MODEL], }, { - ( - "/models(0): one and only one of cadDataUrl and " - "cadDataBlob is required" - ), "/models(0)/name: element is required", + "/models(0)/views: element is required", }, id="model-missing-fields", ), @@ -773,8 +1022,8 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: }, { ( - "/models(0): one and only one of cadDataUrl and " - "cadDataBlob is required" + "model 'model-name' is invalid. One of `cadDataBlob`, " + "`cadDataUrl`, `cadDataUuid` need to be provided" ), }, id="model-without-cad-data", @@ -792,8 +1041,9 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: }, { ( - "/models(0): one and only one of cadDataUrl and " - "cadDataBlob is required" + "model 'model-name' is invalid. Only one of " + "`cadDataBlob`, `cadDataUrl`, `cadDataUuid` need to " + "be provided" ), }, id="model-with-both-cad-data-sources", @@ -838,7 +1088,14 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: **_UNAUTHENTICATED_DATASET_REQUEST, "models": [{**_MODEL, "cadDataFormat": "gltf"}], }, - {"/models(0)/cadDataFormat: error.expected.validenum"}, + { + ( + "Unrecognized cadDataFormat 'GLTF'. Allowed values " + "are: ZIP, GLB, DRC_GLB, DRC_GLTF, DAE, FBX, IGES, " + "OBJ, PVS, PVZ, STL, VRML, or specify no " + "cadDataFormat to auto-detect GLB and zipped glTFs." + ), + }, id="model-cad-data-format-not-in-enum", ), pytest.param( @@ -862,7 +1119,12 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: **_UNAUTHENTICATED_DATASET_REQUEST, "models": [{**_MODEL, "simplify": "sometimes"}], }, - {"/models(0)/simplify: error.expected.validenum"}, + { + ( + "invalid simplify. Should be one of 'never', " + "'always', 'auto'. You provided 'Some(sometimes)'" + ), + }, id="model-simplify-not-in-enum", ), pytest.param( @@ -870,7 +1132,12 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: **_UNAUTHENTICATED_DATASET_REQUEST, "models": [{**_MODEL, "automaticColoring": "sometimes"}], }, - {"/models(0)/automaticColoring: error.expected.validenum"}, + { + ( + "invalid automaticColoring. Should be one of 'never', " + "'always', 'auto'. You provided 'sometimes'" + ), + }, id="model-automatic-coloring-not-in-enum", ), pytest.param( @@ -878,7 +1145,13 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: **_UNAUTHENTICATED_DATASET_REQUEST, "models": [{**_MODEL, "motionHint": "still"}], }, - {"/models(0)/motionHint: error.expected.validenum"}, + { + ( + "`motionHint` and `trackingMode` are no longer " + "supported when using `targetsSdk` 10.9 or later. " + "Please use the `optimizeTrackingFor` setting instead." + ), + }, id="model-motion-hint-not-in-enum", ), pytest.param( @@ -886,7 +1159,12 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: **_UNAUTHENTICATED_DATASET_REQUEST, "models": [{**_MODEL, "optimizeTrackingFor": "cars"}], }, - {"/models(0)/optimizeTrackingFor: error.expected.validenum"}, + { + ( + "`optimizeTrackingFor` must be one of " + "default,low_feature_objects,ar_controller" + ), + }, id="model-optimize-tracking-for-not-in-enum", ), pytest.param( @@ -894,7 +1172,13 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: **_UNAUTHENTICATED_DATASET_REQUEST, "models": [{**_MODEL, "trackingMode": "boat"}], }, - {"/models(0)/trackingMode: error.expected.validenum"}, + { + ( + "`motionHint` and `trackingMode` are no longer " + "supported when using `targetsSdk` 10.9 or later. " + "Please use the `optimizeTrackingFor` setting instead." + ), + }, id="model-tracking-mode-not-in-enum", ), pytest.param( @@ -909,8 +1193,15 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: ], }, { - "/models(0)/motionHint: error.expected.validenum", - "/models(0)/simplify: error.expected.validenum", + ( + "`motionHint` and `trackingMode` are no longer " + "supported when using `targetsSdk` 10.9 or later. " + "Please use the `optimizeTrackingFor` setting instead." + ), + ( + "invalid simplify. Should be one of 'never', " + "'always', 'auto'. You provided 'Some(sometimes)'" + ), }, id="model-multiple-enum-errors", ), @@ -927,7 +1218,7 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: **_UNAUTHENTICATED_DATASET_REQUEST, "models": [{**_MODEL, "views": ["view-name"]}], }, - {"/models(0)/views(0): error.expected.jsobject"}, + {"/models(0)/views(0)/name: element is required"}, id="view-not-object", ), pytest.param( @@ -935,13 +1226,7 @@ def test_invalid_basic_auth_header(*, authorization: str) -> None: **_UNAUTHENTICATED_DATASET_REQUEST, "models": [{**_MODEL, "views": [_EMPTY_VIEW]}], }, - { - ( - "/models(0)/views(0)/guideViewPosition: " - "element is required" - ), - "/models(0)/views(0)/name: element is required", - }, + {"/models(0)/views(0)/name: element is required"}, id="view-missing-fields", ), pytest.param( @@ -1147,6 +1432,42 @@ def test_invalid_dataset_request( for detail in error["details"]: assert detail["code"] == "VALIDATION_ERROR" + @staticmethod + def test_advanced_model_count_exceeds_limit( + *, + verify_model_target_mock_vuforia: VuforiaBackend, + ) -> None: + """Advanced datasets reject more than 20 uniquely named models.""" + models = [{**_MODEL, "name": f"model-{index}"} for index in range(21)] + # Include a duplicate to verify the two validation details which real + # Vuforia returns together for this request. + models[-1]["name"] = models[0]["name"] + body = {**_UNAUTHENTICATED_DATASET_REQUEST, "models": models} + credentials = credentials_for_backend( + backend=verify_model_target_mock_vuforia, + ) + access_token = get_access_token( + credentials=credentials, + backend=verify_model_target_mock_vuforia, + ) + response = requests.post( + url=f"{_VWS_HOST}/modeltargets/advancedDatasets", + headers={"Authorization": f"Bearer {access_token}"}, + json=body, + timeout=30, + ) + + assert response.status_code == HTTPStatus.BAD_REQUEST + error = response.json()["error"] + assert error["code"] == "BAD_REQUEST" + assert {detail["message"] for detail in error["details"]} == { + "names of models must be unique within a Target.", + "total number of models must be maximum 20", + } + assert all( + detail["code"] == "VALIDATION_ERROR" for detail in error["details"] + ) + @staticmethod @pytest.mark.parametrize( argnames=("method", "path"), @@ -1201,12 +1522,8 @@ def test_unknown_dataset( assert error["target"].startswith("userId:") -class TestMockOnlyErrors: - """Mock-only Model Target Web API error paths. - - These cases cannot easily be verified against real Vuforia with the - currently available test account and are kept mock-only by design. - """ +class TestStateBasedDatasets: + """Verified fake tests for State-Based Model Targets.""" @staticmethod @pytest.mark.parametrize( @@ -1231,7 +1548,7 @@ class TestMockOnlyErrors: ) def test_state_based_dataset( *, - model_target_mock_only_vuforia: VuforiaBackend, + verify_model_target_mock_vuforia: VuforiaBackend, dataset_path: str, view_updates: dict[str, object], ) -> None: @@ -1251,7 +1568,7 @@ def test_state_based_dataset( ], } access_token = _access_token_for_backend( - backend=model_target_mock_only_vuforia, + backend=verify_model_target_mock_vuforia, ) headers = {"Authorization": f"Bearer {access_token}"} create_response = requests.post( @@ -1270,6 +1587,46 @@ def test_state_based_dataset( ) assert delete_response.status_code == HTTPStatus.OK + @staticmethod + def test_view_states_are_a_subset( + *, + verify_model_target_mock_vuforia: VuforiaBackend, + ) -> None: + """A view cannot select a state absent from the configuration.""" + body = { + **_UNAUTHENTICATED_DATASET_REQUEST, + "models": [ + { + **_MODEL, + "stateBasedConfigurationJsonString": ( + _STATE_CONFIGURATION + ), + "views": [{**_VIEW, "states": ["unknown"]}], + }, + ], + } + access_token = _access_token_for_backend( + backend=verify_model_target_mock_vuforia, + ) + response = requests.post( + url=f"{_VWS_HOST}/modeltargets/datasets", + headers={"Authorization": f"Bearer {access_token}"}, + json=body, + timeout=30, + ) + + assert response.status_code == HTTPStatus.BAD_REQUEST + error = response.json()["error"] + assert error["code"] == "BAD_REQUEST" + assert [detail["message"] for detail in error["details"]] == [ + "states in entrypoint view-name' must be a subset of all states", + ] + assert error["details"][0]["code"] == "VALIDATION_ERROR" + + +class TestAdditionalBehaviors: + """Additional verified and mock-only Model Target behaviors.""" + @staticmethod @pytest.mark.parametrize( argnames=("model_updates", "view_updates", "expected_message"), @@ -1322,12 +1679,6 @@ def test_state_based_dataset( ("/models(0)/views(0)/states(1): error.expected.jsstring"), id="view-state-not-string", ), - pytest.param( - {"stateBasedConfigurationJsonString": _STATE_CONFIGURATION}, - {"states": ["unknown"]}, - ("/models(0)/views(0)/states(0): error.expected.validenum"), - id="view-state-not-declared", - ), pytest.param( {}, {"states": ["assembled"]}, @@ -1376,86 +1727,62 @@ def test_invalid_state_based_dataset( assert error["details"][0]["code"] == "VALIDATION_ERROR" @staticmethod - def test_advanced_model_count_exceeds_limit() -> None: - """Advanced dataset requests with too many models are rejected. - - Real Vuforia returns a 403 for the currently available test account - because the account lacks the advanced-dataset scope, so the - validation-error shape cannot be observed end-to-end. The mock - therefore enforces the documented advanced-dataset model count - limit on its own. - """ - body = { - **_UNAUTHENTICATED_DATASET_REQUEST, - "models": [*_UNAUTHENTICATED_DATASET_REQUEST["models"]] * 21, - } - with MockVWS(): - response = requests.post( - url=f"{_VWS_HOST}/modeltargets/advancedDatasets", - headers={"Authorization": f"Bearer {_MOCK_BEARER_TOKEN}"}, - json=body, - timeout=30, - ) - - assert response.status_code == HTTPStatus.BAD_REQUEST - error = response.json()["error"] - assert error["code"] == "BAD_REQUEST" - assert error["details"][0]["code"] == "VALIDATION_ERROR" - - @staticmethod - def test_advanced_realistic_appearance_not_in_enum() -> None: + def test_advanced_realistic_appearance_not_in_enum( + *, + verify_model_target_mock_vuforia: VuforiaBackend, + ) -> None: """Advanced dataset requests with a ``realisticAppearance`` value outside the documented enumeration are rejected. The Model Target OpenAPI specification documents ``realisticAppearance`` as a model field for advanced datasets - only, so standard dataset creation does not validate it. This is - mock-only because the available test account lacks the - advanced-dataset scope, so real Vuforia rejects the request with a - 403 before validating the body. + only. """ + credentials = credentials_for_backend( + backend=verify_model_target_mock_vuforia, + ) body = { **_UNAUTHENTICATED_DATASET_REQUEST, - "models": [{**_MODEL, "realisticAppearance": "yes"}], + "models": [ + { + **_MODEL, + "cadDataUrl": credentials.cad_data_url, + "realisticAppearance": "yes", + }, + ], } - headers = {"Authorization": f"Bearer {_MOCK_BEARER_TOKEN}"} - with MockVWS(): - advanced_response = requests.post( - url=f"{_VWS_HOST}/modeltargets/advancedDatasets", - headers=headers, - json=body, - timeout=30, - ) - standard_response = requests.post( - url=f"{_VWS_HOST}/modeltargets/datasets", - headers=headers, - json=body, - timeout=30, - ) + access_token = get_access_token( + credentials=credentials, + backend=verify_model_target_mock_vuforia, + ) + advanced_response = requests.post( + url=f"{_VWS_HOST}/modeltargets/advancedDatasets", + headers={"Authorization": f"Bearer {access_token}"}, + json=body, + timeout=30, + ) assert advanced_response.status_code == HTTPStatus.BAD_REQUEST error = advanced_response.json()["error"] assert error["code"] == "BAD_REQUEST" assert [detail["message"] for detail in error["details"]] == [ - "/models(0)/realisticAppearance: error.expected.validenum", + '`realisticAppearance` must be one of "true", "false", "auto".` ', ] assert error["details"][0]["code"] == "VALIDATION_ERROR" - assert standard_response.status_code == HTTPStatus.CREATED - @staticmethod def test_oauth2_token_body_not_utf_8( *, - model_target_mock_only_vuforia: VuforiaBackend, + verify_model_target_mock_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. + Real Vuforia also treats a body that cannot be decoded as an empty + form. """ credentials = credentials_for_backend( - backend=model_target_mock_only_vuforia, + backend=verify_model_target_mock_vuforia, ) response = requests.post( @@ -1561,66 +1888,67 @@ def test_failed_dataset_cannot_be_downloaded() -> None: ), ], ) - def test_dataset_is_not_visible_to_the_other_dataset_type( + def test_dataset_is_visible_to_the_other_dataset_type( *, + verify_model_target_mock_vuforia: VuforiaBackend, created_path: str, other_path: str, ) -> None: - """A dataset is not reachable through the other type's routes. - - Standard and advanced datasets are separate resources in real - Vuforia, with separate OAuth scopes. This is mock-only because the - available test account lacks the advanced-dataset scope, so real - Vuforia rejects advanced routes with a 403 before looking a dataset - up. - """ - headers = {"Authorization": f"Bearer {_MOCK_BEARER_TOKEN}"} - with MockVWS(): + """Standard and advanced routes share datasets by UUID.""" + access_token = _access_token_for_backend( + backend=verify_model_target_mock_vuforia, + ) + headers = {"Authorization": f"Bearer {access_token}"} + dataset_uuid: str | None = None + try: create_response = requests.post( url=f"{_VWS_HOST}{created_path}", headers=headers, - json=_UNAUTHENTICATED_DATASET_REQUEST, + json=_dataset_request( + cad_data_url=credentials_for_backend( + backend=verify_model_target_mock_vuforia, + ).cad_data_url, + ), timeout=30, ) assert create_response.status_code == HTTPStatus.CREATED dataset_uuid = create_response.json()["uuid"] - other_responses = [ - requests.get( - url=f"{_VWS_HOST}{other_path}/{dataset_uuid}/status", - headers=headers, - timeout=30, - ), - requests.get( - url=f"{_VWS_HOST}{other_path}/{dataset_uuid}/dataset", - headers=headers, - timeout=30, - ), - requests.delete( - url=f"{_VWS_HOST}{other_path}/{dataset_uuid}", - headers=headers, - timeout=30, - ), - ] - - # The dataset survives the delete attempt made through the other - # type's routes. - own_status_response = requests.get( - url=f"{_VWS_HOST}{created_path}/{dataset_uuid}/status", + other_status_response = requests.get( + url=f"{_VWS_HOST}{other_path}/{dataset_uuid}/status", headers=headers, timeout=30, ) - - for response in other_responses: - assert response.status_code == HTTPStatus.NOT_FOUND - error = response.json()["error"] - assert error["code"] == "NOT_FOUND" - assert error["message"] == ( - "Could not find a model-view database with uuid " - f"{dataset_uuid}" + other_delete_response = requests.delete( + url=f"{_VWS_HOST}{other_path}/{dataset_uuid}", + headers=headers, + timeout=30, + ) + own_status_response = requests.get( + url=( + f"{_VWS_HOST}{created_path}/" + f"{create_response.json()['uuid']}/status" + ), + headers=headers, + timeout=30, ) - assert error["target"].startswith("userId:") + finally: + if dataset_uuid is not None: # pragma: no branch + delete_response = requests.delete( + url=f"{_VWS_HOST}{created_path}/{dataset_uuid}", + headers=headers, + timeout=30, + ) + assert delete_response.status_code in { + HTTPStatus.OK, + HTTPStatus.NO_CONTENT, + } + assert other_status_response.status_code == HTTPStatus.OK + assert other_delete_response.status_code in { + HTTPStatus.OK, + HTTPStatus.NO_CONTENT, + } assert own_status_response.status_code == HTTPStatus.OK @@ -1632,7 +1960,12 @@ def test_create_status_and_delete( *, verify_model_target_mock_vuforia: VuforiaBackend, ) -> None: - """A standard Model Target dataset can be created and deleted.""" + """A standard dataset works through the shared advanced routes. + + Standard generation is fast enough for verified CI coverage. Real + Vuforia exposes its completed artifact through both route families, + so this also verifies the advanced status and download endpoints. + """ credentials = credentials_for_backend( backend=verify_model_target_mock_vuforia, ) @@ -1661,7 +1994,8 @@ def test_create_status_and_delete( status_response = requests.get( url=( - f"{_VWS_HOST}/modeltargets/datasets/{dataset_uuid}/status" + f"{_VWS_HOST}/modeltargets/advancedDatasets/" + f"{dataset_uuid}/status" ), headers=headers, timeout=30, @@ -1677,6 +2011,52 @@ def test_create_status_and_delete( "failed", } assert isinstance(status_response_json["createdAt"], str) + + deadline = time.monotonic() + 60 + while ( + status_response_json["status"] == "processing" + and time.monotonic() < deadline + ): + time.sleep(1) + status_response = requests.get( + url=( + f"{_VWS_HOST}/modeltargets/advancedDatasets/" + f"{dataset_uuid}/status" + ), + headers=headers, + timeout=30, + ) + assert status_response.status_code == HTTPStatus.OK + status_response_json = status_response.json() + + assert status_response_json["status"] == "done" + assert isinstance(status_response_json["completedAt"], str) + assert set(status_response_json) == { + "completedAt", + "createdAt", + "status", + "uuid", + } + + download_response = requests.get( + url=( + f"{_VWS_HOST}/modeltargets/advancedDatasets/" + f"{dataset_uuid}/dataset" + ), + headers=headers, + timeout=30, + ) + assert download_response.status_code == HTTPStatus.OK + assert download_response.headers["Content-Type"] == ( + "application/zip" + ) + assert download_response.headers["Content-Disposition"] == ( + "attachment; filename=full-dataset.zip" + ) + with zipfile.ZipFile( + file=io.BytesIO(initial_bytes=download_response.content), + ) as archive: + assert archive.namelist() == ["MTDataset.dat", "MTDataset.xml"] finally: if dataset_uuid is not None: # pragma: no branch delete_response = requests.delete( @@ -1764,3 +2144,269 @@ def test_status_uses_matching_time_field( assert body["status"] == status assert body["uuid"] == "dataset-uuid" assert {"eta", "completedAt"} & body.keys() == {time_field} + + +class TestMockOnlyOAuth2EdgeCases: + """Cover mock-only OAuth2 and validation error paths.""" + + @staticmethod + def _management_token() -> str: + """Return a token which can manage client credentials.""" + response = requests.post( + url=f"{_VWS_HOST}/oauth2/token", + data={ + "grant_type": "password", + "username": "user@example.com", + "password": "password", + "scope": "oauth2.clientcredentials.all", + }, + timeout=30, + ) + assert response.status_code == HTTPStatus.OK + access_token: object = response.json()["access_token"] + assert isinstance(access_token, str) + return access_token + + @staticmethod + def test_oauth2_grant_errors() -> None: + """Invalid passwords and scopes are rejected.""" + with MockVWS(): + invalid_password = requests.post( + url=f"{_VWS_HOST}/oauth2/token", + data={ + "grant_type": "password", + "username": "user@example.com", + "password": "wrong", + }, + timeout=30, + ) + assert invalid_password.status_code == HTTPStatus.UNAUTHORIZED + assert invalid_password.json()["error"] == "invalid_grant" + + invalid_scope = requests.post( + url=f"{_VWS_HOST}/oauth2/token", + auth=("client-id", "client-secret"), + data={"scope": "not.a.scope"}, + timeout=30, + ) + assert invalid_scope.status_code == HTTPStatus.BAD_REQUEST + assert invalid_scope.json()["error"] == "invalid_scope" + + @staticmethod + def test_client_credential_authentication_errors() -> None: + """Credential-management routes enforce bearer-token validity and + scope. + """ + headers = [ + {}, + {"Authorization": "Bearer malformed"}, + {"Authorization": "Bearer e30.e30.signature"}, + {"Authorization": f"Bearer {_MOCK_BEARER_TOKEN}"}, + ] + with MockVWS(): + for request_headers in headers: + response = requests.get( + url=f"{_VWS_HOST}/oauth2/clientcredentials", + headers=request_headers, + timeout=30, + ) + assert response.status_code in { + HTTPStatus.UNAUTHORIZED, + HTTPStatus.FORBIDDEN, + } + + delete_response = requests.delete( + url=f"{_VWS_HOST}/oauth2/clientcredentials/client-id", + timeout=30, + ) + assert delete_response.status_code == HTTPStatus.UNAUTHORIZED + + create_response = requests.post( + url=f"{_VWS_HOST}/oauth2/clientcredentials", + json={"scopes": []}, + timeout=30, + ) + assert create_response.status_code == HTTPStatus.UNAUTHORIZED + + update_response = requests.put( + url=f"{_VWS_HOST}/oauth2/clientcredentials/client-id/scopes", + json=[], + timeout=30, + ) + assert update_response.status_code == HTTPStatus.UNAUTHORIZED + + @staticmethod + def test_non_string_token_scope() -> None: + """A token with a non-string scope has no usable scopes.""" + encoded_header = ( + base64.urlsafe_b64encode( + s=b'{"alg":"mock"}', + ) + .decode(encoding="ascii") + .rstrip("=") + ) + encoded_payload = ( + base64.urlsafe_b64encode( + s=b'{"scope":[]}', + ) + .decode(encoding="ascii") + .rstrip("=") + ) + token = f"{encoded_header}.{encoded_payload}.c2lnbmF0dXJl" + with MockVWS(): + response = requests.get( + url=f"{_VWS_HOST}/oauth2/clientcredentials", + headers={"Authorization": f"Bearer {token}"}, + timeout=30, + ) + assert response.status_code == HTTPStatus.FORBIDDEN + + @staticmethod + def test_client_credential_validation_errors() -> None: + """Credential creation and updates reject invalid request + bodies. + """ + with MockVWS(): + headers = { + "Authorization": ( + f"Bearer {TestMockOnlyOAuth2EdgeCases._management_token()}" + ), + } + for content in (b"{", b'{"scopes":"scope"}', b'{"scopes":[1]}'): + response = requests.post( + url=f"{_VWS_HOST}/oauth2/clientcredentials", + headers={**headers, "Content-Type": "application/json"}, + data=content, + timeout=30, + ) + assert response.status_code == HTTPStatus.BAD_REQUEST + + missing = requests.put( + url=f"{_VWS_HOST}/oauth2/clientcredentials/missing/scopes", + headers=headers, + json=[], + timeout=30, + ) + assert missing.status_code == HTTPStatus.NOT_FOUND + + created = requests.post( + url=f"{_VWS_HOST}/oauth2/clientcredentials", + headers=headers, + json={"scopes": []}, + timeout=30, + ) + client_id = created.json()["client_id"] + for content in (b"{", b'"scope"', b"[1]"): + response = requests.put( + url=( + f"{_VWS_HOST}/oauth2/clientcredentials/" + f"{client_id}/scopes" + ), + headers={**headers, "Content-Type": "application/json"}, + data=content, + timeout=30, + ) + assert response.status_code == HTTPStatus.BAD_REQUEST + + @staticmethod + def test_client_credential_limit( + *, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Credential creation rejects stores at their configured + limit. + """ + monkeypatch.setattr( + target=_model_target_web_api, + name="_MAX_CLIENT_CREDENTIALS", + value=0, + ) + with MockVWS(): + headers = { + "Authorization": ( + f"Bearer {TestMockOnlyOAuth2EdgeCases._management_token()}" + ), + } + response = requests.post( + url=f"{_VWS_HOST}/oauth2/clientcredentials", + headers=headers, + json={"scopes": []}, + timeout=30, + ) + assert response.status_code == HTTPStatus.CONFLICT + + @staticmethod + def test_dataset_scope_and_shape_errors() -> None: + """State-based scope and less common model shapes are + validated. + """ + with MockVWS(): + state_based = { + **_UNAUTHENTICATED_DATASET_REQUEST, + "models": [ + { + **_MODEL, + "stateBasedConfigurationJsonString": ( + _STATE_CONFIGURATION + ), + }, + ], + } + forbidden = requests.post( + url=f"{_VWS_HOST}/modeltargets/datasets", + headers={"Authorization": f"Bearer {_MOCK_BEARER_TOKEN}"}, + json=state_based, + timeout=30, + ) + assert forbidden.status_code == HTTPStatus.FORBIDDEN + + invalid_view = requests.post( + url=f"{_VWS_HOST}/modeltargets/datasets", + headers={"Authorization": f"Bearer {_MOCK_BEARER_TOKEN}"}, + json={ + **_UNAUTHENTICATED_DATASET_REQUEST, + "models": [{**_MODEL, "views": [1]}], + }, + timeout=30, + ) + assert invalid_view.status_code == HTTPStatus.BAD_REQUEST + + advanced_empty = requests.post( + url=f"{_VWS_HOST}/modeltargets/advancedDatasets", + headers={ + "Authorization": ( + "Bearer eyJhbGciOiJtb2NrIn0." + "eyJzY29wZSI6Im1vZGVsdGFyZ2V0cy5hZHZhbmNlZG1vZGVs" + "dGFyZ2V0LmFsbCJ9.c2lnbmF0dXJl" + ), + }, + json={"name": "name", "targetSdk": "10.18", "models": []}, + timeout=30, + ) + assert advanced_empty.status_code == HTTPStatus.BAD_REQUEST + + @staticmethod + def test_view_helper_rejects_non_objects() -> None: + """The view validator reports view values which are not + objects. + """ + # pylint: disable=protected-access + details = _model_target_web_api._view_details( # noqa: SLF001 + models=[{"views": [1]}], + ) + assert details == [ + { + "code": "VALIDATION_ERROR", + "message": "/models(0)/views(0): error.expected.jsobject", + }, + ] + + @staticmethod + def test_target_manager_missing_credential_delete() -> None: + """The internal target manager returns 404 for an unknown + credential. + """ + response = flask_target_manager.remove_oauth2_client_credential( + client_id="missing", + ) + assert response.status_code == HTTPStatus.NOT_FOUND diff --git a/tests/mock_vws/test_requests_mock_usage.py b/tests/mock_vws/test_requests_mock_usage.py index cc1586163..489b3a02d 100644 --- a/tests/mock_vws/test_requests_mock_usage.py +++ b/tests/mock_vws/test_requests_mock_usage.py @@ -52,7 +52,11 @@ processing_time_seconds, ) -_MODEL_TARGET_AUTHORIZATION = "Bearer eyJhbGciOiJtb2NrIn0.e30.c2lnbmF0dXJl" +_MODEL_TARGET_AUTHORIZATION = ( + "Bearer eyJhbGciOiJtb2NrIn0." + "eyJzY29wZSI6Im1vZGVsdGFyZ2V0cy5hbGwifQ." + "c2lnbmF0dXJl" +) _MODEL_TARGET_DATASET_REQUEST = { "name": "dataset-name", "targetSdk": "10.18", @@ -1756,7 +1760,7 @@ def test_standard_dataset_workflow() -> None: with zipfile.ZipFile( file=io.BytesIO(initial_bytes=dataset_response.content), ) as dataset_zip: - assert dataset_zip.namelist() == ["dataset.json"] + assert dataset_zip.namelist() == ["MTDataset.dat", "MTDataset.xml"] @staticmethod def test_advanced_dataset_workflow() -> None: diff --git a/tests/mock_vws/test_respx_mock_usage.py b/tests/mock_vws/test_respx_mock_usage.py index 3b3c5d5a9..9d5bc9d2f 100644 --- a/tests/mock_vws/test_respx_mock_usage.py +++ b/tests/mock_vws/test_respx_mock_usage.py @@ -19,7 +19,11 @@ from mock_vws.image_matchers import ExactMatcher from mock_vws.target import VuMarkTarget -_MODEL_TARGET_AUTHORIZATION = "Bearer eyJhbGciOiJtb2NrIn0.e30.c2lnbmF0dXJl" +_MODEL_TARGET_AUTHORIZATION = ( + "Bearer eyJhbGciOiJtb2NrIn0." + "eyJzY29wZSI6Im1vZGVsdGFyZ2V0cy5zdGFuZGFyZG1vZGVsdGFyZ2V0LmFsbCJ9." + "c2lnbmF0dXJl" +) _MODEL_TARGET_DATASET_REQUEST = { "name": "dataset-name", "targetSdk": "10.18", diff --git a/vuforia_secrets.env.example b/vuforia_secrets.env.example index 4b4284a2c..2fb384ab0 100644 --- a/vuforia_secrets.env.example +++ b/vuforia_secrets.env.example @@ -29,3 +29,5 @@ INACTIVE_VUMARK_VUFORIA_SERVER_SECRET_KEY=example_inactive_vumark_server_secret_ MODEL_TARGET_VUFORIA_CLIENT_ID=example_model_target_client_id MODEL_TARGET_VUFORIA_CLIENT_SECRET=example_model_target_client_secret MODEL_TARGET_VUFORIA_CAD_DATA_URL=https://example.com/model.glb +MODEL_TARGET_VUFORIA_USERNAME=user@example.com +MODEL_TARGET_VUFORIA_PASSWORD=example_model_target_password