security: harden granthi-link + client against 7 codex findings

1. CRITICAL account-takeover by login collision: persist zitadel_sub ->
   gitea_login identity map (state.json, 0600, atomic); mapping wins,
   deleted logins re-created only if service-created, existing unmapped
   logins bind only on verified email match, else 409; token never
   minted before binding passes
2. test_mode now gated behind GRANTHI_LINK_ALLOW_TEST_MODE=1 env
3. refuse startup unless config.json is 0600/0400 and owned by service
4. client config created O_CREAT 0600 (no write-then-chmod window)
5. credential-helper command paths shlex-quoted
6. POST bodies capped at 64KB (413); missing/invalid Content-Length rejected
7. Gitea 409 on user create handled idempotently (re-fetch + verify email)

Co-Authored-By: Claude Fable 5 <[email protected]>

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", credential_helper_value())
-    git(folder, "config", "credential.helper", helper)
 
 
 # --------------------------------------------------------------------------

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.
+    config must be 0600/0400 and owned by the service user or startup is
-  * test_mode: when config "test_mode" is true, a /v1/link body may carry
+    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):
+    def get_user(self, login):
-        status, _ = http_json(
+        """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)
+        sub = str(userinfo.get("sub") or "").strip()
-        if not login:
+        if not sub:
-            return 422, {"error": "could not derive a login from userinfo"}
+            return 422, {"error": "userinfo has no sub"}
-        if not self.user_exists(login):
+        try:
-            err = self.create_user(login, userinfo)
+            with self.state.lock:
-            if err:
+                status, err_resp, login = self._bind_identity(sub, userinfo)
-                return 502, {"error": err}
+        except RuntimeError as e:  # identity state unreadable: fail closed
-            LOG.info("created gitea user %s", login)
+            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)
 
 

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 = []

tests/test_server.py

@@ -1,12 +1,16 @@
-"""Unit tests for granthi-link: login derivation, link/repos flows against a
+"""Unit tests for granthi-link: login derivation, identity binding rules,
-stub HTTP server that plays both Zitadel userinfo and the Gitea API."""
+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):
                                         "[email protected]", "email":
                                         "[email protected]", "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(),
+        StubUpstream.state = {"users": {}, "created": [], "repos": set(),
-                              "token_reqs": []}
+                              "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):
+    def test_link_records_identity_mapping(self):
-        StubUpstream.state["users"].add("alice.smith")
+        self.svc.link({"zitadel_access_token": "good-token",
-        status, resp = self.svc.link({"zitadel_access_token": "good-token",
+                       "device_name": "d"})
-                                      "device_name": "d"})
+        state = self.read_state()
-        self.assertEqual(status, 200)
+        rec = state["identities"]["123"]
-        self.assertEqual(StubUpstream.state["created"], [])
+        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):
+    def test_link_missing_sub_422(self):
-        # disabled -> stub ignored, token required
+        self.enable_test_mode()
-        status, _ = self.svc.link({"test_userinfo": {"sub": "1",
+        status, _ = self.svc.link({"test_userinfo":
-                                                     "preferred_username": "x"}})
+                                   {"preferred_username": "nosub"},
-        self.assertEqual(status, 400)
+                                   "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
-        status, resp = self.svc.link({"test_userinfo": {
+        env = {k: v for k, v in os.environ.items()
-            "sub": "1", "preferred_username": "evetest"}, "device_name": "d"})
+               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")