diff --git a/README.md b/README.md index b080ced..e58c5a0 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,38 @@ 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, + "rules": {"/v1/link": [5, 3600], "/v1/repos": [60, 3600]} +} +``` + +* A malformed rule **refuses startup** rather than silently meaning + unlimited; `[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 key store + is capped (`MAX_RATE_KEYS`); at capacity it drops least-recent windows and + logs loudly, which is fail-open, chosen over an unbounded dict that is a + memory DoS. +* `trust_forwarded_for` is **off** by default. Turn it on only 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. + #### Identity binding (`state.json`) `/v1/link` originally bound purely by `preferred_username` / email @@ -226,8 +258,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 diff --git a/server/config.example.json b/server/config.example.json index 6b6b016..40f1fb6 100644 --- a/server/config.example.json +++ b/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]} + } } diff --git a/server/granthi_link.py b/server/granthi_link.py index 83f3157..5ac9560 100644 --- a/server/granthi_link.py +++ b/server/granthi_link.py @@ -61,6 +61,7 @@ import signal import string import sys import threading +import time import urllib.error import urllib.request from datetime import datetime, timezone @@ -79,6 +80,93 @@ 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) + self.max_keys = max_keys + self._clock = clock or time.monotonic + self._hits = {} + 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 + now = self._clock() + key = (route, client) + with self._lock: + hits = [t for t in self._hits.get(key, ()) if now - t < window] + if len(hits) >= limit: + self._hits[key] = hits + return False, max(1, int(window - (now - hits[0])) + 1) + hits.append(now) + self._hits[key] = hits + if len(self._hits) > self.max_keys: + self._prune(now) + return True, 0 + + def _prune(self, now): + """Caller holds the lock. Drop expired keys; if still over the cap, + drop the least-recently-hit. Dropping RESETS those clients' windows, + which is fail-open -- but an unbounded dict is a memory DoS, and a + bounded one is the lesser failure. Say so loudly either way.""" + for k in [k for k, v in self._hits.items() + if not v or now - v[-1] >= self.rules[k[0]][1]]: + del self._hits[k] + if len(self._hits) > self.max_keys: + over = len(self._hits) - self.max_keys + for k in sorted(self._hits, key=lambda k: self._hits[k][-1])[:over]: + del self._hits[k] + LOG.warning("rate limiter at capacity (%d keys): dropped %d " + "least-recent windows; those clients start fresh", + self.max_keys, over) + + +def client_ip(handler, trust_forwarded_for): + """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. A trusted proxy + APPENDS the peer it actually saw, so the LAST entry is the only one we did + not let the client choose -- take that, never the first. Off by default; + only enable when something trustworthy really is in front. + """ + if trust_forwarded_for: + xff = handler.headers.get("X-Forwarded-For", "") + parts = [p.strip() for p in xff.split(",") if p.strip()] + if parts: + return parts[-1] + return handler.client_address[0] + # -------------------------------------------------------------------------- # HTTP helper (patchable in tests) @@ -203,6 +291,31 @@ 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") or {} + self.trust_forwarded_for = bool(rl.get("trust_forwarded_for", False)) + if rl.get("enabled", True): + rules = dict(DEFAULT_RATE_RULES) + for route, spec in (rl.get("rules") or {}).items(): + if (not isinstance(spec, (list, tuple)) or len(spec) != 2 + or not all(isinstance(x, 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)", + {k: f"{v[0]}/{v[1]}s" for k, v in rules.items()}, + self.trust_forwarded_for) + else: + self.limiter = None + LOG.warning("rate limiting DISABLED by config -- /v1/link will " + "mint tokens without any throttle") + 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 +565,28 @@ 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) + 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 diff --git a/tests/test_server.py b/tests/test_server.py index 2582ef7..da3a883 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -482,3 +482,176 @@ 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), 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), 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): + 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), "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), "203.0.113.9") + + def test_trusted_proxy_falls_back_when_header_absent(self): + h = _FakeHandler("10.0.0.1") + self.assertEqual(granthi_link.client_ip(h, True), "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()