From b3f07a4df6c490f0bc0272c5a59e2337602e7a1a Mon Sep 17 00:00:00 2001 From: jabdikadyr Date: Fri, 25 Sep 2026 10:41:41 +0600 Subject: [PATCH] fix(sdk): harden spec conformance across core, agent, service, and platform --- scripts/conformance_adapter.py | 1069 ++++++++++++++--- .../adapters/asgi.py | 53 +- .../core/__init__.py | 12 +- src/agent_enrollment_protocol/core/did_web.py | 7 +- src/agent_enrollment_protocol/core/http.py | 42 +- src/agent_enrollment_protocol/core/models.py | 66 +- src/agent_enrollment_protocol/core/openapi.py | 64 + .../platform/document.py | 15 + .../platform/platform.py | 47 +- .../service/credentials.py | 14 + .../service/service.py | 10 + tests/test_asgi.py | 45 +- tests/test_conformance_falsifiability.py | 160 +++ tests/test_core_http.py | 154 ++- tests/test_core_models.py | 63 + tests/test_platform.py | 38 +- tests/test_platform_boundaries.py | 51 +- tests/test_service.py | 36 + tests/test_service_credentials.py | 27 + 19 files changed, 1735 insertions(+), 238 deletions(-) create mode 100644 tests/test_conformance_falsifiability.py diff --git a/scripts/conformance_adapter.py b/scripts/conformance_adapter.py index c19721e..c8ec3d3 100755 --- a/scripts/conformance_adapter.py +++ b/scripts/conformance_adapter.py @@ -5,18 +5,23 @@ import base64 import json import sys -from collections.abc import Mapping +from collections.abc import Callable, Mapping from dataclasses import replace from datetime import UTC, datetime, timedelta from inspect import Parameter, signature -from typing import Any +from typing import Any, cast from urllib.parse import parse_qs, urlsplit +from cryptography.hazmat.primitives.asymmetric import ec from pydantic import BaseModel from agent_enrollment_protocol.agent import ( - InspectCacheEntry, - MemoryInspectCache, + Agent, + AgentCommandError, + AgentOptions, + AssertionSigner, + AuthenticationOptions, + GrantOptions, OperationKey, PlatformCommandError, PlatformContextProvider, @@ -26,10 +31,12 @@ PlatformSignPendingError, RandomIdempotencyKeyProvider, ServiceIdentity, + ServiceSession, ) from agent_enrollment_protocol.agent.types import IdentityRequest from agent_enrollment_protocol.core import ( AEP_MEDIA_TYPE, + AEP_PROBLEM_MEDIA_TYPE, AepAssertionError, AepAuthorizationError, AepValidationError, @@ -58,6 +65,7 @@ PlatformDiscoveryDocument, PlatformLifecycleRequest, PlatformProvisionRequest, + PlatformSignCompleted, PlatformSignRequest, PlatformVerificationRequest, PlatformVerificationResponse, @@ -79,27 +87,22 @@ parse_json_model, render_authorization, require_service_origin_binding, + resolve_openapi_security, resolve_openapi_url, - same_origin, + sign_client_assertion, ) from agent_enrollment_protocol.platform import ( AuthorizationRequest, DidVerificationMethod, DiscoveryOptions, - IdempotentOperation, IdentityListQuery, IdentityRecord, MemoryIdentityStore, - MemoryPlatformIdempotencyStore, MemoryReplayStore, Platform, - PlatformIdempotencyInput, PlatformOptions, + PlatformResult, RequestContext, - create_service_scoped_agent_did, -) -from agent_enrollment_protocol.platform import ( - StoredResponse as PlatformStoredResponse, ) from agent_enrollment_protocol.service import ( AssertionVerificationContext, @@ -492,7 +495,9 @@ def evaluate_openapi(identifier: str, case: JsonObject) -> bool: scheme = parse_json_model( json.dumps(source["security_scheme"]), OpenApiAepSecurityScheme, "OpenAPI security" ) - return scheme.authentication_method == "aep-jwt" + if scheme.authentication_method != "aep-jwt": + return False + return _resolves_security_inheritance(expected) if identifier == "url-resolution": relative = resolve_openapi_url(source["final_inspect_url"], source["relative"]) cross_origin = resolve_openapi_url(source["final_inspect_url"], source["cross_origin"]) @@ -502,6 +507,35 @@ def evaluate_openapi(identifier: str, case: JsonObject) -> bool: return False +def _resolves_security_inheritance(expected: JsonObject) -> bool: + """Drives every inheritance shape expected names (CORE-OAS-011/013/014/015).""" + aep, key, mtls = "aepScheme", "apiKeyScheme", "mtlsScheme" + root: tuple[dict[str, tuple[str, ...]], ...] = ({aep: ()},) + supported = (aep, key) + shapes: tuple[tuple[str, tuple[dict[str, tuple[str, ...]], ...] | None], ...] = ( + ("root_inherited", None), + ("operation_empty_array", ()), + ("empty_requirement_alternative", ({}, {aep: ()})), + ("multiple_objects", ({aep: ()}, {key: ()})), + ("multiple_schemes_one_object", ({aep: (), key: ()},)), + ) + for name, operation in shapes: + resolved = resolve_openapi_security( + root=root, operation=operation, supported_schemes=supported + ) + if resolved.disposition.value != expected[name]: + return False + # CORE-OAS-013 and CORE-OAS-015: a compound requirement with an unsatisfiable member is + # unsupported as a whole and is never reduced to the member the Agent does support. + unsupported = resolve_openapi_security( + root=root, operation=({aep: (), mtls: ()},), supported_schemes=supported + ) + return ( + not unsupported.satisfiable + and unsupported.disposition.value == expected["unsupported_compound"] + ) + + def evaluate_authorization(identifier: str, case: JsonObject) -> bool: expected = case["expected"] if identifier == "authorization-carriers": @@ -536,78 +570,577 @@ def evaluate_authorization(identifier: str, case: JsonObject) -> bool: return False +AGENT_ORIGIN = "https://api.example.com" +Clock = Callable[[], datetime] + + +def _header(headers: Mapping[str, str], name: str) -> str | None: + expected = name.lower() + return next((value for key, value in headers.items() if key.lower() == expected), None) + + +def agent_document( + *, + commands: tuple[str, ...] = ("inspect",), + methods: tuple[str, ...] = (), + grant_types: tuple[str, ...] = (), + api_key_header: str | None = None, +) -> JsonObject: + document: JsonObject = { + "aep_version": "1.0", + "bindings": {"supported": ["http"]}, + "claims": {"optional": [], "preferred": [], "required": []}, + "commands": {"supported": list(commands)}, + "core": {"signing_algorithms": ["EdDSA", "ES256"]}, + "http": {"endpoint_base": "/aep/"}, + "identity": {"methods": ["did:web"]}, + "service": {"did": "did:web:api.example.com"}, + } + if methods: + document["authentication"] = {"methods": list(methods)} + if grant_types: + commands_value = document["commands"] + assert isinstance(commands_value, dict) + commands_value["grant_types"] = list(grant_types) + commands_value["grant_types_config"] = { + value: {"supports_per_credential_revoke": "true"} for value in grant_types + } + if api_key_header is not None: + config = commands_value["grant_types_config"] + assert isinstance(config, dict) + config["api-key"] = { + "header_names": [api_key_header], + "supports_per_credential_revoke": "true", + } + return document + + +class AgentSigner: + """An identity provider whose signer returns a well-formed assertion for its claims.""" + + def __init__(self) -> None: + self.claims: list[ClientAssertionClaims] = [] + + async def get_or_create_identity(self, request: IdentityRequest) -> ServiceIdentity: + return ServiceIdentity( + agent_did=AGENT_DID, + identity_method="did:web", + service_did=request.service_did, + signing_algorithms=(SigningAlgorithm.EDDSA,), + ) + + async def signer_for(self, identity: ServiceIdentity) -> AssertionSigner: + async def sign( + claims: ClientAssertionClaims, algorithms: tuple[SigningAlgorithm, ...] + ) -> str: + self.claims.append(claims) + header = {"alg": algorithms[0].value, "kid": claims.iss, "typ": "JWT"} + + def encode(value: object) -> str: + raw = json.dumps(value, separators=(",", ":")).encode() + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode() + + return f"{encode(header)}.{encode(claims.to_wire())}.signature" + + return sign + + +def agent_session( + *inspect_responses: HttpResponse, + commands: tuple[HttpResponse, ...] = (), + clock: Clock | None = None, + inspect_transport: QueueTransport | None = None, +) -> tuple[ServiceSession, AgentSigner, QueueTransport]: + provider = AgentSigner() + command_transport = QueueTransport(*commands) + instance = Agent( + AgentOptions( + clock=clock or (lambda: NOW), + command_transport=command_transport, + identity_provider=provider, + inspect_transport=inspect_transport or QueueTransport(*inspect_responses), + ) + ) + return instance.service(AGENT_ORIGIN), provider, command_transport + + +def inspect_response(**headers: str) -> HttpResponse: + values = {name.replace("_", "-"): value for name, value in headers.items()} + return HttpResponse( + status=200, + headers={"Content-Type": AEP_MEDIA_TYPE, **values}, + body=json.dumps(agent_document(), separators=(",", ":")).encode(), + ) + + +def agent_honours_discovery_caching(expected: JsonObject) -> bool: + """Drives the five caching behaviours expected names for a public discovery document.""" + + async def run() -> bool: + return ( + await agent_reuses_within_default_freshness(int(expected["default_freshness_seconds"])) + and expected["no_cache"] == "revalidate" + and await agent_revalidates_no_cache() + and expected["no_store"] == "do-not-persist" + and await agent_discards_no_store() + and expected["status_304"] == "reuse-validated-representation" + and await agent_reuses_a_validated_representation() + and expected["cache_key"] == "final-safe-redirect-url" + and await agent_revalidates_the_final_url() + ) + + return asyncio.run(run()) + + +async def agent_reuses_within_default_freshness(seconds: int) -> bool: + """A response carrying no Cache-Control is reusable for the SDK's own default lifetime.""" + current = [NOW] + transport = QueueTransport(inspect_response(), inspect_response()) + session, _, _ = agent_session(clock=lambda: current[0], inspect_transport=transport) + await session.inspect() + current[0] = NOW + timedelta(seconds=seconds - 1) + await session.inspect() + if len(transport.requests) != 1: + return False + current[0] = NOW + timedelta(seconds=seconds + 1) + await session.inspect() + return len(transport.requests) == 2 + + +async def agent_revalidates_no_cache() -> bool: + """no-cache stores the document but forces a conditional request before reuse.""" + transport = QueueTransport( + inspect_response(cache_control="no-cache", etag='"inspect-1"'), + HttpResponse(status=304, headers={}, body=b""), + ) + session, _, _ = agent_session(inspect_transport=transport) + await session.inspect() + await session.inspect() + return ( + len(transport.requests) == 2 + and _header(transport.requests[1].headers, "if-none-match") == '"inspect-1"' + ) + + +async def agent_discards_no_store() -> bool: + """no-store forbids persisting the document, so the next call has nothing to revalidate.""" + transport = QueueTransport( + inspect_response(cache_control="no-store", etag='"inspect-1"'), + inspect_response(cache_control="no-store", etag='"inspect-1"'), + ) + session, _, _ = agent_session(inspect_transport=transport) + await session.inspect() + await session.inspect() + return ( + len(transport.requests) == 2 + and _header(transport.requests[1].headers, "if-none-match") is None + ) + + +async def agent_reuses_a_validated_representation() -> bool: + """A 304 reuses the stored representation rather than returning an empty document.""" + transport = QueueTransport( + inspect_response(cache_control="no-cache", etag='"inspect-1"'), + HttpResponse(status=304, headers={}, body=b""), + ) + session, _, _ = agent_session(inspect_transport=transport) + await session.inspect() + revalidated = await session.inspect() + return revalidated.document.service.did == "did:web:api.example.com" + + +async def agent_revalidates_the_final_url() -> bool: + """The cache is keyed on the final URL a safe redirect settled on, not the requested one.""" + final_url = f"{AGENT_ORIGIN}/discovery/aep" + transport = QueueTransport( + HttpResponse(status=308, headers={"Location": final_url}, body=b""), + inspect_response(cache_control="no-cache", etag='"inspect-1"'), + HttpResponse(status=304, headers={}, body=b""), + ) + session, _, _ = agent_session(inspect_transport=transport) + inspection = await session.inspect() + await session.inspect() + return ( + inspection.final_url == final_url + and len(transport.requests) == 3 + and transport.requests[2].url == final_url + ) + + +def agent_headers( + document: JsonObject, options: AuthenticationOptions +) -> Mapping[str, str] | Exception: + async def run() -> Mapping[str, str] | Exception: + session, _, _ = agent_session(json_response(document)) + try: + return await session.authentication_headers(options) + except (AepAuthorizationError, AepValidationError, ValueError) as error: + return error + + return asyncio.run(run()) + + +def agent_assertion_claims( + document: JsonObject, options: AuthenticationOptions +) -> ClientAssertionClaims | None: + async def run() -> ClientAssertionClaims | None: + session, provider, _ = agent_session(json_response(document)) + try: + await session.authentication_headers(options) + except (AepAuthorizationError, AepValidationError, ValueError): + return None + return provider.claims[-1] if provider.claims else None + + return asyncio.run(run()) + + def evaluate_protected(identifier: str, case: JsonObject) -> bool: + """Drives the Agent's own protected-resource behaviour. + + These mappings evaluate the Agent role: the party that selects a carrier, mints an assertion + bound to a resource, and presents a stored credential in the field the Service selected. + """ source = case["input"] expected = case["expected"] if identifier == "api-key-wrong-header-rejected": - return bool( - source["issued_header"].lower() != source["presented_header"].lower() - and expected["accepted"] is False - ) + return agent_presents_api_key_in_the_issued_header(source, expected) if identifier == "authenticate-assertion": - claims = parse_json_model( - json.dumps(expected["claims"]), ClientAssertionClaims, "Client assertion" - ) - return claims.op is AssertionOperation.AUTHENTICATE + return agent_mints_an_authenticate_assertion(expected) if identifier == "authorization-ambiguity": - try: - parse_authorization("AEP first,AEP second", AuthorizationCarrier.DEDICATED) - except (AepAuthorizationError, AepValidationError, ValueError): - return expected["fallback"] is False and expected["selected_credential"] is None - return False + return agent_refuses_every_ambiguous_presentation(source, expected) if identifier == "authorization-field-safety": - return bool( - authorization_header_name(AuthorizationCarrier.DEDICATED).lower() - == source["field_name"].lower() - and "Authorization" in expected["strip_on_disallowed_redirect"] - and "AEP-Authorization" in expected["strip_on_disallowed_redirect"] - ) + return agent_selects_the_named_carrier(source, expected) if identifier == "authorization-payment-composition": - aep = ProtectedResourceAuthorization( - carrier=AuthorizationCarrier.DEDICATED, - scheme=AuthorizationScheme.AEP, - credentials="compact-jws", - ) - carrier, value = render_authorization(aep) - return bool( - carrier == "AEP-Authorization" - and value == "AEP compact-jws" - and expected["mpp"]["ambiguous"] is False - and expected["x402"]["ambiguous"] is False - ) + return agent_composes_with_a_payment_credential(expected) if identifier == "operation-substitution-rejected": - return set(expected["allowed"]) == { - "enroll:enroll", - "grant:grant", - "revoke:revoke", - "status:status", - "authenticate:protected-resource", - } + return agent_binds_each_assertion_to_its_operation(expected) if identifier == "redirect-safety": - return same_origin(source["source"], source["same_origin"]) and not same_origin( - source["source"], source["cross_origin"] - ) + return agent_rebinds_an_assertion_after_a_redirect(source, expected) if identifier == "unadvertised-authentication-method": - authentication = case["input"]["advertised_methods"] + return agent_fails_closed_without_an_advertised_method(source, expected) + if identifier == "assertion-and-credential-failures": + return agent_surfaces_the_service_error_code(expected) + return False + + +def agent_presents_api_key_in_the_issued_header(source: JsonObject, expected: JsonObject) -> bool: + """CORE-PR-009: an API key is presented in exactly the Service-selected header.""" + issued, presented = source["issued_header"], source["presented_header"] + document = agent_document( + commands=("inspect", "enroll", "grant"), + methods=("api-key",), + grant_types=("api-key",), + api_key_header=issued, + ) + + async def run() -> bool: + session, _, _ = agent_session( + json_response(document), + commands=( + json_response({"status": "active"}), + json_response( + { + "api_key": "opaque-api-key", + "credential_id": "credential-1", + "expires_at": "2027-01-01T00:00:00Z", + "header": issued, + "scopes": [], + } + ), + ), + ) + await session.enroll() + await session.grant(GrantOptions(grant_type="api-key")) + headers = await session.authentication_headers( + AuthenticationOptions(resource=f"{AGENT_ORIGIN}/orders") + ) + names = {name.lower() for name in headers} + # The key goes in the issued header and nowhere else, so a Service reading the presented + # header finds nothing to accept. + return names == {issued.lower()} and presented.lower() not in names + + return asyncio.run(run()) and expected["accepted"] is False + + +def agent_mints_an_authenticate_assertion(expected: JsonObject) -> bool: + """CORE-PR-024: the Agent answers a challenge with an assertion bound to the resource.""" + wanted = parse_json_model( + json.dumps(expected["claims"]), ClientAssertionClaims, "Client assertion" + ) + resource = wanted.resource + if resource is None: + return False + document = agent_document(methods=("aep-jwt",)) + claims = agent_assertion_claims(document, AuthenticationOptions(resource=resource)) + return ( + claims is not None + and claims.op is AssertionOperation.AUTHENTICATE + and claims.resource == resource + and claims.aud == "did:web:api.example.com" + and claims.iss == claims.sub + ) + + +def agent_refuses_every_ambiguous_presentation(source: JsonObject, expected: JsonObject) -> bool: + """input.cases names four ambiguous presentations; expected names one outcome for all.""" + if expected["fallback"] is not False or expected["selected_credential"] is not None: + return False + code = expected["code"] + return all(agent_refuses_presentation(name, code) for name in source["cases"]) + + +def agent_refuses_presentation(name: str, code: str) -> bool: + if name == "both-aep-recognized": + # CORE-PR-010: the Agent uses at most one carrier, so it cannot be the party that fills + # both. Whichever carrier is selected, exactly one field is rendered. + document = agent_document(methods=("aep-jwt",)) + for carrier in AuthorizationCarrier: + headers = agent_headers( + document, + AuthenticationOptions(resource=f"{AGENT_ORIGIN}/orders", carrier=carrier.value), + ) + if isinstance(headers, Exception) or set(headers) != {carrier.value}: + return False + return True + values = { + "invalid-dedicated-plus-valid-standard": "AEP", + "duplicate-dedicated-lines": "AEP first,AEP second", + "combined-dedicated-values": "AEP first Bearer second", + } + if name not in values: + raise ValueError(f"Unknown authorization ambiguity case: {name}") + try: + parse_authorization(values[name], AuthorizationCarrier.DEDICATED) + except AepAuthorizationError as error: + return bool(error.code == code) + return False + + +def agent_selects_the_named_carrier(source: JsonObject, expected: JsonObject) -> bool: + """expected.redirect_new_assertion_carrier names the field a fresh assertion goes in.""" + carrier = expected["redirect_new_assertion_carrier"] + if carrier.lower() != source["field_name"].lower(): + return False + document = agent_document(methods=("aep-jwt",)) + headers = agent_headers( + document, AuthenticationOptions(resource=f"{AGENT_ORIGIN}/orders", carrier=carrier) + ) + return not isinstance(headers, Exception) and set(headers) == {carrier} + + +def agent_composes_with_a_payment_credential(expected: JsonObject) -> bool: + """CORE-PR-014: a dedicated-carrier AEP credential composes with a payment credential.""" + document = agent_document(methods=("aep-jwt",)) + for name in ("mpp", "x402"): + composition = expected[name] + if composition["ambiguous"] is not False: + return False + aep_field = next( + field + for field, value in composition.items() + if field != "ambiguous" and str(value).split(" ", 1)[0] in {"AEP", "Bearer"} + ) + headers = agent_headers( + document, + AuthenticationOptions(resource=f"{AGENT_ORIGIN}/orders", carrier=aep_field), + ) + if isinstance(headers, Exception): + return False + # The Agent fills the field the composition names for it and claims none of the others, + # which is what leaves the payment credential beside it unambiguous. + others = {field for field in composition if field not in {"ambiguous", aep_field}} + if set(headers) != {aep_field} or others & set(headers): + return False + return True + + +def agent_binds_each_assertion_to_its_operation(expected: JsonObject) -> bool: + """Only a command's own assertion reaches it, and only authenticate carries a resource.""" + allowed = {value.split(":", 1)[0]: value.split(":", 1)[1] for value in expected["allowed"]} + + async def run() -> bool: + for command in (Command.ENROLL, Command.STATUS): + document = agent_document(commands=("inspect", command.value), methods=("aep-jwt",)) + session, provider, _ = agent_session( + json_response(document), commands=(json_response({"status": "active"}),) + ) + if command is Command.ENROLL: + await session.enroll() + else: + await session.status() + minted = provider.claims[-1] + if minted.op.value != allowed.get(command.value) or minted.resource is not None: + return False + session, provider, _ = agent_session(json_response(agent_document(methods=("aep-jwt",)))) + await session.authentication_headers( + AuthenticationOptions(resource=f"{AGENT_ORIGIN}/orders") + ) + claims = provider.claims[-1] return ( - source["unadvertised_credential"] not in authentication - and source["unadvertised_grant_type"] not in authentication - and expected["inferred_method"] is None + allowed.get("authenticate") == "protected-resource" + and claims.op is AssertionOperation.AUTHENTICATE + and claims.resource is not None ) - if identifier == "assertion-and-credential-failures": - return all( - valid( - ProblemDetails, - { - "code": code, - "status": 401, - "title": code.replace("_", " ").title(), - "type": f"urn:aep:error:{code}", - }, + + return asyncio.run(run()) + + +def agent_rebinds_an_assertion_after_a_redirect(source: JsonObject, expected: JsonObject) -> bool: + """CORE-PR-032: a same-origin redirect is followed with a new assertion bound to the target.""" + same_origin_expected = expected["same_origin"] + if same_origin_expected["credential_forwarded"] is not False: + return False + document = agent_document(methods=("aep-jwt",)) + first = agent_assertion_claims(document, AuthenticationOptions(resource=source["source"])) + target = same_origin_expected["new_authenticate_assertion_resource"] + second = agent_assertion_claims(document, AuthenticationOptions(resource=target)) + if first is None or second is None: + return False + if first.resource != source["source"] or second.resource != target: + return False + if first.jti == second.jti: + return False + # CORE-PR-031: the cross-origin resource is refused outright rather than reached with the + # credential minted for this origin. + cross = agent_headers(document, AuthenticationOptions(resource=source["cross_origin"])) + return isinstance(cross, Exception) and bool( + expected["cross_origin"]["assertion_forwarded"] is False + ) + + +def agent_fails_closed_without_an_advertised_method( + source: JsonObject, expected: JsonObject +) -> bool: + """CORE-IDM-003: a method the Service does not advertise cannot be inferred or presented.""" + advertised = tuple(source["advertised_methods"]) + document = agent_document(methods=advertised) + unadvertised = source["unadvertised_grant_type"] + if unadvertised in advertised or source["unadvertised_credential"] in advertised: + return False + refused = agent_headers( + document, + AuthenticationOptions(resource=f"{AGENT_ORIGIN}/orders", grant_type=unadvertised), + ) + if not isinstance(refused, Exception): + return False + # With no advertised method at all the Agent has nothing to present and must not invent one. + empty = agent_headers( + agent_document(), AuthenticationOptions(resource=f"{AGENT_ORIGIN}/orders") + ) + return isinstance(empty, Exception) and expected["inferred_method"] is None + + +def agent_surfaces_the_service_error_code(expected: JsonObject) -> bool: + """The Agent carries the Service's Problem Details code through rather than collapsing it.""" + document = agent_document(commands=("inspect", "status"), methods=("aep-jwt",)) + + async def run() -> bool: + for code in dict.fromkeys(expected.values()): + session, _, _ = agent_session( + json_response(document), + commands=( + json_response( + { + "code": code, + "status": 401, + "title": code.replace("_", " ").title(), + "type": f"urn:aep:error:{code}", + }, + 401, + AEP_PROBLEM_MEDIA_TYPE, + ), + ), ) - for code in set(expected.values()) + try: + await session.status() + except AgentCommandError as error: + if error.problem is None or error.problem.code != code: + return False + else: + return False + return True + + return asyncio.run(run()) + + +class SigningKeyStore: + """A key store that mints real ES256 assertions, so signing paths are genuinely exercised.""" + + def __init__(self) -> None: + self.key = ec.generate_private_key(ec.SECP256R1()) + + async def create_key(self, identity: IdentityRecord) -> None: + del identity + + async def did_verification_method(self, identity: IdentityRecord) -> DidVerificationMethod: + numbers = self.key.public_key().public_numbers() + size = (self.key.key_size + 7) // 8 + + def encode(value: int) -> str: + return base64.urlsafe_b64encode(value.to_bytes(size, "big")).rstrip(b"=").decode() + + return DidVerificationMethod( + controller=identity.agent_did, + id=identity.key_id, + public_key_jwk={ + "crv": "P-256", + "kty": "EC", + "x": encode(numbers.x), + "y": encode(numbers.y), + }, + type="JsonWebKey2020", ) - return False + + async def sign(self, identity: IdentityRecord, claims: ClientAssertionClaims) -> str: + return sign_client_assertion( + claims, key=self.key, algorithm=SigningAlgorithm.ES256, key_id=identity.key_id + ) + + async def verification_key(self, identity: IdentityRecord) -> Any: + del identity + return self.key.public_key() + + +class AllowAuthorizer: + async def authorize(self, request: AuthorizationRequest, context: RequestContext) -> bool: + del request, context + return True + + +def working_platform(**changes: Any) -> Platform: + identifiers = iter([f"pai_01J0AEPPLATFORM{index:012d}" for index in range(1, 9)]) + agent_did_ids = iter([f"agent-{index}" for index in range(1, 9)]) + values: dict[str, Any] = { + "agent_did_id_generator": lambda: next(agent_did_ids), + "authorizer": AllowAuthorizer(), + "clock": lambda: NOW, + "did_host": "p.example", + "did_path_prefix": "a", + "did_url_template": "https://p.example/a/{agent_did_id}/did.json", + "discovery": DiscoveryOptions( + endpoint_base="/v1/aep", + hosted_verification_endpoint="/v1/aep/verifications", + lifecycle_endpoint="/v1/aep/agent-identities/{agent_identity_id}", + list_endpoint="/v1/aep/agent-identities", + platform_did="did:web:p.example", + platform_name="Example Platform", + provision_endpoint="/v1/aep/agent-identities", + sign_endpoint="/v1/aep/agent-identities/{agent_identity_id}/sign", + ), + "hosted_verification": True, + "identifier": lambda: next(identifiers), + "identity_store": MemoryIdentityStore(), + "key_store": SigningKeyStore(), + "replay_store": MemoryReplayStore(), + "service_did_resolver": ResolvedServiceDid(), + "signing_algorithms": (SigningAlgorithm.ES256,), + } + values.update(changes) + return Platform(PlatformOptions(**values)) + + +def platform_context(key: str, principal: str = "stable-principal-123") -> RequestContext: + return RequestContext(current_time=NOW, idempotency_key=key, principal=principal) def evaluate_platform(identifier: str, case: JsonObject) -> bool: @@ -620,11 +1153,7 @@ def evaluate_platform(identifier: str, case: JsonObject) -> bool: if identifier in {"provision-response", "lifecycle-response"}: return valid(PlatformAgentIdentity, expected) if identifier == "provision-response-distinct-services": - first = source["first_request"]["service_did"] - second = source["second_request"]["service_did"] - return create_service_scoped_agent_did( - "platform.example", "agents", first - ) != create_service_scoped_agent_did("platform.example", "agents", second) + return asyncio.run(exercise_distinct_service_scopes(source, expected)) if identifier == "list-response": return valid(PlatformAgentIdentityListResponse, expected) if identifier == "lifecycle-request": @@ -644,10 +1173,174 @@ def evaluate_platform(identifier: str, case: JsonObject) -> bool: if identifier == "authorization-required": return asyncio.run(exercise_platform_authorization(source, expected)) if identifier in IDEMPOTENCY_CASES: - return asyncio.run(exercise_platform_idempotency()) + return asyncio.run(exercise_platform_idempotency(source, expected)) return False +async def exercise_distinct_service_scopes(source: JsonObject, expected: JsonObject) -> bool: + """PLT-DID-003: each Service gets its own Agent DID, revealing nothing about the other.""" + platform = working_platform() + responses = [] + for name in ("first_request", "second_request"): + request = source[name] + result = await platform.provision( + PlatformProvisionRequest(service_did=request["service_did"]), + platform_context(request["idempotency_key_header"]), + ) + if result.status != 200 or not isinstance(result.body, PlatformAgentIdentity): + return False + responses.append(result.body) + first, second = responses + if first.service_did == second.service_did: + return False + # The DIDs differ, and neither carries the Service DID it was scoped to, which is what keeps + # one Service from recognizing the Agent it sees at another. + for identity, request in zip(responses, ("first_request", "second_request"), strict=True): + service_did = source[request]["service_did"] + label = service_did.removeprefix("did:web:") + if label in identity.agent_did or label in identity.did_document_url: + return False + shape = {name: expected[name]["status"] for name in ("first_response", "second_response")} + return ( + first.agent_did != second.agent_did + and first.agent_identity_id != second.agent_identity_id + and first.did_document_url != second.did_document_url + and shape == {"first_response": first.status.value, "second_response": second.status.value} + ) + + +async def exercise_platform_idempotency(source: JsonObject, expected: JsonObject) -> bool: + """Every replayable operation replays its stored result and conflicts on changed material.""" + conflict_status = expected["changed_input_or_operation_status"] + conflict_code = expected["changed_input_or_operation_code"] + if expected["exact_replay"] != "stored_http_result": + return False + for operation in source["operations"]: + if not await replays_and_conflicts(operation, source, conflict_status, conflict_code): + return False + return await retains_a_stored_result(int(expected["retention_seconds_minimum"])) + + +async def replays_and_conflicts( + operation: str, source: JsonObject, conflict_status: int, conflict_code: str +) -> bool: + platform = working_platform() + principal = source["principal"] + provisioned = await platform.provision( + PlatformProvisionRequest(service_did=SERVICE_DID), + platform_context("provision-setup", principal), + ) + identity = provisioned.body + if not isinstance(identity, PlatformAgentIdentity): + return False + key = source["initial_sign_key"] + if operation == "provision": + first = await platform.provision( + PlatformProvisionRequest(service_did=SERVICE_DID), platform_context(key, principal) + ) + replay = await platform.provision( + PlatformProvisionRequest(service_did=SERVICE_DID), platform_context(key, principal) + ) + conflict = await platform.provision( + PlatformProvisionRequest(service_did="did:web:other.service.example"), + platform_context(key, principal), + ) + elif operation == "sign": + request = PlatformSignRequest( + jti="01J0AEPASSERTION0000000001", + op=AssertionOperation.ENROLL, + service_did=SERVICE_DID, + ) + first = await platform.sign( + identity.agent_identity_id, request, platform_context(key, principal) + ) + replay = await platform.sign( + identity.agent_identity_id, request, platform_context(key, principal) + ) + conflict = await platform.sign( + identity.agent_identity_id, + PlatformSignRequest( + jti="01J0AEPASSERTION0000000002", + op=AssertionOperation.ENROLL, + service_did=SERVICE_DID, + ), + platform_context(key, principal), + ) + elif operation == "hosted_verification": + signed = await platform.sign( + identity.agent_identity_id, + PlatformSignRequest( + jti="01J0AEPASSERTION0000000003", + op=AssertionOperation.ENROLL, + service_did=SERVICE_DID, + ), + platform_context("sign-setup", principal), + ) + if not isinstance(signed.body, PlatformSignCompleted): + return False + request = PlatformVerificationRequest( + client_assertion=signed.body.client_assertion, + op=AssertionOperation.ENROLL, + service_did=SERVICE_DID, + ) + first = await platform.verify(request, platform_context(key, principal)) + replay = await platform.verify(request, platform_context(key, principal)) + conflict = await platform.verify( + PlatformVerificationRequest( + client_assertion=signed.body.client_assertion, + op=AssertionOperation.STATUS, + service_did=SERVICE_DID, + ), + platform_context(key, principal), + ) + else: + raise ValueError(f"Unknown replayable Platform operation: {operation}") + return ( + first.status == 200 + and _stored_body(replay) == _stored_body(first) + and conflict.status == conflict_status + and conflict.problem is not None + and conflict.problem.code == conflict_code + ) + + +async def retains_a_stored_result(seconds: int) -> bool: + """A stored result stays replayable for at least the published retention window. + + Signing is used rather than provisioning because a second provision for the same principal and + Service returns the same identity whether or not the key was remembered, so it could not tell a + retained result from a fresh one. A forgotten sign key mints a new assertion. + """ + current = [NOW] + platform = working_platform(clock=lambda: current[0]) + provisioned = await platform.provision( + PlatformProvisionRequest(service_did=SERVICE_DID), platform_context("provision-setup") + ) + identity = provisioned.body + if not isinstance(identity, PlatformAgentIdentity): + return False + request = PlatformSignRequest( + jti="01J0AEPASSERTION0000000009", + op=AssertionOperation.ENROLL, + service_did=SERVICE_DID, + ) + first = await platform.sign(identity.agent_identity_id, request, platform_context("retained")) + current[0] = NOW + timedelta(seconds=seconds - 1) + replay = await platform.sign( + identity.agent_identity_id, + request, + replace(platform_context("retained"), current_time=current[0]), + ) + return first.status == replay.status and _stored_body(replay) == _stored_body(first) + + +def _stored_body(result: PlatformResult[Any]) -> JsonObject: + value = result.problem if result.problem is not None else result.body + if isinstance(value, BaseModel): + return cast(JsonObject, value.model_dump(by_alias=True, exclude_unset=True, mode="json")) + return cast(JsonObject, value) + + async def exercise_platform_authorization(source: JsonObject, expected: JsonObject) -> bool: authorizer = DenyAuthorizer() store = MemoryIdentityStore() @@ -1048,28 +1741,6 @@ async def operation() -> StoredResponse: ) -async def exercise_platform_idempotency() -> bool: - store = MemoryPlatformIdempotencyStore(lambda: NOW) - source = PlatformIdempotencyInput( - "same", IdempotentOperation.PROVISION, "principal", "sha256:one" - ) - - async def operation() -> PlatformStoredResponse: - return PlatformStoredResponse(200, AEP_MEDIA_TYPE, b"{}", NOW, {}) - - created = await store.execute(source, operation) - replayed = await store.execute(source, operation) - conflict = await store.execute( - PlatformIdempotencyInput("same", IdempotentOperation.PROVISION, "principal", "sha256:two"), - operation, - ) - return ( - created.state.value == "created" - and replayed.state.value == "replayed" - and conflict.state.value == "conflict" - ) - - class AssertionVerifier: async def verify( self, assertion: str, context: AssertionVerificationContext @@ -1135,14 +1806,16 @@ def service(enrollment_store: MemoryEnrollmentStore | None = None) -> Service: async def exercise_service_case(identifier: str, case: JsonObject) -> bool: if identifier == "grant-before-enroll-rejected": + expected = case["expected"] grant_result = await service().grant( b'{"grant_type":"api-key"}', command_options(AssertionOperation.GRANT, "grant", "grant"), ) return ( - grant_result.status == 401 + grant_result.status == expected["status"] and grant_result.problem is not None - and grant_result.problem.code == "not_recognized" + and grant_result.problem.code == expected["code"] + and expected["implicit_enrollment"] is False ) if identifier == "repeated-existing": source = case["input"] @@ -1180,21 +1853,12 @@ async def exercise_service_case(identifier: str, case: JsonObject) -> bool: and restored == record ) if identifier == "operation-substitution-rejected": - substitution_result = await service().enroll( - b'{"agent_did":"did:web:agent.example.com:agents:123"}', - command_options(AssertionOperation.STATUS, "substitution", "substitution"), - ) - return ( - substitution_result.status == 401 - and substitution_result.problem is not None - and substitution_result.problem.code == "not_recognized" - ) - if identifier in { - "authenticate-assertion", - "authorization-ambiguity", - "authorization-payment-composition", - "assertion-and-credential-failures", - }: + return await service_rejects_operation_substitution(case["expected"]) + if identifier == "authorization-ambiguity": + return await service_refuses_every_ambiguous_presentation(case) + if identifier == "assertion-and-credential-failures": + return await service_reports_each_failure_code(case["expected"]) + if identifier in {"authenticate-assertion", "authorization-payment-composition"}: instance = service() enrolled = await instance.enroll( b'{"agent_did":"did:web:agent.example.com:agents:123"}', @@ -1205,43 +1869,136 @@ async def exercise_service_case(identifier: str, case: JsonObject) -> bool: if enrolled.status != 200: return False resource = "https://api.example.com/orders" - headers: dict[str, str] = {} - if identifier == "authorization-ambiguity": - headers = { - "AEP-Authorization": ( - f"AEP {assertion(AssertionOperation.AUTHENTICATE, 'first', resource)}" - ), - "Authorization": ( - f"AEP {assertion(AssertionOperation.AUTHENTICATE, 'second', resource)}" - ), - } - elif identifier == "assertion-and-credential-failures": - headers = {"Authorization": "AEP malformed"} - else: - carrier = ( - "AEP-Authorization" - if identifier == "authorization-payment-composition" - else "Authorization" - ) - headers[carrier] = ( - f"AEP {assertion(AssertionOperation.AUTHENTICATE, identifier, resource)}" - ) - if identifier == "authorization-payment-composition": - headers["Authorization"] = "Payment payment-credential" + carrier = ( + "AEP-Authorization" + if identifier == "authorization-payment-composition" + else "Authorization" + ) + headers = { + carrier: f"AEP {assertion(AssertionOperation.AUTHENTICATE, identifier, resource)}" + } + if identifier == "authorization-payment-composition": + # CORE-PR-014: an unrelated payment credential beside the dedicated carrier is not + # ambiguous, and CORE-PR-015 leaves it untouched. + headers["Authorization"] = "Payment payment-credential" authentication_result = await instance.authenticate_protected_resource( ProtectedResourceRequest(headers=headers, method="GET", url=resource) ) - if identifier in {"authenticate-assertion", "authorization-payment-composition"}: - return authentication_result.authenticated - return ( - not authentication_result.authenticated - and authentication_result.response is not None - and authentication_result.response.problem is not None - and authentication_result.response.problem.code == "not_recognized" - ) + return authentication_result.authenticated return evaluate_protected(identifier, case) +async def service_rejects_operation_substitution(expected: JsonObject) -> bool: + """Only a command's own assertion reaches that command (expected.all_other_pairs).""" + code = expected["all_other_pairs"] + allowed = {value.split(":", 1)[0]: value.split(":", 1)[1] for value in expected["allowed"]} + if allowed.get("enroll") != "enroll" or allowed.get("authenticate") != "protected-resource": + return False + # An enroll assertion opens enroll; a status assertion presented to enroll does not. + accepted = await service().enroll( + b'{"agent_did":"did:web:agent.example.com:agents:123"}', + command_options(AssertionOperation.ENROLL, "allowed", "allowed"), + ) + substituted = await service().enroll( + b'{"agent_did":"did:web:agent.example.com:agents:123"}', + command_options(AssertionOperation.STATUS, "substitution", "substitution"), + ) + return ( + accepted.status == 200 + and substituted.status == 401 + and substituted.problem is not None + and substituted.problem.code == code + ) + + +async def service_refuses_every_ambiguous_presentation(case: JsonObject) -> bool: + """input.cases names four ambiguous presentations; expected names one outcome for all.""" + source, expected = case["input"], case["expected"] + if expected["fallback"] is not False or expected["selected_credential"] is not None: + return False + code = expected["code"] + resource = "https://api.example.com/orders" + for name in source["cases"]: + instance = service() + enrolled = await instance.enroll( + b'{"agent_did":"did:web:agent.example.com:agents:123"}', + command_options(AssertionOperation.ENROLL, f"enroll-{name}", f"enroll-{name}"), + ) + if enrolled.status != 200: + return False + result = await instance.authenticate_protected_resource( + ProtectedResourceRequest( + headers=_ambiguous_headers(name, resource), method="GET", url=resource + ) + ) + if ( + result.authenticated + or result.response is None + or result.response.problem is None + or result.response.problem.code != code + ): + return False + return True + + +def _ambiguous_headers(name: str, resource: str) -> dict[str, str]: + first = f"AEP {assertion(AssertionOperation.AUTHENTICATE, 'first', resource)}" + second = f"AEP {assertion(AssertionOperation.AUTHENTICATE, 'second', resource)}" + if name == "both-aep-recognized": + return {"AEP-Authorization": first, "Authorization": second} + if name == "invalid-dedicated-plus-valid-standard": + # The malformed dedicated field is the presentation; the valid standard one beside it is + # not a fallback the Service may authenticate instead. + return {"AEP-Authorization": "AEP", "Authorization": second} + if name == "duplicate-dedicated-lines": + return {"AEP-Authorization": f"{first},{second}"} + if name == "combined-dedicated-values": + return {"AEP-Authorization": f"{first} extra-credential"} + raise ValueError(f"Unknown authorization ambiguity case: {name}") + + +async def service_reports_each_failure_code(expected: JsonObject) -> bool: + """expected maps each failure mode to the code the Service reports for it.""" + resource = "https://api.example.com/orders" + modes: dict[str, dict[str, str]] = { + # CORE-PR-038: a value naming an AEP scheme is a malformed presentation, not an absent one. + "malformed_credential": {"Authorization": "AEP"}, + # A method the Service does not advertise is a different answer from an unrecognized one. + "unsupported_method": {"Authorization": "Bearer opaque-token"}, + "wrong_audience": { + "Authorization": f"AEP {assertion(AssertionOperation.AUTHENTICATE, 'aud', resource)}" + }, + } + for mode, headers in modes.items(): + instance = service() + enrolled = await instance.enroll( + b'{"agent_did":"did:web:agent.example.com:agents:123"}', + command_options(AssertionOperation.ENROLL, f"enroll-{mode}", f"enroll-{mode}"), + ) + if enrolled.status != 200: + return False + if mode == "wrong_audience": + # The assertion is well formed but minted for another Service, so the audience check + # is what refuses it. + headers = { + "Authorization": "AEP " + + assertion( + AssertionOperation.AUTHENTICATE, "elsewhere", "https://other.example/orders" + ) + } + result = await instance.authenticate_protected_resource( + ProtectedResourceRequest(headers=headers, method="GET", url=resource) + ) + if ( + result.authenticated + or result.response is None + or result.response.problem is None + or result.response.problem.code != expected[mode] + ): + return False + return True + + def evaluate_generic(role: str, category: str, identifier: str, case: JsonObject) -> bool: source = case["input"] expected = case["expected"] @@ -1259,25 +2016,7 @@ def evaluate_generic(role: str, category: str, identifier: str, case: JsonObject if role == "service" and identifier in service_cases: return asyncio.run(exercise_service_case(identifier, case)) if identifier == "public-discovery-cache": - - async def exercise_cache() -> bool: - cache = MemoryInspectCache() - document = parse_json_model( - json.dumps(minimal_document("1.0")), InspectDocument, "Inspect document" - ) - entry = InspectCacheEntry( - cached_at=datetime.now(UTC), - document=document, - final_url="https://api.example.com/discovery/aep", - cache_control="no-cache", - etag='"inspect-1"', - last_modified="Wed, 03 Sep 2026 00:00:00 GMT", - ) - await cache.save_inspect(entry.final_url, entry) - restored = await cache.find_inspect(entry.final_url) - return restored == entry and restored is not entry - - return asyncio.run(exercise_cache()) + return agent_honours_discovery_caching(expected) if category == "claims": return evaluate_claims(identifier, case) if category == "client-assertion": diff --git a/src/agent_enrollment_protocol/adapters/asgi.py b/src/agent_enrollment_protocol/adapters/asgi.py index 56d5419..fc34406 100644 --- a/src/agent_enrollment_protocol/adapters/asgi.py +++ b/src/agent_enrollment_protocol/adapters/asgi.py @@ -37,6 +37,11 @@ DEFAULT_INSPECT_CACHE_CONTROL = "public, max-age=300" DEFAULT_MAXIMUM_REQUEST_BODY_BYTES = 1 << 20 AEP_PRINCIPAL_SCOPE_KEY = "aep.principal" +# RFC 9110 section 9.1 requires every general-purpose server to support GET and HEAD, and section +# 9.3.2 makes HEAD identical to GET apart from the content. A resource that answers GET answers +# HEAD as well, so allowing only GET turned every cache revalidation and link check into a refusal. +_SAFE_METHODS = ("GET", "HEAD") +_SAFE_ALLOW = ", ".join(_SAFE_METHODS) _HTTP_TOKEN_CHARACTERS = frozenset( "!#$%&'*+-.^_`|~0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" ) @@ -111,16 +116,20 @@ async def __call__(self, scope: AsgiScope, receive: AsgiReceive, send: AsgiSend) await self._serve_command(scope, receive, send, command) async def _serve_inspect(self, scope: AsgiScope, send: AsgiSend) -> None: - if scope.get("method") != "GET": + method = scope.get("method") + if method not in _SAFE_METHODS: + # RFC 9110 section 15.5.5: a method the target resource does not support is 405, and + # section 15.5.6 requires the Allow field naming the ones it does. await _send_problem( send, - 400, - "Invalid request", + 405, + "Method not allowed", code="invalid_request", - headers={"Allow": "GET"}, + headers={"Allow": _SAFE_ALLOW}, ) return + head = method == "HEAD" headers = _request_headers(scope) if _etag_matches(headers.get("if-none-match"), self._inspect_etag): await _send_response( @@ -142,6 +151,7 @@ async def _serve_inspect(self, scope: AsgiScope, send: AsgiSend) -> None: "Content-Type": AEP_MEDIA_TYPE, "ETag": self._inspect_etag, }, + head=head, ) async def _serve_command( @@ -151,22 +161,25 @@ async def _serve_command( send: AsgiSend, command: Command, ) -> None: - expected_method = "GET" if command is Command.STATUS else "POST" - if scope.get("method") != expected_method: + safe = command is Command.STATUS + method = scope.get("method") + allowed = _SAFE_METHODS if safe else ("POST",) + if method not in allowed: await _send_problem( send, - 400, - "Invalid request", + 405, + "Method not allowed", code="invalid_request", - headers={"Allow": expected_method}, + headers={"Allow": _SAFE_ALLOW if safe else "POST"}, ) return + head = method == "HEAD" headers = _request_headers(scope) assertion = _command_assertion(headers.get("authorization")) if command is Command.STATUS: status_result = await self._service.status(CommandOptions(client_assertion=assertion)) - await _send_service_result(send, status_result) + await _send_service_result(send, status_result, head=head) return if not _single_media_type(headers.get("content-type"), AEP_MEDIA_TYPE): @@ -177,7 +190,9 @@ async def _serve_command( except _RequestDisconnected: return if body is None: - await _send_problem(send, 400, "Invalid request", code="invalid_request") + # RFC 9110 section 15.5.14: a body over the configured limit is 413, which tells the + # caller not to retry it unchanged. 400 claimed the request was malformed instead. + await _send_problem(send, 413, "Content too large", code="invalid_request") return options = CommandOptions( client_assertion=assertion, @@ -276,7 +291,9 @@ async def _read_body(receive: AsgiReceive, maximum: int) -> bytes | None: return bytes(body) -async def _send_service_result(send: AsgiSend, result: ServiceResult[Any]) -> None: +async def _send_service_result( + send: AsgiSend, result: ServiceResult[Any], *, head: bool = False +) -> None: value: object = result.problem if result.problem is not None else result.body if isinstance(value, BaseModel): value = value.model_dump(by_alias=True, exclude_none=True, mode="json") @@ -284,7 +301,7 @@ async def _send_service_result(send: AsgiSend, result: ServiceResult[Any]) -> No headers = dict(result.headers) headers["Cache-Control"] = "no-store" headers["Content-Type"] = result.content_type - await _send_response(send, result.status, body, headers) + await _send_response(send, result.status, body, headers, head=head) async def _send_problem( @@ -313,7 +330,15 @@ async def _send_response( status: int, body: bytes, headers: Mapping[str, str], + *, + head: bool = False, ) -> None: + """Sends one response, omitting the content when the request method was HEAD. + + RFC 9110 section 9.3.2 keeps the header fields a HEAD response carries identical to the GET + it stands in for, including an accurate Content-Length, so a caller sizing a resource or + revalidating a cached copy reads the same metadata either way. + """ values = { name.lower(): value for name, value in headers.items() if name.lower() != "content-length" } @@ -331,7 +356,7 @@ async def _send_response( raise ValueError("AEP ASGI response contains an invalid HTTP field") encoded.append((name.encode("ascii"), value.encode("latin-1"))) await send({"type": "http.response.start", "status": status, "headers": encoded}) - await send({"type": "http.response.body", "body": body}) + await send({"type": "http.response.body", "body": b"" if head else body}) def _request_headers(scope: Mapping[str, Any]) -> Mapping[str, str | Sequence[str]]: diff --git a/src/agent_enrollment_protocol/core/__init__.py b/src/agent_enrollment_protocol/core/__init__.py index 7dfc12f..20ab76d 100644 --- a/src/agent_enrollment_protocol/core/__init__.py +++ b/src/agent_enrollment_protocol/core/__init__.py @@ -120,7 +120,14 @@ SigningAlgorithm, StatusResponse, ) -from .openapi import OpenApiPathMatch, match_openapi_path, resolve_openapi_url +from .openapi import ( + OpenApiPathMatch, + OpenApiSecurityDisposition, + OpenApiSecurityResolution, + match_openapi_path, + resolve_openapi_security, + resolve_openapi_url, +) from .validation import ( ValidationResult, parse_json_model, @@ -195,6 +202,8 @@ "OpenApiPathMatch", "OpenApiPathMatching", "OpenApiReference", + "OpenApiSecurityDisposition", + "OpenApiSecurityResolution", "OpenApiTrailingSlash", "PlatformAgentIdentity", "PlatformAgentIdentityListResponse", @@ -241,6 +250,7 @@ "parse_platform_sign_response", "render_authorization", "require_service_origin_binding", + "resolve_openapi_security", "resolve_openapi_url", "same_origin", "select_did_web_public_jwk", diff --git a/src/agent_enrollment_protocol/core/did_web.py b/src/agent_enrollment_protocol/core/did_web.py index 693ec50..b3c30ee 100644 --- a/src/agent_enrollment_protocol/core/did_web.py +++ b/src/agent_enrollment_protocol/core/did_web.py @@ -7,6 +7,7 @@ from urllib.parse import unquote, urlsplit _INVALID_PERCENT_ENCODING = re.compile(r"%(?![0-9A-Fa-f]{2})") +_VISIBLE_ASCII_PATTERN = re.compile(r"\A[\x21-\x7e]+\Z") def did_web_document_url(did: str, *, allow_insecure_loopback: bool = False) -> str: @@ -53,7 +54,11 @@ def _did_web_parts(did: str) -> tuple[str, tuple[str, ...]]: ): raise ValueError(f"Invalid did:web identifier: {did}") host = unquote(encoded_parts[0]) - if not host.isascii(): + # urlsplit strips CR, LF, and tab before parsing, so a host carrying one would be inspected in + # its sanitized form and then concatenated into the document URL in its raw form. Requiring + # visible US-ASCII here keeps the value that is validated and the value that is used identical, + # and keeps whitespace and control characters out of a URL handed to an HTTP client. + if not host.isascii() or not _VISIBLE_ASCII_PATTERN.fullmatch(host): raise ValueError(f"Invalid did:web identifier: {did}") parsed = urlsplit(f"//{host}") try: diff --git a/src/agent_enrollment_protocol/core/http.py b/src/agent_enrollment_protocol/core/http.py index 77b6250..c3b948e 100644 --- a/src/agent_enrollment_protocol/core/http.py +++ b/src/agent_enrollment_protocol/core/http.py @@ -15,6 +15,14 @@ ProtectedResourceAuthorization, ) +_CREDENTIALS_PATTERN = re.compile(r"\A[\x21-\x7e]+\Z") +# CORE-AUTH-001 spells the presentation as `"AEP" 1*SP compact-jws`, so repeated separators belong +# to the grammar, a field value may carry trailing optional whitespace, and the credential itself is +# visible US-ASCII with no space. Requiring a single separator rejected a conforming presentation, +# and allowing a space inside the credential let one field value carry two of them. +_AUTHORIZATION_PATTERN = re.compile( + r"\A(AEP|Bearer|Basic)[ \t]+([\x21-\x7e]+)[ \t]*\Z", re.IGNORECASE +) _TOKEN = r"[!#$%&'*+.^_`|~0-9A-Za-z-]+" _QUOTED_STRING = r'"(?:[\t !#-\[\]-~\x80-\xff]|\\[\t !-~\x80-\xff])*"' _MEDIA_TYPE_PATTERN = re.compile( @@ -52,11 +60,25 @@ def media_type_essence(value: str) -> str: def normalize_endpoint_base(endpoint_base: str = DEFAULT_HTTP_ENDPOINT_BASE) -> str: - if not endpoint_base.startswith("/") or endpoint_base.startswith("//"): + if not is_endpoint_path(endpoint_base): raise ValueError("AEP endpoint_base must be an origin-relative absolute path") return endpoint_base if endpoint_base.endswith("/") else f"{endpoint_base}/" +def is_endpoint_path(value: str) -> bool: + """Reports whether a value can carry a command path appended to it. + + CORE-HTTP-004 builds a command URL by appending the relative path with exactly one `/` + separator. A base carrying a query, a fragment, or a dot segment cannot satisfy that: the + command name would land in the query string or walk out of the advertised base. + """ + if not value.startswith("/") or value.startswith("//"): + return False + if "?" in value or "#" in value: + return False + return not any(segment in {".", ".."} for segment in value.split("/")) + + def command_path(command: Command, endpoint_base: str = DEFAULT_HTTP_ENDPOINT_BASE) -> str: if command is Command.INSPECT: raise ValueError("Inspect does not have a command endpoint path") @@ -72,6 +94,13 @@ def render_authorization(value: ProtectedResourceAuthorization) -> tuple[str, st raise AepAuthorizationError( "Authorization credentials must not be empty.", "invalid_request" ) + # Visible US-ASCII excluding space keeps CR, LF, NUL, and other control characters out of a + # rendered field value, and keeps one field value from carrying two credentials. + if _CREDENTIALS_PATTERN.fullmatch(value.credentials) is None: + raise AepAuthorizationError( + "Authorization credentials must not contain whitespace or control characters.", + "invalid_request", + ) return value.carrier.value, f"{value.scheme.value} {value.credentials}" @@ -82,17 +111,16 @@ def parse_authorization( raise AepAuthorizationError( "The dedicated authorization field is ambiguous.", "not_recognized" ) - scheme_text, separator, credentials = value.partition(" ") - schemes = {item.value.lower(): item for item in AuthorizationScheme} - scheme = schemes.get(scheme_text.lower()) - if not separator or scheme is None or not credentials or credentials[0].isspace(): + match = _AUTHORIZATION_PATTERN.fullmatch(value) + if match is None: raise AepAuthorizationError( "The authorization presentation was not recognized.", "not_recognized" ) + schemes = {item.value.lower(): item for item in AuthorizationScheme} return ProtectedResourceAuthorization( carrier=carrier, - credentials=credentials, - scheme=scheme, + credentials=match.group(2), + scheme=schemes[match.group(1).lower()], ) diff --git a/src/agent_enrollment_protocol/core/models.py b/src/agent_enrollment_protocol/core/models.py index 77aad42..234b727 100644 --- a/src/agent_enrollment_protocol/core/models.py +++ b/src/agent_enrollment_protocol/core/models.py @@ -7,6 +7,7 @@ from email.headerregistry import Address from enum import StrEnum from typing import Annotated, Any, ClassVar, Literal, Self +from urllib.parse import urlsplit from pydantic import ( BaseModel, @@ -247,9 +248,10 @@ class HttpConfiguration(AepModel): @model_validator(mode="after") def validate_endpoint_base(self) -> Self: - if self.endpoint_base is not None and ( - not self.endpoint_base.startswith("/") or self.endpoint_base.startswith("//") - ): + # CORE-HTTP-004 appends the command path to this value with one `/` separator, so a base + # carrying a query, a fragment, or a dot segment would steer every command URL away from + # the advertised base. + if self.endpoint_base is not None and not _is_endpoint_path(self.endpoint_base): raise ValueError("endpoint_base must be an origin-relative absolute path") return self @@ -591,7 +593,7 @@ def validate_paths(self) -> Self: self.list, self.hosted_verification, ): - if value is not None and (not value.startswith("/") or value.startswith("//")): + if value is not None and not _is_endpoint_path(value): raise ValueError("Platform endpoint paths must be origin-relative absolute paths") if self.lifecycle.count("{agent_identity_id}") != 1: raise ValueError("lifecycle must contain one {agent_identity_id} placeholder") @@ -609,11 +611,7 @@ class PlatformHttp(AepModel): @model_validator(mode="after") def validate_endpoint_base(self) -> Self: - if ( - not self.endpoint_base.startswith("/") - or self.endpoint_base.startswith("//") - or "{" in self.endpoint_base - ): + if not _is_endpoint_path(self.endpoint_base) or "{" in self.endpoint_base: raise ValueError("endpoint_base must be an origin-relative absolute path") return self @@ -845,6 +843,14 @@ def parse_full_date(value: str) -> date: return date.fromisoformat(value) +def _is_endpoint_path(value: str) -> bool: + if not value.startswith("/") or value.startswith("//"): + return False + if "?" in value or "#" in value: + return False + return not any(segment in {".", ".."} for segment in value.split("/")) + + def _require_identifiers(values: tuple[str, ...], name: str, *, unique: bool = False) -> None: if any(IDENTIFIER_PATTERN.fullmatch(value) is None for value in values): raise ValueError(f"{name} uses the registered identifier syntax") @@ -865,16 +871,18 @@ def _require_did(value: str, name: str) -> None: def _require_https_uri(value: str, name: str) -> None: - from urllib.parse import urlsplit - parsed = urlsplit(value) - if parsed.scheme != "https" or not parsed.hostname or parsed.fragment: - raise ValueError(f"{name} must be an absolute HTTPS URI without a fragment") + if ( + parsed.scheme != "https" + or not parsed.hostname + or parsed.fragment + or parsed.username + or parsed.password + ): + raise ValueError(f"{name} must be an absolute HTTPS URI without a fragment or user info") def _require_resource_uri(value: str, name: str, *, allow_insecure_loopback: bool = False) -> None: - from urllib.parse import urlsplit - parsed = urlsplit(value) secure = parsed.scheme == "https" loopback = ( @@ -882,8 +890,17 @@ def _require_resource_uri(value: str, name: str, *, allow_insecure_loopback: boo and parsed.scheme == "http" and parsed.hostname in {"localhost", "127.0.0.1", "::1"} ) - if (not secure and not loopback) or not parsed.hostname or parsed.fragment: - raise ValueError(f"{name} must be an absolute HTTPS URI without a fragment") + # User information in a resource URL carries credentials the assertion is not bound to and + # makes the authority ambiguous to a reader, so it is refused here as it is on every other + # URL this package accepts. + if ( + (not secure and not loopback) + or not parsed.hostname + or parsed.fragment + or parsed.username + or parsed.password + ): + raise ValueError(f"{name} must be an absolute HTTPS URI without a fragment or user info") def _validate_operation_resource(op: AssertionOperation, resource: str | None) -> None: @@ -896,7 +913,10 @@ def _validate_operation_resource(op: AssertionOperation, resource: str | None) - def _is_mailbox(value: str) -> bool: - if len(value.encode()) > 320 or "@" not in value: + # CLM-CAT-001 binds contact.email to the RFC 5321 Mailbox rule as the JSON Schema `email` + # format references it, which is ASCII. A Unicode mailbox is the separate `idn-email` format, + # and accepting one here would let exámple.com stand in for example.com. + if not value.isascii() or len(value.encode()) > 320 or "@" not in value: return False try: address = Address(addr_spec=value) @@ -911,13 +931,17 @@ def _is_mailbox(value: str) -> bool: return all( label and len(label) <= 63 - and label[0].isalnum() - and label[-1].isalnum() - and all(character.isalnum() or character == "-" for character in label) + and _is_ascii_alphanumeric(label[0]) + and _is_ascii_alphanumeric(label[-1]) + and all(_is_ascii_alphanumeric(character) or character == "-" for character in label) for label in labels ) +def _is_ascii_alphanumeric(value: str) -> bool: + return value.isascii() and value.isalnum() + + def _is_absolute_uri(value: str) -> bool: if any(character.isspace() for character in value): return False diff --git a/src/agent_enrollment_protocol/core/openapi.py b/src/agent_enrollment_protocol/core/openapi.py index e77754c..73b94e3 100644 --- a/src/agent_enrollment_protocol/core/openapi.py +++ b/src/agent_enrollment_protocol/core/openapi.py @@ -1,13 +1,17 @@ from __future__ import annotations import re +from collections.abc import Collection, Mapping, Sequence from dataclasses import dataclass +from enum import StrEnum from urllib.parse import urljoin, urlsplit from .models import OpenApiTrailingSlash _PATH_EXPRESSION = re.compile(r"\{[^{}]+\}") +SecurityRequirements = Sequence[Mapping[str, Sequence[str]]] + @dataclass(frozen=True, slots=True) class OpenApiPathMatch: @@ -15,6 +19,66 @@ class OpenApiPathMatch: template: str +class OpenApiSecurityDisposition(StrEnum): + """How an Agent must treat the security an operation resolves to.""" + + PUBLIC = "public" + OPTIONAL_ANONYMOUS = "optional-anonymous" + REQUIRED = "required" + ALTERNATIVES = "alternatives" + ALL_REQUIRED = "all-required" + FALLBACK_LIVE_CHALLENGE = "fallback-live-challenge" + + +@dataclass(frozen=True, slots=True) +class OpenApiSecurityResolution: + """The resolved security for one operation. + + `satisfiable` lists the complete requirement objects the Agent can satisfy, each as the scheme + names it must present together. It is empty unless the disposition selects credentials. + """ + + disposition: OpenApiSecurityDisposition + satisfiable: tuple[tuple[str, ...], ...] = () + + +def resolve_openapi_security( + *, + root: SecurityRequirements | None, + operation: SecurityRequirements | None, + supported_schemes: Collection[str], +) -> OpenApiSecurityResolution: + """Resolves the effective security for one operation (CORE-OAS-011/013/014/015). + + Operation-level security replaces the root value rather than adding to it, an empty list makes + the operation public, an empty requirement object is an anonymous alternative, several + requirement objects are alternatives, and several schemes inside one object are compound. An + Agent that cannot satisfy every member of a compound requirement treats that requirement as + unsupported and never reduces it to the members it does support, so the requirements reported + as satisfiable are always whole requirement objects. + """ + supported = frozenset(supported_schemes) + effective = operation if operation is not None else (root if root is not None else ()) + if not effective: + return OpenApiSecurityResolution(OpenApiSecurityDisposition.PUBLIC) + requirements = tuple(tuple(requirement) for requirement in effective) + if any(not requirement for requirement in requirements): + return OpenApiSecurityResolution(OpenApiSecurityDisposition.OPTIONAL_ANONYMOUS) + satisfiable = tuple( + requirement for requirement in requirements if supported.issuperset(requirement) + ) + if not satisfiable: + return OpenApiSecurityResolution(OpenApiSecurityDisposition.FALLBACK_LIVE_CHALLENGE) + if len(requirements) > 1: + return OpenApiSecurityResolution(OpenApiSecurityDisposition.ALTERNATIVES, satisfiable) + disposition = ( + OpenApiSecurityDisposition.ALL_REQUIRED + if len(satisfiable[0]) > 1 + else OpenApiSecurityDisposition.REQUIRED + ) + return OpenApiSecurityResolution(disposition, satisfiable) + + def match_openapi_path( templates: tuple[str, ...], *, diff --git a/src/agent_enrollment_protocol/platform/document.py b/src/agent_enrollment_protocol/platform/document.py index e2832ef..9796ff1 100644 --- a/src/agent_enrollment_protocol/platform/document.py +++ b/src/agent_enrollment_protocol/platform/document.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Mapping from typing import Any from urllib.parse import quote, urlsplit @@ -22,6 +23,8 @@ WELL_KNOWN_PATH = AEP_PLATFORM_WELL_KNOWN_PATH _DID_CONTEXT = "https://www.w3.org/ns/did/v1" _DID_PLACEHOLDER = "{agent_did_id}" +# RFC 7517 and RFC 7518 private key members, plus the symmetric key value. +_PRIVATE_JWK_MEMBERS = frozenset({"d", "dp", "dq", "k", "oth", "p", "q", "qi"}) def create_service_scoped_agent_did(host: str, path_prefix: str, agent_did_id: str) -> str: @@ -54,6 +57,7 @@ def create_did_document(identity: IdentityRecord, method: DidVerificationMethod) or not method.public_key_jwk ): raise ValueError("AEP Platform DID verification method does not match the managed identity") + _require_public_jwk(method.public_key_jwk) return { "@context": [_DID_CONTEXT], "assertionMethod": [method.id], @@ -154,5 +158,16 @@ def _validate_endpoint_path(name: str, path: str) -> None: raise ValueError(f"AEP Platform {name} endpoint must be an absolute path") +def _require_public_jwk(jwk: Mapping[str, Any]) -> None: + """Refuses to publish anything but public verification material (PLT-DID-002). + + A DID document is served to anyone who resolves the Agent DID. A key store that returns a JWK + still carrying its private members would publish the signing key itself, so the document is + never built from one. + """ + if jwk.get("kty") == "oct" or _PRIVATE_JWK_MEMBERS & set(jwk): + raise ValueError("AEP Platform DID verification method must carry only public key material") + + def _encode_did_component(value: str) -> str: return quote(value, safe="-._~!$&'()*+,;=") diff --git a/src/agent_enrollment_protocol/platform/platform.py b/src/agent_enrollment_protocol/platform/platform.py index 71371d0..fcc7110 100644 --- a/src/agent_enrollment_protocol/platform/platform.py +++ b/src/agent_enrollment_protocol/platform/platform.py @@ -6,6 +6,7 @@ from collections.abc import Awaitable, Callable from dataclasses import replace from datetime import UTC, datetime, timedelta +from email.utils import format_datetime from typing import Any, TypeVar, cast from pydantic import BaseModel @@ -124,9 +125,24 @@ def __init__(self, options: PlatformOptions) -> None: self._service_did_resolver = options.service_did_resolver self._sign_handler = options.sign_handler self._signing_algorithms = options.signing_algorithms + published = json.dumps( + self._discovery.to_wire(), sort_keys=True, separators=(",", ":") + ).encode() + self._discovery_etag = f'"{hashlib.sha256(published).hexdigest()[:32]}"' + self._discovery_last_modified = _http_date(_aware(self._clock())) def discovery(self) -> PlatformResult[PlatformDiscoveryDocument]: - return _success(200, self._discovery, {"Cache-Control": "max-age=300"}) + # PLT-DISC-012: an Agent honouring PLT-DISC-013 revalidates with If-None-Match and + # If-Modified-Since, which needs the Platform to publish validators alongside freshness. + return _success( + 200, + self._discovery, + { + "Cache-Control": "max-age=300", + "ETag": self._discovery_etag, + "Last-Modified": self._discovery_last_modified, + }, + ) async def did_document(self, agent_did_id: str) -> PlatformResult[dict[str, Any]]: identity = await self._identity_store.find_by_agent_did_id(agent_did_id) @@ -226,6 +242,13 @@ async def sign( context: RequestContext, ) -> PlatformResult[PlatformSignResponse]: lifetime = self._sign_lifetime(request) + if lifetime is None: + # PLT-SIGN-011 lets a Platform enforce a shorter local maximum. A caller-supplied value + # above it is a rejected request, answered ahead of the idempotency store so an invalid + # request neither consumes a key nor escapes as an unhandled failure. + return _problem( + 400, "invalid_request", "lifetime_seconds exceeds the configured maximum" + ) material = {"agent_identity_id": agent_identity_id, "request": request.to_wire()} return await self._idempotent( IdempotentOperation.SIGN, @@ -468,15 +491,13 @@ async def _authorized_identity( return None return identity - def _sign_lifetime(self, request: PlatformSignRequest) -> int: + def _sign_lifetime(self, request: PlatformSignRequest) -> int | None: lifetime = ( self._default_lifetime if request.lifetime_seconds is None else int(request.lifetime_seconds) ) - if lifetime > self._maximum_lifetime: - raise ValueError("lifetime_seconds exceeds the configured maximum") - return lifetime + return None if lifetime > self._maximum_lifetime else lifetime def _request_time(self, context: RequestContext) -> datetime: return context.current_time or self._clock() @@ -662,12 +683,26 @@ def _is_did(value: str) -> bool: ) +def _http_date(value: datetime) -> str: + return format_datetime(value.astimezone(UTC), usegmt=True) + + def _rfc3339(value: datetime) -> str: return value.astimezone(UTC).isoformat(timespec="seconds").replace("+00:00", "Z") def _lifecycle_error(status: ManagedAgentStatus) -> str: - return f"identity_{status.value}" + """Maps a managed status to one of the registered AEP error codes (CORE-ERR-009). + + The registry defines `identity_suspended`, `identity_terminated`, and `identity_unavailable`; + there is no `identity_revoked` or `identity_active`, so a revoked identity reports the + suspended code and anything else reports the non-punitive one. + """ + if status is ManagedAgentStatus.TERMINATED: + return "identity_terminated" + if status in {ManagedAgentStatus.SUSPENDED, ManagedAgentStatus.REVOKED}: + return "identity_suspended" + return "identity_unavailable" def _clone_model(value: ModelT) -> ModelT: diff --git a/src/agent_enrollment_protocol/service/credentials.py b/src/agent_enrollment_protocol/service/credentials.py index 21fcd81..1a98fae 100644 --- a/src/agent_enrollment_protocol/service/credentials.py +++ b/src/agent_enrollment_protocol/service/credentials.py @@ -98,6 +98,7 @@ async def grant(self, request: GrantRequest, context: GrantContext) -> bytes: credential = await self._issue(request, context) parsed = _validate_built_in_credential(self._grant_type, credential) _validate_issued_credential_config(parsed, self._config) + _require_encodable_basic(parsed) expires_at = _parse_expiry(parsed.expires_at) if expires_at <= context.current_time: raise ValueError("AEP issued credential must expire after issuance") @@ -325,6 +326,19 @@ def _validate_issued_credential_config( raise ValueError("AEP issued API-key header is not advertised by the Service") +def _require_encodable_basic(credential: BuiltInCredential) -> None: + """BSC-RSP-003: issued Basic credentials must survive RFC 7617 encoding unchanged. + + RFC 7617 leaves the user-pass charset undefined without a `charset` parameter, so a client + encoding a non-ASCII value as ISO-8859-1 and a Service storing it as UTF-8 derive different + bytes from the same credential and authentication fails. + """ + if not isinstance(credential, BasicGrantResponse): + return + if not credential.username.isascii() or not credential.password.isascii(): + raise ValueError("AEP issued Basic credentials must be encodable as US-ASCII") + + def _configured_api_key_headers(config: GrantTypeConfig) -> frozenset[str] | None: value = config.to_wire().get("header_names") if value is None: diff --git a/src/agent_enrollment_protocol/service/service.py b/src/agent_enrollment_protocol/service/service.py index 85b9534..07ac589 100644 --- a/src/agent_enrollment_protocol/service/service.py +++ b/src/agent_enrollment_protocol/service/service.py @@ -750,6 +750,11 @@ def _select_presentation( if any(_recognized(value) for value in standard): return None, False return selected, True + # CORE-PR-038 separates a missing credential from a malformed one. An Authorization value + # naming an AEP scheme is an AEP presentation even when the rest of it does not parse, so it + # fails closed here; values belonging to other protocols stay available to them. + if any(_names_aep_scheme(value) and not _recognized(value) for value in standard): + return None, False recognized = [value for value in standard if _recognized(value)] if len(recognized) > 1 or (recognized and len(standard) > 1): return None, False @@ -766,6 +771,11 @@ def _recognized(value: str) -> bool: return False +def _names_aep_scheme(value: str) -> bool: + scheme = value.split(" ", 1)[0].split("\t", 1)[0] + return scheme.lower() in {item.value.lower() for item in AuthorizationScheme} + + def _absolute_url(value: str | None, allow_insecure_loopback: bool) -> str: if value is None: raise ValueError("AEP URL is required") diff --git a/tests/test_asgi.py b/tests/test_asgi.py index 1bcd521..e99b901 100644 --- a/tests/test_asgi.py +++ b/tests/test_asgi.py @@ -142,17 +142,37 @@ async def test_asgi_application_serves_and_revalidates_inspect() -> None: ) assert wildcard.status == 304 + # RFC 9110 section 15.5.5 answers an unsupported method with 405, and section 15.5.6 requires + # the Allow field to name the ones the resource does support. rejected = await invoke(application, method="POST", path="/.well-known/aep") - assert rejected.status == 400 - assert rejected.headers["allow"] == "GET" + assert rejected.status == 405 + assert rejected.headers["allow"] == "GET, HEAD" assert rejected.headers["content-type"] == "application/problem+json" assert json.loads(rejected.body) == { "code": "invalid_request", - "status": 400, - "title": "Invalid request", + "status": 405, + "title": "Method not allowed", "type": "urn:aep:error:invalid_request", } + # RFC 9110 section 9.1 requires HEAD wherever GET is supported, and section 9.3.2 keeps its + # header fields identical to the GET it stands in for, content aside. + served = await invoke(application, method="GET", path="/.well-known/aep") + head = await invoke(application, method="HEAD", path="/.well-known/aep") + assert head.status == served.status == 200 + assert head.headers == served.headers + assert head.headers["content-length"] == str(len(served.body)) + assert head.body == b"" + + conditional_head = await invoke( + application, + method="HEAD", + path="/.well-known/aep", + headers=[(b"if-none-match", served.headers["etag"].encode())], + ) + assert conditional_head.status == 304 + assert conditional_head.body == b"" + @pytest.mark.asyncio async def test_asgi_application_delegates_or_returns_not_found() -> None: @@ -267,9 +287,19 @@ async def test_asgi_application_rejects_invalid_command_requests() -> None: application = AepAsgiApplication(service, maximum_request_body_bytes=4) wrong_method = await invoke(application, method="GET", path="/aep/enroll") - assert wrong_method.status == 400 + assert wrong_method.status == 405 assert wrong_method.headers["allow"] == "POST" + # Status is safe, so it answers HEAD as well as GET and names both in Allow. + status_head = await invoke(application, method="HEAD", path="/aep/status") + status_get = await invoke(application, method="GET", path="/aep/status") + assert status_head.status == status_get.status + assert status_head.headers == status_get.headers + assert status_head.body == b"" + status_method = await invoke(application, method="POST", path="/aep/status") + assert status_method.status == 405 + assert status_method.headers["allow"] == "GET, HEAD" + for headers in ( [], [(b"content-type", b"application/json")], @@ -294,8 +324,11 @@ async def test_asgi_application_rejects_invalid_command_requests() -> None: method="POST", path="/aep/enroll", ) - assert large.status == 400 + # RFC 9110 section 15.5.14: a body over the configured limit is 413, which tells the caller not + # to retry it unchanged, rather than 400 claiming the request itself was malformed. + assert large.status == 413 assert json.loads(large.body)["code"] == "invalid_request" + assert json.loads(large.body)["title"] == "Content too large" disconnected = await invoke( application, diff --git a/tests/test_conformance_falsifiability.py b/tests/test_conformance_falsifiability.py new file mode 100644 index 0000000..4241a84 --- /dev/null +++ b/tests/test_conformance_falsifiability.py @@ -0,0 +1,160 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +from copy import deepcopy +from pathlib import Path +from typing import Any + +import pytest + +REPOSITORY = Path(__file__).resolve().parent.parent +ADAPTER = REPOSITORY / "scripts" / "conformance_adapter.py" + +JsonObject = dict[str, Any] +Corruption = tuple[str, str, str, str, Any] + +# Each entry corrupts one member of a vector's published expectation. A mapping that reads its +# expectation flips from passed to failed; a mapping that compares two constants, or two fields of +# the vector to each other, stays passed and fails this test. +CORRUPTIONS: tuple[Corruption, ...] = ( + ("agent", "protected-resource", "api-key-wrong-header-rejected", "accepted", True), + ("agent", "protected-resource", "authorization-ambiguity", "fallback", True), + ("agent", "protected-resource", "authorization-ambiguity", "code", "invalid_request"), + ( + "agent", + "protected-resource", + "authorization-field-safety", + "redirect_new_assertion_carrier", + "Authorization", + ), + ("agent", "protected-resource", "unadvertised-authentication-method", "inferred_method", "x"), + ("agent", "openapi", "security-inheritance", "multiple_schemes_one_object", "required"), + ("agent", "openapi", "security-inheritance", "unsupported_compound", "required"), + ("agent", "openapi", "security-inheritance", "root_inherited", "public"), + ("agent", "caching", "public-discovery-cache", "default_freshness_seconds", "30"), + ( + "platform", + "platform", + "idempotency-replay-conflict", + "changed_input_or_operation_status", + 400, + ), + ( + "platform", + "platform", + "idempotency-replay-conflict", + "changed_input_or_operation_code", + "invalid_request", + ), + ("platform", "platform", "idempotency-replay-conflict", "retention_seconds_minimum", "36000"), + ("platform", "platform", "idempotency-replay-conflict", "exact_replay", "fresh_result"), + ( + "platform", + "platform", + "provision-response-distinct-services", + "first_response", + {"status": "revoked"}, + ), + ("service", "grant-revoke", "grant-before-enroll-rejected", "code", "invalid_request"), + ("service", "grant-revoke", "grant-before-enroll-rejected", "status", 400), + ( + "service", + "protected-resource", + "operation-substitution-rejected", + "all_other_pairs", + "authentication_required", + ), + ("service", "protected-resource", "authorization-ambiguity", "code", "authentication_required"), + ("service", "protected-resource", "authorization-ambiguity", "fallback", True), + ( + "service", + "protected-resource", + "assertion-and-credential-failures", + "malformed_credential", + "authentication_required", + ), + ( + "service", + "protected-resource", + "assertion-and-credential-failures", + "unsupported_method", + "not_recognized", + ), +) + + +def _spec_vectors() -> Path | None: + configured = os.environ.get("AEP_SPECS_DIR") + candidates = [Path(configured)] if configured else [] + candidates += [REPOSITORY.parent / "aep-specs", REPOSITORY / ".conformance" / "aep-specs"] + for candidate in candidates: + vectors = candidate / "ietf" / "test-vectors" + if vectors.is_dir(): + return vectors + return None + + +def _request(sequence: int, role: str, vector: JsonObject, case: JsonObject) -> JsonObject: + applicability = vector["applicability"][role] + return { + "sequence": sequence, + "role": role, + "profile": applicability.get("profile", ""), + "expectation": applicability.get("expectation", ""), + "vector": {"category": vector["category"], "id": vector["id"]}, + "case": case, + } + + +def _evaluate(role: str, requests: list[JsonObject]) -> list[str]: + completed = subprocess.run( + [sys.executable, str(ADAPTER), role], + input="\n".join(json.dumps(item) for item in requests), + capture_output=True, + text=True, + timeout=180, + check=False, + ) + if completed.returncode != 0: + raise AssertionError(f"conformance adapter failed: {completed.stderr[-2000:]}") + return [json.loads(line)["status"] for line in completed.stdout.splitlines() if line.strip()] + + +def test_conformance_mappings_read_the_expectation_they_publish() -> None: + """A conformance mapping that cannot fail reports every implementation as conformant. + + Each case runs twice in one adapter process: once against the vector the specification + publishes, and once against the same vector with one member of its expectation corrupted. + """ + vectors = _spec_vectors() + if vectors is None: + pytest.skip( + "aep-specs is not checked out; set AEP_SPECS_DIR to run the falsifiability check" + ) + by_role: dict[str, tuple[list[JsonObject], list[str]]] = {} + for role, category, identifier, member, replacement in CORRUPTIONS: + vector = json.loads((vectors / category / f"{identifier}.json").read_text()) + published = {"input": vector.get("input", {}), "expected": vector.get("expected", {})} + assert member in published["expected"], f"{identifier}: {member} is not a published member" + corrupted = deepcopy(published) + corrupted["expected"][member] = replacement + requests, labels = by_role.setdefault(role, ([], [])) + requests.append(_request(len(requests) + 1, role, vector, published)) + labels.append(f"{role} {identifier}.{member}: published") + requests.append(_request(len(requests) + 1, role, vector, corrupted)) + labels.append(f"{role} {identifier}.{member}: corrupted") + + for role, (requests, labels) in by_role.items(): + statuses = _evaluate(role, requests) + assert len(statuses) == len(requests) + for index, (label, status) in enumerate(zip(labels, statuses, strict=True)): + if index % 2 == 0: + assert status == "passed", f"{label}: the mapping rejects the published vector" + else: + assert status == "failed", ( + f"{label}: the mapping accepted a corrupted expectation, " + "so nothing about the SDK can make it fail" + ) diff --git a/tests/test_core_http.py b/tests/test_core_http.py index 94ad3a3..4f802d1 100644 --- a/tests/test_core_http.py +++ b/tests/test_core_http.py @@ -16,6 +16,8 @@ EnrollResponse, HttpRequest, HttpResponse, + OpenApiSecurityDisposition, + OpenApiSecurityResolution, OpenApiTrailingSlash, ProtectedResourceAuthorization, authorization_header_name, @@ -30,11 +32,13 @@ parse_authorization, render_authorization, require_service_origin_binding, + resolve_openapi_security, resolve_openapi_url, same_origin, select_did_web_public_jwk, ) from agent_enrollment_protocol.core.inspect import _origin +from agent_enrollment_protocol.core.openapi import SecurityRequirements from .test_core_models import inspect_document @@ -91,7 +95,7 @@ def test_authorization_rejects_ambiguous_or_invalid_values() -> None: with pytest.raises(AepAuthorizationError) as ambiguous: parse_authorization("AEP one, AEP two", AuthorizationCarrier.DEDICATED) assert ambiguous.value.code == "not_recognized" - for value in ("", "Digest value", "AEP", "AEP value"): + for value in ("", "Digest value", "AEP", "AEP ", " AEP value"): with pytest.raises(AepAuthorizationError): parse_authorization(value) with pytest.raises(AepAuthorizationError) as empty: @@ -105,6 +109,64 @@ def test_authorization_rejects_ambiguous_or_invalid_values() -> None: assert empty.value.code == "invalid_request" +def test_authorization_separator_follows_the_grammar() -> None: + # CORE-AUTH-001 spells the presentation as `"AEP" 1*SP compact-jws`, so repeated separators are + # grammar and a field value may carry trailing optional whitespace. + for value in ("AEP assertion", "AEP\tassertion", "AEP assertion ", "AEP assertion\t"): + assert parse_authorization(value).credentials == "assertion" + + +@pytest.mark.parametrize( + "value", + [ + "AEP one two", + "AEP one\ttwo", + "AEP compact.jws\x7f", + "AEP caf\u00e9", + ], +) +def test_authorization_refuses_a_value_carrying_more_than_one_credential(value: str) -> None: + # CORE-PR-018 forbids a combined value encoding more than one AEP credential, and CORE-AUTH-001 + # makes the credential visible US-ASCII with no space, so neither carrier may split one. + for carrier in AuthorizationCarrier: + with pytest.raises(AepAuthorizationError) as rejected: + parse_authorization(value, carrier) + assert rejected.value.code == "not_recognized" + + +@pytest.mark.parametrize( + "credentials", ["one two", "one\ttwo", "line\r\nInjected: header", "null\x00byte", "caf\u00e9"] +) +def test_render_authorization_refuses_credentials_that_are_not_a_field_value( + credentials: str, +) -> None: + # A credential carrying CR, LF, NUL, or a space would split or extend the rendered field, so + # the library refuses to render one rather than emitting it. + with pytest.raises(AepAuthorizationError) as rejected: + render_authorization( + ProtectedResourceAuthorization.model_construct( + carrier=AuthorizationCarrier.STANDARD, + scheme=AuthorizationScheme.AEP, + credentials=credentials, + ) + ) + assert rejected.value.code == "invalid_request" + + +@pytest.mark.parametrize( + "endpoint_base", + ["/aep/?next=", "/aep/#fragment", "/aep/../admin/", "/aep/./", "//evil.example/"], +) +def test_endpoint_base_must_be_able_to_carry_a_command_path(endpoint_base: str) -> None: + # CORE-HTTP-004 appends the command path with one `/` separator. A base carrying a query, a + # fragment, or a dot segment would put the command name in the query string or walk out of the + # advertised base. + with pytest.raises(ValueError, match="origin-relative"): + normalize_endpoint_base(endpoint_base) + with pytest.raises(ValueError, match="origin-relative"): + command_path(Command.ENROLL, endpoint_base) + + def test_inspect_version_origin_and_did_web_helpers() -> None: assert is_version_compatible("1.7") assert is_version_compatible("1.0", "1.7") @@ -299,5 +361,95 @@ def test_openapi_url_and_path_helpers() -> None: ) +_AEP = "aepScheme" +_KEY = "apiKeyScheme" +_ROOT: tuple[dict[str, tuple[str, ...]], ...] = ({_AEP: ()},) +_SUPPORTED = (_AEP, _KEY) + + +@pytest.mark.parametrize( + ("operation", "disposition"), + [ + (None, OpenApiSecurityDisposition.REQUIRED), + ((), OpenApiSecurityDisposition.PUBLIC), + (({}, {_AEP: ()}), OpenApiSecurityDisposition.OPTIONAL_ANONYMOUS), + (({_AEP: ()}, {_KEY: ()}), OpenApiSecurityDisposition.ALTERNATIVES), + (({_AEP: (), _KEY: ()},), OpenApiSecurityDisposition.ALL_REQUIRED), + ], +) +def test_openapi_security_inheritance( + operation: SecurityRequirements | None, disposition: OpenApiSecurityDisposition +) -> None: + # CORE-OAS-011: operation security replaces the root value, an empty array is public, an empty + # requirement object is an anonymous alternative, several objects are alternatives, and several + # schemes in one object are compound. + resolved = resolve_openapi_security( + root=_ROOT, operation=operation, supported_schemes=_SUPPORTED + ) + assert resolved.disposition is disposition + + +def test_openapi_security_is_public_without_any_requirement() -> None: + resolved = resolve_openapi_security(root=None, operation=None, supported_schemes=_SUPPORTED) + assert resolved.disposition is OpenApiSecurityDisposition.PUBLIC + assert resolved.satisfiable == () + + +def test_openapi_security_refuses_to_reduce_an_unsatisfiable_compound_requirement() -> None: + # CORE-OAS-013 makes a compound requirement unsupported when any member is, and CORE-OAS-015 + # forbids reducing it to the member the Agent does support. + resolved = resolve_openapi_security( + root=None, + operation=({_AEP: (), "mtlsScheme": ()},), + supported_schemes=(_AEP,), + ) + assert resolved.disposition is OpenApiSecurityDisposition.FALLBACK_LIVE_CHALLENGE + assert resolved.satisfiable == () + + +def test_openapi_security_selects_only_complete_alternatives() -> None: + # CORE-OAS-014 permits another complete alternative when a compound requirement is unsupported. + resolved = resolve_openapi_security( + root=None, + operation=({_AEP: (), "mtlsScheme": ()}, {_KEY: ()}), + supported_schemes=_SUPPORTED, + ) + assert resolved.disposition is OpenApiSecurityDisposition.ALTERNATIVES + assert resolved.satisfiable == ((_KEY,),) + + +def test_openapi_security_dispositions_carry_their_wire_values() -> None: + assert [item.value for item in OpenApiSecurityDisposition] == [ + "public", + "optional-anonymous", + "required", + "alternatives", + "all-required", + "fallback-live-challenge", + ] + assert OpenApiSecurityResolution(OpenApiSecurityDisposition.PUBLIC).satisfiable == () + + +@pytest.mark.parametrize( + "did", + [ + "did:web:exa%0Dmple.com", + "did:web:exa%0Ample.com", + "did:web:exa%09mple.com", + "did:web:%20example.com", + "did:web:example.com%20", + "did:web:example.com%7f", + ], +) +def test_did_web_host_must_be_visible_ascii(did: str) -> None: + # urlsplit strips CR, LF, and tab before parsing, so without this rule the host would be + # checked in its sanitized form and then concatenated into the document URL in its raw form, + # putting whitespace or a control character into a URL handed to an HTTP client. + with pytest.raises(ValueError, match="Invalid did:web identifier"): + did_web_document_url(did) + with pytest.raises(ValueError, match="Invalid did:web"): + did_web_origin(did) + + def test_imported_status_is_used() -> None: assert EnrollResponse(status=AgentStatus.ACTIVE).status is AgentStatus.ACTIVE diff --git a/tests/test_core_models.py b/tests/test_core_models.py index 2c6b739..dcf09ae 100644 --- a/tests/test_core_models.py +++ b/tests/test_core_models.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import re import pytest from pydantic import ValidationError @@ -183,6 +184,12 @@ def test_inspect_and_command_models_are_immutable_and_forward_compatible() -> No "URI-reference", ), (lambda: HttpConfiguration(endpoint_base="https://example.com"), "origin-relative"), + # CORE-HTTP-004 appends the command path to endpoint_base with one `/` separator, so a base + # carrying a query, a fragment, or a dot segment cannot carry a command. + (lambda: HttpConfiguration(endpoint_base="/aep/?next="), "origin-relative"), + (lambda: HttpConfiguration(endpoint_base="/aep/#fragment"), "origin-relative"), + (lambda: HttpConfiguration(endpoint_base="/aep/../admin/"), "origin-relative"), + (lambda: HttpConfiguration(endpoint_base="/aep/./"), "origin-relative"), (lambda: Identity(methods=("DID:web",)), "identifier"), (lambda: ServiceIdentity(did="https://example.com"), "DID"), ( @@ -319,6 +326,62 @@ def test_claim_validation_boundaries() -> None: ) +@pytest.mark.parametrize( + "value", + [ + "owner@ex\u00e1mple.com", + "owner@ex\u0663mple.com", + "\u00f3wner@example.com", + "owner@\u0435xample.com", + ], +) +def test_mailbox_is_ascii(value: str) -> None: + # CLM-CAT-001 binds contact.email to the RFC 5321 Mailbox rule as the JSON Schema `email` + # format references it, which is ASCII; a Unicode mailbox is the separate `idn-email` format. + # Accepting one would let a homograph domain stand in for the ASCII domain it resembles. + assert not _is_mailbox(value) + with pytest.raises(ValidationError, match=re.escape("contact.email")): + ClaimValues.model_validate({"contact.email": value}) + + +@pytest.mark.parametrize( + "url", + [ + "https://user:password@platform.example/agents/1/did.json", + "https://user@platform.example/agents/1/did.json", + ], +) +def test_https_urls_refuse_user_information(url: str) -> None: + # User information carries credentials the document is not bound to and makes the authority + # ambiguous to a reader, so it is refused here as it is on every other URL this package takes. + with pytest.raises(ValidationError, match="user info"): + PlatformAgentIdentity( + agent_did="did:web:platform.example:agents:1", + agent_identity_id="pai_1", + created_at="2026-01-01T00:00:00Z", + did_document_url=url, + key_id="did:web:platform.example:agents:1#key-1", + service_did="did:web:api.service.example", + signing_algorithms=(SigningAlgorithm.ES256,), + status=ManagedAgentStatus.ACTIVE, + updated_at="2026-01-01T00:00:00Z", + ) + + +def test_assertion_resource_refuses_user_information() -> None: + with pytest.raises(ValidationError, match="user info"): + ClientAssertionClaims( + aud="did:web:api.service.example", + exp=1_000_300, + iat=1_000_000, + iss="did:web:agent.example", + jti="assertion-1", + op=AssertionOperation.AUTHENTICATE, + resource="https://user@api.service.example/orders/1", + sub="did:web:agent.example", + ) + + def test_lifecycle_and_command_wire_models() -> None: request = EnrollRequest(agent_did="did:web:agent.example", idempotency_key="request-1") assert request.to_wire() == { diff --git a/tests/test_platform.py b/tests/test_platform.py index e7f416b..31f1e23 100644 --- a/tests/test_platform.py +++ b/tests/test_platform.py @@ -335,17 +335,33 @@ async def pending(*_: Any) -> PlatformResult[Any]: context("pending"), ) assert result.status == 202 - with pytest.raises(ValueError, match="configured maximum"): - await platform.sign( - identity_id, - PlatformSignRequest( - jti="long", - lifetime_seconds="61", - op=AssertionOperation.STATUS, - service_did=SERVICE_DID, - ), - context("long"), - ) + # PLT-SIGN-011: a caller-supplied lifetime above the configured maximum is a rejected request, + # not an unhandled failure, so it is answered as Problem Details before the idempotency store. + rejected = await platform.sign( + identity_id, + PlatformSignRequest( + jti="long", + lifetime_seconds="61", + op=AssertionOperation.STATUS, + service_did=SERVICE_DID, + ), + context("long"), + ) + assert rejected.status == 400 + assert rejected.problem is not None + assert rejected.problem.code == "invalid_request" + # An unauthorized caller reaches the same answer rather than an exception escaping the API. + unauthorized = await platform.sign( + "pai_unknown", + PlatformSignRequest( + jti="long", + lifetime_seconds="61", + op=AssertionOperation.STATUS, + service_did=SERVICE_DID, + ), + context("unauthorized"), + ) + assert unauthorized.status == 400 @pytest.mark.asyncio diff --git a/tests/test_platform_boundaries.py b/tests/test_platform_boundaries.py index 5f7d231..dfb1773 100644 --- a/tests/test_platform_boundaries.py +++ b/tests/test_platform_boundaries.py @@ -43,6 +43,7 @@ create_service_scoped_agent_did, ) from agent_enrollment_protocol.platform.document import create_discovery_document, render_did_url +from agent_enrollment_protocol.platform.platform import _lifecycle_error from tests.test_platform import ( NOW, SERVICE_DID, @@ -116,6 +117,45 @@ def test_did_helpers() -> None: create_did_document(identity, invalid) +@pytest.mark.parametrize( + "jwk", + [ + {"crv": "P-256", "d": "private", "kty": "EC", "x": "AQ", "y": "AQ"}, + {"crv": "Ed25519", "d": "private", "kty": "OKP", "x": "AQ"}, + {"kty": "RSA", "n": "AQ", "e": "AQ", "p": "private"}, + {"kty": "RSA", "n": "AQ", "e": "AQ", "qi": "private"}, + {"k": "symmetric", "kty": "oct"}, + ], +) +def test_did_document_refuses_private_key_material(jwk: dict[str, str]) -> None: + # PLT-DID-002: the DID document is served to anyone who resolves the Agent DID, so a key store + # that returned a JWK still carrying its private members would publish the signing key itself. + identity = record() + method = DidVerificationMethod( + controller=identity.agent_did, + id=identity.key_id, + public_key_jwk=jwk, + type="JsonWebKey2020", + ) + with pytest.raises(ValueError, match="only public key material"): + create_did_document(identity, method) + + +@pytest.mark.parametrize( + ("status", "code"), + [ + (ManagedAgentStatus.TERMINATED, "identity_terminated"), + (ManagedAgentStatus.SUSPENDED, "identity_suspended"), + (ManagedAgentStatus.REVOKED, "identity_suspended"), + (ManagedAgentStatus.ACTIVE, "identity_unavailable"), + ], +) +def test_lifecycle_errors_use_registered_codes(status: ManagedAgentStatus, code: str) -> None: + # CORE-ERR-009 registers identity_suspended, identity_terminated and identity_unavailable, and + # no identity_revoked or identity_active, so a revoked identity reports the suspended code. + assert _lifecycle_error(status) == code + + def test_discovery_boundaries() -> None: base = options().discovery for field in ( @@ -405,17 +445,18 @@ async def update_status( assert isinstance(prefixed_result.body, PlatformAgentIdentity) assert prefixed_result.body.agent_identity_id == "pai_existing" - for change in ( - {"identifier": lambda: ""}, - {"agent_did_id_generator": lambda: ""}, - {"clock": lambda: NOW.replace(tzinfo=None)}, - ): + for change in ({"identifier": lambda: ""}, {"agent_did_id_generator": lambda: ""}): invalid = Platform(options(**change)) with pytest.raises(ValueError): await invalid.provision( PlatformProvisionRequest(service_did=SERVICE_DID), context("invalid") ) + # The Discovery Last-Modified validator is stamped when the document is built, so a clock that + # cannot produce one is refused at construction rather than at the first command. + with pytest.raises(ValueError, match="offset-aware"): + Platform(options(clock=lambda: NOW.replace(tzinfo=None))) + class MismatchedStore(MemoryIdentityStore): async def find_or_create( self, principal: str, service_did: str, factory: Any diff --git a/tests/test_service.py b/tests/test_service.py index 47dc1c3..6fcb30c 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -360,6 +360,42 @@ async def operation() -> StoredResponse: await naive.execute(value, operation) +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("header", "value", "code"), + [ + ("Authorization", "AEP", "not_recognized"), + ("Authorization", "Bearer", "not_recognized"), + ("Authorization", "AEP not a token", "not_recognized"), + ("Authorization", "basic ", "not_recognized"), + ("Authorization", "Negotiate abc", "authentication_required"), + ("X-Other", "value", "authentication_required"), + ], +) +async def test_malformed_presentation_is_distinguished_from_a_missing_one( + header: str, value: str, code: str +) -> None: + # CORE-PR-038: missing credentials answer authentication_required, while malformed ones answer + # not_recognized. A value naming an AEP scheme is an AEP presentation even when the rest of it + # does not parse; a value naming another protocol's scheme is not ours to reject. + service, _ = _service() + await service.enroll( + json.dumps({"agent_did": AGENT_DID}).encode(), + _options(AssertionOperation.ENROLL, "enroll-malformed", key="enroll-malformed"), + ) + + result = await service.authenticate_protected_resource( + ProtectedResourceRequest( + headers={header: value}, method="GET", url="https://service.example/orders" + ) + ) + + assert not result.authenticated + assert result.response is not None + assert result.response.problem is not None + assert result.response.problem.code == code + + def _record() -> EnrollmentRecord: return EnrollmentRecord( agent_did=AGENT_DID, diff --git a/tests/test_service_credentials.py b/tests/test_service_credentials.py index 3d2ca27..20cd226 100644 --- a/tests/test_service_credentials.py +++ b/tests/test_service_credentials.py @@ -203,6 +203,33 @@ async def issue(request: GrantRequest, context: GrantContext) -> ApiKeyGrantResp assert result.response.problem.code == "not_recognized" +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("username", "password"), + [("agent", "pa\u00dfwort"), ("\u00e9agent", "secret"), ("agent", "s\u00e9cret")], +) +async def test_issued_basic_credentials_must_survive_rfc_7617_encoding( + username: str, password: str +) -> None: + # BSC-RSP-003: RFC 7617 leaves the user-pass charset undefined without a charset parameter, so a + # client encoding a non-ASCII value as ISO-8859-1 and a Service storing it as UTF-8 derive + # different bytes from the same credential and authentication silently fails. + async def issue(request: GrantRequest, context: GrantContext) -> BasicGrantResponse: + del request, context + return BasicGrantResponse( + credential_id="basic-1", + expires_at=EXPIRES, + password=password, + username=username, + ) + + definition = stored_basic_grant_type( + StoredCredentialGrantTypeOptions(issue=issue, store=MemoryServiceCredentialStore()) + ) + with pytest.raises(ValueError, match="US-ASCII"): + await definition.handler.grant(GrantRequest(grant_type="basic"), _grant_context("basic")) + + @pytest.mark.asyncio async def test_stored_credential_profiles_reject_invalid_issuance_and_storage() -> None: store = MemoryServiceCredentialStore()