From 0177fda43304244c00b08d730c232212d40ee1b0 Mon Sep 17 00:00:00 2001 From: Nacl-HCl Date: Mon, 9 Nov 2026 02:34:58 +0000 Subject: [PATCH] server: Accept spec-compliant SpecificAssetId in GET /shells?assetIds Currently, GET /shells?assetIds only understands a custom wrapper {"name": "specificAssetId", "value": ""}. A spec-compliant base64url-encoded SpecificAssetId is silently skipped, so the filter is ignored and all shells are returned; malformed JSON escapes as a 500 instead of a 400. Each assetIds value is now decoded as a SpecificAssetId (AAS Part 2, v3.1.1): the reserved name "globalAssetId" matches the shell's globalAssetId, any other name matches by (name, value) against the shell's specific_asset_id. Malformed input is rejected with a 400. Note: clients sending the old wrapper will no longer match anything, as the wrapper now decodes to a SpecificAssetId named "specificAssetId". Fixes #668 --- server/app/interfaces/repository.py | 38 +++++++------- .../test/interfaces/test_shells_asset_ids.py | 52 ++++++++++++++++++- 2 files changed, 69 insertions(+), 21 deletions(-) diff --git a/server/app/interfaces/repository.py b/server/app/interfaces/repository.py index fc7b4b0a0..b046d19e1 100644 --- a/server/app/interfaces/repository.py +++ b/server/app/interfaces/repository.py @@ -9,7 +9,6 @@ """ import io -import json from typing import Callable, Dict, Iterable, Iterator, List, Optional, Tuple, Type, Union import werkzeug.exceptions @@ -25,7 +24,7 @@ from app._config import API_BASE_PATH from app.interfaces.base import PagingMetadata from app.model import ServiceDescription, ServiceSpecificationProfileEnum -from app.util.converters import IdentifierToBase64URLConverter, IdShortPathConverter, base64url_decode +from app.util.converters import IdentifierToBase64URLConverter, IdShortPathConverter from .base import APIResponse, HTTPApiDecoder, ObjectStoreWSGIApp, T, is_stripped_request @@ -450,28 +449,29 @@ def _get_shells( global_asset_ids = [] for asset_id in asset_ids: - asset_id_json = base64url_decode(asset_id) - asset_dict = json.loads(asset_id_json) - try: - name = asset_dict["name"] - value = asset_dict["value"] - except KeyError as e: - raise BadRequest(f"Invalid assetId format: missing field {e}") from e - - if name == "specificAssetId": - decoded_specific_id = HTTPApiDecoder.json_list(value, model.SpecificAssetId, False, True)[0] - specific_asset_ids.append(decoded_specific_id) - elif name == "globalAssetId": - global_asset_ids.append(value) - - # Filter AAS based on both SpecificAssetIds and globalAssetIds + # Each assetIds value is a base64url-encoded SpecificAssetId (AAS Part 2, v3.1.1). + # The reserved name "globalAssetId" matches AssetInformation/globalAssetId, any other + # name matches an entry of AssetInformation/specific_asset_id. Malformed input + # raises BadRequest (400) instead of escaping as a 500. + decoded_asset_id = HTTPApiDecoder.base64url_json(asset_id, model.SpecificAssetId, False) + if decoded_asset_id.name == "globalAssetId": + global_asset_ids.append(decoded_asset_id.value) + else: + specific_asset_ids.append(decoded_asset_id) + + # Filter AAS based on both SpecificAssetIds and globalAssetIds. A queried SpecificAssetId + # matches by (name, value): full object equality would also require matching + # external_subject_id/semantic_id, which a plain {"name", "value"} query does not carry. aas = filter( lambda shell: ( ( not specific_asset_ids or all( - specific_asset_id in shell.asset_information.specific_asset_id - for specific_asset_id in specific_asset_ids + any( + stored.name == queried.name and stored.value == queried.value + for stored in shell.asset_information.specific_asset_id + ) + for queried in specific_asset_ids ) ) and (not global_asset_ids or shell.asset_information.global_asset_id in global_asset_ids) diff --git a/server/test/interfaces/test_shells_asset_ids.py b/server/test/interfaces/test_shells_asset_ids.py index b534373db..cbe7e50c2 100644 --- a/server/test/interfaces/test_shells_asset_ids.py +++ b/server/test/interfaces/test_shells_asset_ids.py @@ -22,6 +22,10 @@ def _encode_asset_id(name: str, value: str) -> str: return base64.urlsafe_b64encode(payload.encode()).decode() +def _result_ids(response) -> list: + return [shell["id"] for shell in json.loads(response.data)["result"]] + + class ShellsAssetIdsTest(unittest.TestCase): def setUp(self) -> None: self.example_data = create_full_example() @@ -37,11 +41,55 @@ def test_multiple_global_asset_ids_returns_matching_results(self) -> None: id2 = _encode_asset_id("globalAssetId", unknown_id) response = self.client.get(f"{BASE_PATH}/shells?assetIds={id1}&assetIds={id2}") self.assertEqual(200, response.status_code) - result = json.loads(response.data) - returned_ids = [r["id"] for r in result] + returned_ids = _result_ids(response) self.assertIn(aas_list[0].id, returned_ids) def test_malformed_asset_id_missing_field_returns_400(self) -> None: bad_payload = base64.urlsafe_b64encode(b'{"name": "globalAssetId"}').decode() response = self.client.get(f"{BASE_PATH}/shells?assetIds={bad_payload}") self.assertEqual(400, response.status_code) + + def test_spec_specific_asset_id_without_match_returns_empty(self) -> None: + # AAS Part 2, v3.1.1: each assetIds value is a base64url-encoded SpecificAssetId. + # A query for an id no shell carries must return an empty result, not all shells. + spec_id = _encode_asset_id("serialNumber", "does-not-exist") + response = self.client.get(f"{BASE_PATH}/shells?assetIds={spec_id}") + self.assertEqual(200, response.status_code) + self.assertEqual([], _result_ids(response)) + + def test_spec_specific_asset_id_with_match_returns_shell(self) -> None: + aas_list = [obj for obj in self.example_data if isinstance(obj, model.AssetAdministrationShell)] + known_specific_id = next(iter(aas_list[0].asset_information.specific_asset_id)) + spec_id = _encode_asset_id(known_specific_id.name, known_specific_id.value) + response = self.client.get(f"{BASE_PATH}/shells?assetIds={spec_id}") + self.assertEqual(200, response.status_code) + self.assertEqual([aas_list[0].id], _result_ids(response)) + + def test_spec_specific_asset_ids_all_must_match(self) -> None: + aas_list = [obj for obj in self.example_data if isinstance(obj, model.AssetAdministrationShell)] + known_specific_id = next(iter(aas_list[0].asset_information.specific_asset_id)) + matching_id = _encode_asset_id(known_specific_id.name, known_specific_id.value) + non_matching_id = _encode_asset_id("serialNumber", "does-not-exist") + response = self.client.get(f"{BASE_PATH}/shells?assetIds={matching_id}&assetIds={non_matching_id}") + self.assertEqual(200, response.status_code) + self.assertEqual([], _result_ids(response)) + + def test_malformed_asset_id_json_returns_400(self) -> None: + # A value that is not valid JSON must be rejected with 400, not escape as a 500. + bad_payload = base64.urlsafe_b64encode(b"foo").decode() + response = self.client.get(f"{BASE_PATH}/shells?assetIds={bad_payload}") + self.assertEqual(400, response.status_code) + + def test_old_specific_asset_id_wrapper_no_longer_matches(self) -> None: + # The pre-spec wrapper {"name": "specificAssetId", "value": ""} + # is no longer understood: it decodes as a SpecificAssetId named "specificAssetId", which no + # shell carries, so the result is empty. This documents the breaking change of this fix. + aas_list = [obj for obj in self.example_data if isinstance(obj, model.AssetAdministrationShell)] + known_specific_id = next(iter(aas_list[0].asset_information.specific_asset_id)) + wrapper = _encode_asset_id( + "specificAssetId", + json.dumps({"name": known_specific_id.name, "value": known_specific_id.value}), + ) + response = self.client.get(f"{BASE_PATH}/shells?assetIds={wrapper}") + self.assertEqual(200, response.status_code) + self.assertEqual([], _result_ids(response))