import hashlib import importlib.machinery import importlib.util import io import json import os from pathlib import Path import subprocess import tarfile import tempfile import unittest from contextlib import redirect_stdout from unittest import mock SCRIPT_DIR = Path(__file__).resolve().parent RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy" BUILDER_PATH = ( SCRIPT_DIR / "build-engine-mcp-l1-credential-reuse-artifact.mjs" ) ENGINE_ROOT = SCRIPT_DIR.parent.parent.parent / "NODEDC_ENGINE_INFRA" PATCH_ID = "engine-mcp-l1-credential-reuse-20991231-999" def load_runner(): loader = importlib.machinery.SourceFileLoader( "nodedc_engine_mcp_l1_credential_reuse", str(RUNNER_PATH), ) spec = importlib.util.spec_from_loader(loader.name, loader) module = importlib.util.module_from_spec(spec) loader.exec_module(module) return module RUNNER = load_runner() class EngineMcpL1CredentialReuseTest(unittest.TestCase): def require_current_target_source(self): for relative_path, expected in ( RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_TARGET_SHA256.items() ): path = ENGINE_ROOT / relative_path if ( not path.is_file() or hashlib.sha256(path.read_bytes()).hexdigest() != expected ): self.skipTest("L1 credential reuse source has advanced") def build(self, artifact_dir, engine_root=ENGINE_ROOT): environment = os.environ.copy() environment["NODEDC_ENGINE_SOURCE_ROOT"] = str(engine_root) environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir) completed = subprocess.run( ["node", str(BUILDER_PATH), PATCH_ID], check=True, capture_output=True, text=True, env=environment, ) return json.loads(completed.stdout) def test_builder_emits_exact_deterministic_backend_only_successor(self): self.require_current_target_source() with tempfile.TemporaryDirectory( prefix="nodedc-engine-mcp-l1-credential-reuse-" ) as directory: root = Path(directory) first = self.build(root / "first") second = self.build(root / "second") first_artifact = Path(first["artifact"]) second_artifact = Path(second["artifact"]) self.assertEqual(first_artifact.read_bytes(), second_artifact.read_bytes()) self.assertEqual(first["services"], ["nodedc-backend"]) self.assertEqual(first["mcpVersion"], "0.11.0") self.assertEqual(first["tool"], "engine_list_l2_credential_refs") self.assertEqual(first["credentialScope"], "same-l1-workflow") self.assertFalse(first["crossL1Sharing"]) self.assertEqual(first["managedGrants"], "target-local") self.assertFalse(first["credentialValuesIncluded"]) extract = root / "extract" with tarfile.open(first_artifact, "r:gz") as archive: archive.extractall(extract, filter="data") entries = tuple( (extract / "files.txt").read_text(encoding="utf-8").splitlines() ) payload = extract / "payload" self.assertEqual( entries, RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_ARTIFACT_ENTRIES, ) self.assertTrue( RUNNER.is_engine_mcp_l1_credential_reuse_slice( "engine", entries, ) ) self.assertEqual( RUNNER.component_services("engine", entries), ("nodedc-backend",), ) RUNNER.validate_engine_mcp_l1_credential_reuse_slice( payload, entries, ) (root / "load").mkdir() manifest, loaded_entries, _ = RUNNER.load_artifact( first_artifact, root / "load", ) self.assertEqual(manifest["component"], "engine") self.assertEqual(tuple(loaded_entries), entries) def test_builder_rejects_source_drift(self): self.require_current_target_source() with tempfile.TemporaryDirectory( prefix="nodedc-engine-mcp-l1-credential-reuse-drift-" ) as directory: root = Path(directory) engine_copy = root / "engine" for relative_path in ( RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_ARTIFACT_ENTRIES ): source = ENGINE_ROOT / relative_path destination = engine_copy / relative_path destination.parent.mkdir(parents=True, exist_ok=True) destination.write_bytes(source.read_bytes()) route = engine_copy / "nodedc-source/server/routes/n8n.js" route.write_text( route.read_text(encoding="utf-8") + "\n// drift\n", encoding="utf-8", ) with self.assertRaises(subprocess.CalledProcessError): self.build(root / "artifact", engine_copy) def test_descriptor_closes_identity_and_cross_l1_boundaries(self): self.require_current_target_source() descriptor = json.loads( ( ENGINE_ROOT / RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_DESCRIPTOR_REL ).read_text(encoding="utf-8") ) self.assertEqual(descriptor["mcpVersion"], "0.11.0") self.assertEqual( descriptor["visibilityProof"]["scope"], "same-l1-workflow", ) self.assertEqual( descriptor["binding"]["applyOperation"], "assignCredentialRef", ) self.assertFalse(descriptor["crossL1Sharing"]) self.assertEqual(descriptor["managedGrants"], "target-local") self.assertIn( "credential-value", descriptor["binding"]["neverReturns"], ) def test_preflight_requires_exact_foundation_and_absent_descriptor(self): with tempfile.TemporaryDirectory( prefix="nodedc-engine-mcp-l1-credential-reuse-preflight-" ) as directory: root = Path(directory) hashes = {} for mapping in ( RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_PREDECESSOR_SHA256, RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_FOUNDATION_SHA256, ): for relative_path, expected in mapping.items(): path = root / relative_path path.parent.mkdir(parents=True, exist_ok=True) path.write_text("installed\n", encoding="utf-8") hashes[path] = expected with ( mock.patch.object(RUNNER, "component_root", return_value=root), mock.patch.object( RUNNER, "sha256_file", side_effect=lambda path: hashes[Path(path)], ), mock.patch.object( RUNNER, "preflight_engine_credential_backend_runtime", return_value={"mode": "verified-derived-retry"}, ), ): result = ( RUNNER .preflight_engine_mcp_l1_credential_reuse_predecessor() ) self.assertEqual( result["mode"], "normalized-identity-search-v1-to-l1-credential-reuse-v1", ) self.assertEqual( result["foundation_sha256"], RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_FOUNDATION_SHA256, ) def test_plan_renders_closed_same_l1_boundary(self): self.require_current_target_source() with tempfile.TemporaryDirectory( prefix="nodedc-engine-mcp-l1-credential-reuse-plan-" ) as directory: root = Path(directory) artifact = Path(self.build(root / "artifacts")["artifact"]) live_root = root / "live" live_root.mkdir() preflight = { "mode": "normalized-identity-search-v1-to-l1-credential-reuse-v1", "predecessor_sha256": dict( RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_PREDECESSOR_SHA256 ), "foundation_sha256": dict( RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_FOUNDATION_SHA256 ), "target_sha256": dict( RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_TARGET_SHA256 ), "new_paths": tuple( RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_NEW_PATHS ), "backend_mode": "verified-derived-retry", } output = io.StringIO() with ( mock.patch.object(RUNNER, "validate_artifact_location"), mock.patch.object(RUNNER, "ensure_layout"), mock.patch.object(RUNNER, "TMP_DIR", root), mock.patch.object(RUNNER, "component_root", return_value=live_root), mock.patch.object( RUNNER, "component_compose_root", return_value=live_root, ), mock.patch.object( RUNNER, "preflight_engine_mcp_l1_credential_reuse_predecessor", return_value=preflight, ), mock.patch.object( RUNNER, "preflight_engine_credential_backend_runtime", return_value={"mode": "verified-derived-retry"}, ), mock.patch.object(RUNNER, "state_has_sha", return_value=False), mock.patch.object(RUNNER, "state_has_patch_id", return_value=False), redirect_stdout(output), ): RUNNER.plan_artifact(artifact) plan = output.getvalue() self.assertIn("services=nodedc-backend", plan) self.assertIn( "engine_mcp_tool=engine_list_l2_credential_refs", plan, ) self.assertIn( "engine_mcp_apply_operation=assignCredentialRef", plan, ) self.assertIn("engine_mcp_credential_scope=same-l1-workflow", plan) self.assertIn("engine_mcp_cross_l1_sharing=no", plan) self.assertIn("engine_mcp_managed_grants=target-local", plan) self.assertIn("engine_mcp_credential_values_included=no", plan) self.assertIn("l2_graph=untouched", plan) self.assertIn("state=new", plan) def test_healthchecks_dispatch_l1_credential_reuse_acceptance(self): entries = RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_ARTIFACT_ENTRIES with tempfile.TemporaryDirectory( prefix="nodedc-engine-mcp-l1-credential-reuse-health-" ) as directory: root = Path(directory) all_hashes = { **RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_TARGET_SHA256, **RUNNER.ENGINE_MCP_L1_CREDENTIAL_REUSE_FOUNDATION_SHA256, } for relative_path in all_hashes: path = root / relative_path path.parent.mkdir(parents=True, exist_ok=True) path.write_text("target\n", encoding="utf-8") def target_hash(path): relative_path = Path(path).relative_to(root).as_posix() return all_hashes[relative_path] with ( mock.patch.object(RUNNER, "component_root", return_value=root), mock.patch.object(RUNNER, "sha256_file", side_effect=target_hash), mock.patch.object(RUNNER, "healthcheck_compose_service"), mock.patch.object(RUNNER, "component_healthchecks", return_value=()), mock.patch.object( RUNNER, "preflight_engine_credential_backend_runtime", return_value={"mode": "verified-derived-retry"}, ), mock.patch.object( RUNNER, "accept_engine_mcp_l1_credential_reuse_runtime", return_value={"live": "accepted"}, ) as acceptance, mock.patch.object(RUNNER, "healthcheck_container"), ): RUNNER.run_healthchecks( "engine", entries, ("nodedc-backend",), ) acceptance.assert_called_once_with() def test_live_acceptance_probes_only_safe_opaque_same_l1_scope(self): expected_live = ( "engine-mcp-l1-credential-reuse:" "0.11.0:same-l1:opaque-ref:v1" ) with ( mock.patch.object(RUNNER, "component_root", return_value=ENGINE_ROOT), mock.patch.object( RUNNER, "validate_engine_mcp_l1_credential_reuse_slice", ), mock.patch.object( RUNNER, "engine_backend_container_id", return_value="backend", ), mock.patch.object( RUNNER, "run_engine_backend_probe", return_value=expected_live, ) as backend_probe, ): result = RUNNER.accept_engine_mcp_l1_credential_reuse_runtime() self.assertEqual(result["live"], expected_live) probe_source = backend_probe.call_args.args[0][-1] self.assertIn("engineAgentCandidateScope", probe_source) self.assertIn("engineAgentCredentialMayReuseWithinL1", probe_source) self.assertIn("untrusted.example", probe_source) if __name__ == "__main__": unittest.main(verbosity=2)