fix: bind agent and MCP sessions to their owner

CHANGELOG.md

@@ -312,6 +312,17 @@ Entries before v0.5.0 were written retroactively as summaries.
 
 ### Security
 
+- **Agent and MCP runtime sessions are bound to their owner.** New session ids created by
+  the Web UI start with a prefix derived from the signed-in user (`shared/session_owner.py`,
+  mirrored in `web-ui/src/lib/sessionId.ts`), and the agent and the MCP server admit such an
+  id only for that user. Each runtime process serves one user. Existing decks keep their
+  chat: a chat id from before this change is admitted when AgentCore Memory holds that
+  user's history for it, and a deck without history continues under the user's own id. The
+  agent forwards an owner-prefixed id to the MCP runtime (also for composer groups,
+  `<id>-gN`). A refused call gets an error event with no details; a chat tab opened before
+  the update may be asked to reload the page once (new error code
+  `session_reload_required`). Deploy the Web UI first, then SdpmRuntime, then SdpmAgent.
+  Local mode is unchanged.
 - **The AgentCore runtimes allow only the operations the app uses.** The agent and MCP
   runtimes, their `DEFAULT` endpoints and the named endpoints `sdpm_agent_endpoint` and
   `sdpm_endpoint` now carry a resource-based policy (`AWS::BedrockAgentCore::ResourcePolicy`,

agent/Dockerfile

@@ -27,7 +27,8 @@ RUN uv export --frozen --directory /tmp/workspace --package sdpm-agent --no-dev
 RUN useradd -m -u 1000 bedrock_agentcore
 USER bedrock_agentcore
 
-COPY agent/basic_agent.py agent/cost_logger.py agent/errors.py agent/factory.py agent/composition.py agent/mcp_clients.py agent/message_hooks.py agent/model_profiles.py agent/resilience.py agent/session.py agent/stop_guard.py agent/streaming.py agent/turn_fence.py ./
+COPY agent/basic_agent.py agent/cost_logger.py agent/errors.py agent/factory.py agent/composition.py agent/mcp_clients.py agent/message_hooks.py agent/model_profiles.py agent/resilience.py agent/session.py agent/session_guard.py agent/stop_guard.py agent/streaming.py agent/turn_fence.py ./
+COPY shared/__init__.py shared/session_owner.py shared/
 COPY agent/prompts/ prompts/
 COPY agent/modes/ modes/
 COPY agent/tools/ tools/

agent/README.md

@@ -44,7 +44,7 @@ Three connection patterns are supported. Each returns an `MCPClient` instance ad
 
 ### Pattern 1: AgentCore Runtime + JWT Bearer
 
-For MCP servers deployed on Amazon Bedrock AgentCore Runtime. The caller's JWT is forwarded as-is for authentication and user_id propagation, and the caller's session id is forwarded as `Mcp-Session-Id` so that AgentCore routes the requests of one session to the same microVM.
+For MCP servers deployed on Amazon Bedrock AgentCore Runtime. The caller's JWT is forwarded as-is for authentication and user_id propagation, and the caller's session id is forwarded as `Mcp-Session-Id`, with the owner prefix of the token's user, so that AgentCore routes the requests of one session to the same microVM.
 
 ```python
 def _mcp_agentcore_runtime(jwt_token: str, session_id: str = "") -> MCPClient:

agent/README_ja.md

@@ -32,7 +32,7 @@ User → AgentCore Runtime → Agent
 
 ### Pattern 1: AgentCore Runtime + JWT Bearer
 
-AgentCore Runtime 上の MCP サーバーに JWT を転送して接続する。あわせて呼び出し元のセッション ID を `Mcp-Session-Id` として転送し、同一セッションのリクエストが同じ microVM に routing されるようにする。
+AgentCore Runtime 上の MCP サーバーに JWT を転送して接続する。あわせて呼び出し元のセッション ID を、トークンの利用者の持ち主プレフィックス付きで `Mcp-Session-Id` として転送し、同一セッションのリクエストが同じ microVM に routing されるようにする。
 
 ```python
 def _mcp_agentcore_runtime(jwt_token: str, session_id: str = "") -> MCPClient:

agent/basic_agent.py

@@ -19,6 +19,7 @@
 
 from bedrock_agentcore.runtime import BedrockAgentCoreApp
 
+import session_guard
 import stop_guard
 from errors import BAD_INPUT, AUTH, error_event
 from factory import create_agent
@@ -80,6 +81,13 @@ async def agent_stream(payload, context):
         yield error_event("Could not extract user_id from JWT sub claim", code=AUTH)
         return
 
+    # Only the session's owner gets in (session_guard.py); before the cancel registry,
+    # so a refused call never touches a running turn.
+    session_id, refusal = await session_guard.admit(session_id, getattr(context, "session_id", None), user_id)
+    if refusal is not None:
+        yield refusal
+        return
+
     # Cancel previous request for the same session
     prev_cancel = _cancel_events.get(session_id)
     if prev_cancel is not None:

agent/errors.py

@@ -17,6 +17,8 @@
 BAD_INPUT = "bad_input"
 MAX_OUTPUT = "max_output"
 INTERNAL = "internal"
+# Sent only with an explicit code: the page holds a session id it has to replace.
+SESSION_RELOAD_REQUIRED = "session_reload_required"
 
 _PATTERNS: list[tuple[tuple[str, ...], str]] = [
     (("Too much media", "too long"), CONVERSATION_TOO_LONG),

agent/mcp_clients.py

@@ -13,6 +13,8 @@
 from mcp.client.streamable_http import streamablehttp_client
 from strands.tools.mcp import MCPClient
 
+from shared.session_owner import owned_mcp_session_id, sub_from_jwt
+
 logger = logging.getLogger("sdpm.agent")
 
 # MCP server definitions: (display_name, required)
@@ -61,7 +63,10 @@ def mcp_agentcore_runtime(
         # after is back to 0.18s. No re-initialize fallback is needed. This holds
         # for the stateless server we run (`stateless_http=True` in
         # servers/remote/server.py); stateful MCP is documented to 404 instead.
-        headers["Mcp-Session-Id"] = session_id
+        #
+        # The id always carries the owner prefix of the token's user: an id of an
+        # earlier format gets it prepended (shared/session_owner.py).
+        headers["Mcp-Session-Id"] = owned_mcp_session_id(session_id, sub_from_jwt(jwt_token))
 
     kwargs: dict = {}
     if tool_filters:

agent/session_guard.py

@@ -0,0 +1,88 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: MIT-0
+"""Admit an invocation only for the owner of its session.
+
+The AgentCore Runtime authorizer has validated the JWT before the request reaches
+this container; ``sub`` comes from that token. A session id is admitted when:
+
+- it carries the owner prefix of ``sub`` (``shared.session_owner``), or
+- it has an earlier format and AgentCore Memory already holds this user's events
+  for it (sessions created before ids carried the prefix);
+
+and this process serves no other user. Refusals carry a code and no details.
+"""
+
+import asyncio
+import logging
+import os
+
+from errors import AUTH, BAD_INPUT, SERVICE_UNAVAILABLE, SESSION_RELOAD_REQUIRED, error_event
+from shared.session_owner import ProcessOwner, belongs_to, is_owner_prefixed
+
+logger = logging.getLogger("sdpm.agent")
+
+# The one user this process serves (set by the first admitted invocation).
+_process_owner = ProcessOwner()
+# (sub, session id) pairs of earlier-format ids confirmed through Memory.
+_owned_legacy: set[tuple[str, str]] = set()
+_client = None
+
+
+def _memory_client():
+    global _client
+    if _client is None:
+        import boto3
+        from botocore.config import Config
+
+        region = os.environ.get("AWS_REGION", os.environ.get("AWS_DEFAULT_REGION", "us-east-1"))
+        _client = boto3.client(
+            "bedrock-agentcore",
+            region_name=region,
+            config=Config(connect_timeout=2, read_timeout=5, retries={"max_attempts": 2}),
+        )
+    return _client
+
+
+def _has_own_events(memory_id: str, sub: str, session_id: str) -> bool:
+    response = _memory_client().list_events(
+        memoryId=memory_id, actorId=sub, sessionId=session_id, maxResults=1, includePayloads=False,
+    )
+    return bool(response.get("events"))
+
+
+def _refuse(reason: str, session_id: str, event: dict) -> tuple[None, dict]:
+    logger.warning("Session refused (%s) for session %s", reason, (session_id or "")[:12])
+    return None, event
+
+
+async def admit(
+    payload_session_id: str, header_session_id: str | None, sub: str,
+) -> tuple[str | None, dict | None]:
+    """Return ``(session_id, None)`` when admitted, else ``(None, error_event)``.
+
+    The platform routes by the session header, so that id is the one checked;
+    without the header (local runs) the payload id is used.
+    """
+    if header_session_id and header_session_id != payload_session_id:
+        return _refuse("id mismatch", header_session_id, error_event("Session id mismatch", code=BAD_INPUT))
+    session_id = header_session_id or payload_session_id
+
+    if is_owner_prefixed(session_id):
+        if not belongs_to(session_id, sub):
+            return _refuse("owner", session_id, error_event("Session not available", code=AUTH))
+    elif (sub, session_id) not in _owned_legacy:
+        memory_id = os.environ.get("MEMORY_ID", "")
+        if not memory_id or memory_id == "PLACEHOLDER":
+            return _refuse("legacy", session_id, error_event("Reload the page", code=SESSION_RELOAD_REQUIRED))
+        try:
+            owned = await asyncio.to_thread(_has_own_events, memory_id, sub, session_id)
+        except Exception:
+            logger.exception("Session check failed for session %s", session_id[:12])
+            return None, error_event("Session check unavailable", code=SERVICE_UNAVAILABLE)
+        if not owned:
+            return _refuse("legacy", session_id, error_event("Reload the page", code=SESSION_RELOAD_REQUIRED))
+        _owned_legacy.add((sub, session_id))
+
+    if not _process_owner.claim(sub):
+        return _refuse("process", session_id, error_event("Session not available", code=AUTH))
+    return session_id, None

docs/en/security.md

@@ -29,7 +29,9 @@ These settings come with every deployment of the CDK stacks in `infra/`:
 - **AgentCore Runtime**: the agent and MCP runtimes validate a JWT on every call
   (`customJwtAuthorizer`). The user ID comes from the token's `sub` claim. A resource-based
   policy on each runtime and its endpoints allows only `InvokeAgentRuntime` and
-  `StopRuntimeSession`, the operations the app uses.
+  `StopRuntimeSession`, the operations the app uses. The agent and the MCP server admit a
+  session only for its owner: the session id carries a prefix derived from the token's
+  `sub`, and each runtime process serves one user (`shared/session_owner.py`).
 - **Per-deck authorization**: a user is the deck's owner, a viewer (the deck is public), or
   has no access. The rules live in one place, `shared/authz.py`, used by both the API and the
   MCP server. See [Architecture — Authorization](architecture.md#authorization-role-based-access-control).

servers/remote/server.py

@@ -33,6 +33,7 @@
 
 from shared.authz import authorize  # noqa: E402
 from shared import user_library  # noqa: E402
+from shared.session_owner import ProcessOwner, belongs_to, is_owner_prefixed, sub_from_jwt  # noqa: E402
 from storage import write_fence  # noqa: E402
 from storage.aws import AwsStorage  # noqa: E402
 from tools import assets, reference, preview, generate  # noqa: E402
@@ -279,6 +280,44 @@ def _client_quiet() -> None:
     _start_soffice_warm_up_once()
 
 
+# --- Session ownership ---
+#
+# A session is admitted only for its owner (shared/session_owner.py). An
+# owner-prefixed id must carry the prefix of the token's user; ids of other
+# formats are admitted under the process binding alone. Every deck operation
+# still goes through _check_deck_access.
+
+_process_owner = ProcessOwner()
+_SESSION_HEADERS = ("mcp-session-id", "x-amzn-bedrock-agentcore-runtime-session-id")
+
+
+def _session_refusal(headers: dict) -> str | None:
+    """Return the refused session id (or ``"-"``) when the request may not proceed, else None."""
+    auth = headers.get("authorization")
+    if auth is None:
+        return None  # platform health checks; every data tool requires the JWT itself
+    token = auth[7:].strip() if auth.lower().startswith("bearer ") else ""
+    sub = sub_from_jwt(token)
+    if not sub:
+        return "-"
+    for name in _SESSION_HEADERS:
+        session_id = headers.get(name)
+        if session_id and is_owner_prefixed(session_id) and not belongs_to(session_id, sub):
+            return session_id
+    if not _process_owner.claim(sub):
+        return headers.get("mcp-session-id") or "-"
+    return None
+
+
+async def _send_forbidden(send) -> None:  # type: ignore
+    await send({
+        "type": "http.response.start",
+        "status": 403,
+        "headers": [(b"content-type", b"application/json")],
+    })
+    await send({"type": "http.response.body", "body": b'{"error": "forbidden"}'})
+
+
 class _CaptureHeadersMiddleware:
     """Raw ASGI middleware to capture HTTP headers into a ContextVar.
 
@@ -298,6 +337,11 @@ async def __call__(self, scope, receive, send):  # type: ignore
         """Capture headers from HTTP requests into ContextVar."""
         if scope["type"] == "http":
             headers = {k.decode(): v.decode() for k, v in scope.get("headers", [])}
+            refused = _session_refusal(headers)
+            if refused is not None:
+                logger.warning("Session refused for session %s", refused[:12])
+                await _send_forbidden(send)
+                return
             client = scope.get("method") == "POST" and "mcp-session-id" in headers
             if client:
                 _client_request_started()

shared/__init__.py

@@ -8,6 +8,7 @@
 - artifact_keys: key layout of deck artifacts (previews, compose, HTML) and the latest-epoch selector
 - review: change request contract of the Web UI comment tool
 - user_library: the user's style / template library (S3 layout, pins, template lists)
+- session_owner: runtime session ids bound to their owner (also used by agent/)
 
 Upload file conversion (PDF/DOCX/XLSX/PPTX → Markdown/JSON) lives in the core,
 ``sdpm.tools.attachment.ingest``.

shared/session_owner.py

@@ -0,0 +1,91 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: MIT-0
+"""Session ownership — bind runtime sessions to the user who owns them.
+
+A session id created by the Web UI starts with ``owner_prefix(sub)``, the first 16
+hex characters of SHA-256 over the signed-in user's ``sub``. The agent and the
+MCP server admit an owner-prefixed id only for that user, and each runtime
+process serves a single user (``ProcessOwner``).
+
+The JWT has already been validated by the AgentCore Runtime authorizer before
+the request reaches the container, so ``sub_from_jwt`` only decodes the payload.
+
+The Web UI mirror is ``web-ui/src/lib/sessionId.ts``; both test suites pin the
+same test vector.
+"""
+
+import base64
+import hashlib
+import hmac
+import json
+import re
+import threading
+
+PREFIX_LEN = 16
+
+# A prefixed id: 16 lowercase hex characters, then "-". No id of the earlier
+# formats matches: a UUID has its first "-" at index 8, and a padded deck id
+# has no "-" at all.
+_PREFIXED = re.compile(r"^[0-9a-f]{16}-")
+
+
+def owner_prefix(sub: str) -> str:
+    """Return the session id prefix of ``sub`` (16 lowercase hex characters)."""
+    return hashlib.sha256(sub.encode("utf-8")).hexdigest()[:PREFIX_LEN]
+
+
+def is_owner_prefixed(session_id: str) -> bool:
+    """True when ``session_id`` has the owner-prefixed shape."""
+    return bool(_PREFIXED.match(session_id or ""))
+
+
+def belongs_to(session_id: str, sub: str) -> bool:
+    """True when ``session_id`` is owner-prefixed and its prefix is ``sub``'s.
+
+    Suffixes after the prefix (composer groups ``-gN``) do not matter.
+    """
+    if not sub or not is_owner_prefixed(session_id):
+        return False
+    return hmac.compare_digest(session_id[:PREFIX_LEN], owner_prefix(sub))
+
+
+def owned_mcp_session_id(session_id: str, sub: str) -> str:
+    """Return the ``Mcp-Session-Id`` the agent sends for ``session_id``.
+
+    An id that already belongs to ``sub`` is sent unchanged; any other id gets
+    ``sub``'s prefix in front, so the MCP session always carries its owner.
+    Without a ``sub`` the id is returned unchanged.
+    """
+    if not sub or not session_id or belongs_to(session_id, sub):
+        return session_id
+    return f"{owner_prefix(sub)}-{session_id}"
+
+
+def sub_from_jwt(token: str) -> str:
+    """Return the ``sub`` claim of a JWT (payload decode only), or ``""``."""
+    try:
+        payload = token.split(".")[1]
+        payload += "=" * (-len(payload) % 4)
+        claims = json.loads(base64.urlsafe_b64decode(payload))
+        sub = claims.get("sub", "") if isinstance(claims, dict) else ""
+    except (IndexError, ValueError, AttributeError):
+        return ""
+    return sub if isinstance(sub, str) else ""
+
+
+class ProcessOwner:
+    """The one user a runtime process serves: the first ``sub`` that claims it."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._sub: str | None = None
+
+    def claim(self, sub: str) -> bool:
+        """Bind the process to ``sub`` if unbound; True when ``sub`` is the owner."""
+        if not sub:
+            return False
+        with self._lock:
+            if self._sub is None:
+                self._sub = sub
+                return True
+            return hmac.compare_digest(self._sub.encode("utf-8"), sub.encode("utf-8"))

tests/agent_stop_server.py

@@ -11,7 +11,7 @@
 from pathlib import Path
 
 ROOT = Path(__file__).resolve().parents[1]
-sys.path[:0] = [str(ROOT / "agent"), str(ROOT / "tests")]
+sys.path[:0] = [str(ROOT / "agent"), str(ROOT / "tests"), str(ROOT)]
 os.environ["MEMORY_ID"] = "memory-test"
 os.environ.setdefault("AWS_REGION", "us-east-1")