261 lines
12 KiB
Python
261 lines
12 KiB
Python
#!/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 = SCRIPT_DIR / "nodedc-deploy"
|
|
BUILDER_PATH = SCRIPT_DIR / "build-n8n-private-extension-artifact.mjs"
|
|
PATCH_ID = "n8n-nodes-ndc-release-20260716-003"
|
|
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",
|
|
]
|
|
EXPECTED_CREDENTIAL_TYPES = [
|
|
"ndcDataProductWriterApi",
|
|
"ndcDataProductReaderApi",
|
|
"ndcFoundryBindingApi",
|
|
]
|
|
|
|
|
|
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\.2-[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.2")
|
|
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.2")
|
|
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)
|