audit-labs/audit-tools

A collection of scripts, queries, and other goodies you can use in an audit.

clone: git clone https://gitbay.org/audit-labs/audit-tools.git

04d10073de15548aa96bbf4bfb2b72de8ea1addf

verified · cmc

author: Christian Cleberg <hello@cleberg.net> · 2026-07-29T16:52:39Z

fix: detect ruleset-based branch protection

The branch_protections collector only queried the classic
/branches/{branch}/protection endpoint, so a branch protected solely by a
ruleset (org- or repo-level) was reported protected=False. Also query the
per-branch rules endpoint (/repos/{org}/{repo}/rules/branches/{branch}), which
aggregates the rules enforced from all applicable rulesets, and merge the two.

- New protection_source column: "branch protection", "ruleset", or
  "branch protection + ruleset".
- Review/status-check details are pulled from the ruleset's pull_request and
  required_status_checks rules when classic protection is absent.
- Add tests covering ruleset-only, classic-only, both, and neither.
 applications/github/README.md                      |   2 +-
 .../github/collectors/branch_protections.py        | 160 ++++++++++++++-------
 tui/tests/test_branch_protections.py               | 109 ++++++++++++++
 3 files changed, 218 insertions(+), 53 deletions(-)

diff --git a/applications/github/README.md b/applications/github/README.md
index 75fe202..e9685cf 100644
--- a/applications/github/README.md
+++ b/applications/github/README.md
@@ -44,7 +44,7 @@ Creates a directory: `<out>/github_audit_<org>_<YYYY-MM-DD>/`
 | `pending_invitations.csv` | Invitations not yet accepted, with age in days |
 | `team_permissions.csv` | Teams, their repos, permissions, and members |
 | `permission_matrix.csv` | Full user/repo/permission cross-reference |
-| `branch_protections.csv` | Branch protection settings across all repos |
+| `branch_protections.csv` | Per-branch protection across all repos, from classic branch protection **and** rulesets (`protection_source` records which) |
 | `commits.csv` | Commit history across all repos for the target branch |
 | `audit_log.csv` | Branch protection and repository ruleset audit-log changes from the last 180 days (Enterprise Cloud only) |
 | `summary.txt` | Row counts per section |
diff --git a/applications/github/collectors/branch_protections.py b/applications/github/collectors/branch_protections.py
index 3c87723..07a8550 100644
--- a/applications/github/collectors/branch_protections.py
+++ b/applications/github/collectors/branch_protections.py
@@ -1,5 +1,15 @@
 """
-Collect branch protection and ruleset data across all repos in an org.
+Collect branch protection across all repos in an org.
+
+Protection can come from two independent systems:
+
+- **Classic branch protection** (``/branches/{branch}/protection``)
+- **Rulesets** (org- or repo-level) — a branch protected only by a ruleset does
+  not appear in the classic endpoint at all.
+
+Both are checked per branch and merged, so ruleset-only protection is no longer
+reported as unprotected. ``protection_source`` records where the protection
+comes from.
 """
 
 import sys
@@ -8,76 +18,122 @@ import requests
 
 from .api import paginate
 
+_NO_PROTECTION = {
+    "required_reviews": None,
+    "dismiss_stale_reviews": None,
+    "require_code_owner_reviews": None,
+    "required_status_checks": None,
+    "enforce_admins": None,
+    "restrictions": None,
+}
+
 
 def branch_protections(org, cfg):
-    """
-    For each repo, return protection settings per branch and any rulesets.
-    Branches with no protection are included with protected=False.
-    Repos that return 403 on the branches endpoint are skipped with a warning.
+    """For each repo, return protection settings per branch (classic + ruleset).
+
+    Repos whose branches endpoint returns 403/404 are skipped with a warning.
     """
     repos = paginate(f"https://api.github.com/orgs/{org}/repos", cfg)
     rows = []
 
     for repo in repos:
         repo_name = repo["name"]
-
         try:
             branches = paginate(
                 f"https://api.github.com/repos/{org}/{repo_name}/branches", cfg
             )
         except requests.HTTPError as e:
-            if e.response is not None and e.response.status_code == 403:
+            if e.response is not None and e.response.status_code in (403, 404):
                 print(
-                    f"  Skipping {repo_name}: branches endpoint returned 403",
+                    f"  Skipping {repo_name}: branches endpoint returned "
+                    f"{e.response.status_code}",
                     file=sys.stderr,
                 )
                 continue
             raise
 
         for branch in branches:
-            branch_name = branch["name"]
-            url = (
-                f"https://api.github.com/repos/{org}/{repo_name}"
-                f"/branches/{branch_name}/protection"
-            )
-            resp = requests.get(url, headers=cfg["headers"], timeout=cfg["timeout"])
-
-            if resp.status_code in (403, 404):
-                rows.append(
-                    {
-                        "repo": repo_name,
-                        "branch": branch_name,
-                        "protected": False,
-                        "required_reviews": None,
-                        "dismiss_stale_reviews": None,
-                        "require_code_owner_reviews": None,
-                        "required_status_checks": None,
-                        "enforce_admins": None,
-                        "restrictions": None,
-                    }
-                )
-                continue
-
-            resp.raise_for_status()
-            p = resp.json()
-            reviews = p.get("required_pull_request_reviews", {})
-            checks = p.get("required_status_checks", {})
-
-            rows.append(
-                {
-                    "repo": repo_name,
-                    "branch": branch_name,
-                    "protected": True,
-                    "required_reviews": reviews.get("required_approving_review_count"),
-                    "dismiss_stale_reviews": reviews.get("dismiss_stale_reviews"),
-                    "require_code_owner_reviews": reviews.get(
-                        "require_code_owner_reviews"
-                    ),
-                    "required_status_checks": ", ".join(checks.get("contexts", []))
-                    or None,
-                    "enforce_admins": p.get("enforce_admins", {}).get("enabled"),
-                    "restrictions": bool(p.get("restrictions")),
-                }
-            )
+            rows.append(_branch_row(org, repo_name, branch["name"], cfg))
 
     return rows
+
+
+def _branch_row(org, repo, branch, cfg):
+    classic = _classic_protection(org, repo, branch, cfg)
+    ruleset = _ruleset_protection(org, repo, branch, cfg)
+
+    if classic and ruleset:
+        source = "branch protection + ruleset"
+    elif classic:
+        source = "branch protection"
+    elif ruleset:
+        source = "ruleset"
+    else:
+        source = ""
+
+    # Prefer classic values where present, otherwise fall back to ruleset.
+    details = classic or ruleset or _NO_PROTECTION
+    return {
+        "repo": repo,
+        "branch": branch,
+        "protected": bool(classic or ruleset),
+        "protection_source": source,
+        **details,
+    }
+
+
+def _classic_protection(org, repo, branch, cfg):
+    """Return classic branch-protection details, or None if not protected."""
+    url = f"https://api.github.com/repos/{org}/{repo}/branches/{branch}/protection"
+    resp = requests.get(url, headers=cfg["headers"], timeout=cfg["timeout"])
+    if resp.status_code in (403, 404):
+        return None
+    resp.raise_for_status()
+    p = resp.json()
+    reviews = p.get("required_pull_request_reviews", {})
+    checks = p.get("required_status_checks", {})
+    return {
+        "required_reviews": reviews.get("required_approving_review_count"),
+        "dismiss_stale_reviews": reviews.get("dismiss_stale_reviews"),
+        "require_code_owner_reviews": reviews.get("require_code_owner_reviews"),
+        "required_status_checks": ", ".join(checks.get("contexts", [])) or None,
+        "enforce_admins": p.get("enforce_admins", {}).get("enabled"),
+        "restrictions": bool(p.get("restrictions")),
+    }
+
+
+def _ruleset_protection(org, repo, branch, cfg):
+    """Return protection derived from the rulesets active on a branch, or None.
+
+    The per-branch rules endpoint aggregates the rules enforced on the branch
+    from every applicable org- and repo-level ruleset.
+    """
+    url = f"https://api.github.com/repos/{org}/{repo}/rules/branches/{branch}"
+    try:
+        rules = paginate(url, cfg)
+    except requests.HTTPError as e:
+        if e.response is not None and e.response.status_code in (403, 404):
+            return None
+        raise
+    if not rules:
+        return None
+
+    params = {}
+    for rule in rules:
+        params.setdefault(rule.get("type"), rule.get("parameters") or {})
+
+    pull_request = params.get("pull_request", {})
+    status_checks = params.get("required_status_checks", {})
+    contexts = [
+        c.get("context", "") for c in status_checks.get("required_status_checks", [])
+    ]
+    return {
+        "required_reviews": pull_request.get("required_approving_review_count"),
+        "dismiss_stale_reviews": pull_request.get("dismiss_stale_reviews_on_push"),
+        "require_code_owner_reviews": pull_request.get("require_code_owner_review"),
+        "required_status_checks": ", ".join(contexts) or None,
+        # Rulesets model admin enforcement and push restrictions via bypass
+        # actors, which the per-branch rules endpoint does not return.
+        "enforce_admins": None,
+        "restrictions": None,
+    }
diff --git a/tui/tests/test_branch_protections.py b/tui/tests/test_branch_protections.py
new file mode 100644
index 0000000..24bd3b0
--- /dev/null
+++ b/tui/tests/test_branch_protections.py
@@ -0,0 +1,109 @@
+"""Tests for branch-protection collection, covering classic + ruleset merge."""
+
+import types
+
+from applications.github.collectors import branch_protections as bp
+
+CFG = {"headers": {}, "timeout": 30}
+
+
+def _classic_get(status, payload=None):
+    """Fake requests.get for the classic protection endpoint."""
+
+    def _get(*args, **kwargs):
+        return types.SimpleNamespace(
+            status_code=status,
+            raise_for_status=lambda: None,
+            json=lambda: payload or {},
+        )
+
+    return _get
+
+
+def _paginate(rules):
+    """Fake api.paginate returning one repo, one 'main' branch, and `rules`."""
+
+    def _p(url, cfg, params=None):
+        if url.endswith("/orgs/acme/repos"):
+            return [{"name": "repo1"}]
+        if url.endswith("/repos/acme/repo1/branches"):
+            return [{"name": "main"}]
+        if "/rules/branches/main" in url:
+            return rules
+        return []
+
+    return _p
+
+
+def _run(monkeypatch, rules, classic_status, classic_payload=None):
+    monkeypatch.setattr(bp, "paginate", _paginate(rules))
+    monkeypatch.setattr(
+        bp.requests, "get", _classic_get(classic_status, classic_payload)
+    )
+    rows = bp.branch_protections("acme", CFG)
+    assert len(rows) == 1
+    return rows[0]
+
+
+def test_ruleset_only_is_reported_protected(monkeypatch):
+    rules = [
+        {
+            "type": "pull_request",
+            "parameters": {
+                "required_approving_review_count": 2,
+                "require_code_owner_review": True,
+                "dismiss_stale_reviews_on_push": True,
+            },
+        },
+        {"type": "non_fast_forward", "parameters": {}},
+    ]
+    row = _run(monkeypatch, rules, classic_status=404)
+    assert row["protected"] is True
+    assert row["protection_source"] == "ruleset"
+    assert row["required_reviews"] == 2
+    assert row["require_code_owner_reviews"] is True
+    assert row["dismiss_stale_reviews"] is True
+
+
+def test_ruleset_status_checks(monkeypatch):
+    rules = [
+        {
+            "type": "required_status_checks",
+            "parameters": {
+                "required_status_checks": [
+                    {"context": "build"},
+                    {"context": "lint"},
+                ]
+            },
+        }
+    ]
+    row = _run(monkeypatch, rules, classic_status=404)
+    assert row["required_status_checks"] == "build, lint"
+
+
+def test_classic_only(monkeypatch):
+    payload = {
+        "required_pull_request_reviews": {"required_approving_review_count": 1},
+        "enforce_admins": {"enabled": True},
+    }
+    row = _run(monkeypatch, rules=[], classic_status=200, classic_payload=payload)
+    assert row["protected"] is True
+    assert row["protection_source"] == "branch protection"
+    assert row["required_reviews"] == 1
+    assert row["enforce_admins"] is True
+
+
+def test_both_sources(monkeypatch):
+    payload = {"required_pull_request_reviews": {"required_approving_review_count": 3}}
+    rules = [{"type": "pull_request", "parameters": {}}]
+    row = _run(monkeypatch, rules, classic_status=200, classic_payload=payload)
+    assert row["protection_source"] == "branch protection + ruleset"
+    # Classic values win when both are present.
+    assert row["required_reviews"] == 3
+
+
+def test_no_protection(monkeypatch):
+    row = _run(monkeypatch, rules=[], classic_status=404)
+    assert row["protected"] is False
+    assert row["protection_source"] == ""
+    assert row["required_reviews"] is None