#!/usr/bin/env python3 """granthi-link: provisioning service for the granthi-sync product spine. Bridges shre-id (Zitadel) identity to a Granthi (Gitea) forge: POST /v1/link {zitadel_access_token, device_name} -> validates the token against Zitadel userinfo -> 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) -> {gitea_base, login, token, token_name} POST /v1/repos {token, name, private} -> creates a user repo with the USER token, returns clone URLs rebased onto the public gitea_base (container ROOT_URL may not resolve for clients on the tailnet). GET /health -> {"status": "ok", ...} Design decisions (documented per spec): * User creation uses source_id 0 (local). Gitea 1.27.1 has no /api/v1/admin/identity-auth-sources endpoint (404); the shre-id OAuth2 source is ID 1 (discovered via `gitea admin auth list`), but users created against an OAuth2 source cannot basic-auth and admin-created users get no external_login_user row anyway, so SSO linking happens on first OIDC web login (by email) regardless. Local + random password is the simplest correct v1. * 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 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. 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+. """ import base64 import ipaddress import json import logging import os import re import secrets import signal import string import sys import threading import time import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer VERSION = "1.1.0" LOG = logging.getLogger("granthi-link") DEFAULT_CONFIG = "/opt/granthi-link/config.json" DEFAULT_STATE = "/opt/granthi-link/state.json" DEFAULT_AUDIT = "/opt/granthi-link/audit.jsonl" 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._-]+") # Rate limiting. /v1/link is the expensive endpoint -- it round-trips Zitadel, # can CREATE a forge account and always mints a token -- so its default is # deliberately tight. /v1/repos only spends the caller's own token. DEFAULT_RATE_RULES = {"/v1/link": (5, 3600), "/v1/repos": (60, 3600), # Reads are cheap but still authenticated work. "/v1/devices": (120, 3600), "/v1/audit": (120, 3600), # Sharing is an ordinary act; inviting reaches a person # who does not exist yet, so it is the tighter of the two. "/v1/grants": (120, 3600), "/v1/invite": (60, 3600), # Revocation is a safety action, so its limit is set # high rather than tight. It is NOT 0: in this limiter # a limit of 0 DISABLES the endpoint outright (see # RateLimiter.check), which would mean nobody could # ever sign a lost laptop out. Caught in live QA -- # unit tests called the service method directly and so # never went through the limiter at all. "/v1/devices/revoke": (600, 3600)} MAX_RATE_KEYS = 10000 # -------------------------------------------------------------------------- # Rate limiting # -------------------------------------------------------------------------- class RateLimiter: """Sliding-window limiter keyed by (route, client). In-process and lock-guarded: granthi-link is ONE ThreadingHTTPServer process, so a dict is the entire store -- no redis, no shared cache, and nothing to keep consistent across nodes. If this ever runs multi-process the limiter must move with it; that is why the store is behind this class rather than sprinkled through the handler. """ def __init__(self, rules=None, max_keys=MAX_RATE_KEYS, clock=None): self.rules = dict(rules or DEFAULT_RATE_RULES) # Per-ROUTE budget, not one global table. A shared cap lets a flood of # cheap /v1/repos keys exhaust the table and lock brand-new /v1/link # clients out -- turning the fail-closed capacity guard into a # cross-route denial of service. Each route gets its own space. self.max_keys = max_keys self._clock = clock or time.monotonic self._hits = {route: {} for route in self.rules} self._lock = threading.Lock() def check(self, route, client): """-> (allowed: bool, retry_after: int). Records the hit when allowed. Denied requests are NOT recorded: a client that keeps hammering must not push its own window forward and lock itself out indefinitely. """ rule = self.rules.get(route) if not rule: return True, 0 limit, window = rule if limit <= 0: # 0 = endpoint disabled entirely return False, window with self._lock: # Clock read INSIDE the lock: taken outside, two racing threads # can append out of order, and both hits[0] (retry_after) and # v[-1] (reclamation age) assume the list is chronological. now = self._clock() table = self._hits.setdefault(route, {}) hits = [t for t in table.get(client, ()) if now - t < window] if len(hits) >= limit: table[client] = hits return False, max(1, int(window - (now - hits[0])) + 1) if client not in table and len(table) >= self.max_keys: # At capacity, reclaim expired keys first... self._reclaim_expired(route, now) if len(table) >= self.max_keys: # ...and if every window is still live, this is an # identity flood, not organic load. Evicting here would # let an attacker reset their own limit on demand, so # refuse the NEW key instead. Fail closed: /v1/link is # invite-only and low-volume, so hitting this cap is an # attack, and turning tokens away beats minting them. LOG.error("rate limiter at capacity for %s (%d keys) with " "no expired windows: refusing new client %r", route, self.max_keys, client) return False, window hits.append(now) table[client] = hits return True, 0 def _reclaim_expired(self, route, now): """Caller holds the lock. Drop only keys whose window has fully expired -- never a live one, or reclamation becomes the bypass.""" window = self.rules[route][1] table = self._hits.get(route, {}) dead = [c for c, v in table.items() if not v or now - v[-1] >= window] for c in dead: del table[c] if dead: LOG.info("rate limiter reclaimed %d expired windows on %s", len(dead), route) def client_ip(handler, trust_forwarded_for, trusted_proxies=()): """The address to rate-limit on. Behind cloudflared every request arrives from the tunnel, so limiting on the socket peer would let one abuser starve everyone. X-Forwarded-For is client-controlled, though: a caller can prepend anything. Two guards: 1. The header is honored ONLY when the socket peer is itself a configured trusted proxy. Without that, anyone who can reach the origin directly -- and the origin also listens on the tailnet -- picks their own rate-limit key and rotates it at will. 2. A trusted proxy APPENDS the peer it actually saw, so the LAST entry is the only one the client did not choose. Take that, never the first. The value must parse as a real IP; junk falls back to the socket peer rather than becoming a key of its own. """ peer = handler.client_address[0] if not trust_forwarded_for: return peer if not _ip_in_any(peer, trusted_proxies): LOG.warning("X-Forwarded-For ignored: peer %s is not a trusted proxy", peer) return peer xff = handler.headers.get("X-Forwarded-For", "") parts = [p.strip() for p in xff.split(",") if p.strip()] if not parts: return peer try: return str(ipaddress.ip_address(parts[-1])) except ValueError: LOG.warning("X-Forwarded-For last hop %r is not an IP; using peer", parts[-1][:60]) return peer def _ip_in_any(addr, networks): try: ip = ipaddress.ip_address(addr) except ValueError: return False for net in networks: # Pre-parsed at startup by LinkService._parse_trusted_proxies, so a # malformed entry can never reach here as a silent per-request skip. if ip.version == net.version and ip in net: return True return False # -------------------------------------------------------------------------- # HTTP helper (patchable in tests) # -------------------------------------------------------------------------- def http_json(method, url, headers=None, body=None, timeout=20): """Minimal JSON-over-HTTP helper. Returns (status, parsed_or_text).""" data = None hdrs = dict(headers or {}) # Cloudflare in front of id.shre.ai 403s the default Python-urllib UA. hdrs.setdefault("User-Agent", f"granthi-link/{VERSION}") if body is not None: data = json.dumps(body).encode() hdrs.setdefault("Content-Type", "application/json") req = urllib.request.Request(url, data=data, headers=hdrs, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read() status = resp.status except urllib.error.HTTPError as e: raw = e.read() status = e.code except (urllib.error.URLError, OSError) as e: return 599, {"error": str(e)} try: return status, json.loads(raw) if raw else {} except ValueError: return status, {"raw": raw.decode(errors="replace")} def _basic(login, password): tok = base64.b64encode(f"{login}:{password}".encode()).decode() 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) # -- device registry --------------------------------------------------- # Devices hang off the identity that owns them, so "which computers can # reach my files" has exactly one answer and it cannot drift from the # identity map. Callers hold self.lock. def record_device(self, sub, device_id, record): data = self._load() ident = data["identities"].get(str(sub)) if ident is None: return False devices = ident.setdefault("devices", {}) existing = devices.get(device_id, {}) # Re-linking the same device REPLACES its token name and clears any # previous revocation: the user just proved identity again. existing.update(record) existing["revoked_at"] = None devices[device_id] = existing data["identities"][str(sub)] = ident self._write(data) return True def devices_for_login(self, login): """(sub, devices) for a Gitea login, or (None, {}). Keyed on login because the caller authenticates with a forge token, which proves a login -- not a Zitadel sub. """ data = self._load() for sub, ident in data["identities"].items(): if ident.get("login") == login: return sub, dict(ident.get("devices") or {}) return None, {} # -- pending invites --------------------------------------------------- # Keyed by VERIFIED email. An invite is a promise of access made before the # person has a forge account; it is applied the first time they link. The # key must be an identity the IdP vouches for, or anyone could claim # someone else's invite by asserting their address. def add_invite(self, email, grants, invited_by): data = self._load() book = data.setdefault("invites", {}) entry = book.setdefault(email.strip().lower(), []) for g in grants: # Re-inviting the same repo updates the permission rather than # stacking duplicates that would be applied twice. entry[:] = [e for e in entry if e["repo"] != g["repo"]] entry.append({"repo": g["repo"], "permission": g["permission"], "invited_by": invited_by, "invited_at": datetime.now(timezone.utc).isoformat( timespec="seconds")}) self._write(data) def take_invites(self, email): """Read AND clear the invites for an email, atomically under the caller's lock. Returns [] when there are none.""" if not email: return [] data = self._load() book = data.get("invites") or {} pending = book.pop(email.strip().lower(), []) if pending: self._write(data) return pending def peek_invites(self, email): book = self._load().get("invites") or {} return list(book.get((email or "").strip().lower(), [])) def mark_revoked(self, sub, device_id, when): data = self._load() ident = data["identities"].get(str(sub)) or {} device = (ident.get("devices") or {}).get(device_id) if device is None: return False device["revoked_at"] = when self._write(data) return True # -------------------------------------------------------------------------- # Audit log: append-only JSONL, one line per security-relevant event # -------------------------------------------------------------------------- class AuditLog: """Who linked a computer, from where, when, and what was refused. Append-only and separate from the identity map on purpose: state.json is rewritten atomically on every change, so anything kept only there is one bad write away from gone. An audit trail that can be rewritten by the thing it audits is not an audit trail. NOTE what this can and cannot see. Every event here is one granthi-link handled. Git pushes and pulls do NOT pass through this service -- they go straight to the forge -- so they are recorded by Gitea, not here. Reading this file and believing it lists file activity would be a real mistake. """ def __init__(self, path, keep_bytes=64 * 1024 * 1024): self.path = path self.keep_bytes = keep_bytes self.lock = threading.Lock() def write(self, event, **fields): line = {"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"), "event": event} line.update({k: v for k, v in fields.items() if v is not None}) try: with self.lock: self._rotate_if_needed() fd = os.open(self.path, os.O_CREAT | os.O_WRONLY | os.O_APPEND, 0o600) with os.fdopen(fd, "a") as f: f.write(json.dumps(line, sort_keys=True) + "\n") except OSError as e: # Never let auditing break the request it is auditing, but say so # loudly -- a silently dead audit log is worse than none. LOG.error("AUDIT WRITE FAILED (%s): %s", e, line) def _rotate_if_needed(self): try: if os.path.getsize(self.path) < self.keep_bytes: return except OSError: return os.replace(self.path, self.path + ".1") def read_for(self, login, limit=200): """Events belonging to one login, newest first.""" out = [] for path in (self.path, self.path + ".1"): try: with open(path) as f: for raw in f: try: rec = json.loads(raw) except ValueError: continue if rec.get("login") == login: out.append(rec) except OSError: continue out.sort(key=lambda r: r.get("ts") or "", reverse=True) return out[:limit] # -------------------------------------------------------------------------- # Core logic (class so tests can instantiate with a stub config) # -------------------------------------------------------------------------- class LinkService: def __init__(self, config): self.cfg = config # gitea_base: URL this service uses to reach the forge (loopback ok) # public_gitea_base: URL handed to clients (tailnet / public) self.gitea = config["gitea_base"].rstrip("/") 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)) self.audit = AuditLog(config.get("audit_path", DEFAULT_AUDIT)) # Rate limiting is ON by default: this endpoint creates accounts and # mints tokens, so the safe default is limited, and disabling it has # to be a deliberate config act rather than an omission. rl = config.get("rate_limit", {}) if rl is None: rl = {} if not isinstance(rl, dict): raise SystemExit("config rate_limit must be an object") # Real JSON booleans only. `"enabled": null` or `0` must not quietly # turn limiting off, and the string "false" must not turn XFF trust # ON (every non-empty string is truthy). for flag, default in (("enabled", True), ("trust_forwarded_for", False)): if flag in rl and not isinstance(rl[flag], bool): raise SystemExit( f"config rate_limit.{flag} must be true or false, " f"got {rl[flag]!r}") self.trust_forwarded_for = rl.get("trust_forwarded_for", False) self.trusted_proxies = self._parse_trusted_proxies( rl.get("trusted_proxies")) if self.trust_forwarded_for and not self.trusted_proxies: raise SystemExit( "config rate_limit.trust_forwarded_for requires a non-empty " "trusted_proxies list -- trusting the header from any peer " "lets callers choose their own rate-limit key") if rl.get("enabled", True): rules = dict(DEFAULT_RATE_RULES) for route, spec in (rl.get("rules") or {}).items(): # `type(x) is int`, NOT isinstance: bool subclasses int, so # isinstance lets [5, true] through as a 1-SECOND window -- # an hourly limit silently becomes ~5/sec -- and false in the # limit slot disables the endpoint. if (not isinstance(spec, (list, tuple)) or len(spec) != 2 or not all(type(x) is int for x in spec) or spec[1] <= 0): # A malformed rule must not silently mean "unlimited". raise SystemExit( f"config rate_limit.rules[{route!r}] must be " f"[max_requests, window_seconds] with window > 0") rules[route] = (spec[0], spec[1]) self.limiter = RateLimiter(rules) LOG.info("rate limiting active: %s (trust_forwarded_for=%s, " "trusted_proxies=%s)", {k: f"{v[0]}/{v[1]}s" for k, v in rules.items()}, self.trust_forwarded_for, list(self.trusted_proxies)) else: self.limiter = None LOG.warning("rate limiting DISABLED by config -- /v1/link will " "mint tokens without any throttle") @staticmethod def _parse_trusted_proxies(raw): """Validate at STARTUP, not per request. This setting gates a spoofable identity, so every failure mode has to be loud and early: a bare string would be iterated character by character (each char a "network"), malformed entries would only surface as a per-request log line, and a wildcard like 0.0.0.0/0 or ::/0 quietly restores "trust X-Forwarded-For from anyone" -- the exact hole trusted_proxies exists to close. """ if raw is None: return () if isinstance(raw, str) or not isinstance(raw, (list, tuple)): raise SystemExit( "config rate_limit.trusted_proxies must be a list of CIDRs, " f"got {type(raw).__name__}") nets = [] for entry in raw: if not isinstance(entry, str): raise SystemExit( f"config rate_limit.trusted_proxies entry {entry!r} " "must be a string") try: net = ipaddress.ip_network(entry, strict=False) except ValueError as e: raise SystemExit( f"config rate_limit.trusted_proxies entry {entry!r} " f"is not a valid network: {e}") if net.prefixlen == 0: raise SystemExit( f"config rate_limit.trusted_proxies entry {entry!r} " "matches every address, which is the same as trusting " "X-Forwarded-For from any peer -- refusing") nets.append(net) return tuple(nets) 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): status, info = http_json( "GET", self.userinfo_url, headers={"Authorization": f"Bearer {access_token}"}) if status != 200: return None, f"zitadel userinfo rejected token (HTTP {status})" if not isinstance(info, dict) or "sub" not in info: return None, "zitadel userinfo returned no sub" return info, None @staticmethod def derive_login(userinfo): cand = userinfo.get("preferred_username") or "" if not cand: email = userinfo.get("email") or "" cand = email.split("@", 1)[0] cand = cand.split("@", 1)[0] # strip org domain from zitadel logins cand = LOGIN_SAFE.sub("-", cand).strip("-._").lower() return cand or None # -- gitea admin ------------------------------------------------------- def _admin_hdr(self): return {"Authorization": f"token {self.cfg['admin_token']}"} 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()) 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)) body = { "username": login, "email": self.intended_email(login, userinfo), "password": password, "must_change_password": False, "visibility": "private", "full_name": userinfo.get("name", ""), # source_id intentionally omitted -> local user; see module doc. } 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 def mint_token(self, login, device_name, device_id=None): """Admin basic auth + Sudo header. Verified on Gitea 1.27.1: token-auth sudo (header or ?sudo=) -> 401; basic+Sudo -> 201. The device id is part of the token NAME because revocation deletes by name. With only device_name + a one-second timestamp, one user linking two machines called "macbook" in the same second would collide -- and a collision means signing out one laptop kills the other's access. """ safe_dev = LOGIN_SAFE.sub("-", device_name or "device")[:40] suffix = LOGIN_SAFE.sub("-", str(device_id or ""))[:12] token_name = "granthi-sync-{}-{}{}".format( safe_dev, datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"), f"-{suffix}" if suffix else "") status, resp = http_json( "POST", f"{self.gitea}/api/v1/users/{login}/tokens", headers={ "Authorization": _basic( self.cfg["admin_login"], self.cfg["admin_password"]), "Sudo": login, }, body={"name": token_name, "scopes": ["write:repository", "write:user"]}) if status != 201: 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, client_ip=None): device_name = body.get("device_name") or "device" 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: token = body.get("zitadel_access_token") if not token: return 400, {"error": "zitadel_access_token required"} userinfo, err = self.validate_token(token) if err: return 401, {"error": err} 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 wanted_device = str(body.get("device_id") or "").strip() or None gitea_token, token_name, err = self.mint_token(login, device_name, wanted_device) if err: return 502, {"error": err} LOG.info("minted token %s for %s (sub %s)", token_name, login, sub) # A device id the client generated once and keeps. Clients older than # v1.2 do not send one; fall back to the token name, which is unique # per link, so every device still lands in the registry. device_id = wanted_device or token_name with self.state.lock: self.state.record_device(sub, device_id, { "name": device_name, "token_name": token_name, "linked_at": datetime.now(timezone.utc).isoformat( timespec="seconds"), }) granted = self.apply_invites(login, userinfo, client_ip) self.audit.write("device.link", login=login, device_id=device_id, device_name=device_name, token_name=token_name, client_ip=client_ip) resp = {"gitea_base": self.public_gitea, "login": login, "token": gitea_token, "token_name": token_name, "device_id": device_id} if granted: resp["granted_repos"] = granted return 200, resp # -- device registry endpoints ----------------------------------------- def whoami(self, token): """The login a forge token belongs to, or None. Authorisation for every device endpoint rests on this: the caller proves who they are by holding a working token for that account, and the forge is the one that decides. No login is ever taken from the request body. """ if not token or not isinstance(token, str): return None status, resp = http_json( "GET", f"{self.gitea}/api/v1/user", headers={"Authorization": f"token {token}"}) if status == 200 and isinstance(resp, dict): return resp.get("login") return None def devices(self, body, client_ip=None): login = self.whoami(body.get("token")) if not login: return 401, {"error": "invalid or revoked token"} with self.state.lock: _, devices = self.state.devices_for_login(login) out = [] for device_id, rec in sorted( devices.items(), key=lambda kv: kv[1].get("linked_at") or ""): out.append({"device_id": device_id, "name": rec.get("name") or "device", "linked_at": rec.get("linked_at"), "token_name": rec.get("token_name"), "revoked_at": rec.get("revoked_at")}) return 200, {"login": login, "devices": out} def revoke_device(self, body, client_ip=None): """Sign one computer out. Immediate: the forge token is deleted, so the next fetch or push from that machine fails at the server. There is no 'stop syncing' flag for a client to honour or ignore.""" login = self.whoami(body.get("token")) if not login: return 401, {"error": "invalid or revoked token"} device_id = str(body.get("device_id") or "").strip() if not device_id: return 400, {"error": "device_id required"} with self.state.lock: sub, devices = self.state.devices_for_login(login) record = devices.get(device_id) if sub is None or record is None: # Do not confirm the existence of ids on other accounts. self.audit.write("device.revoke.denied", login=login, device_id=device_id, client_ip=client_ip, reason="not a device of this account") return 404, {"error": "no such device on this account"} token_name = record.get("token_name") err = self.delete_forge_token(login, token_name) if err: self.audit.write("device.revoke.failed", login=login, device_id=device_id, token_name=token_name, client_ip=client_ip, reason=err) # Do NOT mark it revoked: the token still works, and a # registry claiming otherwise is worse than an honest error. return 502, {"error": err} when = datetime.now(timezone.utc).isoformat(timespec="seconds") self.state.mark_revoked(sub, device_id, when) self.audit.write("device.revoke", login=login, device_id=device_id, token_name=token_name, client_ip=client_ip) LOG.info("revoked device %s (%s) for %s", device_id, token_name, login) return 200, {"revoked": device_id, "revoked_at": when} def delete_forge_token(self, login, token_name): """Gitea refuses token-auth deletion of tokens (403); admin basic auth + Sudo works (verified 204 on 1.27.2).""" if not token_name: return "device has no recorded token to revoke" status, resp = http_json( "DELETE", f"{self.gitea}/api/v1/users/{login}/tokens/" f"{urllib.parse.quote(token_name, safe='')}", headers={"Authorization": _basic(self.cfg["admin_login"], self.cfg["admin_password"]), "Sudo": login}) if status in (204, 404): # 404 = already gone. The caller wanted it dead; it is dead. return None return f"forge refused token deletion (HTTP {status}): {resp}" # -- sharing ----------------------------------------------------------- PERMISSIONS = ("read", "write", "admin") def _repo_admin_check(self, token, full_name): """Can this caller administer that repo? Ask the FORGE. Listing collaborators requires repo admin, so a 200 here is Gitea's own answer to 'may you share this?'. Deciding it ourselves from the repo name would be a second opinion about someone else's authorisation -- and the wrong one the first time a repo is transferred. """ status, _ = http_json( "GET", f"{self.gitea}/api/v1/repos/{full_name}/collaborators", headers={"Authorization": f"token {token}"}) return status == 200 def grants(self, body, client_ip=None): """Share a repo with someone who already has a forge account.""" login = self.whoami(body.get("token")) if not login: return 401, {"error": "invalid or revoked token"} token = body.get("token") action = (body.get("action") or "add").lower() if action not in ("add", "remove", "list"): return 400, {"error": "action must be add, remove or list"} repo = str(body.get("repo") or "").strip() if not repo: return 400, {"error": "repo required"} full = repo if "/" in repo else f"{login}/{repo}" if not self._repo_admin_check(token, full): self.audit.write("grant.denied", login=login, repo=full, client_ip=client_ip, reason="caller cannot administer this repo") return 403, {"error": f"you cannot administer {full}"} if action == "list": status, resp = http_json( "GET", f"{self.gitea}/api/v1/repos/{full}/collaborators", headers={"Authorization": f"token {token}"}) people = [u.get("login") for u in resp] if isinstance(resp, list) else [] return 200, {"repo": full, "collaborators": people} who = str(body.get("login") or "").strip() if not who: return 400, {"error": "login required"} permission = (body.get("permission") or "write").lower() if permission not in self.PERMISSIONS: return 400, {"error": f"permission must be one of " f"{', '.join(self.PERMISSIONS)}"} if action == "add": status, resp = http_json( "PUT", f"{self.gitea}/api/v1/repos/{full}/collaborators/" f"{urllib.parse.quote(who, safe='')}", headers={"Authorization": f"token {token}"}, body={"permission": permission}) ok = status in (200, 204) else: status, resp = http_json( "DELETE", f"{self.gitea}/api/v1/repos/{full}/collaborators/" f"{urllib.parse.quote(who, safe='')}", headers={"Authorization": f"token {token}"}) ok = status in (200, 204, 404) # already gone is the wanted state if not ok: self.audit.write("grant.failed", login=login, repo=full, grantee=who, action=action, client_ip=client_ip, reason=f"HTTP {status}") return 502, {"error": f"forge refused (HTTP {status}): {resp}"} self.audit.write(f"grant.{action}", login=login, repo=full, grantee=who, permission=permission, client_ip=client_ip) return 200, {"repo": full, "login": who, "action": action, "permission": permission} def invite(self, body, client_ip=None): """Promise access to someone who has no forge account yet. Nothing is created for them here -- no account, no token. The grant is recorded against their VERIFIED email and applied the first time they link. If they never link, nothing ever existed. """ login = self.whoami(body.get("token")) if not login: return 401, {"error": "invalid or revoked token"} email = str(body.get("email") or "").strip().lower() if "@" not in email: return 400, {"error": "a valid email is required"} repos = body.get("repos") if not isinstance(repos, list) or not repos: return 400, {"error": "repos must be a non-empty list"} wanted = [] for entry in repos: if isinstance(entry, str): entry = {"name": entry} if not isinstance(entry, dict) or not entry.get("name"): return 400, {"error": "each repo needs a name"} name = str(entry["name"]).strip() full = name if "/" in name else f"{login}/{name}" permission = (entry.get("permission") or "write").lower() if permission not in self.PERMISSIONS: return 400, {"error": f"permission must be one of " f"{', '.join(self.PERMISSIONS)}"} if not self._repo_admin_check(body.get("token"), full): self.audit.write("invite.denied", login=login, repo=full, invitee=email, client_ip=client_ip, reason="caller cannot administer this repo") return 403, {"error": f"you cannot administer {full}"} wanted.append({"repo": full, "permission": permission}) with self.state.lock: self.state.add_invite(email, wanted, login) self.audit.write("invite", login=login, invitee=email, repos=[w["repo"] for w in wanted], client_ip=client_ip) return 200, {"invited": email, "repos": wanted, "note": "applied the first time they sign in with a " "verified email that matches"} def apply_invites(self, login, userinfo, client_ip=None): """Turn recorded invites into real collaborator rows at link time. Uses the ADMIN credential deliberately: the inviter authorised this when they issued the invite, and their session is long gone by now. Nothing here can fail the link -- someone signing in must not be blocked because a repo they were promised has since been deleted. """ email = (userinfo.get("email") or "").strip().lower() if not email or not userinfo.get("email_verified"): # Unverified email must never collect an invite: the address is # the only thing tying the promise to this person. return [] with self.state.lock: pending = self.state.take_invites(email) applied = [] for grant in pending: status, resp = http_json( "PUT", f"{self.gitea}/api/v1/repos/{grant['repo']}/collaborators/" f"{urllib.parse.quote(login, safe='')}", headers={"Authorization": _basic(self.cfg["admin_login"], self.cfg["admin_password"])}, body={"permission": grant.get("permission", "write")}) if status in (200, 204): applied.append(grant["repo"]) self.audit.write("invite.applied", login=login, repo=grant["repo"], permission=grant.get("permission"), client_ip=client_ip) else: self.audit.write("invite.apply.failed", login=login, repo=grant["repo"], client_ip=client_ip, reason=f"HTTP {status}: {resp}") LOG.warning("invite for %s on %s could not be applied " "(HTTP %s)", login, grant["repo"], status) return applied def audit_read(self, body, client_ip=None): login = self.whoami(body.get("token")) if not login: return 401, {"error": "invalid or revoked token"} try: limit = min(int(body.get("limit") or 100), 500) except (TypeError, ValueError): limit = 100 return 200, {"login": login, "events": self.audit.read_for(login, limit=limit), "note": "granthi-link events only; git pushes and pulls " "go straight to the forge and are recorded " "there, not here"} def repos(self, body, client_ip=None): token = body.get("token") name = body.get("name") if not token or not name: return 400, {"error": "token and name required"} status, resp = http_json( "POST", f"{self.gitea}/api/v1/user/repos", headers={"Authorization": f"token {token}"}, body={"name": name, "private": bool(body.get("private", True)), "default_branch": "main", "auto_init": False}) if status == 409: return 409, {"error": f"repo {name} already exists"} if status != 201: return 502, {"error": f"repo create failed (HTTP {status}): {resp}"} full_name = resp.get("full_name", "") return 200, { "name": resp.get("name"), "full_name": full_name, "private": resp.get("private"), "clone_url": f"{self.public_gitea}/{full_name}.git", "html_url": f"{self.public_gitea}/{full_name}", } # -------------------------------------------------------------------------- # HTTP server plumbing # -------------------------------------------------------------------------- class Handler(BaseHTTPRequestHandler): service = None # set by serve() server_version = f"granthi-link/{VERSION}" def _send(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): if self.path == "/health": self._send(200, {"status": "ok", "service": "granthi-link", "version": VERSION}) else: self._send(404, {"error": "not found"}) def do_POST(self): # Rate-limit BEFORE reading the body or doing any work -- the point is # to spend nothing on an abusive caller. Same close-the-connection # treatment the 413 path uses, for the same reason: we are not going # to drain a body we already decided to reject. limiter = getattr(self.service, "limiter", None) if limiter is not None: who = client_ip(self, self.service.trust_forwarded_for, self.service.trusted_proxies) allowed, retry_after = limiter.check(self.path, who) if not allowed: LOG.warning("rate limited %s %s (retry after %ss)", who, self.path, retry_after) self.close_connection = True payload = json.dumps({ "error": "rate limit exceeded", "retry_after": retry_after}).encode() self.send_response(429) self.send_header("Content-Type", "application/json") self.send_header("Retry-After", str(retry_after)) self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) return 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: 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"}) # Recomputed rather than reused from the rate-limit block above: # that block is skipped entirely when no limiter is configured, and # an audit trail must not have holes because limiting was off. peer = client_ip(self, self.service.trust_forwarded_for, self.service.trusted_proxies) if self.path == "/v1/link": status, resp = self.service.link(body, peer) elif self.path == "/v1/repos": status, resp = self.service.repos(body, peer) elif self.path == "/v1/devices": status, resp = self.service.devices(body, peer) elif self.path == "/v1/devices/revoke": status, resp = self.service.revoke_device(body, peer) elif self.path == "/v1/grants": status, resp = self.service.grants(body, peer) elif self.path == "/v1/invite": status, resp = self.service.invite(body, peer) elif self.path == "/v1/audit": status, resp = self.service.audit_read(body, peer) else: return self._send(404, {"error": "not found"}) self._send(status, resp) def log_message(self, fmt, *args): # route to logging, redact nothing else LOG.info("%s %s", self.address_string(), fmt % args) def serve(config): service = LinkService(config) Handler.service = service servers = [] for host, port in config.get("binds", [["127.0.0.1", 3042]]): srv = ThreadingHTTPServer((host, int(port)), Handler) servers.append(srv) t = threading.Thread(target=srv.serve_forever, daemon=True) t.start() LOG.info("listening on %s:%s", host, port) stop = threading.Event() def on_term(signum, frame): LOG.info("signal %s, shutting down", signum) stop.set() signal.signal(signal.SIGTERM, on_term) signal.signal(signal.SIGINT, on_term) stop.wait() for srv in servers: srv.shutdown() 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}") serve(config) if __name__ == "__main__": main()