import argparse import importlib.util import json from pathlib import Path import shutil import tempfile import unittest from unittest import mock MODULE_PATH = Path(__file__).with_name("l1_installer.py") SPEC = importlib.util.spec_from_file_location("l1_installer", MODULE_PATH) assert SPEC and SPEC.loader l1 = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(l1) ROOT = Path(__file__).resolve().parents[2] RELEASE = ROOT / "releases" / "io.nexuscrm.sfcrm" / "0.31.0" def arguments(temp: Path) -> argparse.Namespace: runtime = temp / "runtime.json" catalog = temp / "catalog.json" runtime.write_text(json.dumps({ "schema": "nexuscrm.runtime-host-compatibility/v1", "protocols": [{"protocolId": "nexuscrm.runtime.v1", "major": 1, "minor": 0}], "engineCapabilities": json.loads((RELEASE / "app-package.json").read_text())["canonical_manifest"]["required_engine_capabilities"], })) package = json.loads((RELEASE / "app-package.json").read_text()) publisher_key = json.loads((RELEASE / "publisher-public-key.json").read_text()) kinds = sorted({ (row["component_kind"], row["component_format_version"]) for row in package["canonical_metadata_bundle"]["components"] }) catalog.write_text(json.dumps({ "schema": "nexuscrm.catalog-host-compatibility/v1", "componentKinds": [ {"componentKind": kind, "componentFormatVersion": version} for kind, version in kinds ], })) return argparse.Namespace( release_dir=RELEASE, profile=RELEASE / "l1-install-profile.json", runtime_host_compatibility=runtime, catalog_host_compatibility=catalog, org_id="test-org", runtime_audience="nexuscrm-runtime-api", service_release_digest="sha256:" + "1" * 64, frontend_release_digest="sha256:" + "2" * 64, frontend_compatibility_receipt_digest="sha256:" + "3" * 64, idempotency_key="test-install", operation_kind="PREPARE", app_installation_id="", expected_catalog_revision=0, expected_security_epoch=0, schema_generation=1, engine_generation=1, publisher_valid_from="2026-01-01T00:00:00Z", publisher_valid_until="2036-01-01T00:00:00Z", trusted_publisher_key_digest=l1.canonical_digest(publisher_key), publication_command_id="019fd2b6-e515-77dd-a866-2e2c9c4a2aef", ) class L1InstallerTest(unittest.TestCase): def test_renders_direct_l1_authority_without_bom(self): with tempfile.TemporaryDirectory() as raw: config = l1.render_config(arguments(Path(raw)), verifier=lambda *_: None) self.assertEqual(config["metadataSourceDigest"], "sha256:a17220b0d225bc48a60c9d9eb5db7ad90f59f5e6018d4a7b91346d390861cd39") self.assertEqual(config["authorizedWorkloadIds"], ["integration-service", "sfcrm-app-service"]) self.assertEqual( {grant["capabilityId"] for grant in config["consumerGrants"]}, { "catalog.consumer.app-metadata-view.v1", "catalog.consumer.app-metadata-view.v3", "catalog.consumer.integration-binding-view.v1", }, ) self.assertNotIn("bom", json.dumps(config).lower()) self.assertEqual(config["integrationBindings"], []) def test_rejects_modified_install_policy(self): package = json.loads((RELEASE / "app-package.json").read_text()) profile = json.loads((RELEASE / "l1-install-profile.json").read_text()) profile["consumerGrants"] = profile["consumerGrants"][:1] with self.assertRaisesRegex(l1.InstallError, "approved policy"): l1.validate_release_closure(package, profile) def test_rejects_publisher_key_outside_deployment_trust(self): with tempfile.TemporaryDirectory() as raw: args = arguments(Path(raw)) args.trusted_publisher_key_digest = "sha256:" + "f" * 64 with self.assertRaisesRegex(l1.InstallError, "not approved"): l1.render_config(args, verifier=lambda *_: None) def test_render_uses_one_validated_profile_snapshot(self): with tempfile.TemporaryDirectory() as raw: temp = Path(raw) args = arguments(temp) profile_path = temp / "profile.json" profile_path.write_bytes((RELEASE / "l1-install-profile.json").read_bytes()) args.profile = profile_path snapshot = l1.load_json(profile_path, "L1 install profile") changed = dict(snapshot) changed["authorizedWorkloadIds"] = ["attacker-workload"] profile_path.write_text(json.dumps(changed)) config = l1.render_config( args, verifier=lambda *_: None, profile=snapshot, ) self.assertEqual( config["authorizedWorkloadIds"], ["integration-service", "sfcrm-app-service"], ) def test_required_engine_capability_is_not_ambient_authority(self): with tempfile.TemporaryDirectory() as raw: config = l1.render_config(arguments(Path(raw)), verifier=lambda *_: None) self.assertIn( "runtime.commands.accept.v1", config["hostCompatibility"]["engineCapabilities"], ) self.assertNotIn( "runtime.commands.accept.v1", {grant["capabilityId"] for grant in config["consumerGrants"]}, ) def test_receipt_is_bound_to_request(self): with tempfile.TemporaryDirectory() as raw: temp = Path(raw) config = l1.render_config(arguments(temp), verifier=lambda *_: None) receipt = temp / "receipt.json" receipt.write_text(json.dumps({ "orgId": "test-org", "appInstallationId": "019fd2b6-e515-77dd-a866-2e2c9c4a2aef", "appReleaseDigest": config["expectedAppReleaseDigest"], "serviceReleaseDigest": config["serviceReleaseDigest"], "metadataSourceDigest": config["metadataSourceDigest"], "workloadId": config["workloadId"], "catalogRevision": 1, "securityEpoch": 1, "schemaGeneration": 1, "engineGeneration": 1, "catalogSnapshotDigest": "sha256:" + "4" * 64, })) self.assertEqual(l1.validate_receipt(config, receipt)["catalogRevision"], 1) value = json.loads(receipt.read_text()) value["serviceReleaseDigest"] = "sha256:" + "5" * 64 receipt.write_text(json.dumps(value)) with self.assertRaisesRegex(l1.InstallError, "serviceReleaseDigest"): l1.validate_receipt(config, receipt) def test_upgrade_receipt_must_match_requested_installation(self): with tempfile.TemporaryDirectory() as raw: temp = Path(raw) args = arguments(temp) args.operation_kind = "UPGRADE" args.app_installation_id = "019fd2b6-e515-77dd-a866-2e2c9c4a2aef" args.expected_catalog_revision = 1 args.expected_security_epoch = 1 config = l1.render_config(args, verifier=lambda *_: None) receipt = temp / "receipt.json" receipt.write_text(json.dumps({ "orgId": "test-org", "appInstallationId": "019fd2b6-e515-7000-a866-2e2c9c4a2aef", "appReleaseDigest": config["expectedAppReleaseDigest"], "serviceReleaseDigest": config["serviceReleaseDigest"], "metadataSourceDigest": config["metadataSourceDigest"], "workloadId": config["workloadId"], "catalogRevision": 2, "securityEpoch": 2, "schemaGeneration": 1, "engineGeneration": 1, "catalogSnapshotDigest": "sha256:" + "4" * 64, })) with self.assertRaisesRegex(l1.InstallError, "appInstallationId"): l1.validate_receipt(config, receipt) def test_serving_config_is_derived_only_from_receipt_and_profile(self): with tempfile.TemporaryDirectory() as raw: temp = Path(raw) args = arguments(temp) config = l1.render_config(args, verifier=lambda *_: None) profile = json.loads((RELEASE / "l1-install-profile.json").read_text()) outputs = [] for index, selected in enumerate(profile["capabilities"], start=1): outputs.append({ "appCapabilityId": selected["appCapabilityId"], "runtimeCapabilityId": selected["runtimeCapabilityId"], "ownershipEpoch": 2, "compatibilityReceiptDigest": "sha256:" + f"{index:064x}", }) receipt = { "orgId": "test-org", "appInstallationId": "019fd2b6-e515-77dd-a866-2e2c9c4a2aef", "appReleaseDigest": config["expectedAppReleaseDigest"], "serviceReleaseDigest": config["serviceReleaseDigest"], "metadataSourceDigest": config["metadataSourceDigest"], "catalogRevision": 2, "securityEpoch": 2, "schemaGeneration": 1, "engineGeneration": 1, "catalogSnapshotDigest": "sha256:" + "4" * 64, "capabilities": outputs, } serving = l1.render_serving_config( profile, receipt, replica_id="local-l1", catalog_store_root="/var/lib/nexuscrm/catalog", ) self.assertEqual(serving["expected_service_release_digest"], config["serviceReleaseDigest"]) self.assertEqual(serving["authorized_workload_ids"], ["integration-service", "sfcrm-app-service"]) self.assertEqual(serving["catalog_consumer_capabilities"], []) self.assertEqual(serving["workload_capabilities"], [{ "runtime_capability_id": "runtime.identity.session-lifecycle.v1", "authorized_workload_id": "sfcrm-app-service", }]) def test_catalog_reconcile_image_must_be_published_and_immutable(self): good = ( "ghcr.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile@sha256:" + "a" * 64 ) self.assertEqual(l1.require_catalog_reconcile_image(good), good) for bad in ( "ghcr.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile:latest", "example.invalid/catalog-reconcile@sha256:" + "a" * 64, ): with self.assertRaisesRegex(l1.InstallError, "immutable digest"): l1.require_catalog_reconcile_image(bad) def test_setup_uses_the_same_database_environment(self): with tempfile.TemporaryDirectory() as raw: path = Path(raw) / "database.env" path.write_text( "TIDB_HOST=host.docker.internal\n" "TIDB_PORT=4406\n" "TIDB_USER=root\n" "TIDB_PASSWORD=secret\n" "TIDB_DATABASE=l1_test\n" "TIDB_TLS_DISABLE=1\n" ) environment = l1.setup_database_environment(path) self.assertEqual(environment["TIDB_HOST"], "127.0.0.1") self.assertEqual(environment["TIDB_DATABASE"], "l1_test") def test_pack_source_rebuilds_to_the_signed_app_release(self): l1.verify_pack_source(RELEASE, ROOT / "salesforce-crm-standard", "salesforce-crm-standard") def test_install_snapshots_every_later_consumed_path(self): with tempfile.TemporaryDirectory() as raw: temp = Path(raw) source = temp / "source" operation = temp / "operation" release = source / "release" pack = source / "pack" release.mkdir(parents=True) pack.mkdir(parents=True) (release / "app-package.json").write_text("release-before") profile = source / "profile.json" runtime = source / "runtime.json" catalog = source / "catalog.json" database = source / "database.env" setup = source / "setup" (pack / "pack.json").write_text("pack-before") for path, value in ( (profile, "profile-before"), (runtime, "runtime-before"), (catalog, "catalog-before"), (database, "database-before"), (setup, "setup-before"), ): path.write_text(value) args = argparse.Namespace( release_dir=release, profile=profile, runtime_host_compatibility=runtime, catalog_host_compatibility=catalog, database_env=database, setup_binary=setup, pack_source=str(pack), pack_name="pack", output_config=source / "request.json", output_receipt=source / "receipt.json", ) snapshot = l1.snapshot_install_inputs(args, operation) for path in (release / "app-package.json", profile, runtime, catalog, database, setup, pack / "pack.json"): path.write_text("changed-after-snapshot") self.assertEqual( "release-before", (snapshot.release_dir / "app-package.json").read_text(), ) self.assertEqual("profile-before", snapshot.profile.read_text()) self.assertEqual("runtime-before", snapshot.runtime_host_compatibility.read_text()) self.assertEqual("catalog-before", snapshot.catalog_host_compatibility.read_text()) self.assertEqual("database-before", snapshot.database_env.read_text()) self.assertEqual("setup-before", snapshot.setup_binary.read_text()) self.assertEqual("pack-before", (Path(snapshot.pack_source) / "pack.json").read_text()) self.assertEqual(operation / "catalog-request.json", snapshot.output_config) self.assertEqual( operation / "output" / "catalog-output.json", snapshot.output_receipt, ) def test_install_has_no_unverified_pack_skip(self): root = l1.parser() subcommands = next( action for action in root._actions if isinstance(action, argparse._SubParsersAction) ) install = subcommands.choices["install"] options = { option for action in install._actions for option in action.option_strings } self.assertNotIn("--skip-pack-install", options) def test_install_executes_only_from_the_private_snapshot(self): with tempfile.TemporaryDirectory() as raw: temp = Path(raw) release = temp / "release" pack = temp / "pack" shutil.copytree(RELEASE, release) shutil.copytree(ROOT / "salesforce-crm-standard", pack) args = arguments(temp) args.release_dir = release args.profile = release / "l1-install-profile.json" args.database_env = temp / "database.env" args.database_env.write_text( "TIDB_HOST=host.docker.internal\n" "TIDB_PORT=4000\n" "TIDB_USER=root\n" "TIDB_PASSWORD=\n" "TIDB_DATABASE=snapshot_test\n" "TIDB_TLS_DISABLE=1\n" ) args.setup_binary = temp / "setup" args.setup_binary.write_text("setup-before") args.pack_source = str(pack) args.pack_name = "salesforce-crm-standard" args.output_config = temp / "evidence" / "catalog-request.json" args.output_receipt = temp / "evidence" / "catalog-output.json" args.catalog_store = temp / "catalog" args.catalog_reconcile_image = ( "ghcr.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile@sha256:" + "a" * 64 ) args.docker_network = "host" args.replica_id = "snapshot-test" args.runtime_catalog_store_root = "/var/lib/nexuscrm/catalog" args.serving_config_output = temp / "evidence" / "serving.json" args.installation_receipt_output = temp / "evidence" / "installation.json" observed: list[tuple[str, ...]] = [] mutated = False def execute(command, *, env=None): nonlocal mutated command = tuple(command) observed.append(command) if not mutated: mutated = True args.database_env.write_text("TIDB_HOST=changed.invalid\n") args.profile.write_text("{}") (pack / "pack.json").write_text("{}") (release / "app-package.json").write_text("{}") args.setup_binary.write_text("changed") args.output_config.unlink() args.output_config.write_text("{}") if command[:2] != ("docker", "run"): return mounts = { item.split(",dst=", 1)[1].split(",", 1)[0]: Path(item.split("src=", 1)[1].split(",dst=", 1)[0]) for item in command if item.startswith("type=bind,src=") } config = json.loads(mounts["/etc/nexuscrm/catalog-config.json"].read_text()) output = mounts["/output"] / "catalog-output.json" output.write_text(json.dumps({ "orgId": config["orgId"], "appInstallationId": "019fd2b6-e515-77dd-a866-2e2c9c4a2aef", "appReleaseDigest": config["expectedAppReleaseDigest"], "serviceReleaseDigest": config["serviceReleaseDigest"], "frontendReleaseDigest": config["frontend"]["releaseDigest"], "metadataSourceDigest": config["metadataSourceDigest"], "workloadId": config["workloadId"], "catalogRevision": config["expectedCatalogRevision"] + 1, "securityEpoch": config["expectedSecurityEpoch"] + 1, "schemaGeneration": config["schemaGeneration"], "engineGeneration": config["engineGeneration"], "catalogSnapshotDigest": "sha256:" + "4" * 64, "capabilities": [{ "appCapabilityId": selection["appCapabilityId"], "runtimeCapabilityId": selection["runtimeCapabilityId"], "ownershipEpoch": 1, "compatibilityReceiptDigest": "sha256:" + f"{index:064x}", } for index, selection in enumerate(config["capabilities"], start=1)], })) with mock.patch.object(l1, "run_checked", side_effect=execute): result = l1.apply(args) setup_commands = [command for command in observed if command[:2] != ("docker", "run")] self.assertEqual(3, len(setup_commands)) self.assertTrue(all(command[0] != str(args.setup_binary) for command in setup_commands)) docker = next(command for command in observed if command[:2] == ("docker", "run")) self.assertNotIn(str(args.database_env), docker) self.assertFalse(any(str(args.release_dir) in value for value in docker)) self.assertFalse(any(str(args.output_config) in value for value in docker)) self.assertEqual( ["integration-service", "sfcrm-app-service"], json.loads(args.output_config.read_text())["authorizedWorkloadIds"], ) self.assertEqual( "019fd2b6-e515-77dd-a866-2e2c9c4a2aef", result["appInstallationId"], ) if __name__ == "__main__": unittest.main()