fix(security): close TOCTOU race in webhook rate limiter (#644)

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

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):

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:
@@ -322,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]}…",

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}")