#!/usr/bin/env python3 # # gitctl-helper: server helper for gitctl, run as the git user. # Copyright (C) 2026 Danilo M. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License along # with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. """gitctl-helper: server helper for gitctl, run as the git user. Dispatches on SSH_ORIGINAL_COMMAND.""" import os, re, sys, shlex, shutil, subprocess, argparse, datetime REPO_BASE = "/var/lib/gitolite3/repositories" CGITRC = "/etc/cgitrc" SYNC_SCRIPT = "/usr/local/bin/sync-cgit-descs.py" # Backups go here, not next to CGITRC: the helper runs as the git user, which # owns /etc/cgitrc (the file) but cannot create new files in /etc. Must be a # git-writable directory. BACKUP_DIR = "/var/lib/gitolite3/cgitrc-backups" # Deleted bare repos are MOVED here, not removed, so delete is reversible. Must # be OUTSIDE REPO_BASE (so gitolite and cgit never see it) and on the same # filesystem as REPO_BASE for the move to be atomic. git-writable. TRASH_DIR = "/var/lib/gitolite3/trash" DEFAULT_OWNER = "by Your Name" BANNER = "# ---------- {name} ------------#" URL_RE = re.compile(r"^[A-Za-z0-9._-]+$") def die(msg, code=1): print(msg, file=sys.stderr) sys.exit(code) def validate_url(u): if not u or ".." in u or not URL_RE.match(u): die(f"error: invalid repo name: {u!r}") return u def validate_section(s): if not s or "\n" in s or "\r" in s: die(f"error: invalid section name: {s!r}") return s SECTION_RE = re.compile(r"^section=(.*)$") URL_LINE_RE = re.compile(r"^repo\.url=(.+)$") # a section banner comment, e.g. "# ---------- Linux ------------#". It belongs # to the section below it, so it bounds the preceding section's span. BANNER_RE = re.compile(r"^#\s*-{3,}.*-{3,}#\s*$") def list_sections_from(lines): out = [] for line in lines: m = SECTION_RE.match(line.rstrip("\n")) if m: out.append(m.group(1)) return out def repo_sections_from(lines): """Map every repo.url= in cgitrc to the section= it falls under (file order). cgitrc lists only the repos exposed to the web (cgit auto-discovery is off), so membership here is exactly what makes a repo public.""" out = {} current = None for line in lines: s = SECTION_RE.match(line.rstrip("\n")) if s: current = s.group(1) continue u = URL_LINE_RE.match(line.rstrip("\n")) if u: out[u.group(1)] = current return out def find_repo_section(lines, url): """Return the section a repo.url= belongs to, or None if absent.""" return repo_sections_from(lines).get(url) DESC_MAX = 60 def read_description(url): """First line of the bare repo's description file, truncated. '' if missing or the gitolite placeholder.""" try: with open(os.path.join(repo_path_for(url), "description")) as f: d = f.readline().strip() except OSError: return "" if d.startswith("Unnamed repository"): return "" return d[:DESC_MAX - 1] + "…" if len(d) > DESC_MAX else d def list_repos(repo_base, cgitrc_lines): """Every bare repo under repo_base (the full set, public and private), each tagged public/private and with its section + description. Returns a list of (name, public_bool, section, desc) sorted by name.""" public = repo_sections_from(cgitrc_lines) out = [] for entry in sorted(os.listdir(repo_base)): if not entry.endswith(".git"): continue if not os.path.isdir(os.path.join(repo_base, entry)): continue name = entry[:-4] out.append((name, name in public, public.get(name) or "", read_description(name))) return out def repo_path_for(url): return os.path.join(REPO_BASE, url + ".git") def build_repo_block(url, owner): return [ f"repo.url={url}\n", f"repo.path={repo_path_for(url)}\n", f"repo.owner={owner}\n", ] def insert_repo_block(lines, section, block): """Insert block at the end of `section`'s span. Raise KeyError if absent.""" start = None for i, line in enumerate(lines): m = SECTION_RE.match(line.rstrip("\n")) if m and m.group(1) == section: start = i break if start is None: raise KeyError(section) # end of span = next section= line OR next banner comment (a banner belongs # to the section it precedes), or EOF end = len(lines) for j in range(start + 1, len(lines)): stripped = lines[j].rstrip("\n") if SECTION_RE.match(stripped) or BANNER_RE.match(stripped): end = j break # trim trailing blank lines inside the span so we control spacing insert_at = end while insert_at > start + 1 and lines[insert_at - 1].strip() == "": insert_at -= 1 payload = ["\n"] + block + ["\n"] return lines[:insert_at] + payload + lines[insert_at:] REPO_KV_RE = re.compile(r"^repo\.[a-z]+=") def remove_repo_block(lines, url): """Inverse of insert_repo_block: drop the repo.url= block (its url/path/ owner lines) plus the one blank line insert_repo_block put before it, so the section's spacing is left as if the repo was never added. Idempotent: if the url is absent, return lines unchanged. Section banners and headers are never touched, so an emptied section keeps its header.""" start = None for i, line in enumerate(lines): m = URL_LINE_RE.match(line.rstrip("\n")) if m and m.group(1) == url: start = i break if start is None: return list(lines) # block body = this repo.url= and the contiguous repo.*= lines under it. end = start + 1 while end < len(lines) and REPO_KV_RE.match(lines[end].rstrip("\n")) \ and not URL_LINE_RE.match(lines[end].rstrip("\n")): end += 1 # also swallow one trailing blank line (the spacer insert added after the # block) and one leading blank line (the spacer before it), but never more, # so unrelated spacing is preserved. if end < len(lines) and lines[end].strip() == "": end += 1 if start > 0 and lines[start - 1].strip() == "": start -= 1 return lines[:start] + lines[end:] def write_cgitrc_lines(path, lines, backup_dir=None): """Back up then write `path` in place. Not atomic: the helper runs as the git user, which owns /etc/cgitrc but cannot create temp inodes in /etc, so an os.replace rename is impossible. The pre-write backup makes a truncated write (crash mid-write) recoverable. backup_dir defaults to the file's own directory; for the real /etc/cgitrc the caller passes BACKUP_DIR, since git cannot create files in /etc.""" if os.path.exists(path): bdir = backup_dir or os.path.dirname(os.path.abspath(path)) os.makedirs(bdir, exist_ok=True) ts = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") shutil.copy2(path, os.path.join(bdir, os.path.basename(path) + ".bak." + ts)) with open(path, "w") as f: # in place, preserves ownership f.writelines(lines) def ensure_section(cgitrc, name, backup_dir=None): """Append a section banner+header if absent. Return True if added.""" with open(cgitrc) as f: lines = f.readlines() if name in list_sections_from(lines): return False addition = ["\n", BANNER.format(name=name) + "\n", f"section={name}\n"] write_cgitrc_lines(cgitrc, lines + addition, backup_dir) return True def do_add_repo(cgitrc, url, section, owner, run_sync, backup_dir=None): with open(cgitrc) as f: lines = f.readlines() existing = find_repo_section(lines, url) if existing is not None: if existing == section: print(f"repo {url} already present under section {section}") return 0 print(f"error: repo {url} already exists under section {existing!r}, " f"refusing to move it to {section!r}", file=sys.stderr) return 1 if section not in list_sections_from(lines): print(f"error: section {section!r} does not exist; run add-section first", file=sys.stderr) return 1 new = insert_repo_block(lines, section, build_repo_block(url, owner)) write_cgitrc_lines(cgitrc, new, backup_dir) print(f"added repo {url} under section {section}") run_sync() return 0 def do_set_desc(desc_path, desc, run_sync, url=None): if not os.path.exists(desc_path): what = f"repo {url!r} not found" if url else f"description file not found: {desc_path}" print(f"error: {what}", file=sys.stderr) return 1 st = os.stat(desc_path) with open(desc_path, "w") as f: # in place, preserves owner f.write(desc.rstrip("\n") + "\n") try: os.chown(desc_path, st.st_uid, st.st_gid) # insurance except PermissionError: pass print(f"wrote description to {desc_path}") run_sync() return 0 def trash_repo(url, repo_base, trash_dir): """Move the bare repo out of repo_base into trash_dir under a timestamped name, instead of deleting it, so the operation is reversible. Returns the destination path, or None if the repo was already gone (idempotent). Refuses to move anything whose resolved path is not directly under repo_base.""" src = os.path.join(repo_base, url + ".git") # defense in depth: validate_url already blocks .. and /, but re-check the # resolved path really sits one level under repo_base before moving. if os.path.dirname(os.path.realpath(src)) != os.path.realpath(repo_base): die(f"error: refusing to trash {url!r}: resolved path escapes repo base") if not os.path.isdir(src): return None os.makedirs(trash_dir, exist_ok=True) ts = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") dst = os.path.join(trash_dir, f"{url}.git.{ts}") shutil.move(src, dst) # leave a breadcrumb for review / a future prune tool. with open(os.path.join(dst, ".trashinfo"), "w") as f: f.write(f"origin={src}\ndeleted={ts}\nby=gitctl\n") return dst def do_delete_repo(cgitrc, url, repo_base, trash_dir, run_sync, backup_dir=None): """Remove a repo from cgit (block) and move its bare repo to the trash. Each step is idempotent and reports what it did. The gitolite stanza is NOT touched here: the client removes and pushes that separately (phase 1).""" with open(cgitrc) as f: lines = f.readlines() new = remove_repo_block(lines, url) if new != lines: write_cgitrc_lines(cgitrc, new, backup_dir) print(f"removed cgit block for {url}") run_sync() else: print(f"no cgit block for {url}, skipping") dst = trash_repo(url, repo_base, trash_dir) if dst: print(f"moved bare repo to {dst}") else: print(f"no bare repo {url}.git on disk, skipping") return 0 def _self_test(): assert validate_url("publisher") == "publisher" assert validate_url("a.b_c-1") == "a.b_c-1" for bad in ["", "../etc", "a/b", "a b", "a;b", "a$b", "a\nb"]: try: validate_url(bad); assert False, bad except SystemExit: pass assert validate_section("Generic Projects") == "Generic Projects" for bad in ["", "a\nb"]: try: validate_section(bad); assert False, bad except SystemExit: pass fixture = [ "section=Generic Projects\n", "repo.url=cad-projects\n", "repo.path=/var/lib/gitolite3/repositories/cad-projects.git\n", "\n", "# ---------- SlackBuilds ------------#\n", "section=SlackBuilds\n", "repo.url=my-slackbuilds\n", "repo.path=/var/lib/gitolite3/repositories/my-slackbuilds.git\n", "\n", ] assert list_sections_from(fixture) == ["Generic Projects", "SlackBuilds"] assert find_repo_section(fixture, "cad-projects") == "Generic Projects" assert find_repo_section(fixture, "my-slackbuilds") == "SlackBuilds" assert find_repo_section(fixture, "nope") is None assert repo_sections_from(fixture) == { "cad-projects": "Generic Projects", "my-slackbuilds": "SlackBuilds", } # list_repos: scans a fake REPO_BASE, tags public/private by cgitrc and # reads each description file. cad-projects is in the fixture (public), # secret-stuff is not (private). import tempfile rb = tempfile.mkdtemp() for n, d in [("cad-projects", "CAD work"), ("secret-stuff", "")]: os.mkdir(os.path.join(rb, n + ".git")) if d: open(os.path.join(rb, n + ".git", "description"), "w").write(d + "\n") open(os.path.join(rb, "not-a-repo"), "w").write("ignore me\n") # non-.git skipped # repo_path_for/read_description read from the real REPO_BASE; point it here. global REPO_BASE saved_base, REPO_BASE = REPO_BASE, rb rows = list_repos(rb, fixture) REPO_BASE = saved_base assert rows == [ ("cad-projects", True, "Generic Projects", "CAD work"), ("secret-stuff", False, "", ""), ], rows # truncation: a long description is cut to DESC_MAX chars (ellipsis included). os.mkdir(os.path.join(rb, "wordy.git")) open(os.path.join(rb, "wordy.git", "description"), "w").write("x" * 100 + "\n") REPO_BASE = rb long = read_description("wordy") REPO_BASE = saved_base assert len(long) == DESC_MAX and long.endswith("…"), long block = build_repo_block("publisher", "by Your Name") assert block == [ "repo.url=publisher\n", "repo.path=/var/lib/gitolite3/repositories/publisher.git\n", "repo.owner=by Your Name\n", ] # insert under the FIRST section: must land before the SlackBuilds banner, # NOT between the banner and its section= line (the banner belongs to the # section it precedes) new = insert_repo_block(fixture, "Generic Projects", block) text = "".join(new) assert "repo.url=publisher" in text # publisher must precede the SlackBuilds banner (and thus its section header) assert text.index("repo.url=publisher") < text.index("# ---------- SlackBuilds") # original repo still there assert "repo.url=cad-projects" in text # insert under the LAST section: lands before EOF new2 = insert_repo_block(fixture, "SlackBuilds", block) assert "repo.url=publisher" in "".join(new2) # missing section raises try: insert_repo_block(fixture, "Nope", block); assert False except KeyError: pass # remove_repo_block is the inverse of insert: insert then remove == original, # for the first section and the last. This is the strongest guard. assert remove_repo_block(insert_repo_block(fixture, "Generic Projects", block), "publisher") == fixture assert remove_repo_block(insert_repo_block(fixture, "SlackBuilds", block), "publisher") == fixture # remove an EXISTING repo from the fixture (middle of the file): the block # goes, the SlackBuilds banner/header and the other repo stay. minus = remove_repo_block(fixture, "cad-projects") mtext = "".join(minus) assert "repo.url=cad-projects" not in mtext assert "section=Generic Projects" in mtext # emptied section keeps header assert "repo.url=my-slackbuilds" in mtext # sibling untouched assert "# ---------- SlackBuilds" in mtext # last repo in the file assert "repo.url=my-slackbuilds" not in "".join(remove_repo_block(fixture, "my-slackbuilds")) # missing url: unchanged, idempotent assert remove_repo_block(fixture, "ghost") == fixture assert remove_repo_block(remove_repo_block(fixture, "cad-projects"), "cad-projects") == remove_repo_block(fixture, "cad-projects") # trash_repo: moves the bare repo into a trash dir, leaves a .trashinfo, and # is idempotent (second call finds nothing). import tempfile base = tempfile.mkdtemp(); trash = tempfile.mkdtemp() os.makedirs(os.path.join(base, "doomed.git")) dst = trash_repo("doomed", base, trash) assert dst and os.path.isdir(dst) and not os.path.exists(os.path.join(base, "doomed.git")) assert os.path.exists(os.path.join(dst, ".trashinfo")) assert trash_repo("doomed", base, trash) is None # already gone # do_delete_repo: removes the cgit block, moves the repo, runs sync once. dd = tempfile.mkdtemp() cg = os.path.join(dd, "cgitrc") open(cg, "w").writelines([ "# ---------- Generic ------------#\n", "section=Generic\n", "repo.url=gone\n", "repo.path=/x/gone.git\n", "repo.owner=me\n", "\n", ]) rbase = tempfile.mkdtemp(); rtrash = tempfile.mkdtemp() os.makedirs(os.path.join(rbase, "gone.git")) dcalls = [] rc = do_delete_repo(cg, "gone", rbase, rtrash, run_sync=lambda: dcalls.append(1), backup_dir=dd) assert rc == 0 assert "repo.url=gone" not in open(cg).read() assert "section=Generic" in open(cg).read() # header survives assert not os.path.exists(os.path.join(rbase, "gone.git")) assert dcalls == [1] # idempotent: second delete is all skips, no sync, still rc 0 dcalls.clear() rc = do_delete_repo(cg, "gone", rbase, rtrash, run_sync=lambda: dcalls.append(1), backup_dir=dd) assert rc == 0 and dcalls == [] import tempfile as _tf d = _tf.mkdtemp() p = os.path.join(d, "cgitrc") open(p, "w").write("section=X\nrepo.url=a\n") write_cgitrc_lines(p, ["section=X\n", "repo.url=a\n", "repo.url=b\n"], backup_dir=d) assert open(p).read() == "section=X\nrepo.url=a\nrepo.url=b\n" baks = [f for f in os.listdir(d) if f.startswith("cgitrc.bak.")] assert len(baks) == 1, baks p2 = os.path.join(d, "cgitrc2") open(p2, "w").write("section=Existing\nrepo.url=z\n") added = ensure_section(p2, "NewSec") assert added is True txt = open(p2).read() assert "section=NewSec" in txt assert "# ---------- NewSec ------------#" in txt # idempotent: second call is a no-op added2 = ensure_section(p2, "NewSec") assert added2 is False assert open(p2).read().count("section=NewSec") == 1 p3 = os.path.join(d, "cgitrc3") open(p3, "w").writelines([ "# ---------- Generic ------------#\n", "section=Generic\n", "repo.url=cad\n", "repo.path=/var/lib/gitolite3/repositories/cad.git\n", "\n", "# ---------- SlackBuilds ------------#\n", "section=SlackBuilds\n", "repo.url=sps\n", "repo.path=/var/lib/gitolite3/repositories/sps.git\n", "\n", ]) calls = [] rc = do_add_repo(p3, "publisher", "Generic", "by Your Name", run_sync=lambda: calls.append(1)) assert rc == 0 t = open(p3).read() assert t.index("repo.url=publisher") < t.index("section=SlackBuilds") assert calls == [1] # sync ran after insert # idempotent: re-add same url under same section -> no duplicate, success, no sync calls.clear() rc = do_add_repo(p3, "publisher", "Generic", "x", run_sync=lambda: calls.append(1)) assert rc == 0 assert open(p3).read().count("repo.url=publisher") == 1 assert calls == [] # exists under a DIFFERENT section -> report, non-zero, no write rc = do_add_repo(p3, "publisher", "SlackBuilds", "x", run_sync=lambda: calls.append(1)) assert rc != 0 # missing section -> non-zero rc = do_add_repo(p3, "newrepo", "Nope", "x", run_sync=lambda: calls.append(1)) assert rc != 0 repo_dir = os.path.join(d, "publisher.git") os.makedirs(repo_dir) desc_path = os.path.join(repo_dir, "description") open(desc_path, "w").write("Unnamed repository; edit this file ...\n") calls.clear() rc = do_set_desc(desc_path, "my real description", run_sync=lambda: calls.append(1)) assert rc == 0 assert open(desc_path).read() == "my real description\n" assert calls == [1] # missing repo -> non-zero, no sync calls.clear() rc = do_set_desc(os.path.join(d, "ghost.git", "description"), "x", run_sync=lambda: calls.append(1), url="ghost") assert rc != 0 assert calls == [] print("self-test OK") def run_sync(): r = subprocess.run([sys.executable, SYNC_SCRIPT]) if r.returncode != 0: die(f"error: sync exited {r.returncode}", r.returncode) def main(): raw = os.environ.get("SSH_ORIGINAL_COMMAND") argv = shlex.split(raw) if raw else sys.argv[1:] if argv and argv[0] == "--self-test": _self_test(); return 0 p = argparse.ArgumentParser( prog="gitctl-helper", description="Server helper for gitctl, run as the git user. Normally invoked by the " "client over a command=-restricted SSH key, which forces this " "script and passes the verb plus arguments in SSH_ORIGINAL_COMMAND. " "It is the single privileged writer of /etc/cgitrc and of each bare " "repo's description file. Can also be run directly for local " "testing; pass --self-test to run the built-in assertions.") sub = p.add_subparsers(dest="verb", required=True, help="server-side operation to perform") sub.add_parser("list-sections", help="print every cgit section name, one per line", description="Read /etc/cgitrc and print each section= name in file order.") sp = sub.add_parser("add-section", help="append a new cgit section header if absent", description="Append a banner and section= header to /etc/cgitrc. " "Idempotent: a section that already exists is left " "untouched. Prints 'added' or 'exists'.") sp.add_argument("--name", required=True, help="section title as shown in cgit, e.g. 'Generic Projects'") ar = sub.add_parser("add-repo", help="insert a repo block under a section in cgitrc, then sync", description="Insert a repo.url/path/owner block at the end of the " "named section's span in /etc/cgitrc (positionally, " "before the next section or EOF), then run the cgit " "description sync. Idempotent under the same section; " "refuses to move a repo that already exists under a " "different section.") ar.add_argument("--url", required=True, help="repo name as it appears in cgit and on disk (becomes .git)") ar.add_argument("--section", required=True, help="existing section to file the repo under; create it first with add-section") ar.add_argument("--owner", default=DEFAULT_OWNER, help="repo.owner string shown in cgit (default: %(default)r)") sd = sub.add_parser("set-desc", help="write a bare repo's description file, then sync", description="Write the bare repo's description file in place " "(preserving its git:gitolite3 ownership) and run the " "sync script. This is the only supported way to change " "a cgit repo.desc=; it is never written directly.") sd.add_argument("--url", required=True, help="repo name (the bare repo is .git)") sd.add_argument("--desc", required=True, help="description text (a single trailing newline is enforced)") sub.add_parser("sync", help="run the cgit description sync script", description="Run sync-cgit-descs.py, which copies each bare repo's " "description into cgit's repo.desc=. The single writer of repo.desc=.") re_ = sub.add_parser("repo-exists", help="exit 0 if the bare repo exists, 1 otherwise", description="Check whether .git exists under the repository " "base. Used by the client to poll after a gitolite " "push until gitolite has built the bare repo.") re_.add_argument("--url", required=True, help="repo name to test for (as .git)") dr = sub.add_parser("delete-repo", help="remove a repo from cgit and move its bare repo to the trash", description="Inverse of add-repo plus the bare repo: remove the " "repo's cgit block (and sync), then MOVE its bare repo " "out of the repository base into the trash dir under a " "timestamped name (not rm, so it is reversible). The " "gitolite stanza is removed by the client, not here. " "Idempotent: a piece already gone is skipped, not an error.") dr.add_argument("--url", required=True, help="repo name to delete (the bare repo is .git)") sub.add_parser("list-repos", help="list every bare repo, tagged public/private", description="Scan the repository base for all bare repos (the full " "set, not just the cgit-exposed ones), tag each public or " "private by cgitrc membership, and print one " "namePUB|PRIVsectiondescription line each.") a = p.parse_args(argv) if a.verb == "list-sections": with open(CGITRC) as f: for s in list_sections_from(f.readlines()): print(s) return 0 if a.verb == "add-section": validate_section(a.name) print("added" if ensure_section(CGITRC, a.name, BACKUP_DIR) else "exists") return 0 if a.verb == "add-repo": validate_url(a.url); validate_section(a.section) return do_add_repo(CGITRC, a.url, a.section, a.owner, run_sync, BACKUP_DIR) if a.verb == "set-desc": validate_url(a.url) return do_set_desc(os.path.join(repo_path_for(a.url), "description"), a.desc, run_sync, url=a.url) if a.verb == "sync": run_sync(); return 0 if a.verb == "repo-exists": validate_url(a.url) return 0 if os.path.isdir(repo_path_for(a.url)) else 1 if a.verb == "list-repos": with open(CGITRC) as f: cgit = f.readlines() for name, pub, section, desc in list_repos(REPO_BASE, cgit): print(f"{name}\t{'PUB' if pub else 'PRIV'}\t{section}\t{desc}") return 0 if a.verb == "delete-repo": validate_url(a.url) return do_delete_repo(CGITRC, a.url, REPO_BASE, TRASH_DIR, run_sync, BACKUP_DIR) return 1 if __name__ == "__main__": if len(sys.argv) > 1 and sys.argv[1] == "--self-test" and not os.environ.get("SSH_ORIGINAL_COMMAND"): _self_test(); sys.exit(0) sys.exit(main())