From 71f399369cf9a5deea8c0deb2374842bdb2cfc92 Mon Sep 17 00:00:00 2001 From: rookiestar28 Date: Mon, 3 Aug 2026 18:14:37 +0800 Subject: [PATCH] fix(ci): classify bootstrap route changes as high risk --- scripts/run_adversarial_gate.py | 15 +- ...st_adversarial_bootstrap_classification.py | 220 ++++++++++++++++++ 2 files changed, 231 insertions(+), 4 deletions(-) create mode 100644 tests/test_adversarial_bootstrap_classification.py diff --git a/scripts/run_adversarial_gate.py b/scripts/run_adversarial_gate.py index 76d8b65..474a6b7 100644 --- a/scripts/run_adversarial_gate.py +++ b/scripts/run_adversarial_gate.py @@ -37,6 +37,9 @@ DEFAULT_HIGH_RISK_PATTERNS = [ "services/access_control.py", "services/tenant_context.py", "api/routes.py", + # CRITICAL: keep route-bootstrap owners exact; broad service globs over-escalate CI. + "services/bootstrap/registration.py", + "services/route_bootstrap_contract.py", "services/security_*.py", "services/startup_profile_gate.py", "services/control_plane.py", @@ -52,7 +55,10 @@ EXTENDED_MUTATION_THRESHOLD = 80.0 def _normalize_rel_path(path: str) -> str: - return pathlib.PurePosixPath(path.replace("\\", "/")).as_posix().lstrip("./") + normalized = path.replace("\\", "/") + while normalized.startswith("./"): + normalized = normalized[2:] + return pathlib.PurePosixPath(normalized).as_posix() def _run_git_diff(base: Optional[str], head: Optional[str]) -> List[str]: @@ -116,10 +122,11 @@ def _collect_changed_files( def _filter_high_risk_files(changed_files: List[str], patterns: List[str]) -> List[str]: matched: Set[str] = set() normalized_patterns = [_normalize_rel_path(p) for p in patterns if p.strip()] - for f in changed_files: + for file_path in changed_files: + normalized_file = _normalize_rel_path(file_path) for pattern in normalized_patterns: - if fnmatch.fnmatch(f, pattern): - matched.add(f) + if fnmatch.fnmatch(normalized_file, pattern): + matched.add(normalized_file) break return sorted(matched) diff --git a/tests/test_adversarial_bootstrap_classification.py b/tests/test_adversarial_bootstrap_classification.py new file mode 100644 index 0000000..cb32f1f --- /dev/null +++ b/tests/test_adversarial_bootstrap_classification.py @@ -0,0 +1,220 @@ +import importlib.util +import subprocess +import unittest +from pathlib import Path +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] +GATE_PATH = ROOT / "scripts" / "run_adversarial_gate.py" + +LEGACY_HIGH_RISK_PATTERNS = { + "services/access_control.py", + "services/tenant_context.py", + "api/routes.py", + "services/security_*.py", + "services/startup_profile_gate.py", + "services/control_plane.py", + "services/endpoint_manifest.py", + "services/webhook_auth.py", + "services/safe_io.py", +} +BOOTSTRAP_BOUNDARIES = [ + "services/bootstrap/registration.py", + "services/route_bootstrap_contract.py", +] + + +def _load_gate_module(): + spec = importlib.util.spec_from_file_location( + "adversarial_bootstrap_classification_gate", GATE_PATH + ) + if spec is None or spec.loader is None: + raise RuntimeError("Unable to load adversarial gate module") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class AdversarialBootstrapClassificationTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.gate = _load_gate_module() + + def test_bootstrap_route_boundary_diff_selects_extended(self): + changed = [ + *BOOTSTRAP_BOUNDARIES, + "tests/test_route_bootstrap_contract.py", + ] + with patch.object( + self.gate, + "_collect_changed_files", + return_value=(changed, "fixture: bootstrap boundary"), + ): + result = self.gate._resolve_effective_profile( + "auto", None, None, self.gate.DEFAULT_HIGH_RISK_PATTERNS + ) + + self.assertEqual(result[0], "extended") + self.assertEqual(result[1], changed) + self.assertEqual(result[2], BOOTSTRAP_BOUNDARIES) + self.assertEqual(result[3], "fixture: bootstrap boundary") + + def test_default_inventory_preserves_legacy_and_adds_only_exact_boundaries(self): + patterns = set(self.gate.DEFAULT_HIGH_RISK_PATTERNS) + self.assertTrue(LEGACY_HIGH_RISK_PATTERNS.issubset(patterns)) + self.assertEqual( + patterns - LEGACY_HIGH_RISK_PATTERNS, + set(BOOTSTRAP_BOUNDARIES), + ) + self.assertNotIn("services/**", patterns) + self.assertNotIn("services/bootstrap/**", patterns) + self.assertNotIn("services/bootstrap/*.py", patterns) + + def test_legacy_high_risk_paths_still_match(self): + candidates = [ + "services/access_control.py", + "services/tenant_context.py", + "api/routes.py", + "services/security_boundary.py", + "services/startup_profile_gate.py", + "services/control_plane.py", + "services/endpoint_manifest.py", + "services/webhook_auth.py", + "services/safe_io.py", + ] + self.assertEqual( + self.gate._filter_high_risk_files( + candidates, self.gate.DEFAULT_HIGH_RISK_PATTERNS + ), + sorted(candidates), + ) + + def test_unrelated_and_neighboring_paths_remain_non_hotspots(self): + candidates = [ + "services/bootstrap/posture.py", + "services/bootstrap/registration_helper.py", + "services/route_bootstrap_contract.py.bak", + "services/other.py", + "tests/test_route_bootstrap_contract.py", + ] + self.assertEqual( + self.gate._filter_high_risk_files( + candidates, self.gate.DEFAULT_HIGH_RISK_PATTERNS + ), + [], + ) + with patch.object( + self.gate, + "_collect_changed_files", + return_value=(candidates, "fixture: non-hotspot"), + ): + result = self.gate._resolve_effective_profile( + "auto", None, None, self.gate.DEFAULT_HIGH_RISK_PATTERNS + ) + self.assertEqual(result, ("smoke", candidates, [], "fixture: non-hotspot")) + + def test_explicit_profiles_take_precedence_without_diff_discovery(self): + for requested in ("smoke", "extended"): + with self.subTest(requested=requested), patch.object( + self.gate, + "_collect_changed_files", + side_effect=AssertionError("explicit profile inspected diff"), + ): + self.assertEqual( + self.gate._resolve_effective_profile( + requested, + "malformed;base", + "malformed|head", + self.gate.DEFAULT_HIGH_RISK_PATTERNS, + ), + (requested, [], [], "explicit profile"), + ) + + def test_candidate_normalization_is_cross_platform_and_deterministic(self): + candidates = [ + r".\services\route_bootstrap_contract.py", + r"services\bootstrap\registration.py", + "services/bootstrap/registration.py", + ] + self.assertEqual( + self.gate._filter_high_risk_files(candidates, BOOTSTRAP_BOUNDARIES), + BOOTSTRAP_BOUNDARIES, + ) + + def test_malformed_paths_cannot_alias_bootstrap_boundaries(self): + candidates = [ + "../services/bootstrap/registration.py", + "/services/bootstrap/registration.py", + r"C:\repo\services\bootstrap\registration.py", + ".../services/bootstrap/registration.py", + ] + self.assertEqual( + self.gate._filter_high_risk_files(candidates, BOOTSTRAP_BOUNDARIES), + [], + ) + self.assertEqual( + self.gate._normalize_rel_path("../services/bootstrap/registration.py"), + "../services/bootstrap/registration.py", + ) + self.assertEqual( + self.gate._normalize_rel_path("/services/bootstrap/registration.py"), + "/services/bootstrap/registration.py", + ) + + def test_custom_appended_pattern_remains_supported(self): + patterns = [*self.gate.DEFAULT_HIGH_RISK_PATTERNS, "custom/policy.py"] + self.assertEqual( + self.gate._filter_high_risk_files( + ["custom/policy.py", "custom/nearby.py"], patterns + ), + ["custom/policy.py"], + ) + + def test_hostile_diff_refs_remain_non_shell_arguments(self): + hostile_base = "main;echo injected" + hostile_head = "HEAD|type secrets" + calls = [] + + def fail_git(command, **kwargs): + calls.append((command, kwargs)) + return subprocess.CompletedProcess(command, 1, stdout="", stderr="bad ref") + + with patch.object(self.gate.shutil, "which", return_value="git"), patch.object( + self.gate.subprocess, "run", side_effect=fail_git + ): + self.assertEqual(self.gate._run_git_diff(hostile_base, hostile_head), []) + + self.assertEqual(len(calls), 2) + self.assertEqual( + calls[0][0], + ["git", "diff", "--name-only", f"{hostile_base}...{hostile_head}"], + ) + self.assertEqual( + calls[1][0], + ["git", "diff", "--name-only", hostile_base, hostile_head], + ) + for _command, kwargs in calls: + self.assertNotIn("shell", kwargs) + self.assertEqual(kwargs, {"capture_output": True, "text": True}) + + def test_missing_diff_context_deterministically_selects_smoke(self): + with patch.object( + self.gate, + "_collect_changed_files", + return_value=([], "no git diff context"), + ): + self.assertEqual( + self.gate._resolve_effective_profile( + "auto", None, None, self.gate.DEFAULT_HIGH_RISK_PATTERNS + ), + ("smoke", [], [], "no git diff context"), + ) + + def test_mutation_thresholds_remain_governed(self): + self.assertEqual(self.gate.SMOKE_MUTATION_THRESHOLD, 20.0) + self.assertEqual(self.gate.EXTENDED_MUTATION_THRESHOLD, 80.0) + + +if __name__ == "__main__": + unittest.main()