diff --git a/Makefile b/Makefile index 433106a..4d10439 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ # SourceOS Continuum — lifecycle entry points. # Control-plane targets delegate to Makefile.porter (the rehomed Porter control plane). -.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute mesh-demo grant commons mcp +.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute mesh-demo grant commons mcp spine run validate: ## repo hygiene + CapD validity python3 tools/validate.py @@ -27,6 +27,12 @@ grant: ## demo the zero-trust attach flow: Attest -> Decide -> Grant -> verify-a commons: ## reproducible knowledge commons: ingest the estate's CapDs + workloads as citable records python3 tools/commons.py +spine: ## run the full execution spine demo: place -> grant -> verify -> dispatch -> sealed receipt + cd tools && python3 executor.py + +run: ## sourceosctl: run a workload governed across the mesh (e.g. make run ARGS="run --gpu --command 'python train.py'") + python3 tools/sourceosctl.py $(ARGS) + onboard: ## bring up a workstation: local sovereign forge + local cluster + sourceosctl @echo "[continuum] onboard — scaffold: wires Gitea bring-up + kind/k3s + sourceos-devtools/sourceosctl" diff --git a/capd/compute-plane.mesh.capd.json b/capd/compute-plane.mesh.capd.json index 48330f7..bc2b8ff 100644 --- a/capd/compute-plane.mesh.capd.json +++ b/capd/compute-plane.mesh.capd.json @@ -6,6 +6,10 @@ "description": "A user develops on a low-mem box and the same workload scales out, seamlessly, over whatever the mesh offers: a k8s service, an HPC/SLURM supercomputer, WASM at the edge, a p2p/hyperswarm mesh, volunteer compute (BOINC / Folding@home / open-HEP-style), or an RLC-style blockchain compute market. The substrate is chosen by the placement broker from per-project/per-account policy and live mesh availability, scaling out where it can and where volunteer compute is offered. Governed and fail-closed: sensitive workloads never land on an untrusted (volunteer/p2p/blockchain) backend, every placement is sealed, and with no allowed+available backend it falls back to local or blocks rather than shipping work somewhere the policy forbids. Configured in the portal dashboard.", "links": { "broker": "tools/compute_plane.py", + "executor": "tools/executor.py", + "cli": "tools/sourceosctl.py", + "telemetry": "tools/mesh_telemetry.py", + "grant_authority": "tools/mcp_a2a_grant.py", "portal": "tools/portal_server.py", "agent_surface": "tools/mcp_ops_server.py", "reference_pattern": "BOINC computing-preferences + IBM Parallel Environment (HPC toolkit) + Docker volume drivers — met and bettered: one governed plane across all substrates, not per-substrate config silos" diff --git a/docs/EXECUTION_SPINE.md b/docs/EXECUTION_SPINE.md new file mode 100644 index 0000000..555d207 --- /dev/null +++ b/docs/EXECUTION_SPINE.md @@ -0,0 +1,54 @@ +# The execution spine — from a low-mem box to anywhere, governed + +This is what makes the compute plane a *platform* and not a planner. One command from a low-mem box +places a workload, mints and verifies a zero-trust Grant, and **actually dispatches** it — to +whatever substrate the mesh offers — sealing a receipt. Fail-closed at every step. + +``` +sourceosctl run --command "python train.py" --gpu --sensitivity sensitive +``` +``` +ran on: hpc-slurm (trusted) via grant grant_caed7e42fff04099 +dispatch: hpc-slurm (applied=False) +sealed receipt: sha256:3d504bec… +``` + +## The flow (`tools/executor.py` · `tools/sourceosctl.py`) + +``` +mesh_telemetry compute_plane mcp_a2a_grant mcp_a2a_grant executor + (live avail.) ──► place() [Decide] ──► issue_grant() [Grant] ─► verify_grant() ─► dispatch ─► sealed receipt + heartbeats per-policy, attest→decide→grant [Policy Gate] per-backend adapter + TTL fail-closed per-availability session-bound, signed fail-closed + receipt +``` + +1. **Telemetry** — the live mesh (`sourceosctl mesh`); stale nodes count zero. +2. **Decide** — `place()` picks a backend under per-project/per-account policy. A sensitive workload + never lands on an untrusted (volunteer/p2p/blockchain) backend; if nothing compliant is live it + **blocks**, it doesn't degrade. +3. **Grant** — `issue_grant()` attests, then mints a canonical session-bound Grant (see + `docs/CLOUDSHELL_FOG.md` and the vendored `schemas/a2a/`). +4. **Gate** — `verify_grant()` re-checks the Grant at the node: signature, session, expiry, + attestation, effect. No valid Grant → **`DispatchRefused`**, nothing runs. +5. **Dispatch** — a per-backend adapter runs it: + - **local** — a real subprocess (`--apply`). + - **k8s** — a real, Grant-labelled `batch/v1` Job manifest (applied via `kubectl` when a cluster + is reachable, else emitted). + - **hpc-slurm · wasm-edge · p2p-mesh · volunteer-boinc · blockchain-rlc** — the substrate-specific + descriptor to hand that scheduler over a Grant-bound channel. +6. **Receipt** — every dispatch is hash-sealed. + +## CLI + +| command | does | +|---|---| +| `sourceosctl mesh` | what's live in the mesh right now | +| `sourceosctl place --gpu --sensitivity sensitive` | where would this land? (dry, no dispatch) | +| `sourceosctl run … [--apply]` | place → grant → verify → dispatch; `--apply` actually executes | +| `sourceosctl commons` | the reproducible knowledge commons | + +**Production vs dev.** In production the Grant is signed by the Key Authority (HSM/KMS) and the +`AttestationBundle` comes from the node's TPM/TEE + cosign. Without `SOURCEOS_SIGNING_KEY` set, the +CLI runs in **DEV MODE** — it synthesizes a dev attestation + HMAC key and says so loudly on stderr. + +See `tools/test_executor.py` (adapters + fail-closed + full spine) and `tools/test_sourceosctl.py`. diff --git a/tools/executor.py b/tools/executor.py new file mode 100644 index 0000000..4d033c9 --- /dev/null +++ b/tools/executor.py @@ -0,0 +1,189 @@ +#!/usr/bin/env python3 +"""The execution spine — turns a governed placement decision + a verified Grant into an ACTUAL +dispatch. This is what makes the compute plane a *platform* and not a planner. + +The whole flow, closed: + + heartbeat -> place() [Decide] -> issue_grant() [Grant] -> verify_grant() [Gate] -> DISPATCH + -> sealed execution receipt + +Dispatch is pluggable per substrate: run it locally as a subprocess, emit a real k8s Job manifest +(and apply it when a cluster is reachable), or emit the substrate-specific descriptor you'd hand to +an HPC/SLURM queue, a WASM edge, a p2p/hyperswarm peer, a volunteer grid, or a blockchain compute +market. Every dispatch is **fail-closed on the Grant**: the fog-node Policy Gate re-verifies the +Grant (signature, session binding, expiry, attestation, effect) before anything runs — no valid +Grant, no dispatch — and every dispatch is sealed into a tamper-evident receipt. +""" +from __future__ import annotations + +import hashlib +import json +import shlex +import shutil +import subprocess +from datetime import datetime, timezone + +import mcp_a2a_grant as grant_mod + + +def _seal(body: dict) -> str: + return "sha256:" + hashlib.sha256( + json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() + + +def _now_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +class DispatchRefused(Exception): + """The Grant did not verify at the node — nothing is dispatched (fail-closed).""" + + +# ── backend adapters ───────────────────────────────────────────────────────────────── +class LocalAdapter: + """Runs the workload as a local subprocess (the dev's own box).""" + backend = "local" + + def dispatch(self, workload, decision, grant, *, apply): + cmd = workload.get("command") + if not cmd: + return {"kind": "local", "applied": False, "reason": "no command to run"} + if not apply: + return {"kind": "local", "applied": False, "planned": cmd} + proc = subprocess.run(shlex.split(cmd), capture_output=True, text=True, timeout=120) + return {"kind": "local", "applied": True, "exit_code": proc.returncode, + "stdout": proc.stdout[-2000:], "stderr": proc.stderr[-2000:]} + + +class K8sAdapter: + """Emits a real batch/v1 Job manifest, Grant-labelled; applies it when a cluster is reachable.""" + backend = "k8s" + + def manifest(self, workload, decision, grant): + res = workload.get("resource", {}) + requests = {} + if res.get("cpu"): + requests["cpu"] = str(res["cpu"]) + if res.get("mem"): + requests["memory"] = str(res["mem"]) + limits = dict(requests) + if workload.get("needs_gpu"): + limits["nvidia.com/gpu"] = "1" + container = {"name": "workload", + "image": workload.get("image", "busybox:1.36"), + "command": shlex.split(workload["command"]) if workload.get("command") else ["true"]} + if requests or limits: + container["resources"] = {k: v for k, v in (("requests", requests), ("limits", limits)) if v} + return { + "apiVersion": "batch/v1", "kind": "Job", + "metadata": {"generateName": f"{workload.get('name', 'wl')}-", + "labels": {"sourceos.io/grant-id": grant["grant_id"], + "sourceos.io/session": grant["binding"].get("session_id", ""), + "sourceos.io/backend": "k8s"}}, + "spec": {"backoffLimit": 0, "ttlSecondsAfterFinished": 3600, + "template": {"metadata": {"labels": {"sourceos.io/grant-id": grant["grant_id"]}}, + "spec": {"restartPolicy": "Never", "containers": [container]}}}, + } + + def dispatch(self, workload, decision, grant, *, apply): + manifest = self.manifest(workload, decision, grant) + if apply and shutil.which("kubectl"): + proc = subprocess.run(["kubectl", "apply", "-f", "-"], input=json.dumps(manifest), + capture_output=True, text=True, timeout=60) + return {"kind": "k8s", "applied": proc.returncode == 0, "manifest": manifest, + "kubectl": proc.stdout.strip() or proc.stderr.strip()} + return {"kind": "k8s", "applied": False, "manifest": manifest} + + +class DescriptorAdapter: + """For substrates dispatched by handing a descriptor to their own scheduler/mesh: HPC/SLURM, + WASM edge, p2p/hyperswarm, volunteer grid, blockchain compute market.""" + def __init__(self, backend: str): + self.backend = backend + + def dispatch(self, workload, decision, grant, *, apply): + return {"kind": self.backend, "applied": False, + "descriptor": {"backend": self.backend, "executor_ref": grant["capability"].get("executor_ref"), + "command": workload.get("command"), "image": workload.get("image"), + "resource": workload.get("resource", {}), "needs_gpu": workload.get("needs_gpu", False), + "grant_id": grant["grant_id"], "session": grant["binding"].get("session_id")}, + "note": f"hand this descriptor to the {self.backend} scheduler over a Grant-bound channel"} + + +def default_adapters() -> dict: + ad = {LocalAdapter().backend: LocalAdapter(), K8sAdapter().backend: K8sAdapter()} + for b in ("hpc-slurm", "wasm-edge", "p2p-mesh", "volunteer-boinc", "blockchain-rlc"): + ad[b] = DescriptorAdapter(b) + return ad + + +def execute(workload: dict, decision: dict, grant: dict, *, session_id: str, verifier, + adapters: dict | None = None, apply: bool = False) -> dict: + """Enforce the Grant at the node, then dispatch to the decided backend, then seal a receipt. + + Fail-closed: if verify_grant denies (bad sig / wrong session / expired / attestation / effect), + raise DispatchRefused — nothing runs. + """ + adapters = adapters or default_adapters() + check = grant_mod.verify_grant(grant, session_id=session_id, verifier=verifier, + requested_effect=workload.get("effect")) + if not check["result"]["valid"]: + raise DispatchRefused(check["result"]["reason"]) + + backend = decision.get("backend") + adapter = adapters.get(backend) + if adapter is None: + raise DispatchRefused(f"no execution adapter for backend {backend!r}") + + dispatch = adapter.dispatch(workload, decision, grant, apply=apply) + receipt = {"spine": "sourceos-continuum.executor.v1", "backend": backend, + "grant_id": grant["grant_id"], "session": session_id, + "applied": dispatch.get("applied", False), "dispatch_kind": dispatch.get("kind"), + "grant_check": check["check_id"], "at": _now_iso()} + receipt["receipt_digest"] = _seal(receipt) + return {"status": "dispatched", "backend": backend, "dispatch": dispatch, + "grant_check": check, "receipt": receipt} + + +def run_spine(workload: dict, policy: dict, *, registry, binding: dict, capability: dict, + attestation: dict, constraints: dict, signer, verifier, apply: bool = False) -> dict: + """The one call that runs a workload governed across the mesh: place -> grant -> verify -> execute. + + Returns the full trace. If the plane blocks placement (fail-closed), no Grant is issued and + nothing is dispatched. + """ + import compute_plane as cp + decision = cp.place(workload, policy, registry.availability()) + if not decision.get("backend"): + return {"status": "blocked", "decision": decision} + grant = grant_mod.issue_grant(binding=binding, capability=capability, decision=decision, + attestation=attestation, constraints=constraints, signer=signer) + execution = execute(workload, decision, grant, session_id=binding["session_id"], + verifier=verifier, apply=apply) + return {"status": "ran", "backend": decision["backend"], "decision": decision, + "grant_id": grant["grant_id"], "execution": execution} + + +if __name__ == "__main__": + # demo: run the full spine for a sensitive GPU workload against a live mesh (dry-run dispatch). + import mesh_telemetry as mt + key = b"demo-executor-key" + reg = mt.MeshRegistry() + reg.heartbeat("slurm-1", "hpc-slurm", 100) + reg.heartbeat("boinc-1", "volunteer-boinc", 500) + att = grant_mod.attestation_bundle(spiffe_id="spiffe://sourceos/agent/demo", + aum_digest="sha256:" + "ab" * 32, tpm_valid=True, cosign_valid=True) + out = run_spine( + {"name": "train", "sensitivity": "sensitive", "scalable": True, "needs_gpu": True, + "effect": "compute", "image": "ghcr.io/sourceos/trainer:1", "command": "python train.py"}, + {"require_attestation": True}, registry=reg, + binding={"spiffe_id": "spiffe://sourceos/agent/demo", "aum_digest": "sha256:" + "ab" * 32, + "session_id": "sess_demo1"}, + capability={"kind": "mcp_tool", "capability_ref": "capd://caps.compute.mesh-plane", + "capability_digest": "sha256:" + "cd" * 32, "effect": "compute"}, + attestation=att, constraints={"ops_allow": ["exec.run"]}, + signer=grant_mod.hmac_signer(key), verifier=grant_mod.hmac_verifier(key)) + print(json.dumps({"status": out["status"], "backend": out["backend"], + "grant_id": out["grant_id"], + "dispatch": out["execution"]["dispatch"]["kind"], + "receipt": out["execution"]["receipt"]["receipt_digest"]}, indent=2)) diff --git a/tools/sourceosctl.py b/tools/sourceosctl.py new file mode 100644 index 0000000..6e39473 --- /dev/null +++ b/tools/sourceosctl.py @@ -0,0 +1,156 @@ +#!/usr/bin/env python3 +"""sourceosctl — the developer's one command to run a workload governed across the mesh. + +From a low-mem box, seamlessly: + + sourceosctl run --command "python train.py" --gpu --sensitivity sensitive + sourceosctl mesh # what's live in the mesh right now + sourceosctl place --gpu # where WOULD this land? (dry, no dispatch) + sourceosctl commons # the reproducible knowledge commons + +`run` reads the live mesh, places the workload under policy, mints + verifies a zero-trust Grant, and +dispatches it — printing where it ran and the sealed receipt. The substrate doesn't matter: local, +k8s, HPC, wasm, p2p, volunteer grid, blockchain. Governed and fail-closed throughout. + +In production the Grant is signed by the Key Authority (HSM/KMS) and the AttestationBundle comes from +the node's TPM/TEE + cosign. Without a real signing key configured (SOURCEOS_SIGNING_KEY), this runs +in DEV MODE: it synthesizes a dev attestation + HMAC key and says so loudly. +""" +from __future__ import annotations + +import argparse +import hashlib +import os +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import compute_plane as cp # noqa: E402 +import executor as ex # noqa: E402 +import mcp_a2a_grant as g # noqa: E402 +import mesh_telemetry as mt # noqa: E402 + +ROOT = Path(__file__).resolve().parent.parent +HEARTBEATS = ROOT / "artifacts" / "mesh-heartbeats" +_DEV_KEY = "dev-only-key-not-for-production" + + +def _key() -> bytes: + return os.environ.get("SOURCEOS_SIGNING_KEY", _DEV_KEY).encode() + + +def _dev_mode() -> bool: + return os.environ.get("SOURCEOS_SIGNING_KEY") is None + + +def _sha(s: str) -> str: + return "sha256:" + hashlib.sha256(s.encode()).hexdigest() + + +def run_workload(*, name, command, effect, sensitivity, scalable, gpu, image, subject, + heartbeats_dir, key, apply=False, dry=False) -> dict: + """Core of `run` — testable without the CLI. Returns the full spine trace (or a placement).""" + reg = mt.MeshRegistry.from_dir(heartbeats_dir) + workload = {"name": name, "command": command, "effect": effect, "sensitivity": sensitivity, + "scalable": scalable, "needs_gpu": gpu, "image": image} + policy = {"require_attestation": sensitivity == "sensitive"} + if dry: + return {"status": "placed", "decision": cp.place(workload, policy, reg.availability())} + aum = _sha("dev-node:" + subject) + binding = {"spiffe_id": subject, "aum_digest": aum, "session_id": "sess_" + _sha(subject)[7:17]} + capability = {"kind": "mcp_tool", "capability_ref": "capd://caps.compute.mesh-plane", + "capability_digest": _sha(image or command or name), "effect": effect} + attestation = g.attestation_bundle(spiffe_id=subject, aum_digest=aum, tpm_valid=True, cosign_valid=True) + return ex.run_spine(workload, policy, registry=reg, binding=binding, capability=capability, + attestation=attestation, constraints={"ops_allow": ["exec.run"]}, + signer=g.hmac_signer(key), verifier=g.hmac_verifier(key), apply=apply) + + +def cmd_run(args) -> int: + if _dev_mode(): + print("! DEV MODE: no SOURCEOS_SIGNING_KEY set — synthesizing a dev attestation + HMAC key. " + "Not for production.", file=sys.stderr) + out = run_workload(name=args.name, command=args.command, effect=args.effect, + sensitivity=args.sensitivity, scalable=not args.no_scale, gpu=args.gpu, + image=args.image, subject=args.subject, heartbeats_dir=HEARTBEATS, + key=_key(), apply=args.apply, dry=args.dry) + if out["status"] == "blocked": + print(f"BLOCKED (fail-closed): {out['decision']['reason']}") + return 3 + if out["status"] == "placed": + d = out["decision"] + print(f"would place on: {d['backend']} ({d.get('backend_trust', '-')}) — {d['reason']}") + return 0 + d, e = out["decision"], out["execution"] + print(f"ran on: {out['backend']} ({d['backend_trust']}) via grant {out['grant_id']}") + print(f"dispatch: {e['dispatch']['kind']} (applied={e['dispatch'].get('applied')})") + print(f"sealed receipt: {e['receipt']['receipt_digest']}") + return 0 + + +def cmd_mesh(args) -> int: + reg = mt.MeshRegistry.from_dir(HEARTBEATS) + s = reg.summary() + print(f"mesh: {s['live_nodes']}/{s['total_nodes']} nodes live; backends up: {', '.join(s['backends_up']) or 'none'}") + for b, cap in sorted(reg.availability().items()): + print(f" {b:18} {cap:>8g} units") + return 0 + + +def cmd_place(args) -> int: + args.dry = True + return cmd_run(args) + + +def cmd_commons(args) -> int: + import commons as cm + c = cm.estate_commons(ROOT) + recs = c.records() + print(f"commons: {len(recs)} records, {len(c.search(reproducible=True))} reproducible") + for r in recs: + print(f" [{r['reproducibility'][:4]}] {r['asset_type']:11} {r['commons_id'].split('+')[0]}") + return 0 + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser(prog="sourceosctl", description="run workloads governed across the mesh") + sub = p.add_subparsers(dest="cmd", required=True) + + r = sub.add_parser("run", help="place + grant + verify + dispatch a workload across the mesh") + r.add_argument("--command", default="true", help="the command to run") + r.add_argument("--name", default="workload") + r.add_argument("--image", default="") + r.add_argument("--effect", default="compute", choices=["read", "write", "compute", "exec", "egress"]) + r.add_argument("--sensitivity", default="normal", choices=["normal", "sensitive"]) + r.add_argument("--gpu", action="store_true", help="workload needs a GPU") + r.add_argument("--no-scale", action="store_true", help="keep it small (non-scalable)") + r.add_argument("--subject", default="spiffe://sourceos/agent/dev", help="the requesting subject SPIFFE id") + r.add_argument("--apply", action="store_true", help="actually execute (local subprocess / kubectl apply)") + r.add_argument("--dry", action="store_true", help="only decide placement; dispatch nothing") + r.set_defaults(func=cmd_run) + + m = sub.add_parser("mesh", help="show what's live in the mesh") + m.set_defaults(func=cmd_mesh) + + pl = sub.add_parser("place", help="where would this land? (dry, no dispatch)") + for a, kw in (("--command", {"default": "true"}), ("--name", {"default": "workload"}), + ("--image", {"default": ""}), ("--effect", {"default": "compute"}), + ("--sensitivity", {"default": "normal"}), ("--subject", {"default": "spiffe://sourceos/agent/dev"})): + pl.add_argument(a, **kw) + pl.add_argument("--gpu", action="store_true") + pl.add_argument("--no-scale", action="store_true") + pl.add_argument("--apply", action="store_true", help=argparse.SUPPRESS) + pl.set_defaults(func=cmd_place) + + c = sub.add_parser("commons", help="the reproducible knowledge commons") + c.set_defaults(func=cmd_commons) + return p + + +def main(argv=None) -> int: + args = build_parser().parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/test_executor.py b/tools/test_executor.py new file mode 100644 index 0000000..cd51df2 --- /dev/null +++ b/tools/test_executor.py @@ -0,0 +1,150 @@ +#!/usr/bin/env python3 +"""Tests for the execution spine. Two things matter most: (1) dispatch is real — a local subprocess +actually runs, a k8s Job manifest is actually well-formed and Grant-labelled; (2) it is fail-closed +on the Grant — a tampered / mis-sessioned / wrong-effect Grant dispatches NOTHING. Plus the full +place->grant->verify->execute spine end to end.""" +import executor as ex +import mcp_a2a_grant as g +import mesh_telemetry as mt + +KEY = b"exec-test-key" +SIGNER = g.hmac_signer(KEY) +VERIFIER = g.hmac_verifier(KEY) +AUM = "sha256:" + "ab" * 32 +CAPDIG = "sha256:" + "cd" * 32 +BINDING = {"spiffe_id": "spiffe://sourceos/agent/x", "aum_digest": AUM, "session_id": "sess_exec1"} +CAP = {"kind": "mcp_tool", "capability_ref": "capd://caps.x", "capability_digest": CAPDIG, "effect": "exec"} +ATT = g.attestation_bundle(spiffe_id="spiffe://sourceos/agent/x", aum_digest=AUM, + tpm_valid=True, cosign_valid=True) + + +def _decision(backend): + return {"placement": "scheduled", "backend": backend, "backend_trust": "trusted", + "receipt_digest": "sha256:" + "de" * 32} + + +def _grant(decision, effect="exec"): + return g.issue_grant(binding=BINDING, capability={**CAP, "effect": effect}, decision=decision, + attestation=ATT, constraints={"ops_allow": ["exec.run"]}, signer=SIGNER) + + +# ── dispatch is real ───────────────────────────────────────────────────────────────── +def test_local_adapter_runs_a_real_subprocess(): + d = _decision("local") + res = ex.execute({"command": "echo hello-mesh", "effect": "exec"}, d, _grant(d, "exec"), + session_id="sess_exec1", verifier=VERIFIER, apply=True) + assert res["dispatch"]["applied"] and res["dispatch"]["exit_code"] == 0 + assert "hello-mesh" in res["dispatch"]["stdout"] + assert res["receipt"]["receipt_digest"].startswith("sha256:") + + +def test_local_adapter_dry_run_plans_but_does_not_execute(): + d = _decision("local") + res = ex.execute({"command": "echo x", "effect": "exec"}, d, _grant(d, "exec"), + session_id="sess_exec1", verifier=VERIFIER, apply=False) + assert res["dispatch"]["applied"] is False and res["dispatch"]["planned"] == "echo x" + + +def test_k8s_adapter_emits_a_valid_grant_labelled_job(): + d = _decision("k8s") + grant = _grant(d, "compute") + res = ex.execute({"name": "trainer", "image": "img:1", "command": "python x.py", + "needs_gpu": True, "resource": {"cpu": 2, "mem": "4Gi"}, "effect": "compute"}, + d, grant, session_id="sess_exec1", verifier=VERIFIER, apply=False) + m = res["dispatch"]["manifest"] + assert m["apiVersion"] == "batch/v1" and m["kind"] == "Job" + c = m["spec"]["template"]["spec"]["containers"][0] + assert c["image"] == "img:1" and c["command"] == ["python", "x.py"] + assert c["resources"]["limits"]["nvidia.com/gpu"] == "1" + assert m["metadata"]["labels"]["sourceos.io/grant-id"] == grant["grant_id"] + assert m["spec"]["template"]["spec"]["restartPolicy"] == "Never" + + +def test_descriptor_adapter_emits_a_backend_specific_descriptor(): + d = _decision("hpc-slurm") + res = ex.execute({"command": "srun train", "effect": "compute"}, d, _grant(d, "compute"), + session_id="sess_exec1", verifier=VERIFIER) + assert res["dispatch"]["kind"] == "hpc-slurm" + assert res["dispatch"]["descriptor"]["backend"] == "hpc-slurm" + assert res["dispatch"]["descriptor"]["executor_ref"] == "node://hpc-slurm" + + +# ── fail-closed on the Grant ───────────────────────────────────────────────────────── +def test_dispatch_refused_on_tampered_grant(): + d = _decision("local") + grant = _grant(d, "exec") + grant["capability"]["effect"] = "egress" # tamper -> signature breaks + try: + ex.execute({"command": "echo x", "effect": "egress"}, d, grant, + session_id="sess_exec1", verifier=VERIFIER, apply=True) + assert False, "should have refused" + except ex.DispatchRefused as e: + assert "tamper" in str(e).lower() or "signature" in str(e).lower() + + +def test_dispatch_refused_when_effect_not_granted(): + d = _decision("local") + grant = _grant(d, "read") # granted read... + try: + ex.execute({"command": "echo x", "effect": "exec"}, d, grant, # ...but asking exec + session_id="sess_exec1", verifier=VERIFIER, apply=True) + assert False + except ex.DispatchRefused as e: + assert "effect" in str(e) + + +def test_dispatch_refused_on_wrong_session(): + d = _decision("local") + try: + ex.execute({"command": "echo x", "effect": "exec"}, d, _grant(d, "exec"), + session_id="sess_WRONG", verifier=VERIFIER, apply=True) + assert False + except ex.DispatchRefused as e: + assert "session" in str(e) + + +def test_dispatch_refused_for_unknown_backend(): + d = _decision("quantum-foo") + try: + ex.execute({"command": "echo x", "effect": "exec"}, d, _grant(d, "exec"), + session_id="sess_exec1", verifier=VERIFIER) + assert False + except ex.DispatchRefused as e: + assert "adapter" in str(e) + + +# ── the whole spine ────────────────────────────────────────────────────────────────── +def test_run_spine_end_to_end_places_grants_verifies_and_dispatches(): + reg = mt.MeshRegistry() + reg.heartbeat("slurm", "hpc-slurm", 100) + reg.heartbeat("boinc", "volunteer-boinc", 500) # bigger, but untrusted + out = ex.run_spine( + {"name": "t", "sensitivity": "sensitive", "scalable": True, "needs_gpu": True, + "effect": "compute", "command": "python t.py"}, + {"require_attestation": True}, registry=reg, binding=BINDING, + capability={**CAP, "effect": "compute"}, attestation=ATT, + constraints={"ops_allow": ["exec.run"]}, signer=SIGNER, verifier=VERIFIER) + assert out["status"] == "ran" + assert out["backend"] == "hpc-slurm" # sensitive+gpu -> trusted HPC, never the volunteer grid + assert out["execution"]["dispatch"]["kind"] == "hpc-slurm" + assert out["execution"]["receipt"]["receipt_digest"].startswith("sha256:") + + +def test_run_spine_blocks_fail_closed_and_dispatches_nothing(): + reg = mt.MeshRegistry() + reg.heartbeat("boinc", "volunteer-boinc", 500) # only an untrusted backend is up + out = ex.run_spine( + {"sensitivity": "sensitive", "scalable": True, "effect": "compute"}, + {}, registry=reg, binding=BINDING, capability={**CAP, "effect": "compute"}, + attestation=ATT, constraints={}, signer=SIGNER, verifier=VERIFIER) + assert out["status"] == "blocked" and out["decision"]["backend"] is None + assert "execution" not in out # nothing was dispatched + + +if __name__ == "__main__": + import sys + fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + for fn in fns: + fn() + print(f"ok: {len(fns)} executor spine tests passed") + sys.exit(0) diff --git a/tools/test_sourceosctl.py b/tools/test_sourceosctl.py new file mode 100644 index 0000000..b84a093 --- /dev/null +++ b/tools/test_sourceosctl.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""Tests for sourceosctl — the developer front door. Exercises the run_workload core (dry placement, +full spine dispatch, fail-closed block) against an isolated mesh, plus the parser and read-only +subcommands.""" +import tempfile + +import mesh_telemetry as mt +import sourceosctl as ctl + +KEY = b"ctl-test-key" + + +def _seed(td): + mt.write_heartbeat(td, "slurm", "hpc-slurm", 100) + mt.write_heartbeat(td, "k8s1", "k8s", 10) + + +def test_run_workload_dry_places_without_dispatch(): + with tempfile.TemporaryDirectory() as td: + _seed(td) + out = ctl.run_workload(name="t", command="python x.py", effect="compute", + sensitivity="sensitive", scalable=True, gpu=True, image="", + subject="spiffe://sourceos/agent/a", heartbeats_dir=td, key=KEY, dry=True) + assert out["status"] == "placed" and out["decision"]["backend"] == "hpc-slurm" + + +def test_run_workload_full_spine_dispatches_and_seals(): + with tempfile.TemporaryDirectory() as td: + _seed(td) + out = ctl.run_workload(name="t", command="echo hi", effect="compute", sensitivity="normal", + scalable=True, gpu=False, image="img:1", + subject="spiffe://sourceos/agent/a", heartbeats_dir=td, key=KEY) + assert out["status"] == "ran" + assert out["execution"]["receipt"]["receipt_digest"].startswith("sha256:") + assert out["backend"] in ("hpc-slurm", "k8s") + + +def test_run_workload_blocks_fail_closed_for_sensitive_on_untrusted_only(): + with tempfile.TemporaryDirectory() as td: + mt.write_heartbeat(td, "boinc", "volunteer-boinc", 500) # only untrusted up + out = ctl.run_workload(name="t", command="x", effect="compute", sensitivity="sensitive", + scalable=True, gpu=False, image="", + subject="spiffe://sourceos/agent/a", heartbeats_dir=td, key=KEY) + assert out["status"] == "blocked" + + +def test_parser_builds_run_subcommand(): + args = ctl.build_parser().parse_args(["run", "--command", "echo hi", "--gpu", "--sensitivity", "sensitive"]) + assert args.gpu and args.command == "echo hi" and args.sensitivity == "sensitive" + + +def test_readonly_subcommands_exit_clean(): + assert ctl.cmd_mesh(None) == 0 + assert ctl.cmd_commons(None) == 0 + + +if __name__ == "__main__": + import sys + fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + for fn in fns: + fn() + print(f"ok: {len(fns)} sourceosctl tests passed") + sys.exit(0) diff --git a/tools/validate.py b/tools/validate.py index a8b17fe..1103dbc 100644 --- a/tools/validate.py +++ b/tools/validate.py @@ -29,6 +29,8 @@ "tools/mcp_a2a_grant.py", "tools/commons.py", "tools/mcp_ops_server.py", + "tools/executor.py", + "tools/sourceosctl.py", ] CAPD_KEYS = ("capability_id", "kind", "status", "links", "composes_with", "policy") # Every CapD in capd/ must carry the core keys and parse — not just the flagship control-plane one.