Commit `62d8887fcc`

62d8887fcc54d3c7fe852c9a03528e254af5f939

parent: 69a3ab8a91

Unsigned

cmc <hello@cleberg.net> · 2026-07-29 16:52 UTC

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.

Layout: unified · split

applications/github/README.md +1 −1

	@@ -44,7 +44,7 @@ Creates a directory: `<out>/github_audit_<org>_<YYYY-MM-DD>/`
44	\| `pending_invitations.csv` \| Invitations not yet accepted, with age in days \|	44	\| `pending_invitations.csv` \| Invitations not yet accepted, with age in days \|
45	\| `team_permissions.csv` \| Teams, their repos, permissions, and members \|	45	\| `team_permissions.csv` \| Teams, their repos, permissions, and members \|
46	\| `permission_matrix.csv` \| Full user/repo/permission cross-reference \|	46	\| `permission_matrix.csv` \| Full user/repo/permission cross-reference \|
47	\| `branch_protections.csv` \| Branch protection settings across all repos \|	47	\| `branch_protections.csv` \| Per-branch protection across all repos, from classic branch protection and rulesets (`protection_source` records which) \|
48	\| `commits.csv` \| Commit history across all repos for the target branch \|	48	\| `commits.csv` \| Commit history across all repos for the target branch \|
49	\| `audit_log.csv` \| Branch protection and repository ruleset audit-log changes from the last 180 days (Enterprise Cloud only) \|	49	\| `audit_log.csv` \| Branch protection and repository ruleset audit-log changes from the last 180 days (Enterprise Cloud only) \|
50	\| `summary.txt` \| Row counts per section \|	50	\| `summary.txt` \| Row counts per section \|

applications/github/collectors/branch_protections.py +108 −52

	@@ -1,5 +1,15 @@
1	"""	1	"""
2	Collect branch protection and ruleset data across all repos in an org.	2	Collect branch protection across all repos in an org.
		3
		4	Protection 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
		10	Both are checked per branch and merged, so ruleset-only protection is no longer
		11	reported as unprotected. ``protection_source`` records where the protection
		12	comes from.
3	"""	13	"""
4		14
5	import sys	15	import sys
	@@ -8,76 +18,122 @@ import requests
8		18
9	from .api import paginate	19	from .api import paginate
10		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
11		30
12	def branch_protections(org, cfg):	31	def branch_protections(org, cfg):
13	"""	32	"""For each repo, return protection settings per branch (classic + ruleset).
14	For each repo, return protection settings per branch and any rulesets.	33
15	Branches with no protection are included with protected=False.	34	Repos whose branches endpoint returns 403/404 are skipped with a warning.
16	Repos that return 403 on the branches endpoint are skipped with a warning.
17	"""	35	"""
18	repos = paginate(f"https://api.github.com/orgs/{org}/repos", cfg)	36	repos = paginate(f"https://api.github.com/orgs/{org}/repos", cfg)
19	rows = []	37	rows = []
20		38
21	for repo in repos:	39	for repo in repos:
22	repo_name = repo["name"]	40	repo_name = repo["name"]
23
24	try:	41	try:
25	branches = paginate(	42	branches = paginate(
26	f"https://api.github.com/repos/{org}/{repo_name}/branches", cfg	43	f"https://api.github.com/repos/{org}/{repo_name}/branches", cfg
27	)	44	)
28	except requests.HTTPError as e:	45	except requests.HTTPError as e:
29	if e.response is not None and e.response.status_code == 403:	46	if e.response is not None and e.response.status_code in (403, 404):
30	print(	47	print(
31	f" Skipping {repo_name}: branches endpoint returned 403",	48	f" Skipping {repo_name}: branches endpoint returned "
		49	f"{e.response.status_code}",
32	file=sys.stderr,	50	file=sys.stderr,
33	)	51	)
34	continue	52	continue
35	raise	53	raise
36		54
37	for branch in branches:	55	for branch in branches:
38	branch_name = branch["name"]	56	rows.append(_branch_row(org, repo_name, branch["name"], cfg))
39	url = (
40	f"https://api.github.com/repos/{org}/{repo_name}"
41	f"/branches/{branch_name}/protection"
42	)
43	resp = requests.get(url, headers=cfg["headers"], timeout=cfg["timeout"])
44
45	if resp.status_code in (403, 404):
46	rows.append(
47	{
48	"repo": repo_name,
49	"branch": branch_name,
50	"protected": False,
51	"required_reviews": None,
52	"dismiss_stale_reviews": None,
53	"require_code_owner_reviews": None,
54	"required_status_checks": None,
55	"enforce_admins": None,
56	"restrictions": None,
57	}
58	)
59	continue
60
61	resp.raise_for_status()
62	p = resp.json()
63	reviews = p.get("required_pull_request_reviews", {})
64	checks = p.get("required_status_checks", {})
65
66	rows.append(
67	{
68	"repo": repo_name,
69	"branch": branch_name,
70	"protected": True,
71	"required_reviews": reviews.get("required_approving_review_count"),
72	"dismiss_stale_reviews": reviews.get("dismiss_stale_reviews"),
73	"require_code_owner_reviews": reviews.get(
74	"require_code_owner_reviews"
75	),
76	"required_status_checks": ", ".join(checks.get("contexts", []))
77	or None,
78	"enforce_admins": p.get("enforce_admins", {}).get("enabled"),
79	"restrictions": bool(p.get("restrictions")),
80	}
81	)
82		57
83	return rows	58	return rows
		59
		60
		61	def _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
		85	def _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
		105	def _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	}

tui/tests/test_branch_protections.py added +109

@@ -0,0 +1,109 @@
	1	"""Tests for branch-protection collection, covering classic + ruleset merge."""
	2
	3	import types
	4
	5	from applications.github.collectors import branch_protections as bp
	6
	7	CFG = {"headers": {}, "timeout": 30}
	8
	9
	10	def _classic_get(status, payload=None):
	11	"""Fake requests.get for the classic protection endpoint."""
	12
	13	def _get(args, *kwargs):
	14	return types.SimpleNamespace(
	15	status_code=status,
	16	raise_for_status=lambda: None,
	17	json=lambda: payload or {},
	18	)
	19
	20	return _get
	21
	22
	23	def _paginate(rules):
	24	"""Fake api.paginate returning one repo, one 'main' branch, and `rules`."""
	25
	26	def _p(url, cfg, params=None):
	27	if url.endswith("/orgs/acme/repos"):
	28	return [{"name": "repo1"}]
	29	if url.endswith("/repos/acme/repo1/branches"):
	30	return [{"name": "main"}]
	31	if "/rules/branches/main" in url:
	32	return rules
	33	return []
	34
	35	return _p
	36
	37
	38	def _run(monkeypatch, rules, classic_status, classic_payload=None):
	39	monkeypatch.setattr(bp, "paginate", _paginate(rules))
	40	monkeypatch.setattr(
	41	bp.requests, "get", _classic_get(classic_status, classic_payload)
	42	)
	43	rows = bp.branch_protections("acme", CFG)
	44	assert len(rows) == 1
	45	return rows[0]
	46
	47
	48	def test_ruleset_only_is_reported_protected(monkeypatch):
	49	rules = [
	50	{
	51	"type": "pull_request",
	52	"parameters": {
	53	"required_approving_review_count": 2,
	54	"require_code_owner_review": True,
	55	"dismiss_stale_reviews_on_push": True,
	56	},
	57	},
	58	{"type": "non_fast_forward", "parameters": {}},
	59	]
	60	row = _run(monkeypatch, rules, classic_status=404)
	61	assert row["protected"] is True
	62	assert row["protection_source"] == "ruleset"
	63	assert row["required_reviews"] == 2
	64	assert row["require_code_owner_reviews"] is True
	65	assert row["dismiss_stale_reviews"] is True
	66
	67
	68	def test_ruleset_status_checks(monkeypatch):
	69	rules = [
	70	{
	71	"type": "required_status_checks",
	72	"parameters": {
	73	"required_status_checks": [
	74	{"context": "build"},
	75	{"context": "lint"},
	76	]
	77	},
	78	}
	79	]
	80	row = _run(monkeypatch, rules, classic_status=404)
	81	assert row["required_status_checks"] == "build, lint"
	82
	83
	84	def test_classic_only(monkeypatch):
	85	payload = {
	86	"required_pull_request_reviews": {"required_approving_review_count": 1},
	87	"enforce_admins": {"enabled": True},
	88	}
	89	row = _run(monkeypatch, rules=[], classic_status=200, classic_payload=payload)
	90	assert row["protected"] is True
	91	assert row["protection_source"] == "branch protection"
	92	assert row["required_reviews"] == 1
	93	assert row["enforce_admins"] is True
	94
	95
	96	def test_both_sources(monkeypatch):
	97	payload = {"required_pull_request_reviews": {"required_approving_review_count": 3}}
	98	rules = [{"type": "pull_request", "parameters": {}}]
	99	row = _run(monkeypatch, rules, classic_status=200, classic_payload=payload)
	100	assert row["protection_source"] == "branch protection + ruleset"
	101	# Classic values win when both are present.
	102	assert row["required_reviews"] == 3
	103
	104
	105	def test_no_protection(monkeypatch):
	106	row = _run(monkeypatch, rules=[], classic_status=404)
	107	assert row["protected"] is False
	108	assert row["protection_source"] == ""
	109	assert row["required_reviews"] is None

Commit 62d8887fcc

Commit `62d8887fcc`