fix(api_server): add optional ADK_API_TOKEN Bearer auth middleware

[f4x0rz]

Adds an optional ADK_API_TOKEN Bearer auth middleware to the ApiServer.

Why

The ApiServer registers /run, /run_sse, and the session CRUD routes (/apps/{app}/users/{user}/sessions/...) without per-route authentication. When an operator binds the server to a network-reachable address with no upstream auth layer (reverse proxy, sidecar, IAP, network policy), those routes reach in-process Python code execution via UnsafeLocalCodeExecutor.execute_code and per-user session state with no caller identity check.

This is the smallest opt-in mitigation that does not break any existing deployment.

Behavior

• ADK_API_TOKEN unset or empty: no-op, every request passes through unchanged. This preserves current behavior for deployments that already gate access upstream.
• ADK_API_TOKEN set: every request other than GET /health and GET /version must carry an Authorization: Bearer <token> header whose token equals ADK_API_TOKEN, otherwise the request is rejected with HTTP 401 and a WWW-Authenticate: Bearer realm="adk" header. Liveness probes do not need credentials.

The middleware mirrors the shape of the existing _OriginCheckMiddleware and _DefaultAppRewriteMiddleware in the same file.

Testing Plan

Setup, on Windows 11, Python 3.13.13:

cd adk-python
python -m venv .venv
.venv/Scripts/python.exe -m pip install -e .
.venv/Scripts/python.exe -m pip install pytest pytest-asyncio uvicorn starlette httpx

Unit tests

.venv/Scripts/python.exe -m pytest tests/unittests/cli/test_bearer_auth_middleware.py -v

Output:

collected 9 items

tests/unittests/cli/test_bearer_auth_middleware.py::test_token_unset_passes_request_through_unchanged PASSED [ 11%]
tests/unittests/cli/test_bearer_auth_middleware.py::test_empty_string_token_is_treated_as_unset PASSED [ 22%]
tests/unittests/cli/test_bearer_auth_middleware.py::test_token_set_request_without_auth_header_is_rejected PASSED [ 33%]
tests/unittests/cli/test_bearer_auth_middleware.py::test_token_set_wrong_bearer_is_rejected PASSED [ 44%]
tests/unittests/cli/test_bearer_auth_middleware.py::test_token_set_correct_bearer_is_accepted PASSED [ 55%]
tests/unittests/cli/test_bearer_auth_middleware.py::test_token_set_public_paths_are_always_open[/health] PASSED [ 66%]
tests/unittests/cli/test_bearer_auth_middleware.py::test_token_set_public_paths_are_always_open[/version] PASSED [ 77%]
tests/unittests/cli/test_bearer_auth_middleware.py::test_non_http_scopes_are_passed_through_unchanged PASSED [ 88%]
tests/unittests/cli/test_bearer_auth_middleware.py::test_session_route_requires_auth_when_token_set PASSED [100%]

======================== 9 passed in 1.94s ========================

Manual integration test via uvicorn + curl

A minimal Starlette app wraps the middleware around a stub /run, /health, and /version (same shape the real ApiServer uses at api_server.py:1048):

# integration_app.py
import os
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from google.adk.cli.api_server import _BearerAuthMiddleware

async def run(request): return JSONResponse({"ok": True})
async def health(request): return JSONResponse({"status": "ok"})
async def version(request): return JSONResponse({"version": "test"})

app = Starlette(routes=[
    Route("/run", run, methods=["POST"]),
    Route("/health", health, methods=["GET"]),
    Route("/version", version, methods=["GET"]),
])

class _Wrap:
    def __init__(self, app):
        self._inner = _BearerAuthMiddleware(app, token=os.environ.get("ADK_API_TOKEN"))
    async def __call__(self, scope, receive, send):
        await self._inner(scope, receive, send)

wrapped_app = _Wrap(app)

Start uvicorn for each scenario, then fire curl.

Scenario 1: ADK_API_TOKEN unset (default). Every route open.

unset ADK_API_TOKEN
python -m uvicorn integration_app:wrapped_app --port 18181 &

curl -sS -w "\nHTTP %{http_code}\n" -X POST http://127.0.0.1:18181/run -d '{}'
# {"ok":true}
# HTTP 200

curl -sS -w "\nHTTP %{http_code}\n" http://127.0.0.1:18181/health
# {"status":"ok"}
# HTTP 200

Scenario 2: ADK_API_TOKEN=secret-12345, no token from client. HTTP 401 with WWW-Authenticate.

export ADK_API_TOKEN=secret-12345
python -m uvicorn integration_app:wrapped_app --port 18181 &

curl -sS -i -X POST http://127.0.0.1:18181/run -d '{}' | head -15
# HTTP/1.1 401 Unauthorized
# server: uvicorn
# content-type: application/json
# www-authenticate: Bearer realm="adk"
# content-length: 35
#
# {"error":"authentication required"}

curl -sS -i -X POST http://127.0.0.1:18181/run \
     -H 'authorization: Bearer wrong-token' -d '{}' | head -15
# HTTP/1.1 401 Unauthorized
# www-authenticate: Bearer realm="adk"
#
# {"error":"authentication required"}

curl -sS -w "\nHTTP %{http_code}\n" http://127.0.0.1:18181/health
# {"status":"ok"}
# HTTP 200   (public path still open)

Scenario 3: ADK_API_TOKEN=secret-12345, correct token from client. HTTP 200, downstream invoked.

curl -sS -w "\nHTTP %{http_code}\n" -X POST http://127.0.0.1:18181/run \
     -H 'authorization: Bearer secret-12345' -d '{}'
# {"ok":true}
# HTTP 200

Why this opens against google/adk-python

Issue 509219988 tracks the underlying auth gap report; per the maintainer's 2026-06-04 note ("the more comprehensive solution would be to allow the server to be deployed with an authentication dependency... If you would like to provide a fix yourself it helps speed up our validation process"), this PR is the proposed fix.

This PR supersedes #5980, which was opened from a fork on the wrong GitHub account.

Files changed

• src/google/adk/cli/api_server.py (+ middleware class, + app.add_middleware wiring)
• tests/unittests/cli/test_bearer_auth_middleware.py (new, 9 tests)

[adk-bot]

🔍 ADK Pull Request Analysis: PR #5981

Title: fix(api_server): add optional ADK_API_TOKEN Bearer auth middleware
Author: @f4x0rz
Status: open
Impact: 280 additions, 0 deletions across 2 files

---

🛡️ Critical Compliance Gate: Google CLA Verification

Before analyzing this PR's diff and local workspace file structure, we verified the contributor's Google Contributor License Agreement (CLA) status:

• Result: ✅ SUCCESS
• Verification: The cla/google status check for commit ec2ff226020aff07469b4b275ec3ccdf097ce7ba was successfully completed on GitHub with status COMPLETED and conclusion SUCCESS.
• Safe to proceed with reading files and completing full analysis.

---

Executive Summary

1. Core Objective: Introduces an optional ADK_API_TOKEN environment variable triggering ASGI Bearer token verification to secure the endpoints of the ASGI/FastAPI ADK API Server without breaking existing deployments.
2. Justification & Value: Valuable Feature - Resolves a significant security concern where endpoints wrapping raw code execution run with zero built-in identity verification by default.
3. Alignment with Principles: Pass with Nits - High-quality compliance with Starlette's middleware conventions and ADK testing protocols; minor style and architectural improvements are noted below.
4. Recommendation: Approve with Nits (specifically addressing the raw WebSockets exclusion and security best practices).

---

Detailed Findings & Analysis

1. Objectives & Impact ("What does it do?")

• Context & Background:
  The ADK API server registers critical endpoints such as /run, /run_sse, and the endpoints under /apps/{app}/users/{user}/sessions/... which facilitate execution of agentic processes, potentially utilizing UnsafeLocalCodeExecutor. This PR addresses Issue 509219988 on Google's Issue Tracker, resolving a gap where public endpoints are bound on-network without default auth gates.
• Implementation Mechanism:
  • Creates class _BearerAuthMiddleware within api_server.py implementing standard ASGI middleware contracts.
  • When the ADK_API_TOKEN environment variable is defined and non-empty, incoming HTTP requests (except public endpoints /health and /version) are checked for a matching Authorization: Bearer <token> header. Mismatched or absent credentials trigger an HTTP 401 with standard WWW-Authenticate signaling.
  • Wires the middleware under api_server.py immediately after _DefaultAppRewriteMiddleware via:

    app.add_middleware(
        _BearerAuthMiddleware,
        token=os.environ.get("ADK_API_TOKEN"),
    )

• Affected Surface: Correctly encapsulates the changes internally as private classes and environment integrations. It maintains full backwards compatibility with existing deployments since the middleware remains a no-op if ADK_API_TOKEN is unset or empty.

2. Justification & Value ("Is it a valid and useful change?")

• Workspace Verification:
  Investigated current workspace file api_server.py. We confirmed that the baseline middleware layout currently features CORS, Origin checking, and App rewrite layers, but entirely lacks security mechanisms. Endpoints can execute local python code executions via web requests with no boundary verifying identity.
• Value Assessment: Highly valuable. Security mitigations directly in the library prevent exposing the host environment to arbitrary users when the API server is bound to a network interface without reverse-proxy layers.
• Alternative Approaches: Wrapping inside standard Starlette middleware is the cleanest path for low-latency request interception prior to route handlers.
• Scope & Depth: ** Systematic Fix (for HTTP)** but Point Fix (collectively). While it safely covers all HTTP routes, it does not secure WebSocket routes (as described in the Architecture checklist).

3. Principle & Style Alignment Checklist ("Does it follow rules?")

• Public API & Visibility Boundaries:

  • Status: Pass
  • Analysis: Uses private class layout prefix _BearerAuthMiddleware in api_server.py to avoid exposure in the public API namespaces.

• Code Quality, Typing & Conventions:

  • Status: Nits
  • Analysis:
    • Nit 1: Missing Future Annotations: The new test module tests/unittests/cli/test_bearer_auth_middleware.py is missing the mandatory from __future__ import annotations declaration immediately following the license block.
    • Nit 2: Timing Attack Risk: Standard string comparison auth_header == expected in _BearerAuthMiddleware.__call__ is vulnerable to character-by-character timing attacks over stable connections. Using hmac.compare_digest is best practice for validating authorization tokens.
    • Nit 3: Legacy Types: In the new test file, typing.Optional and typing.List are used. For brand new files, modern constructs str | None and list[...] are preferred where applicable.
    • Nit 4: Non-Alphabetical Imports: import pytest is sorted below local library imports in test_bearer_auth_middleware.py, violating standard alphabetical organization rules.

• Robustness & Edge Cases:

  • Status: Fail / Critical Security Edge-Case
  • Analysis:
    • The middleware checks if scope["type"] != "http" or self._token is None:. This explicitly exempts WebSocket connections (scope["type"] == "websocket") from authorization checks.
    • As verified in api_server.py:L1591, the server registers a critical WebSocket endpoint @app.websocket("/run_live"). Because WebSockets bypass this check, /run_live is completely unauthenticated and accessible even when ADK_API_TOKEN is set, representing an entry gate bypass.

• Test Integrity & Quality:

  • Status: Pass
  • Analysis: Excellent structure. The new suite tests/unittests/cli/test_bearer_auth_middleware.py utilizes real Starlette structures and minimal, focused CollectingApp inline helpers. Every test cleanly separates Arrange-Act-Assert blocks by blank lines, fully conforming to Rule 9 of the ADK Testing Style.

---

🛠️ Requested Updates / Design Fixes

To achieve full alignment, we recommend providing the following feedback to the contributor on the pull request:

1. Authorize WebSocket Scopes:
   Support checking WebSocket connections for the Authorization header. Update the __call__ interface to intercept websocket scope types if the API token is configured:

   if scope["type"] not in ("http", "websocket") or self._token is None:
     await self._app(scope, receive, send)
     return

   (And ensure the respective WebSocket error close frames are handled if unauthorized).
2. Mitigate Timing Attacks:
   Use hmac.compare_digest instead of direct inequality operators:

   import hmac
   
   # ...
   if auth_header is not None and hmac.compare_digest(auth_header, expected):

3. Modernize Type Hints & Annotations:
   Ensure from __future__ import annotations is added immediately under the license header in tests/unittests/cli/test_bearer_auth_middleware.py, and switch to pure types (list, tuple, | None) instead of standard typing imports.