aios/scripts/verify_gitea_credential_rotation.py

#!/usr/bin/env python3
"""Verify a Gitea credential rotation without printing either credential."""
from __future__ import annotations

import argparse
import json
import subprocess
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path


def token_at(path: Path, line: int) -> str:
    lines = path.read_text(encoding="utf-8").splitlines()
    if line < 1 or line > len(lines) or not lines[line - 1].strip():
        raise RuntimeError("credential reference has no configured token line")
    return lines[line - 1].strip()


def get(api: str, path: str, token: str) -> tuple[int, object | None]:
    request = urllib.request.Request(
        api.rstrip("/") + path,
        headers={
            "Accept": "application/json",
            "Authorization": f"token {token}",
            "User-Agent": "aios-gitops-credential-rotation-verifier/1",
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            return response.status, json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        return exc.code, None


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--old-token-file", type=Path, required=True)
    parser.add_argument("--new-credential-file", type=Path, required=True)
    parser.add_argument("--new-token-line", type=int, default=2)
    parser.add_argument("--api", default="http://127.0.0.1:3000/api/v1")
    parser.add_argument("--repository", default="aios/aios")
    parser.add_argument("--version", default="V2R1C2")
    parser.add_argument("--workspace", type=Path, required=True)
    parser.add_argument("--remote", default="http://127.0.0.1:3000/aios/aios.git")
    parser.add_argument(
        "--branch",
        default="patch/gitea-tag-publication-atomicity-20260723",
    )
    args = parser.parse_args()
    old_token = token_at(args.old_token_file, 1)
    new_token = token_at(args.new_credential_file, args.new_token_line)
    repo_path = f"/repos/{args.repository}"
    old_status, _ = get(args.api, repo_path, old_token)
    new_status, _ = get(args.api, repo_path, new_token)
    encoded = urllib.parse.quote(args.version, safe="")
    ref_status, ref = get(
        args.api,
        f"{repo_path}/git/refs/tags/{encoded}",
        new_token,
    )
    listing_status, listing = get(
        args.api,
        f"{repo_path}/tags?limit=50&page=1",
        new_token,
    )
    refs = ref if isinstance(ref, list) else [ref]
    exact = [
        item
        for item in refs
        if isinstance(item, dict)
        and item.get("ref") in {
            f"refs/tags/{args.version}",
            f"tags/{args.version}",
        }
    ]
    tag_oid = (
        str(((exact[0].get("object") or {}).get("sha")) or "").lower()
        if len(exact) == 1
        else None
    )
    tag_status, tag = (
        get(args.api, f"{repo_path}/git/tags/{tag_oid}", new_token)
        if tag_oid
        else (0, None)
    )
    peeled_target = (
        str(((tag.get("object") or {}).get("sha")) or "").lower()
        if isinstance(tag, dict)
        else None
    )
    listing_target = None
    if isinstance(listing, list):
        for item in listing:
            if item.get("name") == args.version:
                listing_target = str(
                    ((item.get("commit") or {}).get("sha")) or ""
                ).lower()
                break
    remote = subprocess.run(
        [
            "git",
            "-c",
            f"safe.directory={args.workspace}",
            "-c",
            "credential.helper=!/usr/local/libexec/aios-gitea-credential",
            "-c",
            "credential.useHttpPath=true",
            "-C",
            str(args.workspace),
            "ls-remote",
            args.remote,
            f"refs/heads/{args.branch}",
        ],
        text=True,
        capture_output=True,
    )
    remote_oid = (
        remote.stdout.split()[0].lower()
        if remote.returncode == 0 and remote.stdout.split()
        else None
    )
    result = {
        "old_token_repository_http": old_status,
        "new_token_repository_http": new_status,
        "git_ls_remote_oid": remote_oid,
        "gitea_ref_http": ref_status,
        "gitea_tag_oid": tag_oid,
        "gitea_tag_object_http": tag_status,
        "gitea_peeled_target_oid": peeled_target,
        "gitea_listing_http": listing_status,
        "gitea_listing_target_oid": listing_target,
        "credentials_printed": False,
    }
    print(json.dumps(result, indent=2, sort_keys=True))
    passed = (
        old_status in {401, 403}
        and new_status == 200
        and remote_oid is not None
        and ref_status == 200
        and tag_oid is not None
        and tag_status == 200
        and peeled_target is not None
        and listing_status == 200
        and listing_target is not None
    )
    return 0 if passed else 2


if __name__ == "__main__":
    raise SystemExit(main())