atlas-iac/services/gitea/scripts/gitea_branch_protection_check.py

#!/usr/bin/env python3
"""Validate the first effective Gitea protection for an Atlas primary branch."""

from __future__ import annotations

import argparse
import datetime
import json
import os
import stat
from pathlib import Path

MAX_INPUT = 1024 * 1024
MAX_RULES = 100
MAX_RULE_NAME = 255
SPECIAL = frozenset("*?\\[]{}")


class PolicyError(RuntimeError):
    """The protection response is ambiguous or violates human review policy."""


def _read_bounded(path: Path) -> bytes:
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
    try:
        metadata = os.fstat(descriptor)
        if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > MAX_INPUT:
            raise PolicyError("branch protection response exceeds the safe limit")
        value = os.read(descriptor, MAX_INPUT + 1)
    finally:
        os.close(descriptor)
    if len(value) != metadata.st_size:
        raise PolicyError("branch protection response changed while reading")
    return value


def _is_plain(pattern: str) -> bool:
    return not any(character in SPECIAL for character in pattern)


def _literal_rule(pattern: str) -> str:
    """Return one exact rule name or reject Gitea glob interpretation.

    Gitea 1.23.8 delegates matching to gobwas/glob.  Reimplementing that
    grammar and its error quoting rules here would create a second policy
    engine.  The reconciler instead accepts only plain ASCII literals.  This
    deliberately rejects braces, ranges, escapes, malformed globs, and even
    special rules that appear unrelated: an operator must remove ambiguity
    before this job may attest the primary-branch boundary.
    """
    if not pattern.isascii() or not 1 <= len(pattern) <= MAX_RULE_NAME:
        raise PolicyError("branch protection rule name is invalid")
    if not _is_plain(pattern):
        raise PolicyError("glob branch protection rules require human review")
    return pattern


def _created(value: object) -> datetime.datetime:
    if not isinstance(value, str) or len(value) > 64:
        raise PolicyError("branch protection creation time is invalid")
    try:
        result = datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise PolicyError("branch protection creation time is invalid") from exc
    if result.tzinfo is None:
        raise PolicyError("branch protection creation time is invalid")
    return result


def _required(reviewer: str) -> dict[str, object]:
    return {
        "enable_push": True,
        "enable_push_whitelist": True,
        "push_whitelist_usernames": [reviewer],
        "push_whitelist_deploy_keys": False,
        "enable_force_push": False,
        "enable_merge_whitelist": True,
        "merge_whitelist_usernames": [reviewer],
        "enable_approvals_whitelist": True,
        "approvals_whitelist_username": [reviewer],
        "required_approvals": 1,
        "block_on_rejected_reviews": True,
        "block_on_outdated_branch": True,
        "dismiss_stale_approvals": True,
        "block_admin_merge_override": True,
    }


def evaluate(value: bytes, branch: str, reviewer: str) -> str:
    """Return PRESENT/ABSENT or fail on ambiguous/effective policy drift."""
    if len(value) > MAX_INPUT or branch not in {"main", "master"}:
        raise PolicyError("branch protection input is invalid")
    data = json.loads(value)
    if not isinstance(data, list) or len(data) > MAX_RULES:
        raise PolicyError("branch protection response is invalid")
    ordered: list[tuple[int, datetime.datetime, int, dict[str, object]]] = []
    for index, raw in enumerate(data):
        if not isinstance(raw, dict):
            raise PolicyError("branch protection entry is invalid")
        priority = raw.get("priority")
        rule_name = raw.get("rule_name")
        if (
            not isinstance(priority, int)
            or isinstance(priority, bool)
            or not 0 <= priority <= 1_000_000
            or not isinstance(rule_name, str)
        ):
            raise PolicyError("branch protection priority is ambiguous")
        created = _created(raw.get("created_at"))
        literal = _literal_rule(rule_name)
        if literal == branch:
            # With every rule constrained to a plain literal, Gitea's glob
            # comparator cannot affect this result.  Preserve API ordering:
            # priority first (including legacy zero), creation time, then the
            # stable response order for an otherwise exact tie.
            ordered.append((priority, created, index, raw))
    if not ordered:
        return "ABSENT"
    if len(ordered) != 1:
        raise PolicyError("duplicate primary-branch protections are ambiguous")
    effective = ordered[0][3]
    for key, expected in _required(reviewer).items():
        if effective.get(key) != expected:
            raise PolicyError(
                f"effective {branch} protection differs from human-review policy"
            )
    return "PRESENT"


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("response", type=Path)
    parser.add_argument("branch", choices=("main", "master"))
    parser.add_argument("reviewer")
    args = parser.parse_args()
    try:
        value = _read_bounded(args.response)
        print(evaluate(value, args.branch, args.reviewer))
    except (OSError, json.JSONDecodeError, PolicyError) as exc:
        print(f"branch protection check failed: {exc}", file=__import__("sys").stderr)
        return 1
    return 0


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