fix(link): 3 more codex [P2] findings — per-route caps, bool rules, proxy validation

Round 2 of review on the same branch:

- The fail-closed capacity guard was itself a DoS lever. MAX_RATE_KEYS was
  global, and the limiter runs before auth, so a flood of cheap /v1/repos
  keys could exhaust the table and 429 never-seen /v1/link clients until
  live windows expired. Budgets are now per route.
- Rule values accepted booleans: bool subclasses int, so isinstance let
  [5, true] through as a 1-SECOND window (5/hour -> ~5/sec) and false in the
  limit slot disabled the endpoint. Now `type(x) is int`.
- trusted_proxies was unvalidated: a bare string would be iterated character
  by character, malformed entries only surfaced as a per-request log line,
  and 0.0.0.0/0 or ::/0 restored "trust XFF from any peer" — the exact hole
  the setting closes. Now parsed and validated once at startup, wildcards
  refused, and _ip_in_any takes pre-parsed networks so nothing can degrade
  to a silent per-request skip.

Tests 99 -> 108: cross-route flood isolation, per-route reclamation windows,
every bool-in-rule position, bare-string and wildcard proxies, and a v4/v6
mismatch case.

Co-Authored-By: Claude Opus 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01LTARYHX7GPepi3CH3tp5pg

README.md

@@ -119,11 +119,16 @@ header, decided *before* the body is read so an abusive caller costs nothing.
   multi-process or multi-host, the limiter must move with it.** The clock is
   read *inside* the lock; taken outside, racing threads append out of order
   and both `Retry-After` and window reclamation silently go wrong.
-* The key store is capped (`MAX_RATE_KEYS`). At capacity it reclaims expired
+* The key store is capped (`MAX_RATE_KEYS`) **per route**, not globally. At
-  windows, and if every window is still live it **refuses the new key** —
+  capacity it reclaims expired windows, and if every window is still live it
-  fail closed. Evicting a live window would let an attacker who can mint many
+  **refuses the new key** — fail closed. Evicting a live window would let an
-  distinct keys clear their *own* limit on demand, which is worse than
+  attacker who can mint many distinct keys clear their *own* limit on demand.
-  turning a request away on an invite-only endpoint.
+  The per-route budget matters just as much: with one shared table, a flood
+  of cheap `/v1/repos` keys would exhaust it and lock brand-new `/v1/link`
+  clients out, turning the fail-closed guard into a cross-route DoS.
+* Rule values must be real integers. `bool` subclasses `int` in Python, so
+  `[5, true]` would otherwise pass as a **1-second** window — an hourly limit
+  quietly becoming ~5/sec.
 * `trust_forwarded_for` is **off** by default, and turning it on **requires a
   non-empty `trusted_proxies`** — the header is honored only when the socket
   peer is in that list. Without it, anyone reaching the origin directly (it
@@ -133,6 +138,10 @@ header, decided *before* the body is read so an abusive caller costs nothing.
   starve everyone. A caller can *prepend* anything to `X-Forwarded-For`; a
   trusted proxy *appends* the peer it actually saw, so the service reads the
   **last** entry, never the first, and requires it to parse as a real IP.
+  `trusted_proxies` is validated at startup: it must be a list (a bare string
+  would be iterated character by character), every entry a valid network, and
+  wildcards (`0.0.0.0/0`, `::/0`) are refused outright — they would restore
+  exactly the "trust anyone" hole the setting exists to close.
 
 #### Identity binding (`state.json`)
 

server/granthi_link.py

@@ -104,9 +104,13 @@ class RateLimiter:
 
     def __init__(self, rules=None, max_keys=MAX_RATE_KEYS, clock=None):
         self.rules = dict(rules or DEFAULT_RATE_RULES)
+        # Per-ROUTE budget, not one global table. A shared cap lets a flood of
+        # cheap /v1/repos keys exhaust the table and lock brand-new /v1/link
+        # clients out -- turning the fail-closed capacity guard into a
+        # cross-route denial of service. Each route gets its own space.
         self.max_keys = max_keys
         self._clock = clock or time.monotonic
-        self._hits = {}
+        self._hits = {route: {} for route in self.rules}
         self._lock = threading.Lock()
 
     def check(self, route, client):
@@ -121,43 +125,45 @@ class RateLimiter:
         limit, window = rule
         if limit <= 0:                      # 0 = endpoint disabled entirely
             return False, window
-        key = (route, client)
         with self._lock:
             # Clock read INSIDE the lock: taken outside, two racing threads
             # can append out of order, and both hits[0] (retry_after) and
-            # v[-1] (eviction age) assume the list is chronological.
+            # v[-1] (reclamation age) assume the list is chronological.
             now = self._clock()
-            hits = [t for t in self._hits.get(key, ()) if now - t < window]
+            table = self._hits.setdefault(route, {})
+            hits = [t for t in table.get(client, ()) if now - t < window]
             if len(hits) >= limit:
-                self._hits[key] = hits
+                table[client] = hits
                 return False, max(1, int(window - (now - hits[0])) + 1)
-            if key not in self._hits and len(self._hits) >= self.max_keys:
+            if client not in table and len(table) >= self.max_keys:
                 # At capacity, reclaim expired keys first...
-                self._reclaim_expired(now)
+                self._reclaim_expired(route, now)
-                if len(self._hits) >= self.max_keys:
+                if len(table) >= self.max_keys:
                     # ...and if every window is still live, this is an
                     # identity flood, not organic load. Evicting here would
                     # let an attacker reset their own limit on demand, so
                     # refuse the NEW key instead. Fail closed: /v1/link is
                     # invite-only and low-volume, so hitting this cap is an
                     # attack, and turning tokens away beats minting them.
-                    LOG.error("rate limiter at capacity (%d keys) with no "
+                    LOG.error("rate limiter at capacity for %s (%d keys) with "
-                              "expired windows: refusing new client %r on %s",
+                              "no expired windows: refusing new client %r",
-                              self.max_keys, client, route)
+                              route, self.max_keys, client)
                     return False, window
             hits.append(now)
-            self._hits[key] = hits
+            table[client] = hits
             return True, 0
 
-    def _reclaim_expired(self, now):
+    def _reclaim_expired(self, route, now):
         """Caller holds the lock. Drop only keys whose window has fully
-        expired -- never a live one, or eviction becomes the bypass."""
+        expired -- never a live one, or reclamation becomes the bypass."""
-        dead = [k for k, v in self._hits.items()
+        window = self.rules[route][1]
-                if not v or now - v[-1] >= self.rules[k[0]][1]]
+        table = self._hits.get(route, {})
-        for k in dead:
+        dead = [c for c, v in table.items() if not v or now - v[-1] >= window]
-            del self._hits[k]
+        for c in dead:
+            del table[c]
         if dead:
-            LOG.info("rate limiter reclaimed %d expired windows", len(dead))
+            LOG.info("rate limiter reclaimed %d expired windows on %s",
+                     len(dead), route)
 
 
 def client_ip(handler, trust_forwarded_for, trusted_proxies=()):
@@ -202,11 +208,10 @@ def _ip_in_any(addr, networks):
     except ValueError:
         return False
     for net in networks:
-        try:
+        # Pre-parsed at startup by LinkService._parse_trusted_proxies, so a
-            if ip in ipaddress.ip_network(net, strict=False):
+        # malformed entry can never reach here as a silent per-request skip.
-                return True
+        if ip.version == net.version and ip in net:
-        except ValueError:
+            return True
-            LOG.error("ignoring malformed trusted_proxies entry %r", net)
     return False
 
 
@@ -350,7 +355,8 @@ class LinkService:
                     f"config rate_limit.{flag} must be true or false, "
                     f"got {rl[flag]!r}")
         self.trust_forwarded_for = rl.get("trust_forwarded_for", False)
-        self.trusted_proxies = tuple(rl.get("trusted_proxies") or ())
+        self.trusted_proxies = self._parse_trusted_proxies(
+            rl.get("trusted_proxies"))
         if self.trust_forwarded_for and not self.trusted_proxies:
             raise SystemExit(
                 "config rate_limit.trust_forwarded_for requires a non-empty "
@@ -359,8 +365,12 @@ class LinkService:
         if rl.get("enabled", True):
             rules = dict(DEFAULT_RATE_RULES)
             for route, spec in (rl.get("rules") or {}).items():
+                # `type(x) is int`, NOT isinstance: bool subclasses int, so
+                # isinstance lets [5, true] through as a 1-SECOND window --
+                # an hourly limit silently becomes ~5/sec -- and false in the
+                # limit slot disables the endpoint.
                 if (not isinstance(spec, (list, tuple)) or len(spec) != 2
-                        or not all(isinstance(x, int) for x in spec)
+                        or not all(type(x) is int for x in spec)
                         or spec[1] <= 0):
                     # A malformed rule must not silently mean "unlimited".
                     raise SystemExit(
@@ -377,6 +387,43 @@ class LinkService:
             LOG.warning("rate limiting DISABLED by config -- /v1/link will "
                         "mint tokens without any throttle")
 
+    @staticmethod
+    def _parse_trusted_proxies(raw):
+        """Validate at STARTUP, not per request.
+
+        This setting gates a spoofable identity, so every failure mode has to
+        be loud and early: a bare string would be iterated character by
+        character (each char a "network"), malformed entries would only
+        surface as a per-request log line, and a wildcard like 0.0.0.0/0 or
+        ::/0 quietly restores "trust X-Forwarded-For from anyone" -- the exact
+        hole trusted_proxies exists to close.
+        """
+        if raw is None:
+            return ()
+        if isinstance(raw, str) or not isinstance(raw, (list, tuple)):
+            raise SystemExit(
+                "config rate_limit.trusted_proxies must be a list of CIDRs, "
+                f"got {type(raw).__name__}")
+        nets = []
+        for entry in raw:
+            if not isinstance(entry, str):
+                raise SystemExit(
+                    f"config rate_limit.trusted_proxies entry {entry!r} "
+                    "must be a string")
+            try:
+                net = ipaddress.ip_network(entry, strict=False)
+            except ValueError as e:
+                raise SystemExit(
+                    f"config rate_limit.trusted_proxies entry {entry!r} "
+                    f"is not a valid network: {e}")
+            if net.prefixlen == 0:
+                raise SystemExit(
+                    f"config rate_limit.trusted_proxies entry {entry!r} "
+                    "matches every address, which is the same as trusting "
+                    "X-Forwarded-For from any peer -- refusing")
+            nets.append(net)
+        return tuple(nets)
+
     def test_mode_enabled(self):
         """Config test_mode is honored ONLY with the env gate also set."""
         if not self.cfg.get("test_mode"):

tests/test_server.py

@@ -544,7 +544,7 @@ class TestRateLimiterUnit(unittest.TestCase):
         for i in range(500):
             self.now += 0.001
             rl.check("/v1/link", f"10.0.{i // 256}.{i % 256}")
-        self.assertLessEqual(len(rl._hits), 50 + 1)
+        self.assertLessEqual(len(rl._hits["/v1/link"]), 50 + 1)
 
     def test_expired_keys_are_reclaimed(self):
         rl = granthi_link.RateLimiter(rules={"/v1/link": (5, 60)},
@@ -554,7 +554,7 @@ class TestRateLimiterUnit(unittest.TestCase):
         self.now += 120                      # everything expires
         for i in range(10, 25):
             rl.check("/v1/link", f"10.0.0.{i}")
-        self.assertLessEqual(len(rl._hits), 11)
+        self.assertLessEqual(len(rl._hits["/v1/link"]), 11)
 
     def test_concurrent_checks_never_exceed_the_limit(self):
         """The lock has to actually hold under threads: 40 racing callers
@@ -585,7 +585,9 @@ class _FakeHandler:
 
 
 class TestClientIp(unittest.TestCase):
-    PROXY = ("10.0.0.0/8",)
+    # Pre-parsed, exactly as LinkService._parse_trusted_proxies hands them
+    # over -- validation happens once at startup, never per request.
+    PROXY = granthi_link.LinkService._parse_trusted_proxies(["10.0.0.0/8"])
 
     def test_socket_peer_by_default(self):
         h = _FakeHandler("5.5.5.5", xff="1.2.3.4")
@@ -618,10 +620,16 @@ class TestClientIp(unittest.TestCase):
         self.assertEqual(granthi_link.client_ip(h, True, self.PROXY),
                          "10.0.0.1")
 
-    def test_malformed_trusted_proxy_entry_does_not_grant_trust(self):
+    def test_malformed_trusted_proxy_entry_is_rejected_at_startup(self):
+        """It never reaches a request: a bad CIDR kills the process rather
+        than degrading to a per-request log line."""
+        with self.assertRaises(SystemExit):
+            granthi_link.LinkService._parse_trusted_proxies(["not-a-cidr"])
+
+    def test_v4_peer_does_not_match_a_v6_trusted_net(self):
+        nets = granthi_link.LinkService._parse_trusted_proxies(["fd00::/8"])
         h = _FakeHandler("10.0.0.1", xff="9.9.9.9")
-        self.assertEqual(
+        self.assertEqual(granthi_link.client_ip(h, True, nets), "10.0.0.1")
-            granthi_link.client_ip(h, True, ("not-a-cidr",)), "10.0.0.1")
 
 
 class TestRateLimitConfig(ServiceTestBase):
@@ -693,14 +701,14 @@ class TestRateLimitHardening(unittest.TestCase):
                                       max_keys=5, clock=lambda: now[0])
         for i in range(5):
             self.assertTrue(rl.check("/v1/link", f"10.0.0.{i}")[0])
-        victim_hits = list(rl._hits[("/v1/link", "10.0.0.0")])
+        victim_hits = list(rl._hits["/v1/link"]["10.0.0.0"])
         for i in range(100, 140):                 # identity flood
             allowed, retry = rl.check("/v1/link", f"10.0.1.{i}")
             self.assertFalse(allowed)
             self.assertGreater(retry, 0)
         # the earlier client's window survived the flood untouched
-        self.assertEqual(rl._hits[("/v1/link", "10.0.0.0")], victim_hits)
+        self.assertEqual(rl._hits["/v1/link"]["10.0.0.0"], victim_hits)
-        self.assertLessEqual(len(rl._hits), 5)
+        self.assertLessEqual(len(rl._hits["/v1/link"]), 5)
 
     def test_capacity_recovers_once_windows_expire(self):
         now = [1000.0]
@@ -723,7 +731,7 @@ class TestRateLimitHardening(unittest.TestCase):
             t.start()
         for t in ts:
             t.join()
-        hits = rl._hits[("/v1/link", "7.7.7.7")]
+        hits = rl._hits["/v1/link"]["7.7.7.7"]
         self.assertEqual(len(hits), 200)
         self.assertEqual(hits, sorted(hits), "timestamps out of order")
 
@@ -770,3 +778,76 @@ class TestRateLimitConfigTypes(ServiceTestBase):
                          "trusted_proxies": ["10.0.0.0/8", "127.0.0.1/32"]})
         self.assertTrue(svc.trust_forwarded_for)
         self.assertEqual(len(svc.trusted_proxies), 2)
+
+
+class TestRateLimitRound2(unittest.TestCase):
+    """The three findings from the codex re-review."""
+
+    def test_flooding_one_route_cannot_lock_out_another(self):
+        """A shared key table turned the fail-closed capacity guard into a
+        cross-route DoS: cheap /v1/repos keys could exhaust it and shut new
+        /v1/link clients out. Budgets are per route."""
+        now = [1000.0]
+        rl = granthi_link.RateLimiter(
+            rules={"/v1/link": (5, 3600), "/v1/repos": (60, 3600)},
+            max_keys=20, clock=lambda: now[0])
+        for i in range(300):                       # flood the cheap route
+            rl.check("/v1/repos", f"10.1.{i // 256}.{i % 256}")
+        allowed, _ = rl.check("/v1/link", "203.0.113.7")   # never-seen client
+        self.assertTrue(allowed, "/v1/link locked out by a /v1/repos flood")
+
+    def test_reclaim_uses_the_right_window_per_route(self):
+        now = [1000.0]
+        rl = granthi_link.RateLimiter(
+            rules={"/v1/link": (1, 3600), "/v1/repos": (1, 10)},
+            max_keys=2, clock=lambda: now[0])
+        rl.check("/v1/repos", "a")
+        rl.check("/v1/repos", "b")
+        self.assertFalse(rl.check("/v1/repos", "c")[0])
+        now[0] += 11                    # past the /v1/repos window only
+        self.assertTrue(rl.check("/v1/repos", "c")[0])
+
+
+class TestRateLimitConfigRound2(ServiceTestBase):
+    def _svc(self, rl):
+        cfg = dict(self.svc.cfg)
+        cfg["rate_limit"] = rl
+        return granthi_link.LinkService(cfg)
+
+    def test_booleans_are_rejected_in_rule_slots(self):
+        """bool subclasses int, so isinstance let [5, True] through as a
+        1-SECOND window -- 5/hour silently became ~5/sec."""
+        for bad in ([5, True], [True, 3600], [False, 3600], [5, False]):
+            with self.subTest(rule=bad):
+                with self.assertRaises(SystemExit):
+                    self._svc({"rules": {"/v1/link": bad}})
+
+    def test_valid_int_rule_still_accepted(self):
+        svc = self._svc({"rules": {"/v1/link": [5, 3600]}})
+        self.assertEqual(svc.limiter.rules["/v1/link"], (5, 3600))
+
+    def test_trusted_proxies_as_bare_string_refuses_startup(self):
+        """A string would be iterated character by character, each char
+        treated as a network."""
+        with self.assertRaises(SystemExit):
+            self._svc({"trust_forwarded_for": True,
+                       "trusted_proxies": "10.0.0.0/8"})
+
+    def test_wildcard_trusted_proxy_refuses_startup(self):
+        """0.0.0.0/0 or ::/0 restores 'trust XFF from anyone' -- the exact
+        hole trusted_proxies exists to close."""
+        for wild in ("0.0.0.0/0", "::/0"):
+            with self.subTest(cidr=wild):
+                with self.assertRaises(SystemExit):
+                    self._svc({"trust_forwarded_for": True,
+                               "trusted_proxies": [wild]})
+
+    def test_non_string_proxy_entry_refuses_startup(self):
+        with self.assertRaises(SystemExit):
+            self._svc({"trust_forwarded_for": True,
+                       "trusted_proxies": [10, "10.0.0.0/8"]})
+
+    def test_host_address_without_prefix_is_accepted(self):
+        svc = self._svc({"trust_forwarded_for": True,
+                         "trusted_proxies": ["127.0.0.1", "10.0.0.0/8"]})
+        self.assertEqual(len(svc.trusted_proxies), 2)