fix(cli): add optional OIDC verification for Pub/Sub and Eventarc triggers

src/google/adk/cli/api_server.py

@@ -721,6 +721,7 @@ def __init__(
       url_prefix: Optional[str] = None,
       auto_create_session: bool = False,
       trigger_sources: Optional[list[str]] = None,
+      trigger_oidc_audience: Optional[str] = None,
       default_llm_model: Optional[str] = None,
   ):
     self.agent_loader = agent_loader
@@ -741,6 +742,7 @@ def __init__(
     self.url_prefix = url_prefix
     self.auto_create_session = auto_create_session
     self.trigger_sources = trigger_sources
+    self.trigger_oidc_audience = trigger_oidc_audience
     self.default_llm_model = default_llm_model
     self.default_app_name = os.getenv("ADK_DEFAULT_APP_NAME")
 
@@ -1139,7 +1141,11 @@ async def redirect_dev_ui_add_slash():
     if self.trigger_sources:
       from .trigger_routes import TriggerRouter
 
-      trigger_router = TriggerRouter(self, trigger_sources=self.trigger_sources)
+      trigger_router = TriggerRouter(
+          self,
+          trigger_sources=self.trigger_sources,
+          oidc_audience=self.trigger_oidc_audience,
+      )
       trigger_router.register(app)
 
     return app

src/google/adk/cli/fast_api.py

@@ -427,6 +427,7 @@ def get_fast_api_app(
     logo_image_url: str | None = None,
     auto_create_session: bool = False,
     trigger_sources: list[Literal["pubsub", "eventarc"]] | None = None,
+    trigger_oidc_audience: str | None = None,
     default_llm_model: str | None = None,
     gemini_enterprise_app_name: str | None = None,
     express_mode: bool = False,
@@ -478,6 +479,13 @@ def get_fast_api_app(
     trigger_sources: List of trigger sources to enable (e.g. ["pubsub",
       "eventarc"]). When set, registers /trigger/* endpoints for batch and
       event-driven agent invocations. None disables all trigger endpoints.
+    trigger_oidc_audience: When set, every /trigger/* request must carry a
+      Google-signed OIDC bearer token whose audience matches this value.
+      Pub/Sub push subscriptions and Eventarc triggers attach such a token
+      when configured with a service account, so this authenticates the
+      caller as the intended delivery service. When None (the default), the
+      trigger endpoints stay unauthenticated and rely on the deployment
+      platform for access control.
     default_llm_model: Default LLM model to use for the agent.
     gemini_enterprise_app_name: The Gemini Enterprise app name to use for the
       agent.
@@ -586,6 +594,7 @@ def get_fast_api_app(
       url_prefix=url_prefix,
       auto_create_session=auto_create_session,
       trigger_sources=trigger_sources,
+      trigger_oidc_audience=trigger_oidc_audience,
       default_llm_model=default_llm_model,
   )
 

src/google/adk/cli/trigger_routes.py

@@ -42,7 +42,9 @@
 from fastapi import FastAPI
 from fastapi import HTTPException
 from fastapi import Request
+from google.auth.transport import requests as google_auth_requests
 from google.genai import types
+from google.oauth2 import id_token as google_id_token
 from pydantic import BaseModel
 from pydantic import Field
 
@@ -241,8 +243,17 @@ def __init__(
       max_retries: int = DEFAULT_MAX_RETRIES,
       retry_base_delay: float = DEFAULT_RETRY_BASE_DELAY,
       retry_max_delay: float = DEFAULT_RETRY_MAX_DELAY,
+      oidc_audience: Optional[str] = None,
   ):
     self._server = adk_web_server
+    # When set, every /trigger/* request must carry a Google-signed OIDC
+    # bearer token whose audience matches this value. Pub/Sub push
+    # subscriptions and Eventarc triggers attach such a token when configured
+    # with a service account, so verifying it authenticates the caller as the
+    # intended GCP delivery service. When None (the default), the endpoints
+    # stay unauthenticated and rely on the deployment platform for access
+    # control, preserving existing behavior.
+    self._oidc_audience = oidc_audience
     resolved_sources = (
         trigger_sources
         if trigger_sources is not None
@@ -388,6 +399,46 @@ async def _run_agent_with_retry(
         f" {last_error}"
     )
 
+  def _verify_oidc_token(self, request: Request) -> None:
+    """Verifies the request's OIDC bearer token when auth is enabled.
+
+    Pub/Sub push subscriptions and Eventarc triggers, when configured with a
+    service account, attach a Google-signed OIDC token in the
+    ``Authorization: Bearer`` header. When ``oidc_audience`` is set, this
+    verifies that token's signature and audience so an unauthenticated caller
+    cannot invoke an agent run.
+
+    Args:
+      request: The incoming request.
+
+    Raises:
+      HTTPException: 401 if the token is missing, malformed, or fails
+        verification.
+    """
+    if not self._oidc_audience:
+      return
+
+    auth_header = request.headers.get("Authorization", "")
+    scheme, _, token = auth_header.partition(" ")
+    if scheme.lower() != "bearer" or not token.strip():
+      raise HTTPException(
+          status_code=401,
+          detail="Missing or malformed Authorization bearer token.",
+      )
+
+    try:
+      google_id_token.verify_oauth2_token(
+          token.strip(),
+          google_auth_requests.Request(),
+          self._oidc_audience,
+      )
+    except Exception as e:
+      logger.warning("OIDC token verification failed: %s", e)
+      raise HTTPException(
+          status_code=401,
+          detail="OIDC token verification failed.",
+      ) from e
+
   def register(self, app: FastAPI) -> None:
     """Register /trigger/* routes on the FastAPI app.
 
@@ -411,6 +462,7 @@ def register(self, app: FastAPI) -> None:
       async def trigger_pubsub(
           app_name: str, req: PubSubTriggerRequest, request: Request
       ) -> TriggerResponse:
+        self._verify_oidc_token(request)
         subscription = req.subscription or "pubsub-caller"
         user_id = subscription.replace("/", "--")
 
@@ -477,6 +529,7 @@ async def trigger_pubsub(
       async def trigger_eventarc(
           app_name: str, req: EventarcTriggerRequest, request: Request
       ) -> TriggerResponse:
+        self._verify_oidc_token(request)
 
         source = (
             req.source or request.headers.get("ce-source") or "eventarc-caller"

tests/unittests/cli/test_trigger_routes.py

@@ -32,6 +32,7 @@
 from google.adk.agents.base_agent import BaseAgent
 from google.adk.agents.run_config import RunConfig
 from google.adk.cli import fast_api as fast_api_module
+from google.adk.cli import trigger_routes as trigger_routes_module
 from google.adk.cli.fast_api import get_fast_api_app
 from google.adk.cli.trigger_routes import _is_transient_error
 from google.adk.cli.trigger_routes import TransientError
@@ -199,6 +200,7 @@ def _make_test_client(
     mock_memory_service,
     mock_agent_loader,
     trigger_sources: Optional[list[str]] = None,
+    trigger_oidc_audience: Optional[str] = None,
 ) -> TestClient:
   """Build a TestClient with the given trigger setting."""
   with (
@@ -248,6 +250,7 @@ def _make_test_client(
         memory_service_uri="",
         allow_origins=["*"],
         trigger_sources=trigger_sources,
+        trigger_oidc_audience=trigger_oidc_audience,
     )
     return TestClient(app)
 
@@ -286,6 +289,107 @@ def client_no_triggers(
   )
 
 
+@pytest.fixture
+def client_oidc(
+    mock_session_service,
+    mock_artifact_service,
+    mock_memory_service,
+    mock_agent_loader,
+):
+  """TestClient with triggers enabled and OIDC audience verification on."""
+  return _make_test_client(
+      mock_session_service,
+      mock_artifact_service,
+      mock_memory_service,
+      mock_agent_loader,
+      trigger_sources=["pubsub", "eventarc"],
+      trigger_oidc_audience="https://my-service.example.run.app",
+  )
+
+
+class TestTriggerOidcVerification:
+  """Tests for optional OIDC bearer-token verification on trigger routes."""
+
+  _PUBSUB_PAYLOAD = {
+      "message": {
+          "data": base64.b64encode(b"hi").decode("utf-8"),
+          "messageId": "msg-oidc",
+      },
+      "subscription": "projects/p/subscriptions/s",
+  }
+
+  def test_rejects_missing_token(self, client_oidc):
+    resp = client_oidc.post(
+        "/apps/test_app/trigger/pubsub", json=self._PUBSUB_PAYLOAD
+    )
+    assert resp.status_code == 401
+
+  def test_rejects_non_bearer_scheme(self, client_oidc):
+    resp = client_oidc.post(
+        "/apps/test_app/trigger/pubsub",
+        json=self._PUBSUB_PAYLOAD,
+        headers={"Authorization": "Basic abc"},
+    )
+    assert resp.status_code == 401
+
+  def test_rejects_invalid_token(self, client_oidc, monkeypatch):
+    def _fail(*args, **kwargs):
+      raise ValueError("bad token")
+
+    monkeypatch.setattr(
+        trigger_routes_module.google_id_token,
+        "verify_oauth2_token",
+        _fail,
+    )
+    resp = client_oidc.post(
+        "/apps/test_app/trigger/pubsub",
+        json=self._PUBSUB_PAYLOAD,
+        headers={"Authorization": "Bearer forged.jwt.value"},
+    )
+    assert resp.status_code == 401
+
+  def test_accepts_valid_token(self, client_oidc, monkeypatch):
+    captured_audience = []
+
+    def _ok(token, request, audience):
+      captured_audience.append(audience)
+      return {"aud": audience, "email": "svc@project.iam"}
+
+    monkeypatch.setattr(
+        trigger_routes_module.google_id_token,
+        "verify_oauth2_token",
+        _ok,
+    )
+
+    async def dummy_run_async(self, user_id, session_id, new_message, **kwargs):
+      yield _model_event("ok")
+      await asyncio.sleep(0)
+
+    monkeypatch.setattr(Runner, "run_async", dummy_run_async)
+
+    resp = client_oidc.post(
+        "/apps/test_app/trigger/pubsub",
+        json=self._PUBSUB_PAYLOAD,
+        headers={"Authorization": "Bearer good.jwt.value"},
+    )
+    assert resp.status_code == 200
+    assert captured_audience == ["https://my-service.example.run.app"]
+
+  def test_unauthenticated_by_default(self, client, monkeypatch):
+    """With no audience configured, no token is required (existing behavior)."""
+
+    async def dummy_run_async(self, user_id, session_id, new_message, **kwargs):
+      yield _model_event("ok")
+      await asyncio.sleep(0)
+
+    monkeypatch.setattr(Runner, "run_async", dummy_run_async)
+
+    resp = client.post(
+        "/apps/test_app/trigger/pubsub", json=self._PUBSUB_PAYLOAD
+    )
+    assert resp.status_code == 200
+
+
 # ===================================================================
 # /apps/test_app/trigger/pubsub — Pub/Sub Push Subscription
 # ===================================================================