fix(security): require CMCP_POLICY_HASH and CMCP_CATALOG_HASH in production

src/cmcp_gateway/startup.py

@@ -106,6 +106,17 @@ def run_startup(config_path: str) -> GatewayContext:
 
     # Step 4: policy bundle
     policy_expected_hash = os.environ.get("CMCP_POLICY_HASH")
+    if policy_expected_hash is None and not config.dev_mode:
+        # POLICY-001 (CRITICAL): without a pinned hash, a compromised policy bundle
+        # loads silently. Require CMCP_POLICY_HASH in production; set CMCP_DEV_MODE=1
+        # only for local development.
+        _fatal(
+            "POLICY_HASH_REQUIRED",
+            "CMCP_POLICY_HASH env var is not set. "
+            "Set it to the sha256:<hex> of the policy bundle to prevent policy tampering. "
+            "Set CMCP_DEV_MODE=1 only in development to skip this check.",
+        )
+        sys.exit(1)
     try:
         policy_bundle = load_policy_bundle(config.policy_bundle_path, expected_hash=policy_expected_hash)
     except PolicyHashMismatch as exc:
@@ -124,6 +135,16 @@ def run_startup(config_path: str) -> GatewayContext:
 
     # Step 5: catalog
     catalog_expected_hash = os.environ.get("CMCP_CATALOG_HASH")
+    if catalog_expected_hash is None and not config.dev_mode:
+        # POLICY-002 (CRITICAL): without a pinned hash, a compromised catalog loads
+        # silently, allowing unauthorized tools. Require CMCP_CATALOG_HASH in production.
+        _fatal(
+            "CATALOG_HASH_REQUIRED",
+            "CMCP_CATALOG_HASH env var is not set. "
+            "Set it to the sha256:<hex> of the tool catalog to prevent catalog tampering. "
+            "Set CMCP_DEV_MODE=1 only in development to skip this check.",
+        )
+        sys.exit(1)
     try:
         catalog = load_catalog(config.catalog_path, expected_hash=catalog_expected_hash)
     except CatalogHashMismatch as exc:

tests/unit/test_startup.py

@@ -118,3 +118,47 @@ def test_startup_fails_on_catalog_hash_mismatch(complete_setup, monkeypatch):
     with pytest.raises(SystemExit) as exc_info:
         run_startup(complete_setup)
     assert exc_info.value.code == 1
+
+
+def test_startup_fails_when_policy_hash_unset_and_not_dev_mode(tmp_path):
+    """POLICY-001 (CRITICAL): CMCP_POLICY_HASH must be set outside dev mode."""
+    config_path = tmp_path / "cmcp-config.yaml"
+    policy_dir = tmp_path / "policy"
+    policy_dir.mkdir()
+    catalog_path = tmp_path / "catalog.json"
+    config_path.write_text(f"policy_bundle_path: {policy_dir}\ncatalog_path: {catalog_path}\n")
+    (policy_dir / "manifest.json").write_text(json.dumps(MANIFEST))
+    (policy_dir / "allow.cedar").write_text(CEDAR_POLICY)
+    (policy_dir / "schema.cedarschema").write_text(SCHEMA)
+    catalog_path.write_text(json.dumps([CATALOG_ENTRY]))
+
+    env = {"CMCP_DEV_MODE": "0"}
+    with patch.dict(os.environ, env, clear=True), pytest.raises(SystemExit) as exc_info:
+        run_startup(str(config_path))
+    assert exc_info.value.code == 1
+
+
+def test_startup_fails_when_catalog_hash_unset_and_not_dev_mode(tmp_path, monkeypatch):
+    """POLICY-002 (CRITICAL): CMCP_CATALOG_HASH must be set outside dev mode."""
+    config_path = tmp_path / "cmcp-config.yaml"
+    policy_dir = tmp_path / "policy"
+    policy_dir.mkdir()
+    catalog_path = tmp_path / "catalog.json"
+    config_path.write_text(f"policy_bundle_path: {policy_dir}\ncatalog_path: {catalog_path}\n")
+    (policy_dir / "manifest.json").write_text(json.dumps(MANIFEST))
+    (policy_dir / "allow.cedar").write_text(CEDAR_POLICY)
+    (policy_dir / "schema.cedarschema").write_text(SCHEMA)
+    catalog_path.write_text(json.dumps([CATALOG_ENTRY]))
+
+    import json as _json
+
+    from cmcp_gateway.policy.bundle import _canonical_bundle_hash
+    manifest_raw = _json.loads((policy_dir / "manifest.json").read_text())
+    policy_files = {"allow.cedar": CEDAR_POLICY}
+    computed = _canonical_bundle_hash(manifest_raw, policy_files, SCHEMA)
+    policy_hash = f"sha256:{computed}"
+
+    env = {"CMCP_DEV_MODE": "0", "CMCP_POLICY_HASH": policy_hash}
+    with patch.dict(os.environ, env, clear=True), pytest.raises(SystemExit) as exc_info:
+        run_startup(str(config_path))
+    assert exc_info.value.code == 1