139 lines
5.2 KiB
Python
139 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate the first effective Gitea protection for an Atlas primary branch."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import fnmatch
|
|
import json
|
|
import os
|
|
import stat
|
|
from pathlib import Path
|
|
|
|
MAX_INPUT = 1024 * 1024
|
|
MAX_RULES = 100
|
|
MAX_RULE_NAME = 255
|
|
|
|
|
|
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 _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 pattern.count("[") != pattern.count("]") or "\\" in pattern:
|
|
raise PolicyError("branch protection glob is ambiguous")
|
|
# Gitea matches ordinary names case-insensitively and compiles wildcard
|
|
# names as globs with '/' as a separator. Primary branches have no slash,
|
|
# so fnmatchcase is equivalent for the only two targets managed here.
|
|
patterns = [pattern]
|
|
while any("{" in item or "}" in item for item in patterns):
|
|
expanded: list[str] = []
|
|
for item in patterns:
|
|
if "{" not in item and "}" not in item:
|
|
expanded.append(item)
|
|
continue
|
|
start = item.find("{")
|
|
end = item.find("}", start + 1)
|
|
if start < 0 or end < 0 or "{" in item[start + 1 : end]:
|
|
raise PolicyError("branch protection glob is ambiguous")
|
|
choices = item[start + 1 : end].split(",")
|
|
if len(choices) < 2 or any(not choice for choice in choices):
|
|
raise PolicyError("branch protection glob is ambiguous")
|
|
expanded.extend(item[:start] + choice + item[end + 1 :] for choice in choices)
|
|
patterns = expanded
|
|
if len(patterns) > 32:
|
|
raise PolicyError("branch protection glob expands beyond the safe limit")
|
|
return any(
|
|
fnmatch.fnmatchcase(branch.casefold(), item.casefold()) for item in patterns
|
|
)
|
|
|
|
|
|
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, int, dict[str, object]]] = []
|
|
priorities: set[int] = set()
|
|
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 priority in priorities
|
|
or not isinstance(rule_name, str)
|
|
):
|
|
raise PolicyError("branch protection priority is ambiguous")
|
|
priorities.add(priority)
|
|
if _matches(rule_name, branch):
|
|
ordered.append((priority, index, raw))
|
|
if not ordered:
|
|
return "ABSENT"
|
|
effective = min(ordered)[2]
|
|
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())
|