govoplan/tools/repo/bootstrap-repositories.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import os
import re
import subprocess
import sys
from pathlib import Path
from urllib.parse import urlsplit


ROOT = Path(__file__).resolve().parents[2]
REGISTERED_TRANSPORT = "registered"
PUBLIC_HTTPS_TRANSPORT = "public-https"
GITEA_SSH_PREFIX = "git@git.add-ideas.de:"
GITEA_HTTPS_PREFIX = "https://git.add-ideas.de/"
GITEA_CHECKOUT_AUTH_KEY = "http.https://git.add-ideas.de/.extraheader"
GITEA_REPOSITORY_PATH = re.compile(r"(?:GovOPlaN|add-ideas)/[a-z0-9][a-z0-9-]*[.]git")


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Clone GovOPlaN repositories listed in repositories.json."
    )
    parser.add_argument(
        "--check", action="store_true", help="Only report missing repositories."
    )
    parser.add_argument(
        "--parent", type=Path, help="Override checkout parent directory."
    )
    parser.add_argument(
        "--repo",
        action="append",
        default=[],
        metavar="NAME",
        help="Clone only this registered repository; repeat for more than one.",
    )
    parser.add_argument(
        "--exclude-repo",
        action="append",
        default=[],
        metavar="NAME",
        help="Exclude this registered repository; repeat for more than one.",
    )
    parser.add_argument(
        "--transport",
        choices=(REGISTERED_TRANSPORT, PUBLIC_HTTPS_TRANSPORT),
        default=REGISTERED_TRANSPORT,
        help=(
            "Clone registered remotes unchanged, or convert public "
            "git.add-ideas.de SSH remotes to anonymous HTTPS."
        ),
    )
    parser.add_argument(
        "--reuse-checkout-auth",
        action="store_true",
        help=(
            "Reuse the checkout repository's short-lived Gitea HTTP auth "
            "header for child clones without printing or persisting it."
        ),
    )
    return parser.parse_args(argv)


def clone_remote(remote: str, *, transport: str) -> str:
    if transport == REGISTERED_TRANSPORT:
        return remote
    if transport != PUBLIC_HTTPS_TRANSPORT:
        raise ValueError(f"unsupported repository transport: {transport}")
    if remote.startswith(GITEA_SSH_PREFIX):
        repository_path = remote.removeprefix(GITEA_SSH_PREFIX)
    else:
        parsed = urlsplit(remote)
        if (
            parsed.scheme not in {"https", "ssh"}
            or parsed.hostname != "git.add-ideas.de"
            or parsed.port is not None
            or parsed.password is not None
            or parsed.query
            or parsed.fragment
        ):
            raise ValueError(
                "public HTTPS bootstrap only accepts registered "
                "git.add-ideas.de remotes"
            )
        if parsed.scheme == "https" and parsed.username is not None:
            raise ValueError("public HTTPS bootstrap does not accept credentials")
        if parsed.scheme == "ssh" and parsed.username != "git":
            raise ValueError("registered Gitea SSH remotes must use the git account")
        repository_path = parsed.path.removeprefix("/")
    if GITEA_REPOSITORY_PATH.fullmatch(repository_path) is None:
        raise ValueError("registered Gitea repository path is malformed")
    return GITEA_HTTPS_PREFIX + repository_path


def _add_checkout_auth(environment: dict[str, str], *, root: Path) -> None:
    result = subprocess.run(
        [
            "git",
            "-C",
            str(root),
            "config",
            "--local",
            "--get",
            GITEA_CHECKOUT_AUTH_KEY,
        ],
        check=False,
        capture_output=True,
        text=True,
    )
    auth_header = result.stdout.strip()
    if result.returncode != 0 or not auth_header:
        raise ValueError(
            "checkout authentication is unavailable; ensure actions/checkout "
            "persists the GITEA_TOKEN credentials"
        )
    if re.fullmatch(
        r"authorization: basic [A-Za-z0-9+/=]+",
        auth_header,
        flags=re.IGNORECASE,
    ) is None:
        raise ValueError("checkout authentication has an unsupported format")

    try:
        config_count = int(environment.get("GIT_CONFIG_COUNT", "0"))
    except ValueError as exc:
        raise ValueError("GIT_CONFIG_COUNT must be an integer") from exc
    if config_count < 0:
        raise ValueError("GIT_CONFIG_COUNT cannot be negative")
    environment[f"GIT_CONFIG_KEY_{config_count}"] = GITEA_CHECKOUT_AUTH_KEY
    environment[f"GIT_CONFIG_VALUE_{config_count}"] = auth_header
    environment["GIT_CONFIG_COUNT"] = str(config_count + 1)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    manifest = json.loads((ROOT / "repositories.json").read_text(encoding="utf-8"))
    parent = args.parent or Path(manifest["default_parent"])
    missing: list[tuple[dict[str, str], Path, str]] = []
    repositories = manifest["repositories"]
    names = [entry["name"] for entry in repositories]
    if len(names) != len(set(names)):
        raise ValueError("repository registry contains duplicate names")
    requested = set(args.repo)
    excluded = set(args.exclude_repo)
    unknown = (requested | excluded) - set(names)
    if unknown:
        raise ValueError(
            f"unknown registered repositories: {', '.join(sorted(unknown))}"
        )
    overlap = requested & excluded
    if overlap:
        raise ValueError(
            f"repositories cannot be selected and excluded: {', '.join(sorted(overlap))}"
        )

    for entry in repositories:
        if (requested and entry["name"] not in requested) or entry["name"] in excluded:
            continue
        repo = parent / entry["path"]
        if repo.exists():
            continue
        transport = entry.get("bootstrap_transport", args.transport)
        remote = clone_remote(entry["remote"], transport=transport)
        missing.append((entry, repo, remote))

    environment = os.environ.copy()
    environment.update(
        {
            "GIT_ASKPASS": "/bin/false",
            "GIT_TERMINAL_PROMPT": "0",
        }
    )
    if args.reuse_checkout_auth and missing and not args.check:
        _add_checkout_auth(environment, root=ROOT)
    for entry, repo, remote in missing:
        print(f"missing: {entry['name']} -> {repo}")
        if not args.check:
            subprocess.run(
                [
                    "git",
                    "-c",
                    "credential.helper=",
                    "clone",
                    "--",
                    remote,
                    str(repo),
                ],
                check=True,
                env=environment,
            )

    if args.check:
        return 1 if missing else 0
    return 0


if __name__ == "__main__":
    sys.exit(main())