Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -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."
}
Original file line number Diff line number Diff line change
@@ -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."
}
16 changes: 16 additions & 0 deletions fixtures/sourceos-spec/nodestateschema.sourceos-evidence-root.json
Original file line number Diff line number Diff line change
@@ -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."
}
129 changes: 129 additions & 0 deletions scripts/immutable-node-plan.py
Original file line number Diff line number Diff line change
@@ -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
118 changes: 118 additions & 0 deletions scripts/test-immutable-node.py
Original file line number Diff line number Diff line change
@@ -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())
56 changes: 56 additions & 0 deletions scripts/validate-immutable-node-projections.py
Original file line number Diff line number Diff line change
@@ -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
Loading
Loading