From 5cfc0065e5d4081b70d3f1d352ef8ee7ea8bdec1 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Sun, 23 Aug 2026 02:09:10 +0000 Subject: [PATCH 1/3] Refactor `_verify_package_control_files` to fix Too Many Arguments Extract expected_manifest and expected_attestation generation to caller `verify_worker_package` to reduce the number of parameters of `_verify_package_control_files` from 8 down to 4. Also use direct `worker` dict instead of passing `workers` collection and `expected_worker` id to improve encapsulation. Co-authored-by: joy7758 <138868899+joy7758@users.noreply.github.com> --- .../skill_materialization.py | 44 ++++++++----------- 1 file changed, 19 insertions(+), 25 deletions(-) diff --git a/src/titmas_action_gate/skill_materialization.py b/src/titmas_action_gate/skill_materialization.py index 4d7ad96..edf1457 100644 --- a/src/titmas_action_gate/skill_materialization.py +++ b/src/titmas_action_gate/skill_materialization.py @@ -352,29 +352,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 +408,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] From 5047669920ce4f9ae77ab9bebdb512ea5df49313 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Sun, 23 Aug 2026 02:12:53 +0000 Subject: [PATCH 2/3] Fix ruff lint errors in `test_workflow.py`, `test_security_argument_injection.py` and `service.py` Co-authored-by: joy7758 <138868899+joy7758@users.noreply.github.com> --- src/titmas_action_gate/service.py | 2 +- tests/test_security_argument_injection.py | 5 ++--- tests/test_workflow.py | 6 +++--- 3 files changed, 6 insertions(+), 7 deletions(-) diff --git a/src/titmas_action_gate/service.py b/src/titmas_action_gate/service.py index 90d7164..df6bc08 100644 --- a/src/titmas_action_gate/service.py +++ b/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) diff --git a/tests/test_security_argument_injection.py b/tests/test_security_argument_injection.py index 3990256..4d8bb42 100644 --- a/tests/test_security_argument_injection.py +++ b/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") diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 61c356d..461a3bf 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -1,8 +1,8 @@ +import json import tempfile import unittest from pathlib import Path -import json from titmas_action_gate.workflow import validate_agentteams_template, write_demo_report @@ -83,7 +83,7 @@ def test_write_demo_report(self): self.assertEqual(result_path, output_file) self.assertTrue(output_file.exists()) - with open(output_file, "r", encoding="utf-8") as f: + with open(output_file, encoding="utf-8") as f: content = json.load(f) self.assertEqual(content, report_data) @@ -95,7 +95,7 @@ def test_write_demo_report_creates_directories(self): write_demo_report(report_data, output_file) self.assertTrue(output_file.exists()) - with open(output_file, "r", encoding="utf-8") as f: + with open(output_file, encoding="utf-8") as f: content = json.load(f) self.assertEqual(content, report_data) From 7da06a183eebf6cc0d9c87e45d26871a55366826 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Sun, 23 Aug 2026 02:18:20 +0000 Subject: [PATCH 3/3] Fix lint error in tests/test_signing.py after merging origin/main Co-authored-by: joy7758 <138868899+joy7758@users.noreply.github.com> --- .gitignore | 1 + benchmark.py | 40 ---------- .../skill_materialization.py | 47 ++++++----- tests/test_runtime_mcp_server.py | 67 ++++++++++++++++ tests/test_signing.py | 69 ++++++++++++++++ tests/test_store.py | 80 +++++++++++++++++++ 6 files changed, 245 insertions(+), 59 deletions(-) delete mode 100644 benchmark.py create mode 100644 tests/test_runtime_mcp_server.py create mode 100644 tests/test_signing.py create mode 100644 tests/test_store.py diff --git a/.gitignore b/.gitignore index ae69a1f..11676b0 100644 --- a/.gitignore +++ b/.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/ diff --git a/benchmark.py b/benchmark.py deleted file mode 100644 index 9347732..0000000 --- a/benchmark.py +++ /dev/null @@ -1,40 +0,0 @@ -import re -import timeit - -source_commit = "1234567890abcdef1234567890abcdef12345678" -invalid_commit = "1234567890abcdef1234567890abcdef1234567g" -hex_set = set("0123456789abcdef") -hex_pattern = re.compile(r"^[0-9a-f]{40}$") - - -def original(val): - return not isinstance(val, str) or len(val) != 40 or any(v not in "0123456789abcdef" for v in val) - - -def with_set(val): - return not isinstance(val, str) or len(val) != 40 or not set(val).issubset(hex_set) - - -def with_regex(val): - return not isinstance(val, str) or not hex_pattern.match(val) - - -def with_int(val): - if not isinstance(val, str) or len(val) != 40: - return True - try: - # this accepts uppercase and signs, so maybe not exact match - int(val, 16) - return not val.islower() and not val.isdigit() # this is complicated - except ValueError: - return True - - -print("Original (valid):", timeit.timeit("original(source_commit)", globals=globals(), number=100000)) -print("Original (invalid):", timeit.timeit("original(invalid_commit)", globals=globals(), number=100000)) - -print("Set (valid):", timeit.timeit("with_set(source_commit)", globals=globals(), number=100000)) -print("Set (invalid):", timeit.timeit("with_set(invalid_commit)", globals=globals(), number=100000)) - -print("Regex (valid):", timeit.timeit("with_regex(source_commit)", globals=globals(), number=100000)) -print("Regex (invalid):", timeit.timeit("with_regex(invalid_commit)", globals=globals(), number=100000)) diff --git a/src/titmas_action_gate/skill_materialization.py b/src/titmas_action_gate/skill_materialization.py index edf1457..ce52a21 100644 --- a/src/titmas_action_gate/skill_materialization.py +++ b/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 diff --git a/tests/test_runtime_mcp_server.py b/tests/test_runtime_mcp_server.py new file mode 100644 index 0000000..10a572b --- /dev/null +++ b/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() diff --git a/tests/test_signing.py b/tests/test_signing.py new file mode 100644 index 0000000..59390d7 --- /dev/null +++ b/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() diff --git a/tests/test_store.py b/tests/test_store.py new file mode 100644 index 0000000..b8018db --- /dev/null +++ b/tests/test_store.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +import tempfile +import unittest +from datetime import UTC, datetime +from pathlib import Path + +from titmas_action_gate.errors import ConflictError, NotFoundError +from titmas_action_gate.store import AppendOnlyStore + + +class StoreTests(unittest.TestCase): + def setUp(self) -> None: + self.tempdir = tempfile.TemporaryDirectory() + self.path = Path(self.tempdir.name) / "gate.sqlite3" + self.store = AppendOnlyStore(self.path) + + def tearDown(self) -> None: + self.tempdir.cleanup() + + def test_append_and_get_record(self) -> None: + created_at = datetime(2026, 8, 2, 0, 0, tzinfo=UTC) + record = self.store.append_record( + record_type="test", + record_id="r1", + request_id="req1", + payload={"a": 1}, + created_at=created_at, + ) + self.assertEqual(record["record_type"], "test") + self.assertEqual(record["record_id"], "r1") + self.assertEqual(record["request_id"], "req1") + self.assertEqual(record["payload"], {"a": 1}) + + fetched = self.store.get_record("r1") + self.assertEqual(fetched, record) + + def test_get_record_not_found(self) -> None: + with self.assertRaises(NotFoundError): + self.store.get_record("nonexistent") + + def test_append_duplicate_record_id_conflict(self) -> None: + self.store.append_record( + record_type="test", + record_id="r1", + request_id="req1", + payload={"a": 1}, + ) + with self.assertRaises(ConflictError): + self.store.append_record( + record_type="test", + record_id="r1", + request_id="req1", + payload={"a": 2}, + ) + + def test_records_for_request(self) -> None: + self.store.append_record(record_type="type1", record_id="1", request_id="req1", payload={}) + self.store.append_record(record_type="type2", record_id="2", request_id="req1", payload={}) + self.store.append_record(record_type="type1", record_id="3", request_id="req2", payload={}) + + records = self.store.records_for_request("req1") + self.assertEqual(len(records), 2) + self.assertEqual(records[0]["record_id"], "1") + self.assertEqual(records[1]["record_id"], "2") + + def test_latest_for_request(self) -> None: + self.store.append_record(record_type="type1", record_id="1", request_id="req1", payload={"v": 1}) + self.store.append_record(record_type="type1", record_id="2", request_id="req1", payload={"v": 2}) + + latest = self.store.latest_for_request("req1", "type1") + self.assertEqual(latest["record_id"], "2") + self.assertEqual(latest["payload"]["v"], 2) + + def test_latest_for_request_not_found(self) -> None: + with self.assertRaises(NotFoundError): + self.store.latest_for_request("req1", "type1") + +if __name__ == "__main__": + unittest.main()