#!/usr/bin/env python3 import hashlib import importlib.machinery import importlib.util import io import json import os import subprocess import sys import tarfile import tempfile import unittest from pathlib import Path SCRIPT_DIR = Path(__file__).resolve().parent PLATFORM_ROOT = SCRIPT_DIR.parent.parent RUNNER_PATH = Path( os.environ.get("NODEDC_DEPLOY_RUNNER_UNDER_TEST", SCRIPT_DIR / "nodedc-deploy") ).resolve() BUILDER_PATH = SCRIPT_DIR / "build-n8n-private-extension-artifact.mjs" PATCH_ID = "n8n-nodes-ndc-provider-rotating-access-20260718-006" EXPECTED_NODES = [ "dist/nodes/NdcDataProductPublish/NdcDataProductPublish.node.js", "dist/nodes/NdcDataProductRead/NdcDataProductRead.node.js", "dist/nodes/NdcFoundryBinding/NdcFoundryBinding.node.js", ] EXPECTED_NODE_TYPES = [ "n8n-nodes-ndc.ndcDataProductPublish", "n8n-nodes-ndc.ndcDataProductRead", "n8n-nodes-ndc.ndcFoundryBinding", ] EXPECTED_CREDENTIALS = [ "dist/credentials/NdcDataProductWriterApi.credentials.js", "dist/credentials/NdcDataProductReaderApi.credentials.js", "dist/credentials/NdcFoundryBindingApi.credentials.js", "dist/credentials/NdcProviderRotatingAccessApi.credentials.js", ] EXPECTED_CREDENTIAL_TYPES = [ "ndcDataProductWriterApi", "ndcDataProductReaderApi", "ndcFoundryBindingApi", "ndcProviderRotatingAccessApi", ] def load_runner(): loader = importlib.machinery.SourceFileLoader("nodedc_deploy_under_test", 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 N8nPrivateExtensionPolicyTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.temporary = tempfile.TemporaryDirectory(prefix="nodedc-extension-policy-") cls.root = Path(cls.temporary.name) cls.artifacts = [] cls.results = [] for index in range(2): output_dir = cls.root / f"build-{index}" output_dir.mkdir() environment = os.environ.copy() environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(output_dir) command = ["node", str(BUILDER_PATH)] if index == 0: command.append(PATCH_ID) result = subprocess.run( command, cwd=PLATFORM_ROOT, env=environment, check=True, capture_output=True, text=True, ) metadata = json.loads(result.stdout) artifact = Path(metadata["artifact"]) cls.results.append(metadata) cls.artifacts.append(artifact) @classmethod def tearDownClass(cls): cls.temporary.cleanup() def test_runner_source_compiles(self): compile(RUNNER_PATH.read_text(encoding="utf-8"), str(RUNNER_PATH), "exec") def test_builder_is_byte_reproducible(self): first = self.artifacts[0].read_bytes() second = self.artifacts[1].read_bytes() self.assertEqual(first, second) self.assertEqual(self.results[0]["artifactSha256"], self.results[1]["artifactSha256"]) self.assertEqual(self.results[0]["patchId"], PATCH_ID) self.assertEqual(self.results[0]["nodeTypes"], EXPECTED_NODE_TYPES) self.assertEqual(self.results[0]["credentialTypes"], EXPECTED_CREDENTIAL_TYPES) self.assertEqual(first[4:8], b"\0\0\0\0", "gzip mtime must be zero") self.assertEqual(first[3] & 0x08, 0, "gzip header must not carry a host filename") inner = self._inner_package_bytes() self.assertEqual(inner[4:8], b"\0\0\0\0", "inner package gzip mtime must be zero") self.assertEqual(inner[3] & 0x08, 0, "inner package gzip header must not carry a host filename") def test_positive_release_passes_runner_policy(self): with tempfile.TemporaryDirectory(prefix="nodedc-extension-load-") as directory: manifest, entries, _payload = RUNNER.load_artifact(self.artifacts[0], Path(directory)) self.assertEqual(manifest["component"], "n8n-private-extension") self.assertEqual(len(entries), 1) self.assertRegex(entries[0], r"^releases/n8n-nodes-ndc/0\.1\.4-[a-f0-9]{16}$") with tarfile.open(self.artifacts[0], "r:gz") as archive: names = archive.getnames() self.assertFalse(any(Path(name).name.startswith("._") for name in names)) self.assertFalse(any(".." in Path(name).parts for name in names)) def test_release_has_exactly_three_non_tool_runtime_types(self): package = self._inner_package_bytes() with tarfile.open(fileobj=io.BytesIO(package), mode="r:gz") as archive: package_json_member = archive.getmember("package/package.json") package_json = json.loads(archive.extractfile(package_json_member).read()) self.assertEqual(package_json["version"], "0.1.4") self.assertEqual(package_json["n8n"]["nodes"], EXPECTED_NODES) self.assertEqual(package_json["n8n"]["credentials"], EXPECTED_CREDENTIALS) for node_path in EXPECTED_NODES: source = archive.extractfile(archive.getmember(f"package/{node_path}")).read().decode("utf-8") self.assertNotRegex(source, r"\busableAsTool\b") def test_inner_package_stream_and_members_are_canonical(self): package = self._inner_package_bytes() self.assertEqual(package[4:8], b"\0\0\0\0") self.assertEqual(package[3] & 0x08, 0) with tarfile.open(fileobj=io.BytesIO(package), mode="r:gz") as archive: members = archive.getmembers() self.assertEqual([member.name for member in members], sorted(member.name for member in members)) for member in members: self.assertEqual((member.uid, member.gid, member.uname, member.gname), (0, 0, "root", "root")) self.assertEqual(member.mtime, 0) self.assertEqual(member.mode, 0o755 if member.isdir() else 0o644) def test_v2_first_activation_uses_verified_inactive_baseline(self): with tempfile.TemporaryDirectory(prefix="nodedc-extension-v2-") as directory: work = Path(directory) RUNNER.safe_extract(self.artifacts[0], work) entries = RUNNER.parse_files_list(work / "files.txt") release_dir = work / "payload" / entries[0] release_path = release_dir / "release.json" rollback_path = release_dir / "rollback.json" release = json.loads(release_path.read_text(encoding="utf-8")) rollback = json.loads(rollback_path.read_text(encoding="utf-8")) expected_policy = { "allowed": ["previous_verified_immutable_release", "verified_inactive"], "firstActivation": "verified_inactive", "requiresPreActivationVerification": True, } self.assertEqual(release["schemaVersion"], "nodedc.n8n-private-extension-release/v2") self.assertEqual(release["package"]["version"], "0.1.4") self.assertEqual(release["activation"]["rollbackBaselinePolicy"], expected_policy) self.assertEqual(rollback["schemaVersion"], "nodedc.n8n-private-extension-rollback/v2") self.assertEqual(rollback["baselinePolicy"], expected_policy) rollback["baselinePolicy"]["requiresPreActivationVerification"] = False rollback_path.write_text(json.dumps(rollback), encoding="utf-8") with self.assertRaisesRegex(RUNNER.DeployError, "rollback manifest mismatch"): RUNNER.validate_n8n_private_extension_release(work / "payload", entries) def test_outer_appledouble_and_traversal_are_rejected(self): with self.assertRaisesRegex(RUNNER.DeployError, "unexpected tar member"): RUNNER.validate_tar_member(tarfile.TarInfo("._manifest.env")) with self.assertRaisesRegex(RUNNER.DeployError, "path escape rejected"): RUNNER.validate_tar_member(tarfile.TarInfo("payload/../escape")) def test_extra_custom_node_is_rejected(self): def mutate(package_json): package_json["n8n"]["nodes"].append("dist/nodes/Extra/Extra.node.js") package_path, release = self._tampered_package( mutate, {"package/dist/nodes/Extra/Extra.node.js": b"module.exports = {};\n"}, ) with self.assertRaisesRegex(RUNNER.DeployError, "n8n.nodes registration mismatch"): RUNNER.validate_n8n_package_tarball(package_path, release) def test_lifecycle_script_is_rejected(self): def mutate(package_json): package_json.setdefault("scripts", {})["install"] = "echo forbidden" package_path, release = self._tampered_package(mutate) with self.assertRaisesRegex(RUNNER.DeployError, "lifecycle script is forbidden"): RUNNER.validate_n8n_package_tarball(package_path, release) def test_inner_appledouble_is_rejected(self): package_path, release = self._tampered_package( lambda _package_json: None, {"package/dist/._NdcDataProductPublish.node.js": b"forbidden\n"}, ) with self.assertRaisesRegex(RUNNER.DeployError, "hidden member rejected"): RUNNER.validate_n8n_package_tarball(package_path, release) def test_tool_variant_marker_is_rejected_by_runner(self): target = EXPECTED_NODES[0] package_path, release = self._tampered_package( lambda _package_json: None, file_mutations={target: lambda source: source + b"\n// usableAsTool: true\n"}, ) with self.assertRaisesRegex(RUNNER.DeployError, "would generate a tool variant"): RUNNER.validate_n8n_package_tarball(package_path, release) def _tampered_package(self, mutate, extra_files=None, file_mutations=None): package_bytes = self._inner_package_bytes() members = [] with tarfile.open(fileobj=io.BytesIO(package_bytes), mode="r:gz") as archive: for member in archive: if not member.isfile(): continue source = archive.extractfile(member) members.append((member.name, source.read())) rewritten = [] for name, content in members: if name == "package/package.json": package_json = json.loads(content) mutate(package_json) content = (json.dumps(package_json, sort_keys=True) + "\n").encode("utf-8") package_rel = name.removeprefix("package/") if package_rel in (file_mutations or {}): content = file_mutations[package_rel](content) rewritten.append((name, content)) rewritten.extend((extra_files or {}).items()) target = self.root / f"tampered-{len(list(self.root.glob('tampered-*.tgz')))}.tgz" with tarfile.open(target, "w:gz", format=tarfile.PAX_FORMAT) as archive: for name, content in rewritten: member = tarfile.TarInfo(name) member.mode = 0o644 member.mtime = 0 member.size = len(content) archive.addfile(member, io.BytesIO(content)) with tarfile.open(target, "r:gz") as archive: package_member = next(member for member in archive if member.name == "package/package.json") version = json.loads(archive.extractfile(package_member).read())["version"] release = { "package": { "version": version, "sha256": hashlib.sha256(target.read_bytes()).hexdigest(), "bytes": target.stat().st_size, } } return target, release def _inner_package_bytes(self): with tarfile.open(self.artifacts[0], "r:gz") as archive: member = next(item for item in archive if item.name.endswith("/package.tgz")) source = archive.extractfile(member) return source.read() if __name__ == "__main__": unittest.main(verbosity=2)