From db5e770bc0074036d04374a8f5760e0742151e60 Mon Sep 17 00:00:00 2001 From: Oleksii Dolhov Date: Thu, 7 May 2026 11:38:18 +0300 Subject: [PATCH 1/2] fix(security): close TOCTOU race in webhook rate limiter (#644) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pre-fix path issued a separate GET then INCR. N concurrent callers could all observe count < WEBHOOK_RATE_LIMIT before any of them incremented, slipping past the 429 and pushing the actual call rate to limit + N. Switched to INCR-then-compare (Redis INCR is atomic): increment unconditionally, then 429 the caller whose post-increment count crosses the threshold. Trade-off: blocked requests still tick the counter, slightly extending cool-down for an over-limit token. Acceptable for a rate-limiter — we only stop accepting work, we don't unwind. Tests: - tests/unit/test_webhook_rate_limit_toctou.py — pins INCR-first semantics. The structural assertion (r.get() not called) reliably catches a partial revert that re-adds the GET; the wide-window race belongs in integration tests against real Redis. - tests/integration/test_webhook_rate_limit.py — adds concurrent burst test alongside the existing #589 sequential coverage. Verified live in trinity-backend with real Redis: - pre-fix: 15/20 succeeded under 20-thread burst (limit 10) — race reproduced. - post-fix: exactly 10/20 succeeded — limit holds. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/backend/routers/webhooks.py | 19 +- tests/integration/test_webhook_rate_limit.py | 105 ++++++- tests/unit/test_webhook_rate_limit_toctou.py | 292 +++++++++++++++++++ 3 files changed, 398 insertions(+), 18 deletions(-) create mode 100644 tests/unit/test_webhook_rate_limit_toctou.py diff --git a/src/backend/routers/webhooks.py b/src/backend/routers/webhooks.py index 12b53d315..d83b396ce 100644 --- a/src/backend/routers/webhooks.py +++ b/src/backend/routers/webhooks.py @@ -207,18 +207,25 @@ def _check_webhook_rate_limit(token: str) -> None: key = f"webhook_calls:{token}" try: - count = r.get(key) - if count and int(count) >= WEBHOOK_RATE_LIMIT: + # INCR-then-compare avoids the read-then-incr TOCTOU race (#644): + # under concurrency, separate GET + INCR round-trips let N callers + # all observe `count < limit` and all increment, exceeding the limit + # by N. INCR is atomic in Redis, so we increment unconditionally and + # 429 the caller whose post-increment count crosses the threshold. + # Trade-off: blocked requests still tick the counter, slightly + # extending the cool-down for an already-over-limit token. Acceptable + # for a rate-limiter (we only stop accepting work, we don't unwind). + pipe = r.pipeline() + pipe.incr(key) + pipe.expire(key, WEBHOOK_RATE_WINDOW) + new_count, _ = pipe.execute() + if int(new_count) > WEBHOOK_RATE_LIMIT: ttl = r.ttl(key) raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail=f"Webhook rate limit exceeded. Try again in {ttl} seconds.", headers={"Retry-After": str(max(ttl, 1))}, ) - pipe = r.pipeline() - pipe.incr(key) - pipe.expire(key, WEBHOOK_RATE_WINDOW) - pipe.execute() except HTTPException: raise except Exception as e: diff --git a/tests/integration/test_webhook_rate_limit.py b/tests/integration/test_webhook_rate_limit.py index 489de189f..313b3ef24 100644 --- a/tests/integration/test_webhook_rate_limit.py +++ b/tests/integration/test_webhook_rate_limit.py @@ -1,22 +1,27 @@ -"""Issue #589 regression test for webhooks.py Redis client switch. +"""Webhook rate-limit regression tests. -The fix in src/backend/routers/webhooks.py replaced - redis.Redis(host="redis", port=6379) -with - redis.from_url(REDIS_URL) -so the credentials embedded in REDIS_URL are actually used. Without this -test, a regression would silently fail-open and rate limiting would be -disabled. +Two scenarios covered against the live stack: -Self-contained: creates an agent + schedule + webhook token inline so a -fresh token is used (no pre-existing rate-limit state). +1. **Sequential** (#589) — verifies the Redis client uses the credentialed + `REDIS_URL` so rate limiting is actually engaged (the historic regression + was a silent fail-open on bad auth). +2. **Concurrent** (#644) — verifies the limiter is TOCTOU-safe: firing + `WEBHOOK_RATE_LIMIT + 5` simultaneous requests must not let more than + `WEBHOOK_RATE_LIMIT` slip through. The previous read-then-INCR path + allowed each concurrent caller to observe `count < limit` before any + of them incremented — so the actual call rate exceeded the budget by + the concurrency factor. -Marked `integration` (not `smoke`) because it needs the full stack — +Both tests build their own agent + schedule + webhook token so they don't +share rate-limit state with each other. + +Marked `integration` (not `smoke`) because they need the full stack — backend + Redis with auth + scheduler service. The smoke runner targets -~30s and excludes Docker-dependent tests; this goes through +~30s and excludes Docker-dependent tests; these go through tests/run-integration.sh. """ +import asyncio import uuid import httpx @@ -77,3 +82,79 @@ def test_webhook_rate_limit_returns_429_after_threshold(api_client: TrinityApiCl ) finally: api_client.delete(f"/api/agents/{agent_name}") + + +@pytest.mark.integration +def test_webhook_rate_limit_holds_under_concurrency(api_client: TrinityApiClient): + """Concurrent regression for #644. + + Fire `WEBHOOK_RATE_LIMIT + 5` requests simultaneously. The pre-fix + read-then-INCR path could let all N callers observe `count < limit` + before any incremented, exceeding the limit by N. After the + INCR-then-compare fix, at most `WEBHOOK_RATE_LIMIT` calls get a + non-429 response. + """ + agent_name = f"test-644-webhook-{uuid.uuid4().hex[:8]}" + + create_resp = api_client.post("/api/agents", json={"name": agent_name}) + if create_resp.status_code not in (200, 201): + pytest.skip(f"Cannot create test agent: {create_resp.text}") + + try: + sched_resp = api_client.post( + f"/api/agents/{agent_name}/schedules", + json={ + "name": f"wh-{uuid.uuid4().hex[:6]}", + "cron_expression": "0 0 1 1 *", # never fires during tests + "message": "noop", + "enabled": True, + "timezone": "UTC", + }, + ) + assert sched_resp.status_code == 201, sched_resp.text + sid = sched_resp.json()["id"] + + gen_resp = api_client.post( + f"/api/agents/{agent_name}/schedules/{sid}/webhook" + ) + assert gen_resp.status_code == 200, gen_resp.text + webhook_url = gen_resp.json()["webhook_url"] + token = webhook_url.split("/api/webhooks/")[1] + + url = f"http://localhost:8000/api/webhooks/{token}" + n_concurrent = WEBHOOK_RATE_LIMIT + 5 + + async def fire_all(): + async with httpx.AsyncClient(timeout=10.0) as client: + tasks = [client.post(url) for _ in range(n_concurrent)] + return await asyncio.gather(*tasks, return_exceptions=True) + + results = asyncio.run(fire_all()) + + statuses = [] + for r in results: + if isinstance(r, Exception): + pytest.fail(f"Concurrent webhook call raised: {r!r}") + statuses.append(r.status_code) + + accepted = sum(1 for s in statuses if s in (202, 503)) + rate_limited = sum(1 for s in statuses if s == 429) + other = [s for s in statuses if s not in (202, 503, 429)] + + assert not other, f"Unexpected statuses: {other} (full set: {statuses})" + assert accepted <= WEBHOOK_RATE_LIMIT, ( + f"{accepted} calls succeeded under {n_concurrent}-way concurrency, " + f"limit is {WEBHOOK_RATE_LIMIT}. TOCTOU race regressed: {statuses}" + ) + # Sanity: at least some made it through (otherwise the test isn't + # exercising the limiter — e.g., backend down). + assert accepted >= 1, ( + f"No requests accepted ({statuses}) — limiter or backend broken" + ) + # And at least one was rate-limited (proves the limiter ran). + assert rate_limited >= 1, ( + f"No 429 in {n_concurrent}-way burst with limit {WEBHOOK_RATE_LIMIT} — " + f"limiter not engaging: {statuses}" + ) + finally: + api_client.delete(f"/api/agents/{agent_name}") diff --git a/tests/unit/test_webhook_rate_limit_toctou.py b/tests/unit/test_webhook_rate_limit_toctou.py new file mode 100644 index 000000000..00ae4cc4a --- /dev/null +++ b/tests/unit/test_webhook_rate_limit_toctou.py @@ -0,0 +1,292 @@ +"""TOCTOU regression for webhook rate limiter (#644). + +Pre-fix code path was: + + count = r.get(key) + if count and int(count) >= WEBHOOK_RATE_LIMIT: + raise 429 + pipe.incr(key); pipe.expire(...); pipe.execute() + +Concurrent callers could all observe ``count < limit`` between the GET and +the INCR, all skip the 429, and all increment — letting the actual call +rate exceed ``WEBHOOK_RATE_LIMIT`` by the concurrency factor. + +Fix: INCR-then-compare. Increment unconditionally (Redis INCR is atomic), +then 429 the caller whose post-increment count crosses the threshold. + +These tests pin three properties of the new implementation: + +1. Sequential calls past the limit raise 429 — basic behavior. +2. Concurrent calls under the post-fix code don't over-shoot the limit + (sanity check; in-process timing can mask the pre-fix race). +3. ``r.get()`` is no longer on the hot path — INCR-then-compare. This is + the reliable structural regression signal: a partial revert that + re-adds the GET trips this even when timing-based tests don't. + +The wide-window concurrency repro against a real Redis lives in +``tests/integration/test_webhook_rate_limit.py`` (run via run-integration.sh). +""" + +from __future__ import annotations + +import importlib.util +import os +import sys +import threading +import types +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +from fastapi import HTTPException + +# Issue #589: backend/config.py raises at import if REDIS_URL lacks credentials. +os.environ.setdefault("REDIS_URL", "redis://test:test@redis:6379") +os.environ.setdefault("REDIS_PASSWORD", "test") +os.environ.setdefault("REDIS_BACKEND_PASSWORD", "test") + + +pytestmark = pytest.mark.unit + + +def _find_backend_root() -> Path: + """Locate the backend source tree across host and in-container layouts.""" + candidates = [ + Path(__file__).resolve().parent.parent.parent / "src" / "backend", # host + Path("/app"), # trinity-backend container + ] + env_override = os.environ.get("TRINITY_BACKEND_PATH") + if env_override: + candidates.insert(0, Path(env_override)) + for c in candidates: + if (c / "routers" / "webhooks.py").exists(): + return c + raise RuntimeError( + "Cannot locate backend source tree (set TRINITY_BACKEND_PATH)" + ) + + +_BACKEND = _find_backend_root() +_WEBHOOKS_PY = _BACKEND / "routers" / "webhooks.py" + + +def _stub_module(name: str, **attrs) -> types.ModuleType: + mod = types.ModuleType(name) + for k, v in attrs.items(): + setattr(mod, k, v) + return mod + + +@pytest.fixture +def webhooks(monkeypatch): + """Load webhooks.py with the same dep stubs as the in-process unit tests.""" + db_stub = types.SimpleNamespace(get_schedule_by_webhook_token=lambda _t: None) + monkeypatch.setitem(sys.modules, "database", _stub_module("database", db=db_stub)) + + class _AuditEventType: + EXECUTION = "execution" + + class _PlatformAudit: + async def log(self, **_kw): + return None + + services_pkg = _stub_module("services") + audit_stub = _stub_module( + "services.platform_audit_service", + AuditEventType=_AuditEventType, + platform_audit_service=_PlatformAudit(), + ) + monkeypatch.setitem(sys.modules, "services", services_pkg) + monkeypatch.setitem(sys.modules, "services.platform_audit_service", audit_stub) + + backend_str = str(_BACKEND) + if backend_str not in sys.path: + sys.path.insert(0, backend_str) + + spec = importlib.util.spec_from_file_location( + "webhooks_toctou_under_test", str(_WEBHOOKS_PY) + ) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +# ── Atomic-counter Redis stub ──────────────────────────────────────────────── +# +# Real Redis INCR is atomic; the stub mirrors that with a Lock so concurrent +# threads can't observe a stale read. The pre-fix code's GET path is an +# explicit get() method — kept on the stub to assert it isn't called by the +# fixed implementation. + +class _AtomicRedisStub: + def __init__(self): + self._lock = threading.Lock() + self._counts: dict[str, int] = {} + self._ttls: dict[str, int] = {} + self.get_calls = 0 + self.incr_calls = 0 + self.expire_calls = 0 + + # GET should not be called by the post-fix implementation. + def get(self, key): + self.get_calls += 1 + with self._lock: + v = self._counts.get(key) + return None if v is None else str(v).encode() + + def ttl(self, key): + return self._ttls.get(key, -1) + + def pipeline(self): + return _PipelineStub(self) + + +class _PipelineStub: + def __init__(self, parent: _AtomicRedisStub): + self._parent = parent + self._ops: list = [] + + def incr(self, key): + self._ops.append(("incr", key)) + return self + + def expire(self, key, ttl): + self._ops.append(("expire", key, ttl)) + return self + + def execute(self): + results = [] + with self._parent._lock: + for op in self._ops: + if op[0] == "incr": + self._parent.incr_calls += 1 + self._parent._counts[op[1]] = self._parent._counts.get(op[1], 0) + 1 + results.append(self._parent._counts[op[1]]) + elif op[0] == "expire": + self._parent.expire_calls += 1 + self._parent._ttls[op[1]] = op[2] + results.append(True) + self._ops.clear() + return results + + +# ── Tests ──────────────────────────────────────────────────────────────────── + +class TestSequentialBoundary: + """Basic post-fix behavior — N+1 raises 429.""" + + def test_first_n_succeed_then_429(self, webhooks, monkeypatch): + stub = _AtomicRedisStub() + monkeypatch.setattr(webhooks, "_get_redis", lambda: stub) + + token = "boundary-token" + limit = webhooks.WEBHOOK_RATE_LIMIT + + for _ in range(limit): + webhooks._check_webhook_rate_limit(token) + + with pytest.raises(HTTPException) as excinfo: + webhooks._check_webhook_rate_limit(token) + assert excinfo.value.status_code == 429 + assert "Retry-After" in excinfo.value.headers + + +class TestConcurrentLimit: + """Property #2 — the limit must hold under thread concurrency. + + Sanity check that the post-fix INCR-then-compare path doesn't + over-shoot the limit when called from multiple threads against an + atomic-counter stub. This is timing-dependent — the GIL + in-process + Lock can mask the pre-fix race in unit tests, so this is a happy-path + pin only. The reliable structural regression signal lives in + `TestIncrFirstSemantics` below; the wide-window race is exercised + against a real Redis in `tests/integration/test_webhook_rate_limit.py`. + """ + + @pytest.mark.parametrize("concurrency", [16, 32]) + def test_no_more_than_limit_succeed(self, webhooks, monkeypatch, concurrency): + stub = _AtomicRedisStub() + monkeypatch.setattr(webhooks, "_get_redis", lambda: stub) + + token = f"concurrent-{concurrency}" + limit = webhooks.WEBHOOK_RATE_LIMIT + n = limit + concurrency + + def call(): + try: + webhooks._check_webhook_rate_limit(token) + return "ok" + except HTTPException as e: + return e.status_code + + with ThreadPoolExecutor(max_workers=concurrency) as ex: + results = list(ex.map(lambda _: call(), range(n))) + + accepted = sum(1 for r in results if r == "ok") + rate_limited = sum(1 for r in results if r == 429) + + assert accepted == limit, ( + f"expected exactly {limit} accepted under {concurrency}-way " + f"concurrency, got {accepted}. Counts: ok={accepted} 429={rate_limited}. " + f"Pre-fix TOCTOU race regressed." + ) + assert rate_limited == n - limit + + +class TestIncrFirstSemantics: + """Property #3 — INCR is on the hot path, GET is not. + + A partial revert (re-introducing the GET-then-INCR sequence) would + flip this assertion even if the conditional was kept correct, because + real-Redis concurrency would still be racy regardless of whether the + in-process test catches it. + """ + + def test_incr_called_get_not_called(self, webhooks, monkeypatch): + stub = _AtomicRedisStub() + monkeypatch.setattr(webhooks, "_get_redis", lambda: stub) + + token = "incr-first" + for _ in range(3): + webhooks._check_webhook_rate_limit(token) + + assert stub.incr_calls == 3, ( + f"expected 3 INCRs, got {stub.incr_calls} — INCR must run on every call" + ) + assert stub.get_calls == 0, ( + f"r.get() called {stub.get_calls} times — pre-fix code path is back. " + "INCR-then-compare must not GET first." + ) + + def test_pipeline_order_is_incr_then_expire(self, webhooks, monkeypatch): + """Pipeline must batch the INCR + EXPIRE so TTL is set even on first call.""" + # Use a MagicMock-shaped pipeline to capture call order. + recorder = MagicMock() + recorder.execute = MagicMock(return_value=[1, True]) + + class _RecordingRedis: + def __init__(self): + self.pipeline_calls = 0 + + def pipeline(self): + self.pipeline_calls += 1 + return recorder + + def ttl(self, _key): + return -1 + + def get(self, *_a, **_kw): + pytest.fail("r.get() must not be called by the post-fix code") + + rec = _RecordingRedis() + monkeypatch.setattr(webhooks, "_get_redis", lambda: rec) + + webhooks._check_webhook_rate_limit("ordered-token") + + # First two recorded ops on the pipeline are incr then expire. + method_names = [c[0] for c in recorder.method_calls] + # Drop the trailing execute() call from comparison. + ops_before_execute = [m for m in method_names if m != "execute"] + assert ops_before_execute[:2] == ["incr", "expire"] From afed75a3fde1e97bbe623d78acef7ca2815a5678 Mon Sep 17 00:00:00 2001 From: Oleksii Dolhov Date: Thu, 7 May 2026 11:51:11 +0300 Subject: [PATCH 2/2] =?UTF-8?q?fix(webhooks):=20unblock=20trigger=20endpoi?= =?UTF-8?q?nt=20=E2=80=94=20schedule=20model=20+=20audit=20signature=20(#6?= =?UTF-8?q?47=20follow-up)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit While verifying #644 against a live stack, found two additional facade gaps that #648 (the WEBHOOK-001 delegation fix) didn't catch — both crash trigger_webhook before the rate-limiter even runs to completion: 1. `Schedule` pydantic model never carried `webhook_enabled` / `webhook_token` fields. The DB columns exist, but the row mapper discarded them, so `if not schedule.webhook_enabled:` raised AttributeError on every trigger call. 2. `webhooks.py:trigger_webhook` called `platform_audit_service.log()` with `actor_type="system"`. The service derives actor_type internally from actor_user / actor_agent_name / mcp_scope and has no such kwarg; every accepted webhook 500'd in the audit step. Both are tiny: - Add the two fields to `Schedule` (db_models.py). - Pull them through `_row_to_schedule` (db/schedules.py). - Drop the bogus actor_type kwarg, pass actor_ip instead — webhook callers are unauthenticated so caller IP is the only attributable signal. With these, the integration test in tests/integration/test_webhook_rate_limit.py now exercises the full HTTP path end-to-end. Live verification against the running backend: 15-way concurrent burst → 10 × 202 + 5 × 429, limit holds. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/backend/db/schedules.py | 5 ++++- src/backend/db_models.py | 5 +++++ src/backend/routers/webhooks.py | 7 +++++-- 3 files changed, 14 insertions(+), 3 deletions(-) diff --git a/src/backend/db/schedules.py b/src/backend/db/schedules.py index c3c0c82b6..a56de12c9 100644 --- a/src/backend/db/schedules.py +++ b/src/backend/db/schedules.py @@ -101,7 +101,10 @@ def _row_to_schedule(row) -> Schedule: # Validation configuration (VALIDATE-001) validation_enabled=bool(row["validation_enabled"]) if "validation_enabled" in row_keys and row["validation_enabled"] is not None else False, validation_prompt=row["validation_prompt"] if "validation_prompt" in row_keys else None, - validation_timeout_seconds=row["validation_timeout_seconds"] if "validation_timeout_seconds" in row_keys and row["validation_timeout_seconds"] is not None else 120 + validation_timeout_seconds=row["validation_timeout_seconds"] if "validation_timeout_seconds" in row_keys and row["validation_timeout_seconds"] is not None else 120, + # Webhook trigger (WEBHOOK-001 / #647 follow-up) + webhook_enabled=bool(row["webhook_enabled"]) if "webhook_enabled" in row_keys and row["webhook_enabled"] is not None else False, + webhook_token=row["webhook_token"] if "webhook_token" in row_keys else None, ) @staticmethod diff --git a/src/backend/db_models.py b/src/backend/db_models.py index 5e072bd31..5f2968ffb 100644 --- a/src/backend/db_models.py +++ b/src/backend/db_models.py @@ -153,6 +153,11 @@ class Schedule(BaseModel): validation_enabled: bool = False # Enable post-execution validation validation_prompt: Optional[str] = None # Custom auditor instructions (None = default prompt) validation_timeout_seconds: int = 120 # Timeout for validation task (30-600 range) + # Webhook trigger (WEBHOOK-001 / #647 follow-up): the DB column exists and + # is read by `webhooks.py:trigger_webhook`, but the pydantic model never + # carried these fields — every webhook trigger raised AttributeError. + webhook_enabled: bool = False + webhook_token: Optional[str] = None class ScheduleExecution(BaseModel): diff --git a/src/backend/routers/webhooks.py b/src/backend/routers/webhooks.py index d83b396ce..71e18085a 100644 --- a/src/backend/routers/webhooks.py +++ b/src/backend/routers/webhooks.py @@ -329,12 +329,15 @@ async def trigger_webhook( detail="Scheduler service unavailable — try again later", ) - # Audit trail (SEC-001) + # Audit trail (SEC-001). Webhook callers are unauthenticated — the URL + # token IS the credential — so no actor_user / actor_agent_name. The + # service derives actor_type internally; passing it explicitly is a + # TypeError (#647 follow-up). Caller IP is the only attributable signal. await platform_audit_service.log( event_type=AuditEventType.EXECUTION, event_action="task_triggered", source="api", - actor_type="system", + actor_ip=caller_ip, target_type="agent", target_id=schedule.agent_name, endpoint=f"/api/webhooks/{webhook_token[:8]}…",