🧹 Refactor _verify_package_control_files to reduce argument count

.gitignore

@@ -22,3 +22,4 @@ evaluations/results/
 # Installed locally from an exact source lock. Upstream redistribution permission is unresolved.
 skills/alibabacloud-resourcecenter-search/
 venv/
+venv/

src/titmas_action_gate/service.py

@@ -5,6 +5,7 @@
 import hashlib
 import hmac
 import json
+from dataclasses import dataclass
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any
@@ -20,7 +21,6 @@
 from .provider import GitHubProvider
 from .signing import HmacRecordSigner
 from .store import AppendOnlyStore
-from dataclasses import dataclass
 
 
 @dataclass(frozen=True)

src/titmas_action_gate/skill_materialization.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import concurrent.futures
 import hashlib
 import json
 import re
@@ -72,32 +73,40 @@ def _source_inventory(
 
     contents: dict[str, bytes] = {}
     files: list[dict[str, str]] = []
+
+    def _read_and_hash(path: Path) -> tuple[Path, bytes, str]:
+        data = path.read_bytes()
+        return path, data, hashlib.sha256(data).hexdigest()
+
     if skill_name != OFFICIAL_CLOUD_SKILL:
-        for source_path in sorted(path for path in skill_root.rglob("*") if path.is_file()):
-            relative = source_path.relative_to(skill_root).as_posix()
-            package_path = f"skills/{skill_name}/{relative}"
-            data = source_path.read_bytes()
+        source_paths = sorted(path for path in skill_root.rglob("*") if path.is_file())
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            for source_path, data, sha256_hash in executor.map(_read_and_hash, source_paths):
+                relative = source_path.relative_to(skill_root).as_posix()
+                package_path = f"skills/{skill_name}/{relative}"
+                contents[package_path] = data
+                files.append(
+                    {
+                        "package_path": package_path,
+                        "source_path": source_path.relative_to(root).as_posix(),
+                        "sha256": sha256_hash,
+                    }
+                )
+    schema_paths = sorted((root / "schemas").glob("*.json"))
+    if schema_names is not None:
+        schema_paths = [p for p in schema_paths if p.name in schema_names]
+
+    with concurrent.futures.ThreadPoolExecutor() as executor:
+        for schema_path, data, sha256_hash in executor.map(_read_and_hash, schema_paths):
+            package_path = f"schemas/{schema_path.name}"
             contents[package_path] = data
             files.append(
                 {
                     "package_path": package_path,
-                    "source_path": source_path.relative_to(root).as_posix(),
-                    "sha256": hashlib.sha256(data).hexdigest(),
+                    "source_path": schema_path.relative_to(root).as_posix(),
+                    "sha256": sha256_hash,
                 }
             )
-    for schema_path in sorted((root / "schemas").glob("*.json")):
-        if schema_names is not None and schema_path.name not in schema_names:
-            continue
-        package_path = f"schemas/{schema_path.name}"
-        data = schema_path.read_bytes()
-        contents[package_path] = data
-        files.append(
-            {
-                "package_path": package_path,
-                "source_path": schema_path.relative_to(root).as_posix(),
-                "sha256": hashlib.sha256(data).hexdigest(),
-            }
-        )
     return skill_version, files, contents
 
 
@@ -352,29 +361,16 @@ def _verify_package_hashes(archive: zipfile.ZipFile, attestation: dict[str, Any]
 
 def _verify_package_control_files(
     archive: zipfile.ZipFile,
-    workers: dict[str, Any],
-    expected_worker: str,
-    source_commit: str,
-    model: str,
-    runtime: str,
-    expected_version: str,
-    expected_files: list[dict[str, str]],
+    worker: dict[str, Any],
+    expected_manifest: dict[str, Any],
+    expected_attestation: dict[str, Any],
 ) -> dict[str, Any]:
     package_manifest = json.loads(archive.read("manifest.json"))
-    expected_manifest = _package_manifest(workers[expected_worker], source_commit=source_commit, model=model, runtime=runtime)
-    expected_attestation = _attestation(
-        workers[expected_worker],
-        source_commit=source_commit,
-        model=model,
-        runtime=runtime,
-        skill_version=expected_version,
-        source_files=expected_files,
-    )
     expected_special = {
         "manifest.json": _json_bytes(expected_manifest),
         "skill-attestation.json": _json_bytes(expected_attestation),
-        "config/AGENTS.md": _agents_markdown(workers[expected_worker], workers[expected_worker]["skills"][0]),
-        "config/SOUL.md": _soul_markdown(workers[expected_worker]),
+        "config/AGENTS.md": _agents_markdown(worker, worker["skills"][0]),
+        "config/SOUL.md": _soul_markdown(worker),
     }
     if any(archive.read(name) != data for name, data in expected_special.items()):
         raise ActionGateError("SKILL_DIGEST_MISMATCH", "Worker package control files do not match deterministic source.")
@@ -421,15 +417,22 @@ def verify_worker_package(
             expected_model,
         )
         _verify_package_hashes(archive, attestation, expected_hashes)
+
+        expected_manifest = _package_manifest(workers[expected_worker], source_commit=source_commit, model=model, runtime=runtime)
+        expected_attestation = _attestation(
+            workers[expected_worker],
+            source_commit=source_commit,
+            model=model,
+            runtime=runtime,
+            skill_version=expected_version,
+            source_files=expected_files,
+        )
+
         package_manifest = _verify_package_control_files(
             archive,
-            workers,
-            expected_worker,
-            source_commit,
-            model,
-            runtime,
-            expected_version,
-            expected_files,
+            workers[expected_worker],
+            expected_manifest,
+            expected_attestation,
         )
 
     skill_name = workers[expected_worker]["skills"][0]

tests/test_runtime_mcp_server.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import os
+import unittest
+from unittest.mock import MagicMock, patch
+
+from titmas_action_gate.runtime_mcp_server import build_from_environment, configured_runtime_host
+
+
+class RuntimeMcpServerConfigurationTests(unittest.TestCase):
+    @patch.dict(os.environ, {"TITMAS_ACTION_GATE_RUNTIME_MCP_HOST": "127.0.0.1"})
+    def test_configured_runtime_host_valid(self) -> None:
+        self.assertEqual(configured_runtime_host(), "127.0.0.1")
+
+    @patch.dict(os.environ, {"TITMAS_ACTION_GATE_RUNTIME_MCP_HOST": "invalid_host"})
+    def test_configured_runtime_host_invalid(self) -> None:
+        with self.assertRaises(ValueError) as context:
+            configured_runtime_host()
+        self.assertIn("must be loopback or 0.0.0.0 for a disposable run", str(context.exception))
+
+    @patch.dict(os.environ, {}, clear=True)
+    def test_build_from_environment_missing_vars(self) -> None:
+        with self.assertRaises(RuntimeError) as context:
+            build_from_environment()
+        self.assertIn("runtime MCP requires state dir, 0600 credentials file, caller token, and approver token", str(context.exception))
+
+    @patch.dict(
+        os.environ,
+        {
+            "TITMAS_ACTION_GATE_STATE_DIR": "/tmp",
+            "TITMAS_ACTION_GATE_RUNTIME_CREDENTIALS_FILE": "/tmp/creds",
+            "TITMAS_ACTION_GATE_CALLER_TOKEN": "caller",
+            "TITMAS_ACTION_GATE_APPROVER_TOKEN": "approver",
+        },
+        clear=True,
+    )
+    @patch("titmas_action_gate.runtime_mcp_server.RuntimePrincipalRegistry")
+    def test_build_from_environment_missing_demo_mode(self, mock_registry: MagicMock) -> None:
+        mock_registry.from_file.return_value = MagicMock()
+        with self.assertRaises(RuntimeError) as context:
+            build_from_environment()
+        self.assertIn("native M4 runtime server currently permits only explicit disposable demo mode", str(context.exception))
+
+    @patch.dict(
+        os.environ,
+        {
+            "TITMAS_ACTION_GATE_STATE_DIR": "/tmp",
+            "TITMAS_ACTION_GATE_RUNTIME_CREDENTIALS_FILE": "/tmp/creds",
+            "TITMAS_ACTION_GATE_CALLER_TOKEN": "caller",
+            "TITMAS_ACTION_GATE_APPROVER_TOKEN": "approver",
+            "TITMAS_ACTION_GATE_DEMO_MODE": "true",
+            "TITMAS_ALIBABA_CLOUD_PROFILE": "profile",
+            "TITMAS_ALIBABA_RAM_POLICY_OBSERVATION": "observation",
+        },
+        clear=True,
+    )
+    @patch("titmas_action_gate.runtime_mcp_server.RuntimePrincipalRegistry")
+    @patch("titmas_action_gate.runtime_mcp_server.ActionGateService")
+    def test_build_from_environment_partial_cloud_config(self, mock_service: MagicMock, mock_registry: MagicMock) -> None:
+        mock_registry.from_file.return_value = MagicMock()
+        with self.assertRaises(RuntimeError) as context:
+            build_from_environment()
+        self.assertIn("Alibaba Cloud runtime references must be configured as a complete set", str(context.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_security_argument_injection.py

@@ -1,9 +1,8 @@
 import unittest
-import subprocess
-from unittest.mock import patch, MagicMock
+from unittest.mock import MagicMock, patch
 
 from titmas_action_gate.provider import GhCliProvider
-from titmas_action_gate.errors import ActionGateError
+
 
 class ProviderSecurityTests(unittest.TestCase):
     @patch("subprocess.run")

tests/test_signing.py

@@ -0,0 +1,69 @@
+import copy
+import unittest
+
+from titmas_action_gate.signing import HmacRecordSigner
+
+
+class TestHmacRecordSigner(unittest.TestCase):
+    def setUp(self):
+        self.valid_key = b"0123456789abcdef0123456789abcdef"
+        self.payload = {"test": "data", "count": 1}
+
+    def test_init_valid_key(self):
+        signer = HmacRecordSigner(self.valid_key, key_id="test-key")
+        self.assertEqual(signer._key, self.valid_key)
+        self.assertEqual(signer.key_id, "test-key")
+
+    def test_init_invalid_key(self):
+        with self.assertRaises(ValueError) as ctx:
+            HmacRecordSigner(b"too_short")
+        self.assertIn("must be at least 32 bytes", str(ctx.exception))
+
+    def test_sign(self):
+        signer = HmacRecordSigner(self.valid_key, key_id="test-key")
+        signature = signer.sign(self.payload)
+
+        self.assertEqual(signature["algorithm"], "HMAC-SHA256-DEMO_ONLY")
+        self.assertEqual(signature["key_id"], "test-key")
+        self.assertIn("payload_sha256", signature)
+        self.assertIn("value", signature)
+        self.assertIsInstance(signature["value"], str)
+
+    def test_verify_success(self):
+        signer = HmacRecordSigner(self.valid_key, key_id="test-key")
+        signature = signer.sign(self.payload)
+        self.assertTrue(signer.verify(self.payload, signature))
+
+    def test_verify_failure_tampered_payload(self):
+        signer = HmacRecordSigner(self.valid_key, key_id="test-key")
+        signature = signer.sign(self.payload)
+        tampered_payload = {"test": "data", "count": 2}
+        self.assertFalse(signer.verify(tampered_payload, signature))
+
+    def test_verify_failure_tampered_signature_value(self):
+        signer = HmacRecordSigner(self.valid_key, key_id="test-key")
+        signature = signer.sign(self.payload)
+        tampered_signature = copy.deepcopy(signature)
+        tampered_signature["value"] = "0" * 64
+        self.assertFalse(signer.verify(self.payload, tampered_signature))
+
+    def test_verify_failure_different_key_id(self):
+        signer = HmacRecordSigner(self.valid_key, key_id="test-key")
+        signature = signer.sign(self.payload)
+        tampered_signature = copy.deepcopy(signature)
+        tampered_signature["key_id"] = "different-key"
+        self.assertFalse(signer.verify(self.payload, tampered_signature))
+
+    def test_verify_failure_different_algorithm(self):
+        signer = HmacRecordSigner(self.valid_key, key_id="test-key")
+        signature = signer.sign(self.payload)
+        tampered_signature = copy.deepcopy(signature)
+        tampered_signature["algorithm"] = "HMAC-SHA512"
+        self.assertFalse(signer.verify(self.payload, tampered_signature))
+
+    def test_verify_missing_fields(self):
+        signer = HmacRecordSigner(self.valid_key, key_id="test-key")
+        self.assertFalse(signer.verify(self.payload, {}))
+
+if __name__ == '__main__':
+    unittest.main()