fix: align session credential validation

README.md

@@ -60,10 +60,23 @@ The Service module includes stored credential profiles for each built-in Grant T
 Each factory accepts application-owned credential issuance and storage implementations. The included
 memory store is intended for examples and local development.
 
+Basic credentials use UTF-8 to encode `username:password` before Base64 encoding. Usernames and
+passwords may be empty or contain non-ASCII characters; usernames cannot contain a colon, and
+neither field can contain control characters.
+
 Concrete Grant Types and extensions can define additional Grant and Revoke request members. Pass
 those members through `GrantOptions.parameters` or `RevokeOptions.parameters`; the Agent retains
 control of the standard Grant Type and Revoke selector fields.
 
+## API-key credentials
+
+For an `api-key` credential, `api_key` is the complete HTTP header value. If a Service returns
+`header: "Authorization"` and `api_key: "Bearer secret"`, the Agent sends exactly
+`Authorization: Bearer secret`. The SDK does not add or remove a prefix, split the value, or trim it.
+Internal spaces and printable ASCII punctuation are allowed; empty values, leading or trailing
+spaces, control characters, and non-ASCII characters are rejected. The selected API-key header
+is also preserved when the caller selects the dedicated AEP carrier.
+
 ## Protected HTTP requests
 
 After explicitly enrolling an Agent with a Service, use `fetch_protected_resource()` to request a

src/agent_enrollment_protocol/core/models.py

@@ -675,7 +675,13 @@ def normalize_scopes(self) -> Self:
 class OAuthBearerGrantResponse(CredentialResponse):
     access_token: str = Field(min_length=1, repr=False)
     token_format: Literal["opaque", "jwt"] | None = None
-    token_type: Literal["Bearer"]
+    token_type: str
+
+    @model_validator(mode="after")
+    def validate_token_type(self) -> Self:
+        if self.token_type.lower() != "bearer":
+            raise ValueError("token_type must be Bearer (case-insensitive)")
+        return self
 
 
 class ApiKeyGrantResponse(CredentialResponse):
@@ -684,9 +690,10 @@ class ApiKeyGrantResponse(CredentialResponse):
 
     @model_validator(mode="after")
     def validate_api_key(self) -> Self:
-        if any(
-            ord(character) < 0x21 or ord(character) > 0x7E or character in {'"', ",", ";", "\\"}
-            for character in self.api_key
+        if (
+            self.api_key.startswith(" ")
+            or self.api_key.endswith(" ")
+            or any(ord(character) < 0x20 or ord(character) > 0x7E for character in self.api_key)
         ):
             raise ValueError("api_key must be an unambiguous HTTP field value")
         if HTTP_FIELD_NAME_PATTERN.fullmatch(self.header) is None:
@@ -695,9 +702,9 @@ def validate_api_key(self) -> Self:
 
 
 class BasicGrantResponse(CredentialResponse):
-    password: str = Field(min_length=1, repr=False)
-    realm: str | None = Field(default=None, min_length=1)
-    username: str = Field(min_length=1)
+    password: str = Field(repr=False)
+    realm: str | None = None
+    username: str
 
     @model_validator(mode="after")
     def validate_basic(self) -> Self:

src/agent_enrollment_protocol/service/credentials.py

@@ -367,16 +367,13 @@ def _validate_issued_credential_config(
 
 
 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")
+    try:
+        credential.username.encode()
+        credential.password.encode()
+    except UnicodeEncodeError:
+        raise ValueError("AEP issued Basic credentials must be encodable as UTF-8") from None
 
 
 def _configured_api_key_headers(config: GrantTypeConfig) -> frozenset[str] | None:

src/agent_enrollment_protocol/service/service.py

@@ -500,26 +500,24 @@ async def authenticate_protected_resource(
                     authentication_method=AEP_AUTHENTICATION_METHOD_JWT,
                 ),
             )
-        if presentation is not None:
-            method = (
-                AEP_GRANT_TYPE_BASIC
-                if presentation.scheme is AuthorizationScheme.BASIC
-                else AEP_GRANT_TYPE_OAUTH_BEARER
-            )
-            if method not in self._authentication_methods:
-                return self._protected_problem("unsupported_authentication_method", resource)
         authentication_input = CredentialAuthenticationInput(
             current_time=self._now(),
             headers=headers,
             method=request.method,
             url=resource,
         )
-        presented = presentation is not None
+        presented = presentation is not None or any(
+            _names_aep_scheme(value) for value in headers.get("authorization", ())
+        )
+        advertised_presentation = False
+        matched: AuthenticatedPrincipal | None = None
         for method in self._authentication_methods:
             if method == AEP_AUTHENTICATION_METHOD_JWT:
                 continue
             authenticator = cast(CredentialAuthenticator, self._grant_types[method].authenticator)
-            presented = presented or await authenticator.has_presentation(authentication_input)
+            detected = await authenticator.has_presentation(authentication_input)
+            advertised_presentation = advertised_presentation or detected
+            presented = presented or detected
             principal = await authenticator.authenticate(authentication_input)
             if principal is None:
                 continue
@@ -532,7 +530,19 @@ async def authenticate_protected_resource(
                 raise ValueError("AEP credential authenticator returned an invalid principal")
             if not await self._active(principal.agent_did):
                 return self._protected_problem("not_recognized", resource)
-            return ProtectedResourceResult(authenticated=True, principal=principal)
+            if matched is not None:
+                return self._protected_problem("not_recognized", resource)
+            matched = principal
+        if matched is not None:
+            return ProtectedResourceResult(authenticated=True, principal=matched)
+        if presentation is not None and not advertised_presentation:
+            method = (
+                AEP_GRANT_TYPE_BASIC
+                if presentation.scheme is AuthorizationScheme.BASIC
+                else AEP_GRANT_TYPE_OAUTH_BEARER
+            )
+            if method not in self._authentication_methods:
+                return self._protected_problem("unsupported_authentication_method", resource)
         return self._protected_problem(
             "not_recognized" if presented else "authentication_required", resource
         )
@@ -888,11 +898,6 @@ 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

tests/test_agent_boundaries.py

@@ -574,7 +574,7 @@ async def test_credential_presentations_and_metadata_validation() -> None:
             "access_token": "token",
             "credential_id": "bearer",
             "expires_at": expires,
-            "token_type": "Bearer",
+            "token_type": "bEaReR",
         },
         "basic": {
             "credential_id": "basic",
@@ -597,6 +597,8 @@ async def test_credential_presentations_and_metadata_validation() -> None:
         _validate_credential_record(record, record.service_did, now)
         headers = _credential_headers(record, AuthorizationCarrier.DEDICATED)
         assert "AEP-Authorization" in headers
+        if grant_type == "oauth-bearer":
+            assert headers["AEP-Authorization"] == "Bearer token"
     invalid = CredentialRecord(
         "wrong",
         now + timedelta(hours=1),

tests/test_agent_integration.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import base64
 import json
 from collections.abc import AsyncIterator
 from dataclasses import replace
@@ -33,6 +34,7 @@
 from agent_enrollment_protocol.core import (
     ApiKeyGrantResponse,
     AssertionOperation,
+    BasicGrantResponse,
     ClientAssertionClaims,
     GrantRequest,
     GrantTypeConfig,
@@ -53,11 +55,13 @@
     GrantContext,
     GrantTypeDefinition,
     MemoryServiceCredentialStore,
+    ProtectedResourceRequest,
     RevokeContext,
     Service,
     ServiceOptions,
     StoredCredentialGrantTypeOptions,
     stored_api_key_grant_type,
+    stored_basic_grant_type,
 )
 
 from .test_platform import NOW, KeyStore
@@ -68,9 +72,94 @@
 RESOURCE = "https://service.example/private"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "username,password",
+    [
+        ("agent", "pa\u00dfwort"),
+        ("\u00e9agent", "secret"),
+        ("agent", "s\u00e9cret"),
+        ("\u7528\u6237", "\U0001f511:secret"),
+        ("e\u0301agent", "se\u0301cret"),
+    ],
+)
+async def test_basic_utf8_credentials_round_trip_through_agent_and_service(
+    username: str, password: str
+) -> None:
+    key = Ed25519PrivateKey.generate()
+
+    async def issue(request: GrantRequest, context: GrantContext) -> BasicGrantResponse:
+        return BasicGrantResponse(
+            username=username,
+            password=password,
+            credential_id="basic",
+            expires_at=(NOW + timedelta(hours=1)).isoformat(),
+        )
+
+    service = Service(
+        ServiceOptions(
+            service_did=SERVICE_DID,
+            identity_methods=("did:web",),
+            authentication_methods=("basic",),
+            verifier=Verifier(key),
+            clock=lambda: NOW,
+            grant_types=(
+                stored_basic_grant_type(
+                    StoredCredentialGrantTypeOptions(
+                        issue=issue, store=MemoryServiceCredentialStore()
+                    )
+                ),
+            ),
+        )
+    )
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=AepAsgiApplication(service))
+    ) as client:
+        transport = HttpxTransport(client)
+        async with Agent(
+            AgentOptions(
+                identity_provider=LocalIdentityProvider(key),
+                clock=lambda: NOW,
+                command_transport=transport,
+                inspect_transport=transport,
+            )
+        ) as agent:
+            session = agent.service("https://service.example")
+            await session.enroll()
+            await session.grant(GrantOptions(grant_type="basic"))
+            headers = await session.authentication_headers(
+                AuthenticationOptions(resource=RESOURCE, credential_id="basic")
+            )
+            assert headers == {
+                "Authorization": "Basic "
+                + base64.b64encode(f"{username}:{password}".encode()).decode("ascii")
+            }
+            request = ProtectedResourceRequest(headers=headers, method="GET", url=RESOURCE)
+            assert (await service.authenticate_protected_resource(request)).authenticated
+            await session.revoke(RevokeOptions(grant_type="basic", credential_id="basic"))
+            assert not (await service.authenticate_protected_resource(request)).authenticated
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("method", ["aep-jwt", "api-key"])
-async def test_protected_resource_helper_through_real_asgi_service(method: str) -> None:
+@pytest.mark.parametrize(
+    "header,api_key",
+    [("X-Key", "secret")]
+    + [
+        ("Authorization", value)
+        for value in [
+            "Bearer secret",
+            "bearer secret",
+            "Basic secret",
+            "basic secret",
+            "Bearer two words",
+            'key,";\\value',
+        ]
+    ],
+)
+async def test_protected_resource_helper_through_real_asgi_service(
+    method: str, header: str, api_key: str
+) -> None:
     from agent_enrollment_protocol.adapters import AsgiReceive, AsgiScope, AsgiSend
 
     key = Ed25519PrivateKey.generate()
@@ -80,9 +169,9 @@ async def test_protected_resource_helper_through_real_asgi_service(method: str)
     async def issue(request: GrantRequest, context: GrantContext) -> ApiKeyGrantResponse:
         issued[0] += 1
         return ApiKeyGrantResponse(
-            api_key="secret",
+            api_key=api_key,
             credential_id="one",
-            header="X-Key",
+            header=header,
             expires_at=(NOW + timedelta(hours=1)).isoformat(),
         )
 
@@ -148,12 +237,17 @@ async def private(scope: AsgiScope, receive: AsgiReceive, send: AsgiSend) -> Non
                     request=HttpRequest(method="GET", url=RESOURCE),
                     carrier="AEP-Authorization",
                     transport=transport,
-                    additional_authentication_headers={"Authorization": "Payment opaque"},
+                    additional_authentication_headers={"Authorization": "Payment opaque"}
+                    if header != "Authorization"
+                    else {},
                 )
             )
             assert result.status == 200 and result.body == b"authenticated"
             assert [path for path, _ in received] == ["/private", "/result"]
-            assert all(headers[b"authorization"] == b"Payment opaque" for _, headers in received)
+            if header != "Authorization":
+                assert all(
+                    headers[b"authorization"] == b"Payment opaque" for _, headers in received
+                )
             if method == "aep-jwt":
                 tokens = [
                     headers[b"aep-authorization"].decode().split()[1] for _, headers in received
@@ -164,7 +258,9 @@ async def private(scope: AsgiScope, receive: AsgiReceive, send: AsgiSend) -> Non
                 ]
                 assert tokens[0] != tokens[1] and issued[0] == 0
             else:
-                assert all(headers[b"x-key"] == b"secret" for _, headers in received)
+                assert all(
+                    headers[header.lower().encode()] == api_key.encode() for _, headers in received
+                )
                 assert issued[0] == 1