diff --git a/client/__pycache__/granthi_sync_client.cpython-314.pyc b/client/__pycache__/granthi_sync_client.cpython-314.pyc deleted file mode 100644 index a8eceaf..0000000 Binary files a/client/__pycache__/granthi_sync_client.cpython-314.pyc and /dev/null differ diff --git a/client/granthi_sync_client.py b/client/granthi_sync_client.py index f567271..d4f3328 100644 --- a/client/granthi_sync_client.py +++ b/client/granthi_sync_client.py @@ -29,6 +29,7 @@ import base64 import json import os import re +import shlex import signal import subprocess import sys @@ -101,9 +102,12 @@ def load_config(): def save_config(cfg): os.makedirs(CONFIG_DIR, mode=0o700, exist_ok=True) tmp = CONFIG_PATH + ".tmp" - with open(tmp, "w") as f: + # O_CREAT with mode 0600 -- the file is never observable with wider + # permissions (a write-then-chmod sequence leaves a umask-sized window + # in which the token is world-readable). + fd = os.open(tmp, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as f: json.dump(cfg, f, indent=2) - os.chmod(tmp, 0o600) os.replace(tmp, CONFIG_PATH) @@ -220,9 +224,17 @@ def cmd_git_credential(argv): return 0 +def credential_helper_value(): + """Shell command git runs for credentials. Both paths are shlex-quoted: + a Python or script path containing spaces (or shell metacharacters) + must neither break the helper nor inject into the shell.""" + return "!{} {} git-credential".format( + shlex.quote(sys.executable), + shlex.quote(os.path.abspath(__file__))) + + def install_credential_helper(folder): - helper = f"!{sys.executable} {os.path.abspath(__file__)} git-credential" - git(folder, "config", "credential.helper", helper) + git(folder, "config", "credential.helper", credential_helper_value()) # -------------------------------------------------------------------------- diff --git a/server/__pycache__/granthi_link.cpython-314.pyc b/server/__pycache__/granthi_link.cpython-314.pyc deleted file mode 100644 index eb10fe5..0000000 Binary files a/server/__pycache__/granthi_link.cpython-314.pyc and /dev/null differ diff --git a/server/granthi_link.py b/server/granthi_link.py index f12a717..83f3157 100644 --- a/server/granthi_link.py +++ b/server/granthi_link.py @@ -5,7 +5,7 @@ Bridges shre-id (Zitadel) identity to a Granthi (Gitea) forge: POST /v1/link {zitadel_access_token, device_name} -> validates the token against Zitadel userinfo - -> ensures a Gitea user exists (admin API) + -> applies the identity-binding rules (see below) -> mints a scoped Gitea token for that user (admin basic auth + `Sudo:` header -- the only mechanism that works on Gitea 1.27; token-authenticated sudo returns 401) @@ -29,11 +29,24 @@ Design decisions (documented per spec): * Token minting NEEDS the admin password (basic auth + Sudo header). The admin API token alone cannot mint user tokens on 1.27. The config therefore carries admin_login/admin_password alongside admin_token; - config must be root-owned 0600. - * test_mode: when config "test_mode" is true, a /v1/link body may carry + config must be 0600/0400 and owned by the service user or startup is + REFUSED (fail closed). + * test_mode: when config "test_mode" is true AND the service environment + also sets GRANTHI_LINK_ALLOW_TEST_MODE=1, a /v1/link body may carry "test_userinfo" (dict) instead of a Zitadel round-trip. This exists so E2E can exercise the ensure-user+mint path without a human OAuth login. - NEVER enable in production config. + Config alone is NOT enough: without the env gate the flag is logged + loudly and ignored. NEVER enable in production. + * Identity binding: /v1/link persists a server-side map of Zitadel `sub` + -> Gitea login in state.json (0600, atomic writes). Rules: + (a) mapped sub -> always use the mapped login; if that login was + deleted it is re-created only when it was service-created, + otherwise the link is refused; + (b) unmapped sub + login free -> create user, record mapping; + (c) unmapped sub + login taken -> bind ONLY when the Gitea user's + primary email equals the Zitadel userinfo email AND + email_verified is true; otherwise 409. + A token is never minted before the binding rule passes. Stdlib only. Python 3.9+. """ @@ -53,10 +66,16 @@ import urllib.request from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -VERSION = "1.0.0" +VERSION = "1.1.0" LOG = logging.getLogger("granthi-link") DEFAULT_CONFIG = "/opt/granthi-link/config.json" +DEFAULT_STATE = "/opt/granthi-link/state.json" +MAX_BODY_BYTES = 64 * 1024 +TEST_MODE_ENV = "GRANTHI_LINK_ALLOW_TEST_MODE" + +# Sentinel: create_user hit a 409 (someone else created the login first). +USER_CREATE_CONFLICT = object() LOGIN_SAFE = re.compile(r"[^a-zA-Z0-9._-]+") @@ -95,6 +114,80 @@ def _basic(login, password): return f"Basic {tok}" +# -------------------------------------------------------------------------- +# Startup hardening +# -------------------------------------------------------------------------- + +def check_config_perms(path, euid=None): + """Fail-closed config check. Returns an error string or None. + + The config carries the Gitea admin password; it must be 0600/0400 and + owned by the user the service runs as, or startup is refused. + """ + st = os.stat(path) + mode = st.st_mode & 0o777 + if mode not in (0o600, 0o400): + return (f"config {path} has mode {oct(mode)}; refusing to start " + f"(must be 0600 or 0400)") + euid = os.geteuid() if euid is None else euid + if st.st_uid != euid: + return (f"config {path} is owned by uid {st.st_uid} but the service " + f"runs as uid {euid}; refusing to start") + return None + + +# -------------------------------------------------------------------------- +# Identity map: zitadel sub -> gitea login (JSON, 0600, atomic writes) +# -------------------------------------------------------------------------- + +class IdentityStore: + """Persistent map of Zitadel `sub` -> Gitea login binding records. + + Record shape: {"login": str, "created_by_service": bool, + "email": str, "linked_at": iso8601} + """ + + def __init__(self, path): + self.path = path + self.lock = threading.Lock() + + def _load(self): + try: + with open(self.path) as f: + data = json.load(f) + except FileNotFoundError: + return {"identities": {}} + except (ValueError, OSError) as e: + # Corrupt/unreadable state must NOT silently fall back to an + # empty map -- that would re-open the takeover window. + raise RuntimeError(f"identity state {self.path} unreadable: {e}") + if not isinstance(data, dict) or not isinstance( + data.get("identities"), dict): + raise RuntimeError(f"identity state {self.path} malformed") + return data + + def _write(self, data): + tmp = f"{self.path}.tmp.{os.getpid()}" + fd = os.open(tmp, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600) + try: + with os.fdopen(fd, "w") as f: + json.dump(data, f, indent=2, sort_keys=True) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, self.path) + finally: + if os.path.exists(tmp): + os.unlink(tmp) + + def get(self, sub): + return self._load()["identities"].get(str(sub)) + + def set(self, sub, record): + data = self._load() + data["identities"][str(sub)] = record + self._write(data) + + # -------------------------------------------------------------------------- # Core logic (class so tests can instantiate with a stub config) # -------------------------------------------------------------------------- @@ -108,6 +201,20 @@ class LinkService: self.public_gitea = config.get("public_gitea_base", self.gitea).rstrip("/") self.userinfo_url = config.get( "zitadel_userinfo", "https://id.shre.ai/oidc/v1/userinfo") + self.state = IdentityStore(config.get("state_path", DEFAULT_STATE)) + + def test_mode_enabled(self): + """Config test_mode is honored ONLY with the env gate also set.""" + if not self.cfg.get("test_mode"): + return False + if os.environ.get(TEST_MODE_ENV) != "1": + LOG.error( + "config sets test_mode=true but %s=1 is NOT set in the " + "service environment -- IGNORING test_mode (treating as " + "false). Remove test_mode from production config.", + TEST_MODE_ENV) + return False + return True # -- identity ---------------------------------------------------------- def validate_token(self, access_token): @@ -134,18 +241,30 @@ class LinkService: def _admin_hdr(self): return {"Authorization": f"token {self.cfg['admin_token']}"} - def user_exists(self, login): - status, _ = http_json( + def get_user(self, login): + """Admin-view of a Gitea user (includes primary email) or None.""" + status, resp = http_json( "GET", f"{self.gitea}/api/v1/users/{login}", headers=self._admin_hdr()) - return status == 200 + if status == 200 and isinstance(resp, dict): + return resp + return None + + def user_exists(self, login): + return self.get_user(login) is not None + + @staticmethod + def intended_email(login, userinfo): + return userinfo.get("email") or f"{login}@users.noreply.granthi.shre.ai" def create_user(self, login, userinfo): + """Create the Gitea user. Returns None on success, the sentinel + USER_CREATE_CONFLICT on HTTP 409 (concurrent create -- caller + re-fetches and continues idempotently), or an error string.""" pw_alphabet = string.ascii_letters + string.digits password = "".join(secrets.choice(pw_alphabet) for _ in range(30)) - email = userinfo.get("email") or f"{login}@users.noreply.granthi.shre.ai" body = { "username": login, - "email": email, + "email": self.intended_email(login, userinfo), "password": password, "must_change_password": False, "visibility": "private", @@ -155,6 +274,8 @@ class LinkService: status, resp = http_json( "POST", f"{self.gitea}/api/v1/admin/users", headers=self._admin_hdr(), body=body) + if status == 409: + return USER_CREATE_CONFLICT if status != 201: return f"gitea admin user create failed (HTTP {status}): {resp}" return None @@ -178,10 +299,85 @@ class LinkService: return None, None, f"token mint failed (HTTP {status}): {resp}" return resp.get("sha1"), token_name, None + # -- identity binding (finding 1: no minting before binding passes) ---- + def _record_binding(self, sub, login, userinfo, created_by_service): + self.state.set(sub, { + "login": login, + "created_by_service": bool(created_by_service), + "email": userinfo.get("email") or "", + "linked_at": datetime.now(timezone.utc).isoformat( + timespec="seconds"), + }) + + def _bind_identity(self, sub, userinfo): + """Apply the binding rules. Returns (status, error_resp, login). + + login is None unless binding passed. Caller holds self.state.lock. + """ + rec = self.state.get(sub) + if rec: # rule (a): mapping wins, regardless of current userinfo + login = rec["login"] + if not self.user_exists(login): + if not rec.get("created_by_service"): + return 409, {"error": + f"mapped login {login} no longer exists and " + "was not created by this service; refusing " + "to re-create"}, None + err = self.create_user(login, userinfo) + if err and err is not USER_CREATE_CONFLICT: + return 502, {"error": err}, None + LOG.info("re-created service-managed gitea user %s", login) + return 200, None, login + + login = self.derive_login(userinfo) + if not login: + return 422, {"error": "could not derive a login from userinfo"}, None + + if not self.user_exists(login): # rule (b): fresh login + err = self.create_user(login, userinfo) + if err is USER_CREATE_CONFLICT: + # finding 7: concurrent first-link race. Re-fetch and continue + # idempotently -- but only if the user that won the race + # carries the email WE would have set; anything else is a + # foreign identity and must be refused. + user = self.get_user(login) + if not user: + return 502, {"error": "user create conflicted but user " + "not fetchable"}, None + want = self.intended_email(login, userinfo).lower() + if (user.get("email") or "").lower() != want: + return 409, {"error": "login exists and is not linked " + "to this identity"}, None + LOG.info("user %s created concurrently; continuing", login) + elif err: + return 502, {"error": err}, None + else: + LOG.info("created gitea user %s", login) + self._record_binding(sub, login, userinfo, created_by_service=True) + return 200, None, login + + # rule (c): login taken by an unmapped Gitea user -- bind only on + # verified email match. + user = self.get_user(login) + if not user: + return 502, {"error": "gitea user lookup failed"}, None + zemail = (userinfo.get("email") or "").lower() + gemail = (user.get("email") or "").lower() + if zemail and userinfo.get("email_verified") is True and zemail == gemail: + self._record_binding(sub, login, userinfo, created_by_service=False) + LOG.info("bound existing gitea user %s to sub %s via verified " + "email match", login, sub) + return 200, None, login + LOG.warning("refused link: login %s exists, sub %s not mapped, " + "email match=%s verified=%s", login, sub, + zemail == gemail, userinfo.get("email_verified")) + return 409, {"error": "login exists and is not linked to this " + "identity"}, None + # -- endpoints --------------------------------------------------------- def link(self, body): device_name = body.get("device_name") or "device" - if self.cfg.get("test_mode") and isinstance(body.get("test_userinfo"), dict): + if self.test_mode_enabled() and isinstance(body.get("test_userinfo"), dict): LOG.warning("TEST-MODE link request (stubbed userinfo)") userinfo = body["test_userinfo"] else: @@ -191,18 +387,21 @@ class LinkService: userinfo, err = self.validate_token(token) if err: return 401, {"error": err} - login = self.derive_login(userinfo) - if not login: - return 422, {"error": "could not derive a login from userinfo"} - if not self.user_exists(login): - err = self.create_user(login, userinfo) - if err: - return 502, {"error": err} - LOG.info("created gitea user %s", login) + sub = str(userinfo.get("sub") or "").strip() + if not sub: + return 422, {"error": "userinfo has no sub"} + try: + with self.state.lock: + status, err_resp, login = self._bind_identity(sub, userinfo) + except RuntimeError as e: # identity state unreadable: fail closed + LOG.error("%s", e) + return 500, {"error": "identity state unavailable"} + if login is None: + return status, err_resp gitea_token, token_name, err = self.mint_token(login, device_name) if err: return 502, {"error": err} - LOG.info("minted token %s for %s", token_name, login) + LOG.info("minted token %s for %s (sub %s)", token_name, login, sub) return 200, {"gitea_base": self.public_gitea, "login": login, "token": gitea_token, "token_name": token_name} @@ -253,11 +452,29 @@ class Handler(BaseHTTPRequestHandler): self._send(404, {"error": "not found"}) def do_POST(self): + cl = self.headers.get("Content-Length") + if cl is None: + self.close_connection = True + return self._send(411, {"error": "Content-Length required"}) + try: + length = int(cl) + except (ValueError, TypeError): + self.close_connection = True + return self._send(400, {"error": "invalid Content-Length"}) + if length < 0: + self.close_connection = True + return self._send(400, {"error": "invalid Content-Length"}) + if length > MAX_BODY_BYTES: + # body is not read; close so the peer can't stream it anyway + self.close_connection = True + return self._send(413, {"error": f"request body too large " + f"(max {MAX_BODY_BYTES} bytes)"}) try: - length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length) or b"{}") except (ValueError, TypeError): return self._send(400, {"error": "invalid JSON body"}) + if not isinstance(body, dict): + return self._send(400, {"error": "body must be a JSON object"}) if self.path == "/v1/link": status, resp = self.service.link(body) elif self.path == "/v1/repos": @@ -298,14 +515,15 @@ def main(): logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") path = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_CONFIG + perm_err = check_config_perms(path) + if perm_err: # fail CLOSED: a warning here would leak the admin password + LOG.error("%s", perm_err) + sys.exit(2) with open(path) as f: config = json.load(f) for key in ("gitea_base", "admin_token", "admin_login", "admin_password"): if key not in config: sys.exit(f"config missing required key: {key}") - st = os.stat(path) - if st.st_mode & 0o077: - LOG.warning("config %s is group/world readable -- chmod 600 it", path) serve(config) diff --git a/tests/__pycache__/test_client.cpython-314.pyc b/tests/__pycache__/test_client.cpython-314.pyc deleted file mode 100644 index bfe7bef..0000000 Binary files a/tests/__pycache__/test_client.cpython-314.pyc and /dev/null differ diff --git a/tests/__pycache__/test_server.cpython-314.pyc b/tests/__pycache__/test_server.cpython-314.pyc deleted file mode 100644 index 6afe809..0000000 Binary files a/tests/__pycache__/test_server.cpython-314.pyc and /dev/null differ diff --git a/tests/test_client.py b/tests/test_client.py index 1c78aae..5070010 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -3,6 +3,7 @@ config handling, device-flow polling (mocked HTTP). Stdlib unittest only.""" import json import os +import shlex import shutil import subprocess import sys @@ -124,11 +125,65 @@ class TestConfig(unittest.TestCase): self.assertEqual(st.st_mode & 0o777, 0o600) self.assertEqual(client.load_config()["login"], "x") + def test_save_is_0600_even_with_permissive_umask(self): + """Finding 4: the token file must be born 0600 (O_CREAT mode), not + chmod'ed after write -- a wide-open umask must not widen it.""" + old = os.umask(0o000) + try: + client.save_config({"token": "sekrit", "folders": {}}) + finally: + os.umask(old) + st = os.stat(client.CONFIG_PATH) + self.assertEqual(st.st_mode & 0o777, 0o600) + + def test_save_never_calls_chmod(self): + """The 0600 mode must come from creation, not a later chmod (which + would leave a window where the file is world-readable).""" + with mock.patch.object(client.os, "chmod", + side_effect=AssertionError( + "chmod used; file must be created 0600")): + client.save_config({"token": "sekrit", "folders": {}}) + st = os.stat(client.CONFIG_PATH) + self.assertEqual(st.st_mode & 0o777, 0o600) + def test_load_missing_returns_empty(self): with mock.patch.object(client, "CONFIG_PATH", "/nonexistent/nope.json"): self.assertEqual(client.load_config(), {}) +class TestCredentialHelperQuoting(unittest.TestCase): + """Finding 5: helper command paths must be shlex-quoted.""" + + def test_paths_with_spaces_are_quoted(self): + with mock.patch.object(client.sys, "executable", + "/opt/py dir/bin/python3"), \ + mock.patch.object(client, "__file__", + "/home/a user/granthi sync/client.py"): + val = client.credential_helper_value() + self.assertTrue(val.startswith("!")) + self.assertIn("'/opt/py dir/bin/python3'", val) + self.assertIn("'/home/a user/granthi sync/client.py'", val) + # shell round-trip yields exactly [python, script, subcommand] + parts = shlex.split(val[1:]) + self.assertEqual(parts, ["/opt/py dir/bin/python3", + "/home/a user/granthi sync/client.py", + "git-credential"]) + + def test_metacharacters_do_not_inject(self): + evil = "/tmp/x; rm -rf ~; echo/client.py" + with mock.patch.object(client, "__file__", evil): + val = client.credential_helper_value() + parts = shlex.split(val[1:]) + self.assertEqual(parts[1], os.path.abspath(evil)) + self.assertEqual(len(parts), 3) + + def test_plain_paths_still_work(self): + val = client.credential_helper_value() + parts = shlex.split(val[1:]) + self.assertEqual(parts[0], sys.executable) + self.assertEqual(parts[2], "git-credential") + + class TestDeviceFlow(unittest.TestCase): def test_device_flow_polls_until_token(self): calls = [] diff --git a/tests/test_server.py b/tests/test_server.py index 3d02b5a..7ce3789 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,12 +1,16 @@ -"""Unit tests for granthi-link: login derivation, link/repos flows against a -stub HTTP server that plays both Zitadel userinfo and the Gitea API.""" +"""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")) @@ -14,7 +18,12 @@ import granthi_link # noqa: E402 class StubUpstream(BaseHTTPRequestHandler): - """Plays Zitadel (/oidc/v1/userinfo) and Gitea (everything else).""" + """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): @@ -34,10 +43,14 @@ class StubUpstream(BaseHTTPRequestHandler): "Alice.Smith@org.example", "email": "alice@example.com", "name": "Alice"}) return self._json(401, {"error": "invalid token"}) - if self.path.startswith("/api/v1/users/"): + 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}) + return self._json(200, {"login": login, + "email": st["users"][login]}) return self._json(404, {"message": "not found"}) self._json(404, {}) @@ -46,7 +59,9 @@ class StubUpstream(BaseHTTPRequestHandler): length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length) or b"{}") if self.path == "/api/v1/admin/users": - st["users"].add(body["username"]) + 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"): @@ -72,20 +87,42 @@ class StubUpstream(BaseHTTPRequestHandler): class ServiceTestBase(unittest.TestCase): def setUp(self): - StubUpstream.state = {"users": set(), "created": [], "repos": set(), - "token_reqs": []} + 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): @@ -123,12 +160,15 @@ class TestLink(ServiceTestBase): self.assertEqual(sorted(req["body"]["scopes"]), ["write:repository", "write:user"]) - def test_link_existing_user_skips_create(self): - StubUpstream.state["users"].add("alice.smith") - status, resp = self.svc.link({"zitadel_access_token": "good-token", - "device_name": "d"}) - self.assertEqual(status, 200) - self.assertEqual(StubUpstream.state["created"], []) + 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", @@ -139,17 +179,251 @@ class TestLink(ServiceTestBase): status, _ = self.svc.link({"device_name": "d"}) self.assertEqual(status, 400) - def test_test_mode_stub_only_when_enabled(self): - # disabled -> stub ignored, token required - status, _ = self.svc.link({"test_userinfo": {"sub": "1", - "preferred_username": "x"}}) - 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 - status, resp = self.svc.link({"test_userinfo": { - "sub": "1", "preferred_username": "evetest"}, "device_name": "d"}) + 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) + + +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_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): @@ -169,14 +443,10 @@ class TestRepos(ServiceTestBase): self.assertEqual(status, 400) -class TestHealth(ServiceTestBase): +class TestHealth(HandlerTestBase): def test_health_endpoint(self): - granthi_link.Handler.service = self.svc - srv = ThreadingHTTPServer(("127.0.0.1", 0), granthi_link.Handler) - threading.Thread(target=srv.serve_forever, daemon=True).start() - self.addCleanup(srv.shutdown) with urllib.request.urlopen( - f"http://127.0.0.1:{srv.server_address[1]}/health") as r: + 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")