diff --git a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py index f930b548..90f6ceea 100644 --- a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py +++ b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py @@ -799,14 +799,22 @@ def list_messages( return [] def _filter_restored_tool_context(self, messages: list[SessionMessage]) -> list[SessionMessage]: - """Strip historical toolUse/toolResult context from restored messages.""" + """Strip historical toolUse/toolResult context from restored messages. + + Extended-thinking (reasoningContent) blocks are coupled to the tool + calls that follow them. Bedrock rejects an assistant message whose + reasoningContent blocks have been separated from their companion + toolUse blocks, so we must strip both together. + """ filtered_messages: list[SessionMessage] = [] for session_message in messages: message = session_message.to_message() filtered_content = [ content for content in message.get("content", []) - if "toolUse" not in content and "toolResult" not in content + if "toolUse" not in content + and "toolResult" not in content + and "reasoningContent" not in content ] if not filtered_content: diff --git a/src/bedrock_agentcore/runtime/a2a.py b/src/bedrock_agentcore/runtime/a2a.py index 720e0791..d9c4a3b7 100644 --- a/src/bedrock_agentcore/runtime/a2a.py +++ b/src/bedrock_agentcore/runtime/a2a.py @@ -21,10 +21,12 @@ OAUTH2_CALLBACK_URL_HEADER, REQUEST_ID_HEADER, SESSION_HEADER, + USER_ID_HEADER, PingStatus, is_forwardable_header, ) from .tracing import _ensure_baggage_processor_registered +from .utils import extract_sub_from_bearer logger = logging.getLogger(__name__) @@ -179,6 +181,11 @@ def build(self, request: Any) -> Any: session_id = headers.get(SESSION_HEADER) BedrockAgentCoreContext.set_request_context(request_id, session_id) + enduser_id = headers.get(USER_ID_HEADER) or extract_sub_from_bearer( + headers.get(AUTHORIZATION_HEADER) or headers.get(_AUTHORIZATION_HEADER_LOWER) + ) + BedrockAgentCoreContext.set_enduser_id(enduser_id) + workload_access_token = headers.get(ACCESS_TOKEN_HEADER) if workload_access_token: BedrockAgentCoreContext.set_workload_access_token(workload_access_token) diff --git a/src/bedrock_agentcore/runtime/ag_ui.py b/src/bedrock_agentcore/runtime/ag_ui.py index 70ea5ab4..43bc70a6 100644 --- a/src/bedrock_agentcore/runtime/ag_ui.py +++ b/src/bedrock_agentcore/runtime/ag_ui.py @@ -31,10 +31,12 @@ OAUTH2_CALLBACK_URL_HEADER, REQUEST_ID_HEADER, SESSION_HEADER, + USER_ID_HEADER, PingStatus, is_forwardable_header, ) from .tracing import _ensure_baggage_processor_registered +from .utils import extract_sub_from_bearer logger = logging.getLogger(__name__) @@ -163,6 +165,11 @@ def _build_request_context(self, request: Request | WebSocket) -> RequestContext session_id = headers.get(SESSION_HEADER) BedrockAgentCoreContext.set_request_context(request_id, session_id) + enduser_id = headers.get(USER_ID_HEADER) or extract_sub_from_bearer( + headers.get(AUTHORIZATION_HEADER) or headers.get(_AUTHORIZATION_HEADER_LOWER) + ) + BedrockAgentCoreContext.set_enduser_id(enduser_id) + workload_access_token = headers.get(ACCESS_TOKEN_HEADER) if workload_access_token: BedrockAgentCoreContext.set_workload_access_token(workload_access_token) diff --git a/src/bedrock_agentcore/runtime/app.py b/src/bedrock_agentcore/runtime/app.py index c291b255..8f4152c1 100644 --- a/src/bedrock_agentcore/runtime/app.py +++ b/src/bedrock_agentcore/runtime/app.py @@ -46,11 +46,12 @@ TASK_ACTION_FORCE_HEALTHY, TASK_ACTION_JOB_STATUS, TASK_ACTION_PING_STATUS, + USER_ID_HEADER, PingStatus, is_forwardable_header, ) from .tracing import _ensure_baggage_processor_registered -from .utils import convert_complex_objects +from .utils import convert_complex_objects, extract_sub_from_bearer # Sentinel so we only parse OTEL_RESOURCE_ATTRIBUTES once per process. _UNRESOLVED = object() @@ -413,6 +414,11 @@ def _build_request_context(self, request) -> RequestContext: session_id = headers.get(SESSION_HEADER) BedrockAgentCoreContext.set_request_context(request_id, session_id) + enduser_id = headers.get(USER_ID_HEADER) or extract_sub_from_bearer( + headers.get(AUTHORIZATION_HEADER) or headers.get(_AUTHORIZATION_HEADER_LOWER) + ) + BedrockAgentCoreContext.set_enduser_id(enduser_id) + agent_identity_token = headers.get(IDENTITY_WAT_HEADER) or headers.get(ACCESS_TOKEN_HEADER) if agent_identity_token: BedrockAgentCoreContext.set_workload_access_token(agent_identity_token) diff --git a/src/bedrock_agentcore/runtime/context.py b/src/bedrock_agentcore/runtime/context.py index b5f20221..83eba6b5 100644 --- a/src/bedrock_agentcore/runtime/context.py +++ b/src/bedrock_agentcore/runtime/context.py @@ -34,6 +34,7 @@ class BedrockAgentCoreContext: _oauth2_callback_url: ContextVar[Optional[str]] = ContextVar("oauth2_callback_url") _request_id: ContextVar[Optional[str]] = ContextVar("request_id") _session_id: ContextVar[Optional[str]] = ContextVar("session_id") + _enduser_id: ContextVar[Optional[str]] = ContextVar("enduser_id", default=None) _request_headers: ContextVar[Optional[Dict[str, str]]] = ContextVar("request_headers") _routing_experiment_arn: ContextVar[Optional[str]] = ContextVar("routing_experiment_arn", default=None) _routing_experiment_variant: ContextVar[Optional[str]] = ContextVar("routing_experiment_variant", default=None) @@ -93,6 +94,23 @@ def get_session_id(cls) -> Optional[str]: except LookupError: return None + @classmethod + def set_enduser_id(cls, enduser_id: Optional[str]) -> None: + """Set the end-user identity for the current request. + + The value is stamped onto every OpenTelemetry span as the ``enduser.id`` + attribute (OTel semantic convention). It is extracted automatically from + the ``X-Amzn-Bedrock-AgentCore-Runtime-User-Id`` request header, or from + the ``sub`` claim of a Bearer JWT in the ``Authorization`` header. + Applications can also set it explicitly to override the inferred value. + """ + cls._enduser_id.set(enduser_id) + + @classmethod + def get_enduser_id(cls) -> Optional[str]: + """Return the end-user identity for the current request, or None.""" + return cls._enduser_id.get() + @classmethod def set_request_headers(cls, headers: Dict[str, str]): """Set request headers in the context.""" diff --git a/src/bedrock_agentcore/runtime/models.py b/src/bedrock_agentcore/runtime/models.py index db0cef82..cbefad73 100644 --- a/src/bedrock_agentcore/runtime/models.py +++ b/src/bedrock_agentcore/runtime/models.py @@ -17,6 +17,7 @@ class PingStatus(str, Enum): SESSION_HEADER = "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id" SHELL_ID_HEADER = "X-Amzn-Bedrock-AgentCore-Shell-Id" REQUEST_ID_HEADER = "X-Amzn-Bedrock-AgentCore-Runtime-Request-Id" +USER_ID_HEADER = "X-Amzn-Bedrock-AgentCore-Runtime-User-Id" ACCESS_TOKEN_HEADER = "WorkloadAccessToken" # nosec IDENTITY_WAT_HEADER = "X-Amz-Bedrock-AgentCore-Identity-WAT" # nosec OAUTH2_CALLBACK_URL_HEADER = "OAuth2CallbackUrl" diff --git a/src/bedrock_agentcore/runtime/tracing.py b/src/bedrock_agentcore/runtime/tracing.py index 7c0cf9cb..d693787b 100644 --- a/src/bedrock_agentcore/runtime/tracing.py +++ b/src/bedrock_agentcore/runtime/tracing.py @@ -103,7 +103,7 @@ class BaggageSpanProcessor(_get_base_class()): # type: ignore[misc] """ def on_start(self, span: object, parent_context: Optional[object] = None) -> None: - """Set routing experiment attributes on every new span. + """Set routing experiment and end-user identity attributes on every new span. Primary source: ContextVars set by ``_build_request_context`` — covers all spans created after request parsing (agent spans, tool spans, etc.). @@ -134,6 +134,10 @@ def on_start(self, span: object, parent_context: Optional[object] = None) -> Non if variant is not None: span.set_attribute("aws.agentcore.gateway.routing_experiment_variant_name", variant) # type: ignore[union-attr] + enduser_id = _context.get_enduser_id() + if enduser_id is not None: + span.set_attribute("enduser.id", enduser_id) # type: ignore[union-attr] + def on_end(self, span: object) -> None: """No-op.""" diff --git a/src/bedrock_agentcore/runtime/utils.py b/src/bedrock_agentcore/runtime/utils.py index 7a074577..63373616 100644 --- a/src/bedrock_agentcore/runtime/utils.py +++ b/src/bedrock_agentcore/runtime/utils.py @@ -1,7 +1,12 @@ """Bedrock AgentCore runtime utilities for object conversion and serialization.""" +import base64 +import json +import logging from dataclasses import asdict, is_dataclass -from typing import Any +from typing import Any, Optional + +logger = logging.getLogger(__name__) def convert_complex_objects(obj: Any, _depth: int = 0) -> Any: @@ -38,3 +43,33 @@ def convert_complex_objects(obj: Any, _depth: int = 0) -> Any: def is_valid_partition(partition: str) -> bool: """Returns if parsed-arn partition is valid.""" return partition in ("aws", "aws-us-gov") + + +def extract_sub_from_bearer(authorization: Optional[str]) -> Optional[str]: + """Return the 'sub' claim from a Bearer JWT without signature validation. + + Intended only for populating the OTel ``enduser.id`` span attribute. + The token is NOT validated — its signature, expiry, and issuer are not + checked. Trust decisions must be made by the inbound auth layer before + the request reaches agent code. + + Returns ``None`` when the header is absent, malformed, or has no 'sub'. + """ + if not authorization: + return None + parts = authorization.split(" ", 1) + if len(parts) != 2 or parts[0].lower() != "bearer": + return None + segments = parts[1].strip().split(".") + if len(segments) < 2: + return None + payload = segments[1] + # JWT base64url uses no padding; add it back for Python's decoder. + padding = (4 - len(payload) % 4) % 4 + try: + decoded = base64.urlsafe_b64decode(payload + "=" * padding) + sub = json.loads(decoded).get("sub") + return str(sub) if sub is not None else None + except Exception: + logger.debug("Could not decode JWT payload for enduser.id extraction", exc_info=True) + return None diff --git a/tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py b/tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py index b225cfa9..9ef30c84 100644 --- a/tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py +++ b/tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py @@ -3925,4 +3925,104 @@ def test_same_millisecond_different_microseconds_is_a_tie(self, session_manager) # through as a false non-tie. assert r1 == datetime(2024, 1, 1, 12, 0, 0, 0, tzinfo=timezone.utc) assert r2 == datetime(2024, 1, 1, 12, 0, 0, 1000, tzinfo=timezone.utc) - assert r2 > r1 + + +class TestFilterRestoredToolContext: + """Regression tests for _filter_restored_tool_context with extended thinking. + + Regression for https://github.com/aws/bedrock-agentcore-sdk-python/issues/621. + + When extended thinking is enabled, Bedrock rejects assistant messages whose + reasoningContent blocks have been stripped of their companion toolUse blocks + (or vice-versa). The filter must remove reasoningContent blocks alongside + toolUse/toolResult so that no partial assistant message reaches the API. + """ + + def _make_session_message(self, role: str, content: list) -> SessionMessage: + return SessionMessage.from_message({"role": role, "content": content}, 0) + + def test_strips_reasoning_content_alongside_tool_use(self, session_manager): + """reasoningContent blocks paired with toolUse must both be removed.""" + messages = [ + self._make_session_message( + "assistant", + [ + {"reasoningContent": {"reasoningText": {"text": "I should call weather.", "signature": "sig1"}}}, + {"toolUse": {"toolUseId": "tu_1", "name": "get_weather", "input": {"city": "Paris"}}}, + ], + ), + self._make_session_message("user", [{"toolResult": {"toolUseId": "tu_1", "content": [{"text": "22C"}]}}]), + self._make_session_message("assistant", [{"text": "It is 22°C in Paris."}]), + ] + + result = session_manager._filter_restored_tool_context(messages) + + # The tool-use assistant turn and its toolResult are dropped entirely. + # The plain-text assistant turn survives. + assert len(result) == 1 + assert result[0].to_message()["content"] == [{"text": "It is 22°C in Paris."}] + + def test_message_with_only_reasoning_and_tool_use_is_dropped_entirely(self, session_manager): + """An assistant message whose entire content is reasoningContent + toolUse + produces an empty filtered_content and must be excluded from the output.""" + messages = [ + self._make_session_message( + "assistant", + [ + {"reasoningContent": {"reasoningText": {"text": "Reasoning.", "signature": "s"}}}, + {"toolUse": {"toolUseId": "tu_x", "name": "calc", "input": {}}}, + ], + ), + ] + + result = session_manager._filter_restored_tool_context(messages) + assert result == [] + + def test_text_alongside_reasoning_and_tool_use_is_preserved(self, session_manager): + """If the assistant message has text content in addition to reasoningContent + and toolUse, the text survives after the other blocks are stripped.""" + messages = [ + self._make_session_message( + "assistant", + [ + {"reasoningContent": {"reasoningText": {"text": "Let me look this up.", "signature": "s"}}}, + {"text": "Checking now…"}, + {"toolUse": {"toolUseId": "tu_2", "name": "search", "input": {"q": "Paris weather"}}}, + ], + ), + ] + + result = session_manager._filter_restored_tool_context(messages) + + assert len(result) == 1 + content = result[0].to_message()["content"] + assert content == [{"text": "Checking now…"}] + + def test_messages_without_tool_context_are_unchanged(self, session_manager): + """Plain user/assistant messages (no tool or reasoning blocks) pass through.""" + messages = [ + self._make_session_message("user", [{"text": "What is the weather?"}]), + self._make_session_message("assistant", [{"text": "I don't know."}]), + ] + + result = session_manager._filter_restored_tool_context(messages) + + assert len(result) == 2 + assert result[0].to_message()["content"] == [{"text": "What is the weather?"}] + assert result[1].to_message()["content"] == [{"text": "I don't know."}] + + def test_existing_tool_use_filtering_still_works(self, session_manager): + """Original toolUse/toolResult filtering behaviour is preserved.""" + messages = [ + self._make_session_message( + "assistant", + [{"toolUse": {"toolUseId": "tu_3", "name": "calc", "input": {"expr": "1+1"}}}], + ), + self._make_session_message("user", [{"toolResult": {"toolUseId": "tu_3", "content": [{"text": "2"}]}}]), + self._make_session_message("assistant", [{"text": "The answer is 2."}]), + ] + + result = session_manager._filter_restored_tool_context(messages) + + assert len(result) == 1 + assert result[0].to_message()["content"] == [{"text": "The answer is 2."}] diff --git a/tests/bedrock_agentcore/runtime/test_tracing.py b/tests/bedrock_agentcore/runtime/test_tracing.py index 020b4595..4e89d530 100644 --- a/tests/bedrock_agentcore/runtime/test_tracing.py +++ b/tests/bedrock_agentcore/runtime/test_tracing.py @@ -17,6 +17,7 @@ def setup_method(self): from bedrock_agentcore.runtime.context import BedrockAgentCoreContext BedrockAgentCoreContext.set_routing_experiment(None, None) + BedrockAgentCoreContext.set_enduser_id(None) def _make_span(self): span = MagicMock() @@ -169,6 +170,82 @@ def _inner(): assert results["req2"]["aws.agentcore.gateway.routing_experiment_variant_name"] == "green" +class TestBaggageSpanProcessorEnduserIdAttribute: + """enduser.id is stamped when BedrockAgentCoreContext.enduser_id is set.""" + + def setup_method(self): + from bedrock_agentcore.runtime.context import BedrockAgentCoreContext + + BedrockAgentCoreContext.set_routing_experiment(None, None) + BedrockAgentCoreContext.set_enduser_id(None) + + def _make_span(self): + return MagicMock() + + def test_enduser_id_set_on_span(self): + from bedrock_agentcore.runtime.context import BedrockAgentCoreContext + + BedrockAgentCoreContext.set_enduser_id("user-42") + span = self._make_span() + BaggageSpanProcessor().on_start(span) + + calls = {c[0][0]: c[0][1] for c in span.set_attribute.call_args_list} + assert calls["enduser.id"] == "user-42" + + def test_enduser_id_not_set_when_none(self): + from bedrock_agentcore.runtime.context import BedrockAgentCoreContext + + BedrockAgentCoreContext.set_enduser_id(None) + span = self._make_span() + BaggageSpanProcessor().on_start(span) + + set_keys = {c[0][0] for c in span.set_attribute.call_args_list} + assert "enduser.id" not in set_keys + + def test_enduser_id_does_not_interfere_with_routing_experiment(self): + from bedrock_agentcore.runtime.context import BedrockAgentCoreContext + + BedrockAgentCoreContext.set_routing_experiment("arn:aws:bedrock:us-east-1:123:exp/e1", "blue") + BedrockAgentCoreContext.set_enduser_id("alice") + span = self._make_span() + BaggageSpanProcessor().on_start(span) + + calls = {c[0][0]: c[0][1] for c in span.set_attribute.call_args_list} + assert calls["aws.agentcore.gateway.routing_experiment_arn"] == "arn:aws:bedrock:us-east-1:123:exp/e1" + assert calls["aws.agentcore.gateway.routing_experiment_variant_name"] == "blue" + assert calls["enduser.id"] == "alice" + + def test_different_contexts_get_different_enduser_ids(self): + """Concurrent requests must not bleed enduser IDs into each other.""" + import contextvars + + from bedrock_agentcore.runtime.context import BedrockAgentCoreContext + + processor = BaggageSpanProcessor() + results = {} + + def run_in_context(name, uid): + ctx = contextvars.copy_context() + + def _inner(): + BedrockAgentCoreContext.set_enduser_id(uid) + span = MagicMock() + processor.on_start(span) + results[name] = {c[0][0]: c[0][1] for c in span.set_attribute.call_args_list} + + ctx.run(_inner) + + t1 = threading.Thread(target=run_in_context, args=("req1", "user-A")) + t2 = threading.Thread(target=run_in_context, args=("req2", "user-B")) + t1.start() + t2.start() + t1.join() + t2.join() + + assert results["req1"]["enduser.id"] == "user-A" + assert results["req2"]["enduser.id"] == "user-B" + + class TestBaggageSpanProcessorNoOpMethods: def test_on_end_does_not_raise(self): BaggageSpanProcessor().on_end(MagicMock()) @@ -325,6 +402,106 @@ def handler(payload): assert captured["arn"] is None assert captured["variant"] is None + def test_enduser_id_set_from_user_id_header(self): + """X-Amzn-Bedrock-AgentCore-Runtime-User-Id header populates enduser.id ContextVar.""" + from starlette.testclient import TestClient + + from bedrock_agentcore.runtime import BedrockAgentCoreApp + from bedrock_agentcore.runtime.context import BedrockAgentCoreContext + from bedrock_agentcore.runtime.models import USER_ID_HEADER + + app = BedrockAgentCoreApp() + captured = {} + + @app.entrypoint + def handler(payload): + captured["enduser_id"] = BedrockAgentCoreContext.get_enduser_id() + return {"ok": True} + + client = TestClient(app) + client.post("/invocations", json={}, headers={USER_ID_HEADER: "explicit-user-99"}) + + assert captured["enduser_id"] == "explicit-user-99" + + def test_enduser_id_extracted_from_jwt_sub_claim(self): + """Bearer JWT in Authorization header → sub claim sets enduser.id ContextVar.""" + import base64 + import json + + from starlette.testclient import TestClient + + from bedrock_agentcore.runtime import BedrockAgentCoreApp + from bedrock_agentcore.runtime.context import BedrockAgentCoreContext + + header = base64.urlsafe_b64encode(b'{"alg":"none"}').rstrip(b"=").decode() + payload = base64.urlsafe_b64encode(json.dumps({"sub": "cognito-user-42"}).encode()).rstrip(b"=").decode() + jwt = f"{header}.{payload}.fakesig" + + app = BedrockAgentCoreApp() + captured = {} + + @app.entrypoint + def handler(payload_data): + captured["enduser_id"] = BedrockAgentCoreContext.get_enduser_id() + return {"ok": True} + + client = TestClient(app) + client.post("/invocations", json={}, headers={"Authorization": f"Bearer {jwt}"}) + + assert captured["enduser_id"] == "cognito-user-42" + + def test_user_id_header_takes_priority_over_jwt(self): + """Explicit User-Id header wins over JWT sub when both are present.""" + import base64 + import json + + from starlette.testclient import TestClient + + from bedrock_agentcore.runtime import BedrockAgentCoreApp + from bedrock_agentcore.runtime.context import BedrockAgentCoreContext + from bedrock_agentcore.runtime.models import USER_ID_HEADER + + header = base64.urlsafe_b64encode(b'{"alg":"none"}').rstrip(b"=").decode() + payload = base64.urlsafe_b64encode(json.dumps({"sub": "jwt-sub"}).encode()).rstrip(b"=").decode() + jwt = f"{header}.{payload}.fakesig" + + app = BedrockAgentCoreApp() + captured = {} + + @app.entrypoint + def handler(payload_data): + captured["enduser_id"] = BedrockAgentCoreContext.get_enduser_id() + return {"ok": True} + + client = TestClient(app) + client.post( + "/invocations", + json={}, + headers={USER_ID_HEADER: "explicit-wins", "Authorization": f"Bearer {jwt}"}, + ) + + assert captured["enduser_id"] == "explicit-wins" + + def test_enduser_id_is_none_when_no_header_and_no_jwt(self): + """No User-Id header and no Authorization → enduser.id is None.""" + from starlette.testclient import TestClient + + from bedrock_agentcore.runtime import BedrockAgentCoreApp + from bedrock_agentcore.runtime.context import BedrockAgentCoreContext + + app = BedrockAgentCoreApp() + captured = {} + + @app.entrypoint + def handler(payload): + captured["enduser_id"] = BedrockAgentCoreContext.get_enduser_id() + return {"ok": True} + + client = TestClient(app) + client.post("/invocations", json={}) + + assert captured["enduser_id"] is None + def test_extract_baggage_error_clears_experiment_context(self): """When _extract_baggage raises, all_baggage defaults to {} and ContextVars are set to None.""" from starlette.testclient import TestClient diff --git a/tests/bedrock_agentcore/runtime/test_utils.py b/tests/bedrock_agentcore/runtime/test_utils.py index 5c9ace48..e6c8c96b 100644 --- a/tests/bedrock_agentcore/runtime/test_utils.py +++ b/tests/bedrock_agentcore/runtime/test_utils.py @@ -1,11 +1,58 @@ """Tests for Bedrock AgentCore runtime utilities.""" +import base64 +import json from dataclasses import dataclass from typing import List, Optional from pydantic import BaseModel -from bedrock_agentcore.runtime.utils import convert_complex_objects, is_valid_partition +from bedrock_agentcore.runtime.utils import convert_complex_objects, extract_sub_from_bearer, is_valid_partition + + +def _make_jwt(claims: dict) -> str: + """Produce a structurally valid (but unsigned) JWT with the given payload claims.""" + header = base64.urlsafe_b64encode(b'{"alg":"none"}').rstrip(b"=").decode() + payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).rstrip(b"=").decode() + return f"{header}.{payload}.fakesig" + + +class TestExtractSubFromBearer: + def test_returns_sub_from_valid_jwt(self): + token = _make_jwt({"sub": "user-123", "iss": "https://example.com"}) + assert extract_sub_from_bearer(f"Bearer {token}") == "user-123" + + def test_case_insensitive_bearer_prefix(self): + token = _make_jwt({"sub": "user-abc"}) + assert extract_sub_from_bearer(f"bearer {token}") == "user-abc" + assert extract_sub_from_bearer(f"BEARER {token}") == "user-abc" + + def test_returns_none_when_no_sub_claim(self): + token = _make_jwt({"email": "user@example.com"}) + assert extract_sub_from_bearer(f"Bearer {token}") is None + + def test_returns_none_for_none_input(self): + assert extract_sub_from_bearer(None) is None + + def test_returns_none_for_empty_string(self): + assert extract_sub_from_bearer("") is None + + def test_returns_none_for_non_bearer_scheme(self): + assert extract_sub_from_bearer("Basic dXNlcjpwYXNz") is None + + def test_returns_none_for_malformed_jwt(self): + assert extract_sub_from_bearer("Bearer notajwt") is None + + def test_returns_none_for_invalid_base64_payload(self): + assert extract_sub_from_bearer("Bearer abc.!!!.sig") is None + + def test_sub_coerced_to_str(self): + token = _make_jwt({"sub": 42}) + assert extract_sub_from_bearer(f"Bearer {token}") == "42" + + def test_strips_trailing_whitespace_from_token(self): + token = _make_jwt({"sub": "clean-user"}) + assert extract_sub_from_bearer(f"Bearer {token} ") == "clean-user" class TestConvertComplexObjects: