granthi-sync/tests/test_server.py

"""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":
                                        "[email protected]", "email":
                                        "[email protected]", "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"})
            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": "[email protected]"}),
            "alice.smith")

    def test_email_fallback(self):
        self.assertEqual(
            granthi_link.LinkService.derive_login({"email": "[email protected]"}),
            "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"] = "[email protected]"
        status, resp = self.stub_link("s9", "bob", email="[email protected]",
                                      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"] = "[email protected]"
        status, _ = self.stub_link("s9", "bob", email="[email protected]",
                                   verified=False)
        self.assertEqual(status, 409)
        self.assertEqual(StubUpstream.state["token_reqs"], [])

    def test_existing_user_wrong_email_409(self):
        StubUpstream.state["users"]["bob"] = "[email protected]"
        status, _ = self.stub_link("s9", "bob", email="[email protected]",
                                   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"] = "[email protected]"
        status, _ = self.stub_link("s9", "bob", email="[email protected]",
                                   verified=True)
        self.assertEqual(status, 200)
        del StubUpstream.state["users"]["bob"]
        status, resp = self.stub_link("s9", "bob", email="[email protected]",
                                      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 = "[email protected]"
        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"] = "[email protected]"
        StubUpstream.state["hide_once"].add("alice")
        status, resp = self.stub_link("s1", "alice",
                                      email="[email protected]")
        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="[email protected]",
                                    verified=True, device_id="dev-a")[1]
        self.bob = self.stub_link("s2", "bob", email="[email protected]",
                                  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="[email protected]",
                                    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("[email protected]")
        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("[email protected]")
        status, resp = self.stub_link("s9", "carol", email="[email protected]",
                                      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("[email protected]")
        status, resp = self.stub_link("s10", "dave", email="[email protected]",
                                      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("[email protected]"))

    def test_an_invite_is_applied_once(self):
        self.invite("[email protected]")
        self.stub_link("s11", "erin", email="[email protected]", verified=True,
                       device_id="dev-e")
        status, resp = self.stub_link("s11", "erin", email="[email protected]",
                                      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("[email protected]", repos=("bob/secret",))
        self.assertEqual(status, 403)
        self.assertEqual(self.svc.state.peek_invites("[email protected]"), [])

    def test_reinviting_updates_the_permission_instead_of_stacking(self):
        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
                         "repos": [{"name": "notes", "permission": "read"}]})
        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
                         "repos": [{"name": "notes", "permission": "write"}]})
        pending = self.svc.state.peek_invites("[email protected]")
        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("[email protected]", repos=("notes",))
        StubUpstream.state["repo_owner"].pop("alice/notes")
        status, resp = self.stub_link("s12", "gina", email="[email protected]",
                                      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("[email protected]")
        self.stub_link("s13", "hana", email="[email protected]", 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="[email protected]",
                                    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": "[email protected]",
                         "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="[email protected]",
                                      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("[email protected]")],
                         ["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="[email protected]",
                                      verified=True, device_id="dev-c2")
        self.assertEqual(resp.get("granted_repos"), ["alice/notes"])
        self.assertEqual(self.svc.state.peek_invites("[email protected]"), [])

    def test_a_partial_failure_only_consumes_what_landed(self):
        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
                         "repos": ["notes", "reports"]})
        StubUpstream.state["repo_owner"].pop("alice/reports")   # one of two fails
        _, resp = self.stub_link("s10", "dan", email="[email protected]",
                                 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("[email protected]")],
                         ["alice/reports"])

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)