"""Unit tests for granthi-link: login derivation, identity binding rules, link/repos flows against a stub HTTP server that plays both Zitadel userinfo and the Gitea API, plus startup/transport hardening.""" import http.client import json import os import sys import tempfile import threading import unittest import urllib.parse import urllib.request from unittest import mock from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "server")) import granthi_link # noqa: E402 class StubUpstream(BaseHTTPRequestHandler): """Plays Zitadel (/oidc/v1/userinfo) and Gitea (everything else). state["users"]: dict login -> email (existing Gitea users) state["hide_once"]: logins whose next GET 404s (simulates a concurrent create racing between the existence check and the create call) """ state = None # dict injected per-test def _json(self, status, obj): payload = json.dumps(obj).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) def do_GET(self): st = self.state if self.path == "/oidc/v1/userinfo": auth = self.headers.get("Authorization", "") if auth == "Bearer good-token": return self._json(200, {"sub": "123", "preferred_username": "Alice.Smith@org.example", "email": "alice@example.com", "name": "Alice"}) return self._json(401, {"error": "invalid token"}) if self.path == "/api/v1/user": # whoami: the forge decides who a token belongs to tok = self.headers.get("Authorization", "").replace("token ", "") login = st["tokens_by_sha"].get(tok) if not login: return self._json(401, {"message": "unauthorized"}) return self._json(200, {"login": login}) if self.path.endswith("/collaborators") and self.path.startswith("/api/v1/repos/"): full = self.path.split("/api/v1/repos/")[1].rsplit("/collaborators", 1)[0] tok = self.headers.get("Authorization", "").replace("token ", "") caller = st["tokens_by_sha"].get(tok) if st["repo_owner"].get(full) != caller: return self._json(403, {"message": "must have admin rights"}) return self._json(200, [{"login": w} for w in st["collab"].get(full, {})]) if self.path.startswith("/api/v1/users/") and not self.path.endswith("/tokens"): login = self.path.rsplit("/", 1)[1] if login in st["hide_once"]: st["hide_once"].discard(login) return self._json(404, {"message": "not found"}) if login in st["users"]: return self._json(200, {"login": login, "email": st["users"][login]}) return self._json(404, {"message": "not found"}) self._json(404, {}) def do_POST(self): st = self.state length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length) or b"{}") if self.path == "/api/v1/admin/users": if body["username"] in st["users"]: return self._json(409, {"message": "user already exists"}) if body.get("email") in st["users"].values(): # real Gitea: emails are unique across accounts return self._json(422, {"message": f"e-mail already in use [email: {body['email']}]"}) st["users"][body["username"]] = body["email"] st["created"].append(body) return self._json(201, {"login": body["username"]}) if self.path.startswith("/api/v1/users/") and self.path.endswith("/tokens"): # must arrive with basic auth + Sudo (the verified 1.27 mechanism) st["token_reqs"].append({ "auth": self.headers.get("Authorization", ""), "sudo": self.headers.get("Sudo", ""), "body": body}) if not self.headers.get("Authorization", "").startswith("Basic "): return self._json(401, {"message": "auth required"}) login = self.path.split("/")[4] # unique per mint, like a real forge -- two users may legitimately # hold the same token NAME st["sha_seq"] = st.get("sha_seq", 0) + 1 sha = f"SHA-{login}-{st['sha_seq']}-{body['name']}" st["tokens_by_sha"][sha] = login st["tokens"].setdefault(login, set()).add(body["name"]) return self._json(201, {"sha1": sha, "name": body["name"]}) if self.path == "/api/v1/user/repos": if body["name"] in st["repos"]: return self._json(409, {"message": "exists"}) st["repos"].add(body["name"]) return self._json(201, {"name": body["name"], "private": body.get("private"), "full_name": f"alice/{body['name']}"}) self._json(404, {}) def do_PUT(self): """Gitea adds a collaborator with PUT, and the admin path uses basic auth because the inviter's session is long gone by then.""" st = self.state length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length) or b"{}") if "/collaborators/" in self.path and self.path.startswith("/api/v1/repos/"): full = self.path.split("/api/v1/repos/")[1].split("/collaborators/")[0] who = self.path.rsplit("/", 1)[1] if full not in st["repo_owner"]: return self._json(404, {"message": "no such repo"}) tok = self.headers.get("Authorization", "") caller = st["tokens_by_sha"].get(tok.replace("token ", "")) if not tok.startswith("Basic ") and st["repo_owner"].get(full) != caller: return self._json(403, {"message": "forbidden"}) st["collab"].setdefault(full, {})[who] = body.get("permission", "write") return self._json(204, {}) self._json(404, {}) def do_DELETE(self): st = self.state if "/collaborators/" in self.path and self.path.startswith("/api/v1/repos/"): full = self.path.split("/api/v1/repos/")[1].split("/collaborators/")[0] who = self.path.rsplit("/", 1)[1] tok = self.headers.get("Authorization", "").replace("token ", "") caller = st["tokens_by_sha"].get(tok) if st["repo_owner"].get(full) != caller: return self._json(403, {"message": "forbidden"}) st["collab"].get(full, {}).pop(who, None) return self._json(204, {}) if self.path.startswith("/api/v1/users/") and "/tokens/" in self.path: if not self.headers.get("Authorization", "").startswith("Basic "): # matches real Gitea: token auth cannot delete tokens return self._json(403, {"message": "basic auth required"}) login, name = self.path.split("/tokens/") login = login.rsplit("/", 1)[1] name = urllib.parse.unquote(name) if st.get("revoke_fails"): return self._json(500, {"message": "forge exploded"}) if name not in st["tokens"].get(login, set()): return self._json(404, {"message": "not found"}) st["tokens"][login].discard(name) for sha, owner in list(st["tokens_by_sha"].items()): if owner == login and sha.endswith(f"-{name}"): del st["tokens_by_sha"][sha] return self._json(204, {}) self._json(404, {}) def log_message(self, *a): pass class ServiceTestBase(unittest.TestCase): def setUp(self): StubUpstream.state = {"users": {}, "created": [], "repos": set(), "token_reqs": [], "hide_once": set(), "tokens": {}, "tokens_by_sha": {}, "revoke_fails": False, "repo_owner": {}, "collab": {}} self.upstream = ThreadingHTTPServer(("127.0.0.1", 0), StubUpstream) threading.Thread(target=self.upstream.serve_forever, daemon=True).start() self.addCleanup(self.upstream.shutdown) base = f"http://127.0.0.1:{self.upstream.server_address[1]}" self.state_dir = tempfile.mkdtemp(prefix="granthi-link-state-") self.state_path = os.path.join(self.state_dir, "state.json") self.svc = granthi_link.LinkService({ "gitea_base": base, "public_gitea_base": "http://public.example:3041", "zitadel_userinfo": f"{base}/oidc/v1/userinfo", "admin_token": "ADMTOK", "admin_login": "root", "admin_password": "rootpw", "test_mode": False, "state_path": self.state_path, "audit_path": os.path.join(self.state_dir, "audit.jsonl"), }) def enable_test_mode(self): self.svc.cfg["test_mode"] = True patcher = mock.patch.dict( os.environ, {granthi_link.TEST_MODE_ENV: "1"}) patcher.start() self.addCleanup(patcher.stop) def stub_link(self, sub, username, email=None, verified=None, device="d", device_id=None, client_ip=None): ui = {"sub": sub, "preferred_username": username} if email is not None: ui["email"] = email if verified is not None: ui["email_verified"] = verified body = {"test_userinfo": ui, "device_name": device} if device_id: body["device_id"] = device_id return self.svc.link(body, client_ip) def read_state(self): with open(self.state_path) as f: return json.load(f) class TestDeriveLogin(unittest.TestCase): def test_strips_domain_and_sanitizes(self): self.assertEqual( granthi_link.LinkService.derive_login( {"preferred_username": "Alice.Smith@org.example"}), "alice.smith") def test_email_fallback(self): self.assertEqual( granthi_link.LinkService.derive_login({"email": "Bob+x@e.com"}), "bob-x") def test_empty_returns_none(self): self.assertIsNone(granthi_link.LinkService.derive_login({})) class TestLink(ServiceTestBase): def test_link_creates_user_and_mints_token(self): status, resp = self.svc.link({"zitadel_access_token": "good-token", "device_name": "mac studio"}) self.assertEqual(status, 200) self.assertEqual(resp["login"], "alice.smith") # the stub now issues a per-token sha so revocation can be tested self.assertIn("granthi-sync-", resp["token"]) self.assertEqual(resp["gitea_base"], "http://public.example:3041") st = StubUpstream.state self.assertEqual(len(st["created"]), 1) created = st["created"][0] self.assertFalse(created["must_change_password"]) self.assertEqual(created["visibility"], "private") self.assertGreaterEqual(len(created["password"]), 30) req = st["token_reqs"][0] self.assertTrue(req["auth"].startswith("Basic ")) self.assertEqual(req["sudo"], "alice.smith") self.assertEqual(sorted(req["body"]["scopes"]), ["write:repository", "write:user"]) def test_link_records_identity_mapping(self): self.svc.link({"zitadel_access_token": "good-token", "device_name": "d"}) state = self.read_state() rec = state["identities"]["123"] self.assertEqual(rec["login"], "alice.smith") self.assertTrue(rec["created_by_service"]) mode = os.stat(self.state_path).st_mode & 0o777 self.assertEqual(mode, 0o600) def test_link_bad_token_401(self): status, resp = self.svc.link({"zitadel_access_token": "bad", "device_name": "d"}) self.assertEqual(status, 401) def test_link_missing_token_400(self): status, _ = self.svc.link({"device_name": "d"}) self.assertEqual(status, 400) def test_link_missing_sub_422(self): self.enable_test_mode() status, _ = self.svc.link({"test_userinfo": {"preferred_username": "nosub"}, "device_name": "d"}) self.assertEqual(status, 422) class TestIdentityBinding(ServiceTestBase): """Finding 1: account takeover by login collision.""" def setUp(self): super().setUp() self.enable_test_mode() def test_repeat_link_same_sub_reuses_mapping(self): status, resp = self.stub_link("s1", "alice") self.assertEqual(status, 200) # same sub again -- even with a different preferred_username the # mapping wins and no second user is created status, resp = self.stub_link("s1", "totally-different") self.assertEqual(status, 200) self.assertEqual(resp["login"], "alice") self.assertEqual(len(StubUpstream.state["created"]), 1) def test_colliding_username_different_sub_409_no_token(self): status, _ = self.stub_link("s1", "alice") self.assertEqual(status, 200) minted_before = len(StubUpstream.state["token_reqs"]) status, resp = self.stub_link("s2", "alice") # attacker self.assertEqual(status, 409) self.assertIn("not linked to this identity", resp["error"]) # no token minted for the refused identity self.assertEqual(len(StubUpstream.state["token_reqs"]), minted_before) self.assertNotIn("s2", self.read_state()["identities"]) def test_existing_user_binds_on_verified_email_match(self): StubUpstream.state["users"]["bob"] = "bob@example.com" status, resp = self.stub_link("s9", "bob", email="bob@example.com", verified=True) self.assertEqual(status, 200) self.assertEqual(resp["login"], "bob") rec = self.read_state()["identities"]["s9"] self.assertFalse(rec["created_by_service"]) def test_existing_user_unverified_email_409(self): StubUpstream.state["users"]["bob"] = "bob@example.com" status, _ = self.stub_link("s9", "bob", email="bob@example.com", verified=False) self.assertEqual(status, 409) self.assertEqual(StubUpstream.state["token_reqs"], []) def test_existing_user_wrong_email_409(self): StubUpstream.state["users"]["bob"] = "bob@example.com" status, _ = self.stub_link("s9", "bob", email="evil@example.com", verified=True) self.assertEqual(status, 409) self.assertEqual(StubUpstream.state["token_reqs"], []) def test_deleted_service_created_login_is_recreated(self): self.stub_link("s1", "alice") del StubUpstream.state["users"]["alice"] # user deleted in Gitea status, resp = self.stub_link("s1", "alice") self.assertEqual(status, 200) self.assertEqual(resp["login"], "alice") self.assertIn("alice", StubUpstream.state["users"]) def test_deleted_adopted_login_is_refused(self): StubUpstream.state["users"]["bob"] = "bob@example.com" status, _ = self.stub_link("s9", "bob", email="bob@example.com", verified=True) self.assertEqual(status, 200) del StubUpstream.state["users"]["bob"] status, resp = self.stub_link("s9", "bob", email="bob@example.com", verified=True) self.assertEqual(status, 409) self.assertIn("not created by this service", resp["error"]) def test_corrupt_state_fails_closed(self): with open(self.state_path, "w") as f: f.write("{ not json") status, resp = self.svc.link({"test_userinfo": {"sub": "s1", "preferred_username": "alice"}, "device_name": "d"}) self.assertEqual(status, 500) self.assertEqual(StubUpstream.state["token_reqs"], []) class TestConcurrentCreateRace(ServiceTestBase): """Finding 7: Gitea 409 on user create is handled idempotently.""" def setUp(self): super().setUp() self.enable_test_mode() def test_409_on_create_refetches_and_continues(self): # user exists (created by a concurrent request with OUR email) but # the first existence check misses it email = "alice@example.com" StubUpstream.state["users"]["alice"] = email StubUpstream.state["hide_once"].add("alice") status, resp = self.stub_link("s1", "alice", email=email) self.assertEqual(status, 200) self.assertEqual(resp["login"], "alice") self.assertEqual(self.read_state()["identities"]["s1"]["login"], "alice") def test_409_on_create_with_foreign_email_refused(self): StubUpstream.state["users"]["alice"] = "someoneelse@example.com" StubUpstream.state["hide_once"].add("alice") status, resp = self.stub_link("s1", "alice", email="alice@example.com") self.assertEqual(status, 409) self.assertEqual(StubUpstream.state["token_reqs"], []) class TestTestModeGate(ServiceTestBase): """Finding 2: test_mode requires the env gate.""" def test_config_flag_alone_is_ignored(self): self.svc.cfg["test_mode"] = True env = {k: v for k, v in os.environ.items() if k != granthi_link.TEST_MODE_ENV} with mock.patch.dict(os.environ, env, clear=True), \ self.assertLogs("granthi-link", level="ERROR"): status, _ = self.svc.link({"test_userinfo": { "sub": "1", "preferred_username": "x"}}) self.assertEqual(status, 400) # falls through to token-required def test_env_gate_wrong_value_is_ignored(self): self.svc.cfg["test_mode"] = True with mock.patch.dict(os.environ, {granthi_link.TEST_MODE_ENV: "true"}): status, _ = self.svc.link({"test_userinfo": { "sub": "1", "preferred_username": "x"}}) self.assertEqual(status, 400) def test_enabled_with_config_and_env(self): self.enable_test_mode() status, resp = self.stub_link("1", "evetest") self.assertEqual(status, 200) self.assertEqual(resp["login"], "evetest") def test_env_alone_without_config_flag_disabled(self): with mock.patch.dict(os.environ, {granthi_link.TEST_MODE_ENV: "1"}): status, _ = self.svc.link({"test_userinfo": { "sub": "1", "preferred_username": "x"}}) self.assertEqual(status, 400) class TestConfigPerms(unittest.TestCase): """Finding 3: refuse startup on permissive or foreign-owned config.""" def setUp(self): self.tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".json") self.tmp.write(b"{}") self.tmp.close() self.addCleanup(os.unlink, self.tmp.name) def test_0600_ok(self): os.chmod(self.tmp.name, 0o600) self.assertIsNone(granthi_link.check_config_perms(self.tmp.name)) def test_0400_ok(self): os.chmod(self.tmp.name, 0o400) self.assertIsNone(granthi_link.check_config_perms(self.tmp.name)) def test_0644_refused(self): os.chmod(self.tmp.name, 0o644) err = granthi_link.check_config_perms(self.tmp.name) self.assertIn("refusing to start", err) self.assertIn("0o644", err) def test_0640_refused(self): os.chmod(self.tmp.name, 0o640) self.assertIsNotNone(granthi_link.check_config_perms(self.tmp.name)) def test_foreign_owner_refused(self): os.chmod(self.tmp.name, 0o600) not_me = os.geteuid() + 1 err = granthi_link.check_config_perms(self.tmp.name, euid=not_me) self.assertIn("owned by uid", err) def test_main_exits_2_on_permissive_config(self): """The refusal must actually stop startup (exit nonzero), not just return a string -- verified through main().""" with open(self.tmp.name, "w") as f: json.dump({"gitea_base": "http://x", "admin_token": "t", "admin_login": "r", "admin_password": "p"}, f) os.chmod(self.tmp.name, 0o644) with mock.patch.object(granthi_link.sys, "argv", ["granthi_link.py", self.tmp.name]), \ mock.patch.object(granthi_link, "serve") as served, \ self.assertLogs("granthi-link", level="ERROR"): with self.assertRaises(SystemExit) as cm: granthi_link.main() self.assertEqual(cm.exception.code, 2) served.assert_not_called() # never reached serve() class HandlerTestBase(ServiceTestBase): def setUp(self): super().setUp() granthi_link.Handler.service = self.svc self.srv = ThreadingHTTPServer(("127.0.0.1", 0), granthi_link.Handler) threading.Thread(target=self.srv.serve_forever, daemon=True).start() self.addCleanup(self.srv.shutdown) self.port = self.srv.server_address[1] def raw_post(self, path, body_bytes=None, headers=None): conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) self.addCleanup(conn.close) conn.putrequest("POST", path) for k, v in (headers or {}).items(): conn.putheader(k, v) conn.endheaders() if body_bytes: conn.send(body_bytes) resp = conn.getresponse() return resp.status, json.loads(resp.read() or b"{}") class TestBodyLimits(HandlerTestBase): """Finding 6: bounded reads, Content-Length required on POST.""" def test_oversized_content_length_413(self): status, resp = self.raw_post( "/v1/link", headers={"Content-Length": str(granthi_link.MAX_BODY_BYTES + 1)}) self.assertEqual(status, 413) self.assertIn("too large", resp["error"]) def test_oversized_real_body_413(self): """Send an actual over-limit payload on the wire (not just the header), so the response is proven, not the predicate alone.""" body = json.dumps( {"pad": "x" * (granthi_link.MAX_BODY_BYTES + 2000)}).encode() self.assertGreater(len(body), granthi_link.MAX_BODY_BYTES) status, resp = self.raw_post( "/v1/link", body_bytes=body, headers={"Content-Length": str(len(body)), "Content-Type": "application/json"}) self.assertEqual(status, 413) def test_missing_content_length_411(self): status, _ = self.raw_post("/v1/link") self.assertEqual(status, 411) def test_invalid_content_length_400(self): status, _ = self.raw_post("/v1/link", headers={"Content-Length": "banana"}) self.assertEqual(status, 400) def test_normal_post_still_works(self): body = json.dumps({"device_name": "d"}).encode() status, resp = self.raw_post( "/v1/link", body_bytes=body, headers={"Content-Length": str(len(body)), "Content-Type": "application/json"}) self.assertEqual(status, 400) # missing token, but parsed fine self.assertIn("zitadel_access_token", resp["error"]) def test_at_limit_accepted(self): pad = "x" * (granthi_link.MAX_BODY_BYTES - 30) body = json.dumps({"pad": pad}).encode() self.assertLessEqual(len(body), granthi_link.MAX_BODY_BYTES) status, _ = self.raw_post( "/v1/link", body_bytes=body, headers={"Content-Length": str(len(body))}) self.assertEqual(status, 400) # parsed; fails on missing token class TestRepos(ServiceTestBase): def test_repo_create_returns_public_clone_url(self): status, resp = self.svc.repos({"token": "USERTOK", "name": "notes", "private": True}) self.assertEqual(status, 200) self.assertEqual(resp["clone_url"], "http://public.example:3041/alice/notes.git") def test_repo_conflict_409(self): self.svc.repos({"token": "T", "name": "notes"}) status, _ = self.svc.repos({"token": "T", "name": "notes"}) self.assertEqual(status, 409) def test_missing_fields_400(self): status, _ = self.svc.repos({"name": "x"}) self.assertEqual(status, 400) class TestHealth(HandlerTestBase): def test_health_endpoint(self): with urllib.request.urlopen( f"http://127.0.0.1:{self.port}/health") as r: body = json.loads(r.read()) self.assertEqual(body["status"], "ok") self.assertEqual(body["service"], "granthi-link") class TestDeviceRegistry(ServiceTestBase): """Which computers can reach my files, and can I cut one off.""" def setUp(self): super().setUp() self.enable_test_mode() def link_device(self, device_id, name="laptop"): status, resp = self.stub_link("s1", "alice", device=name, device_id=device_id) self.assertEqual(status, 200, resp) return resp def test_devices_are_recorded_and_listed_per_account(self): self.link_device("dev-a", "work-laptop") second = self.link_device("dev-b", "home-mac") status, resp = self.svc.devices({"token": second["token"]}) self.assertEqual(status, 200) self.assertEqual(resp["login"], "alice") ids = [d["device_id"] for d in resp["devices"]] self.assertEqual(sorted(ids), ["dev-a", "dev-b"]) names = {d["device_id"]: d["name"] for d in resp["devices"]} self.assertEqual(names["dev-a"], "work-laptop") def test_listing_needs_a_working_token(self): self.link_device("dev-a") status, _ = self.svc.devices({"token": "not-a-real-token"}) self.assertEqual(status, 401) status, _ = self.svc.devices({}) self.assertEqual(status, 401) def test_login_is_never_taken_from_the_request_body(self): """Authorisation comes from the forge's answer about the token, so a body claiming another account changes nothing.""" first = self.link_device("dev-a") status, resp = self.svc.devices({"token": first["token"], "login": "somebody-else"}) self.assertEqual(resp["login"], "alice") def test_relinking_the_same_device_does_not_duplicate_it(self): self.link_device("dev-a", "laptop") self.link_device("dev-a", "laptop-renamed") status, resp = self.svc.devices({"token": self.link_device("dev-a")["token"]}) self.assertEqual(len(resp["devices"]), 1) def test_old_clients_without_a_device_id_still_register(self): resp = self.stub_link("s1", "alice", device="ancient")[1] self.assertEqual(resp["device_id"], resp["token_name"]) status, listed = self.svc.devices({"token": resp["token"]}) self.assertEqual([d["device_id"] for d in listed["devices"]], [resp["token_name"]]) class TestRevoke(ServiceTestBase): def setUp(self): super().setUp() self.enable_test_mode() self.a = self.stub_link("s1", "alice", device="laptop-a", device_id="dev-a")[1] self.b = self.stub_link("s1", "alice", device="laptop-b", device_id="dev-b")[1] def test_revoking_a_device_kills_its_token_at_the_forge(self): status, resp = self.svc.revoke_device({"token": self.b["token"], "device_id": "dev-a"}) self.assertEqual(status, 200, resp) # the revoked device's token no longer authenticates ANYTHING self.assertEqual(self.svc.whoami(self.a["token"]), None) # the device that did the revoking still works self.assertEqual(self.svc.whoami(self.b["token"]), "alice") def test_a_device_can_revoke_itself(self): status, _ = self.svc.revoke_device({"token": self.a["token"], "device_id": "dev-a"}) self.assertEqual(status, 200) self.assertEqual(self.svc.whoami(self.a["token"]), None) def test_revoked_device_is_marked_not_deleted(self): self.svc.revoke_device({"token": self.b["token"], "device_id": "dev-a"}) status, resp = self.svc.devices({"token": self.b["token"]}) by_id = {d["device_id"]: d for d in resp["devices"]} self.assertIsNotNone(by_id["dev-a"]["revoked_at"]) self.assertIsNone(by_id["dev-b"]["revoked_at"]) def test_cannot_revoke_a_device_on_another_account(self): other = self.stub_link("s2", "mallory", device="theirs", device_id="dev-x")[1] status, resp = self.svc.revoke_device({"token": other["token"], "device_id": "dev-a"}) self.assertEqual(status, 404) # alice's device is untouched self.assertEqual(self.svc.whoami(self.a["token"]), "alice") def test_a_failed_forge_delete_is_not_recorded_as_revoked(self): """A registry that says 'revoked' while the token still works is worse than an honest error.""" StubUpstream.state["revoke_fails"] = True status, _ = self.svc.revoke_device({"token": self.b["token"], "device_id": "dev-a"}) self.assertEqual(status, 502) StubUpstream.state["revoke_fails"] = False _, listed = self.svc.devices({"token": self.b["token"]}) by_id = {d["device_id"]: d for d in listed["devices"]} self.assertIsNone(by_id["dev-a"]["revoked_at"]) self.assertEqual(self.svc.whoami(self.a["token"]), "alice") def test_revoking_twice_is_not_an_error(self): self.svc.revoke_device({"token": self.b["token"], "device_id": "dev-a"}) status, _ = self.svc.revoke_device({"token": self.b["token"], "device_id": "dev-a"}) self.assertEqual(status, 200) # forge 404 = already gone = success class TestAuditLog(ServiceTestBase): def setUp(self): super().setUp() self.enable_test_mode() def test_link_and_revoke_are_recorded_with_who_and_where(self): first = self.stub_link("s1", "alice", device="laptop", device_id="dev-a", client_ip="203.0.113.9")[1] self.stub_link("s1", "alice", device="mac", device_id="dev-b") second = self.svc.devices({"token": first["token"]})[1] self.assertEqual(len(second["devices"]), 2) self.svc.revoke_device({"token": first["token"], "device_id": "dev-b"}, client_ip="198.51.100.4") status, resp = self.svc.audit_read({"token": first["token"]}) self.assertEqual(status, 200) events = resp["events"] kinds = [e["event"] for e in events] self.assertIn("device.link", kinds) self.assertIn("device.revoke", kinds) link_ev = [e for e in events if e["event"] == "device.link" and e["device_id"] == "dev-a"][0] self.assertEqual(link_ev["client_ip"], "203.0.113.9") self.assertEqual(link_ev["login"], "alice") revoke_ev = [e for e in events if e["event"] == "device.revoke"][0] self.assertEqual(revoke_ev["client_ip"], "198.51.100.4") self.assertEqual(revoke_ev["device_id"], "dev-b") def test_events_are_newest_first_and_scoped_to_the_caller(self): mine = self.stub_link("s1", "alice", device_id="dev-a")[1] self.stub_link("s2", "mallory", device_id="dev-x") _, resp = self.svc.audit_read({"token": mine["token"]}) self.assertTrue(resp["events"]) self.assertTrue(all(e["login"] == "alice" for e in resp["events"])) stamps = [e["ts"] for e in resp["events"]] self.assertEqual(stamps, sorted(stamps, reverse=True)) def test_reading_the_log_needs_a_working_token(self): self.stub_link("s1", "alice", device_id="dev-a") status, _ = self.svc.audit_read({"token": "nope"}) self.assertEqual(status, 401) def test_a_refused_revoke_is_recorded_too(self): self.stub_link("s1", "alice", device_id="dev-a") other = self.stub_link("s2", "mallory", device_id="dev-x")[1] self.svc.revoke_device({"token": other["token"], "device_id": "dev-a"}, client_ip="192.0.2.7") _, resp = self.svc.audit_read({"token": other["token"]}) denied = [e for e in resp["events"] if e["event"] == "device.revoke.denied"] self.assertEqual(len(denied), 1) self.assertEqual(denied[0]["client_ip"], "192.0.2.7") def test_the_log_says_what_it_cannot_see(self): """Reading this and believing it lists file activity would be a real mistake: git traffic never passes through this service.""" mine = self.stub_link("s1", "alice", device_id="dev-a")[1] _, resp = self.svc.audit_read({"token": mine["token"]}) self.assertIn("pushes and pulls", resp["note"]) def test_a_broken_audit_file_does_not_break_the_request(self): self.svc.audit.path = "/nonexistent-dir/audit.jsonl" status, _ = self.stub_link("s1", "alice", device_id="dev-a") self.assertEqual(status, 200) # linking still works def test_rotation_keeps_the_previous_file_readable(self): self.svc.audit.keep_bytes = 1500 # one rotation, not many mine = self.stub_link("s1", "alice", device_id="dev-a")[1] for i in range(20): self.svc.audit.write("noise", login="alice", n=i) _, resp = self.svc.audit_read({"token": mine["token"]}, ) self.assertGreater(len(resp["events"]), 15) # spans both files self.assertTrue(os.path.exists(self.svc.audit.path + ".1")) class TestTokenNameUniqueness(ServiceTestBase): def setUp(self): super().setUp() self.enable_test_mode() def test_two_devices_named_the_same_get_different_token_names(self): """Revocation deletes by token name, so a collision would mean signing out one laptop kills the other.""" a = self.stub_link("s1", "alice", device="macbook", device_id="dev-a")[1] b = self.stub_link("s1", "alice", device="macbook", device_id="dev-b")[1] self.assertNotEqual(a["token_name"], b["token_name"]) self.svc.revoke_device({"token": b["token"], "device_id": "dev-a"}) self.assertIsNone(self.svc.whoami(a["token"])) self.assertEqual(self.svc.whoami(b["token"]), "alice") class TestDeviceEndpointsOverHttp(HandlerTestBase): """Through the real handler, not the service method. The first live run of the revoke endpoint returned 429 on every call while every unit test passed: the tests called svc.revoke_device() directly, so nothing ever went through the rate limiter. Any route whose limit is a policy decision needs at least one test that speaks HTTP. """ def setUp(self): super().setUp() self.enable_test_mode() self.a = self.stub_link("s1", "alice", device="laptop", device_id="dev-a")[1] self.b = self.stub_link("s1", "alice", device="desktop", device_id="dev-b")[1] def post(self, path, obj): body = json.dumps(obj).encode() return self.raw_post(path, body, {"Content-Type": "application/json", "Content-Length": str(len(body))}) def test_revoke_is_reachable_and_not_rate_limited_away(self): status, resp = self.post("/v1/devices/revoke", {"token": self.b["token"], "device_id": "dev-a"}) self.assertEqual(status, 200, resp) self.assertIsNone(self.svc.whoami(self.a["token"])) def test_repeated_revokes_keep_working(self): """A person signing several lost machines out in one sitting must not be locked out partway through.""" for i in range(12): self.stub_link("s1", "alice", device=f"d{i}", device_id=f"gone-{i}") for i in range(12): status, resp = self.post("/v1/devices/revoke", {"token": self.b["token"], "device_id": f"gone-{i}"}) self.assertEqual(status, 200, f"revoke {i}: {resp}") def test_devices_and_audit_are_reachable_over_http(self): status, resp = self.post("/v1/devices", {"token": self.a["token"]}) self.assertEqual(status, 200) self.assertEqual(len(resp["devices"]), 2) status, resp = self.post("/v1/audit", {"token": self.a["token"]}) self.assertEqual(status, 200) self.assertTrue(resp["events"]) def test_no_route_a_client_uses_is_configured_to_zero(self): """0 means DISABLED in this limiter, so a zero on a live route is a dead endpoint, not an unlimited one.""" for route, (limit, _window) in granthi_link.DEFAULT_RATE_RULES.items(): self.assertGreater(limit, 0, f"{route} is disabled by its rule") class TestSharing(ServiceTestBase): """Give a person access to one repo -- and to nothing else.""" def setUp(self): super().setUp() self.enable_test_mode() self.alice = self.stub_link("s1", "alice", email="alice@x.test", verified=True, device_id="dev-a")[1] self.bob = self.stub_link("s2", "bob", email="bob@x.test", verified=True, device_id="dev-b")[1] StubUpstream.state["repo_owner"]["alice/notes"] = "alice" StubUpstream.state["repo_owner"]["bob/private"] = "bob" def test_owner_can_share_and_unshare(self): status, resp = self.svc.grants({"token": self.alice["token"], "repo": "notes", "login": "bob"}) self.assertEqual(status, 200, resp) self.assertIn("bob", StubUpstream.state["collab"]["alice/notes"]) status, _ = self.svc.grants({"token": self.alice["token"], "action": "remove", "repo": "notes", "login": "bob"}) self.assertEqual(status, 200) self.assertNotIn("bob", StubUpstream.state["collab"]["alice/notes"]) def test_a_non_owner_cannot_share_someone_elses_repo(self): """The forge decides who may share, not this service.""" status, resp = self.svc.grants({"token": self.bob["token"], "repo": "alice/notes", "login": "bob"}) self.assertEqual(status, 403) self.assertEqual(StubUpstream.state["collab"].get("alice/notes", {}), {}) def test_permission_is_validated(self): status, _ = self.svc.grants({"token": self.alice["token"], "repo": "notes", "login": "bob", "permission": "owner"}) self.assertEqual(status, 400) def test_listing_shows_who_has_access(self): self.svc.grants({"token": self.alice["token"], "repo": "notes", "login": "bob"}) status, resp = self.svc.grants({"token": self.alice["token"], "action": "list", "repo": "notes"}) self.assertEqual((status, resp["collaborators"]), (200, ["bob"])) def test_removing_someone_who_never_had_access_is_not_an_error(self): status, _ = self.svc.grants({"token": self.alice["token"], "action": "remove", "repo": "notes", "login": "nobody"}) self.assertEqual(status, 200) class TestInvites(ServiceTestBase): """A promise of access for someone with no account yet.""" def setUp(self): super().setUp() self.enable_test_mode() self.alice = self.stub_link("s1", "alice", email="alice@x.test", verified=True, device_id="dev-a")[1] StubUpstream.state["repo_owner"]["alice/notes"] = "alice" def invite(self, email, repos=("notes",), token=None): return self.svc.invite({"token": token or self.alice["token"], "email": email, "repos": list(repos)}) def test_invite_creates_nothing_until_they_sign_in(self): status, resp = self.invite("carol@x.test") self.assertEqual(status, 200, resp) # no account, no collaborator row yet self.assertNotIn("carol", StubUpstream.state["users"]) self.assertEqual(StubUpstream.state["collab"].get("alice/notes", {}), {}) def test_access_lands_on_first_link_with_a_verified_email(self): self.invite("carol@x.test") status, resp = self.stub_link("s9", "carol", email="carol@x.test", verified=True, device_id="dev-c") self.assertEqual(status, 200, resp) self.assertEqual(resp.get("granted_repos"), ["alice/notes"]) self.assertIn("carol", StubUpstream.state["collab"]["alice/notes"]) def test_an_unverified_email_collects_nothing(self): """The address is the only thing tying the promise to this person.""" self.invite("dave@x.test") status, resp = self.stub_link("s10", "dave", email="dave@x.test", verified=False, device_id="dev-d") self.assertEqual(status, 200) self.assertIsNone(resp.get("granted_repos")) self.assertEqual(StubUpstream.state["collab"].get("alice/notes", {}), {}) # and the invite is still waiting, not consumed self.assertTrue(self.svc.state.peek_invites("dave@x.test")) def test_an_invite_is_applied_once(self): self.invite("erin@x.test") self.stub_link("s11", "erin", email="erin@x.test", verified=True, device_id="dev-e") status, resp = self.stub_link("s11", "erin", email="erin@x.test", verified=True, device_id="dev-e2") self.assertIsNone(resp.get("granted_repos")) def test_cannot_invite_to_a_repo_you_do_not_administer(self): StubUpstream.state["repo_owner"]["bob/secret"] = "bob" status, _ = self.invite("mallory@x.test", repos=("bob/secret",)) self.assertEqual(status, 403) self.assertEqual(self.svc.state.peek_invites("mallory@x.test"), []) def test_reinviting_updates_the_permission_instead_of_stacking(self): self.svc.invite({"token": self.alice["token"], "email": "f@x.test", "repos": [{"name": "notes", "permission": "read"}]}) self.svc.invite({"token": self.alice["token"], "email": "f@x.test", "repos": [{"name": "notes", "permission": "write"}]}) pending = self.svc.state.peek_invites("f@x.test") self.assertEqual(len(pending), 1) self.assertEqual(pending[0]["permission"], "write") def test_a_bad_email_is_refused(self): status, _ = self.svc.invite({"token": self.alice["token"], "email": "not-an-email", "repos": ["notes"]}) self.assertEqual(status, 400) def test_a_failed_grant_does_not_block_the_person_signing_in(self): """Someone must not be locked out because a repo they were promised has since been deleted.""" self.invite("gina@x.test", repos=("notes",)) StubUpstream.state["repo_owner"].pop("alice/notes") status, resp = self.stub_link("s12", "gina", email="gina@x.test", verified=True, device_id="dev-g") self.assertEqual(status, 200) self.assertIn("token", resp) def test_invite_and_application_are_both_audited(self): self.invite("hana@x.test") self.stub_link("s13", "hana", email="hana@x.test", verified=True, device_id="dev-h") _, mine = self.svc.audit_read({"token": self.alice["token"]}) self.assertIn("invite", [e["event"] for e in mine["events"]]) class TestInviteSurvivesAFailedGrant(ServiceTestBase): """[P2, codex] A transient forge error must not destroy the promise.""" def setUp(self): super().setUp() self.enable_test_mode() self.alice = self.stub_link("s1", "alice", email="alice@x.test", verified=True, device_id="dev-a")[1] StubUpstream.state["repo_owner"]["alice/notes"] = "alice" StubUpstream.state["repo_owner"]["alice/reports"] = "alice" def test_a_failed_grant_leaves_the_invite_pending_for_next_time(self): self.svc.invite({"token": self.alice["token"], "email": "carol@x.test", "repos": ["notes"]}) # the forge loses the repo mid-flight -> the PUT 404s StubUpstream.state["repo_owner"].pop("alice/notes") status, resp = self.stub_link("s9", "carol", email="carol@x.test", verified=True, device_id="dev-c") self.assertEqual(status, 200) # sign-in still succeeds self.assertIsNone(resp.get("granted_repos")) # the promise is STILL THERE rather than silently consumed self.assertEqual([g["repo"] for g in self.svc.state.peek_invites("carol@x.test")], ["alice/notes"]) # and it lands on the next link, once the repo is back StubUpstream.state["repo_owner"]["alice/notes"] = "alice" status, resp = self.stub_link("s9", "carol", email="carol@x.test", verified=True, device_id="dev-c2") self.assertEqual(resp.get("granted_repos"), ["alice/notes"]) self.assertEqual(self.svc.state.peek_invites("carol@x.test"), []) def test_a_partial_failure_only_consumes_what_landed(self): self.svc.invite({"token": self.alice["token"], "email": "dan@x.test", "repos": ["notes", "reports"]}) StubUpstream.state["repo_owner"].pop("alice/reports") # one of two fails _, resp = self.stub_link("s10", "dan", email="dan@x.test", verified=True, device_id="dev-d") self.assertEqual(resp.get("granted_repos"), ["alice/notes"]) self.assertEqual([g["repo"] for g in self.svc.state.peek_invites("dan@x.test")], ["alice/reports"]) class TestDuplicateEmailIsExplained(ServiceTestBase): """A 502 sends someone looking for an outage. The real answer is that the address already belongs to another account here -- say so. (Hit live on 2026-08-23 when an operator moved an email onto a different account.)""" def setUp(self): super().setUp() self.enable_test_mode() # an existing account already holds the address StubUpstream.state["users"]["existing"] = "taken@x.test" def test_it_is_a_409_that_names_the_problem(self): status, resp = self.stub_link("s-new", "brandnew", email="taken@x.test", verified=True, device_id="dev-x") self.assertEqual(status, 409, resp) msg = resp["error"] self.assertIn("taken@x.test", msg) self.assertIn("already belongs to a different account", msg) self.assertIn("brandnew", msg) # names the login it tried self.assertNotIn("502", msg) def test_a_normal_create_is_unaffected(self): status, resp = self.stub_link("s-ok", "fresh", email="fresh@x.test", verified=True, device_id="dev-y") self.assertEqual(status, 200, resp) self.assertIn("token", resp) 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)