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.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.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"})
            return self._json(201, {"sha1": "MINTED", "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 log_message(self, *a):
        pass


class ServiceTestBase(unittest.TestCase):
    def setUp(self):
        StubUpstream.state = {"users": {}, "created": [], "repos": set(),
                              "token_reqs": [], "hide_once": set()}
        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,
        })

    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"):
        ui = {"sub": sub, "preferred_username": username}
        if email is not None:
            ui["email"] = email
        if verified is not None:
            ui["email_verified"] = verified
        return self.svc.link({"test_userinfo": ui, "device_name": device})

    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")
        self.assertEqual(resp["token"], "MINTED")
        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")


if __name__ == "__main__":
    unittest.main()