SILVER/NODEDC_DEVICE_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

chore(deploy): add migration invariant audit

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-22 22:11:35 +03:00
parent 79140ee7bf
commit 71da1dd42a
6 changed files with 411 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deployment/device-control-core-migration-replay-audit-v1.json
+29
View File
@@ -0,0 +1,29 @@
{
"schemaVersion": "nodedc.device-plane.device-control-core-migration-replay-audit.v1",
"mode": "rejected-recovery-044-live-invariants-read-only-audit",
"allowedOperation": "canonical-plan-only",
"applyAllowed": false,
"rejectedRecovery": {
"patchId": "device-control-core-migration-replay-recovery-20260822-044",
"artifactSha256": "b893d8c90f98943797d32f486d6477d58a3be69eb1291e28c4a4bbd2e96774b7"
},
"readOnlyEvidence": [
"invalid-command-kind-count",
"triggering-receipt-count",
"constraint-validated",
"constraint-covers-final-command-kinds",
"host-telemetry-table-absent"
],
"runtimeMutation": "none",
"sourceMutation": "none",
"databaseMutation": "none",
"networkMutation": "none",
"secretRead": "none",
"preservedServices": [
"device-control-core",
"device-manager",
"device-gateway",
"device-postgres",
"device-backhaul-target"
]
}
infra/deploy-runner/build-device-control-core-migration-replay-audit-artifact.mjs
+95
View File
@@ -0,0 +1,95 @@
#!/usr/bin/env node
import { createHash } from "node:crypto";
import { spawnSync } from "node:child_process";
import { cp, mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { dirname, join, resolve } from "node:path";
import { fileURLToPath } from "node:url";
const scriptDir = dirname(fileURLToPath(import.meta.url));
const sourceRoot = resolve(scriptDir, "../..");
const artifactDir = resolve(
process.env.NODEDC_DEPLOY_ARTIFACT_DIR
|| resolve(scriptDir, "../deploy-artifacts"),
);
const [
patchId = "device-control-core-migration-replay-audit-20260822-045",
...extra
] = process.argv.slice(2);
if (extra.length || !/^[A-Za-z0-9._-]{1,96}$/.test(patchId)) {
throw new Error(
"usage: build-device-control-core-migration-replay-audit-artifact.mjs [patch-id]",
);
}
const entry = "deployment/device-control-core-migration-replay-audit-v1.json";
const stage = await mkdtemp(join(tmpdir(), "nodedc-control-core-migration-audit-"));
const payload = join(stage, "payload");
const target = resolve(artifactDir, `nodedc-device-plane-${patchId}.tgz`);
try {
const descriptor = JSON.parse(await readFile(resolve(sourceRoot, entry), "utf8"));
if (
descriptor.schemaVersion
!== "nodedc.device-plane.device-control-core-migration-replay-audit.v1"
|| descriptor.mode
!== "rejected-recovery-044-live-invariants-read-only-audit"
|| descriptor.allowedOperation !== "canonical-plan-only"
|| descriptor.applyAllowed !== false
|| descriptor.runtimeMutation !== "none"
|| descriptor.sourceMutation !== "none"
|| descriptor.databaseMutation !== "none"
|| descriptor.networkMutation !== "none"
|| descriptor.secretRead !== "none"
) {
throw new Error("device_control_core_migration_replay_audit_descriptor_mismatch");
}
await mkdir(dirname(join(payload, entry)), { recursive: true });
await cp(resolve(sourceRoot, entry), join(payload, entry), { force: true });
await writeFile(
join(stage, "manifest.env"),
`id=${patchId}\ncomponent=device-plane\ntype=app-overlay\n`,
"utf8",
);
await writeFile(join(stage, "files.txt"), `${entry}\n`, "utf8");
await mkdir(artifactDir, { recursive: true });
const tar = spawnSync(
"python3",
["-c", canonicalTarScript(), target, stage],
{ encoding: "utf8", maxBuffer: 16 * 1024 * 1024 },
);
if (tar.status !== 0) throw new Error(`tar_failed:${tar.stderr || tar.stdout}`);
const sha256 = createHash("sha256")
.update(await readFile(target))
.digest("hex");
console.log(JSON.stringify({
ok: true,
patchId,
component: "device-plane",
artifact: target,
sha256,
entries: [entry],
build: [],
services: [],
allowedOperation: descriptor.allowedOperation,
runtimeMutation: descriptor.runtimeMutation,
}, null, 2));
} finally {
await rm(stage, { recursive: true, force: true });
}
function canonicalTarScript() {
return [
"import gzip,io,pathlib,sys,tarfile",
"root=pathlib.Path(sys.argv[2])",
"with open(sys.argv[1],'wb') as out:",
" with gzip.GzipFile(filename='',mode='wb',fileobj=out,compresslevel=9,mtime=0) as gz:",
" with tarfile.open(fileobj=gz,mode='w',format=tarfile.PAX_FORMAT) as tar:",
" for top in ('manifest.env','files.txt','payload'):",
" p=root/top; paths=[p]+(sorted(p.rglob('*')) if p.is_dir() else [])",
" for x in paths:",
" info=tar.gettarinfo(str(x),arcname=x.relative_to(root).as_posix()); info.uid=info.gid=0; info.uname=info.gname='root'; info.mtime=0; info.mode=0o755 if info.isdir() else 0o644",
" with (open(x,'rb') if info.isfile() else io.BytesIO()) as src: tar.addfile(info,src if info.isfile() else None)",
].join("\n");
}
infra/deploy-runner/build-device-control-core-release-artifact.mjs
+12
View File
@@ -37,6 +37,18 @@ if (
const isV3 = patchId.startsWith("device-control-core-release-v3-");
const isV2 = patchId.startsWith("device-control-core-release-v2-");
const coreDockerfile = await readFile(
resolve(devicePlaneRoot, "services/device-control-core/Dockerfile"),
"utf8",
);
if (
!isV3
&& coreDockerfile.includes(
"COPY packages/infrastructure-telemetry-contract ./packages/infrastructure-telemetry-contract",
)
) {
throw new Error("historical_device_control_core_builder_has_advanced");
}
const expectedV2Predecessor = Object.freeze({
patchId: predecessorPatchId ?? "device-control-core-release-20260812-024",
artifactSha256: predecessorSha256 ?? "a289e909283109642e6bba3d9822a31f63423cfe0bbcd52705979681bd2bc793",
infra/deploy-runner/build-device-edge-core-channel-bootstrap-artifact.mjs
+11
View File
@@ -19,6 +19,17 @@ const upgradeV4 = patchId.startsWith("device-edge-core-channel-upgrade-v4-");
const upgradeV2 = !upgradeV4 && patchId.startsWith("device-edge-core-channel-upgrade-v2-");
const upgradeV1 = !upgradeV4 && !upgradeV2 && patchId.startsWith("device-edge-core-channel-upgrade-");
const upgrade = upgradeV1 || upgradeV2 || upgradeV4;
const coreDockerfile = await readFile(
resolve(devicePlaneRoot, "services/device-control-core/Dockerfile"),
"utf8",
);
if (
coreDockerfile.includes(
"COPY packages/infrastructure-telemetry-contract ./packages/infrastructure-telemetry-contract",
)
) {
throw new Error("historical_device_edge_core_channel_builder_has_advanced");
}
const descriptorPath = upgradeV4
? "deployment/device-edge-core-channel-upgrade-v4.json"
: upgradeV2
infra/deploy-runner/test_device_control_core_migration_replay_audit_artifact.py
+186
View File
@@ -0,0 +1,186 @@
#!/usr/bin/env python3
import contextlib
import hashlib
import importlib.machinery
import importlib.util
import inspect
import io
import json
import os
import re
import subprocess
import tarfile
import tempfile
import unittest
from pathlib import Path
from unittest import mock
SCRIPT_DIR = Path(__file__).resolve().parent
RUNNER_PATH = Path(
os.environ.get(
"NODEDC_DEPLOY_RUNNER",
SCRIPT_DIR.parents[2] / "platform/infra/deploy-runner/nodedc-deploy",
)
)
BUILDER = (
SCRIPT_DIR
/ "build-device-control-core-migration-replay-audit-artifact.mjs"
)
def load_runner():
loader = importlib.machinery.SourceFileLoader(
"nodedc_control_core_migration_replay_audit_runner_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 DeviceControlCoreMigrationReplayAuditArtifactTest(unittest.TestCase):
def build(self, artifact_dir, patch_id):
environment = os.environ.copy()
environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir)
result = subprocess.run(
["node", str(BUILDER), patch_id],
check=True,
capture_output=True,
text=True,
env=environment,
)
return json.loads(result.stdout)
def test_artifact_is_deterministic_plan_only_audit(self):
with tempfile.TemporaryDirectory(
prefix="nodedc-control-core-migration-replay-audit-",
) as directory:
artifact_dir = Path(directory)
patch_id = "device-control-core-migration-replay-audit-unit-045"
first = self.build(artifact_dir, patch_id)
first_bytes = Path(first["artifact"]).read_bytes()
second = self.build(artifact_dir, patch_id)
second_bytes = Path(second["artifact"]).read_bytes()
self.assertEqual(first_bytes, second_bytes)
self.assertEqual(
first["sha256"],
hashlib.sha256(first_bytes).hexdigest(),
)
self.assertEqual(first["build"], [])
self.assertEqual(first["services"], [])
self.assertEqual(first["allowedOperation"], "canonical-plan-only")
self.assertEqual(first["runtimeMutation"], "none")
with tarfile.open(first["artifact"], "r:gz") as archive:
descriptor = json.loads(
archive.extractfile(
"payload/"
+ RUNNER.DEVICE_PLANE_CONTROL_CORE_MIGRATION_REPLAY_AUDIT_REL
).read().decode("utf-8")
)
self.assertEqual(
descriptor,
RUNNER.expected_device_plane_control_core_migration_replay_audit_descriptor(),
)
entries = RUNNER.DEVICE_PLANE_CONTROL_CORE_MIGRATION_REPLAY_AUDIT_ENTRIES
self.assertEqual(RUNNER.component_services("device-plane", entries), ())
self.assertEqual(RUNNER.component_builds("device-plane", entries), ())
def test_database_audit_returns_all_mismatches_without_mutation(self):
database_result = mock.Mock(
returncode=0,
stdout="2\t0\tfalse\tfalse\tfalse\n",
stderr="",
)
with mock.patch.object(
RUNNER,
"device_plane_service_container_ids",
return_value=("p" * 64,),
), mock.patch.object(
RUNNER.subprocess,
"run",
return_value=database_result,
) as run:
evidence = (
RUNNER.collect_device_plane_control_core_migration_replay_database_evidence(
enforce_recovery_invariants=False,
)
)
self.assertEqual(evidence["invalidCommandKindCount"], 2)
self.assertEqual(evidence["triggeringReceiptCount"], 0)
self.assertFalse(evidence["constraintValidated"])
self.assertFalse(evidence["constraintCoversFinalKinds"])
self.assertFalse(evidence["hostTelemetryTableAbsent"])
self.assertFalse(evidence["recovery044Ready"])
command = run.call_args.args[0]
query = command[command.index("-c") + 1]
self.assertTrue(query.lstrip().lower().startswith("with constraint_state"))
self.assertNotIn(";", query)
self.assertIsNone(
re.search(
r"(?im)^\s*(?:delete|update|insert|truncate|alter|drop|create)\b",
query,
)
)
def test_recovery_rejection_prints_five_bounded_values(self):
database_result = mock.Mock(
returncode=0,
stdout="0\t0\ttrue\ttrue\ttrue\n",
stderr="",
)
output = io.StringIO()
with mock.patch.object(
RUNNER,
"device_plane_service_container_ids",
return_value=("p" * 64,),
), mock.patch.object(
RUNNER.subprocess,
"run",
return_value=database_result,
), contextlib.redirect_stdout(output), self.assertRaises(
RUNNER.DeployError
):
RUNNER.collect_device_plane_control_core_migration_replay_database_evidence()
rendered = output.getvalue()
for marker in (
"invalid_command_kind_count=0",
"triggering_receipt_count=0",
"constraint_validated=true",
"constraint_covers_final_kinds=true",
"host_telemetry_table_absent=true",
):
self.assertIn(marker, rendered)
def test_plan_and_apply_paths_preserve_plan_only_boundary(self):
plan_source = inspect.getsource(RUNNER.plan_artifact)
apply_source = inspect.getsource(RUNNER.apply_artifact)
self.assertIn(
"device_plane_control_core_migration_replay_audit_preflight",
plan_source,
)
self.assertIn(
"emit_device_plane_control_core_migration_replay_database_evidence",
plan_source,
)
self.assertIn(
"is_device_plane_control_core_migration_replay_audit_slice",
apply_source,
)
self.assertLess(
apply_source.index(
"is_device_plane_control_core_migration_replay_audit_slice"
),
apply_source.index("state_has_sha"),
)
if __name__ == "__main__":
unittest.main()
infra/deploy-runner/test_device_manager_control_plane_artifacts.py
+78 -31
View File
@@ -144,6 +144,16 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
== RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V1_COMPOSE_SHA256
)
def historical_control_core_builders_are_current(self):
dockerfile = (
DEVICE_CORE_ROOT / "services/device-control-core/Dockerfile"
).read_text(encoding="utf-8")
return (
"COPY packages/infrastructure-telemetry-contract "
"./packages/infrastructure-telemetry-contract"
not in dockerfile
)
def build(self, script, patch_id, artifact_dir):
environment = os.environ.copy()
environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir)
@@ -741,7 +751,40 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
completed.stderr,
)
def test_historical_core_builders_fail_closed_after_telemetry_dockerfile(self):
if self.historical_control_core_builders_are_current():
self.skipTest("historical Core Dockerfile is still current")
cases = (
(
"build-device-control-core-release-artifact.mjs",
"device-control-core-release-rebuild-forbidden-001",
"historical_device_control_core_builder_has_advanced",
),
(
"build-device-edge-core-channel-bootstrap-artifact.mjs",
"device-edge-core-channel-bootstrap-rebuild-forbidden-001",
"historical_device_edge_core_channel_builder_has_advanced",
),
)
for script, patch_id, expected_error in cases:
with self.subTest(script=script), tempfile.TemporaryDirectory(
prefix="nodedc-device-core-historical-reject-",
) as directory:
environment = os.environ.copy()
environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = directory
completed = subprocess.run(
["node", str(SCRIPT_DIR / script), patch_id],
cwd=DEVICE_CORE_ROOT,
env=environment,
capture_output=True,
text=True,
)
self.assertNotEqual(completed.returncode, 0)
self.assertIn(expected_error, completed.stderr)
def test_edge_core_channel_bootstrap_is_core_only_and_secret_free(self):
if not self.historical_control_core_builders_are_current():
self.skipTest("historical Core channel builder generation is frozen")
manifest, entries, names, result = self.assert_deterministic_artifact(
"build-device-edge-core-channel-bootstrap-artifact.mjs",
"device-edge-core-channel-bootstrap-unit-001",
@@ -778,6 +821,8 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
self.assertEqual(checks[0]["expected_json"]["commandTransport"], "disabled")
def test_edge_core_channel_upgrade_is_core_only_and_pins_bootstrap_018(self):
if not self.historical_control_core_builders_are_current():
self.skipTest("historical Core channel builder generation is frozen")
manifest, entries, names, result = self.assert_deterministic_artifact(
"build-device-edge-core-channel-bootstrap-artifact.mjs",
"device-edge-core-channel-upgrade-unit-001",
@@ -807,6 +852,8 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
)
def test_edge_core_channel_upgrade_v2_is_core_only_and_pins_upgrade_019(self):
if not self.historical_control_core_builders_are_current():
self.skipTest("historical Core channel builder generation is frozen")
patch_id = "device-edge-core-channel-upgrade-v2-unit-001"
manifest, entries, names, result = self.assert_deterministic_artifact(
"build-device-edge-core-channel-bootstrap-artifact.mjs",
@@ -853,6 +900,8 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
)
def test_edge_core_channel_upgrade_v4_is_core_only_and_pins_upgrade_021(self):
if not self.historical_control_core_builders_are_current():
self.skipTest("historical Core channel builder generation is frozen")
patch_id = "device-edge-core-channel-upgrade-v4-unit-001"
manifest, entries, names, result = self.assert_deterministic_artifact(
"build-device-edge-core-channel-bootstrap-artifact.mjs",
@@ -944,6 +993,8 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
)
def test_control_core_release_is_repeatable_core_only_and_compose_free(self):
if not self.historical_control_core_builders_are_current():
self.skipTest("historical Core release v1 builder generation is frozen")
patch_id = "device-control-core-release-unit-001"
manifest, entries, names, result = self.assert_deterministic_artifact(
"build-device-control-core-release-artifact.mjs",
@@ -1019,6 +1070,8 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
)
def test_control_core_release_v2_is_typed_core_only(self):
if not self.historical_control_core_builders_are_current():
self.skipTest("historical Core release v2 builder generation is frozen")
patch_id = "device-control-core-release-v2-unit-001"
manifest, entries, names, result = self.assert_deterministic_artifact(
"build-device-control-core-release-artifact.mjs",
@@ -1055,6 +1108,8 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
self.assertEqual(descriptor["gelios"], "untouched-legacy-only")
def test_control_core_release_builder_supports_release_predecessor(self):
if not self.historical_control_core_builders_are_current():
self.skipTest("historical Core release v1 builder generation is frozen")
with tempfile.TemporaryDirectory(
prefix="nodedc-control-core-successor-",
) as directory:
@@ -1111,6 +1166,8 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
)
def test_control_core_release_v2_builder_supports_v2_release_predecessor(self):
if not self.historical_control_core_builders_are_current():
self.skipTest("historical Core release v2 builder generation is frozen")
with tempfile.TemporaryDirectory(
prefix="nodedc-control-core-v2-successor-",
) as directory:
@@ -2583,7 +2640,6 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
)),
encoding="utf-8",
)
restored_core_id = "f" * 64
with (
mock.patch.object(
RUNNER,
@@ -2596,29 +2652,20 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
),
mock.patch.object(
RUNNER,
"run_component_runtime",
"retag_device_plane_control_core_image",
) as retag_core_image,
mock.patch.object(
RUNNER,
"prepare_component_runtime",
) as prepare_runtime,
mock.patch.object(
RUNNER,
"run_compose",
) as restore_runtime,
mock.patch.object(
RUNNER,
"healthcheck_compose_service_with_grace",
) as restore_health,
mock.patch.object(
RUNNER,
"compose_service_container_id",
return_value=restored_core_id,
),
mock.patch.object(
RUNNER,
"inspect_device_plane_container",
return_value={
"Id": restored_core_id,
"State": {
"Status": "running",
"Running": True,
"Health": {"Status": "healthy"},
},
},
),
"accept_device_plane_control_core_rollback_runtime",
) as rollback_acceptance,
mock.patch.object(
RUNNER,
"validate_device_manager_control_plane_runtime",
@@ -2634,20 +2681,20 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
)
stop.assert_not_called()
restore_runtime.assert_called_once_with(
retag_core_image.assert_called_once_with(
"sha256:" + "a" * 64,
"Device Control Core exact pre-apply rollback image",
)
prepare_runtime.assert_called_once_with(
"device-plane",
existing,
)
restore_runtime.assert_called_once_with(
"device-plane",
("device-control-core",),
existing,
)
self.assertEqual(
[call.args for call in restore_health.call_args_list],
[
("device-plane", "device-manager"),
("device-plane", "device-gateway"),
("device-plane", "device-postgres"),
("device-plane", "device-backhaul-target"),
],
)
rollback_acceptance.assert_called_once()
runtime_acceptance.assert_called_once_with(
require_edge_channel=True,
core_network_mode="private-egress",