SmartestHome/workshop/git_ops.py

"""Git for workshop projects: the assistant may write history, never rewrite it.

THE RULE, STATED ONCE
---------------------
**Everything that ADDS to history is allowed. Nothing that REMOVES from it is.**

Commit, push, branch, tag, merge: yes, unattended. Force-push, rebase, amend, reset
--hard, filter-branch, filter-repo, branch deletion, tag deletion, reflog expiry, gc
--prune: never, by any path, for any reason, including a good one.

The reasoning is a rollback guarantee. If history is append-only, then whatever the
assistant did wrong is *recoverable by looking at an earlier commit* — the worst case
is a bad commit you revert. The moment rewriting is on the table, the worst case
becomes work that no longer exists anywhere, and no amount of care makes that
recoverable after the fact.

WHEN HISTORY GENUINELY HAS TO BE SCRUBBED
------------------------------------------
It does happen — an API token committed by accident is the real case, and leaving it
in history is worse than the rewrite. So this module **writes the commands out and
hands them to you**, and you run them yourself from the repository on the SMB share.
See scrub_instructions().

That is not theatre. The person running `git filter-repo` can see what is about to
happen, has the repo in front of them, and can take a copy first. A service doing it
on a timer at 3am cannot. The manual step *is* the safety mechanism, and automating it
would remove the only thing making it safe.

CLIENT-SIDE REFUSAL IS A POLICY, NOT A CONTROL
-----------------------------------------------
Everything in this file is a convention that a different process with the same token
could ignore. The actual guarantee is **Gitea branch protection**, which is applied on
repo creation (see gitea.py's protect_branch) and which refuses force-pushes and
deletions server-side. This module and that protection say the same thing twice, on
purpose: one of them is the intent, the other is the enforcement.
"""

from __future__ import annotations

import logging
import os
import re
import shlex
import subprocess
from pathlib import Path

LOG = logging.getLogger("workshop.git")

GIT_AUTHOR_NAME = os.environ.get("WORKSHOP_GIT_AUTHOR_NAME", "workshop assistant")
GIT_AUTHOR_EMAIL = os.environ.get("WORKSHOP_GIT_AUTHOR_EMAIL", "workshop@localhost")
GIT_TIMEOUT = float(os.environ.get("WORKSHOP_GIT_TIMEOUT", "120"))
GIT_DEFAULT_BRANCH = os.environ.get("WORKSHOP_GIT_BRANCH", "main")

# Every one of these either destroys history or can be made to. Matched against the
# whole argument list of any git invocation this module builds, as a backstop against
# a future caller assembling one from parameters — the deny-list is cheap and the
# thing it prevents is not recoverable.
FORBIDDEN_ARGS = {
    "--force", "-f", "--force-with-lease", "--force-if-includes",
    "filter-branch", "filter-repo", "rebase", "reset", "gc", "prune", "reflog",
    "--amend", "--hard", "--delete", "--prune", "-D", "--allow-unrelated-histories",
}
# `git push --delete <ref>` and `git branch -d` are spelled without the tokens above in
# some forms, so subcommands that only ever remove are refused outright.
FORBIDDEN_SUBCOMMANDS = {"filter-branch", "filter-repo", "rebase", "reset", "gc", "reflog", "replace"}


class HistoryRewriteRefused(RuntimeError):
    """Raised when a command would remove history. Never caught-and-continued."""


def _check(args: list[str]) -> None:
    if args and args[0] in FORBIDDEN_SUBCOMMANDS:
        raise HistoryRewriteRefused(f"`git {args[0]}` rewrites history and is never run by this service")
    for arg in args:
        if arg in FORBIDDEN_ARGS:
            raise HistoryRewriteRefused(f"`{arg}` can remove history and is never passed by this service")


def _git(repo: Path, args: list[str]) -> tuple[int, str, str]:
    _check(args)
    env = {
        **os.environ,
        "GIT_AUTHOR_NAME": GIT_AUTHOR_NAME,
        "GIT_AUTHOR_EMAIL": GIT_AUTHOR_EMAIL,
        "GIT_COMMITTER_NAME": GIT_AUTHOR_NAME,
        "GIT_COMMITTER_EMAIL": GIT_AUTHOR_EMAIL,
        # Never sit waiting for a passphrase or a username on a service with no tty.
        "GIT_TERMINAL_PROMPT": "0",
    }
    proc = subprocess.run(
        ["git", *args], cwd=repo, env=env, capture_output=True, text=True, timeout=GIT_TIMEOUT
    )
    return proc.returncode, proc.stdout.strip(), proc.stderr.strip()


def _code_dir(project_dir: Path) -> Path:
    """The git working tree is `code/`, NOT the project root.

    The rest of the workspace — datasheets, photos, generated diagrams — deliberately
    stays outside version control. They are large, binary, and already backed by the
    share; more importantly, keeping them out means no git operation of any kind has a
    path to them. `git clean` cannot reach a datasheet that was never in the repo.
    """
    return project_dir / "code"


def ensure_repo(project_dir: Path, remote_url: str = "") -> dict:
    """Initialise `code/` as a repository if it isn't one, and set its remote."""
    code = _code_dir(project_dir)
    try:
        code.mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        return {"ok": False, "reason": "workspace_unavailable", "message": str(exc)}

    if not (code / ".git").is_dir():
        rc, _, err = _git(code, ["init", "-b", GIT_DEFAULT_BRANCH])
        if rc != 0:
            return {"ok": False, "reason": "git_error", "message": err}
        LOG.info("workshop: initialised git repo at %s", code)

    if remote_url:
        rc, out, _ = _git(code, ["remote"])
        if "origin" in out.split():
            _git(code, ["remote", "set-url", "origin", remote_url])
        else:
            _git(code, ["remote", "add", "origin", remote_url])
    return {"ok": True, "path": str(code)}


def status(project_dir: Path) -> dict:
    code = _code_dir(project_dir)
    if not (code / ".git").is_dir():
        return {"ok": True, "initialised": False, "path": str(code)}
    rc, changed, _ = _git(code, ["status", "--porcelain"])
    # `branch --show-current`, not `rev-parse --abbrev-ref HEAD`: on a repo with no
    # commits yet the latter reports the literal string "HEAD", which reads on screen
    # as a detached head rather than as "brand new, nothing committed".
    _, branch, _ = _git(code, ["branch", "--show-current"])
    _, log, _ = _git(code, ["log", "-5", "--pretty=%h %s"])
    return {
        "ok": rc == 0,
        "initialised": True,
        "path": str(code),
        "branch": branch,
        "dirty": bool(changed),
        "changed_files": [line[3:] for line in changed.splitlines() if line[3:]],
        "recent": log.splitlines(),
    }


def commit(project_dir: Path, message: str, push: bool = True) -> dict:
    """Stage everything under `code/`, commit, and push. Additive only.

    No `--amend`, ever, including for "just a typo in the message" — an amend after a
    push is a force-push waiting to happen, and a typo in a commit message costs
    nothing compared to the rule staying simple enough to be true.
    """
    code = _code_dir(project_dir)
    if not (code / ".git").is_dir():
        return {"ok": False, "reason": "not_initialised", "message": "No repository yet — create one first."}
    message = str(message or "").strip()
    if not message:
        return {"ok": False, "reason": "bad_field", "message": "A commit message is required."}

    rc, _, err = _git(code, ["add", "-A"])
    if rc != 0:
        return {"ok": False, "reason": "git_error", "message": err}

    rc, changed, _ = _git(code, ["status", "--porcelain"])
    if not changed:
        return {"ok": True, "committed": False, "message": "Nothing to commit."}

    rc, out, err = _git(code, ["commit", "-m", message])
    if rc != 0:
        return {"ok": False, "reason": "git_error", "message": err or out}
    result = {"ok": True, "committed": True, "detail": out.splitlines()[:2]}

    if push:
        rc, out, err = _git(code, ["push", "-u", "origin", "HEAD"])
        result["pushed"] = rc == 0
        if rc != 0:
            # A failed push is not a failed commit — the work is safe locally on the
            # share, which is the whole reason committing and pushing are reported
            # separately rather than as one boolean.
            result["push_error"] = err or out
            LOG.warning("workshop: commit succeeded but push failed: %s", err or out)
    return result


def branch(project_dir: Path, name: str) -> dict:
    """Create and switch to a branch. Creating only — there is no delete counterpart
    in this module, because deleting a branch is how commits stop being reachable."""
    code = _code_dir(project_dir)
    if not re.match(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,100}$", str(name or "")):
        return {"ok": False, "reason": "bad_field", "message": "That isn't a usable branch name."}
    rc, out, err = _git(code, ["switch", "-c", name])
    return {"ok": rc == 0, "detail": out or err}


def scrub_instructions(project_dir: Path, secret_path: str = "", note: str = "") -> dict:
    """The commands to remove something from history — printed, never run.

    Called when something that must not be in history is found in it: a token, a key, a
    password. This service will not do it. What it does is write out exactly what to
    run, from the repository on the share, so the person doing it can read it first and
    take a copy.

    The instructions deliberately lead with "rotate the secret first". Scrubbing a
    leaked token from git is the *second* thing to do — the commit was pushed, and a
    token that has been on a server for an hour must be assumed compromised whether or
    not the history is cleaned afterwards. A scrub that makes people feel finished
    without rotating is worse than no scrub.
    """
    code = _code_dir(project_dir)
    target = shlex.quote(secret_path) if secret_path else "PATH/TO/FILE"
    return {
        "ok": True,
        "manual": True,
        "why": (
            "Removing anything from history is never done by this service — see git_ops.py. "
            "Run these yourself, from the repository on the share, so you can read them "
            "first and take a copy."
        ),
        "note": note or "",
        "repo_path": str(code),
        "steps": [
            {
                "order": 1,
                "do": "ROTATE THE SECRET FIRST.",
                "why": "It was pushed. Assume it is compromised whether or not you clean the history. "
                       "Everything below is damage limitation, not a fix.",
                "command": "",
            },
            {
                "order": 2,
                "do": "Take a copy of the repository before touching it.",
                "why": "A rewrite is the one operation with no undo. A copy is the undo.",
                "command": f"cp -a {shlex.quote(str(code))} {shlex.quote(str(code) + '.backup')}",
            },
            {
                "order": 3,
                "do": "Remove the file from every commit.",
                "why": "git-filter-repo is the maintained tool; filter-branch is deprecated and slower.",
                "command": f"cd {shlex.quote(str(code))} && git filter-repo --invert-paths --path {target}",
            },
            {
                "order": 4,
                "do": "Re-add the remote and force-push.",
                "why": "filter-repo drops the remote on purpose, so that this step has to be deliberate.",
                "command": "git remote add origin <URL> && git push --force --all && git push --force --tags",
            },
            {
                "order": 5,
                "do": "Turn branch protection back on in Gitea if step 4 required turning it off.",
                "why": "Protection is what stops this from being possible unattended. Leaving it off "
                       "quietly removes the guarantee the whole arrangement rests on.",
                "command": "",
            },
        ],
    }