diff --git a/scripts/quality_governance_common.py b/scripts/quality_governance_common.py index bb43427..cce77df 100644 --- a/scripts/quality_governance_common.py +++ b/scripts/quality_governance_common.py @@ -14,6 +14,7 @@ REQUIRED_HOTSPOT_FAMILIES = ( "config_bootstrap", ) MIN_PROMOTION_REVIEW_CYCLES = 2 +RATCHET55_CRITICAL_FAMILIES = ("safe_io", "security_boundary") @dataclass(frozen=True) @@ -52,6 +53,28 @@ def _validate_hotspot_family( ) +def _validate_ratchet55_readiness(family: dict[str, Any], failures: list[str]) -> None: + family_id = family["id"] + readiness = family.get("ratchet55_readiness") + if not isinstance(readiness, dict): + failures.append( + f"coverage policy: hotspot family {family_id} must include ratchet55_readiness metadata" + ) + return + + for field_name in ( + "targeted_regression_suite", + "ownership_status", + "readiness_notes", + ): + value = readiness.get(field_name) + if not isinstance(value, str) or not value.strip(): + failures.append( + "coverage policy: hotspot family " + f"{family_id} ratchet55_readiness missing {field_name}" + ) + + def load_and_validate_policy(path: Path) -> tuple[dict[str, Any] | None, list[str]]: failures: list[str] = [] if not path.is_file(): @@ -142,6 +165,24 @@ def load_and_validate_policy(path: Path) -> tuple[dict[str, Any] | None, list[st continue _validate_hotspot_family(family, seen_family_ids, failures) + policy_next_stage = None + if current_stage in stage_ids: + for index, raw_stage in enumerate(stages_raw): + if raw_stage.get("id") == current_stage: + if index + 1 < len(stages_raw): + policy_next_stage = stages_raw[index + 1].get("id") + break + if policy_next_stage == "ratchet-55": + families_by_id = { + family.get("id"): family + for family in family_payload + if isinstance(family, dict) and isinstance(family.get("id"), str) + } + for family_id in RATCHET55_CRITICAL_FAMILIES: + family = families_by_id.get(family_id) + if family is not None: + _validate_ratchet55_readiness(family, failures) + missing_declared_required = sorted(set(required_families) - seen_family_ids) if missing_declared_required: failures.append( diff --git a/tests/coverage_governance_policy.json b/tests/coverage_governance_policy.json index 40b6de3..f1d9154 100644 --- a/tests/coverage_governance_policy.json +++ b/tests/coverage_governance_policy.json @@ -50,7 +50,12 @@ "id": "safe_io", "paths": [ "services/safe_io.py" - ] + ], + "ratchet55_readiness": { + "targeted_regression_suite": "tests/test_safe_io.py; tests/test_callback_url_policy.py; tests/test_s36s37r79_egress_hardening.py; tests/test_r185_hotspot_regression_ownership.py", + "ownership_status": "targeted-regression-owned", + "readiness_notes": "SSRF deny-by-default, host allowlist, private-IP blocking, DNS pinning, and redirect revalidation are covered by focused safe_io suites before any ratchet-55 proposal." + } }, { "id": "security_boundary", @@ -59,7 +64,12 @@ "services/access_control.py", "services/csrf_protection.py", "services/secrets_encryption.py" - ] + ], + "ratchet55_readiness": { + "targeted_regression_suite": "tests/security/test_startup_gate.py; tests/security/test_r99_sensitive_contract.py; tests/security/test_rbac.py; tests/test_s57_secrets_encryption.py; tests/test_r185_hotspot_regression_ownership.py", + "ownership_status": "targeted-regression-owned", + "readiness_notes": "Startup gate fail-closed, auth/RBAC, CSRF convenience-mode denial, and secret-envelope tamper/failure behavior have explicit targeted suites before any ratchet-55 proposal." + } }, { "id": "connector_config", diff --git a/tests/test_r180_exception_boundary_governance.py b/tests/test_r180_exception_boundary_governance.py index 7e5c76d..168e46f 100644 --- a/tests/test_r180_exception_boundary_governance.py +++ b/tests/test_r180_exception_boundary_governance.py @@ -20,7 +20,9 @@ class TestExceptionBoundaryGovernance(unittest.TestCase): "services.route_bootstrap._do_full_registration", side_effect=RuntimeError("route-registration-broken"), ), - patch("services.route_bootstrap.logging.getLogger", return_value=MagicMock()), + patch( + "services.route_bootstrap.logging.getLogger", return_value=MagicMock() + ), patch.dict( sys.modules, {"server": SimpleNamespace(PromptServer=prompt_server)}, diff --git a/tests/test_r185_hotspot_regression_ownership.py b/tests/test_r185_hotspot_regression_ownership.py new file mode 100644 index 0000000..41517ad --- /dev/null +++ b/tests/test_r185_hotspot_regression_ownership.py @@ -0,0 +1,160 @@ +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from services.csrf_protection import require_same_origin_if_no_token +from services.safe_io import SSRFError, safe_fetch, validate_outbound_url +from tests.quality_governance_test_utils import ( + sample_policy_payload, + write_governance_baseline_fixture, +) + +ROOT = Path(__file__).resolve().parents[1] +GOVERNANCE_SCRIPT = ROOT / "scripts" / "verify_quality_governance.py" + + +class TestR185HotspotRegressionOwnership(unittest.TestCase): + def _run_governance_script(self, *args): + return subprocess.run( + [sys.executable, str(GOVERNANCE_SCRIPT), *args], + capture_output=True, + text=True, + cwd=str(ROOT), + check=False, + ) + + def test_ratchet55_critical_families_require_targeted_regression_metadata(self): + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + fixture = write_governance_baseline_fixture( + tmp, + fail_under=45.0, + coverage_policy_payload=sample_policy_payload( + current_stage="ratchet-45", + stages=[ + {"id": "baseline-35", "min_fail_under": 35.0}, + {"id": "ratchet-45", "min_fail_under": 45.0}, + {"id": "ratchet-55", "min_fail_under": 55.0}, + ], + hotspot_families=[ + {"id": "safe_io", "paths": ["services/safe_io.py"]}, + { + "id": "security_boundary", + "paths": ["services/security_gate.py"], + }, + { + "id": "connector_config", + "paths": ["connector/config.py"], + }, + {"id": "config_bootstrap", "paths": ["config.py"]}, + ], + ), + ) + + result = self._run_governance_script( + "--pyproject", + str(fixture["pyproject"]), + "--adversarial-gate", + str(fixture["adversarial_gate"]), + "--test-sop", + str(fixture["test_sop"]), + "--mutation-survivor-allowlist", + str(fixture["survivor_allowlist"]), + "--release-policy-doc", + str(fixture["release_policy_doc"]), + "--coverage-policy", + str(fixture["coverage_policy"]), + ) + + self.assertNotEqual(result.returncode, 0) + self.assertIn("ratchet55_readiness", result.stdout) + self.assertIn("safe_io", result.stdout) + self.assertIn("security_boundary", result.stdout) + + def test_safe_io_redirect_revalidates_second_hop_before_connect(self): + class RedirectResponse: + headers = {"Location": "http://169.254.169.254/latest/meta-data"} + + def getcode(self): + return 302 + + def close(self): + pass + + opener = MagicMock() + opener.open.return_value = RedirectResponse() + + with ( + patch("services.safe_io.socket.getaddrinfo") as getaddrinfo, + patch("services.safe_io._build_pinned_opener", return_value=opener), + ): + getaddrinfo.side_effect = [ + [ + ( + None, + None, + None, + None, + ("93.184.216.34", 443), + ) + ], + [ + ( + None, + None, + None, + None, + ("169.254.169.254", 80), + ) + ], + ] + + with self.assertRaises(SSRFError) as ctx: + safe_fetch( + "https://example.com/start", + allow_hosts={"example.com"}, + max_redirects=1, + ) + + self.assertIn("Host not in allowlist", str(ctx.exception)) + opener.open.assert_called_once() + + def test_security_boundary_denies_cross_origin_convenience_request(self): + request = SimpleNamespace( + path="/openclaw/config", + headers={"Origin": "https://attacker.example"}, + ) + + with patch("services.csrf_protection.logger.warning"): + response = require_same_origin_if_no_token( + request, + admin_token_configured=False, + ) + + self.assertIsNotNone(response) + self.assertEqual(response.status, 403) + self.assertIn("csrf_protection", response.text) + + def test_security_gate_fails_closed_on_fatal_error(self): + from services.security_gate import enforce_startup_gate + + with ( + patch( + "services.security_gate.SecurityGate.verify_mandatory_controls", + return_value=(False, [], ["fatal boundary violation"]), + ), + patch("services.security_gate.logger"), + ): + with self.assertRaises(RuntimeError) as ctx: + enforce_startup_gate() + + self.assertIn("fatal boundary violation", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main()