SmartestHome/workshop/gitea.py

"""Gitea repositories for workshop projects.

The workshop assistant produces things that want version control — firmware, scripts,
KiCad files, a config it has been iterating on — and this household already runs Gitea.
So a project can have a repository, created on request, recorded against the project.

WHAT THIS DOES
--------------
Creates repositories, reports what exists, and — the part that matters —
**applies branch protection at creation so history cannot be rewritten.** Committing
and pushing live in `git_ops.py`; the assistant does both, unattended.

The line is not "may it write to git" but "may it *remove* anything": adding to
history is allowed because the worst case is a bad commit you revert, while rewriting
history has no undo. `protect_branch()` below is what makes that a control rather than
a promise — Gitea refuses the force-push regardless of what asked for it, including a
human who typed `--force` out of habit.

The git working tree is `code/` inside the project workspace, never the workspace
root, so no git operation of any kind has a path to the datasheets and photos sitting
beside it.

THE TOKEN
---------
Gitea needs an **API token** with permission to create repositories — Settings →
Applications → Generate New Token, scope `write:repository` (plus `write:user` if you
want it to create repos under your own account rather than an organisation). That
token can create and delete repositories, so:

- It lives in `workshop.env`, 600, like every other credential in this project.
- **Repository deletion is not implemented here at all**, and neither is any history
  rewrite. The token can technically do both; nothing in this service gives them a
  path. When history genuinely has to be scrubbed — a committed token — see
  `git_ops.scrub_instructions()`, which writes the commands out for you to run by
  hand.
- Prefer a token on a **dedicated Gitea user** ("workshop-bot") with access to one
  organisation, over one on your own account with access to everything you own. The
  blast radius of a leaked env file is then one org of generated repos.
"""

from __future__ import annotations

import json
import logging
import os
import re
import urllib.error
import urllib.request

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

GITEA_URL = os.environ.get("GITEA_URL", "").rstrip("/")
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
# Which account or organisation new repositories belong to. Empty means the token's own
# user, which works and is the worse default — see the module docstring.
GITEA_OWNER = os.environ.get("GITEA_OWNER", "").strip()
GITEA_PRIVATE = os.environ.get("GITEA_PRIVATE_REPOS", "true").strip().lower() != "false"
GITEA_TIMEOUT = float(os.environ.get("GITEA_TIMEOUT", "15"))

# Gitea's own constraint on repository names, applied here so a bad name is a clear
# refusal from this service rather than a 422 from somewhere else.
REPO_NAME_RE = re.compile(r"^[A-Za-z0-9._-]{1,100}$")


def configured() -> bool:
    return bool(GITEA_URL and GITEA_TOKEN)


def _request(method: str, path: str, payload: dict | None = None) -> tuple[int, dict]:
    data = json.dumps(payload).encode() if payload is not None else None
    req = urllib.request.Request(f"{GITEA_URL}/api/v1{path}", data=data, method=method)
    req.add_header("Content-Type", "application/json")
    # Gitea accepts `Authorization: token <t>`; the bearer form is not universally
    # honoured across versions, so use the documented one.
    req.add_header("Authorization", f"token {GITEA_TOKEN}")
    try:
        with urllib.request.urlopen(req, timeout=GITEA_TIMEOUT) as resp:
            body = resp.read()
            return resp.status, (json.loads(body) if body else {})
    except urllib.error.HTTPError as exc:
        body = exc.read()
        try:
            parsed = json.loads(body) if body else {}
        except ValueError:
            parsed = {"message": body.decode("utf-8", "replace")[:300]}
        return exc.code, parsed


def create_repo(name: str, description: str = "", private: bool | None = None) -> dict:
    """Create a repository, or report the one that is already there.

    **An existing repository of that name is a success, not a conflict.** Asking twice
    means the second caller wanted a repo to exist, which it does — and returning an
    error there would make every retry look like a failure while leaving the assistant
    with no URL to record.
    """
    if not configured():
        return {
            "ok": False,
            "reason": "not_configured",
            "message": "GITEA_URL and GITEA_TOKEN are not set — see workshop.env.example.",
        }
    name = str(name or "").strip()
    if not REPO_NAME_RE.match(name):
        return {
            "ok": False,
            "reason": "bad_name",
            "message": "A repository name is letters, digits, dot, dash or underscore.",
        }

    payload = {
        "name": name,
        "description": str(description or "")[:255],
        "private": GITEA_PRIVATE if private is None else bool(private),
        # An empty repo is the right starting point: the workspace already has the
        # files, and an auto-generated README is the first merge conflict.
        "auto_init": False,
    }
    path = f"/orgs/{GITEA_OWNER}/repos" if GITEA_OWNER else "/user/repos"
    status, body = _request("POST", path, payload)

    if status in (200, 201):
        LOG.info("workshop: created Gitea repository %s", body.get("full_name") or name)
        summary = _repo_summary(body)
        summary["protection"] = protect_branch(body.get("owner", {}).get("login", ""), name)
        return {"ok": True, "created": True, "repo": summary}

    if status == 409:
        existing = get_repo(name)
        if existing.get("ok"):
            return {"ok": True, "created": False, "repo": existing["repo"]}

    LOG.warning("workshop: Gitea repo creation failed (%s): %s", status, body.get("message"))
    return {
        "ok": False,
        "reason": "gitea_error",
        "status": status,
        "message": body.get("message") or f"Gitea returned {status}.",
    }


def protect_branch(owner: str, repo: str, branch: str = "") -> dict:
    """Refuse force-pushes and branch deletion on the default branch, server-side.

    THIS IS THE ENFORCEMENT. `git_ops.py` refuses to rewrite history, but that is a
    convention any process holding the same token could ignore. Branch protection is
    the control: Gitea rejects the push regardless of what asked for it, including a
    human at a terminal who typed --force out of habit.

    Applied at creation because retrofitting it means a window where it wasn't on, and
    that window is exactly when a new repo gets its messy first pushes.

    Best-effort: a failure here is reported, not raised. A repo without protection is
    still a usable repo, and refusing to create one would trade a real capability for a
    guarantee that git_ops.py already provides in the normal case. **The response says
    which you got** — never assume protection is on because a repo exists.

    VERIFY against your Gitea: the branch-protection payload's field names have changed
    across versions (`branch_name` vs `rule_name` in particular), and this is written
    from the documented shape, not against a live instance.
    """
    if not owner:
        return {"applied": False, "reason": "owner unknown"}
    payload = {
        "rule_name": branch or "main",
        "branch_name": branch or "main",
        "enable_push": True,
        # The two that matter.
        "enable_force_push": False,
        "enable_delete": False,
    }
    status, body = _request("POST", f"/repos/{owner}/{repo}/branch_protections", payload)
    if status in (200, 201):
        return {"applied": True, "branch": payload["rule_name"]}
    LOG.warning(
        "workshop: could not apply branch protection to %s/%s (%s): %s — history is protected "
        "only by this service's own refusal until you set it in Gitea by hand",
        owner, repo, status, body.get("message"),
    )
    return {"applied": False, "status": status, "message": body.get("message")}


def get_repo(name: str) -> dict:
    if not configured():
        return {"ok": False, "reason": "not_configured", "message": "Gitea is not configured."}
    owner = GITEA_OWNER
    if not owner:
        status, user = _request("GET", "/user")
        if status != 200:
            return {"ok": False, "reason": "gitea_error", "message": "Could not resolve the token's own user."}
        owner = user.get("login", "")
    status, body = _request("GET", f"/repos/{owner}/{name}")
    if status == 200:
        return {"ok": True, "repo": _repo_summary(body)}
    return {"ok": False, "reason": "not_found", "message": f"No repository {owner}/{name}."}


def _repo_summary(body: dict) -> dict:
    """Only the fields anything here needs. Gitea's repo object is ~60 keys, and
    storing or echoing all of them would make this service's API shape hostage to
    Gitea's."""
    return {
        "name": body.get("name"),
        "full_name": body.get("full_name"),
        "html_url": body.get("html_url"),
        "clone_url": body.get("clone_url"),
        "ssh_url": body.get("ssh_url"),
        "private": body.get("private"),
    }