#!/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 re 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 _glob_regex(pattern: str, position: int = 0, terminators: str = "") -> tuple[str, int]: """Compile the gobwas/glob syntax Gitea v1.23.8 uses.""" pieces: list[str] = [] while position < len(pattern): character = pattern[position] if character in terminators: break if character == "\\": position += 1 if position >= len(pattern): raise ValueError("trailing escape") pieces.append(re.escape(pattern[position])) elif character == "*": if position + 1 < len(pattern) and pattern[position + 1] == "*": pieces.append(".*") position += 1 else: pieces.append("[^/]*") elif character == "?": pieces.append("[^/]") elif character == "[": end = pattern.find("]", position + 1) if end < 0: raise ValueError("unterminated range") value = pattern[position + 1 : end] if not value: raise ValueError("empty range") negate = value.startswith("!") value = value[1:] if negate else value if not value or "[" in value or "\\" in value: raise ValueError("invalid range") if "-" in value: if value.count("-") != 1 or value.startswith("-") or value.endswith("-"): raise ValueError("invalid range") low, high = value.split("-") if len(low) != 1 or len(high) != 1 or ord(high) < ord(low): raise ValueError("invalid range") content = re.escape(low) + "-" + re.escape(high) else: content = re.escape(value) pieces.append("[" + ("^" if negate else "") + content + "]") position = end elif character == "{": alternatives: list[str] = [] position += 1 while True: alternative, position = _glob_regex(pattern, position, ",}") alternatives.append(alternative) if position >= len(pattern): raise ValueError("unterminated alternatives") if pattern[position] == "}": break position += 1 if len(alternatives) < 2 or any(not item for item in alternatives): raise ValueError("invalid alternatives") pieces.append("(?:" + "|".join(alternatives) + ")") elif character in "]}": # A closing delimiter outside its grammar context is literal in # gobwas/glob's lexer. pieces.append(re.escape(character)) else: pieces.append(re.escape(character)) position += 1 return "".join(pieces), position def _is_plain(pattern: str) -> bool: return not any(character in SPECIAL for character in pattern) def _matches(pattern: str, branch: str) -> bool: if not pattern.isascii() or not 1 <= len(pattern) <= MAX_RULE_NAME: raise PolicyError("branch protection rule name is invalid") if _is_plain(pattern): return pattern.casefold() == branch.casefold() try: expression, position = _glob_regex(pattern) if position != len(pattern): raise ValueError("incomplete glob") except (re.error, ValueError): # Gitea quotes an invalid special pattern and matches it literally. # A literal containing a special byte cannot equal main or master. return False return re.fullmatch(expression, branch) is not None 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, bool, 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 1 <= priority <= 1_000_000 or not isinstance(rule_name, str) ): raise PolicyError("branch protection priority is ambiguous") created = _created(raw.get("created_at")) if _matches(rule_name, branch): # Gitea v1.23.8 sorts by priority, then puts plain names before # globs, then uses creation time. The API exposes every component. ordered.append((priority, not _is_plain(rule_name), created, index, raw)) if not ordered: return "ABSENT" effective = min(ordered)[4] 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())