diff --git a/scripts/policy_selector.py b/scripts/policy_selector.py index 607b514..c02b967 100644 --- a/scripts/policy_selector.py +++ b/scripts/policy_selector.py @@ -469,10 +469,23 @@ def fetch_shared_config(api_root, gh_token): allowlist_repos = [] allowlist_data = {} + # False means "we could not read the policy", which is NOT the same as + # "the policy is empty". Callers use it to fail closed rather than + # silently enforcing a weaker document than the real allowlist requires. + allowlist_ok = False if raw_allowlist: try: - allowlist_data = yaml.safe_load(raw_allowlist) + parsed = yaml.safe_load(raw_allowlist) + if not isinstance(parsed, dict): + # safe_load returns None for an empty/comments-only file and a + # list for a top-level sequence. Assigning either to + # allowlist_data before the .get() below would destroy the {} + # default and poison every downstream consumer. + raise ValueError( + f"allowlist must be a mapping, got {type(parsed).__name__}" + ) + allowlist_data = parsed # Handle nesting under 'license_overrides' -> 'repos' repos_config = allowlist_data.get("license_overrides", {}).get("repos", {}) if not repos_config: @@ -483,10 +496,23 @@ def fetch_shared_config(api_root, gh_token): if r_config.get("require_cla") is False: allowlist_repos.append(r_name) - allowlist_repos.extend(allowlist_data.get("repositories", [])) + extra = allowlist_data.get("repositories") or [] + if isinstance(extra, list): + allowlist_repos.extend(extra) + else: + # A bare string here would be iterated character by character. + debug_log(f"⚠️ 'repositories' must be a list, got {type(extra).__name__}; ignoring.") + allowlist_ok = True debug_log(f"✅ Allowlist loaded via API. Found {len(allowlist_repos)} DCO-only repos.") except Exception as e: - debug_log(f"⚠️ Failed to parse allowlist YAML: {e}") + # Reset rather than leave a half-built or poisoned value behind. + allowlist_data = {} + allowlist_repos = [] + print(f"::error::Failed to parse cla/allowlist.yml: {e}. " + "Enforcing CLA for every repo until this is fixed.") + else: + print("::error::Could not fetch cla/allowlist.yml. " + "Enforcing CLA for every repo until this is fixed.") # B. Licenses licenses_data = fetch_json_with_fallback(api_root, "data/licenses_all.json", "cla/licenses_all.json", gh_token) or [] @@ -497,6 +523,7 @@ def fetch_shared_config(api_root, gh_token): return { "allowlist_data": allowlist_data, "allowlist_repos": allowlist_repos, + "allowlist_ok": allowlist_ok, "licenses_data": licenses_data, "permissive_data": permissive_data, } @@ -553,6 +580,18 @@ def process_single_pr(pr_number, pr_head_sha, pr_user, repo_full_name, gh_token, debug_log(f"⚠️ Logic Module Error: {e}. Defaulting to STRICT mode.") is_strict = True + # An unreadable allowlist is not an empty one. Without its overrides a + # non-permissive licence can look permissive — Broadcom Source Available + # is listed as permissive in the base tables and is held at CLA only by + # the override — so continuing would silently downgrade repos to DCO. + # Default to strict instead, matching the Logic Module Error path above. + # .get() default is True so a hand-built shared_config (tests, callers + # that supply their own data) is treated as deliberate, not as a failure. + if not config.get("allowlist_ok", True): + debug_log("⚠️ Allowlist unavailable. Defaulting to STRICT mode.") + is_strict = True + allowlist_repos = [] + # Allowlist Override if repo_full_name in allowlist_repos: debug_log(f"ℹ️ Repo {repo_full_name} is in Allowlist. Enforcing DCO only.") diff --git a/scripts/requires_cla.py b/scripts/requires_cla.py index 3211e58..a2b8830 100644 --- a/scripts/requires_cla.py +++ b/scripts/requires_cla.py @@ -26,6 +26,7 @@ # --- Overrides Helper --- from pathlib import Path +import fnmatch import re try: import yaml @@ -61,21 +62,45 @@ def _load_allowlist(in_memory_data: Optional[Dict] = None) -> dict: # Silent fail on disk read (expected in new architecture) return {} +def _matches_any(norm_license: str, patterns) -> bool: + """Exact match, or a glob when the pattern contains '*'. + + cla/allowlist.yml has documented "simple '*' wildcards" since it was + written, and ships `LicenseRef-Broadcom*` on that basis — but the match + was plain set membership, so that entry could only ever match a licence + literally named `LicenseRef-Broadcom*`. Any new Broadcom LicenseRef fell + through to the base tables, where the canonical-name matcher strips the + `LicenseRef-` prefix and finds `Broadcom_Proprietary` listed as + permissive — so a proprietary licence resolved to DCO. + + Patterns are matched case-sensitively against the already-normalised + name (both sides are lowercased by _norm_license_name first), so + fnmatchcase avoids fnmatch's platform-dependent case folding. + """ + for pattern in patterns: + if "*" in pattern or "?" in pattern or "[" in pattern: + if fnmatch.fnmatchcase(norm_license, pattern): + return True + elif norm_license == pattern: + return True + return False + + def _override_requires_cla(norm_license: str, allowlist: dict) -> None | bool: """ Return True (force CLA), False (force DCO), or None (no override). """ section = allowlist.get("license_overrides") or {} - req = {_norm_license_name(x) for x in (section.get("require_cla") or [])} - dco = {_norm_license_name(x) for x in (section.get("allow_dco") or [])} + req = [_norm_license_name(x) for x in (section.get("require_cla") or [])] + dco = [_norm_license_name(x) for x in (section.get("allow_dco") or [])] print(f"::warning::[DEBUG OVERRIDE] Checking Normalized License: '{norm_license}'") print(f"::warning:: -> 'require_cla' list contains: {sorted(list(req))}") - if norm_license in req: + if _matches_any(norm_license, req): print(f"::warning:: -> ✅ MATCH FOUND in require_cla! Forcing True.") return True - if norm_license in dco: + if _matches_any(norm_license, dco): print(f"::warning:: -> MATCH FOUND in allow_dco! Forcing False.") return False diff --git a/tests/test_policy_selector.py b/tests/test_policy_selector.py index 48ef37c..6a39af2 100644 --- a/tests/test_policy_selector.py +++ b/tests/test_policy_selector.py @@ -33,6 +33,7 @@ import io import tempfile import unittest +from pathlib import Path # --- Import setup. Must happen before `import policy_selector`. --- # @@ -505,6 +506,44 @@ def test_dco_policy_accepts_signed_off_commits(self): self.assertEqual([s["state"] for s in fake.statuses()], ["success"]) self.assertEqual(fake.statuses()[0]["description"], "DCO Signed") + def test_unreadable_allowlist_forces_cla_not_dco(self): + """The behaviour that matters. requires_CLA says permissive (DCO) and + the repo is in allowlist_repos (DCO), but the allowlist could not be + read — so neither signal is trustworthy and the gate must ask for the + stronger document rather than the weaker one.""" + policy_selector.requires_cla.requires_CLA = lambda *a, **k: False + fake = self.run_pr( + paginated_routes={"/issues/5/comments": [], "/pulls/5/commits": []}, + config=dict(self.SHARED_CONFIG, allowlist_ok=False, + allowlist_repos=["vmware/repo"])) + self.assertEqual(fake.statuses()[0]["description"], "CLA Missing") + + def test_fail_closed_guard_does_not_corrupt_a_shared_config(self): + """cla_sweeper fetches the config ONCE and threads the same dict + through every PR in the sweep. The guard clears allowlist_repos, so if + it mutated the list in place instead of rebinding the local name, the + first PR to hit it would strip the DCO downgrade from every repo + processed afterwards in that sweep. + """ + shared = dict(self.SHARED_CONFIG, allowlist_ok=False, + allowlist_repos=["vmware/repo", "vmware/other"]) + before = list(shared["allowlist_repos"]) + policy_selector.requires_cla.requires_CLA = lambda *a, **k: False + self.run_pr(paginated_routes={"/issues/5/comments": [], "/pulls/5/commits": []}, + config=shared) + self.assertEqual(shared["allowlist_repos"], before, + "the guard must rebind, not mutate the caller's list") + + def test_hand_built_config_without_the_flag_is_treated_as_readable(self): + """Back-compat: a caller supplying its own data deliberately (tests, + license_report.py) has no allowlist_ok key and must not be forced + strict by its absence.""" + policy_selector.requires_cla.requires_CLA = lambda *a, **k: False + fake = self.run_pr( + paginated_routes={"/issues/5/comments": [], "/pulls/5/commits": []}, + config=dict(self.SHARED_CONFIG)) # no allowlist_ok key at all + self.assertEqual(fake.statuses()[0]["description"], "DCO Missing") + def test_allowlisted_repo_is_downgraded_to_dco(self): config = dict(self.SHARED_CONFIG, allowlist_repos=["vmware/repo"]) fake = self.run_pr(paginated_routes={"/issues/5/comments": [], "/pulls/5/commits": []}, config=config) @@ -653,14 +692,15 @@ def test_write_failure_is_reported_to_the_caller(self): # fetch_shared_config (protects the PR #66 per-sweep caching contract) # --------------------------------------------------------------------------- class TestFetchSharedConfig(PolicySelectorTestCase): - def test_returns_all_four_keys_process_single_pr_indexes(self): + def test_returns_all_keys_process_single_pr_indexes(self): # process_single_pr indexes these directly, so a missing key is a # KeyError mid-sweep rather than a soft failure. self.install() config = policy_selector.fetch_shared_config("https://api.invalid", "tok") self.assertEqual( sorted(config.keys()), - ["allowlist_data", "allowlist_repos", "licenses_data", "permissive_data"], + ["allowlist_data", "allowlist_ok", "allowlist_repos", + "licenses_data", "permissive_data"], ) def test_missing_config_degrades_to_empty_rather_than_raising(self): @@ -668,6 +708,11 @@ def test_missing_config_degrades_to_empty_rather_than_raising(self): config = policy_selector.fetch_shared_config("https://api.invalid", "tok") self.assertEqual(config["licenses_data"], []) self.assertEqual(config["permissive_data"], []) + # allowlist_data was previously omitted here — it is the one key that + # could come back poisoned (None, or a list) rather than empty. + self.assertEqual(config["allowlist_data"], {}) + self.assertIs(config["allowlist_ok"], False, + "an unfetchable allowlist must be reported as not-ok, not as empty") self.assertEqual(config["allowlist_repos"], []) @@ -790,6 +835,17 @@ def test_dotgithub_stays_on_dco(self): "update this test in the same commit", ) + def test_broadcom_wildcard_entry_is_present(self): + """Pins the entry itself, not just the matcher. Deleting + `LicenseRef-Broadcom*` from this file silently returns every Broadcom + licence except the one spelled out to the base tables, where they are + listed as permissive — i.e. back to DCO.""" + req = (self.mapping().get("license_overrides") or {}).get("require_cla") or [] + self.assertTrue( + any("*" in str(x) and "roadcom" in str(x) for x in req), + f"expected a LicenseRef-Broadcom* wildcard in require_cla, got {req}", + ) + def test_broadcom_source_available_still_forces_cla(self): """Broadcom Source Available is listed as permissive in the base tables, so without this override it would fall through to DCO. Feed @@ -802,5 +858,200 @@ def test_broadcom_source_available_still_forces_cla(self): self.assertIs(decision, True) + +# --------------------------------------------------------------------------- +# Allowlist robustness: a policy we cannot read must not read as a policy +# that permits everything. +# --------------------------------------------------------------------------- +class TestAllowlistFailClosed(PolicySelectorTestCase): + """cla/allowlist.yml is fetched from `ref: main` on every run. + + Before this, `allowlist_data = yaml.safe_load(...)` was assigned before + the `.get()` that can raise, so a comments-only file left allowlist_data + as None, the AttributeError was swallowed into a debug_log, and every + licence override silently disappeared org-wide. That direction is + dangerous: Broadcom Source Available is listed as permissive in the base + tables and is held at CLA *only* by the override, so losing it downgrades + those repos to DCO. + """ + + def _config(self, body): + self.install(routes={"/contents/cla/allowlist.yml": self._file(body)}) + return policy_selector.fetch_shared_config("https://api.invalid", "tok") + + @staticmethod + def _file(text): + import base64 + return {"content": base64.b64encode(text.encode()).decode()} + + def test_valid_allowlist_is_ok_and_parsed(self): + cfg = self._config( + "license_overrides:\n" + " require_cla: [LicenseRef-Broadcom_Source_Available]\n" + " repos:\n" + " vmware/x:\n" + " require_cla: false\n" + ) + self.assertIs(cfg["allowlist_ok"], True) + self.assertEqual(cfg["allowlist_repos"], ["vmware/x"]) + + def test_comments_only_file_is_not_ok_and_data_stays_a_dict(self): + cfg = self._config("# nothing but a comment\n") + self.assertIs(cfg["allowlist_ok"], False) + self.assertEqual(cfg["allowlist_data"], {}, + "must not leak yaml.safe_load's None into callers") + + def test_top_level_list_is_not_ok(self): + cfg = self._config("- one\n- two\n") + self.assertIs(cfg["allowlist_ok"], False) + self.assertEqual(cfg["allowlist_data"], {}) + + def test_malformed_yaml_is_not_ok(self): + cfg = self._config("license_overrides: [unclosed\n") + self.assertIs(cfg["allowlist_ok"], False) + self.assertEqual(cfg["allowlist_data"], {}) + self.assertEqual(cfg["allowlist_repos"], []) + + def test_repositories_scalar_does_not_explode_into_characters(self): + cfg = self._config("repositories: vmware/foo\n") + self.assertEqual(cfg["allowlist_repos"], [], + "a bare string must be rejected, not iterated per character") + + +class TestOverrideMatchesTheRealCatalogue(unittest.TestCase): + """Guard the seam between two files that must agree but are spelled + differently. + + cla/allowlist.yml says `LicenseRef-Broadcom_Source_Available` + (underscores); data/licenses_all.json's canonical spdx_id is + `LicenseRef-Broadcom-Source-Available` (hyphens). They match ONLY because + _norm_license_name collapses [\\s_]+ to '-'. Change that normaliser, or + regenerate the catalogue with a different spelling, and production stops + forcing CLA on 48 repos while a test that normalises both sides itself + would stay green. So this drives the lookup from the catalogue's own + value, not from a literal. + """ + + @classmethod + def setUpClass(cls): + import yaml + cls.allowlist = yaml.safe_load( + requires_cla._ALLOWLIST_PATH.read_text(encoding="utf-8")) + cls.catalogue = json.loads( + (Path(__file__).resolve().parents[1] / "data" / "licenses_all.json").read_text()) + + def override_for(self, catalogue_key): + entry = self.catalogue[catalogue_key] + spdx = entry.get("spdx_id") or catalogue_key + import contextlib, io + with contextlib.redirect_stdout(io.StringIO()): + return requires_cla._override_requires_cla( + requires_cla._norm_license_name(spdx), self.allowlist) + + def test_allowlist_and_catalogue_spellings_converge(self): + """Assert the convergence directly, not through the matcher. + + Going via _override_requires_cla cannot detect a broken normaliser, + because the LicenseRef-Broadcom* wildcard matches the catalogue form + either way and masks it. The two files genuinely disagree on spelling + — underscores here, hyphens there — and only _norm_license_name makes + them meet. That property is what has to hold. + """ + entry = self.catalogue["Broadcom_Source_Available"] + from_catalogue = requires_cla._norm_license_name(entry["spdx_id"]) + from_allowlist = requires_cla._norm_license_name( + "LicenseRef-Broadcom_Source_Available") + self.assertEqual( + from_catalogue, from_allowlist, + f"catalogue spells it {entry['spdx_id']!r} and the allowlist spells it " + "'LicenseRef-Broadcom_Source_Available'; _norm_license_name is the only " + "thing making them match, so 48 repos depend on this rule", + ) + + def test_broadcom_source_available_resolves_to_cla_from_catalogue_id(self): + self.assertIs(self.override_for("Broadcom_Source_Available"), True) + + def test_broadcom_proprietary_resolves_to_cla_via_the_wildcard(self): + """Without the wildcard this falls through to the base tables, which + list Broadcom_Proprietary as permissive, and the repo gets DCO.""" + self.assertIs(self.override_for("Broadcom_Proprietary"), True) + + +class TestOverrideWildcards(unittest.TestCase): + """`cla/allowlist.yml` has always documented "simple '*' wildcards", and + ships `LicenseRef-Broadcom*` on that basis, but matching was plain set + membership so that entry matched nothing.""" + + ALLOWLIST = {"license_overrides": { + "require_cla": ["LicenseRef-Broadcom_Source_Available", "LicenseRef-Broadcom*"], + "allow_dco": ["LicenseRef-Sample*"], + }} + + def check(self, license_id): + import contextlib, io + with contextlib.redirect_stdout(io.StringIO()): + return requires_cla._override_requires_cla( + requires_cla._norm_license_name(license_id), self.ALLOWLIST) + + def test_exact_entry_still_matches(self): + self.assertIs(self.check("LicenseRef-Broadcom_Source_Available"), True) + + def test_exact_entries_match_when_no_wildcard_covers_them(self): + """Pins the non-glob branch of _matches_any independently. + + Every other fixture here pairs an exact entry with a wildcard over the + same namespace (`LicenseRef-Broadcom_Source_Available` alongside + `LicenseRef-Broadcom*`), so the wildcard masks a broken exact branch. + Verified: with the exact branch removed, an exact-only allowlist stops + matching and every one of those tests still passed. This fixture has + no wildcard, so the exact path has to work on its own. + """ + exact_only = {"license_overrides": {"require_cla": ["MIT", "GPL-2.0"], + "allow_dco": ["Apache-2.0"]}} + import contextlib, io + with contextlib.redirect_stdout(io.StringIO()): + self.assertIs(requires_cla._override_requires_cla("mit", exact_only), True) + self.assertIs(requires_cla._override_requires_cla("gpl-2.0", exact_only), True) + self.assertIs(requires_cla._override_requires_cla("apache-2.0", exact_only), False) + self.assertIsNone(requires_cla._override_requires_cla("bsd-3-clause", exact_only)) + + def test_wildcard_catches_other_broadcom_licences(self): + """The gap this closes: without it, LicenseRef-Broadcom-Proprietary + falls through to the base tables, where the canonical matcher strips + the LicenseRef- prefix and finds Broadcom_Proprietary listed as + permissive — so a proprietary licence resolved to DCO.""" + for lic in ("LicenseRef-Broadcom-Proprietary", + "LicenseRef-Broadcom_Enterprise", + "LicenseRef-BroadcomAnything"): + self.assertIs(self.check(lic), True, lic) + + def test_wildcard_does_not_over_match(self): + for lic in ("MIT", "Apache-2.0", "LicenseRef-Other", "Broadcom-Without-Prefix"): + self.assertIsNone(self.check(lic), lic) + + def test_allow_dco_wildcards_work_too(self): + self.assertIs(self.check("LicenseRef-Sample-Thing"), False) + + def test_matching_is_case_sensitive_on_already_normalised_input(self): + """Both sides are lowercased by _norm_license_name before matching, so + fnmatchcase is correct and fnmatch's platform-dependent case folding + is not wanted. An upper-case pattern must therefore NOT match.""" + upper = {"license_overrides": {"require_cla": ["LICENSEREF-ZZZ*"]}} + import contextlib, io + with contextlib.redirect_stdout(io.StringIO()): + # the pattern is normalised (lowercased) too, so this still matches + self.assertIs(requires_cla._override_requires_cla("licenseref-zzz-a", upper), True) + # but a raw, un-normalised upper-case subject must not + self.assertIsNone(requires_cla._override_requires_cla("LICENSEREF-ZZZ-A", upper)) + + def test_require_cla_wins_over_allow_dco(self): + both = {"license_overrides": {"require_cla": ["LicenseRef-X*"], + "allow_dco": ["LicenseRef-X*"]}} + import contextlib, io + with contextlib.redirect_stdout(io.StringIO()): + d = requires_cla._override_requires_cla("licenseref-xyz", both) + self.assertIs(d, True, "require_cla must take precedence") + + if __name__ == "__main__": unittest.main()