test(coverage): add hotspot readiness governance

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(

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",

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)},

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()