fix(mcp): harden header handling and add config validation

- Add CRLF/TAB to _DANGEROUS_CHARS and sanitize auth headers
- Standardize header merge order (header_provider takes precedence)
- Add model validator for state_header_mapping/state_header_format
- Warn on duplicate header names in combined providers
- Add request_state/state_delta/invocation_id to sync Runner.run()
- Use TYPE_CHECKING guard in types.py

Author: timof1308

src/google/adk/runners.py

@@ -439,7 +439,10 @@ def run(
       *,
       user_id: str,
       session_id: str,
-      new_message: types.Content,
+      new_message: Optional[types.Content] = None,
+      invocation_id: Optional[str] = None,
+      state_delta: Optional[dict[str, Any]] = None,
+      request_state: Optional[dict[str, Any]] = None,
       run_config: Optional[RunConfig] = None,
   ) -> Generator[Event, None, None]:
     """Runs the agent.
@@ -457,6 +460,9 @@ def run(
       user_id: The user ID of the session.
       session_id: The session ID of the session.
       new_message: A new message to append to the session.
+      invocation_id: The invocation id to resume.
+      state_delta: Optional state changes to apply to the session.
+      request_state: Optional ephemeral state for the request.
       run_config: The run config for the agent.
 
     Yields:
@@ -472,6 +478,9 @@ async def _invoke_run_async():
                 user_id=user_id,
                 session_id=session_id,
                 new_message=new_message,
+                invocation_id=invocation_id,
+                state_delta=state_delta,
+                request_state=request_state,
                 run_config=run_config,
             )
         ) as agen:
@@ -561,6 +570,10 @@ async def _run_with_trace(
         is_resumable = (
             self.resumability_config and self.resumability_config.is_resumable
         )
+        # Three-branch decision tree:
+        #   A) invocation_id provided → resume that specific invocation
+        #   B) not resumable → must start a new invocation (requires new_message)
+        #   C) resumable, no explicit invocation_id → resolve or create new
         if invocation_id:
           if not is_resumable:
             raise ValueError(
@@ -614,8 +627,9 @@ async def _run_with_trace(
             if invocation_context.end_of_agents.get(
                 invocation_context.agent.name
             ):
-              # Directly return if the current agent in invocation context is
-              # already final.
+              # Agent already completed in a prior invocation — skip execution
+              # and return immediately. This can happen when resuming a
+              # completed agent in a multi-agent pipeline.
               return
 
         async def execute(ctx: InvocationContext) -> AsyncGenerator[Event]:

src/google/adk/tools/mcp_tool/_internal.py

@@ -20,13 +20,15 @@
 **Security Notes:**
 
 - Header validation implements RFC 7230 §3.2 for proper HTTP header format
-- Only truly dangerous control characters are removed from header values
+- All ASCII control characters (0x00-0x1F) and DEL (0x7F) are removed from
+  header values to prevent injection
 - All functions log security-relevant warnings when appropriate
 
 **RFC 7230 Compliance:**
 
 - Header names: only letters, digits, and hyphens allowed
-- Header values: control characters (0x00-0x1F, 0x7F) are dangerous
+- Header values: control characters including CRLF (0x00-0x1F, 0x7F) are
+  removed to prevent injection
 
 **Attack Prevention:**
 
@@ -58,8 +60,11 @@
     "\x06",
     "\x07",
     "\x08",
+    "\x09",
+    "\x0a",
     "\x0b",
     "\x0c",
+    "\x0d",
     "\x0e",
     "\x0f",
     "\x10",
@@ -133,10 +138,10 @@ def sanitize_header_value(value: Any) -> str:
   if not isinstance(value, str):
     value = str(value)
 
-  # Remove only characters that are truly dangerous for HTTP headers
-  # These are control characters that can break parsing or enable injection
-  # We DON'T remove all \r\n sequences as that would break legitimate multi-line headers
-  # and violate RFC 7230 §3.2.4 which allows header folding
+  # Remove CRLF and control characters to prevent header injection.
+  # Header folding (obs-fold) was deprecated by RFC 7230 and obsoleted
+  # by RFC 9110. CRLF in header values is the primary vector for
+  # header injection and response splitting attacks.
   sanitized_chars = []
   for char in value:
     if char not in _DANGEROUS_CHARS:

src/google/adk/tools/mcp_tool/mcp_toolset.py

@@ -163,9 +163,9 @@ def provider(ctx: ReadonlyContext) -> Dict[str, str]:
 
     validate_header_value(state_key, value, strict=strict)
     formatted_value = header_format.format(value=value)
-    # Strip CRLF from the interpolated value to prevent header injection.
-    # The format string is validated at construction time, but the runtime
-    # value comes from session state and must never contain CRLF here.
+    # Defense in depth: strip CRLF before sanitization. sanitize_header_value
+    # also strips these, but we strip early to prevent injection via format
+    # string interpolation.
     formatted_value = formatted_value.replace("\r", "").replace("\n", "")
     sanitized_value = sanitize_header_value(formatted_value)
 
@@ -193,6 +193,15 @@ def combined_provider(ctx: ReadonlyContext) -> Dict[str, str]:
       try:
         provider_headers = provider(ctx)
         if provider_headers:
+          overlapping = set(headers.keys()) & set(provider_headers.keys())
+          if overlapping:
+            logger.warning(
+                "Duplicate header names %s from header provider "
+                "%d/%d. Last value wins.",
+                overlapping,
+                i + 1,
+                num_providers,
+            )
           headers.update(provider_headers)
       except Exception as e:
         logger.error(f"Header provider {i+1}/{num_providers} failed: {e}")
@@ -301,7 +310,9 @@ def __init__(
         MCP server.
       sampling_capabilities: Optional capabilities for sampling.
       credential_key: A user specified key used to load and save this credential
-        in a credential service. Used with auth_scheme.
+        in a credential service. Used with auth_scheme. Note: when both
+        credential_key and header_provider are configured, header_provider
+        values take precedence over auth headers for the same header names.
     """
 
     # --- BEGIN BOUND TOKEN PATCH ---
@@ -433,6 +444,10 @@ def _get_auth_headers(
           # Default to using scheme name as header
           headers = {self._auth_config.auth_scheme.name: credential.api_key}
 
+    # Sanitize all header values to prevent injection attacks.
+    if headers:
+      headers = {k: sanitize_header_value(v) for k, v in headers.items()}
+
     return headers
 
   async def _execute_with_session(
@@ -444,17 +459,17 @@ async def _execute_with_session(
     """Creates a session and executes a coroutine with it."""
     headers: Dict[str, str] = {}
 
-    # Add headers from header_provider if available
+    # Add auth headers from exchanged credential first
+    auth_headers = self._get_auth_headers(readonly_context)
+    if auth_headers:
+      headers.update(auth_headers)
+
+    # Add headers from header_provider (takes precedence over auth headers)
     if self._header_provider and readonly_context:
       provider_headers = self._header_provider(readonly_context)
       if provider_headers:
         headers.update(provider_headers)
 
-    # Add auth headers from exchanged credential if available
-    auth_headers = self._get_auth_headers(readonly_context)
-    if auth_headers:
-      headers.update(auth_headers)
-
     session = await self._mcp_session_manager.create_session(
         headers=headers if headers else None
     )
@@ -761,3 +776,33 @@ def _check_only_one_params_field(self):
           " set."
       )
     return self
+
+  @model_validator(mode="after")
+  def _validate_state_header_config(self):
+    """Validates state_header_mapping and state_header_format consistency."""
+    if not self.state_header_mapping:
+      if self.state_header_format:
+        raise ValueError(
+            "state_header_format cannot be set without state_header_mapping."
+        )
+      return self
+
+    # Validate header names in state_header_mapping values
+    for state_key, header_name in self.state_header_mapping.items():
+      validate_header_name(header_name)
+
+    # Validate state_header_format keys match header names
+    if self.state_header_format:
+      header_names = set(self.state_header_mapping.values())
+      for format_key in self.state_header_format:
+        if format_key not in header_names:
+          raise ValueError(
+              f'state_header_format key "{format_key}" does not match'
+              " any header name in state_header_mapping values."
+              f" Expected one of: {sorted(header_names)}"
+          )
+      # Validate format strings don't contain CRLF
+      for header_name, fmt in self.state_header_format.items():
+        validate_header_format(fmt)
+
+    return self

src/google/adk/tools/mcp_tool/types.py

@@ -16,7 +16,9 @@
 
 from typing import Callable
 from typing import Dict
+from typing import TYPE_CHECKING
 
-from ...agents.readonly_context import ReadonlyContext
+if TYPE_CHECKING:
+  from ...agents.readonly_context import ReadonlyContext
 
-HeaderProvider = Callable[[ReadonlyContext], Dict[str, str]]
+HeaderProvider = Callable[["ReadonlyContext"], Dict[str, str]]

tests/unittests/tools/mcp_tool/test_jwt_token_propagation.py

@@ -14,6 +14,7 @@
 
 """Tests for JWT token propagation feature in MCP toolset."""
 
+import logging
 import sys
 from unittest.mock import Mock
 
@@ -538,6 +539,10 @@ def test_header_value_sanitization_rfc_compliant(self):
         ("nack\x15syn\x16etb\x17", "nacksynetb"),  # NAK, SYN, ETB
         ("can\x18em\x19sub\x1Aesc", "canemsubesc"),  # CAN, EM, SUB, ESC
         ("fs\x1ags\x1brs\x1cus", "fsgsrsus"),  # FS, GS, RS, US
+        ("tok\r\nX-Injected: evil", "tokX-Injected: evil"),  # CRLF
+        ("tok\nInjected: bad", "tokInjected: bad"),  # LF
+        ("tok\rAnother: header", "tokAnother: header"),  # CR
+        ("tab\ttest", "tabtest"),  # TAB should be removed
         ("space\x20test", "space test"),  # Space should be preserved
         (
             "normal!@#$%^&*()test",
@@ -671,3 +676,102 @@ def test_header_format_crlf_injection_protection(self):
           header_name="Authorization",
           header_format="Bearer {value}\r\nX-Injected: evil",
       )
+
+
+class TestMcpToolsetConfigValidation:
+  """Test suite for McpToolsetConfig state header validation."""
+
+  def test_format_without_mapping_raises(self):
+    """Test that state_header_format without mapping raises ValueError."""
+    with pytest.raises(ValueError, match="state_header_format cannot be set"):
+      McpToolsetConfig(
+          stdio_server_params=StdioServerParameters(
+              command="test_command", args=[]
+          ),
+          state_header_format={"Authorization": "Bearer {value}"},
+      )
+
+  def test_format_key_not_in_mapping_values_raises(self):
+    """Test that format key not matching any mapping value raises."""
+    with pytest.raises(ValueError, match="does not match"):
+      McpToolsetConfig(
+          stdio_server_params=StdioServerParameters(
+              command="test_command", args=[]
+          ),
+          state_header_mapping={"jwt_token": "Authorization"},
+          state_header_format={"X-Wrong-Header": "Bearer {value}"},
+      )
+
+  def test_invalid_header_name_in_mapping_raises(self):
+    """Test that invalid header name in mapping value raises ValueError."""
+    with pytest.raises(ValueError, match="invalid characters"):
+      McpToolsetConfig(
+          stdio_server_params=StdioServerParameters(
+              command="test_command", args=[]
+          ),
+          state_header_mapping={"jwt_token": "Authorization\n"},
+      )
+
+  def test_crlf_in_format_value_raises(self):
+    """Test that CRLF in format string raises ValueError."""
+    with pytest.raises(ValueError, match="CRLF"):
+      McpToolsetConfig(
+          stdio_server_params=StdioServerParameters(
+              command="test_command", args=[]
+          ),
+          state_header_mapping={"jwt_token": "Authorization"},
+          state_header_format={
+              "Authorization": "Bearer {value}\r\nX-Injected: evil"
+          },
+      )
+
+  def test_valid_config_passes_validation(self):
+    """Test that valid config passes all validation."""
+    config = McpToolsetConfig(
+        stdio_server_params=StdioServerParameters(
+            command="test_command", args=[]
+        ),
+        state_header_mapping={
+            "jwt_token": "Authorization",
+            "tenant_id": "X-Tenant-ID",
+        },
+        state_header_format={"Authorization": "Bearer {value}"},
+    )
+    assert config.state_header_mapping is not None
+
+
+class TestCombinedHeaderProviderDuplicateWarning:
+  """Test suite for duplicate header warning in combined provider."""
+
+  def test_warns_on_duplicate_headers(self, caplog):
+    """Test that duplicate header names trigger a warning."""
+    from google.adk.tools.mcp_tool.mcp_toolset import create_combined_header_provider
+
+    provider1 = lambda ctx: {"Authorization": "Bearer token1"}  # noqa: E731
+    provider2 = lambda ctx: {"Authorization": "Bearer token2"}  # noqa: E731
+
+    combined = create_combined_header_provider([provider1, provider2])
+
+    with caplog.at_level(logging.WARNING, logger="google_adk"):
+      headers = combined(Mock(spec=ReadonlyContext))
+
+    assert headers["Authorization"] == "Bearer token2"
+    assert "Duplicate header names" in caplog.text
+
+  def test_no_warning_without_duplicates(self, caplog):
+    """Test that no warning is logged when headers don't overlap."""
+    from google.adk.tools.mcp_tool.mcp_toolset import create_combined_header_provider
+
+    provider1 = lambda ctx: {"Authorization": "Bearer token1"}  # noqa: E731
+    provider2 = lambda ctx: {"X-Tenant-ID": "tenant-123"}  # noqa: E731
+
+    combined = create_combined_header_provider([provider1, provider2])
+
+    with caplog.at_level(logging.WARNING, logger="google_adk"):
+      headers = combined(Mock(spec=ReadonlyContext))
+
+    assert "Duplicate header names" not in caplog.text
+    assert headers == {
+        "Authorization": "Bearer token1",
+        "X-Tenant-ID": "tenant-123",
+    }