From 1305a690e05950c70a5daf45b2005e6a444c0236 Mon Sep 17 00:00:00 2001 From: aeyeopsdev <275853971+aeyeopsdev@users.noreply.github.com> Date: Sun, 31 May 2026 19:36:32 +0000 Subject: [PATCH] feat(gateway): route platform HTTP event callbacks --- gateway/platforms/api_server.py | 122 ++++++++++++++++++++++ gateway/run.py | 4 +- plugins/platforms/google_chat/adapter.py | 91 ++++++++++++++++ plugins/platforms/google_chat/plugin.yaml | 8 ++ tests/gateway/test_api_server.py | 94 +++++++++++++++++ tests/gateway/test_google_chat.py | 104 +++++++++++++++++- 6 files changed, 420 insertions(+), 3 deletions(-) diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 7a33519ead40..97add25763ba 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -995,6 +995,10 @@ class APIServerAdapter(BasePlatformAdapter): # (the /v1/runs path tracks its own in-flight set via # _active_run_tasks). self._inflight_agent_runs: int = 0 + # Back-reference to the owning GatewayRunner (set by gateway/run.py) + # so /api/platforms/{platform}/events can resolve sibling adapters. + # BasePlatformAdapter declares the class-level default of None. + self.gateway_runner: Optional[Any] = None # Requests admitted before their handler reaches agent bookkeeping. # Shutdown counts this reservation so the request cannot slip through # the drain between its first await and _run_agent()/task registration. @@ -1254,6 +1258,118 @@ class APIServerAdapter(BasePlatformAdapter): status=401, ) + @staticmethod + def _normalize_callback_platform(value: str) -> str: + normalized = (value or "").strip().lower().replace("-", "_") + if not re.fullmatch(r"[a-z0-9_]+", normalized): + return "" + return normalized + + def _get_platform_callback_adapter( + self, + request: "web.Request", + platform_name: str, + ) -> Optional[Any]: + injected = request.app.get("platform_event_adapters") + if isinstance(injected, dict): + adapter = injected.get(platform_name) + if adapter is not None: + return adapter + + adapter = request.app.get(f"{platform_name}_adapter") + if adapter is not None: + return adapter + + runner = self.gateway_runner or request.app.get("gateway_runner") + adapters = getattr(runner, "adapters", None) + if not adapters: + return None + + try: + from gateway.config import Platform as _Platform + return adapters.get(_Platform(platform_name)) + except Exception: + for platform, candidate in adapters.items(): + if getattr(platform, "value", platform) == platform_name: + return candidate + return None + + async def _handle_platform_event_callback(self, request: "web.Request") -> "web.Response": + platform_name = self._normalize_callback_platform( + request.match_info.get("platform", "") + ) + if not platform_name: + return web.json_response( + _openai_error( + "Invalid platform name", + code="invalid_platform", + ), + status=400, + ) + + adapter = self._get_platform_callback_adapter(request, platform_name) + if adapter is None: + return web.json_response( + _openai_error( + "Platform adapter is not connected", + code="platform_unavailable", + ), + status=503, + ) + + verifier = getattr(adapter, "verify_http_event_request", None) + dispatcher = getattr(adapter, "dispatch_http_event", None) + if verifier is None or dispatcher is None: + return web.json_response( + _openai_error( + "Platform adapter does not support HTTP events", + code="platform_http_events_unsupported", + ), + status=503, + ) + + ok, code = verifier(request.headers.get("Authorization", "")) + if not ok: + return web.json_response( + _openai_error( + "Invalid platform event authorization", + code=code or "invalid_platform_event_authorization", + ), + status=401, + ) + + try: + payload = await request.json() + except Exception: + return web.json_response( + _openai_error("Invalid JSON in platform event", code="invalid_json"), + status=400, + ) + + if not isinstance(payload, dict): + return web.json_response( + _openai_error( + "Platform event must be a JSON object", + code="invalid_request", + ), + status=400, + ) + + try: + result = await dispatcher(payload) + except Exception: + logger.exception("Platform HTTP event dispatch failed for %s", platform_name) + return web.json_response( + _openai_error( + "Platform event dispatch failed", + err_type="server_error", + code="platform_event_dispatch_failed", + ), + status=500, + ) + + return web.json_response(result if isinstance(result, dict) else {}) + # ------------------------------------------------------------------ # Multi-profile multiplexing (/p//…) # ------------------------------------------------------------------ @@ -1364,6 +1480,10 @@ class APIServerAdapter(BasePlatformAdapter): ("POST", "/v1/responses", self._handle_responses), ("GET", "/v1/responses/{response_id}", self._handle_get_response), ("DELETE", "/v1/responses/{response_id}", self._handle_delete_response), + # Generic platform HTTP event callback ingress. Authenticated by + # the target adapter's own verifier (platform-signed bearer), NOT + # API_SERVER_KEY — external platforms hold no API server key. + ("POST", "/api/platforms/{platform}/events", self._handle_platform_event_callback), ("GET", "/api/jobs", self._handle_list_jobs), ("POST", "/api/jobs", self._handle_create_job), ("GET", "/api/jobs/{job_id}", self._handle_get_job), @@ -5174,6 +5294,8 @@ class APIServerAdapter(BasePlatformAdapter): # native routes first lets those shims no-op instead of shadowing the # upstream session-control handlers. self._app["api_server_adapter"] = self + if self.gateway_runner is not None: + self._app["gateway_runner"] = self.gateway_runner # Start background sweep to clean up orphaned (unconsumed) run streams sweep_task = asyncio.create_task(self._sweep_orphaned_runs()) diff --git a/gateway/run.py b/gateway/run.py index c970ef93e582..6cf4566e044d 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -9042,7 +9042,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if not check_api_server_requirements(): logger.warning("API Server: aiohttp not installed") return None - return APIServerAdapter(config) + adapter = APIServerAdapter(config) + adapter.gateway_runner = self + return adapter elif platform == Platform.WEBHOOK: from gateway.platforms.webhook import WebhookAdapter, check_webhook_requirements diff --git a/plugins/platforms/google_chat/adapter.py b/plugins/platforms/google_chat/adapter.py index ae8b80a635ff..43958d722914 100644 --- a/plugins/platforms/google_chat/adapter.py +++ b/plugins/platforms/google_chat/adapter.py @@ -73,6 +73,20 @@ MediaFileUpload: Any = None # type: ignore _google_modules_loaded: bool = False +def _verify_google_id_token(token: str, audience: str) -> Dict[str, Any]: + try: + from google.auth.transport import requests as google_requests + from google.oauth2 import id_token + except ImportError as exc: + raise RuntimeError("google-auth is required for Google Chat HTTP callbacks") from exc + + return id_token.verify_oauth2_token( + token, + google_requests.Request(), + audience, + ) + + def _load_google_modules() -> bool: """Lazily import the heavy google-cloud + googleapiclient stack. @@ -690,6 +704,21 @@ class GoogleChatAdapter(BasePlatformAdapter): self._max_bytes = int(os.getenv("GOOGLE_CHAT_MAX_BYTES", str(16 * 1024 * 1024))) except (ValueError, TypeError): self._max_bytes = 16 * 1024 * 1024 + self._http_events_url = ( + self.config.extra.get("http_events_url") + or os.getenv("GOOGLE_CHAT_HTTP_EVENTS_URL", "") + or "" + ).strip() + self._http_events_audience = ( + self.config.extra.get("http_events_audience") + or os.getenv("GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE", "") + or self._http_events_url + ).strip() + self._http_events_service_account_email = ( + self.config.extra.get("http_events_service_account_email") + or os.getenv("GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL", "") + or "" + ).strip().lower() # ------------------------------------------------------------------ # Configuration loading and validation @@ -1422,6 +1451,62 @@ class GoogleChatAdapter(BasePlatformAdapter): except Exception: pass + async def dispatch_http_event(self, envelope: Dict[str, Any]) -> Dict[str, Any]: + extracted = self._extract_message_payload(envelope) + if extracted is None: + return {} + + msg, space, _fmt = extracted + sender = msg.get("sender") or {} + if sender.get("type") == "BOT": + return {} + + msg_name = msg.get("name") or "" + if msg_name and self._dedup.is_duplicate(msg_name): + return {} + + msg_with_space = dict(msg) + if "space" not in msg_with_space and space: + msg_with_space["space"] = space + + enriched_env = dict(envelope) + if "space" not in enriched_env and space: + enriched_env["space"] = space + + await self._dispatch_message(msg_with_space, enriched_env) + return {} + + def verify_http_event_request(self, auth_header: str) -> Tuple[bool, str]: + if not self._http_events_audience or not self._http_events_service_account_email: + return False, "google_chat_http_events_not_configured" + + if not auth_header.startswith("Bearer "): + return False, "missing_google_bearer" + + token = auth_header[7:].strip() + if not token: + return False, "missing_google_bearer" + + try: + claims = _verify_google_id_token(token, self._http_events_audience) + except Exception as exc: + logger.warning( + "[GoogleChat] HTTP event bearer verification failed: %s", + _redact_sensitive(str(exc)), + ) + return False, "invalid_google_bearer" + + expected = { + item.strip().lower() + for item in self._http_events_service_account_email.split(",") + if item.strip() + } + claim_email = str(claims.get("email") or "").strip().lower() + if not claim_email or claim_email not in expected: + return False, "unexpected_google_bearer_identity" + + return True, "" + async def _dispatch_message(self, msg: Dict[str, Any], envelope: Dict[str, Any]) -> None: """Translate a Chat message payload to a MessageEvent and hand off. @@ -3270,6 +3355,12 @@ def _env_enablement() -> Optional[Dict[str, Any]]: seed["subscription_name"] = subscription if http_events_url: seed["http_events_url"] = http_events_url + http_events_audience = os.getenv("GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE") + if http_events_audience: + seed["http_events_audience"] = http_events_audience + http_events_sa_email = os.getenv("GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL") + if http_events_sa_email: + seed["http_events_service_account_email"] = http_events_sa_email sa_json = ( os.getenv("GOOGLE_CHAT_SERVICE_ACCOUNT_JSON") or os.getenv("GOOGLE_APPLICATION_CREDENTIALS") diff --git a/plugins/platforms/google_chat/plugin.yaml b/plugins/platforms/google_chat/plugin.yaml index 6cf99cb8a0a0..d178a53085d4 100644 --- a/plugins/platforms/google_chat/plugin.yaml +++ b/plugins/platforms/google_chat/plugin.yaml @@ -23,6 +23,14 @@ optional_env: description: "Authenticated HTTP endpoint for Chat message events." prompt: "HTTP events callback URL" password: false + - name: GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE + description: "Expected audience for Google-signed HTTP event bearer tokens. Defaults to GOOGLE_CHAT_HTTP_EVENTS_URL." + prompt: "HTTP events token audience" + password: false + - name: GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL + description: "Expected Google service account email for HTTP event bearer tokens." + prompt: "HTTP events service account email" + password: false - name: GOOGLE_CHAT_PROJECT_ID description: "GCP project ID for optional Pub/Sub inbound mode. Falls back to GOOGLE_CLOUD_PROJECT." prompt: "GCP project ID" diff --git a/tests/gateway/test_api_server.py b/tests/gateway/test_api_server.py index dc8ff81a992d..aa48e5323daa 100644 --- a/tests/gateway/test_api_server.py +++ b/tests/gateway/test_api_server.py @@ -652,9 +652,28 @@ def _create_app(adapter: APIServerAdapter) -> web.Application: app.router.add_post("/v1/responses", adapter._handle_responses) app.router.add_get("/v1/responses/{response_id}", adapter._handle_get_response) app.router.add_delete("/v1/responses/{response_id}", adapter._handle_delete_response) + app.router.add_post( + "/api/platforms/{platform}/events", + adapter._handle_platform_event_callback, + ) return app +class _FakeGoogleChatAdapter: + def __init__(self, *, verify_ok: bool = True, verify_code: str = ""): + self.verify_ok = verify_ok + self.verify_code = verify_code + self.dispatched = [] + + def verify_http_event_request(self, auth_header: str): + self.auth_header = auth_header + return self.verify_ok, self.verify_code + + async def dispatch_http_event(self, payload): + self.dispatched.append(payload) + return {"ok": True} + + @pytest.fixture def adapter(): return _make_adapter() @@ -2850,6 +2869,81 @@ class TestSendMethod: assert "HTTP request/response" in result.error +class TestPlatformEventCallbackEndpoint: + @pytest.mark.asyncio + async def test_dispatches_authorized_google_chat_event(self, adapter): + app = _create_app(adapter) + google_adapter = _FakeGoogleChatAdapter() + app["platform_event_adapters"] = {"google_chat": google_adapter} + + async with TestClient(TestServer(app)) as cli: + resp = await cli.post( + "/api/platforms/google_chat/events", + headers={"Authorization": "Bearer google-token"}, + json={"type": "MESSAGE", "message": {"text": "hi"}}, + ) + body = await resp.json() + + assert resp.status == 200 + assert body == {"ok": True} + assert google_adapter.auth_header == "Bearer google-token" + assert google_adapter.dispatched == [ + {"type": "MESSAGE", "message": {"text": "hi"}} + ] + + @pytest.mark.asyncio + async def test_rejects_invalid_google_chat_auth(self, adapter): + app = _create_app(adapter) + app["platform_event_adapters"] = { + "google_chat": _FakeGoogleChatAdapter( + verify_ok=False, + verify_code="invalid_google_bearer", + ) + } + + async with TestClient(TestServer(app)) as cli: + resp = await cli.post( + "/api/platforms/google_chat/events", + headers={"Authorization": "Bearer bad"}, + json={"type": "MESSAGE"}, + ) + body = await resp.json() + + assert resp.status == 401 + assert body["error"]["code"] == "invalid_google_bearer" + + @pytest.mark.asyncio + async def test_requires_connected_google_chat_adapter(self, adapter): + app = _create_app(adapter) + + async with TestClient(TestServer(app)) as cli: + resp = await cli.post( + "/api/platforms/google_chat/events", + headers={"Authorization": "Bearer google-token"}, + json={"type": "MESSAGE"}, + ) + body = await resp.json() + + assert resp.status == 503 + assert body["error"]["code"] == "platform_unavailable" + + @pytest.mark.asyncio + async def test_rejects_malformed_platform_event_json(self, adapter): + app = _create_app(adapter) + app["platform_event_adapters"] = {"google_chat": _FakeGoogleChatAdapter()} + + async with TestClient(TestServer(app)) as cli: + resp = await cli.post( + "/api/platforms/google_chat/events", + headers={"Authorization": "Bearer google-token"}, + data="{", + ) + body = await resp.json() + + assert resp.status == 400 + assert body["error"]["code"] == "invalid_json" + + # --------------------------------------------------------------------------- # GET /v1/responses/{response_id} # --------------------------------------------------------------------------- diff --git a/tests/gateway/test_google_chat.py b/tests/gateway/test_google_chat.py index 2f739b51fff3..14d8726146f7 100644 --- a/tests/gateway/test_google_chat.py +++ b/tests/gateway/test_google_chat.py @@ -254,6 +254,8 @@ class TestEnvConfigLoading: "GOOGLE_CHAT_SUBSCRIPTION_NAME", "GOOGLE_CHAT_SUBSCRIPTION", "GOOGLE_CHAT_HTTP_EVENTS_URL", + "GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE", + "GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL", "GOOGLE_CHAT_SERVICE_ACCOUNT_JSON", "GOOGLE_APPLICATION_CREDENTIALS", "GOOGLE_CHAT_HOME_CHANNEL", @@ -284,10 +286,32 @@ class TestEnvConfigLoading: def test_http_events_enable_without_pubsub(self, monkeypatch): self._clean_env(monkeypatch) - monkeypatch.setenv("GOOGLE_CHAT_HTTP_EVENTS_URL", "https://example.test/google-chat/events") + monkeypatch.setenv( + "GOOGLE_CHAT_HTTP_EVENTS_URL", + "https://example.test/google-chat/events", + ) + monkeypatch.setenv( + "GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE", + "https://callback.example.test/events", + ) + monkeypatch.setenv( + "GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL", + "chat-callback@example.iam.gserviceaccount.com", + ) cfg = load_gateway_config() assert _GC in cfg.platforms - assert cfg.platforms[_GC].extra["http_events_url"] == "https://example.test/google-chat/events" + assert ( + cfg.platforms[_GC].extra["http_events_url"] + == "https://example.test/google-chat/events" + ) + assert ( + cfg.platforms[_GC].extra["http_events_audience"] + == "https://callback.example.test/events" + ) + assert ( + cfg.platforms[_GC].extra["http_events_service_account_email"] + == "chat-callback@example.iam.gserviceaccount.com" + ) # =========================================================================== @@ -414,6 +438,82 @@ class TestValidateConfig: assert sub == "projects/inferred/subscriptions/sub" +class TestHttpEventIngress: + def test_verify_http_event_request_accepts_expected_google_identity(self, monkeypatch): + cfg = PlatformConfig(enabled=True) + cfg.extra.update( + { + "http_events_url": "https://example.test/google-chat/events", + "http_events_service_account_email": ( + "chat-callback@example.iam.gserviceaccount.com" + ), + } + ) + a = GoogleChatAdapter(cfg) + + def fake_verify(token, audience): + assert token == "signed-token" + assert audience == "https://example.test/google-chat/events" + return {"email": "chat-callback@example.iam.gserviceaccount.com"} + + monkeypatch.setattr(_gc_mod, "_verify_google_id_token", fake_verify) + + assert a.verify_http_event_request("Bearer signed-token") == (True, "") + + def test_verify_http_event_request_rejects_unexpected_identity(self, monkeypatch): + cfg = PlatformConfig(enabled=True) + cfg.extra.update( + { + "http_events_url": "https://example.test/google-chat/events", + "http_events_service_account_email": ( + "expected@example.iam.gserviceaccount.com" + ), + } + ) + a = GoogleChatAdapter(cfg) + monkeypatch.setattr( + _gc_mod, + "_verify_google_id_token", + lambda _token, _audience: { + "email": "other@example.iam.gserviceaccount.com" + }, + ) + + ok, code = a.verify_http_event_request("Bearer signed-token") + + assert ok is False + assert code == "unexpected_google_bearer_identity" + + def test_verify_http_event_request_requires_callback_identity_config(self): + cfg = PlatformConfig(enabled=True) + cfg.extra["http_events_url"] = "https://example.test/google-chat/events" + a = GoogleChatAdapter(cfg) + + assert a.verify_http_event_request("Bearer signed-token") == ( + False, + "google_chat_http_events_not_configured", + ) + + @pytest.mark.asyncio + async def test_dispatch_http_event_routes_message_payload(self, adapter): + envelope = _make_chat_envelope(text="hello from http") + + result = await adapter.dispatch_http_event(envelope) + + assert result == {} + adapter.handle_message.assert_awaited_once() + event = adapter.handle_message.await_args.args[0] + assert event.text == "hello from http" + assert event.source.chat_id == "spaces/S" + + @pytest.mark.asyncio + async def test_dispatch_http_event_ignores_bot_messages(self, adapter): + envelope = _make_chat_envelope(text="bot echo", sender_type="BOT") + + assert await adapter.dispatch_http_event(envelope) == {} + adapter.handle_message.assert_not_awaited() + + class TestConnectModes: @pytest.mark.asyncio async def test_connect_http_mode_skips_pubsub_subscriber(self, tmp_path, monkeypatch):