fix(control-plane): honor enforced org security config (#39)

Author: xiaojiou176

apps/orchestrator/tests/test_github_control_plane.py

@@ -35,6 +35,12 @@ def _policy_payload() -> dict:
             "secret_scanning_push_protection": {"required": True, "mode": "api"},
             "secret_scanning_non_provider_patterns": {"required": True, "mode": "api"},
             "secret_scanning_validity_checks": {"required": True, "mode": "api"},
+            "org_code_security_configuration": {
+                "required": True,
+                "mode": "api",
+                "configuration_id": 240284,
+                "required_repository_status": "enforced",
+            },
             "dependabot_config": {"required": True, "mode": "repo_file", "path": ".github/dependabot.yml"},
             "codeql_workflow": {"required": True, "mode": "repo_file", "path": ".github/workflows/codeql.yml"},
             "codeql_config": {"required": True, "mode": "repo_file", "path": ".github/codeql/codeql-config.yml"},
@@ -48,6 +54,7 @@ def _impl(path: str) -> tuple[int, dict]:
             return (
                 0,
                 {
+                    "id": 123,
                     "default_branch": "main",
                     "security_and_analysis": {
                         "secret_scanning": {"status": "enabled"},
@@ -71,6 +78,61 @@ def _impl(path: str) -> tuple[int, dict]:
             return 0, {}
         if path == "repos/example/repo/dependabot/alerts?per_page=1":
             return 0, []
+        if path == "orgs/example/code-security/configurations/240284":
+            return 0, {
+                "id": 240284,
+                "secret_scanning": "enabled",
+                "secret_scanning_push_protection": "enabled",
+                "secret_scanning_non_provider_patterns": "enabled",
+                "secret_scanning_validity_checks": "enabled",
+            }
+        if path == "orgs/example/code-security/configurations/240284/repositories":
+            return 0, [{"status": "enforced", "repository": {"id": 123}}]
+        raise AssertionError(path)
+
+    return _impl
+
+
+def _fake_gh_json_without_org_enforcement(secret_non_provider: str, validity_checks: str):
+    def _impl(path: str) -> tuple[int, dict]:
+        if path == "repos/example/repo":
+            return (
+                0,
+                {
+                    "id": 123,
+                    "default_branch": "main",
+                    "security_and_analysis": {
+                        "secret_scanning": {"status": "enabled"},
+                        "secret_scanning_push_protection": {"status": "enabled"},
+                        "secret_scanning_non_provider_patterns": {"status": secret_non_provider},
+                        "secret_scanning_validity_checks": {"status": validity_checks},
+                    },
+                },
+            )
+        if path == "repos/example/repo/actions/permissions":
+            return 0, {"enabled": True, "allowed_actions": "all", "sha_pinning_required": True}
+        if path == "repos/example/repo/branches/main/protection":
+            return 0, {"required_status_checks": {"contexts": ["Quick Feedback"]}}
+        if path == "repos/example/repo/environments":
+            return 0, {"environments": [{"name": "owner-approved-sensitive"}]}
+        if path == "repos/example/repo/private-vulnerability-reporting":
+            return 0, {"enabled": True}
+        if path == "repos/example/repo/vulnerability-alerts":
+            return 0, {}
+        if path == "repos/example/repo/code-scanning/default-setup":
+            return 0, {}
+        if path == "repos/example/repo/dependabot/alerts?per_page=1":
+            return 0, []
+        if path == "orgs/example/code-security/configurations/240284":
+            return 0, {
+                "id": 240284,
+                "secret_scanning": "enabled",
+                "secret_scanning_push_protection": "enabled",
+                "secret_scanning_non_provider_patterns": "enabled",
+                "secret_scanning_validity_checks": "enabled",
+            }
+        if path == "orgs/example/code-security/configurations/240284/repositories":
+            return 0, []
         raise AssertionError(path)
 
     return _impl
@@ -83,7 +145,7 @@ def test_github_control_plane_requires_secret_scanning_subfeatures(tmp_path: Pat
     policy_path.write_text(json.dumps(_policy_payload(), ensure_ascii=False, indent=2), encoding="utf-8")
 
     monkeypatch.setattr(module, "ROOT", tmp_path)
-    monkeypatch.setattr(module, "_gh_json", _fake_gh_json("disabled", "disabled"))
+    monkeypatch.setattr(module, "_gh_json", _fake_gh_json_without_org_enforcement("disabled", "disabled"))
     monkeypatch.setattr(module, "_repo_path_exists", lambda _path: True)
     monkeypatch.setattr(sys, "argv", ["check_github_control_plane.py", "--policy", str(policy_path), "--output", str(output_path)])
 
@@ -110,3 +172,21 @@ def test_github_control_plane_accepts_enabled_secret_scanning_subfeatures(tmp_pa
     assert rc == 0
     assert report["errors"] == []
     assert report["security_and_analysis"]["secret_scanning_non_provider_patterns"]["status"] == "enabled"
+
+
+def test_github_control_plane_accepts_enforced_org_config_fallback(tmp_path: Path, monkeypatch) -> None:
+    module = _load_module()
+    policy_path = tmp_path / "policy.json"
+    output_path = tmp_path / "report.json"
+    policy_path.write_text(json.dumps(_policy_payload(), ensure_ascii=False, indent=2), encoding="utf-8")
+
+    monkeypatch.setattr(module, "ROOT", tmp_path)
+    monkeypatch.setattr(module, "_gh_json", _fake_gh_json("disabled", "disabled"))
+    monkeypatch.setattr(module, "_repo_path_exists", lambda _path: True)
+    monkeypatch.setattr(sys, "argv", ["check_github_control_plane.py", "--policy", str(policy_path), "--output", str(output_path)])
+
+    rc = module.main()
+    report = json.loads(output_path.read_text(encoding="utf-8"))
+    assert rc == 0
+    assert report["errors"] == []
+    assert report["org_code_security_configuration"]["id"] == 240284

configs/github_control_plane_policy.json

@@ -42,6 +42,12 @@
       "required": true,
       "mode": "api"
     },
+    "org_code_security_configuration": {
+      "required": true,
+      "mode": "api",
+      "configuration_id": 240284,
+      "required_repository_status": "enforced"
+    },
     "dependabot_config": {
       "required": true,
       "mode": "repo_file",

scripts/check_github_control_plane.py

@@ -42,6 +42,41 @@ def _security_feature_status(repo_payload: dict, feature_name: str) -> str:
     return str(feature.get("status") or "").strip()
 
 
+def _org_configuration_id(platform_evidence: dict) -> str:
+    cfg = platform_evidence.get("org_code_security_configuration")
+    if not isinstance(cfg, dict):
+        return ""
+    return str(cfg.get("configuration_id") or "").strip()
+
+
+def _org_config_proves_feature(
+    org_config_payload: dict,
+    org_repo_payload: list[dict] | dict,
+    *,
+    repo_id: int | None,
+    feature_name: str,
+    required_repository_status: str,
+) -> bool:
+    if repo_id is None:
+        return False
+    feature_value = str(org_config_payload.get(feature_name) or "").strip()
+    if feature_value != "enabled":
+        return False
+    if not isinstance(org_repo_payload, list):
+        return False
+    for item in org_repo_payload:
+        if not isinstance(item, dict):
+            continue
+        repo = item.get("repository")
+        if not isinstance(repo, dict):
+            continue
+        if repo.get("id") != repo_id:
+            continue
+        status = str(item.get("status") or "").strip()
+        return status == required_repository_status
+    return False
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description="Validate live GitHub control-plane settings against repo policy.")
     parser.add_argument("--policy", default=str(DEFAULT_POLICY))
@@ -63,6 +98,11 @@ def main() -> int:
     vuln_alerts_code, vuln_alerts_payload = _gh_json(f"repos/{owner}/{repo}/vulnerability-alerts")
     codeql_code, codeql_payload = _gh_json(f"repos/{owner}/{repo}/code-scanning/default-setup")
     dependabot_code, dependabot_payload = _gh_json(f"repos/{owner}/{repo}/dependabot/alerts?per_page=1")
+    org_config_id = ""
+    org_config_code = 0
+    org_config_payload: dict = {}
+    org_config_repos_code = 0
+    org_config_repos_payload: list[dict] | dict = []
 
     if repo_code != 0:
         errors.append(f"gh api repo fetch failed: {repo_payload}")
@@ -105,13 +145,24 @@ def main() -> int:
 
     platform_evidence = policy.get("platform_evidence") if isinstance(policy.get("platform_evidence"), dict) else {}
     if platform_evidence:
+        org_config_id = _org_configuration_id(platform_evidence)
+        if org_config_id:
+            org_config_code, org_config_payload = _gh_json(f"orgs/{owner}/code-security/configurations/{org_config_id}")
+            org_config_repos_code, org_config_repos_payload = _gh_json(
+                f"orgs/{owner}/code-security/configurations/{org_config_id}/repositories"
+            )
         pvr_required = bool((platform_evidence.get("private_vulnerability_reporting") or {}).get("required"))
         if pvr_required:
             if pvr_code != 0:
                 errors.append(f"private vulnerability reporting not proven: {pvr_payload}")
         vulnerability_alerts_required = bool((platform_evidence.get("vulnerability_alerts") or {}).get("required"))
         if vulnerability_alerts_required and vuln_alerts_code != 0:
             errors.append(f"vulnerability alerts not proven: {vuln_alerts_payload}")
+        repo_id = repo_payload.get("id") if isinstance(repo_payload, dict) else None
+        required_repo_status = str(
+            ((platform_evidence.get("org_code_security_configuration") or {}) if isinstance(platform_evidence.get("org_code_security_configuration"), dict) else {}).get("required_repository_status")
+            or "enforced"
+        ).strip()
         for feature_name in (
             "secret_scanning",
             "secret_scanning_push_protection",
@@ -121,10 +172,22 @@ def main() -> int:
             feature_rule = platform_evidence.get(feature_name) if isinstance(platform_evidence.get(feature_name), dict) else {}
             if feature_rule.get("required"):
                 status = _security_feature_status(repo_payload, feature_name)
-                if status != "enabled":
+                if status != "enabled" and not _org_config_proves_feature(
+                    org_config_payload,
+                    org_config_repos_payload,
+                    repo_id=repo_id if isinstance(repo_id, int) else None,
+                    feature_name=feature_name,
+                    required_repository_status=required_repo_status,
+                ):
                     errors.append(
                         f"{feature_name} drift: actual={status or 'missing'!r} expected='enabled'"
                     )
+        org_cfg_rule = platform_evidence.get("org_code_security_configuration") if isinstance(platform_evidence.get("org_code_security_configuration"), dict) else {}
+        if org_cfg_rule.get("required"):
+            if org_config_code != 0:
+                errors.append(f"org code-security configuration not proven: {org_config_payload}")
+            elif org_config_repos_code != 0:
+                errors.append(f"org code-security configuration repository binding not proven: {org_config_repos_payload}")
         dependabot_rule = platform_evidence.get("dependabot_config") if isinstance(platform_evidence.get("dependabot_config"), dict) else {}
         dependabot_path = str(dependabot_rule.get("path") or "").strip()
         if dependabot_path and not _repo_path_exists(dependabot_path):
@@ -161,6 +224,8 @@ def main() -> int:
         "private_vulnerability_reporting": pvr_payload if pvr_code == 0 else {"error": pvr_payload},
         "vulnerability_alerts": {"enabled": True} if vuln_alerts_code == 0 else {"error": vuln_alerts_payload},
         "security_and_analysis": repo_payload.get("security_and_analysis") if repo_code == 0 else {"error": repo_payload},
+        "org_code_security_configuration": org_config_payload if org_config_id and org_config_code == 0 else {"error": org_config_payload} if org_config_id else {},
+        "org_code_security_configuration_repositories": org_config_repos_payload if org_config_id and org_config_repos_code == 0 else {"error": org_config_repos_payload} if org_config_id else {},
         "codeql_default_setup": codeql_payload if codeql_code == 0 else {"error": codeql_payload},
         "dependabot_alerts": dependabot_payload if dependabot_code == 0 else {"error": dependabot_payload},
         "errors": errors,