fix(api): rate-limit states/leases/claims/mcp coordination routes (#357)

duyet · duyetbot · web-flow · commit f25b2dcd3d34 · 2026-07-17T10:40:30.000+07:00
* fix(api): rate-limit the states router Every write-heavy states route (watch, query, lease, events, PUT, GET, DELETE) attached scopedAuth per-route but never chained rateLimitMiddleware, so an authenticated key could drive unlimited D1 writes and Durable Object notify traffic. Add the limiter after auth on every route, matching the convention used by conversations/tags/etc. Refs #340 Co-Authored-By: Duyet Le <me@duyet.net> Co-Authored-By: duyetbot <bot@duyet.net> * fix(api): rate-limit the leases router POST /:id/renew and DELETE /:id had scopedAuth but no rate limiter, letting a single key drive unbounded lease renew/release traffic. Refs #340 Co-Authored-By: Duyet Le <me@duyet.net> Co-Authored-By: duyetbot <bot@duyet.net> * fix(api): rate-limit the claims router router.use("*", scopedAuth(...)) covered auth for all claim routes but no limiter followed it; chain rateLimitMiddleware right after so every claim write/read goes through the same per-key limit as other routers. Refs #340 Co-Authored-By: Duyet Le <me@duyet.net> Co-Authored-By: duyetbot <bot@duyet.net> * fix(api): rate-limit the MCP endpoint router.use("*", mcpAuth) set apiKeyHash for both API keys and capability/OAuth tokens but nothing enforced a request limit, so a leaked capability token could drive unlimited tool-call traffic. Chain rateLimitMiddleware right after mcpAuth. Refs #340, #349, #350 Co-Authored-By: Duyet Le <me@duyet.net> Co-Authored-By: duyetbot <bot@duyet.net> * test(api): cover 429s on states/leases/claims/mcp routes Seeds the rate_limits table directly at the configured limit (avoiding 10k real requests per assertion) and verifies PUT /states, POST /leases/:id/renew, POST /claims, and MCP tools/call all return the standard RATE_LIMITED envelope once exceeded, plus a states PUT still succeeds under the limit. Refs #340, #349, #350 Co-Authored-By: Duyet Le <me@duyet.net> Co-Authored-By: duyetbot <bot@duyet.net> --------- Co-authored-by: duyetbot <bot@duyet.net>
diff --git a/packages/api/src/routes/claims/index.ts b/packages/api/src/routes/claims/index.ts
@@ -6,6 +6,7 @@ import {
   parseLimitParam,
   parseOrderParam,
 } from "../../lib/helpers";
+import { rateLimitMiddleware } from "../../middleware/rate-limit";
 import { scopedAuth } from "../../middleware/scoped-auth";
 import {
   type CreateEvidenceInput,
@@ -84,6 +85,7 @@ const CreateClaimSchema = z.object({
 const router = new Hono<{ Bindings: Bindings; Variables: Variables }>();
 
 router.use("*", scopedAuth({ scope: "claim:write" }));
+router.use("*", rateLimitMiddleware);
 
 router.post("/", async (c) => {
   const { data, error } = await parseAndValidateBody(c, CreateClaimSchema);
diff --git a/packages/api/src/routes/leases/index.ts b/packages/api/src/routes/leases/index.ts
@@ -1,6 +1,7 @@
 import { Hono } from "hono";
 import { errorResponse, parseAndValidateBody } from "../../lib/helpers";
 import { RenewLeaseSchema } from "../../lib/validation";
+import { rateLimitMiddleware } from "../../middleware/rate-limit";
 import { scopedAuth } from "../../middleware/scoped-auth";
 import { releaseLease, renewLease } from "../../services/leases";
 import type { Bindings, Variables } from "../../types";
@@ -13,7 +14,7 @@ const router = new Hono<{ Bindings: Bindings; Variables: Variables }>();
 // of being duplicated here. This router only operates on an existing lease
 // by id (renew/release).
 
-router.post("/:id/renew", scopedAuth({ scope: "lease:write" }), async (c) => {
+router.post("/:id/renew", scopedAuth({ scope: "lease:write" }), rateLimitMiddleware, async (c) => {
   const { data, error } = await parseAndValidateBody(c, RenewLeaseSchema);
   if (error) return error;
   if (!data) return errorResponse(c, "BAD_REQUEST", "Invalid request body", 400);
@@ -26,7 +27,7 @@ router.post("/:id/renew", scopedAuth({ scope: "lease:write" }), async (c) => {
   return c.json(result.lease);
 });
 
-router.delete("/:id", scopedAuth({ scope: "lease:write" }), async (c) => {
+router.delete("/:id", scopedAuth({ scope: "lease:write" }), rateLimitMiddleware, async (c) => {
   const error = await releaseLease(c.get("db"), c.get("projectId"), c.req.param("id"));
   if (error) return errorResponse(c, error.code, error.message, error.status);
   return c.body(null, 204);
diff --git a/packages/api/src/routes/mcp/index.ts b/packages/api/src/routes/mcp/index.ts
@@ -1,6 +1,7 @@
 import { Hono } from "hono";
 import { scopeSatisfies } from "../../lib/scopes";
 import { mcpAuth } from "../../middleware/mcp-auth";
+import { rateLimitMiddleware } from "../../middleware/rate-limit";
 import type { Bindings, Variables } from "../../types";
 import { TOOLS, TOOLS_BY_NAME, type ToolContext, ToolError } from "./tools";
 
@@ -55,6 +56,7 @@ function rpcError(id: string | number | null, code: number, message: string) {
 }
 
 router.use("*", mcpAuth);
+router.use("*", rateLimitMiddleware);
 
 // Server-initiated SSE is not supported in stateless mode.
 router.get("/", (c) =>
diff --git a/packages/api/src/routes/states/index.ts b/packages/api/src/routes/states/index.ts
@@ -2,6 +2,7 @@ import { Hono } from "hono";
 import { streamSSE } from "hono/streaming";
 import { errorResponse, parseAndValidateBody, parseLimitParam } from "../../lib/helpers";
 import { CreateLeaseSchema, QueryStatesSchema, UpsertStateSchema } from "../../lib/validation";
+import { rateLimitMiddleware } from "../../middleware/rate-limit";
 import { scopedAuth } from "../../middleware/scoped-auth";
 import { createLease } from "../../services/leases";
 import {
@@ -47,45 +48,52 @@ function notifyStateEvent(c: any, event: StateEventResponse) {
 // GET /watch — SSE state event stream
 // ---------------------------------------------------------------------------
 
-router.get("/watch", scopedAuth({ scope: "state:watch", allowQueryToken: true }), async (c) => {
-  const after =
-    parsePositiveInt(c.req.query("after")) ?? parsePositiveInt(c.req.header("Last-Event-ID")) ?? 0;
-  const once = c.req.query("once") === "true";
-  const hub = stateHub(c);
-
-  if (hub && !once) {
-    const url = new URL("https://state-stream.local/watch");
-    url.searchParams.set("after", String(after));
-    return hub.fetch(
-      new Request(url, {
-        headers: { "X-Project-Id": c.get("projectId") },
-      }),
-    );
-  }
+router.get(
+  "/watch",
+  scopedAuth({ scope: "state:watch", allowQueryToken: true }),
+  rateLimitMiddleware,
+  async (c) => {
+    const after =
+      parsePositiveInt(c.req.query("after")) ??
+      parsePositiveInt(c.req.header("Last-Event-ID")) ??
+      0;
+    const once = c.req.query("once") === "true";
+    const hub = stateHub(c);
+
+    if (hub && !once) {
+      const url = new URL("https://state-stream.local/watch");
+      url.searchParams.set("after", String(after));
+      return hub.fetch(
+        new Request(url, {
+          headers: { "X-Project-Id": c.get("projectId") },
+        }),
+      );
+    }
 
-  return streamSSE(c, async (stream) => {
-    const events = await c
-      .get("d1Db")
-      .prepare(
-        `SELECT sequence, id, state_key, agent_id, event_type, data, metadata, tags, idempotency_key, created_at
+    return streamSSE(c, async (stream) => {
+      const events = await c
+        .get("d1Db")
+        .prepare(
+          `SELECT sequence, id, state_key, agent_id, event_type, data, metadata, tags, idempotency_key, created_at
          FROM state_events
          WHERE project_id = ? AND sequence > ?
          ORDER BY sequence ASC
          LIMIT 1000`,
-      )
-      .bind(c.get("projectId"), after)
-      .all<any>();
-
-    for (const event of events.results ?? []) {
-      await stream.writeSSE({
-        id: String(event.sequence),
-        event: `state.${event.event_type}`,
-        data: JSON.stringify(event),
-      });
-    }
-    stream.close();
-  });
-});
+        )
+        .bind(c.get("projectId"), after)
+        .all<any>();
+
+      for (const event of events.results ?? []) {
+        await stream.writeSSE({
+          id: String(event.sequence),
+          event: `state.${event.event_type}`,
+          data: JSON.stringify(event),
+        });
+      }
+      stream.close();
+    });
+  },
+);
 
 // ---------------------------------------------------------------------------
 // POST /query — Rich state query
@@ -98,7 +106,7 @@ router.get("/watch", scopedAuth({ scope: "state:watch", allowQueryToken: true })
 // whose `after` cursor is ascending (`sequence > after`).
 // ---------------------------------------------------------------------------
 
-router.post("/query", scopedAuth({ scope: "state:read" }), async (c) => {
+router.post("/query", scopedAuth({ scope: "state:read" }), rateLimitMiddleware, async (c) => {
   const { data, error } = await parseAndValidateBody(c, QueryStatesSchema);
   if (error) return error;
   if (!data) return errorResponse(c, "BAD_REQUEST", "Invalid request body", 400);
@@ -122,55 +130,65 @@ router.post("/query", scopedAuth({ scope: "state:read" }), async (c) => {
 // POST /:state_key/lease — Acquire lease
 // ---------------------------------------------------------------------------
 
-router.post("/:state_key/lease", scopedAuth({ scope: "lease:write" }), async (c) => {
-  const { data, error } = await parseAndValidateBody(c, CreateLeaseSchema);
-  if (error) return error;
-  if (!data) return errorResponse(c, "BAD_REQUEST", "Invalid request body", 400);
-
-  const result = await createLease(
-    c.get("db"),
-    c.get("projectId"),
-    c.req.param("state_key"),
-    data.holder,
-    data.ttl_ms,
-  );
+router.post(
+  "/:state_key/lease",
+  scopedAuth({ scope: "lease:write" }),
+  rateLimitMiddleware,
+  async (c) => {
+    const { data, error } = await parseAndValidateBody(c, CreateLeaseSchema);
+    if (error) return error;
+    if (!data) return errorResponse(c, "BAD_REQUEST", "Invalid request body", 400);
+
+    const result = await createLease(
+      c.get("db"),
+      c.get("projectId"),
+      c.req.param("state_key"),
+      data.holder,
+      data.ttl_ms,
+    );
 
-  if (result.error) {
-    return errorResponse(c, result.error.code, result.error.message, result.error.status);
-  }
+    if (result.error) {
+      return errorResponse(c, result.error.code, result.error.message, result.error.status);
+    }
 
-  return c.json(result.lease, 201);
-});
+    return c.json(result.lease, 201);
+  },
+);
 
 // ---------------------------------------------------------------------------
 // GET /:state_key/events — WAL events
 // ---------------------------------------------------------------------------
 
-router.get("/:state_key/events", scopedAuth({ scope: "state:read" }), async (c) => {
-  const after = parsePositiveInt(c.req.query("after")) ?? 0;
-  const limit = parseLimitParam(c.req.query("limit"), 50, 500);
-  const events = await listStateEvents(
-    c.get("d1Db"),
-    c.get("projectId"),
-    c.req.param("state_key"),
-    after,
-    limit,
-  );
+router.get(
+  "/:state_key/events",
+  scopedAuth({ scope: "state:read" }),
+  rateLimitMiddleware,
+  async (c) => {
+    const after = parsePositiveInt(c.req.query("after")) ?? 0;
+    const limit = parseLimitParam(c.req.query("limit"), 50, 500);
+    const events = await listStateEvents(
+      c.get("d1Db"),
+      c.get("projectId"),
+      c.req.param("state_key"),
+      after,
+      limit,
+    );
 
-  return c.json({
-    data: events,
-    pagination: {
-      limit: events.length,
-      next_cursor: events.length ? String(events[events.length - 1].sequence) : null,
-    },
-  });
-});
+    return c.json({
+      data: events,
+      pagination: {
+        limit: events.length,
+        next_cursor: events.length ? String(events[events.length - 1].sequence) : null,
+      },
+    });
+  },
+);
 
 // ---------------------------------------------------------------------------
 // PUT /:state_key — Upsert full state
 // ---------------------------------------------------------------------------
 
-router.put("/:state_key", scopedAuth({ scope: "state:write" }), async (c) => {
+router.put("/:state_key", scopedAuth({ scope: "state:write" }), rateLimitMiddleware, async (c) => {
   const { data, error } = await parseAndValidateBody(c, UpsertStateSchema);
   if (error) return error;
   if (!data) return errorResponse(c, "BAD_REQUEST", "Invalid request body", 400);
@@ -219,7 +237,7 @@ router.put("/:state_key", scopedAuth({ scope: "state:write" }), async (c) => {
 // GET /:state_key — Latest or historical state
 // ---------------------------------------------------------------------------
 
-router.get("/:state_key", scopedAuth({ scope: "state:read" }), async (c) => {
+router.get("/:state_key", scopedAuth({ scope: "state:read" }), rateLimitMiddleware, async (c) => {
   const atSequence = parsePositiveInt(c.req.query("at_sequence"));
   const atTime = parsePositiveInt(c.req.query("at_time"));
   const state =
@@ -241,46 +259,53 @@ router.get("/:state_key", scopedAuth({ scope: "state:read" }), async (c) => {
 // DELETE /:state_key — Tombstone state
 // ---------------------------------------------------------------------------
 
-router.delete("/:state_key", scopedAuth({ scope: "state:write" }), async (c) => {
-  const stateKey = c.req.param("state_key");
-  const leaseId = c.req.header("X-Lease-Id") ?? c.req.query("lease_id");
-  const idempotencyKey = c.req.header("Idempotency-Key");
-  const requestHash = await buildIdempotencyHash("DELETE", stateKey, { lease_id: leaseId ?? null });
-  const cached = await readIdempotency(
-    c.get("d1Db"),
-    c.get("projectId"),
-    idempotencyKey,
-    requestHash,
-  );
-  if (cached.error)
-    return errorResponse(c, cached.error.code, cached.error.message, cached.error.status);
-  if (cached.replay) return cached.replay;
-
-  const result = await deleteState(
-    c.get("d1Db"),
-    c.get("projectId"),
-    stateKey,
-    leaseId,
-    idempotencyKey,
-    requestHash,
-  );
-  if (result.error)
-    return errorResponse(c, result.error.code, result.error.message, result.error.status);
-  // A concurrent duplicate carried the same Idempotency-Key: this request's
-  // mutation was rolled back atomically; serve the winner's stored response.
-  if (result.replay) return result.replay;
-  if (!result.result) return errorResponse(c, "INTERNAL_ERROR", "State delete failed", 500);
-
-  await storeIdempotency(
-    c.get("d1Db"),
-    c.get("projectId"),
-    idempotencyKey,
-    requestHash,
-    result.result.status,
-    result.result.body,
-  );
-  notifyStateEvent(c, result.result.event);
-  return c.json(result.result.body, 200);
-});
+router.delete(
+  "/:state_key",
+  scopedAuth({ scope: "state:write" }),
+  rateLimitMiddleware,
+  async (c) => {
+    const stateKey = c.req.param("state_key");
+    const leaseId = c.req.header("X-Lease-Id") ?? c.req.query("lease_id");
+    const idempotencyKey = c.req.header("Idempotency-Key");
+    const requestHash = await buildIdempotencyHash("DELETE", stateKey, {
+      lease_id: leaseId ?? null,
+    });
+    const cached = await readIdempotency(
+      c.get("d1Db"),
+      c.get("projectId"),
+      idempotencyKey,
+      requestHash,
+    );
+    if (cached.error)
+      return errorResponse(c, cached.error.code, cached.error.message, cached.error.status);
+    if (cached.replay) return cached.replay;
+
+    const result = await deleteState(
+      c.get("d1Db"),
+      c.get("projectId"),
+      stateKey,
+      leaseId,
+      idempotencyKey,
+      requestHash,
+    );
+    if (result.error)
+      return errorResponse(c, result.error.code, result.error.message, result.error.status);
+    // A concurrent duplicate carried the same Idempotency-Key: this request's
+    // mutation was rolled back atomically; serve the winner's stored response.
+    if (result.replay) return result.replay;
+    if (!result.result) return errorResponse(c, "INTERNAL_ERROR", "State delete failed", 500);
+
+    await storeIdempotency(
+      c.get("d1Db"),
+      c.get("projectId"),
+      idempotencyKey,
+      requestHash,
+      result.result.status,
+      result.result.body,
+    );
+    notifyStateEvent(c, result.result.event);
+    return c.json(result.result.body, 200);
+  },
+);
 
 export default router;
diff --git a/packages/api/test/coordination-rate-limit.test.ts b/packages/api/test/coordination-rate-limit.test.ts
