From 54140f816abb54c3de485c02776d201986ea1695 Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Tue, 5 May 2026 21:13:25 -0400 Subject: [PATCH 1/7] feat(immutable-node): add host capability projection fixture --- ...pabilityplacement.sourceos-supervisor.json | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) create mode 100644 fixtures/sourceos-spec/hostcapabilityplacement.sourceos-supervisor.json diff --git a/fixtures/sourceos-spec/hostcapabilityplacement.sourceos-supervisor.json b/fixtures/sourceos-spec/hostcapabilityplacement.sourceos-supervisor.json new file mode 100644 index 0000000..75febae --- /dev/null +++ b/fixtures/sourceos-spec/hostcapabilityplacement.sourceos-supervisor.json @@ -0,0 +1,21 @@ +{ + "id": "urn:srcos:host-capability-placement:sourceos-supervisor", + "type": "HostCapabilityPlacement", + "specVersion": "2.2.0", + "name": "SourceOS host supervisor", + "placementClass": "image-baked-host-capability", + "authority": "sourceos-base", + "primaryPlane": "node-substrate", + "mandatoryForBaseNode": true, + "requiresEnrollment": false, + "lifecycleCoupling": "host-release", + "pathHints": [ + "/usr/libexec/sourceos/supervisor", + "/usr/lib/systemd/system/sourceos-supervisor.service" + ], + "evidenceRefs": [ + "urn:srcos:release-receipt:sourceos-supervisor-bootstrap" + ], + "policyRef": "urn:srcos:policy:immutable-node-host-supervisor", + "notes": "Base SourceOS node substrate capability. This is not a desktop-owned feature and not a Socios enrollment dependency." +} From 1e82a51e4890ab5d90d2b28e740f17852d01f2e5 Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Tue, 5 May 2026 21:19:05 -0400 Subject: [PATCH 2/7] feat(immutable-node): add node state projection fixture --- .../nodestateschema.sourceos-evidence-root.json | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) create mode 100644 fixtures/sourceos-spec/nodestateschema.sourceos-evidence-root.json diff --git a/fixtures/sourceos-spec/nodestateschema.sourceos-evidence-root.json b/fixtures/sourceos-spec/nodestateschema.sourceos-evidence-root.json new file mode 100644 index 0000000..a4cbf4b --- /dev/null +++ b/fixtures/sourceos-spec/nodestateschema.sourceos-evidence-root.json @@ -0,0 +1,16 @@ +{ + "id": "urn:srcos:node-state-schema:sourceos-evidence-root", + "type": "NodeStateSchema", + "specVersion": "2.2.0", + "name": "SourceOS evidence root", + "rootPath": "/var/lib/sourceos/evidence", + "stateClass": "append-only-evidence", + "primaryPlane": "execution-evidence", + "schemaRef": "urn:srcos:release-receipt:sourceos-evidence", + "rollbackCompatibility": "required-n-and-n-minus-1", + "mutability": "append-only", + "owner": "agent-machine", + "evidenceRequired": true, + "desktopVisible": true, + "notes": "Durable evidence root survives host rollback. Desktop surfaces may display summarized posture, but Agent Machine and evidence consumers own the state contract." +} From a1630dd7fbe25a01aacf87aa9d64e32b617e4756 Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Tue, 5 May 2026 21:32:40 -0400 Subject: [PATCH 3/7] feat(immutable-node): add immutable node profile projection fixture --- ...lenodeprofile.m2-asahi-agent-node-dev.json | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) create mode 100644 fixtures/sourceos-spec/immutablenodeprofile.m2-asahi-agent-node-dev.json diff --git a/fixtures/sourceos-spec/immutablenodeprofile.m2-asahi-agent-node-dev.json b/fixtures/sourceos-spec/immutablenodeprofile.m2-asahi-agent-node-dev.json new file mode 100644 index 0000000..886bd86 --- /dev/null +++ b/fixtures/sourceos-spec/immutablenodeprofile.m2-asahi-agent-node-dev.json @@ -0,0 +1,45 @@ +{ + "id": "urn:srcos:immutable-node-profile:m2-asahi-agent-node-dev", + "type": "ImmutableNodeProfile", + "specVersion": "2.2.0", + "name": "M2 Asahi SourceOS agent node dev profile", + "channel": "m2-asahi-dev", + "primaryPlane": "agent-runtime-substrate", + "substrate": { + "strategy": "bootc-first", + "hostMutationPosture": "staged-updates-only", + "sociosRequired": false + }, + "bootReleaseSetRef": "urn:srcos:boot-release-set:m2-asahi-dev", + "releaseSetRef": "urn:srcos:release-set:m2-asahi-dev", + "agentMachineProfileRef": "urn:srcos:agent-machine-profile:m2-asahi-dev", + "agentPlaneRuntimeRef": "urn:srcos:agentplane-runtime:m2-asahi-dev", + "workstationProfileRef": "urn:srcos:workstation-profile:workstation-v0", + "desktopConsumerRefs": [ + "urn:srcos:desktop-consumer:sourceos-settings-node-status", + "urn:srcos:desktop-consumer:sourceos-shell-node-posture" + ], + "hostCapabilityPlacementRefs": [ + "urn:srcos:host-capability-placement:sourceos-supervisor" + ], + "nodeStateSchemaRefs": [ + "urn:srcos:node-state-schema:sourceos-evidence-root" + ], + "optionalSociosCapabilityPackRefs": [ + "urn:socios:capability-pack:automation-commons-opt-in" + ], + "validation": { + "planCommand": "sourceosctl immutable-node plan --profile m2-asahi-agent-node-dev --dry-run", + "validateCommand": "sourceosctl immutable-node validate --profile m2-asahi-agent-node-dev", + "inspectCommand": "sourceosctl immutable-node inspect --profile m2-asahi-agent-node-dev", + "evidenceCommand": "agent-machine release evidence inspect --profile m2-asahi-agent-node-dev" + }, + "policyRefs": [ + "urn:srcos:policy:immutable-node-base", + "urn:srcos:policy:socios-opt-in-required" + ], + "evidenceRefs": [ + "urn:srcos:release-receipt:m2-asahi-dev" + ], + "notes": "Node/agent-runtime substrate profile. Workstation and shell references are consumers only. Optional Socios pack reference does not imply enrollment." +} From 170af9249f8e010139a1d477710b52b520bf57b8 Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Tue, 5 May 2026 21:50:41 -0400 Subject: [PATCH 4/7] feat(immutable-node): add functional immutable node planner and applier --- src/agent_machine/immutable_node.py | 300 ++++++++++++++++++++++++++++ 1 file changed, 300 insertions(+) create mode 100644 src/agent_machine/immutable_node.py diff --git a/src/agent_machine/immutable_node.py b/src/agent_machine/immutable_node.py new file mode 100644 index 0000000..fb7fcf4 --- /dev/null +++ b/src/agent_machine/immutable_node.py @@ -0,0 +1,300 @@ +"""Functional immutable-node profile planning and guarded apply helpers. + +This module consumes SourceOS immutable-node projection fixtures and supports a +real staged host mutation path. Mutation is denied by default and requires both +`--execute` and `--policy-ok` in the CLI wrapper. + +Initial supported mutation classes: +- state-roots: create declared /var/lib or /var/cache state roots. +- staging-artifacts: write plan/evidence artifacts under Agent Machine state. + +The module intentionally does not start services, pull images, enroll Socios, +change boot entries, or run systemctl in this first tranche. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from agent_machine.contracts import load_json + +GENERATOR_VERSION = "0.1.0" +PLAN_SPEC_VERSION = "0.1.0" +EVIDENCE_SPEC_VERSION = "0.1.0" +VALID_PROFILE_PLANES = {"node-substrate", "agent-runtime-substrate"} +VALID_STATE_ROOT_PREFIXES = ("/var/lib/", "/var/cache/") +FORBIDDEN_STATE_ROOT_PREFIXES = ("/etc", "/usr") +SUPPORTED_MUTATION_CLASSES = {"state-roots", "staging-artifacts"} +DEFAULT_STAGING_ROOT = "/var/lib/agent-machine/immutable-node" + + +@dataclass(frozen=True) +class ApplyOptions: + target_root: Path + execute: bool + policy_ok: bool + mutation_classes: tuple[str, ...] + evidence_out: Path | None = None + + +class ProjectionIndex(dict[str, dict[str, Any]]): + """Simple id-keyed projection index.""" + + +def utc_now() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def canonical_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":")) + + +def sha256_json(value: Any) -> str: + return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest() + + +def slug_from_urn(value: str) -> str: + tail = value.rsplit(":", 1)[-1] + return re.sub(r"[^a-zA-Z0-9._-]+", "-", tail).strip("-") or "immutable-node" + + +def under_target_root(target_root: Path, absolute_path: str) -> Path: + if not absolute_path.startswith("/"): + raise AssertionError(f"expected absolute path, got {absolute_path!r}") + return target_root / absolute_path.lstrip("/") + + +def ensure_safe_target_root(target_root: Path, execute: bool) -> None: + if execute and target_root == Path("/") and hasattr(os, "geteuid") and os.geteuid() != 0: + raise AssertionError("applying to / requires root privileges") + + +def require_object(path: Path, value: Any) -> dict[str, Any]: + if not isinstance(value, dict): + raise AssertionError(f"{path}: root must be a JSON object") + return value + + +def load_projection_index(fixtures_dir: Path) -> ProjectionIndex: + if not fixtures_dir.exists(): + raise AssertionError(f"fixtures directory does not exist: {fixtures_dir}") + index: ProjectionIndex = ProjectionIndex() + for path in sorted(fixtures_dir.glob("*.json")): + data = require_object(path, load_json(path)) + identifier = data.get("id") + if not isinstance(identifier, str) or not identifier.startswith("urn:srcos:"): + raise AssertionError(f"{path}: id must be a urn:srcos string") + if identifier in index: + raise AssertionError(f"duplicate projection id: {identifier}") + index[identifier] = data + return index + + +def validate_host_capability(capability: dict[str, Any]) -> None: + if capability.get("type") != "HostCapabilityPlacement": + raise AssertionError(f"{capability.get('id')}: expected HostCapabilityPlacement") + if capability.get("mandatoryForBaseNode") is True and capability.get("requiresEnrollment") is True: + raise AssertionError(f"{capability.get('id')}: mandatory base node capability must not require enrollment") + if capability.get("mandatoryForBaseNode") is True and capability.get("primaryPlane") == "desktop-consumer": + raise AssertionError(f"{capability.get('id')}: desktop cannot own mandatory base node capability") + if capability.get("authority") == "socios-optional-pack" and capability.get("mandatoryForBaseNode") is True: + raise AssertionError(f"{capability.get('id')}: Socios optional pack cannot be mandatory for base node") + + +def validate_state_root(state_root: dict[str, Any]) -> None: + if state_root.get("type") != "NodeStateSchema": + raise AssertionError(f"{state_root.get('id')}: expected NodeStateSchema") + root_path = state_root.get("rootPath") + if not isinstance(root_path, str): + raise AssertionError(f"{state_root.get('id')}: rootPath must be a string") + if root_path.startswith(FORBIDDEN_STATE_ROOT_PREFIXES): + raise AssertionError(f"{state_root.get('id')}: rootPath must not live under /etc or /usr") + if not root_path.startswith(VALID_STATE_ROOT_PREFIXES): + raise AssertionError(f"{state_root.get('id')}: rootPath must live under /var/lib or /var/cache") + + +def validate_profile(profile: dict[str, Any], index: ProjectionIndex) -> None: + if profile.get("type") != "ImmutableNodeProfile": + raise AssertionError("profile type must be ImmutableNodeProfile") + primary_plane = profile.get("primaryPlane") + if primary_plane not in VALID_PROFILE_PLANES: + raise AssertionError("ImmutableNodeProfile.primaryPlane must be node-substrate or agent-runtime-substrate") + substrate = profile.get("substrate") + if not isinstance(substrate, dict): + raise AssertionError("ImmutableNodeProfile.substrate must be an object") + if substrate.get("sociosRequired") is not False: + raise AssertionError("Base SourceOS immutable nodes must not require Socios enrollment") + for ref in profile.get("hostCapabilityPlacementRefs", []): + capability = index.get(ref) + if capability is None: + raise AssertionError(f"missing HostCapabilityPlacement projection: {ref}") + validate_host_capability(capability) + for ref in profile.get("nodeStateSchemaRefs", []): + state_root = index.get(ref) + if state_root is None: + raise AssertionError(f"missing NodeStateSchema projection: {ref}") + validate_state_root(state_root) + + +def render_capability(capability: dict[str, Any]) -> dict[str, Any]: + return { + "id": capability.get("id"), + "name": capability.get("name"), + "placementClass": capability.get("placementClass"), + "authority": capability.get("authority"), + "primaryPlane": capability.get("primaryPlane"), + "mandatoryForBaseNode": capability.get("mandatoryForBaseNode", False), + "requiresEnrollment": capability.get("requiresEnrollment", False), + "lifecycleCoupling": capability.get("lifecycleCoupling"), + "pathHints": capability.get("pathHints", []), + "policyRef": capability.get("policyRef"), + "evidenceRefs": capability.get("evidenceRefs", []), + } + + +def render_state_root(state_root: dict[str, Any]) -> dict[str, Any]: + return { + "id": state_root.get("id"), + "name": state_root.get("name"), + "rootPath": state_root.get("rootPath"), + "stateClass": state_root.get("stateClass"), + "primaryPlane": state_root.get("primaryPlane"), + "rollbackCompatibility": state_root.get("rollbackCompatibility"), + "mutability": state_root.get("mutability"), + "owner": state_root.get("owner"), + "evidenceRequired": state_root.get("evidenceRequired", False), + "desktopVisible": state_root.get("desktopVisible", False), + "schemaRef": state_root.get("schemaRef"), + } + + +def render_plan(profile_path: Path, profile: dict[str, Any], index: ProjectionIndex) -> dict[str, Any]: + validate_profile(profile, index) + capabilities = [render_capability(index[ref]) for ref in profile.get("hostCapabilityPlacementRefs", [])] + state_roots = [render_state_root(index[ref]) for ref in profile.get("nodeStateSchemaRefs", [])] + substrate = profile.get("substrate", {}) + return { + "kind": "ImmutableNodePlan", + "specVersion": PLAN_SPEC_VERSION, + "generator": {"name": "agent-machine immutable-node", "version": GENERATOR_VERSION}, + "profile": { + "id": profile.get("id"), + "name": profile.get("name"), + "channel": profile.get("channel"), + "primaryPlane": profile.get("primaryPlane"), + "sourcePath": str(profile_path), + }, + "substrate": { + "strategy": substrate.get("strategy"), + "hostMutationPosture": substrate.get("hostMutationPosture"), + "sociosRequired": substrate.get("sociosRequired"), + }, + "releaseRefs": {"bootReleaseSetRef": profile.get("bootReleaseSetRef"), "releaseSetRef": profile.get("releaseSetRef")}, + "runtimeRefs": {"agentMachineProfileRef": profile.get("agentMachineProfileRef"), "agentPlaneRuntimeRef": profile.get("agentPlaneRuntimeRef")}, + "desktopConsumers": {"workstationProfileRef": profile.get("workstationProfileRef"), "desktopConsumerRefs": profile.get("desktopConsumerRefs", []), "desktopOwnsSubstrate": False}, + "capabilityPlacements": capabilities, + "stateRoots": state_roots, + "optionalSociosCapabilityPackRefs": profile.get("optionalSociosCapabilityPackRefs", []), + "policyRefs": profile.get("policyRefs", []), + "evidenceRefs": profile.get("evidenceRefs", []), + "mutationPlan": { + "supportedMutationClasses": sorted(SUPPORTED_MUTATION_CLASSES), + "defaultMutationClasses": ["state-roots", "staging-artifacts"], + "requiresExecuteFlag": True, + "requiresPolicyOkFlag": True, + }, + "safety": { + "hostMutationPerformed": False, + "sociosEnrollmentPerformed": False, + "sociosRequired": False, + "desktopOwnsSubstrate": False, + "rawSecretsIncluded": False, + }, + "verdict": "planned", + } + + +def preflight_plan(plan: dict[str, Any], target_root: Path, mutation_classes: tuple[str, ...]) -> dict[str, Any]: + unsupported = sorted(set(mutation_classes) - SUPPORTED_MUTATION_CLASSES) + if unsupported: + raise AssertionError(f"unsupported mutation classes: {unsupported}") + state_root_checks = [] + for root in plan.get("stateRoots", []): + root_path = str(root.get("rootPath")) + validate_state_root({"type": "NodeStateSchema", "id": root.get("id"), "rootPath": root_path}) + target_path = under_target_root(target_root, root_path) + state_root_checks.append({"rootPath": root_path, "targetPath": str(target_path), "exists": target_path.exists(), "willCreate": "state-roots" in mutation_classes and not target_path.exists()}) + return { + "kind": "ImmutableNodePreflight", + "specVersion": EVIDENCE_SPEC_VERSION, + "profileRef": plan["profile"]["id"], + "targetRoot": str(target_root), + "mutationClasses": list(mutation_classes), + "stateRootChecks": state_root_checks, + "hostMutationPerformed": False, + "verdict": "preflight-passed", + } + + +def apply_plan(plan: dict[str, Any], options: ApplyOptions) -> dict[str, Any]: + if not options.execute or not options.policy_ok: + raise AssertionError("apply requires both --execute and --policy-ok") + ensure_safe_target_root(options.target_root, options.execute) + preflight = preflight_plan(plan, options.target_root, options.mutation_classes) + profile_slug = slug_from_urn(str(plan["profile"]["id"])) + touched: list[dict[str, str]] = [] + + if "state-roots" in options.mutation_classes: + for check in preflight["stateRootChecks"]: + target_path = Path(check["targetPath"]) + target_path.mkdir(parents=True, exist_ok=True) + os.chmod(target_path, 0o750) + touched.append({"action": "ensure-directory", "path": str(target_path)}) + + staging_dir = under_target_root(options.target_root, f"{DEFAULT_STAGING_ROOT}/{profile_slug}") + if "staging-artifacts" in options.mutation_classes: + staging_dir.mkdir(parents=True, exist_ok=True) + os.chmod(staging_dir, 0o750) + plan_path = staging_dir / "immutable-node-plan.json" + plan_path.write_text(json.dumps(plan, indent=2, sort_keys=True) + "\n", encoding="utf-8") + touched.append({"action": "write-plan", "path": str(plan_path)}) + + evidence = { + "kind": "ImmutableNodeApplyEvidence", + "specVersion": EVIDENCE_SPEC_VERSION, + "issuedAt": utc_now(), + "profileRef": plan["profile"]["id"], + "profileChannel": plan["profile"].get("channel"), + "targetRoot": str(options.target_root), + "mutationClasses": list(options.mutation_classes), + "planDigestSha256": sha256_json(plan), + "hostMutationPerformed": True, + "sociosEnrollmentPerformed": False, + "sociosRequired": False, + "desktopOwnsSubstrate": False, + "rawSecretsIncluded": False, + "touched": touched, + "verdict": "applied", + } + + evidence_out = options.evidence_out + if evidence_out is None and "staging-artifacts" in options.mutation_classes: + evidence_out = staging_dir / "immutable-node-apply-evidence.json" + if evidence_out is not None: + evidence_out.parent.mkdir(parents=True, exist_ok=True) + evidence_out.write_text(json.dumps(evidence, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return evidence + + +def emit_json(payload: dict[str, Any], pretty: bool) -> None: + if pretty: + print(json.dumps(payload, indent=2, sort_keys=True)) + else: + print(json.dumps(payload, sort_keys=True, separators=(",", ":"))) From 81ede9bb6a1248f917562f1418263b7a8f39c6c0 Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Fri, 8 May 2026 11:02:22 -0400 Subject: [PATCH 5/7] feat(immutable-node): add guarded plan preflight apply script --- scripts/immutable-node-plan.py | 129 +++++++++++++++++++++++++++++++++ 1 file changed, 129 insertions(+) create mode 100644 scripts/immutable-node-plan.py diff --git a/scripts/immutable-node-plan.py b/scripts/immutable-node-plan.py new file mode 100644 index 0000000..d65322d --- /dev/null +++ b/scripts/immutable-node-plan.py @@ -0,0 +1,129 @@ +#!/usr/bin/env python3 +"""Plan, preflight, and guarded-apply SourceOS immutable-node profiles.""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from agent_machine.contracts import load_json +from agent_machine.immutable_node import ( + ApplyOptions, + emit_json, + load_projection_index, + preflight_plan, + render_plan, + validate_profile, +) + +DEFAULT_PROFILE = ROOT / "fixtures" / "sourceos-spec" / "immutablenodeprofile.m2-asahi-agent-node-dev.json" +DEFAULT_FIXTURES_DIR = ROOT / "fixtures" / "sourceos-spec" +DEFAULT_MUTATION_CLASSES = ("state-roots", "staging-artifacts") + + +def _load_plan(args: argparse.Namespace) -> dict: + profile_path = args.profile_json + profile = load_json(profile_path) + index = load_projection_index(args.fixtures_dir) + return render_plan(profile_path, profile, index) + + +def cmd_validate(args: argparse.Namespace) -> int: + profile = load_json(args.profile_json) + index = load_projection_index(args.fixtures_dir) + validate_profile(profile, index) + emit_json( + { + "kind": "ImmutableNodeValidation", + "specVersion": "0.1.0", + "profileRef": profile["id"], + "fixturesDir": str(args.fixtures_dir), + "verdict": "valid", + }, + args.pretty, + ) + return 0 + + +def cmd_plan(args: argparse.Namespace) -> int: + emit_json(_load_plan(args), args.pretty) + return 0 + + +def cmd_preflight(args: argparse.Namespace) -> int: + plan = _load_plan(args) + emit_json(preflight_plan(plan, args.target_root, tuple(args.mutation_class)), args.pretty) + return 0 + + +def cmd_apply(args: argparse.Namespace) -> int: + from agent_machine.immutable_node import apply_plan + + plan = _load_plan(args) + evidence = apply_plan( + plan, + ApplyOptions( + target_root=args.target_root, + execute=args.execute, + policy_ok=args.policy_ok, + mutation_classes=tuple(args.mutation_class), + evidence_out=args.evidence_out, + ), + ) + emit_json(evidence, args.pretty) + return 0 + + +def add_common_arguments(parser: argparse.ArgumentParser) -> None: + parser.add_argument("profile_json", type=Path, nargs="?", default=DEFAULT_PROFILE) + parser.add_argument("--fixtures-dir", type=Path, default=DEFAULT_FIXTURES_DIR) + parser.add_argument("--pretty", action="store_true") + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="SourceOS immutable-node plan/preflight/apply helper") + subcommands = parser.add_subparsers(dest="command", required=True) + + validate = subcommands.add_parser("validate", help="Validate immutable-node profile and referenced projection fixtures") + add_common_arguments(validate) + validate.set_defaults(func=cmd_validate) + + plan = subcommands.add_parser("plan", help="Render ImmutableNodePlan JSON") + add_common_arguments(plan) + plan.set_defaults(func=cmd_plan) + + preflight = subcommands.add_parser("preflight", help="Preflight state-root and staging-artifact mutation without writing") + add_common_arguments(preflight) + preflight.add_argument("--target-root", type=Path, required=True) + preflight.add_argument("--mutation-class", action="append", choices=sorted(DEFAULT_MUTATION_CLASSES), default=list(DEFAULT_MUTATION_CLASSES)) + preflight.set_defaults(func=cmd_preflight) + + apply = subcommands.add_parser("apply", help="Apply guarded immutable-node mutation") + add_common_arguments(apply) + apply.add_argument("--target-root", type=Path, required=True) + apply.add_argument("--mutation-class", action="append", choices=sorted(DEFAULT_MUTATION_CLASSES), default=list(DEFAULT_MUTATION_CLASSES)) + apply.add_argument("--execute", action="store_true", help="Required to permit mutation") + apply.add_argument("--policy-ok", action="store_true", help="Required bootstrap policy assertion") + apply.add_argument("--evidence-out", type=Path) + apply.set_defaults(func=cmd_apply) + + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + return int(args.func(args)) + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except AssertionError as exc: + print(str(exc), file=sys.stderr) + raise SystemExit(1) from exc From 3b4efbd5d11af7157a1c51650bd0ce0bbe035fff Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Fri, 8 May 2026 13:16:54 -0400 Subject: [PATCH 6/7] feat(immutable-node): add projection safety validator --- .../validate-immutable-node-projections.py | 56 +++++++++++++++++++ 1 file changed, 56 insertions(+) create mode 100644 scripts/validate-immutable-node-projections.py diff --git a/scripts/validate-immutable-node-projections.py b/scripts/validate-immutable-node-projections.py new file mode 100644 index 0000000..c988c9a --- /dev/null +++ b/scripts/validate-immutable-node-projections.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +"""Validate SourceOS immutable-node projection fixtures and safety boundaries.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from agent_machine.contracts import load_json +from agent_machine.immutable_node import load_projection_index, validate_profile + +FIXTURE_DIR = ROOT / "fixtures" / "sourceos-spec" +PROFILE = FIXTURE_DIR / "immutablenodeprofile.m2-asahi-agent-node-dev.json" + + +def main() -> int: + index = load_projection_index(FIXTURE_DIR) + profile = load_json(PROFILE) + validate_profile(profile, index) + + if profile.get("primaryPlane") == "desktop-consumer": + raise AssertionError("ImmutableNodeProfile must not be desktop-owned") + if profile.get("substrate", {}).get("sociosRequired") is not False: + raise AssertionError("ImmutableNodeProfile must keep Socios optional for base SourceOS nodes") + + for ref in profile.get("hostCapabilityPlacementRefs", []): + capability = index[ref] + if capability.get("mandatoryForBaseNode") is True and capability.get("requiresEnrollment") is True: + raise AssertionError(f"{ref}: mandatory base capability must not require enrollment") + if capability.get("authority") == "socios-optional-pack" and capability.get("mandatoryForBaseNode") is True: + raise AssertionError(f"{ref}: Socios optional pack cannot be mandatory") + + for ref in profile.get("nodeStateSchemaRefs", []): + state_root = index[ref] + root_path = state_root.get("rootPath", "") + if not str(root_path).startswith(("/var/lib/", "/var/cache/")): + raise AssertionError(f"{ref}: state root must live under /var/lib or /var/cache") + if str(root_path).startswith(("/etc", "/usr")): + raise AssertionError(f"{ref}: state root must not live under /etc or /usr") + + print(f"VALID immutable-node profile {PROFILE.relative_to(ROOT)}") + print("VALID immutable-node safety boundary: Socios optional, desktop consumer-only, /var-only state roots") + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except AssertionError as exc: + print(str(exc), file=sys.stderr) + raise SystemExit(1) from exc From 896aae7d66b8a26c7c4a3921744f69dc011d5469 Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Fri, 8 May 2026 13:19:06 -0400 Subject: [PATCH 7/7] test(immutable-node): add guarded mutation safety tests --- scripts/test-immutable-node.py | 118 +++++++++++++++++++++++++++++++++ 1 file changed, 118 insertions(+) create mode 100644 scripts/test-immutable-node.py diff --git a/scripts/test-immutable-node.py b/scripts/test-immutable-node.py new file mode 100644 index 0000000..35ef4b6 --- /dev/null +++ b/scripts/test-immutable-node.py @@ -0,0 +1,118 @@ +#!/usr/bin/env python3 +"""Executable safety tests for immutable-node planning and guarded apply.""" + +from __future__ import annotations + +import copy +import json +import sys +import tempfile +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from agent_machine.contracts import load_json +from agent_machine.immutable_node import ( + ApplyOptions, + load_projection_index, + preflight_plan, + render_plan, + sha256_json, + apply_plan, + validate_host_capability, + validate_profile, + validate_state_root, +) + +FIXTURE_DIR = ROOT / "fixtures" / "sourceos-spec" +PROFILE = FIXTURE_DIR / "immutablenodeprofile.m2-asahi-agent-node-dev.json" + + +def expect_failure(label: str, fn) -> None: + try: + fn() + except AssertionError: + return + raise AssertionError(f"expected failure did not occur: {label}") + + +def main() -> int: + index = load_projection_index(FIXTURE_DIR) + profile = load_json(PROFILE) + validate_profile(profile, index) + + plan = render_plan(PROFILE, profile, index) + assert plan["kind"] == "ImmutableNodePlan" + assert plan["safety"]["hostMutationPerformed"] is False + assert plan["safety"]["sociosRequired"] is False + assert plan["desktopConsumers"]["desktopOwnsSubstrate"] is False + assert sha256_json(plan) == sha256_json(copy.deepcopy(plan)) + + with tempfile.TemporaryDirectory(prefix="agent-machine-immutable-node-") as tmp: + target_root = Path(tmp) + preflight = preflight_plan(plan, target_root, ("state-roots", "staging-artifacts")) + assert preflight["kind"] == "ImmutableNodePreflight" + assert preflight["hostMutationPerformed"] is False + assert preflight["stateRootChecks"][0]["willCreate"] is True + assert not (target_root / "var/lib/sourceos/evidence").exists() + + expect_failure( + "apply requires --execute and --policy-ok", + lambda: apply_plan( + plan, + ApplyOptions( + target_root=target_root, + execute=False, + policy_ok=False, + mutation_classes=("state-roots", "staging-artifacts"), + ), + ), + ) + + evidence = apply_plan( + plan, + ApplyOptions( + target_root=target_root, + execute=True, + policy_ok=True, + mutation_classes=("state-roots", "staging-artifacts"), + ), + ) + assert evidence["kind"] == "ImmutableNodeApplyEvidence" + assert evidence["hostMutationPerformed"] is True + assert evidence["sociosEnrollmentPerformed"] is False + assert evidence["rawSecretsIncluded"] is False + assert evidence["planDigestSha256"] == sha256_json(plan) + assert (target_root / "var/lib/sourceos/evidence").is_dir() + assert (target_root / "var/lib/agent-machine/immutable-node/m2-asahi-agent-node-dev/immutable-node-plan.json").is_file() + assert (target_root / "var/lib/agent-machine/immutable-node/m2-asahi-agent-node-dev/immutable-node-apply-evidence.json").is_file() + + bad_profile = copy.deepcopy(profile) + bad_profile["substrate"]["sociosRequired"] = True + expect_failure("sociosRequired must be false", lambda: validate_profile(bad_profile, index)) + + bad_profile = copy.deepcopy(profile) + bad_profile["primaryPlane"] = "desktop-consumer" + expect_failure("desktop cannot own immutable node profile", lambda: validate_profile(bad_profile, index)) + + bad_state = copy.deepcopy(index["urn:srcos:node-state-schema:sourceos-evidence-root"]) + bad_state["rootPath"] = "/etc/sourceos/evidence" + expect_failure("state root under /etc rejected", lambda: validate_state_root(bad_state)) + + bad_state = copy.deepcopy(index["urn:srcos:node-state-schema:sourceos-evidence-root"]) + bad_state["rootPath"] = "/usr/lib/sourceos/evidence" + expect_failure("state root under /usr rejected", lambda: validate_state_root(bad_state)) + + bad_capability = copy.deepcopy(index["urn:srcos:host-capability-placement:sourceos-supervisor"]) + bad_capability["requiresEnrollment"] = True + expect_failure("mandatory capability cannot require enrollment", lambda: validate_host_capability(bad_capability)) + + print(json.dumps({"kind": "ImmutableNodeSafetyTest", "verdict": "passed"}, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())