Files
granthi-sync/server/granthi_link.py
T

667 lines
28 KiB
Python
Raw Normal View History

#!/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 json
import logging
import os
import re
import secrets
import signal
import string
import sys
import threading
import time
import urllib.error
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"
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)}
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)
self.max_keys = max_keys
self._clock = clock or time.monotonic
self._hits = {}
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
now = self._clock()
key = (route, client)
with self._lock:
hits = [t for t in self._hits.get(key, ()) if now - t < window]
if len(hits) >= limit:
self._hits[key] = hits
return False, max(1, int(window - (now - hits[0])) + 1)
hits.append(now)
self._hits[key] = hits
if len(self._hits) > self.max_keys:
self._prune(now)
return True, 0
def _prune(self, now):
"""Caller holds the lock. Drop expired keys; if still over the cap,
drop the least-recently-hit. Dropping RESETS those clients' windows,
which is fail-open -- but an unbounded dict is a memory DoS, and a
bounded one is the lesser failure. Say so loudly either way."""
for k in [k for k, v in self._hits.items()
if not v or now - v[-1] >= self.rules[k[0]][1]]:
del self._hits[k]
if len(self._hits) > self.max_keys:
over = len(self._hits) - self.max_keys
for k in sorted(self._hits, key=lambda k: self._hits[k][-1])[:over]:
del self._hits[k]
LOG.warning("rate limiter at capacity (%d keys): dropped %d "
"least-recent windows; those clients start fresh",
self.max_keys, over)
def client_ip(handler, trust_forwarded_for):
"""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. A trusted proxy
APPENDS the peer it actually saw, so the LAST entry is the only one we did
not let the client choose -- take that, never the first. Off by default;
only enable when something trustworthy really is in front.
"""
if trust_forwarded_for:
xff = handler.headers.get("X-Forwarded-For", "")
parts = [p.strip() for p in xff.split(",") if p.strip()]
if parts:
return parts[-1]
return handler.client_address[0]
# --------------------------------------------------------------------------
# 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)
# --------------------------------------------------------------------------
# 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))
# 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") or {}
self.trust_forwarded_for = bool(rl.get("trust_forwarded_for", False))
if rl.get("enabled", True):
rules = dict(DEFAULT_RATE_RULES)
for route, spec in (rl.get("rules") or {}).items():
if (not isinstance(spec, (list, tuple)) or len(spec) != 2
or not all(isinstance(x, 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)",
{k: f"{v[0]}/{v[1]}s" for k, v in rules.items()},
self.trust_forwarded_for)
else:
self.limiter = None
LOG.warning("rate limiting DISABLED by config -- /v1/link will "
"mint tokens without any throttle")
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):
"""Admin basic auth + Sudo header. Verified on Gitea 1.27.1:
token-auth sudo (header or ?sudo=) -> 401; basic+Sudo -> 201."""
safe_dev = LOGIN_SAFE.sub("-", device_name or "device")[:40]
token_name = "granthi-sync-{}-{}".format(
safe_dev, datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"))
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):
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
gitea_token, token_name, err = self.mint_token(login, device_name)
if err:
return 502, {"error": err}
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}
def repos(self, body):
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)
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"})
if self.path == "/v1/link":
status, resp = self.service.link(body)
elif self.path == "/v1/repos":
status, resp = self.service.repos(body)
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()