diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index bd69e203f2423..d81aab5a9bd27 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -5615,12 +5615,29 @@ class APIServerAdapter(BasePlatformAdapter): token = auth[7:].strip() if auth.startswith("Bearer ") else "" cfg = load_config() - claims = get_fire_verifier()( + verifier = get_fire_verifier() + verify_kwargs = dict( token=token, expected_audience=cfg_get(cfg, "cron", "chronos", "expected_audience", default=""), jwks_or_key=cfg_get(cfg, "cron", "chronos", "nas_jwks_url", default="") or None, issuer=cfg_get(cfg, "cron", "chronos", "portal_url", default="") or None, ) + try: + if asyncio.iscoroutinefunction(verifier): + claims = await verifier(**verify_kwargs) + else: + # The verifier resolves the NAS signing key from a JWKS URL, + # which is a synchronous HTTP GET on a cache miss (cold client + # or a rotated kid) — keep that blocking I/O off the event loop + # so a slow or rate-limited portal can't stall every other + # adapter sharing this loop. Same hardening the platform HTTP + # event verifier already got. + claims = await asyncio.to_thread(verifier, **verify_kwargs) + except Exception: + # Fail closed: a crashing verifier must never admit a fire — this + # is the only inbound that can trigger remote job execution. + logger.exception("cron fire: verifier crashed; rejecting token") + claims = None if claims is None: logger.warning( "cron fire: rejected invalid token: %s", diff --git a/tests/gateway/test_cron_fire_webhook.py b/tests/gateway/test_cron_fire_webhook.py index fe61cea767f6a..b8f8a3d4cf175 100644 --- a/tests/gateway/test_cron_fire_webhook.py +++ b/tests/gateway/test_cron_fire_webhook.py @@ -123,3 +123,118 @@ async def test_missing_job_id_400(adapter, monkeypatch): assert spy.fired == [] +@pytest.mark.asyncio +async def test_fire_does_not_require_api_server_key(adapter, monkeypatch): + """The fire endpoint must NOT gate on API_SERVER_KEY — auth is the NAS-JWT. + A request with NO API key header but a valid fire token still succeeds.""" + spy = _SpyProvider() + monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", lambda: spy) + monkeypatch.setattr( + "plugins.cron_providers.chronos.verify.get_fire_verifier", + lambda: (lambda **kw: {"purpose": "cron_fire"}), + ) + + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + # Bearer is the FIRE token, not the API_SERVER_KEY "sk-secret". + resp = await cli.post("/api/cron/fire", + headers={"Authorization": "Bearer nas-jwt"}, + json={"job_id": "j9"}) + assert resp.status == 202 + for _ in range(50): + if spy.fired: + break + await asyncio.sleep(0.01) + assert spy.fired == ["j9"] + + +@pytest.mark.asyncio +async def test_sync_verifier_runs_off_the_event_loop(adapter, monkeypatch): + """The verifier resolves the signing key from a JWKS URL — a synchronous + HTTP GET on a cache miss. It must run via asyncio.to_thread, NOT inline on + the event loop, or a slow/rate-limited portal stalls every other adapter + sharing the loop. Proof: the sync verifier executes on a worker thread, not + the loop thread. + """ + loop_thread_id = threading.get_ident() + seen = {} + + def blocking_verifier(**kw): + seen["thread_id"] = threading.get_ident() + return {"purpose": "cron_fire"} + + spy = _SpyProvider() + monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", lambda: spy) + monkeypatch.setattr( + "plugins.cron_providers.chronos.verify.get_fire_verifier", + lambda: blocking_verifier, + ) + + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + resp = await cli.post("/api/cron/fire", + headers={"Authorization": "Bearer good"}, + json={"job_id": "off-loop"}) + assert resp.status == 202 + + # If the verifier had run inline on the loop, its thread id would equal the + # loop thread's; to_thread puts it on a distinct worker thread. + assert seen["thread_id"] != loop_thread_id + + +@pytest.mark.asyncio +async def test_crashing_verifier_fails_closed_401(adapter, monkeypatch): + """A verifier that raises must be treated as a rejection (401), never admit + the fire, and never surface as a 500 — this is the only inbound that can + trigger remote job execution, so it fails closed. + """ + spy = _SpyProvider() + monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", lambda: spy) + + def exploding_verifier(**kw): + raise RuntimeError("JWKS endpoint unreachable") + + monkeypatch.setattr( + "plugins.cron_providers.chronos.verify.get_fire_verifier", + lambda: exploding_verifier, + ) + + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + resp = await cli.post("/api/cron/fire", + headers={"Authorization": "Bearer boom"}, + json={"job_id": "abc123"}) + assert resp.status == 401 + + await asyncio.sleep(0.05) + assert spy.fired == [] + + +@pytest.mark.asyncio +async def test_async_verifier_is_awaited(adapter, monkeypatch): + """A coroutine verifier (a future async escape-hatch) is awaited directly + rather than dispatched to a thread — a valid async verify still fires. + """ + spy = _SpyProvider() + monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", lambda: spy) + + async def async_verifier(**kw): + return {"purpose": "cron_fire", "aud": "agent:x"} + + monkeypatch.setattr( + "plugins.cron_providers.chronos.verify.get_fire_verifier", + lambda: async_verifier, + ) + + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + resp = await cli.post("/api/cron/fire", + headers={"Authorization": "Bearer good"}, + json={"job_id": "async-ok"}) + assert resp.status == 202 + + for _ in range(50): + if spy.fired: + break + await asyncio.sleep(0.01) + assert spy.fired == ["async-ok"]