feat: add git-setup-remotes and git-sign-push CLIs
Ship the two maintainer workflow helpers as installable Python CLIs so
any fork-based OSS project can reuse them without vendoring shell
scripts:
* git-setup-remotes configures origin/fork/main-tracking/pushDefault
for a fork-based clone. URLs are parameterized (--canonical / --fork
or CANONICAL_URL / FORK_URL) so the same binary bootstraps any
maintainer's repo. Idempotent.
* git-sign-push GPG-signs every unpushed commit on the current branch
and pushes, resolving the target remote from remote.pushDefault
(falling back to origin) for branches without upstream.
Both refuse to run when CLAUDE_CODE/CLAUDECODE is set, since the Claude
sandbox blocks .git/config writes and access to ~/.gnupg: failing fast
beats failing late.
Other additions:
* Makefile: install / install-dev / test / lint / clean targets.
* .github/workflows/test.yml: pytest + ruff matrix for py 3.10/11/12.
* MIRRORS: github, git.veen.world:2201, code.infinito.nexus:2201, pypi.
* LICENSE switched to MIT; README records the extraction origin
(s.infinito.nexus/code) and adds author + license sections.
* Tests cover sandbox-refusal guards and the fork-URL resolution
preference order (CLI arg > env var > existing fork remote > origin
when not canonical).
This commit is contained in:
Kevin Veen-Birkenbach
4
src/git_maintainer_tools/__init__.py
Normal file
4
src/git_maintainer_tools/__init__.py
Normal file
@@ -0,0 +1,4 @@
|
||||
"""git-maintainer-tools: small CLIs for fork-based OSS maintainer workflows."""
|
||||
|
||||
__all__ = ["__version__"]
|
||||
__version__ = "0.1.0"
|
||||
130
src/git_maintainer_tools/_git.py
Normal file
130
src/git_maintainer_tools/_git.py
Normal file
@@ -0,0 +1,130 @@
|
||||
"""Shared helpers for interacting with git and the Claude sandbox."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import Iterable, Sequence
|
||||
|
||||
|
||||
class GitError(RuntimeError):
|
||||
"""Raised when a git subprocess returns a non-zero exit status."""
|
||||
|
||||
|
||||
def run_git(
|
||||
*args: str,
|
||||
check: bool = True,
|
||||
capture: bool = True,
|
||||
quiet: bool = False,
|
||||
) -> subprocess.CompletedProcess:
|
||||
"""Run `git <args>` and return the completed-process handle.
|
||||
|
||||
`check=True` raises `GitError` on non-zero exit. `capture=True` routes
|
||||
stdout/stderr through pipes so callers can inspect output; setting it
|
||||
to False lets git write straight to the terminal (useful for
|
||||
long-running commands like `git push` or `git rebase`).
|
||||
"""
|
||||
cmd: list[str] = ["git", *args]
|
||||
kwargs: dict = {}
|
||||
if capture:
|
||||
kwargs["stdout"] = subprocess.PIPE
|
||||
kwargs["stderr"] = subprocess.PIPE
|
||||
kwargs["text"] = True
|
||||
proc = subprocess.run(cmd, **kwargs)
|
||||
if check and proc.returncode != 0:
|
||||
stderr = (proc.stderr or "") if capture else ""
|
||||
if not quiet:
|
||||
sys.stderr.write(stderr)
|
||||
raise GitError(
|
||||
f"git {' '.join(args)} exited {proc.returncode}: {stderr.strip()}"
|
||||
)
|
||||
return proc
|
||||
|
||||
|
||||
def git_out(*args: str) -> str:
|
||||
"""Run a git command that is expected to succeed and return stdout (stripped)."""
|
||||
return run_git(*args).stdout.strip()
|
||||
|
||||
|
||||
def has_remote(name: str) -> bool:
|
||||
try:
|
||||
run_git("remote", "get-url", name, quiet=True)
|
||||
return True
|
||||
except GitError:
|
||||
return False
|
||||
|
||||
|
||||
def remote_url(name: str) -> str | None:
|
||||
if not has_remote(name):
|
||||
return None
|
||||
return git_out("remote", "get-url", name)
|
||||
|
||||
|
||||
def ensure_not_sandboxed(tool_name: str, reason: str) -> None:
|
||||
"""Abort the CLI when invoked inside the Claude sandbox.
|
||||
|
||||
Claude Code exports `CLAUDE_CODE` / `CLAUDECODE` inside the sandbox.
|
||||
Writes to `.git/config` (and `~/.gnupg` access) are blocked there per
|
||||
the Git Safety Protocol, so running these tools inside would always
|
||||
fail late. Fail fast instead.
|
||||
"""
|
||||
if os.environ.get("CLAUDE_CODE") or os.environ.get("CLAUDECODE"):
|
||||
sys.stderr.write(
|
||||
f"ERROR: {tool_name} must run outside the Claude sandbox ({reason}).\n"
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def ensure_clean_tree() -> None:
|
||||
if run_git("diff", "--quiet", check=False, capture=False).returncode != 0 \
|
||||
or run_git("diff", "--cached", "--quiet", check=False, capture=False).returncode != 0:
|
||||
sys.stderr.write(
|
||||
"ERROR: uncommitted changes present. Commit or stash before signing.\n"
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def current_branch() -> str:
|
||||
branch = git_out("rev-parse", "--abbrev-ref", "HEAD")
|
||||
if branch == "HEAD":
|
||||
sys.stderr.write("ERROR: detached HEAD.\n")
|
||||
sys.exit(1)
|
||||
return branch
|
||||
|
||||
|
||||
def first_existing_rev(candidates: Iterable[str]) -> str | None:
|
||||
for cand in candidates:
|
||||
proc = run_git("rev-parse", "--verify", "--quiet", cand, check=False)
|
||||
if proc.returncode == 0:
|
||||
return cand
|
||||
return None
|
||||
|
||||
|
||||
def resolve_env_or_arg(
|
||||
arg_value: str | None, env_var: str, label: str
|
||||
) -> str | None:
|
||||
"""Return the first non-empty value from the CLI arg or the env var."""
|
||||
if arg_value:
|
||||
return arg_value
|
||||
value = os.environ.get(env_var, "").strip()
|
||||
return value or None
|
||||
|
||||
|
||||
def abbrev_ref(rev: str) -> str | None:
|
||||
proc = run_git(
|
||||
"rev-parse", "--abbrev-ref", "--symbolic-full-name", rev, check=False
|
||||
)
|
||||
if proc.returncode != 0:
|
||||
return None
|
||||
value = (proc.stdout or "").strip()
|
||||
return value or None
|
||||
|
||||
|
||||
def rev_count(range_spec: str) -> int:
|
||||
return int(git_out("rev-list", "--count", range_spec))
|
||||
|
||||
|
||||
def format_cmd_preview(cmd: Sequence[str]) -> str:
|
||||
"""Render a command list for the help messages that tell the user what ran."""
|
||||
return " ".join(cmd)
|
||||
169
src/git_maintainer_tools/setup_remotes.py
Normal file
169
src/git_maintainer_tools/setup_remotes.py
Normal file
@@ -0,0 +1,169 @@
|
||||
"""`git-setup-remotes`: configure the fork-based remote layout.
|
||||
|
||||
Target state after a successful run:
|
||||
|
||||
* `origin` points at the canonical repository (the project's upstream).
|
||||
* `fork` points at the maintainer's personal fork.
|
||||
* `main` tracks `origin/main` so `git pull` takes from canonical.
|
||||
* `remote.pushDefault` = `fork`, `push.default` = `current` so every
|
||||
`git push` without args (and new-branch pushes via sign-push) lands
|
||||
on the fork, not on the canonical repo.
|
||||
|
||||
The tool is idempotent: re-running on a correctly configured repo is a
|
||||
no-op. URLs are taken from CLI args or environment variables; there is
|
||||
no hardcoded project-specific default so the same binary can bootstrap
|
||||
any maintainer's repo.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
|
||||
from . import _git as g
|
||||
|
||||
|
||||
TOOL_NAME = "git-setup-remotes"
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog=TOOL_NAME,
|
||||
description=__doc__.strip().splitlines()[0],
|
||||
)
|
||||
parser.add_argument(
|
||||
"--canonical",
|
||||
help=(
|
||||
"URL of the canonical repository (assigned to `origin`). "
|
||||
"Falls back to the $CANONICAL_URL environment variable."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--fork",
|
||||
help=(
|
||||
"URL of the maintainer's personal fork (assigned to `fork`). "
|
||||
"Falls back to the $FORK_URL environment variable; if neither "
|
||||
"is set, the current `fork` remote or a non-canonical `origin` "
|
||||
"is used."
|
||||
),
|
||||
)
|
||||
return parser
|
||||
|
||||
|
||||
def resolve_fork_url(arg_value: str | None, canonical_url: str) -> str | None:
|
||||
"""Preference order:
|
||||
1. --fork CLI arg or $FORK_URL env var.
|
||||
2. Existing `fork` remote (already configured).
|
||||
3. Existing `origin` remote *if* it does not point at canonical
|
||||
(clone-from-fork case: origin currently holds the fork URL).
|
||||
"""
|
||||
explicit = g.resolve_env_or_arg(arg_value, "FORK_URL", "--fork")
|
||||
if explicit:
|
||||
return explicit
|
||||
fork = g.remote_url("fork")
|
||||
if fork:
|
||||
return fork
|
||||
origin = g.remote_url("origin")
|
||||
if origin and origin != canonical_url:
|
||||
return origin
|
||||
return None
|
||||
|
||||
|
||||
def configure_origin(canonical_url: str) -> None:
|
||||
"""Ensure `origin` points at the canonical URL.
|
||||
|
||||
If `origin` already points somewhere else (typically the fork), move
|
||||
it aside. Use `git remote rename origin fork` when possible so that
|
||||
per-branch tracking refs under `remotes/origin/*` migrate to
|
||||
`remotes/fork/*` automatically.
|
||||
"""
|
||||
if not g.has_remote("origin"):
|
||||
g.run_git("remote", "add", "origin", canonical_url, capture=False)
|
||||
return
|
||||
|
||||
current = g.remote_url("origin")
|
||||
if current == canonical_url:
|
||||
return
|
||||
|
||||
if g.has_remote("fork"):
|
||||
# Both `origin` (= fork) and `fork` already exist. Drop the stale
|
||||
# `origin` entry and add canonical fresh.
|
||||
g.run_git("remote", "remove", "origin", capture=False)
|
||||
else:
|
||||
g.run_git("remote", "rename", "origin", "fork", capture=False)
|
||||
g.run_git("remote", "add", "origin", canonical_url, capture=False)
|
||||
|
||||
|
||||
def cleanup_legacy_upstream(canonical_url: str) -> None:
|
||||
"""Drop a legacy `upstream` remote when it duplicates `origin`."""
|
||||
if g.has_remote("upstream") and g.remote_url("upstream") == canonical_url:
|
||||
g.run_git("remote", "remove", "upstream", capture=False)
|
||||
|
||||
|
||||
def configure_fork(fork_url: str) -> None:
|
||||
if g.has_remote("fork"):
|
||||
if g.remote_url("fork") != fork_url:
|
||||
g.run_git("remote", "set-url", "fork", fork_url, capture=False)
|
||||
else:
|
||||
g.run_git("remote", "add", "fork", fork_url, capture=False)
|
||||
|
||||
|
||||
def wire_main_tracking() -> None:
|
||||
"""Point `main` at `origin/main` if both exist."""
|
||||
if g.run_git("rev-parse", "--verify", "--quiet", "main", check=False).returncode != 0:
|
||||
return
|
||||
if g.run_git("rev-parse", "--verify", "--quiet", "origin/main", check=False).returncode != 0:
|
||||
return
|
||||
g.run_git("branch", "--set-upstream-to=origin/main", "main", capture=False)
|
||||
|
||||
|
||||
def apply_push_config() -> None:
|
||||
g.run_git("config", "remote.pushDefault", "fork", capture=False)
|
||||
g.run_git("config", "push.default", "current", capture=False)
|
||||
|
||||
|
||||
def report_state() -> None:
|
||||
print("Remotes:")
|
||||
print(g.git_out("remote", "-v"))
|
||||
print()
|
||||
print(f"remote.pushDefault = {g.git_out('config', '--get', 'remote.pushDefault')}")
|
||||
print(f"push.default = {g.git_out('config', '--get', 'push.default')}")
|
||||
main_u = g.abbrev_ref("main@{u}")
|
||||
if main_u:
|
||||
print(f"main tracks = {main_u}")
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
g.ensure_not_sandboxed(TOOL_NAME, "it writes .git/config")
|
||||
|
||||
args = build_parser().parse_args(argv)
|
||||
canonical_url = g.resolve_env_or_arg(args.canonical, "CANONICAL_URL", "--canonical")
|
||||
if not canonical_url:
|
||||
sys.stderr.write(
|
||||
"ERROR: missing canonical URL. Pass --canonical URL or set CANONICAL_URL.\n"
|
||||
)
|
||||
return 1
|
||||
|
||||
fork_url = resolve_fork_url(args.fork, canonical_url)
|
||||
if not fork_url:
|
||||
sys.stderr.write(
|
||||
"ERROR: cannot determine the fork URL.\n"
|
||||
" Pass --fork URL or set FORK_URL, and re-run.\n"
|
||||
)
|
||||
return 1
|
||||
|
||||
configure_origin(canonical_url)
|
||||
cleanup_legacy_upstream(canonical_url)
|
||||
configure_fork(fork_url)
|
||||
|
||||
g.run_git("fetch", "--quiet", "origin", capture=False)
|
||||
g.run_git("fetch", "--quiet", "fork", check=False, capture=False)
|
||||
|
||||
wire_main_tracking()
|
||||
apply_push_config()
|
||||
report_state()
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
sys.exit(main())
|
||||
121
src/git_maintainer_tools/sign_push.py
Normal file
121
src/git_maintainer_tools/sign_push.py
Normal file
@@ -0,0 +1,121 @@
|
||||
"""`git-sign-push`: GPG-sign every unpushed commit on the current branch and push.
|
||||
|
||||
Intended to replace a direct `git push` from inside an agent sandbox so
|
||||
that only the operator's GPG key (in `~/.gnupg`, unreadable from the
|
||||
sandbox) ever signs shipped commits. For a branch with no upstream, the
|
||||
push target is resolved from `remote.pushDefault` (falling back to
|
||||
`origin`), so the fork-based workflow from `git-setup-remotes` routes
|
||||
new branches to the personal fork rather than the canonical repo.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
from . import _git as g
|
||||
|
||||
|
||||
TOOL_NAME = "git-sign-push"
|
||||
DEFAULT_PUSH_REMOTE = "origin"
|
||||
FALLBACK_BASE_REFS = ("origin/main", "origin/master")
|
||||
|
||||
|
||||
def resolve_push_remote() -> str:
|
||||
"""Return the configured push default, falling back to `origin`."""
|
||||
proc = g.run_git(
|
||||
"config", "--default", DEFAULT_PUSH_REMOTE,
|
||||
"--get", "remote.pushDefault",
|
||||
check=False,
|
||||
)
|
||||
if proc.returncode == 0:
|
||||
value = (proc.stdout or "").strip()
|
||||
if value:
|
||||
return value
|
||||
return DEFAULT_PUSH_REMOTE
|
||||
|
||||
|
||||
def resolve_base_commit() -> tuple[str, bool]:
|
||||
"""Return (base_commit, needs_upstream).
|
||||
|
||||
`needs_upstream` is True when the current branch has no upstream,
|
||||
which means the eventual push must be `git push -u <remote> <branch>`.
|
||||
"""
|
||||
upstream = g.abbrev_ref("@{u}")
|
||||
if upstream:
|
||||
return g.git_out("merge-base", "HEAD", upstream), False
|
||||
|
||||
base_ref = g.first_existing_rev(FALLBACK_BASE_REFS)
|
||||
if not base_ref:
|
||||
sys.stderr.write(
|
||||
"ERROR: could not determine base commit (no upstream, no "
|
||||
"origin/main|master).\n"
|
||||
)
|
||||
sys.exit(1)
|
||||
return g.git_out("merge-base", "HEAD", base_ref), True
|
||||
|
||||
|
||||
def count_unsigned(base: str) -> tuple[int, int]:
|
||||
"""Return (total_commits_in_range, unsigned_count)."""
|
||||
total = g.rev_count(f"{base}..HEAD")
|
||||
if total == 0:
|
||||
return 0, 0
|
||||
signed_status = g.git_out("log", "--format=%G?", f"{base}..HEAD").splitlines()
|
||||
unsigned = sum(1 for s in signed_status if s != "G")
|
||||
return total, unsigned
|
||||
|
||||
|
||||
def resign(base: str) -> None:
|
||||
"""Rewrite every commit in base..HEAD with a GPG signature.
|
||||
|
||||
`GIT_SEQUENCE_EDITOR=:` turns the interactive rebase TODO-list editor
|
||||
into a no-op so the rebase just replays the existing commits without
|
||||
reordering or squashing. `-S` signs each replayed commit.
|
||||
"""
|
||||
env = os.environ.copy()
|
||||
env["GIT_SEQUENCE_EDITOR"] = ":"
|
||||
import subprocess
|
||||
proc = subprocess.run(
|
||||
["git", "rebase", "--rebase-merges", "-S", base],
|
||||
env=env,
|
||||
)
|
||||
if proc.returncode != 0:
|
||||
sys.stderr.write("ERROR: GPG re-sign rebase failed.\n")
|
||||
sys.exit(proc.returncode)
|
||||
|
||||
|
||||
def push(push_remote: str, branch: str, needs_upstream: bool) -> None:
|
||||
if needs_upstream:
|
||||
g.run_git("push", "-u", push_remote, branch, capture=False)
|
||||
else:
|
||||
g.run_git("push", "--force-with-lease", capture=False)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
g.ensure_not_sandboxed(
|
||||
TOOL_NAME, "gpg-agent/pinentry need access to ~/.gnupg"
|
||||
)
|
||||
g.ensure_clean_tree()
|
||||
|
||||
branch = g.current_branch()
|
||||
g.run_git("fetch", "--quiet", "origin", capture=False)
|
||||
|
||||
base, needs_upstream = resolve_base_commit()
|
||||
total, unsigned = count_unsigned(base)
|
||||
if total == 0:
|
||||
print("Nothing to sign or push.")
|
||||
return 0
|
||||
|
||||
if unsigned > 0:
|
||||
print(f"Signing {unsigned} of {total} commit(s) in {base}..HEAD")
|
||||
resign(base)
|
||||
else:
|
||||
print(f"All {total} commit(s) already GPG-signed; skipping re-sign.")
|
||||
|
||||
push_remote = resolve_push_remote()
|
||||
push(push_remote, branch, needs_upstream)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user