IBM
diff --git a/‎.secrets.baseline‎
Lines changed: 50 additions & 50 deletions b/‎.secrets.baseline‎
Lines changed: 50 additions & 50 deletions
diff --git a/‎CHANGELOG.md‎
Lines changed: 43 additions & 0 deletions b/‎CHANGELOG.md‎
Lines changed: 43 additions & 0 deletions
diff --git a/‎docs/docs/architecture/multitenancy.md‎
Lines changed: 2 additions & 1 deletion b/‎docs/docs/architecture/multitenancy.md‎
Lines changed: 2 additions & 1 deletion
diff --git a/‎docs/docs/manage/rbac.md‎
Lines changed: 5 additions & 2 deletions b/‎docs/docs/manage/rbac.md‎
Lines changed: 5 additions & 2 deletions
diff --git a/‎mcpgateway/admin.py‎
Lines changed: 40 additions & 12 deletions b/‎mcpgateway/admin.py‎
Lines changed: 40 additions & 12 deletions
diff --git a/‎mcpgateway/auth.py‎
Lines changed: 55 additions & 4 deletions b/‎mcpgateway/auth.py‎
Lines changed: 55 additions & 4 deletions
@@ -39,6 +39,49 @@
 
 `CONTENT_PATTERN_DETECTION_ENABLED=true` and `CONTENT_VALIDATE_PROMPT_TEMPLATES=true` ship as defaults (in contrast to `CONTENT_STRICT_MIME_VALIDATION=false` which had a soft-launch default for US-2). Existing deployments containing any of the default blocked patterns in stored resources or prompts (e.g. Jinja2 `{{ config }}` access, shell metacharacters, `UNION SELECT`) will start returning 400s on subsequent update calls. Set either flag to `false` temporarily to audit and clean existing content before re-enabling.
 
+#### **🔒 Admin bypass no longer reveals other users' private resources** ([#4323](https://github.com/IBM/mcp-context-forge/issues/4323), [#4341](https://github.com/IBM/mcp-context-forge/pull/4341))
+
+**Action Required for integrators relying on admin-bypass reads of other users' private resources.**
+
+Admin bypass (`is_admin=true` with `teams: null` in the JWT, or dev-mode basic-auth admin) now grants access only to **public** and **team** resources via the public HTTP routes. Another user's private resources (visibility=`private`) are only accessible to their owner — admin bypass can no longer read, update, delete, list, or enumerate another user's private tools, prompts, resources, servers, gateways, or A2A agents.
+
+The service layer additionally implements an own-private carve-out for the DB-resolved admin shape `(email, None)`: a session that resolves to admin in the database AND has not been narrowed by a token scope can still access its own private rows. The verified path that exercises this carve-out today is the trusted internal A2A endpoint (`mcpgateway/main.py::_get_internal_a2a_scope_context`), which forwards the admin email through to the service layer. Other internal/in-process callers will hit the same carve-out *only* if they preserve the email on the `(email, None)` shape; OAuth token refresh, for example, performs its own owner check at `token_storage_service._refresh_access_token` and does not exercise the hybrid branch. The carve-out is **not** reachable from the public HTTP routes: `mcpgateway.auth_context.get_scoped_resource_access_context` collapses HTTP admin requests to `(None, None)`, so a normal browser-driven admin gets the same anonymous-bypass treatment as everyone else and is denied their own private rows on `GET /tools/{my_private_tool}`. Use `team`-scoped tokens or own-the-resource workflows if you need a HTTP admin to see their own private rows directly.
+
+**Enforcement applied at the service layer for:**
+
+- `ToolService.get_tool`, `list_tools`
+- `PromptService.get_prompt`, `get_prompt_details`, `list_prompts`
+- `ResourceService.get_resource_by_id`, `read_resource`, `list_resources`
+- `ServerService.get_server`, `list_servers`
+- `GatewayService.get_gateway`, `list_gateways`
+- `A2AAgentService.get_agent`, `get_agent_by_name`, `get_agent_card`, `cancel_task`, `get_task`
+- `BaseService._apply_access_control` and `BaseService._apply_visibility_scope` (list endpoints inheriting from `BaseService`, plus completion / tag enumeration)
+
+**Behavior for denied access:**
+
+- Direct-ID reads return `404 Not Found` (not 403) to avoid disclosing the existence of private resources.
+- A structured log event (`*_access_denied`, e.g. `tool_access_denied`) is emitted for forensics.
+
+**What's unchanged:**
+
+- Public-resource access for admin bypass — unchanged.
+- Team-resource access for admin bypass — unchanged.
+- Resource owners can still access their own private resources via owner-email matching at the service layer.
+- DB-resolved admin sessions can still see their own private rows via internal / non-HTTP call paths that preserve the admin email on a `(email, None)` shape — the trusted internal A2A endpoint is the verified example today. HTTP admin requests are intentionally collapsed to `(None, None)` and do not fire the own-private carve-out — by design, to avoid HTTP being a stealthy escalation surface.
+- Scoped tokens (`teams: [...]`) continue to use their scoped team list; admin-bypass detection still requires both `is_admin=true` **and** `teams: null`.
+
+**Migration guidance for integrators:**
+
+- Audit tokens/scripts that currently rely on an admin listing or reading another user's private data. Transfer resource ownership or switch them to `team`-scoped visibility if the cross-user access is intentional.
+- If an admin genuinely needs cross-user visibility for an operational scenario, prefer a properly scoped token (`teams: ["<target_team>"]`) over relying on bypass.
+- Callers of `server_service.get_server`, `gateway_service.get_gateway`, `prompt_service.get_prompt_details`, `resource_service.get_resource_by_id`, and `a2a_service.get_agent_by_name`/`get_agent_card` now accept new optional `user_email` / `token_teams` parameters. Omitting them evaluates as admin-bypass (public + team access, other-users' private denied). Call sites in `mcpgateway/main.py` and `mcpgateway/admin.py` have been updated to forward the caller's scope via `get_scoped_resource_access_context` (now in `mcpgateway/auth_context.py`).
+
+**Related security invariants (see `AGENTS.md`):**
+
+- `public` is platform-public scope, not internet-anonymous.
+- Token-team interpretation continues to flow through `normalize_token_teams()` / `resolve_session_teams()` in `mcpgateway/auth.py`.
+- Non-JWT admin (basic-auth / dev-mode) retains unrestricted access to public and team resources, but is now also denied direct reads of other users' private resources.
+
 ## [1.0.0-RC3] - 2026-04-14 - Auth Hardening, Plugin Multi-Tenancy, Rust Runtime & Multi-Arch
 
 ### Overview
 
@@ -289,7 +289,8 @@ Applies to Tools, Servers, Resources, Prompts, and A2A Agents. All resources are
   - Who sees it: Only the resource owner (owner_email), and only when their token is not a public-only token (token_teams must be non-empty).
   - Team members cannot see or use it unless they are the owner.
   - Public-only tokens (token_teams=[]) cannot access private resources even as the owner.
-  - Mutations: Owner and Platform Admin can update/delete; team owners may be allowed by policy (see Enhancements).
+  - Platform Admin bypass (token teams=null, is_admin=true) does NOT grant visibility or mutation access to another user's private resources (see [#4341](https://github.com/IBM/mcp-context-forge/pull/4341)). Since direct-ID reads return 404 under bypass, update/delete of another user's private resource also fails because the read-before-write gate hides the resource.
+  - Mutations: Only the owner. If cross-user administrative access is required, prefer changing visibility to `team` or using a properly scoped token; admin bypass alone is insufficient.
 
 - Team:
 
 
@@ -317,16 +317,19 @@ Resources in ContextForge have three visibility levels:
 |------------|-------------|-------------|
 | `public` | Accessible to all authenticated users | Everyone with valid token |
 | `team` | Accessible to team members only | Team members + admins (with bypass) |
-| `private` | Accessible to owner only | Resource owner + admins (with bypass) |
+| `private` | Accessible to owner only | Resource owner only — **never** admin bypass |
 
 ### Access Matrix by Token Type
 
 | Token Type | Public Resources | Team Resources | Private Resources |
 |------------|-----------------|----------------|-------------------|
-| Admin Bypass (`teams=null`, `is_admin=true`) | ✅ | ✅ (all teams) | ✅ (all) |
+| Admin Bypass (`teams=null`, `is_admin=true`) | ✅ | ✅ (all teams) | ❌ (owner-only, see note) |
 | Team-Scoped (`teams=["t1"]`) | ✅ | ✅ (own team) | ✅ (own only) |
 | Public-Only (`teams=[]`) | ✅ | ❌ | ❌ |
 
+!!! warning "Admin Bypass Does Not Include Other Users' Private Resources"
+    Since [#4341](https://github.com/IBM/mcp-context-forge/pull/4341), admin bypass **cannot** read, list, update, or delete another user's private resources. Private resources (visibility=`private`) are strictly owner-scoped. If cross-user access is intentional, prefer `team` visibility or a scoped token over relying on bypass. See `docs/architecture/multitenancy.md` for the canonical multi-tenancy model.
+
 !!! warning "Public-Only Token Limitations"
     **Public-only tokens (`teams=[]`) cannot access private resources, even if the resource is owned by the token's user.**
 
 
@@ -66,6 +66,7 @@
 
 # Authentication and password-related imports
 from mcpgateway.auth import get_current_user, get_user_team_roles
+from mcpgateway.auth_context import get_scoped_resource_access_context
 from mcpgateway.cache.a2a_stats_cache import a2a_stats_cache
 from mcpgateway.cache.global_config_cache import global_config_cache
 from mcpgateway.common.models import LogLevel
@@ -2796,12 +2797,13 @@ async def admin_servers_partial_html(
 
 @admin_router.get("/servers/{server_id}", response_model=ServerRead)
 @require_permission("servers.read", allow_admin_bypass=False)
-async def admin_get_server(server_id: str, db: Session = Depends(get_db), user=Depends(get_current_user_with_permissions)) -> Dict[str, Any]:
+async def admin_get_server(server_id: str, request: Request, db: Session = Depends(get_db), user=Depends(get_current_user_with_permissions)) -> Dict[str, Any]:
     """
     Retrieve server details for the admin UI.
 
     Args:
         server_id (str): The ID of the server to retrieve.
+        request (Request): Incoming FastAPI request (for visibility scope resolution).
         db (Session): The database session dependency.
         user (str): The authenticated user dependency.
 
@@ -2820,7 +2822,8 @@ async def admin_get_server(server_id: str, db: Session = Depends(get_db), user=D
     """
     try:
         LOGGER.debug(f"User {get_user_email(user)} requested details for server ID {server_id}")
-        server = await server_service.get_server(db, server_id)
+        auth_user_email, auth_token_teams = get_scoped_resource_access_context(request, user)
+        server = await server_service.get_server(db, server_id, user_email=auth_user_email, token_teams=auth_token_teams)
         return server.masked().model_dump(by_alias=True)
     except ServerNotFoundError as e:
         raise HTTPException(status_code=404, detail=str(e))
@@ -11312,7 +11315,7 @@ async def _safe_entity_search(search_callable, empty_key: str, **kwargs: Any) ->
 
 @admin_router.get("/tools/{tool_id}", response_model=ToolRead)
 @require_permission("tools.read", allow_admin_bypass=False)
-async def admin_get_tool(tool_id: str, db: Session = Depends(get_db), user=Depends(get_current_user_with_permissions)) -> Dict[str, Any]:
+async def admin_get_tool(tool_id: str, request: Request, db: Session = Depends(get_db), user=Depends(get_current_user_with_permissions)) -> Dict[str, Any]:
     """
     Retrieve specific tool details for the admin UI.
 
@@ -11322,6 +11325,7 @@ async def admin_get_tool(tool_id: str, db: Session = Depends(get_db), user=Depen
 
     Args:
         tool_id (str): The ID of the tool to retrieve.
+        request (Request): Incoming FastAPI request (for visibility scope resolution).
         db (Session): Database session dependency.
         user (str): Authenticated user dependency.
 
@@ -11339,11 +11343,19 @@ async def admin_get_tool(tool_id: str, db: Session = Depends(get_db), user=Depen
         'admin_get_tool'
     """
     LOGGER.debug(f"User {get_user_email(user)} requested details for tool ID {tool_id}")
+    auth_user_email, auth_token_teams = get_scoped_resource_access_context(request, user)
     _user_email = get_user_email(user)
     _is_admin = bool(user.get("is_admin", False) if isinstance(user, dict) else getattr(user, "is_admin", False))
     _team_roles = _get_user_team_roles(db, _user_email) if not _is_admin else {}
     try:
-        tool = await tool_service.get_tool(db, tool_id, requesting_user_email=_user_email, requesting_user_is_admin=_is_admin, requesting_user_team_roles=_team_roles)
+        tool = await tool_service.get_tool(
+            db,
+            tool_id,
+            requesting_user_email=auth_user_email,
+            requesting_user_is_admin=_is_admin,
+            requesting_user_team_roles=_team_roles,
+            token_teams=auth_token_teams,
+        )
         return tool.model_dump(by_alias=True)
     except ToolNotFoundError as e:
         raise HTTPException(status_code=404, detail=str(e))
@@ -11945,11 +11957,12 @@ async def admin_set_tool_state(
 
 @admin_router.get("/gateways/{gateway_id}", response_model=GatewayRead)
 @require_permission("gateways.read", allow_admin_bypass=False)
-async def admin_get_gateway(gateway_id: str, db: Session = Depends(get_db), user=Depends(get_current_user_with_permissions)) -> Dict[str, Any]:
+async def admin_get_gateway(gateway_id: str, request: Request, db: Session = Depends(get_db), user=Depends(get_current_user_with_permissions)) -> Dict[str, Any]:
     """Get gateway details for the admin UI.
 
     Args:
         gateway_id: Gateway ID.
+        request: Incoming FastAPI request (for visibility scope resolution).
         db: Database session.
         user: Authenticated user.
 
@@ -11968,7 +11981,8 @@ async def admin_get_gateway(gateway_id: str, db: Session = Depends(get_db), user
     """
     LOGGER.debug(f"User {get_user_email(user)} requested details for gateway ID {gateway_id}")
     try:
-        gateway = await gateway_service.get_gateway(db, gateway_id)
+        auth_user_email, auth_token_teams = get_scoped_resource_access_context(request, user)
+        gateway = await gateway_service.get_gateway(db, gateway_id, user_email=auth_user_email, token_teams=auth_token_teams)
         return gateway.model_dump(by_alias=True)
     except GatewayNotFoundError as e:
         raise HTTPException(status_code=404, detail=str(e))
@@ -12544,11 +12558,12 @@ async def admin_test_resource(resource_uri: str, db: Session = Depends(get_db),
 
 @admin_router.get("/resources/{resource_id}")
 @require_permission("resources.read", allow_admin_bypass=False)
-async def admin_get_resource(resource_id: str, db: Session = Depends(get_db), user=Depends(get_current_user_with_permissions)) -> Dict[str, Any]:
+async def admin_get_resource(resource_id: str, request: Request, db: Session = Depends(get_db), user=Depends(get_current_user_with_permissions)) -> Dict[str, Any]:
     """Get resource details for the admin UI.
 
     Args:
         resource_id: Resource ID.
+        request: Incoming FastAPI request (for visibility scope resolution).
         db: Database session.
         user: Authenticated user.
 
@@ -12567,9 +12582,15 @@ async def admin_get_resource(resource_id: str, db: Session = Depends(get_db), us
     """
     LOGGER.debug(f"User {get_user_email(user)} requested details for resource ID {resource_id}")
     try:
-        resource = await resource_service.get_resource_by_id(db, resource_id, include_inactive=True)
-        # content = await resource_service.read_resource(db, resource_id=resource_id)
-        return {"resource": resource.model_dump(by_alias=True)}  # , "content": None}
+        auth_user_email, auth_token_teams = get_scoped_resource_access_context(request, user)
+        resource = await resource_service.get_resource_by_id(
+            db,
+            resource_id,
+            include_inactive=True,
+            user_email=auth_user_email,
+            token_teams=auth_token_teams,
+        )
+        return {"resource": resource.model_dump(by_alias=True)}
     except ResourceNotFoundError as e:
         raise HTTPException(status_code=404, detail=str(e))
     except Exception as e:
@@ -12942,11 +12963,12 @@ async def admin_set_resource_state(
 
 @admin_router.get("/prompts/{prompt_id}")
 @require_permission("prompts.read", allow_admin_bypass=False)
-async def admin_get_prompt(prompt_id: str, db: Session = Depends(get_db), user=Depends(get_current_user_with_permissions)) -> Dict[str, Any]:
+async def admin_get_prompt(prompt_id: str, request: Request, db: Session = Depends(get_db), user=Depends(get_current_user_with_permissions)) -> Dict[str, Any]:
     """Get prompt details for the admin UI.
 
     Args:
         prompt_id: Prompt ID.
+        request: Incoming FastAPI request (for visibility scope resolution).
         db: Database session.
         user: Authenticated user.
 
@@ -12965,7 +12987,13 @@ async def admin_get_prompt(prompt_id: str, db: Session = Depends(get_db), user=D
     """
     LOGGER.info(f"User {get_user_email(user)} requested details for prompt ID {prompt_id}")
     try:
-        prompt_details = await prompt_service.get_prompt_details(db, prompt_id)
+        auth_user_email, auth_token_teams = get_scoped_resource_access_context(request, user)
+        prompt_details = await prompt_service.get_prompt_details(
+            db,
+            prompt_id,
+            user_email=auth_user_email,
+            token_teams=auth_token_teams,
+        )
         prompt = PromptRead.model_validate(prompt_details)
         return prompt.model_dump(by_alias=True)
     except PromptNotFoundError as e:
 
@@ -4,10 +4,61 @@
 SPDX-License-Identifier: Apache-2.0
 Authors: Mihai Criveti
 
-Shared authentication utilities.
-
-This module provides common authentication functions that can be shared
-across different parts of the application without creating circular imports.
+Authentication primitives: JWT, sessions, API tokens, and team membership.
+
+Purpose (for future implementers)
+---------------------------------
+``mcpgateway`` separates authentication concerns into two modules:
+
+- ``mcpgateway.auth`` (this module) - the **token / session / team model
+  layer**. Helpers here operate on stored artifacts: JWT payloads, session
+  records, API tokens, revocation records, and team-membership rows. They
+  return DB-shaped or dict results and never take a FastAPI ``Request``.
+  Because they are pure in that sense, they can be reused from any context
+  (tests, background tasks, transport hops, RBAC middleware).
+
+- ``mcpgateway.auth_context`` - the **per-request resolution layer**. Helpers
+  there take a FastAPI ``Request`` plus the ``user`` produced by the auth
+  dependency and compute what the caller is allowed to see on this request
+  (Layer 1 visibility context). See ``auth_context.py`` for its purpose block
+  and the ``AGENTS.md`` "Authentication & RBAC Overview" section for the
+  two-layer policy model.
+
+Rule of thumb
+-------------
+- Input is a ``Request``? -> belongs in ``auth_context.py``.
+- Input is a JWT payload, user email, token hash, or team ID? -> belongs here.
+
+Public surface
+--------------
+The names below form this module's stable public API. ``main.py``, routers,
+transports, middleware, and tests all consume them.
+
+    User and session resolution
+        get_current_user(...) - FastAPI dependency
+        get_user_team_roles(db, email) -> dict[team_id, role]
+
+    Token claim normalization (canonical per AGENTS.md)
+        normalize_token_teams(payload) -> list[str] | None
+        resolve_session_teams(...) -> list[str] | None
+
+Private-but-cross-module surface
+--------------------------------
+A few leading-underscore helpers are intentionally imported by other modules
+for a specific reason: they are the **synchronous variants** of async DB
+lookups, wrapped in ``asyncio.to_thread`` by callers in FastAPI hot paths.
+The underscore is a convention marker that says "prefer the async wrapper;
+if you must go sync, use this one" - not a "don't import" signal. The
+callers (``main.py``, ``transports/streamablehttp_transport.py``,
+``utils/verify_credentials.py``) all follow the same ``asyncio.to_thread``
+pattern.
+
+    _check_token_revoked_sync(jti) -> bool
+    _lookup_api_token_sync(token_hash) -> dict | None
+
+If you are tempted to import any other underscore-prefixed name from this
+module, stop and ask whether the caller should really go through a public
+wrapper or whether the helper genuinely deserves promotion to the public API.
 """
 
 # Standard