Merge feat/rate-limit: throttle /v1/link and /v1/repos

Closes the last unbuilt item on the promotion-window hardening checklist.
Three codex rounds: 4 findings, then 3, then clean. 108 tests.

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

README.md

@@ -90,6 +90,59 @@ window. The service **refuses to start** (exit 2) unless `config.json` is
 mode 0600/0400 and owned by the user it runs as — the config carries the
 forge admin password, so permissive perms fail closed, not open.
 
+#### Rate limiting
+
+Sliding-window, per `(route, client)`, **on by default** — `/v1/link` creates
+accounts and mints tokens, so unlimited has to be a deliberate config act,
+never an omission. Defaults: `/v1/link` 5/hour, `/v1/repos` 60/hour. `GET
+/health` is never limited. Over the limit → **429** with a `Retry-After`
+header, decided *before* the body is read so an abusive caller costs nothing.
+
+```json
+"rate_limit": {
+  "enabled": true,
+  "trust_forwarded_for": false,
+  "trusted_proxies": [],
+  "rules": {"/v1/link": [5, 3600], "/v1/repos": [60, 3600]}
+}
+```
+
+* Anything malformed **refuses startup** rather than silently meaning
+  unlimited: a bad rule, a non-boolean `enabled`/`trust_forwarded_for` (JSON
+  `null`, `0`, or the *string* `"false"` — every non-empty string is truthy),
+  or a `rate_limit` that is not an object. `[0, N]` disables an endpoint
+  outright.
+* Denied requests are *not* recorded, so a client that keeps hammering cannot
+  push its own window forward and lock itself out permanently.
+* State is an in-process dict behind a lock — granthi-link is one
+  `ThreadingHTTPServer`, so that is the entire store. **If this ever runs
+  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`) **per route**, not globally. At
+  capacity it reclaims expired windows, and if every window is still live it
+  **refuses the new key** — fail closed. Evicting a live window would let an
+  attacker who can mint many distinct keys clear their *own* limit on demand.
+  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
+  also listens on the tailnet) could pick and rotate their own rate-limit key
+  just by sending a header. Turn it on when exposing behind cloudflared,
+  where every request otherwise arrives from the tunnel and one abuser would
+  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`)
 
 `/v1/link` originally bound purely by `preferred_username` / email
@@ -226,8 +279,11 @@ Every linked folder becomes a private repo under their account.
    app is host-independent. Rotate the beta admin token/password out of the
    config when pointing at prod (prod forge is READ-ONLY to this estate —
    promotion is an operator action, not an agent action).
-4. Add rate limiting / abuse controls before public exposure (one token mint
-   per link call today).
+4. ~~Add rate limiting / abuse controls before public exposure.~~ **DONE** —
+   see "Rate limiting" above (5 links/hour per client by default). When
+   exposing behind cloudflared, set `trust_forwarded_for: true` in the same
+   change, or every request will look like the tunnel and one abuser will
+   throttle everybody.
 5. **Hardening checklist (must all hold before exposing):**
    - [ ] `config.json` is 0600 (or 0400) and owned by the service user —
          the service refuses to start otherwise; verify with

server/config.example.json

@@ -6,5 +6,10 @@
   "admin_login": "nirpa",
   "admin_password": "FROM /opt/gitea-beta/.admin-creds (required: Gitea 1.27 token minting only works via basic auth + Sudo header)",
   "binds": [["127.0.0.1", 3042], ["100.111.127.127", 3042]],
-  "test_mode": false
+  "test_mode": false,
+  "rate_limit": {
+    "enabled": true,
+    "trust_forwarded_for": false,
+    "rules": {"/v1/link": [5, 3600], "/v1/repos": [60, 3600]}
+  }
 }

server/granthi_link.py

@@ -52,6 +52,7 @@ Stdlib only. Python 3.9+.
 """
 
 import base64
+import ipaddress
 import json
 import logging
 import os
@@ -61,6 +62,7 @@ import signal
 import string
 import sys
 import threading
+import time
 import urllib.error
 import urllib.request
 from datetime import datetime, timezone
@@ -79,6 +81,139 @@ USER_CREATE_CONFLICT = object()
 
 LOGIN_SAFE = re.compile(r"[^a-zA-Z0-9._-]+")
 
+# Rate limiting. /v1/link is the expensive endpoint -- it round-trips Zitadel,
+# can CREATE a forge account and always mints a token -- so its default is
+# deliberately tight. /v1/repos only spends the caller's own token.
+DEFAULT_RATE_RULES = {"/v1/link": (5, 3600), "/v1/repos": (60, 3600)}
+MAX_RATE_KEYS = 10000
+
+
+# --------------------------------------------------------------------------
+# Rate limiting
+# --------------------------------------------------------------------------
+
+class RateLimiter:
+    """Sliding-window limiter keyed by (route, client).
+
+    In-process and lock-guarded: granthi-link is ONE ThreadingHTTPServer
+    process, so a dict is the entire store -- no redis, no shared cache, and
+    nothing to keep consistent across nodes. If this ever runs multi-process
+    the limiter must move with it; that is why the store is behind this class
+    rather than sprinkled through the handler.
+    """
+
+    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 = {route: {} for route in self.rules}
+        self._lock = threading.Lock()
+
+    def check(self, route, client):
+        """-> (allowed: bool, retry_after: int). Records the hit when allowed.
+
+        Denied requests are NOT recorded: a client that keeps hammering must
+        not push its own window forward and lock itself out indefinitely.
+        """
+        rule = self.rules.get(route)
+        if not rule:
+            return True, 0
+        limit, window = rule
+        if limit <= 0:                      # 0 = endpoint disabled entirely
+            return False, window
+        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] (reclamation age) assume the list is chronological.
+            now = self._clock()
+            table = self._hits.setdefault(route, {})
+            hits = [t for t in table.get(client, ()) if now - t < window]
+            if len(hits) >= limit:
+                table[client] = hits
+                return False, max(1, int(window - (now - hits[0])) + 1)
+            if client not in table and len(table) >= self.max_keys:
+                # At capacity, reclaim expired keys first...
+                self._reclaim_expired(route, now)
+                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 for %s (%d keys) with "
+                              "no expired windows: refusing new client %r",
+                              route, self.max_keys, client)
+                    return False, window
+            hits.append(now)
+            table[client] = hits
+            return True, 0
+
+    def _reclaim_expired(self, route, now):
+        """Caller holds the lock. Drop only keys whose window has fully
+        expired -- never a live one, or reclamation becomes the bypass."""
+        window = self.rules[route][1]
+        table = self._hits.get(route, {})
+        dead = [c for c, v in table.items() if not v or now - v[-1] >= window]
+        for c in dead:
+            del table[c]
+        if dead:
+            LOG.info("rate limiter reclaimed %d expired windows on %s",
+                     len(dead), route)
+
+
+def client_ip(handler, trust_forwarded_for, trusted_proxies=()):
+    """The address to rate-limit on.
+
+    Behind cloudflared every request arrives from the tunnel, so limiting on
+    the socket peer would let one abuser starve everyone. X-Forwarded-For is
+    client-controlled, though: a caller can prepend anything. Two guards:
+
+    1. The header is honored ONLY when the socket peer is itself a configured
+       trusted proxy. Without that, anyone who can reach the origin directly
+       -- and the origin also listens on the tailnet -- picks their own
+       rate-limit key and rotates it at will.
+    2. A trusted proxy APPENDS the peer it actually saw, so the LAST entry is
+       the only one the client did not choose. Take that, never the first.
+
+    The value must parse as a real IP; junk falls back to the socket peer
+    rather than becoming a key of its own.
+    """
+    peer = handler.client_address[0]
+    if not trust_forwarded_for:
+        return peer
+    if not _ip_in_any(peer, trusted_proxies):
+        LOG.warning("X-Forwarded-For ignored: peer %s is not a trusted proxy",
+                    peer)
+        return peer
+    xff = handler.headers.get("X-Forwarded-For", "")
+    parts = [p.strip() for p in xff.split(",") if p.strip()]
+    if not parts:
+        return peer
+    try:
+        return str(ipaddress.ip_address(parts[-1]))
+    except ValueError:
+        LOG.warning("X-Forwarded-For last hop %r is not an IP; using peer",
+                    parts[-1][:60])
+        return peer
+
+
+def _ip_in_any(addr, networks):
+    try:
+        ip = ipaddress.ip_address(addr)
+    except ValueError:
+        return False
+    for net in networks:
+        # Pre-parsed at startup by LinkService._parse_trusted_proxies, so a
+        # malformed entry can never reach here as a silent per-request skip.
+        if ip.version == net.version and ip in net:
+            return True
+    return False
+
 
 # --------------------------------------------------------------------------
 # HTTP helper (patchable in tests)
@@ -203,6 +338,92 @@ class LinkService:
             "zitadel_userinfo", "https://id.shre.ai/oidc/v1/userinfo")
         self.state = IdentityStore(config.get("state_path", DEFAULT_STATE))
 
+        # Rate limiting is ON by default: this endpoint creates accounts and
+        # mints tokens, so the safe default is limited, and disabling it has
+        # to be a deliberate config act rather than an omission.
+        rl = config.get("rate_limit", {})
+        if rl is None:
+            rl = {}
+        if not isinstance(rl, dict):
+            raise SystemExit("config rate_limit must be an object")
+        # Real JSON booleans only. `"enabled": null` or `0` must not quietly
+        # turn limiting off, and the string "false" must not turn XFF trust
+        # ON (every non-empty string is truthy).
+        for flag, default in (("enabled", True), ("trust_forwarded_for", False)):
+            if flag in rl and not isinstance(rl[flag], bool):
+                raise SystemExit(
+                    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 = 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 "
+                "trusted_proxies list -- trusting the header from any peer "
+                "lets callers choose their own rate-limit key")
+        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(type(x) is int for x in spec)
+                        or spec[1] <= 0):
+                    # A malformed rule must not silently mean "unlimited".
+                    raise SystemExit(
+                        f"config rate_limit.rules[{route!r}] must be "
+                        f"[max_requests, window_seconds] with window > 0")
+                rules[route] = (spec[0], spec[1])
+            self.limiter = RateLimiter(rules)
+            LOG.info("rate limiting active: %s (trust_forwarded_for=%s, "
+                     "trusted_proxies=%s)",
+                     {k: f"{v[0]}/{v[1]}s" for k, v in rules.items()},
+                     self.trust_forwarded_for, list(self.trusted_proxies))
+        else:
+            self.limiter = None
+            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"):
@@ -452,6 +673,29 @@ class Handler(BaseHTTPRequestHandler):
             self._send(404, {"error": "not found"})
 
     def do_POST(self):
+        # Rate-limit BEFORE reading the body or doing any work -- the point is
+        # to spend nothing on an abusive caller. Same close-the-connection
+        # treatment the 413 path uses, for the same reason: we are not going
+        # to drain a body we already decided to reject.
+        limiter = getattr(self.service, "limiter", None)
+        if limiter is not None:
+            who = client_ip(self, self.service.trust_forwarded_for,
+                            self.service.trusted_proxies)
+            allowed, retry_after = limiter.check(self.path, who)
+            if not allowed:
+                LOG.warning("rate limited %s %s (retry after %ss)",
+                            who, self.path, retry_after)
+                self.close_connection = True
+                payload = json.dumps({
+                    "error": "rate limit exceeded",
+                    "retry_after": retry_after}).encode()
+                self.send_response(429)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Retry-After", str(retry_after))
+                self.send_header("Content-Length", str(len(payload)))
+                self.end_headers()
+                self.wfile.write(payload)
+                return
         cl = self.headers.get("Content-Length")
         if cl is None:
             self.close_connection = True

tests/test_server.py

@@ -482,3 +482,372 @@ class TestHealth(HandlerTestBase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestRateLimiterUnit(unittest.TestCase):
+    """The limiter in isolation, on a fake clock -- no sleeping in tests."""
+
+    def setUp(self):
+        self.now = 1000.0
+        self.rl = granthi_link.RateLimiter(
+            rules={"/v1/link": (3, 60)}, clock=lambda: self.now)
+
+    def test_allows_up_to_the_limit_then_denies(self):
+        for i in range(3):
+            allowed, _ = self.rl.check("/v1/link", "1.1.1.1")
+            self.assertTrue(allowed, f"request {i} should pass")
+        allowed, retry = self.rl.check("/v1/link", "1.1.1.1")
+        self.assertFalse(allowed)
+        self.assertGreater(retry, 0)
+        self.assertLessEqual(retry, 61)
+
+    def test_window_slides(self):
+        for _ in range(3):
+            self.rl.check("/v1/link", "1.1.1.1")
+        self.assertFalse(self.rl.check("/v1/link", "1.1.1.1")[0])
+        self.now += 61
+        self.assertTrue(self.rl.check("/v1/link", "1.1.1.1")[0])
+
+    def test_denied_requests_do_not_extend_the_window(self):
+        """A client that keeps hammering must not push its own window
+        forward and lock itself out forever."""
+        for _ in range(3):
+            self.rl.check("/v1/link", "1.1.1.1")
+        for _ in range(20):          # hammer while denied
+            self.now += 1
+            self.assertFalse(self.rl.check("/v1/link", "1.1.1.1")[0])
+        self.now = 1000.0 + 61       # just past the ORIGINAL window
+        self.assertTrue(self.rl.check("/v1/link", "1.1.1.1")[0])
+
+    def test_clients_are_isolated(self):
+        for _ in range(3):
+            self.rl.check("/v1/link", "1.1.1.1")
+        self.assertFalse(self.rl.check("/v1/link", "1.1.1.1")[0])
+        self.assertTrue(self.rl.check("/v1/link", "2.2.2.2")[0])
+
+    def test_routes_are_isolated_and_unknown_routes_pass(self):
+        for _ in range(3):
+            self.rl.check("/v1/link", "1.1.1.1")
+        self.assertFalse(self.rl.check("/v1/link", "1.1.1.1")[0])
+        self.assertTrue(self.rl.check("/v1/repos", "1.1.1.1")[0])
+        for _ in range(50):
+            self.assertTrue(self.rl.check("/health", "1.1.1.1")[0])
+
+    def test_zero_limit_disables_the_endpoint(self):
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (0, 60)},
+                                      clock=lambda: self.now)
+        self.assertFalse(rl.check("/v1/link", "1.1.1.1")[0])
+
+    def test_key_store_stays_bounded(self):
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (5, 60)},
+                                      max_keys=50, clock=lambda: self.now)
+        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["/v1/link"]), 50 + 1)
+
+    def test_expired_keys_are_reclaimed(self):
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (5, 60)},
+                                      max_keys=10, clock=lambda: self.now)
+        for i in range(10):
+            rl.check("/v1/link", f"10.0.0.{i}")
+        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["/v1/link"]), 11)
+
+    def test_concurrent_checks_never_exceed_the_limit(self):
+        """The lock has to actually hold under threads: 40 racing callers
+        against a limit of 10 must yield exactly 10 allows."""
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (10, 60)})
+        results, lock = [], threading.Lock()
+
+        def hit():
+            a, _ = rl.check("/v1/link", "9.9.9.9")
+            with lock:
+                results.append(a)
+
+        ts = [threading.Thread(target=hit) for _ in range(40)]
+        for t in ts:
+            t.start()
+        for t in ts:
+            t.join()
+        self.assertEqual(sum(results), 10)
+
+
+class _FakeHandler:
+    def __init__(self, peer, xff=None):
+        self.client_address = (peer, 12345)
+        self.headers = {} if xff is None else {"X-Forwarded-For": xff}
+        if xff is not None:
+            self.headers = type("H", (), {"get": lambda s, k, d="":
+                                          xff if k == "X-Forwarded-For" else d})()
+
+
+class TestClientIp(unittest.TestCase):
+    # 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")
+        self.assertEqual(granthi_link.client_ip(h, False, self.PROXY), "5.5.5.5")
+
+    def test_trusted_proxy_uses_the_last_hop_not_the_client_supplied_first(self):
+        """A caller can PREPEND anything to X-Forwarded-For; a trusted proxy
+        appends the peer it really saw. Only the last entry is trustworthy."""
+        h = _FakeHandler("10.0.0.1", xff="1.2.3.4, 203.0.113.9")
+        self.assertEqual(granthi_link.client_ip(h, True, self.PROXY),
+                         "203.0.113.9")
+
+    def test_untrusted_peer_cannot_choose_its_own_key(self):
+        """The origin also listens on the tailnet. Anyone reaching it
+        directly must not be able to pick (and rotate) their rate-limit key
+        just by sending a header."""
+        h = _FakeHandler("203.0.113.50", xff="9.9.9.9")
+        self.assertEqual(granthi_link.client_ip(h, True, self.PROXY),
+                         "203.0.113.50")
+
+    def test_non_ip_last_hop_falls_back_to_peer(self):
+        for junk in ("not-an-ip", "', OR 1=1", "x" * 500, "10.0.0.1:8080"):
+            with self.subTest(junk=junk):
+                h = _FakeHandler("10.0.0.1", xff=f"1.2.3.4, {junk}")
+                self.assertEqual(
+                    granthi_link.client_ip(h, True, self.PROXY), "10.0.0.1")
+
+    def test_trusted_proxy_falls_back_when_header_absent(self):
+        h = _FakeHandler("10.0.0.1")
+        self.assertEqual(granthi_link.client_ip(h, True, self.PROXY),
+                         "10.0.0.1")
+
+    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(granthi_link.client_ip(h, True, nets), "10.0.0.1")
+
+
+class TestRateLimitConfig(ServiceTestBase):
+    def _svc(self, rl):
+        cfg = dict(self.svc.cfg)
+        cfg["rate_limit"] = rl
+        return granthi_link.LinkService(cfg)
+
+    def test_enabled_by_default_when_key_absent(self):
+        self.assertIsNotNone(self.svc.limiter)
+
+    def test_explicit_disable_is_honored(self):
+        self.assertIsNone(self._svc({"enabled": False}).limiter)
+
+    def test_custom_rule_overrides_default(self):
+        svc = self._svc({"rules": {"/v1/link": [99, 120]}})
+        self.assertEqual(svc.limiter.rules["/v1/link"], (99, 120))
+
+    def test_malformed_rule_refuses_startup_rather_than_meaning_unlimited(self):
+        for bad in ({"rules": {"/v1/link": [5]}},
+                    {"rules": {"/v1/link": "5/hour"}},
+                    {"rules": {"/v1/link": [5, 0]}},
+                    {"rules": {"/v1/link": [5, -1]}},
+                    {"rules": {"/v1/link": ["5", "60"]}}):
+            with self.subTest(cfg=bad):
+                with self.assertRaises(SystemExit):
+                    self._svc(bad)
+
+
+class TestRateLimitOverHttp(HandlerTestBase):
+    """Proven on the wire: real 429, real Retry-After."""
+
+    def setUp(self):
+        super().setUp()
+        self.svc.limiter = granthi_link.RateLimiter(rules={"/v1/link": (2, 60)})
+
+    def test_third_request_gets_429_with_retry_after(self):
+        body = json.dumps({"zitadel_access_token": "x"}).encode()
+        hdrs = {"Content-Length": str(len(body)),
+                "Content-Type": "application/json"}
+        for _ in range(2):
+            st, _ = self.raw_post("/v1/link", body, hdrs)
+            self.assertNotEqual(st, 429)
+        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+        self.addCleanup(conn.close)
+        conn.request("POST", "/v1/link", body=body, headers=hdrs)
+        resp = conn.getresponse()
+        self.assertEqual(resp.status, 429)
+        self.assertTrue(resp.getheader("Retry-After"))
+        self.assertIn("rate limit", json.loads(resp.read())["error"])
+
+    def test_health_is_never_rate_limited(self):
+        for _ in range(30):
+            conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+            conn.request("GET", "/health")
+            self.assertEqual(conn.getresponse().status, 200)
+            conn.close()
+
+
+class TestRateLimitHardening(unittest.TestCase):
+    """The four codex [P2] findings, each pinned by a test."""
+
+    def test_capacity_fails_closed_instead_of_resetting_a_live_window(self):
+        """Evicting a live window would let an attacker who can mint many
+        distinct keys clear their OWN limit on demand. New keys are refused
+        instead while every window is still live."""
+        now = [1000.0]
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (1, 3600)},
+                                      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"])
+        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.assertLessEqual(len(rl._hits["/v1/link"]), 5)
+
+    def test_capacity_recovers_once_windows_expire(self):
+        now = [1000.0]
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (1, 60)},
+                                      max_keys=3, clock=lambda: now[0])
+        for i in range(3):
+            rl.check("/v1/link", f"10.0.0.{i}")
+        self.assertFalse(rl.check("/v1/link", "10.0.9.9")[0])
+        now[0] += 61
+        self.assertTrue(rl.check("/v1/link", "10.0.9.9")[0])
+
+    def test_hits_stay_chronological_under_thread_contention(self):
+        """retry_after uses hits[0] and reclamation uses v[-1]; both assume
+        the list is ordered. Reading the clock outside the lock let racing
+        threads append out of order."""
+        rl = granthi_link.RateLimiter(rules={"/v1/link": (500, 3600)})
+        ts = [threading.Thread(target=lambda: rl.check("/v1/link", "7.7.7.7"))
+              for _ in range(200)]
+        for t in ts:
+            t.start()
+        for t in ts:
+            t.join()
+        hits = rl._hits["/v1/link"]["7.7.7.7"]
+        self.assertEqual(len(hits), 200)
+        self.assertEqual(hits, sorted(hits), "timestamps out of order")
+
+
+class TestRateLimitConfigTypes(ServiceTestBase):
+    def _svc(self, rl):
+        cfg = dict(self.svc.cfg)
+        cfg["rate_limit"] = rl
+        return granthi_link.LinkService(cfg)
+
+    def test_non_boolean_enabled_refuses_startup(self):
+        """`"enabled": null` or `0` must not quietly mean unlimited."""
+        for bad in (None, 0, "", "false", "no", []):
+            with self.subTest(enabled=bad):
+                with self.assertRaises(SystemExit):
+                    self._svc({"enabled": bad})
+
+    def test_string_false_does_not_enable_forwarded_trust(self):
+        """Every non-empty string is truthy -- "false" used to mean True."""
+        with self.assertRaises(SystemExit):
+            self._svc({"trust_forwarded_for": "false",
+                       "trusted_proxies": ["10.0.0.0/8"]})
+
+    def test_rate_limit_must_be_an_object(self):
+        for bad in ("yes", 5, ["/v1/link"]):
+            with self.subTest(rl=bad):
+                with self.assertRaises(SystemExit):
+                    self._svc(bad)
+
+    def test_null_rate_limit_means_defaults_not_disabled(self):
+        svc = self._svc(None)
+        self.assertIsNotNone(svc.limiter)
+
+    def test_forwarded_trust_without_trusted_proxies_refuses_startup(self):
+        """Trusting the header from ANY peer lets callers choose their own
+        rate-limit key -- that must not be reachable by omission."""
+        with self.assertRaises(SystemExit):
+            self._svc({"trust_forwarded_for": True})
+        with self.assertRaises(SystemExit):
+            self._svc({"trust_forwarded_for": True, "trusted_proxies": []})
+
+    def test_forwarded_trust_with_proxies_is_accepted(self):
+        svc = self._svc({"trust_forwarded_for": True,
+                         "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)