Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion gateway/platforms/api_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -5615,12 +5615,29 @@ async def _handle_cron_fire(self, request: "web.Request") -> "web.Response":
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",
Expand Down
115 changes: 115 additions & 0 deletions tests/gateway/test_cron_fire_webhook.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
Loading