Prevent OAuth token leakage via static Authorization headers (HolmesGPT#2095)

## Summary

This PR adds security guards to prevent privilege escalation when OAuth
is configured but no user token is available. It ensures that static
Authorization headers from configuration cannot silently substitute for
missing per-user OAuth tokens, which could leak shared service-account
credentials.

## Key Changes

### 1. Server-Mode User ID Guard in OAuthTokenManager
- Added `_is_server_mode()` method to detect when running against the
DB-backed token store (multi-user server)
- Added guards in `get_access_token()`, `has_token()`, and
`store_token()` to refuse cache operations without a valid user_id in
server mode
- Prevents cache key collisions where one caller's token could be served
to another via the DEFAULT_CLI_USER fallback
- CLI mode (DiskTokenStore) continues to work normally with
DEFAULT_CLI_USER as the legitimate single-user identity

### 2. Static Authorization Header Stripping in RemoteMCPToolset
- Modified `_render_headers()` to strip any Authorization header
(case-insensitive) when OAuth is configured but no token is available
- Only strips headers when OAuth is enabled AND no cached token exists
- Preserves other headers and only strips Authorization to prevent
service-account credential substitution
- Logs appropriate warnings distinguishing between "no token" and "no
token + header stripped" cases

### 3. Comprehensive Test Coverage
- Added `TestServerModeUserIdGuard` class with 5 tests covering:
  - Token retrieval/storage refusal without user_id in server mode
  - Legitimate per-user token access still works
  - CLI mode continues to allow DEFAULT_CLI_USER
- Added 3 tests for Authorization header stripping:
  - Strips static Authorization when OAuth enabled but no token
- Case-insensitive matching (handles "Authorization", "authorization",
"AUTHORIZATION")
  - Preserves static Authorization when OAuth is disabled

## Implementation Details

- The server-mode guard mirrors the existing guard in
`DalTokenStore.get_token()` to maintain consistency
- Authorization header stripping uses case-insensitive key matching to
handle all header case variations
- Logging distinguishes between scenarios to aid debugging (token
missing vs. token missing + header stripped)
- No changes to the OAuth flow itself—only prevents unsafe fallback
behavior

https://claude.ai/code/session_01AAECGd71rhfPJfqgDWHtmp

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Bug Fixes**
* Prevented cross-user token use; token checks and stores now fail
safely when no user ID is present.
* Static Authorization headers are suppressed when per-user OAuth tokens
are required, with clearer warnings for likely 401s.

* **Chores**
* CLI and interactive flows now consistently supply a default user ID in
request contexts.

* **Tests**
* Expanded coverage for per-user token behavior, header suppression, and
CLI/interactive flows.

<!-- review_stack_entry_start -->

[![Review Change
Stack](https://storage.googleapis.com/coderabbit_public_assets/review-stack-in-coderabbit-ui.svg)](https://app.coderabbit.ai/change-stack/HolmesGPT/holmesgpt/pull/2095?utm_source=github_walkthrough&utm_medium=github&utm_campaign=change_stack)

<!-- review_stack_entry_end -->
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Signed-off-by: Claude <noreply@anthropic.com>
Co-authored-by: Claude <noreply@anthropic.com>

Author: naomi-robusta

holmes/core/oauth_utils.py

@@ -12,7 +12,6 @@
 from urllib.parse import parse_qs, urlencode, urlparse
 
 import httpx
-from holmes.common.env_vars import DEFAULT_CLI_USER
 from mcp.client.auth.oauth2 import PKCEParameters
 from mcp.client.auth.utils import (
     build_oauth_authorization_server_metadata_discovery_urls,
@@ -76,8 +75,7 @@ def eager_load_oauth_tools(executor: Any) -> None:
         if not ts._mcp_config.oauth.authorization_url:
             continue
         for user_id in token_mgr.get_cached_user_ids(ts._mcp_config.oauth):
-            request_ctx = {"user_id": user_id} if user_id != DEFAULT_CLI_USER else None
-            executor.oauth_connector.load_tools_for_user(user_id, ts, request_ctx)
+            executor.oauth_connector.load_tools_for_user(user_id, ts, {"user_id": user_id})
 
 
 # exchange_code_for_tokens is re-exported from oauth_config (imported above)

holmes/core/tools_utils/oauth_tool_connector.py

@@ -150,9 +150,10 @@ def store_user_tools(self, user_id: str, toolset_name: str, tools: List[Tool]) -
 
     def resolve_tools(self, user_id: Optional[str]) -> Optional[Dict[str, List[Tool]]]:
         """Return per-user OAuth tools if available, or None."""
-        key = user_id or _get_token_manager().require_user_id(None)
+        if not user_id:
+            return None
         with self._lock:
-            user_tools = self._user_tools.get(key)
+            user_tools = self._user_tools.get(user_id)
             return dict(user_tools) if user_tools else None
 
     def apply_user_tools(
@@ -190,18 +191,20 @@ def apply_user_tools(
 
     def find_tool(self, name: str, user_id: Optional[str]) -> Optional[Tool]:
         """Look up a tool in the per-user OAuth tools store."""
-        key = user_id or _get_token_manager().require_user_id(None)
+        if not user_id:
+            return None
         with self._lock:
-            for toolset_tools in self._user_tools.get(key, {}).values():
+            for toolset_tools in self._user_tools.get(user_id, {}).values():
                 for tool in toolset_tools:
                     if tool.name == name:
                         return tool
         return None
 
     def get_toolset(self, tool_name: str, user_id: Optional[str]) -> Optional[Any]:
         """Return the toolset for a per-user OAuth tool, or None."""
-        key = user_id or _get_token_manager().require_user_id(None)
-        return self._user_tool_to_toolset.get(key, {}).get(tool_name)
+        if not user_id:
+            return None
+        return self._user_tool_to_toolset.get(user_id, {}).get(tool_name)
 
 
     # ── Error handling helpers ─────────────────────────────────────────

holmes/interactive.py

@@ -48,6 +48,7 @@
 from rich.table import Table
 from rich.text import Text
 
+from holmes.common.env_vars import DEFAULT_CLI_USER
 from holmes.config import Config
 from holmes.core.config import config_path_dir
 from holmes.core.init_event import StatusEvent, StatusEventKind, ToolsetStatus
@@ -2638,6 +2639,7 @@ def _run_ai_stream(
                                 cancel_event=_cancel_event,
                                 tool_number_offset=_tool_number_offset,
                                 iteration_offset=_iteration_offset,
+                                request_context={"user_id": DEFAULT_CLI_USER},
                             )
                             tool_decisions = None
                             last_event = None

holmes/main.py

@@ -38,7 +38,13 @@
 from holmes.core.tool_calling_llm import LLMResult, ToolCallingLLM
 from holmes.core.tools import PrerequisiteCacheMode, ToolsetTag, pretty_print_toolset_status
 from holmes.core.tools_utils.filesystem_result_storage import tool_result_storage
+from holmes.common.env_vars import DEFAULT_CLI_USER
 from holmes.core.oauth_utils import enable_disk_token_store
+
+# CLI runs as a single local identity. Setting this explicitly on every LLM
+# invocation keeps the OAuth token manager mode-agnostic: it never has to
+# guess whether a missing user_id means "CLI" or "buggy server caller".
+_CLI_REQUEST_CONTEXT = {"user_id": DEFAULT_CLI_USER}
 from holmes.core.tracing import SpanType, TracingFactory
 from holmes.interactive import InitProgressRenderer, run_interactive_loop, silence_display_loggers
 from holmes.plugins.destinations import DestinationType
@@ -160,6 +166,9 @@ def _investigate_issue(
     config: Config,
 ) -> LLMResult:
     """Investigate an issue using the standard ask system prompt with investigation additions."""
+    # Enable disk-based OAuth token storage so MCP tokens previously
+    # authorized via `holmes ask` are reused here (idempotent).
+    enable_disk_token_store()
     investigation_additions = f"Provide a terse analysis of the following {issue.source_type} alert/issue and why it is firing."
     system_prompt = build_system_prompt(
         toolsets=ai.tool_executor.toolsets,
@@ -177,7 +186,7 @@ def _investigate_issue(
         {"role": "system", "content": system_prompt},
         {"role": "user", "content": user_prompt},
     ]
-    return ai.call(messages)
+    return ai.call(messages, request_context=_CLI_REQUEST_CONTEXT)
 
 
 # TODO: add streaming output
@@ -400,7 +409,7 @@ def ask(
             f'holmes ask "{prompt}"', span_type=SpanType.TASK
         ) as trace_span:
             trace_span.log(input=prompt, metadata={"type": "user_question"})
-            response = ai.call(messages, trace_span=trace_span)
+            response = ai.call(messages, trace_span=trace_span, request_context=_CLI_REQUEST_CONTEXT)
             trace_span.log(
                 output=response.result,
             )
@@ -730,6 +739,9 @@ def ticket(
             tool_results_dir=tool_results_dir,
         )
 
+        # Enable disk-based OAuth token storage for CLI mode
+        enable_disk_token_store()
+
         # Render ticket-specific additions
         ticket_additions = load_and_render_prompt(
             prompt="builtin://_ticket_additions.jinja2",
@@ -760,7 +772,7 @@ def ticket(
             {"role": "system", "content": system_prompt},
             {"role": "user", "content": ticket_user_prompt},
         ]
-        result = ai.call(messages)
+        result = ai.call(messages, request_context=_CLI_REQUEST_CONTEXT)
 
         console.print(Rule())
         console.print(

holmes/plugins/toolsets/mcp/oauth_token_manager.py

@@ -135,10 +135,18 @@ def get_access_token(
     ) -> Optional[str]:
         """Return a valid access token, checking cache → refresh → persistent store.
 
+        A user_id must be present on request_context. CLI callers must pass
+        DEFAULT_CLI_USER explicitly; the server must pass the authenticated
+        user. Without a user_id the cache cannot be safely keyed, so we
+        refuse to serve any token (mirrors DalTokenStore.get_token's guard).
+
         Returns None if no token is available anywhere (caller should initiate OAuth flow).
         """
-        cache_key = self._get_cache_key(oauth_config, request_context)
         user_id = _get_user_id(request_context)
+        if not user_id:
+            return None
+
+        cache_key = self._build_cache_key(user_id, oauth_config.authorization_url)
 
         # 1. Check in-memory cache
         cached = self._cache.get_valid_access_token(cache_key)
@@ -178,7 +186,10 @@ def has_token(
         request_context: Optional[Dict[str, Any]] = None,
     ) -> bool:
         """Check if any token (access or refreshable) is available in cache."""
-        cache_key = self._get_cache_key(oauth_config, request_context)
+        user_id = _get_user_id(request_context)
+        if not user_id:
+            return False
+        cache_key = self._build_cache_key(user_id, oauth_config.authorization_url)
         return self._cache.has_token_or_refresh(cache_key)
 
     def store_token(
@@ -190,8 +201,12 @@ def store_token(
         store_to_disk: bool = False,
     ) -> None:
         """Store a token to cache and persistent store."""
-        cache_key = self._get_cache_key(oauth_config, request_context)
         user_id = _get_user_id(request_context)
+        if not user_id:
+            logger.warning("OAuthTokenManager: refusing to store token without a user_id")
+            return
+
+        cache_key = self._build_cache_key(user_id, oauth_config.authorization_url)
         access_token = token_data.get("access_token")
         if not access_token:
             logger.warning("OAuthTokenManager: store_token called with no access_token")
@@ -226,17 +241,17 @@ def store_token(
         )
 
     def require_user_id(self, request_context: Optional[Dict[str, Any]]) -> str:
-        """Return a user_id, using DEFAULT_CLI_USER in CLI mode or raising in server mode.
+        """Return the user_id from request_context, or raise if absent.
 
-        CLI mode (DiskTokenStore / no store): returns DEFAULT_CLI_USER.
-        Server mode (DalTokenStore): raises ValueError if user_id is missing.
+        Callers (CLI, server, conversation worker) are responsible for putting
+        a real user_id on request_context — CLI uses DEFAULT_CLI_USER, server
+        uses the authenticated user. Missing user_id is a programming error
+        (would silently corrupt downstream per-user storage), so we fail fast.
         """
         user_id = _get_user_id(request_context)
-        if user_id:
-            return user_id
-        if isinstance(self._store, DalTokenStore):
-            return None
-        return DEFAULT_CLI_USER
+        if not user_id:
+            raise ValueError("OAuthTokenManager: user_id is required in request_context")
+        return user_id
 
     def shutdown(self) -> None:
         """Stop the background refresh thread."""
@@ -400,7 +415,12 @@ def _do_refresh_request(
     # ── Key helpers ────────────────────────────────────────────────────
 
     def _get_cache_key(self, oauth_config: Any, request_context: Optional[Dict[str, Any]]) -> str:
-        user_id = _get_user_id(request_context) or DEFAULT_CLI_USER
+        """Build a cache key from request_context. Raises if user_id is missing —
+        callers must put a user_id on the context (DEFAULT_CLI_USER in CLI mode,
+        authenticated user in server mode)."""
+        user_id = _get_user_id(request_context)
+        if not user_id:
+            raise ValueError("OAuthTokenManager: user_id is required in request_context")
         return self._build_cache_key(user_id, oauth_config.authorization_url)
 
     @staticmethod

holmes/plugins/toolsets/mcp/toolset_mcp.py

@@ -834,7 +834,24 @@ def _render_headers(
                 final_headers["Authorization"] = f"Bearer {cached_token}"
                 logger.debug("OAuth token injected for MCP server %s", self.name)
             else:
-                logger.warning("OAuth MCP server %s: no cached token — request will likely 401", self.name)
+                # OAuth is configured for this toolset but we have no user token
+                # (e.g. user_id is missing in server mode, or no one has authenticated).
+                # Strip any Authorization header coming from static `headers` /
+                # `extra_headers` so a shared service-account credential cannot
+                # silently substitute for the absent per-user OAuth token.
+                stripped = [k for k in list(final_headers.keys()) if k.lower() == "authorization"]
+                for k in stripped:
+                    del final_headers[k]
+                if stripped:
+                    logger.warning(
+                        "OAuth MCP server %s: no OAuth token available and static Authorization header suppressed — request will 401",
+                        self.name,
+                    )
+                else:
+                    logger.warning(
+                        "OAuth MCP server %s: no OAuth token available — request will 401",
+                        self.name,
+                    )
 
         return final_headers if final_headers else None