audit-labs/audit-tools

  1"""
  2Collect branch protection across all repos in an org.
  3
  4Protection can come from two independent systems:
  5
  6- **Classic branch protection** (``/branches/{branch}/protection``)
  7- **Rulesets** (org- or repo-level) — a branch protected only by a ruleset does
  8  not appear in the classic endpoint at all.
  9
 10Both are checked per branch and merged, so ruleset-only protection is no longer
 11reported as unprotected. ``protection_source`` records where the protection
 12comes from.
 13"""
 14
 15import sys
 16
 17import requests
 18
 19from .api import paginate
 20
 21_NO_PROTECTION = {
 22    "required_reviews": None,
 23    "dismiss_stale_reviews": None,
 24    "require_code_owner_reviews": None,
 25    "required_status_checks": None,
 26    "enforce_admins": None,
 27    "restrictions": None,
 28}
 29
 30
 31def branch_protections(org, cfg):
 32    """For each repo, return protection settings per branch (classic + ruleset).
 33
 34    Repos whose branches endpoint returns 403/404 are skipped with a warning.
 35    """
 36    repos = paginate(f"https://api.github.com/orgs/{org}/repos", cfg)
 37    rows = []
 38
 39    for repo in repos:
 40        repo_name = repo["name"]
 41        try:
 42            branches = paginate(
 43                f"https://api.github.com/repos/{org}/{repo_name}/branches", cfg
 44            )
 45        except requests.HTTPError as e:
 46            if e.response is not None and e.response.status_code in (403, 404):
 47                print(
 48                    f"  Skipping {repo_name}: branches endpoint returned "
 49                    f"{e.response.status_code}",
 50                    file=sys.stderr,
 51                )
 52                continue
 53            raise
 54
 55        for branch in branches:
 56            rows.append(_branch_row(org, repo_name, branch["name"], cfg))
 57
 58    return rows
 59
 60
 61def _branch_row(org, repo, branch, cfg):
 62    classic = _classic_protection(org, repo, branch, cfg)
 63    ruleset = _ruleset_protection(org, repo, branch, cfg)
 64
 65    if classic and ruleset:
 66        source = "branch protection + ruleset"
 67    elif classic:
 68        source = "branch protection"
 69    elif ruleset:
 70        source = "ruleset"
 71    else:
 72        source = ""
 73
 74    # Prefer classic values where present, otherwise fall back to ruleset.
 75    details = classic or ruleset or _NO_PROTECTION
 76    return {
 77        "repo": repo,
 78        "branch": branch,
 79        "protected": bool(classic or ruleset),
 80        "protection_source": source,
 81        **details,
 82    }
 83
 84
 85def _classic_protection(org, repo, branch, cfg):
 86    """Return classic branch-protection details, or None if not protected."""
 87    url = f"https://api.github.com/repos/{org}/{repo}/branches/{branch}/protection"
 88    resp = requests.get(url, headers=cfg["headers"], timeout=cfg["timeout"])
 89    if resp.status_code in (403, 404):
 90        return None
 91    resp.raise_for_status()
 92    p = resp.json()
 93    reviews = p.get("required_pull_request_reviews", {})
 94    checks = p.get("required_status_checks", {})
 95    return {
 96        "required_reviews": reviews.get("required_approving_review_count"),
 97        "dismiss_stale_reviews": reviews.get("dismiss_stale_reviews"),
 98        "require_code_owner_reviews": reviews.get("require_code_owner_reviews"),
 99        "required_status_checks": ", ".join(checks.get("contexts", [])) or None,
100        "enforce_admins": p.get("enforce_admins", {}).get("enabled"),
101        "restrictions": bool(p.get("restrictions")),
102    }
103
104
105def _ruleset_protection(org, repo, branch, cfg):
106    """Return protection derived from the rulesets active on a branch, or None.
107
108    The per-branch rules endpoint aggregates the rules enforced on the branch
109    from every applicable org- and repo-level ruleset.
110    """
111    url = f"https://api.github.com/repos/{org}/{repo}/rules/branches/{branch}"
112    try:
113        rules = paginate(url, cfg)
114    except requests.HTTPError as e:
115        if e.response is not None and e.response.status_code in (403, 404):
116            return None
117        raise
118    if not rules:
119        return None
120
121    params = {}
122    for rule in rules:
123        params.setdefault(rule.get("type"), rule.get("parameters") or {})
124
125    pull_request = params.get("pull_request", {})
126    status_checks = params.get("required_status_checks", {})
127    contexts = [
128        c.get("context", "") for c in status_checks.get("required_status_checks", [])
129    ]
130    return {
131        "required_reviews": pull_request.get("required_approving_review_count"),
132        "dismiss_stale_reviews": pull_request.get("dismiss_stale_reviews_on_push"),
133        "require_code_owner_reviews": pull_request.get("require_code_owner_review"),
134        "required_status_checks": ", ".join(contexts) or None,
135        # Rulesets model admin enforcement and push restrictions via bypass
136        # actors, which the per-branch rules endpoint does not return.
137        "enforce_admins": None,
138        "restrictions": None,
139    }