fix(mcp): enhance CRLF injection protection in header handling

Author: timof1308

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

@@ -152,9 +152,9 @@ def validate_header_format(header_format: str) -> None:
   """
   if "\r" in header_format or "\n" in header_format:
     raise ValueError(
-        "Header format string cannot contain CRLF (carriage return or line feed) "
-        "characters due to header injection risk. "
-        f"Invalid format: {repr(header_format)}"
+        "Header format string cannot contain CRLF (carriage return or line"
+        " feed) characters due to header injection risk. Invalid format:"
+        f" {repr(header_format)}"
     )
 
 

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

@@ -63,6 +63,8 @@
 
 logger = logging.getLogger("google_adk." + __name__)
 
+T = TypeVar("T")
+
 
 def create_session_state_header_provider(
     state_key: str,
@@ -94,7 +96,8 @@ def create_session_state_header_provider(
   **Security Best Practices:**
 
   1. **Token Security**: Use ``request_state`` for sensitive, short-lived tokens
-    (JWTs, API keys) instead of ``session.state`` to avoid persisting sensitive data.
+    (JWTs, API keys) instead of ``session.state`` to avoid persisting
+    sensitive data.
 
   2. **Header Validation**: Header names and values are automatically validated
     according to RFC 7230 to prevent injection attacks.
@@ -141,7 +144,9 @@ def create_session_state_header_provider(
       response = await agent.run(
           RunAgentRequest(
               session_id="user-123",
-              request_state={"jwt_token": "eyJhbG..."}  # Ephemeral, not persisted
+              request_state={  # Ephemeral, not persisted
+                  "jwt_token": "eyJhbG..."
+              }
           )
       )
   """
@@ -157,6 +162,10 @@ 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.
+    formatted_value = formatted_value.replace("\r", "").replace("\n", "")
     sanitized_value = sanitize_header_value(formatted_value)
 
     return {header_name: sanitized_value}

tests/unittests/tools/mcp_tool/test_jwt_token_propagation.py

@@ -549,24 +549,35 @@ def test_header_value_sanitization_rfc_compliant(self):
       result = sanitize_header_value(input_val)
       assert result == expected
 
-  def test_header_value_preserves_rfc_folding(self):
-    """Test that legitimate CRLF sequences for header folding are preserved."""
-    from google.adk.tools.mcp_tool._internal import sanitize_header_value
+  def test_header_value_crlf_stripped_in_provider(self):
+    """Test that CRLF in state values is stripped to prevent header injection."""
+    from google.adk.tools.mcp_tool.mcp_toolset import create_session_state_header_provider
 
-    # Multi-line headers with proper folding should be preserved (RFC 7230 §3.2.4)
-    folding_cases = [
-        ("Authorization: Bearer token", "Authorization: Bearer token"),
-        ("X-Custom: value1\r\n\tvalue2", "X-Custom: value1\r\n\tvalue2"),
-        ("X-Header: line1\r\n line2", "X-Header: line1\r\n line2"),
-        (
-            "Content-Type: application/json\r\n\tcharset=utf-8",
-            "Content-Type: application/json\r\n\tcharset=utf-8",
-        ),
+    # CRLF in the token value should be stripped from the final header
+    malicious_cases = [
+        ("tok\r\nX-Injected: evil", "Bearer tokX-Injected: evil"),
+        ("tok\nInjected: bad", "Bearer tokInjected: bad"),
+        ("tok\rAnother: header", "Bearer tokAnother: header"),
+        ("tok\r\n", "Bearer tok"),
+        ("tok\n", "Bearer tok"),
+        ("tok\r", "Bearer tok"),
     ]
 
-    for folding_case in folding_cases:
-      result = sanitize_header_value(folding_case[0])
-      assert result == folding_case[1]
+    for malicious_value, expected_header in malicious_cases:
+      mock_context = Mock(spec=ReadonlyContext)
+      mock_context.state = {"jwt_token": malicious_value}
+      provider = create_session_state_header_provider(state_key="jwt_token")
+      headers = provider(mock_context)
+      assert headers == {
+          "Authorization": expected_header
+      }, f"Failed for value {malicious_value!r}"
+
+    # Clean values should be unchanged
+    mock_context = Mock(spec=ReadonlyContext)
+    mock_context.state = {"jwt_token": "clean-token-123"}
+    provider = create_session_state_header_provider(state_key="jwt_token")
+    headers = provider(mock_context)
+    assert headers == {"Authorization": "Bearer clean-token-123"}
 
   def test_header_value_validation_rfc_compliant(self):
     """Test that header value validation is RFC 7230 compliant."""
@@ -659,8 +670,8 @@ def test_binary_data_handling(self):
 
   def test_header_format_crlf_injection_protection(self):
     """Test that header format strings with CRLF sequences are rejected."""
-    from google.adk.tools.mcp_tool.mcp_toolset import create_session_state_header_provider
     from google.adk.tools.mcp_tool._internal import validate_header_format
+    from google.adk.tools.mcp_tool.mcp_toolset import create_session_state_header_provider
 
     # Valid format strings should be accepted
     valid_formats = [
@@ -691,5 +702,5 @@ def test_header_format_crlf_injection_protection(self):
       create_session_state_header_provider(
           state_key="token",
           header_name="Authorization",
-          header_format="Bearer {value}\r\nX-Injected: evil"
+          header_format="Bearer {value}\r\nX-Injected: evil",
       )