diff --git a/.gitignore b/.gitignore index 5febcc0..0f65165 100644 --- a/.gitignore +++ b/.gitignore @@ -15,6 +15,9 @@ node_modules/ # Promotion-gate runtime evidence (emitted at rollout, not committed; dir kept via .gitkeep) artifacts/gate-decisions/*.json +# Live mesh telemetry heartbeats (emitted by nodes at runtime, not committed; dir kept via .gitkeep) +artifacts/mesh-heartbeats/*.json + # Porter control-plane build outputs (compiled locally, not committed) artifacts/porter-shim/shim artifacts/cloudshell-hardened-pack/culler/culler diff --git a/Makefile b/Makefile index ded63fc..8fa51dc 100644 --- a/Makefile +++ b/Makefile @@ -1,10 +1,32 @@ # 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 rollout promotion-gate +.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute validate: ## repo hygiene + CapD validity python3 tools/validate.py +tools-test: ## unit-test the governed tools (portal router, compute plane, promotion gate, MCP surface) + cd tools && python3 -m pytest -q + +portal: ## developer portal: read-only web console over the governed surface (scale-to-zero, stdlib only) + python3 tools/portal_server.py $(PORT) + +compute: ## compute plane: route one workload across the mesh under per-project policy + availability + python3 tools/compute_plane.py + +mesh-demo: ## seed a live demo mesh (heartbeats) so the portal shows live telemetry + @python3 tools/mesh_telemetry.py heartbeat artifacts/mesh-heartbeats k8s-a k8s 8 >/dev/null + @python3 tools/mesh_telemetry.py heartbeat artifacts/mesh-heartbeats slurm-login hpc-slurm 120 >/dev/null + @python3 tools/mesh_telemetry.py heartbeat artifacts/mesh-heartbeats edge-1 wasm-edge 20 >/dev/null + @python3 tools/mesh_telemetry.py heartbeat artifacts/mesh-heartbeats boinc-grid volunteer-boinc 400 >/dev/null + @python3 tools/mesh_telemetry.py view artifacts/mesh-heartbeats + +grant: ## demo the zero-trust attach flow: Attest -> Decide -> Grant -> verify-at-node + python3 tools/mcp_a2a_grant.py + +commons: ## reproducible knowledge commons: ingest the estate's CapDs + workloads as citable records + python3 tools/commons.py + 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/artifacts/mesh-heartbeats/.gitkeep b/artifacts/mesh-heartbeats/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/capd/cloudshell-fog.capd.json b/capd/cloudshell-fog.capd.json new file mode 100644 index 0000000..16cac7a --- /dev/null +++ b/capd/cloudshell-fog.capd.json @@ -0,0 +1,49 @@ +{ + "capability_id": "caps.compute.cloudshell-fog@0.1.0", + "kind": "compute.zero-trust-fog", + "status": "experimental", + "name": "Sovereign Agentic Cloud-Shell — zero-trust MCP-A2A fog deployment", + "description": "The full Attest -> Decide -> Grant -> verify-at-node -> enforce-constraints flow from the cloud-shell fog spec, made first-class over the continuum. A low-mem box starts a session; the Control-Plane Agent (compute_plane.place) decides a fog node from live HyperSwarm discovery (mesh_telemetry); the MCP-A2A Policy Authority attests (TPM/TEE + cosign) then mints a signed, session-bound Grant carrying its constraints (mcp_a2a_grant.issue_grant); the fog node's Policy Gate re-verifies that Grant on attach and on every PTY/FS op (mcp_a2a_grant.verify_grant). Fail-closed at every step; every decision sealed to the append-only ledger.", + "links": { + "control_plane_agent": "tools/compute_plane.py", + "hyperswarm_discovery": "tools/mesh_telemetry.py", + "policy_authority": "tools/mcp_a2a_grant.py", + "policy_gate": "tools/mcp_a2a_grant.py", + "capability_registry": "capd/", + "agent_surface": "tools/mcp_ops_server.py", + "portal": "tools/portal_server.py", + "supply_chain_gate": "prophet-platform/tools/advisory_check.py + attest.py (estate)", + "ledger": "artifacts/gate-decisions + mcp-receipts (sealed)", + "spec_witness": "docs/CLOUDSHELL_FOG.md", + "shape_authority": "SourceOS-Linux/mcp-a2a-zero-trust (canonical Grant/AttestationBundle/QuorumProof; vendored hash-pinned at schemas/a2a/)" + }, + "composes_with": { + "compute_plane": "caps.compute.mesh-plane@0.1.0", + "inner_loop": "caps.dev.devspace-inner-loop@0.1.0", + "control_plane": "caps.infra.paas.continuum-local@0.1.0", + "scales_up_to": "caps.infra.cluster-scaleup.hyperswarm@0.1.0" + }, + "policy": { + "zero_trust": true, + "attest_decide_grant": true, + "fail_closed": true, + "evidence_emitting": true, + "session_bound_grants": true, + "quorum_when_required": true, + "read_only_portal": true + }, + "flow": [ + "0 login / refresh (IdP OIDC/OAuth2 + optional FIDO2)", + "1 start shell / agent session (Browser Terminal UI over WSS/WebTransport)", + "2 SessionRequest: capability + agent (Console API -> Control-Plane Agent)", + "3 resolve CapD + MCP tool schemas (Capability Registry)", + "4 verify artifact: OCI digest + SBOM + signature (Supply-chain Gate)", + "5 find candidate nodes (HyperSwarm Mesh discovery / mesh_telemetry)", + "6 route to selected node", + "7 policy eval: user + node + capability (MCP-A2A Policy Authority)", + "8 issue Grant, session-bound; +QuorumProof when required (Key Authority HSM/KMS)", + "9 TriRPC attach over a Grant-bound channel (Edge Gateway <-> Fog Node)", + "10 authorize attach (fog-node MCP-A2A Policy Gate: verify Grant)", + "11 allow PTY/FS ops, enforce constraints (Shell Runtime Container/Pod sandbox)" + ] +} diff --git a/capd/compute-plane.mesh.capd.json b/capd/compute-plane.mesh.capd.json new file mode 100644 index 0000000..48330f7 --- /dev/null +++ b/capd/compute-plane.mesh.capd.json @@ -0,0 +1,27 @@ +{ + "capability_id": "caps.compute.mesh-plane@0.1.0", + "kind": "compute.placement-plane", + "status": "experimental", + "name": "Compute plane — one governed door to any substrate", + "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", + "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" + }, + "composes_with": { + "inner_loop": "caps.dev.devspace-inner-loop@0.1.0", + "control_plane": "caps.infra.paas.continuum-local@0.1.0", + "scales_up_to": "caps.infra.cluster-scaleup.hyperswarm@0.1.0" + }, + "policy": { + "fail_closed": true, + "evidence_emitting": true, + "sensitive_never_untrusted": true, + "per_project_config": true, + "per_account_config": true, + "read_only_portal": true + }, + "backends": ["local", "k8s", "hpc-slurm", "wasm-edge", "p2p-mesh", "volunteer-boinc", "blockchain-rlc"] +} diff --git a/capd/devspace.local-dev.capd.json b/capd/devspace.local-dev.capd.json new file mode 100644 index 0000000..c688288 --- /dev/null +++ b/capd/devspace.local-dev.capd.json @@ -0,0 +1,24 @@ +{ + "capability_id": "caps.dev.devspace-inner-loop@0.1.0", + "kind": "dev.inner-loop", + "status": "experimental", + "name": "SourceOS DevSpace — inner-loop dev-environment", + "description": "Nocalhost-style inner-loop over the continuum: an isolated per-user namespace, hot code sync into a running workload, a sidecar dev container, and port-forward + local debug — shortening the modify->result feedback loop. Consumes the control plane; surfaced to humans via the portal and to agents via the MCP ops surface (one governed source, two views).", + "links": { + "portal": "tools/portal_server.py", + "agent_surface": "tools/mcp_ops_server.py", + "reference_pattern": "Nocalhost DevSpace (inner-loop) — met and bettered: sovereign + evidence-emitting" + }, + "composes_with": { + "control_plane": "caps.infra.paas.continuum-local@0.1.0", + "operator_surface": "SourceOS-Linux/sourceos-devtools", + "compute_plane": "caps.compute.mesh-plane@0.1.0", + "scales_up_to": "caps.infra.cluster-scaleup.hyperswarm@0.1.0" + }, + "policy": { + "local_first": true, + "evidence_emitting": true, + "isolation": "per-user-namespace", + "read_only_portal": true + } +} diff --git a/capd/knowledge-commons.mesh.capd.json b/capd/knowledge-commons.mesh.capd.json new file mode 100644 index 0000000..12b92ea --- /dev/null +++ b/capd/knowledge-commons.mesh.capd.json @@ -0,0 +1,30 @@ +{ + "capability_id": "caps.knowledge.commons@0.1.0", + "kind": "knowledge.reproducible-commons", + "status": "experimental", + "name": "Reproducible Knowledge Commons", + "description": "A Zenodo-style citable, content-addressed, reproducible, reusable layer over the whole estate. Every capability, workload, model, dataset, or semantic action becomes a commons record with a DOI-like id (commons:/@+), its reproducibility provenance, and its reuse history. Folds three demonstrated systems into one governed plane: Zenodo/reproducible-fusion (content-addressed + citable + reproducibility-graded, fail-closed on the claim), ARM / Asset Reuse Manager (Domain->Category->Asset + recommendation + use/evaluate feedback), and the Semantic API/ontology (records carry a declarative semantic_action). Ingests the estate's own CapDs + suite workloads, so the compute-mesh / cloud-shell fog work is itself first-class citable here.", + "links": { + "engine": "tools/commons.py", + "portal": "tools/portal_server.py", + "agent_surface": "tools/mcp_ops_server.py", + "ledger": "artifacts/gate-decisions + mcp-receipts (sealed)", + "spec_witness": "docs/KNOWLEDGE_COMMONS.md", + "reference_pattern": "IBM Asset Reuse Manager (ARM) + MLOps 3-phase reproducible pipeline + Semantic API/ontology — met and bettered: one governed, content-addressed, reproducibility-graded commons instead of three separate systems" + }, + "composes_with": { + "compute_plane": "caps.compute.mesh-plane@0.1.0", + "fog": "caps.compute.cloudshell-fog@0.1.0", + "control_plane": "caps.infra.paas.continuum-local@0.1.0", + "scales_up_to": "caps.infra.cluster-scaleup.hyperswarm@0.1.0" + }, + "policy": { + "content_addressed": true, + "citable": true, + "reproducibility_gate": true, + "arm_reuse_feedback": true, + "semantic_action_catalog": true, + "evidence_emitting": true, + "read_only_portal": true + } +} diff --git a/docs/CLOUDSHELL_FOG.md b/docs/CLOUDSHELL_FOG.md new file mode 100644 index 0000000..407b21f --- /dev/null +++ b/docs/CLOUDSHELL_FOG.md @@ -0,0 +1,58 @@ +# Sovereign Agentic Cloud-Shell — fog deployment, made first-class + +This witnesses the cloud-shell fog spec (*Zero-Trust MCP-A2A + TriRPC transport + HyperSwarm +placement*) against its implementation in this repo. Every box in the diagram points at real code — +no box is decoration. + +## Planes → implementation + +| Spec plane / box | Implementation | Status | +|---|---|---| +| **Capability Build / Packaging** — OCI image (digest+SBOM+sig), publish CapD + tool schema | estate build + `attest.py` (SLSA/in-toto/DSSE), `capd/*.capd.json` | estate | +| **Capability Registry** (CapDs + MCP tool catalog) | `capd/` + `tools/mcp_ops_server.py` | ✅ | +| **Control Plane Agent** (placement + quotas + admission; session lifecycle) | `tools/compute_plane.py` — `place()` is the *Decide* | ✅ | +| **HyperSwarm Mesh** (discovery + gossip + rendezvous) / *find candidate nodes* | `tools/mesh_telemetry.py` — live liveness registry | ✅ | +| **Supply-chain Gate** (OCI digest + SBOM + signatures) | `prophet-platform/tools/advisory_check.py` + `attest.py` | estate | +| **MCP-A2A Policy Authority** (Attest → Decide → Grant, +QuorumProof) | `tools/mcp_a2a_grant.py` — `issue_grant()` | ✅ | +| **Key Authority** (HSM/KMS — sign grants / derive session keys) | `hmac_signer` interface in `mcp_a2a_grant.py` (swap in HSM/ed25519) | ✅ iface | +| **fog-node MCP-A2A Policy Gate** (verify Grant + bind to session + streams) | `tools/mcp_a2a_grant.py` — `verify_grant()` | ✅ | +| **Shell Runtime Container/Pod** (PTY + FS sandbox) — enforce constraints | gate returns per-op authorization + redactions; runtime sandbox = estate | ✅ policy | +| **Attestation Verifier / Node Attestor** (TPM/TEE + cosign) | attestation inputs consumed fail-closed by `issue_grant` | ✅ iface | +| **Observability Sink** (OTEL) | estate OTel collector (`prophet-platform/deploy/superiority-march/observability`) | estate | +| **Ledger / Audit** (append-only evidence) | sealed receipts → `artifacts/{gate-decisions,mcp-receipts}` | ✅ | +| **Browser Terminal UI / Console API** | portal (read-only view) `tools/portal_server.py`; mutations via MCP surface | ✅ view | + +## The numbered flow (0–11) + +`0` login/refresh (IdP) → `1` start session (UI) → `2` SessionRequest {capability, agent} → +`3` resolve CapD + tool schemas (registry) → `4` verify artifact digest/SBOM/sig (supply-chain gate) → +`5` **find candidate nodes** (HyperSwarm / `mesh_telemetry`) → `6` route to selected node → +`7` **policy eval** user+node+capability (`place()` Decide) → `8` **issue Grant** session-bound, ++QuorumProof when required (`issue_grant`) → `9` TriRPC attach over the Grant-bound channel → +`10` **authorize attach** (`verify_grant`) → `11` **allow PTY/FS ops**, enforce constraints +(`verify_grant(requested_op=...)`). + +## Why this is zero-trust, not just RBAC + +A placement decision is **not** permission to run. The node re-verifies the Grant itself — signature, +session binding, expiry, attestation, and that the *specific* op is inside the granted constraints — +on attach **and on every PTY/FS op**. A missing, stale, tampered, or over-reaching Grant is denied, +fail-closed. Sensitive workloads never even get a Grant for an untrusted (volunteer/p2p/blockchain) +node — the Decide stage refuses them first (`compute_plane`), and the Authority refuses to mint a +Grant against a blocked or unattested decision. + +See `tools/test_mcp_a2a_grant.py` (Authority + Gate) and `tools/test_compute_plane.py` (Decide). + +## Shape conformance — we consume the canonical spec, we do not fork it + +The Grant, AttestationBundle, and QuorumProof that `tools/mcp_a2a_grant.py` emits/consumes are the +**canonical** shapes owned by `SourceOS-Linux/mcp-a2a-zero-trust` (the estate's zero-trust authority +repo — "owns the zero-trust authority model … grant request, grant decision, and grant ledger +contracts"). Those schemas are vendored here under `schemas/a2a/`, **hash-pinned** to the authority's +`schemas/index.json` (see `schemas/a2a/PROVENANCE.md`), and `test_mcp_a2a_grant.py` validates every +emitted Grant / QuorumProof / AttestationBundle against them. So `issue_grant()` produces a `Grant` +with the canonical `binding` / `capability` / `constraints` / `policy_hash` / `quorum_proof` / +`evidence_refs` / `sig{issuer,sig}`, and `verify_grant()` returns a canonical `tool_grant.validate` +result `{valid, expired, revoked, reason}` — identical to `examples/grant.example.json` and +`examples/tool_grant_check.example.json` in the authority repo. If the authority updates a schema, +re-vendor and re-pin; the authority stays the source of truth for the shape. diff --git a/docs/KNOWLEDGE_COMMONS.md b/docs/KNOWLEDGE_COMMONS.md new file mode 100644 index 0000000..67e201c --- /dev/null +++ b/docs/KNOWLEDGE_COMMONS.md @@ -0,0 +1,35 @@ +# Reproducible Knowledge Commons — three demonstrated systems, one governed plane + +The commons folds the three systems those diagrams show — IBM's **Asset Reuse Manager**, the 3-phase +**MLOps** reproducible pipeline, and the **Semantic API / ontology** — into one content-addressed, +citable, reproducibility-graded plane (`tools/commons.py`). It ingests the estate's own CapDs and +suite workloads, so the compute-mesh / cloud-shell fog work is itself first-class and citable here. + +## System → implementation + +| Demonstrated system (diagram) | In the commons | +|---|---| +| **Zenodo / reproducible-fusion commons** — versioned, citable, DOI-like, reproducible | `mint_id()` → content-addressed `commons:/@+`; every record carries a `cite` string | +| **ARM: Domain → Category → Asset** | every record has `domain` / `category` / `asset_type`; `search()` navigates them | +| **ARM: Recommendation** | `recommend()` ranks by reuse score, within a domain | +| **ARM: Use / Evaluate → Feedback** | `record_use(id, outcome)` raises `reuse.score = evaluations / uses` | +| **MLOps: reproducible pipeline + model registry + monitoring** | the **reproducibility gate**: a record is `reproducible` only if `provenance` carries a `source_digest` **and** an `attestation_ref` or `sbom_digest` — else honestly `declared` | +| **Semantic API: declarative action + ontology constraints + executor** | a record may carry a `semantic_action` (signature + policy/ontology constraints); the suite workloads ingest with their governing policy as the semantic action | + +## The reproducibility gate is fail-closed on the *claim* + +This is the point that makes it a *reproducible* commons and not just a catalog: you cannot mint a +record that claims `reproducible` unless the provenance actually carries what you would need to +reproduce it. A model deposited with only weights is `declared`; a model deposited with its +`source_digest` + a cosign/SLSA `attestation_ref` is `reproducible`. The portal shows the split +honestly (`N reproducible` of the total), so the dashboard never overstates. + +## The estate ingests itself + +`estate_commons(root)` deposits every `capd/*.capd.json` and every `mesh/suite-workloads.json` entry. +So `caps.compute.mesh-plane`, `caps.compute.cloudshell-fog`, `caps.dev.devspace-inner-loop`, and the +five app-suite workloads (Noetica, memory-mesh, TurtleTern, Goose Notes, BearBrowser) are all citable +commons records. The cloud-shell fog capability ingests as `reproducible` because its CapD references +the sealed ledger + the mcp-a2a-zero-trust shape authority — provenance enough to reproduce the claim. + +See `tools/test_commons.py`. Surfaced read-only at the portal's `/api/commons`. diff --git a/mesh/suite-workloads.json b/mesh/suite-workloads.json new file mode 100644 index 0000000..7768dc5 --- /dev/null +++ b/mesh/suite-workloads.json @@ -0,0 +1,41 @@ +{ + "schema": "sourceos-continuum.suite-workloads.v1", + "note": "The app suite as first-class citizens of the one governed compute mesh. Each product declares its characteristic workload and the policy that governs where the compute plane may place it. A low-mem box runs the front-end; the heavy work scales out over the mesh under these policies. The dashboard runs place() for each of these against LIVE telemetry so you can see, right now, where each product's work would land.", + "workloads": [ + { + "id": "noetica.reasoning", + "product": "Noetica", + "description": "Desktop reasoning app offloads heavy graph/LLM reasoning off the local box to the mesh. The reasoning corpus is the user's own — sensitive — so it may only run on trusted infra, attested.", + "workload": {"sensitivity": "sensitive", "scalable": true, "needs_gpu": true}, + "policy": {"forbid_untrusted_for_sensitive": true, "require_attestation": true} + }, + { + "id": "memory-mesh.holographic-recall", + "product": "memory-mesh", + "description": "The distributed memory / holographic-message fabric: embedding, recall, and holographic ops over the estate's memory. It holds memory, so it is sensitive and trusted-only + attested; it both provides nodes to the mesh and consumes it.", + "workload": {"sensitivity": "sensitive", "scalable": true, "needs_gpu": true}, + "policy": {"forbid_untrusted_for_sensitive": true, "require_attestation": true, "prefer": ["k8s", "hpc-slurm"]} + }, + { + "id": "turtletern.trinomial-turbulence", + "product": "TurtleTern", + "description": "The trinomial regime-aware pricer + Lorenz low-dim turbulence engine: embarrassingly-parallel numeric simulation over public math, no private data. The poster child for volunteer/HPC scale-out — free to burst onto the biggest grid the mesh offers.", + "workload": {"sensitivity": "normal", "scalable": true, "needs_gpu": true}, + "policy": {"forbid_untrusted_for_sensitive": true} + }, + { + "id": "goosenotes.index", + "product": "Goose Notes", + "description": "Personal-notes embedding + reindex: local-first and small most of the time, but a full reindex can batch-scale. Notes are personal, so trusted-only.", + "workload": {"sensitivity": "sensitive", "scalable": true, "needs_gpu": false}, + "policy": {"forbid_untrusted_for_sensitive": true, "prefer": ["wasm-edge", "k8s"]} + }, + { + "id": "bearbrowser.scan", + "product": "BearBrowser", + "description": "Offensive crawl/scan workloads. Not private data, but running attack tooling on volunteer/p2p boxes is a liability — governed onto trusted infra only, never the volunteer grid, regardless of scale pressure.", + "workload": {"sensitivity": "normal", "scalable": true, "needs_gpu": false}, + "policy": {"allowed_backends": ["local", "k8s", "hpc-slurm", "wasm-edge"], "forbid_untrusted_for_sensitive": true} + } + ] +} diff --git a/schemas/a2a/PROVENANCE.md b/schemas/a2a/PROVENANCE.md new file mode 100644 index 0000000..edb4baa --- /dev/null +++ b/schemas/a2a/PROVENANCE.md @@ -0,0 +1,18 @@ +# A2A zero-trust schemas — vendored, pinned by hash + +These are the **canonical** MCP-A2A zero-trust contracts, owned by +`SourceOS-Linux/mcp-a2a-zero-trust` (the estate's zero-trust authority repo). They are vendored here +so continuum's `tools/mcp_a2a_grant.py` conforms to the canonical shape and its conformance is +testable in-repo, without a cross-repo build dependency. + +Pinned by sha256 (matches the authority's `schemas/index.json` — verified on vendor): + +| file | title | sha256 | +|---|---|---| +| `grant.schema.json` | Grant | `sha256:2aac20b5fc9ce2ef72c0609bc1687f2b4b17a2167ef3148fa8ad3c4c1494f0b1` | +| `quorum_proof.schema.json` | QuorumProof | `sha256:d3ceec20d3268c30c1f0fda17f0981654a850ad39073ac5b1ff4aff62a0b2bb2` | +| `attestation_bundle.schema.json` | AttestationBundle | `sha256:485d0ed689cc1b3184a18d555bb3eba75c4110f7087d0c3c6c54c575447a4272` | +| `runtime_evidence_refs.schema.json` | RuntimeEvidenceRefs | (governance/) | + +**Do not hand-edit.** If the authority updates a schema, re-vendor and update the pins. `mcp-a2a-zero-trust` +remains the source of truth for the shape; continuum consumes it, it does not fork it. diff --git a/schemas/a2a/attestation_bundle.schema.json b/schemas/a2a/attestation_bundle.schema.json new file mode 100644 index 0000000..099ce9e --- /dev/null +++ b/schemas/a2a/attestation_bundle.schema.json @@ -0,0 +1,67 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://sourceos.local/schemas/canonical/attestation_bundle.schema.json", + "title": "AttestationBundle", + "type": "object", + "required": [ + "subject", + "results", + "evidence_refs" + ], + "properties": { + "subject": { + "type": "object", + "required": [ + "spiffe_id", + "aum_digest" + ], + "properties": { + "spiffe_id": { + "type": "string", + "minLength": 1 + }, + "aum_digest": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + } + }, + "additionalProperties": false + }, + "results": { + "type": "object", + "required": [ + "tpm_valid", + "cosign_valid" + ], + "properties": { + "tpm_valid": { + "type": "boolean" + }, + "cosign_valid": { + "type": "boolean" + }, + "fido2_valid": { + "type": "boolean", + "default": false + } + }, + "additionalProperties": false + }, + "evidence_refs": { + "type": "object", + "properties": { + "tpm_quote_ref": { + "type": "string" + }, + "cosign_bundle_ref": { + "type": "string" + }, + "fido2_attest_ref": { + "type": "string" + } + }, + "additionalProperties": false + } + }, + "additionalProperties": false +} diff --git a/schemas/a2a/grant.schema.json b/schemas/a2a/grant.schema.json new file mode 100644 index 0000000..4e59d31 --- /dev/null +++ b/schemas/a2a/grant.schema.json @@ -0,0 +1,143 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://sourceos.local/schemas/canonical/grant.schema.json", + "title": "Grant", + "type": "object", + "required": [ + "grant_id", + "issued_at", + "expires_at", + "binding", + "capability", + "constraints", + "policy_hash" + ], + "properties": { + "grant_id": { + "type": "string", + "minLength": 8 + }, + "issued_at": { + "type": "string", + "format": "date-time" + }, + "expires_at": { + "type": "string", + "format": "date-time" + }, + "binding": { + "type": "object", + "required": [ + "spiffe_id", + "aum_digest" + ], + "properties": { + "spiffe_id": { + "type": "string", + "minLength": 1 + }, + "aum_digest": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "session_id": { + "type": "string", + "minLength": 6 + } + }, + "additionalProperties": false + }, + "capability": { + "type": "object", + "required": [ + "kind", + "capability_ref", + "capability_digest", + "effect" + ], + "properties": { + "kind": { + "type": "string", + "enum": [ + "mcp_tool", + "a2a_skill", + "deployment", + "runner_action" + ] + }, + "capability_ref": { + "type": "string", + "minLength": 1 + }, + "capability_digest": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "server": { + "type": "string" + }, + "tool": { + "type": "string" + }, + "agent_id": { + "type": "string" + }, + "skill": { + "type": "string" + }, + "deployment_id": { + "type": "string" + }, + "operation": { + "type": "string" + }, + "executor_ref": { + "type": "string" + }, + "effect": { + "type": "string", + "enum": [ + "read", + "write", + "compute", + "exec", + "egress" + ] + } + }, + "additionalProperties": false + }, + "constraints": { + "type": "object" + }, + "policy_hash": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "quorum_proof": { + "$ref": "quorum_proof.schema.json" + }, + "evidence_refs": { + "$ref": "https://socioprophet.dev/schema/governance/runtime_evidence_refs.schema.json" + }, + "sig": { + "type": "object", + "required": [ + "issuer", + "sig" + ], + "properties": { + "issuer": { + "type": "string", + "minLength": 1 + }, + "sig": { + "type": "string", + "minLength": 16 + } + }, + "additionalProperties": false + } + }, + "additionalProperties": false +} diff --git a/schemas/a2a/quorum_proof.schema.json b/schemas/a2a/quorum_proof.schema.json new file mode 100644 index 0000000..97f8d69 --- /dev/null +++ b/schemas/a2a/quorum_proof.schema.json @@ -0,0 +1,60 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://sourceos.local/schemas/canonical/quorum_proof.schema.json", + "title": "QuorumProof", + "type": "object", + "required": [ + "rule", + "validators", + "signed_payload_hash", + "signatures" + ], + "properties": { + "rule": { + "type": "string", + "minLength": 1 + }, + "validators": { + "type": "array", + "minItems": 1, + "items": { + "type": "string", + "minLength": 1 + } + }, + "signed_payload_hash": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "signatures": { + "type": "array", + "minItems": 1, + "items": { + "type": "object", + "required": [ + "kind", + "spiffe_id", + "sig" + ], + "properties": { + "kind": { + "type": "string", + "enum": [ + "human" + ] + }, + "spiffe_id": { + "type": "string", + "minLength": 1 + }, + "sig": { + "type": "string", + "minLength": 16 + } + }, + "additionalProperties": false + } + } + }, + "additionalProperties": false +} diff --git a/schemas/a2a/runtime_evidence_refs.schema.json b/schemas/a2a/runtime_evidence_refs.schema.json new file mode 100644 index 0000000..8a494ed --- /dev/null +++ b/schemas/a2a/runtime_evidence_refs.schema.json @@ -0,0 +1,42 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://socioprophet.dev/schema/governance/runtime_evidence_refs.schema.json", + "title": "RuntimeEvidenceRefs", + "description": "Reference set that lets grants, policy decisions, or ledger events carry semantic-proof and HDT-summary bindings without embedding full artifacts.", + "type": "object", + "additionalProperties": false, + "properties": { + "event_ir_ref": { + "type": "string", + "minLength": 1 + }, + "event_ir_hash": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "semantic_proof_ref": { + "type": "string", + "minLength": 1 + }, + "semantic_proof_hash": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "hdt_decision_ref": { + "type": "string", + "minLength": 1 + }, + "hdt_decision_hash": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + }, + "attestation_bundle_ref": { + "type": "string", + "minLength": 1 + }, + "attestation_bundle_hash": { + "type": "string", + "pattern": "^sha256:[a-f0-9]{64}$" + } + } +} diff --git a/tools/commons.py b/tools/commons.py new file mode 100644 index 0000000..8b54633 --- /dev/null +++ b/tools/commons.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +"""Reproducible Knowledge Commons — a Zenodo-style, citable, reproducible, reusable layer over the +whole estate. + +Every capability, workload, model, dataset, or semantic action becomes a **commons record**: a +content-addressed, citable deposit (a DOI-like `commons:/@+`), carrying +its reproducibility provenance and its reuse history. Three demonstrated systems, folded into one: + + * Zenodo / reproducible-fusion commons: content-addressed + citable + versioned; a record is only + marked `reproducible` if its provenance actually carries the digests to reproduce it — otherwise + it is honestly `declared`. Fail-closed on the reproducibility claim. + * ARM (Asset Reuse Manager): Domain → Category → Asset, with recommendation + a use/evaluate + feedback loop that raises a record's reuse score. + * Semantic API / ontology: a record may carry a declarative `semantic_action` (signature + + ontology constraints + executor), so the commons is also the semantic-action catalog. + +The commons ingests the estate's own CapDs and suite workloads, so the compute-mesh / cloud-shell +fog work we built is itself first-class here — citable and reusable — not a thing off to the side. +""" +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + + +def _canon(obj) -> bytes: + return json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8") + + +def mint_id(domain: str, name: str, version: str, content) -> tuple[str, str]: + """Content-addressed, citable id. Same inputs -> same id (reproducible; Zenodo-style versioning).""" + digest = hashlib.sha256(_canon({"domain": domain, "name": name, "version": version, + "content": content})).hexdigest() + return f"commons:{domain}/{name}@{version}+{digest[:12]}", "sha256:" + digest + + +def _domain_of(capability_id: str) -> str: + # caps.compute.mesh-plane@0.1.0 -> "compute"; caps.dev.x -> "dev"; caps.infra.y -> "infra" + parts = capability_id.split("@")[0].split(".") + return parts[1] if len(parts) > 1 else "misc" + + +class Commons: + """The deposit store + ARM reuse graph + semantic-action catalog.""" + + def __init__(self): + self._records: dict[str, dict] = {} + + # ── deposit (Zenodo) + reproducibility gate (MLOps provenance) ─────────────────── + def deposit(self, *, domain: str, name: str, version: str, asset_type: str, content, + category: str | None = None, provenance: dict | None = None, + semantic_action: dict | None = None, reuse: dict | None = None) -> dict: + provenance = provenance or {} + cid, content_digest = mint_id(domain, name, version, content) + # a record may only claim `reproducible` if it carries what you'd need to reproduce it: + # a source digest AND either an attestation reference or an SBOM digest. Else: `declared`. + reproducible = bool(provenance.get("source_digest")) and \ + bool(provenance.get("attestation_ref") or provenance.get("sbom_digest")) + rec = { + "commons_id": cid, + "content_digest": content_digest, + "domain": domain, "name": name, "version": version, + "asset_type": asset_type, "category": category or asset_type, + "provenance": provenance, + "reproducibility": "reproducible" if reproducible else "declared", + "semantic_action": semantic_action, + "reuse": reuse or {"uses": 0, "evaluations": 0, "score": 0.0}, + "cite": f"{name} {version}. Commons {cid}.", + } + self._records[cid] = rec + return rec + + # ── ingest the estate: its own CapDs and suite workloads become first-class records ── + def ingest_capds(self, capd_dir) -> int: + d = Path(capd_dir) + n = 0 + for f in sorted(d.glob("*.capd.json")) if d.is_dir() else []: + try: + data = json.loads(f.read_text()) + except (OSError, json.JSONDecodeError): + continue + cap_id = data.get("capability_id", f.stem) + name, _, version = cap_id.partition("@") + links = data.get("links", {}) + # reproducible iff the CapD points at a supply-chain/attestation provenance + provenance = {"source_ref": f"capd://{f.name}", "kind": data.get("kind"), + "links": links} + if any(k in links for k in ("supply_chain_gate", "shape_authority", "ledger")): + provenance["attestation_ref"] = links.get("supply_chain_gate") or links.get("ledger") + provenance["source_digest"] = "sha256:" + hashlib.sha256(f.read_bytes()).hexdigest() + self.deposit(domain=_domain_of(cap_id), name=name or f.stem, + version=version or "0.0.0", asset_type="capability", + category=data.get("kind", "capability"), content=data, + provenance=provenance) + n += 1 + return n + + def ingest_workloads(self, path) -> int: + try: + profiles = json.loads(Path(path).read_text())["workloads"] + except (OSError, json.JSONDecodeError, KeyError): + return 0 + for p in profiles: + self.deposit(domain="workload", name=p.get("id", "unknown"), version="0.1.0", + asset_type="workload", category=p.get("product", "workload"), + content=p, provenance={"source_ref": "mesh/suite-workloads.json"}, + semantic_action={"policy": p.get("policy", {}), "workload": p.get("workload", {})}) + return len(profiles) + + # ── resolve / search / ARM recommend / feedback ────────────────────────────────── + def resolve(self, commons_id: str) -> dict | None: + return self._records.get(commons_id) + + def records(self) -> list[dict]: + return list(self._records.values()) + + def search(self, *, domain=None, category=None, asset_type=None, reproducible=None) -> list[dict]: + out = [] + for r in self._records.values(): + if domain and r["domain"] != domain: + continue + if category and r["category"] != category: + continue + if asset_type and r["asset_type"] != asset_type: + continue + if reproducible is not None and (r["reproducibility"] == "reproducible") != reproducible: + continue + out.append(r) + return out + + def recommend(self, *, domain=None, asset_type=None, limit=5) -> list[dict]: + """ARM recommendation: most-reused, most-evaluated records first (optionally within a domain).""" + pool = self.search(domain=domain, asset_type=asset_type) + return sorted(pool, key=lambda r: (r["reuse"]["score"], r["reuse"]["uses"]), reverse=True)[:limit] + + def record_use(self, commons_id: str, outcome: str = "ok") -> dict | None: + """ARM use/evaluate feedback loop — raises the record's reuse score.""" + rec = self._records.get(commons_id) + if rec is None: + return None + rec["reuse"]["uses"] += 1 + if outcome == "ok": + rec["reuse"]["evaluations"] += 1 + rec["reuse"]["score"] = round(rec["reuse"]["evaluations"] / max(rec["reuse"]["uses"], 1), 3) + return rec + + +def estate_commons(root) -> Commons: + """A Commons populated from the estate: this repo's CapDs + the suite workloads.""" + root = Path(root) + c = Commons() + c.ingest_capds(root / "capd") + c.ingest_workloads(root / "mesh" / "suite-workloads.json") + return c + + +if __name__ == "__main__": + import sys + root = Path(__file__).resolve().parent.parent + c = estate_commons(root) + recs = c.records() + print(json.dumps({ + "total": len(recs), + "by_asset_type": {t: len(c.search(asset_type=t)) for t in {r["asset_type"] for r in recs}}, + "reproducible": len(c.search(reproducible=True)), + "sample": [{"commons_id": r["commons_id"], "reproducibility": r["reproducibility"], + "cite": r["cite"]} for r in recs[:6]], + }, indent=2)) + sys.exit(0) diff --git a/tools/compute_plane.py b/tools/compute_plane.py new file mode 100644 index 0000000..e7a0e1a --- /dev/null +++ b/tools/compute_plane.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""The compute plane — one governed door to any execution substrate. + +A user develops on a low-mem box (this M2) and the SAME workload scales out, seamlessly, over +whatever compute 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 does not matter — the plane routes by PER-PROJECT / +PER-ACCOUNT policy and live mesh availability, scaling out where it can and where volunteer compute +is offered. + +It is GOVERNED, which is the whole point and the differentiator: a sensitive workload NEVER lands +on an untrusted volunteer/p2p/blockchain backend (fail-closed), every placement is sealed into a +tamper-evident receipt, and if no allowed+available backend exists it falls back to local or blocks +— it never silently ships work somewhere the policy forbids. Configured in the portal dashboard, +by project and by account. +""" +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone + +# The mesh of substrates the plane can target. `trust: untrusted` = volunteer / third-party nodes +# (BOINC, Folding@home, p2p mesh, blockchain market) — never given sensitive data. `elasticity` is +# a coarse scale-out capacity rank (1 = a single laptop, 10 = a global volunteer grid). +BACKENDS = { + "local": {"kind": "local", "trust": "trusted", "elasticity": 1, "gpu": False}, + "k8s": {"kind": "container", "trust": "trusted", "elasticity": 6, "gpu": True}, + "hpc-slurm": {"kind": "hpc", "trust": "trusted", "elasticity": 9, "gpu": True}, + "wasm-edge": {"kind": "wasm", "trust": "trusted", "elasticity": 7, "gpu": False}, + "p2p-mesh": {"kind": "hyperswarm", "trust": "untrusted", "elasticity": 8, "gpu": False}, + "volunteer-boinc": {"kind": "volunteer", "trust": "untrusted", "elasticity": 10, "gpu": False}, + "blockchain-rlc": {"kind": "compute-market", "trust": "untrusted", "elasticity": 8, "gpu": True}, +} + + +def _seal(body: dict) -> str: + return "sha256:" + hashlib.sha256( + json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() + + +def place(workload: dict, policy: dict, availability: dict) -> dict: + """Route one workload to a backend under per-project/per-account policy and live availability. + + workload: {sensitivity: 'sensitive'|'normal', scalable: bool, needs_gpu: bool} + policy: {allowed_backends: [...], forbid_untrusted_for_sensitive: bool (default True), + prefer: [...], require_attestation: bool} + availability: {backend_id: capacity_units} — which backends are up right now, and how big. + + Returns a sealed placement decision. Fail-closed: sensitive work never goes untrusted; no + allowed+available candidate -> fall back to local if permitted, else block. + """ + sensitive = workload.get("sensitivity") == "sensitive" + needs_gpu = bool(workload.get("needs_gpu")) + scalable = bool(workload.get("scalable")) + allowed = set(policy.get("allowed_backends") or BACKENDS.keys()) + forbid_untrusted = policy.get("forbid_untrusted_for_sensitive", True) + + excluded: dict[str, str] = {} + candidates = [] + for bid, spec in BACKENDS.items(): + if bid not in allowed: + excluded[bid] = "not in project policy allowed_backends" + elif availability.get(bid, 0) <= 0: + excluded[bid] = "not available in the mesh right now" + elif needs_gpu and not spec["gpu"]: + excluded[bid] = "no GPU" + elif sensitive and forbid_untrusted and spec["trust"] == "untrusted": + excluded[bid] = "GOVERNANCE: sensitive workload may not run on an untrusted backend" + else: + candidates.append(bid) + + decision = {"plane": "sourceos-continuum.compute_plane.v1", + "workload": workload, "excluded": excluded, + "decided_at": datetime.now(timezone.utc).isoformat()} + + if not candidates: + # fail-closed: nothing allowed+available satisfies the workload's constraints. We do NOT + # silently degrade (e.g. run GPU work on a non-GPU local box) — we block. `local`, when it + # is genuinely usable, is always already a candidate, so there is nothing to fall back to. + decision.update({"backend": None, "placement": "blocked", + "reason": "no allowed+available backend satisfies the workload — blocked, not shipped anywhere (fail-closed)"}) + decision["receipt_digest"] = _seal({k: v for k, v in decision.items() if k != "receipt_digest"}) + return decision + + # honour explicit preference order, then scale out where we can (highest elasticity) for + # scalable work, else keep it cheap/local (lowest elasticity). + prefer = [b for b in (policy.get("prefer") or []) if b in candidates] + if prefer: + chosen = prefer[0] + why = "project preference" + elif scalable: + chosen = max(candidates, key=lambda b: (BACKENDS[b]["elasticity"], availability.get(b, 0))) + why = ("scale-out: highest available elasticity in the mesh" + if BACKENDS[chosen]["elasticity"] > 1 + else "mesh had no scale-out capacity available; ran locally") + else: + chosen = min(candidates, key=lambda b: BACKENDS[b]["elasticity"]) + why = "kept small: lowest-cost candidate (non-scalable workload)" + + decision.update({ + "backend": chosen, + "backend_kind": BACKENDS[chosen]["kind"], + "backend_trust": BACKENDS[chosen]["trust"], + "placement": "scheduled", + "reason": why, + "alternatives": sorted(b for b in candidates if b != chosen), + "attestation_required": bool(policy.get("require_attestation")) or BACKENDS[chosen]["trust"] == "untrusted", + }) + decision["receipt_digest"] = _seal({k: v for k, v in decision.items() if k != "receipt_digest"}) + return decision + + +def backends_view(availability: dict | None = None) -> dict: + """The dashboard view of the mesh: every substrate, its trust/elasticity, and live availability.""" + availability = availability or {} + return {"backends": [ + {"id": bid, **spec, "available": availability.get(bid, 0)} for bid, spec in BACKENDS.items()]} + + +if __name__ == "__main__": + # demo: a sensitive, scalable, GPU workload against a full mesh — untrusted backends are refused. + demo = place({"sensitivity": "sensitive", "scalable": True, "needs_gpu": True}, + {"forbid_untrusted_for_sensitive": True}, + {b: 100 for b in BACKENDS}) + print(json.dumps(demo, indent=2, sort_keys=True)) diff --git a/tools/mcp_a2a_grant.py b/tools/mcp_a2a_grant.py new file mode 100644 index 0000000..a91db48 --- /dev/null +++ b/tools/mcp_a2a_grant.py @@ -0,0 +1,194 @@ +#!/usr/bin/env python3 +"""MCP-A2A Policy Authority + fog-node Policy Gate — CONFORMS to the canonical mcp-a2a-zero-trust +schemas (see schemas/a2a/, vendored + hash-pinned from SourceOS-Linux/mcp-a2a-zero-trust, the +estate's zero-trust authority). This module does NOT invent a grant shape; it emits the canonical +`Grant`, consumes the canonical `AttestationBundle`, honours the canonical `QuorumProof`, and returns +a canonical `tool_grant.validate` check result. + +The cloud-shell fog flow it implements: + + Policy Authority (issue_grant): Attest (AttestationBundle.results.tpm_valid & cosign_valid) → + Decide (a real scheduled compute_plane placement) → Grant (mint a canonical Grant: session-bound + via `binding`, structured `capability`, `constraints`, `policy_hash`, optional `quorum_proof`, + `evidence_refs`, signed with `sig{issuer,sig}`). Refuses to mint without attestation, without a + valid placement, or without a required quorum. + + fog-node Policy Gate (verify_grant): on attach and on every op, re-check the signature, the + session binding, expiry, attestation binding, and that the requested effect/op is within the + Grant's capability + constraints. Returns a canonical tool_grant.validate result + {valid, expired, revoked, reason}. Fail-closed. + +HMAC stands in for the Key Authority (HSM/KMS); the `sig{issuer,sig}` shape is canonical — swap in +ed25519/HSM without touching the flow. +""" +from __future__ import annotations + +import hashlib +import hmac +import json +import re +import uuid +from datetime import datetime, timedelta, timezone + +ISSUER = "spiffe://sourceos/policyarbiter" +_SHA256_RE = re.compile(r"^sha256:[a-f0-9]{64}$") + + +def _canon(body: dict) -> bytes: + return json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8") + + +def _sha256(body: dict) -> str: + return "sha256:" + hashlib.sha256(_canon(body)).hexdigest() + + +def _iso(dt: datetime) -> str: + return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _parse_iso(s: str) -> datetime: + return datetime.fromisoformat(s.replace("Z", "+00:00")) + + +def hmac_signer(key: bytes): + """Stand-in for the Key Authority (HSM/KMS) 'sign grant' op. Returns a hex sig (>=16 chars).""" + return lambda body: hmac.new(key, _canon(body), hashlib.sha256).hexdigest() + + +def hmac_verifier(key: bytes): + signer = hmac_signer(key) + return lambda body, sig: hmac.compare_digest(signer(body), sig) + + +def attestation_bundle(*, spiffe_id: str, aum_digest: str, tpm_valid: bool, cosign_valid: bool, + fido2_valid: bool = False, tpm_quote_ref: str = "worm://quotes/none", + cosign_bundle_ref: str = "worm://bundles/none") -> dict: + """Build a canonical AttestationBundle (schemas/a2a/attestation_bundle.schema.json).""" + return {"subject": {"spiffe_id": spiffe_id, "aum_digest": aum_digest}, + "results": {"tpm_valid": bool(tpm_valid), "cosign_valid": bool(cosign_valid), + "fido2_valid": bool(fido2_valid)}, + "evidence_refs": {"tpm_quote_ref": tpm_quote_ref, "cosign_bundle_ref": cosign_bundle_ref}} + + +class GrantRefused(Exception): + """Attest / Decide / quorum precondition failed — no Grant is minted (fail-closed).""" + + +def issue_grant(*, binding: dict, capability: dict, decision: dict, attestation: dict, + constraints: dict, signer, issuer: str = ISSUER, ttl_s: float = 900.0, + quorum_proof: dict | None = None, evidence_refs: dict | None = None, + now: datetime | None = None) -> dict: + """Attest → Decide → Grant. Returns a canonical Grant (sig embedded) or raises GrantRefused. + + binding: {"spiffe_id", "aum_digest" (sha256:...), "session_id"} + capability: {"kind" (mcp_tool|a2a_skill|deployment|runner_action), "capability_ref", + "capability_digest" (sha256:...), "effect" (read|write|compute|exec|egress), ...} + decision: a compute_plane.place() result — must be a real scheduled placement. + attestation: a canonical AttestationBundle. + constraints: free-form object; `require_quorum`/`quorum_threshold` are Authority-side directives + (stripped from the emitted Grant, which only carries enforceable constraints). + """ + now = datetime.now(timezone.utc) if now is None else now + + # ── Attest ────────────────────────────────────────────────────────────────────── + res = attestation.get("results", {}) + if not (res.get("tpm_valid") and res.get("cosign_valid")): + raise GrantRefused("attestation failed: results.tpm_valid and results.cosign_valid required") + + # ── Decide ────────────────────────────────────────────────────────────────────── + if decision.get("placement") != "scheduled" or not decision.get("backend"): + raise GrantRefused(f"no valid placement to grant (placement={decision.get('placement')!r})") + + # ── Quorum (only when the capability demands it) ───────────────────────────────── + if constraints.get("require_quorum"): + threshold = int(constraints.get("quorum_threshold", 2)) + sigs = (quorum_proof or {}).get("signatures", []) + if not quorum_proof or len(sigs) < threshold: + raise GrantRefused(f"quorum required: need {threshold} validator signatures, got {len(sigs)}") + + cap = dict(capability) + cap.setdefault("executor_ref", f"node://{decision['backend']}") # bind the chosen fog node + + enforceable = {k: v for k, v in constraints.items() if k not in ("require_quorum", "quorum_threshold")} + enforceable.setdefault("ttl_sec", int(ttl_s)) + + ev = dict(evidence_refs or {}) + ev.setdefault("attestation_bundle_ref", "worm://attest/inline") + ev.setdefault("attestation_bundle_hash", _sha256(attestation)) + if decision.get("receipt_digest"): + ev.setdefault("hdt_decision_ref", "worm://hdt/decision") + ev.setdefault("hdt_decision_hash", decision["receipt_digest"]) + + grant = { + "grant_id": "grant_" + uuid.uuid4().hex[:16], + "issued_at": _iso(now), + "expires_at": _iso(now + timedelta(seconds=float(ttl_s))), + "binding": {k: binding[k] for k in ("spiffe_id", "aum_digest", "session_id") if k in binding}, + "capability": cap, + "constraints": enforceable, + "policy_hash": _sha256({"binding": binding, "capability": cap, "constraints": enforceable, + "placement": decision.get("backend"), "attestation": res}), + } + if quorum_proof: + grant["quorum_proof"] = quorum_proof + if ev: + grant["evidence_refs"] = ev + grant["sig"] = {"issuer": issuer, "sig": signer(grant)} # sign the grant-minus-sig + return grant + + +def verify_grant(grant: dict, *, session_id: str, verifier, actor: dict | None = None, + requested_effect: str | None = None, requested_op: str | None = None, + trust_boundary_id: str | None = None, now: datetime | None = None) -> dict: + """fog-node Policy Gate → canonical tool_grant.validate result. Fail-closed; re-check on attach + and on every op. Pass requested_effect/requested_op on an op to enforce it is within the Grant.""" + now = datetime.now(timezone.utc) if now is None else now + + def check(valid, expired, revoked, reason): + out = {"check_id": "chk_" + uuid.uuid4().hex[:8], "operation": "tool_grant.validate", + "grant_id": grant.get("grant_id"), "checked_at": _iso(now), + "actor": actor or {"spiffe_id": ISSUER, + "aum_digest": grant.get("binding", {}).get("aum_digest", "sha256:" + "0" * 64)}, + "result": {"valid": valid, "expired": expired, "revoked": revoked, "reason": reason}, + "policy_hash": grant.get("policy_hash")} + if trust_boundary_id: + out["trust_boundary_id"] = trust_boundary_id + return out + + sig = grant.get("sig", {}) + body = {k: v for k, v in grant.items() if k != "sig"} + if not (sig.get("sig") and verifier(body, sig["sig"])): + return check(False, False, False, "signature invalid — Grant tampered or wrong key") + if grant.get("binding", {}).get("session_id") != session_id: + return check(False, False, False, "session mismatch — Grant is not bound to this session") + if now > _parse_iso(grant.get("expires_at", "1970-01-01T00:00:00Z")): + return check(False, True, False, "Grant expired") + if not grant.get("evidence_refs", {}).get("attestation_bundle_hash"): + return check(False, False, False, "no attestation bound in Grant") + granted_effect = grant.get("capability", {}).get("effect") + if requested_effect is not None and requested_effect != granted_effect: + return check(False, False, False, f"effect {requested_effect!r} not granted (granted: {granted_effect!r})") + ops_allow = grant.get("constraints", {}).get("ops_allow") + if requested_op is not None and ops_allow is not None and requested_op not in ops_allow: + return check(False, False, False, f"op {requested_op!r} not in granted constraints ops_allow={ops_allow}") + return check(True, False, False, "Grant is active and within TTL; session-bound; op permitted") + + +if __name__ == "__main__": + # demo: the full canonical attach flow for one session. + key = b"demo-key-not-for-real-use" + dec = {"placement": "scheduled", "backend": "hpc-slurm", "backend_trust": "trusted", + "receipt_digest": "sha256:" + "de" * 32} + att = attestation_bundle(spiffe_id="spiffe://sourceos/agent/noetica", aum_digest="sha256:" + "ab" * 32, + tpm_valid=True, cosign_valid=True) + grant = issue_grant( + binding={"spiffe_id": "spiffe://sourceos/agent/noetica", "aum_digest": "sha256:" + "ab" * 32, + "session_id": "sess_abc123"}, + capability={"kind": "mcp_tool", "capability_ref": "capd://caps.dev.devspace-inner-loop", + "capability_digest": "sha256:" + "cd" * 32, "effect": "exec", + "server": "shell.runtime", "tool": "pty"}, + decision=dec, attestation=att, constraints={"ops_allow": ["pty.attach", "fs.read"], "paths_allow": ["$HOME/**"]}, + signer=hmac_signer(key)) + result = verify_grant(grant, session_id="sess_abc123", verifier=hmac_verifier(key), + requested_effect="exec", requested_op="pty.attach") + print(json.dumps({"grant": grant, "check": result}, indent=2)) diff --git a/tools/mesh_telemetry.py b/tools/mesh_telemetry.py new file mode 100644 index 0000000..e4df914 --- /dev/null +++ b/tools/mesh_telemetry.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +"""Live mesh telemetry — the fail-closed liveness registry the compute plane places against. + +"Per mesh availability" has to be *live*, not a static snapshot. Every node in the mesh — a k8s +worker, an HPC login node, a WASM edge, a volunteer BOINC/Folding@home box, a p2p peer — emits a +heartbeat carrying its backend kind and free capacity. This registry turns those heartbeats into +the availability dict `compute_plane.place()` consumes. + +Staleness is the signal; TTL is the enforcement. A node that stops beating stops counting: once its +last heartbeat ages past the TTL it contributes zero capacity, so the plane will not schedule there. +Fail-closed by construction — no heartbeat means unavailable, never "assume it's still up." The +read-only portal never ingests; nodes write heartbeat files (sovereign, no broker), the registry +reads them. +""" +from __future__ import annotations + +import json +import time +from pathlib import Path + + +class MeshRegistry: + """Aggregates node heartbeats into live per-backend availability, expiring stale nodes by TTL.""" + + def __init__(self, ttl_seconds: float = 30.0, clock=time.time): + self.ttl = float(ttl_seconds) + self._clock = clock + self._nodes: dict[str, dict] = {} # node_id -> {backend, capacity, ts} + + def heartbeat(self, node_id: str, backend: str, capacity: float, *, ts: float | None = None) -> None: + self._nodes[node_id] = {"backend": backend, "capacity": float(capacity), + "ts": float(ts if ts is not None else self._clock())} + + @classmethod + def from_dir(cls, path, ttl_seconds: float = 30.0, clock=time.time) -> "MeshRegistry": + """Load heartbeat files (.json = {node_id, backend, capacity, ts}) written by nodes.""" + reg = cls(ttl_seconds=ttl_seconds, clock=clock) + p = Path(path) + if p.is_dir(): + for f in p.glob("*.json"): + try: + hb = json.loads(f.read_text()) + reg.heartbeat(hb["node_id"], hb["backend"], hb["capacity"], ts=hb.get("ts")) + except (OSError, json.JSONDecodeError, KeyError, TypeError): + continue # a malformed heartbeat is a dead node, not a crash — fail-closed + return reg + + def _live(self, now: float | None = None) -> dict[str, dict]: + now = self._clock() if now is None else now + return {nid: n for nid, n in self._nodes.items() if now - n["ts"] <= self.ttl} + + def availability(self, now: float | None = None) -> dict[str, float]: + """Live free capacity summed per backend. Backends with no live node are simply absent (0).""" + out: dict[str, float] = {} + for n in self._live(now).values(): + out[n["backend"]] = out.get(n["backend"], 0.0) + n["capacity"] + return out + + def nodes(self, now: float | None = None) -> list[dict]: + """Per-node view for the dashboard: which nodes are live, their age, and capacity.""" + now = self._clock() if now is None else now + rows = [] + for nid, n in sorted(self._nodes.items()): + age = round(now - n["ts"], 1) + rows.append({"node_id": nid, "backend": n["backend"], "capacity": n["capacity"], + "age_s": age, "live": age <= self.ttl}) + return rows + + def summary(self, now: float | None = None) -> dict: + live = self._live(now) + return {"ttl_s": self.ttl, "total_nodes": len(self._nodes), "live_nodes": len(live), + "backends_up": sorted({n["backend"] for n in live.values()})} + + +def write_heartbeat(dir_path, node_id: str, backend: str, capacity: float, *, ts: float | None = None) -> Path: + """Emit one heartbeat file — what a mesh node (or the demo agent) calls to announce itself.""" + d = Path(dir_path) + d.mkdir(parents=True, exist_ok=True) + f = d / f"{node_id}.json" + f.write_text(json.dumps({"node_id": node_id, "backend": backend, + "capacity": float(capacity), + "ts": float(ts if ts is not None else time.time())})) + return f + + +if __name__ == "__main__": + # `mesh_telemetry.py heartbeat ` — announce a node. + # `mesh_telemetry.py view ` — show live availability + per-node liveness. + import sys + if len(sys.argv) >= 6 and sys.argv[1] == "heartbeat": + p = write_heartbeat(sys.argv[2], sys.argv[3], sys.argv[4], float(sys.argv[5])) + print(f"wrote {p}") + elif len(sys.argv) >= 3 and sys.argv[1] == "view": + reg = MeshRegistry.from_dir(sys.argv[2]) + print(json.dumps({"summary": reg.summary(), "availability": reg.availability(), + "nodes": reg.nodes()}, indent=2, sort_keys=True)) + else: + print("usage: mesh_telemetry.py heartbeat | view ") + sys.exit(2) diff --git a/tools/portal_server.py b/tools/portal_server.py new file mode 100644 index 0000000..6ee2a7c --- /dev/null +++ b/tools/portal_server.py @@ -0,0 +1,217 @@ +#!/usr/bin/env python3 +"""Developer portal — the human view of continuum's governed surface (superiority-march move #8). + +A dependency-free, self-hosted, read-only web console over the SAME data the MCP ops surface (move +#1) exposes to agents: CapD capabilities, the lifecycle, and the sealed evidence bundle. One +governed source, two views — agent via MCP, human via this portal. Fully open: stdlib only, no +external CDN (inline HTML/CSS/JS), scale-to-zero (a plain HTTP server, spawned on demand). + +Read-only by construction: mutating actions flow through the governed MCP surface + the fail-closed +promotion gate, never the portal. `route(path) -> (status, content_type, body)` is the pure core +(unit-tested); serve() wraps it in http.server. +""" +from __future__ import annotations + +import json +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +_ROOT = Path(__file__).resolve().parent.parent +_TOOLS = Path(__file__).resolve().parent +_HEARTBEATS = _ROOT / "artifacts" / "mesh-heartbeats" + + +def _sib(name: str): + """Load a sibling tool module by path (keeps the portal import-light and relocatable).""" + import importlib.util + spec = importlib.util.spec_from_file_location(name, _TOOLS / f"{name}.py") + m = importlib.util.module_from_spec(spec) + spec.loader.exec_module(m) + return m + + +def _registry(): + return _sib("mesh_telemetry").MeshRegistry.from_dir(_HEARTBEATS) + + +def _capabilities() -> dict: + caps = [] + d = _ROOT / "capd" + for f in sorted(d.glob("*.capd.json")) if d.is_dir() else []: + try: + data = json.loads(f.read_text()) + caps.append({"file": f.name, "capability_id": data.get("capability_id"), + "kind": data.get("kind"), "status": data.get("status"), + "name": data.get("name"), "policy": data.get("policy", {})}) + except (OSError, json.JSONDecodeError): + continue + return {"capabilities": caps} + + +def _lifecycle() -> dict: + return {"lifecycle": [ + {"stage": "onboard", "note": "local sovereign forge + cluster + sourceosctl"}, + {"stage": "develop", "note": "inner-loop dev-environments (see caps.dev.devspace)"}, + {"stage": "cloud-native-test", "note": "ephemeral preview env + evidence bundle"}, + {"stage": "rollout", "note": "fail-closed promotion gate on a sealed APPROVE verdict"}, + ]} + + +def _evidence(limit: int = 20) -> dict: + bundles = [] + for name in ("gate-decisions", "mcp-receipts"): + p = _ROOT / "artifacts" / name + if p.is_dir(): + for f in sorted(p.glob("*.json"), reverse=True)[:limit]: + bundles.append({"bundle": name, "name": f.name}) + return {"evidence": bundles[:limit]} + + +def _compute() -> dict: + """The compute-mesh view: every substrate the plane can target, with trust and LIVE availability + summed from mesh telemetry (the HyperSwarm discovery/liveness substrate; stale nodes count 0).""" + cp = _sib("compute_plane") + reg = _registry() + view = cp.backends_view(reg.availability()) + view["telemetry"] = reg.summary() + return view + + +def _mesh() -> dict: + """Per-node liveness: which fog nodes are beating, their backend, capacity, and age + (the spec's HyperSwarm Mesh + Node Identity 'find candidate nodes').""" + reg = _registry() + return {"summary": reg.summary(), "nodes": reg.nodes()} + + +def _placements() -> dict: + """The app suite on the mesh: run place() (the Control-Plane-Agent Decide) for each product's + declared workload against LIVE availability, so the dashboard shows where every product's work + would land right now.""" + cp = _sib("compute_plane") + avail = _registry().availability() + try: + profiles = json.loads((_ROOT / "mesh" / "suite-workloads.json").read_text())["workloads"] + except (OSError, json.JSONDecodeError, KeyError): + profiles = [] + out = [] + for p in profiles: + d = cp.place(p.get("workload", {}), p.get("policy", {}), avail) + out.append({"product": p.get("product"), "id": p.get("id"), + "backend": d.get("backend"), "placement": d.get("placement"), + "backend_trust": d.get("backend_trust"), "reason": d.get("reason")}) + return {"placements": out, "availability": avail} + + +def _commons() -> dict: + """The Reproducible Knowledge Commons: every estate capability + workload as a citable, + content-addressed record, honestly graded reproducible vs. declared.""" + c = _sib("commons").estate_commons(_ROOT) + recs = c.records() + return {"total": len(recs), "reproducible": len(c.search(reproducible=True)), + "records": [{"commons_id": r["commons_id"], "domain": r["domain"], + "asset_type": r["asset_type"], "reproducibility": r["reproducibility"], + "cite": r["cite"]} for r in recs]} + + +_CONSOLE = """ +SourceOS Continuum — Console + +

SourceOS Continuum — Developer Console

+
Read-only view of the governed surface. Actions run through the MCP surface + fail-closed promotion gate.
+
+

Capabilities

loading…
+

Lifecycle

loading…
+

Compute mesh — scale out anywhere, governed

+
Develop local; the compute plane routes each workload by per-project policy + live mesh telemetry. Untrusted (volunteer/p2p/blockchain) backends never receive sensitive work.
+
+
loading…
+

App suite on the mesh — live placement

+
Where each product's workload lands right now, under its policy and current availability. BLOCKED = fail-closed, no compliant node live.
+
loading…
+

Reproducible Knowledge Commons

+
Every capability + workload as a citable, content-addressed record (Zenodo-style). reproducible = provenance carries the digests to reproduce it; declared = registered but not yet reproducibility-backed.
+
+
loading…
+

Sealed evidence (latest)

loading…
+
+""" + + +def route(path: str) -> tuple[int, str, str]: + """Pure request router: read-only endpoints only. Returns (status, content_type, body).""" + path = path.split("?", 1)[0] + if path == "/": + return 200, "text/html; charset=utf-8", _CONSOLE + if path == "/healthz": + return 200, "text/plain", "ok" + api = {"/api/capabilities": _capabilities, "/api/lifecycle": _lifecycle, + "/api/evidence": _evidence, "/api/compute": _compute, + "/api/mesh": _mesh, "/api/placements": _placements, "/api/commons": _commons} + if path in api: + return 200, "application/json", json.dumps(api[path](), indent=2, sort_keys=True) + return 404, "text/plain", "not found" + + +class _Handler(BaseHTTPRequestHandler): + def do_GET(self): # read-only: only GET is served + status, ctype, body = route(self.path) + payload = body.encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def log_message(self, *_): # quiet by default + pass + + +def serve(port: int = 8088) -> None: + server = ThreadingHTTPServer(("127.0.0.1", port), _Handler) + print(f"[continuum] portal on http://127.0.0.1:{port} (read-only; Ctrl-C to stop)") + try: + server.serve_forever() + except KeyboardInterrupt: + server.shutdown() + + +if __name__ == "__main__": + import sys + serve(int(sys.argv[1]) if len(sys.argv) > 1 else 8088) diff --git a/tools/test_commons.py b/tools/test_commons.py new file mode 100644 index 0000000..8aecac9 --- /dev/null +++ b/tools/test_commons.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +"""Tests for the Reproducible Knowledge Commons. Load-bearing: (1) the reproducibility gate — a +record may only claim `reproducible` if its provenance actually carries the digests; (2) the ARM +use/evaluate feedback loop; (3) the estate ingests itself, so the compute-mesh/fog work is a +first-class citable record.""" +import pathlib + +import commons as cm + +ROOT = pathlib.Path(__file__).resolve().parent.parent + + +def test_mint_id_is_content_addressed_and_deterministic(): + a, da = cm.mint_id("compute", "mesh-plane", "0.1.0", {"x": 1}) + b, db = cm.mint_id("compute", "mesh-plane", "0.1.0", {"x": 1}) + c, _ = cm.mint_id("compute", "mesh-plane", "0.1.0", {"x": 2}) + assert a == b and da == db and a.startswith("commons:compute/mesh-plane@0.1.0+") + assert c != a # different content -> different citable id (reproducible versioning) + + +def test_reproducibility_gate_requires_real_provenance(): + c = cm.Commons() + weak = c.deposit(domain="ml", name="model", version="1", asset_type="model", content={"w": 1}) + strong = c.deposit(domain="ml", name="model", version="2", asset_type="model", content={"w": 2}, + provenance={"source_digest": "sha256:" + "a" * 64, "attestation_ref": "worm://a1"}) + half = c.deposit(domain="ml", name="model", version="3", asset_type="model", content={"w": 3}, + provenance={"source_digest": "sha256:" + "b" * 64}) # no attestation/sbom + assert weak["reproducibility"] == "declared" + assert strong["reproducibility"] == "reproducible" + assert half["reproducibility"] == "declared" # source alone is not enough + + +def test_deposit_is_citable(): + c = cm.Commons() + rec = c.deposit(domain="data", name="corpus", version="0.2.0", asset_type="dataset", content={"n": 10}) + assert rec["cite"].startswith("corpus 0.2.0. Commons commons:data/corpus@0.2.0+") + + +def test_arm_use_evaluate_feedback_raises_score(): + c = cm.Commons() + rec = c.deposit(domain="d", name="a", version="1", asset_type="capability", content={}) + cid = rec["commons_id"] + c.record_use(cid, "ok") + c.record_use(cid, "ok") + c.record_use(cid, "fail") + r = c.resolve(cid) + assert r["reuse"]["uses"] == 3 and r["reuse"]["evaluations"] == 2 and r["reuse"]["score"] == 0.667 + + +def test_recommend_orders_by_reuse(): + c = cm.Commons() + lo = c.deposit(domain="d", name="lo", version="1", asset_type="capability", content={"a": 1}) + hi = c.deposit(domain="d", name="hi", version="1", asset_type="capability", content={"a": 2}) + for _ in range(3): + c.record_use(hi["commons_id"], "ok") + top = c.recommend(domain="d") + assert top[0]["name"] == "hi" and top[-1]["name"] == "lo" + + +def test_estate_ingests_itself_including_the_fog_work(): + c = cm.estate_commons(ROOT) + caps = c.search(asset_type="capability") + ids = {r["commons_id"].split("+")[0] for r in caps} + # the compute-mesh + cloud-shell fog capabilities are first-class citable records + assert any("caps.compute.mesh-plane" in i for i in ids), ids + assert any("caps.compute.cloudshell-fog" in i for i in ids), ids + # the cloudshell-fog CapD references a ledger/shape-authority, so it ingests as reproducible + fog = next(r for r in caps if "cloudshell-fog" in r["commons_id"]) + assert fog["reproducibility"] == "reproducible" + + +def test_estate_ingests_the_suite_workloads_with_semantic_actions(): + c = cm.estate_commons(ROOT) + workloads = c.search(asset_type="workload") + assert len(workloads) >= 5 + bear = next((r for r in workloads if "bearbrowser" in r["name"]), None) + assert bear is not None and bear["semantic_action"]["policy"] # policy carried as the semantic action + + +def test_search_filters_by_domain_and_reproducibility(): + c = cm.estate_commons(ROOT) + assert all(r["domain"] == "compute" for r in c.search(domain="compute")) + assert all(r["reproducibility"] == "reproducible" for r in c.search(reproducible=True)) + + +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)} commons tests passed") + sys.exit(0) diff --git a/tools/test_compute_plane.py b/tools/test_compute_plane.py new file mode 100644 index 0000000..387610a --- /dev/null +++ b/tools/test_compute_plane.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +"""Tests for the governed compute plane — the placement broker is the differentiator, so its +governance (sensitive work never goes untrusted; fail-closed when nothing is allowed) is nailed +down hard, alongside the scale-out and preference logic.""" +import compute_plane as cp + +FULL = {b: 100 for b in cp.BACKENDS} # whole mesh up, plenty of capacity + + +def test_sensitive_workload_refuses_every_untrusted_backend(): + d = cp.place({"sensitivity": "sensitive", "scalable": True}, {}, FULL) + assert d["backend_trust"] == "trusted" + for bid in ("p2p-mesh", "volunteer-boinc", "blockchain-rlc"): + assert "untrusted" in d["excluded"][bid] + + +def test_sensitive_scalable_gpu_lands_on_hpc_not_volunteer(): + # highest-elasticity *trusted* GPU backend is hpc-slurm; volunteer-boinc (elasticity 10) is barred. + d = cp.place({"sensitivity": "sensitive", "scalable": True, "needs_gpu": True}, {}, FULL) + assert d["backend"] == "hpc-slurm" + + +def test_normal_scalable_scales_out_to_the_biggest_grid(): + # no sensitivity bar -> the global volunteer grid (elasticity 10) wins. + d = cp.place({"sensitivity": "normal", "scalable": True}, {}, FULL) + assert d["backend"] == "volunteer-boinc" + assert d["attestation_required"] is True # untrusted backend always demands attestation + + +def test_nonscalable_workload_stays_small(): + d = cp.place({"sensitivity": "normal", "scalable": False}, {}, FULL) + assert d["backend"] == "local" + + +def test_project_policy_restricts_allowed_backends(): + d = cp.place({"sensitivity": "normal", "scalable": True}, + {"allowed_backends": ["local", "k8s"]}, FULL) + assert d["backend"] == "k8s" + assert "hpc-slurm" in d["excluded"] + + +def test_explicit_preference_wins_over_scale_out(): + d = cp.place({"sensitivity": "normal", "scalable": True}, + {"prefer": ["wasm-edge"]}, FULL) + assert d["backend"] == "wasm-edge" + assert d["reason"] == "project preference" + + +def test_needs_gpu_excludes_non_gpu_backends(): + d = cp.place({"sensitivity": "normal", "scalable": True, "needs_gpu": True}, + {"allowed_backends": ["local", "wasm-edge", "k8s"]}, FULL) + assert d["backend"] == "k8s" # only GPU-capable candidate + assert "no GPU" in d["excluded"]["wasm-edge"] + + +def test_scale_out_wanted_but_only_local_up_runs_local_and_says_so(): + d = cp.place({"sensitivity": "normal", "scalable": True}, {}, {"local": 1}) + assert d["backend"] == "local" + assert d["placement"] == "scheduled" + assert "locally" in d["reason"] # honest: we couldn't actually scale out + + +def test_fail_closed_blocks_when_no_candidate_and_no_local(): + # sensitive workload, only untrusted backends up, local not available -> blocked, shipped nowhere. + d = cp.place({"sensitivity": "sensitive", "scalable": True}, {}, + {"volunteer-boinc": 500}) + assert d["backend"] is None + assert d["placement"] == "blocked" + + +def test_fail_closed_blocks_gpu_work_rather_than_degrade_to_nongpu_local(): + # only a non-GPU local box is up; we must NOT silently run GPU work on it. + d = cp.place({"sensitivity": "normal", "scalable": True, "needs_gpu": True}, {}, {"local": 1}) + assert d["backend"] is None and d["placement"] == "blocked" + + +def test_receipt_seal_is_deterministic_and_covers_the_decision(): + d = cp.place({"sensitivity": "normal", "scalable": True}, {"prefer": ["k8s"]}, FULL) + assert d["receipt_digest"].startswith("sha256:") + reseal = cp._seal({k: v for k, v in d.items() if k != "receipt_digest"}) + assert reseal == d["receipt_digest"] + + +def test_backends_view_exposes_the_whole_mesh_with_availability(): + view = cp.backends_view({"volunteer-boinc": 200}) + ids = {b["id"] for b in view["backends"]} + assert {"local", "hpc-slurm", "wasm-edge", "p2p-mesh", "volunteer-boinc", "blockchain-rlc"} <= ids + boinc = next(b for b in view["backends"] if b["id"] == "volunteer-boinc") + assert boinc["trust"] == "untrusted" and boinc["available"] == 200 + + +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)} compute-plane tests passed") + sys.exit(0) diff --git a/tools/test_mcp_a2a_grant.py b/tools/test_mcp_a2a_grant.py new file mode 100644 index 0000000..387c7a9 --- /dev/null +++ b/tools/test_mcp_a2a_grant.py @@ -0,0 +1,201 @@ +#!/usr/bin/env python3 +"""Tests for the MCP-A2A Policy Authority (issue_grant) + fog-node Policy Gate (verify_grant). + +Two things are proven: (1) the emitted Grant / QuorumProof / AttestationBundle CONFORM to the +canonical mcp-a2a-zero-trust schemas (validated against the vendored schemas/a2a/*.schema.json — so +this is real conformance, not a shape I made up); (2) every fail-closed edge holds — no Grant without +attestation / a real placement / a required quorum, and at the node a tampered / mis-sessioned / +expired / over-reaching Grant is denied.""" +import json +import pathlib +import re +from datetime import datetime, timezone + +import mcp_a2a_grant as g + +KEY = b"unit-test-key" +SIGNER = g.hmac_signer(KEY) +VERIFIER = g.hmac_verifier(KEY) +NOW = datetime(2026, 1, 7, 0, 0, 0, tzinfo=timezone.utc) +AUM = "sha256:" + "ab" * 32 +CAPDIG = "sha256:" + "cd" * 32 +SCHED = {"placement": "scheduled", "backend": "hpc-slurm", "backend_trust": "trusted", + "receipt_digest": "sha256:" + "de" * 32} + +BINDING = {"spiffe_id": "spiffe://sourceos/agent/x", "aum_digest": AUM, "session_id": "sess_unit1"} +CAPABILITY = {"kind": "mcp_tool", "capability_ref": "capd://caps.dev.devspace-inner-loop", + "capability_digest": CAPDIG, "effect": "exec", "server": "shell.runtime", "tool": "pty"} + + +def _att(): + return g.attestation_bundle(spiffe_id="spiffe://sourceos/agent/x", aum_digest=AUM, + tpm_valid=True, cosign_valid=True) + + +def _quorum(nsigs): + return {"rule": "2of3-human", + "validators": ["spiffe://validators/h1", "spiffe://validators/h2", "spiffe://validators/h3"], + "signed_payload_hash": "sha256:" + "11" * 32, + "signatures": [{"kind": "human", "spiffe_id": f"spiffe://validators/h{i + 1}", + "sig": "MEUCIQD" + "fake" * 4} for i in range(nsigs)]} + + +def _issue(**over): + kw = dict(binding=BINDING, capability=CAPABILITY, decision=SCHED, attestation=_att(), + constraints={"ops_allow": ["pty.attach", "fs.read"], "paths_allow": ["$HOME/**"]}, + signer=SIGNER, now=NOW) + kw.update(over) + return g.issue_grant(**kw) + + +# ── canonical-schema conformance (dependency-free JSON-Schema subset validator) ─────────── +def _load_schema(name): + p = pathlib.Path(__file__).resolve().parent.parent / "schemas" / "a2a" / name + return json.loads(p.read_text()) + + +def _conformance_errors(node, schema, path="$"): + errs = [] + t = schema.get("type") + if t == "object": + if not isinstance(node, dict): + return [f"{path}: expected object"] + for r in schema.get("required", []): + if r not in node: + errs.append(f"{path}: missing required '{r}'") + props = schema.get("properties", {}) + if schema.get("additionalProperties") is False: + for k in node: + if k not in props: + errs.append(f"{path}: unexpected property '{k}'") + for k, v in node.items(): + if k in props and "$ref" not in props[k]: # $ref'd subschemas validated on their own + errs += _conformance_errors(v, props[k], f"{path}.{k}") + elif t == "array": + if not isinstance(node, list): + return [f"{path}: expected array"] + if schema.get("items"): + for i, e in enumerate(node): + errs += _conformance_errors(e, schema["items"], f"{path}[{i}]") + if "enum" in schema and node not in schema["enum"]: + errs.append(f"{path}: {node!r} not in enum {schema['enum']}") + if "pattern" in schema and isinstance(node, str) and not re.search(schema["pattern"], node): + errs.append(f"{path}: {node!r} does not match {schema['pattern']}") + return errs + + +def test_emitted_grant_conforms_to_canonical_grant_schema(): + errs = _conformance_errors(_issue(), _load_schema("grant.schema.json")) + assert errs == [], errs + + +def test_grant_with_quorum_still_conforms_and_quorum_matches_its_schema(): + grant = _issue(constraints={"ops_allow": ["deploy"], "require_quorum": True, "quorum_threshold": 2}, + quorum_proof=_quorum(2)) + assert _conformance_errors(grant, _load_schema("grant.schema.json")) == [] + assert _conformance_errors(grant["quorum_proof"], _load_schema("quorum_proof.schema.json")) == [] + + +def test_attestation_bundle_conforms_to_canonical_schema(): + assert _conformance_errors(_att(), _load_schema("attestation_bundle.schema.json")) == [] + + +# ── Policy Authority: Attest → Decide → Grant, fail-closed ──────────────────────────────── +def test_issue_mints_a_session_bound_signed_grant(): + grant = _issue() + assert grant["binding"]["session_id"] == "sess_unit1" + assert grant["capability"]["executor_ref"] == "node://hpc-slurm" # bound the chosen fog node + assert grant["issued_at"] == "2026-01-07T00:00:00Z" and grant["expires_at"] == "2026-01-07T00:15:00Z" + body = {k: v for k, v in grant.items() if k != "sig"} + assert VERIFIER(body, grant["sig"]["sig"]) and grant["sig"]["issuer"] == g.ISSUER + + +def test_issue_refused_without_attestation(): + bad = g.attestation_bundle(spiffe_id="s", aum_digest=AUM, tpm_valid=True, cosign_valid=False) + try: + _issue(attestation=bad) + assert False + except g.GrantRefused as e: + assert "attestation" in str(e) + + +def test_issue_refused_on_blocked_placement(): + try: + _issue(decision={"placement": "blocked", "backend": None}) + assert False + except g.GrantRefused as e: + assert "placement" in str(e) + + +def test_issue_refused_when_quorum_required_but_insufficient(): + try: + _issue(constraints={"ops_allow": ["deploy"], "require_quorum": True, "quorum_threshold": 2}, + quorum_proof=_quorum(1)) + assert False + except g.GrantRefused as e: + assert "quorum" in str(e) + + +# ── fog-node Policy Gate: canonical tool_grant.validate, fail-closed ────────────────────── +def _at(minute): + return datetime(2026, 1, 7, 0, minute, 0, tzinfo=timezone.utc) + + +def test_gate_authorizes_a_valid_grant_for_a_permitted_effect_and_op(): + res = g.verify_grant(_issue(), session_id="sess_unit1", verifier=VERIFIER, + requested_effect="exec", requested_op="pty.attach", now=_at(5)) + assert res["operation"] == "tool_grant.validate" + assert res["result"] == {"valid": True, "expired": False, "revoked": False, + "reason": "Grant is active and within TTL; session-bound; op permitted"} + + +def test_gate_denies_a_tampered_grant(): + grant = _issue() + grant["capability"]["executor_ref"] = "node://volunteer-boinc" # redirect to an untrusted node + res = g.verify_grant(grant, session_id="sess_unit1", verifier=VERIFIER, now=_at(5)) + assert res["result"]["valid"] is False and "tamper" in res["result"]["reason"].lower() + + +def test_gate_denies_wrong_session(): + res = g.verify_grant(_issue(), session_id="sess_other", verifier=VERIFIER, now=_at(5)) + assert res["result"]["valid"] is False and "session" in res["result"]["reason"] + + +def test_gate_denies_expired_grant(): + res = g.verify_grant(_issue(), session_id="sess_unit1", verifier=VERIFIER, now=_at(20)) + assert res["result"]["valid"] is False and res["result"]["expired"] is True + + +def test_gate_denies_effect_outside_grant(): + res = g.verify_grant(_issue(), session_id="sess_unit1", verifier=VERIFIER, + requested_effect="write", now=_at(5)) + assert res["result"]["valid"] is False and "write" in res["result"]["reason"] + + +def test_gate_denies_op_outside_constraints(): + res = g.verify_grant(_issue(), session_id="sess_unit1", verifier=VERIFIER, + requested_op="fs.write", now=_at(5)) + assert res["result"]["valid"] is False and "fs.write" in res["result"]["reason"] + + +def test_end_to_end_place_then_grant_then_gate(): + import compute_plane as cp + decision = cp.place({"sensitivity": "sensitive", "scalable": True, "needs_gpu": True}, + {"require_attestation": True}, {"hpc-slurm": 100}) + assert decision["backend"] == "hpc-slurm" + grant = g.issue_grant(binding=BINDING, capability={**CAPABILITY, "effect": "compute"}, + decision=decision, attestation=_att(), + constraints={"ops_allow": ["exec.run"]}, signer=SIGNER, now=NOW) + assert _conformance_errors(grant, _load_schema("grant.schema.json")) == [] + res = g.verify_grant(grant, session_id="sess_unit1", verifier=VERIFIER, + requested_effect="compute", requested_op="exec.run", now=_at(1)) + assert res["result"]["valid"] is True + + +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)} mcp-a2a-grant conformance+gate tests passed") + sys.exit(0) diff --git a/tools/test_mesh_telemetry.py b/tools/test_mesh_telemetry.py new file mode 100644 index 0000000..eb1ce5d --- /dev/null +++ b/tools/test_mesh_telemetry.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +"""Tests for the live mesh telemetry registry. The load-bearing behavior is fail-closed liveness: +a node that stops beating must stop counting toward availability. Time is injected so expiry is +deterministic.""" +import json + +import mesh_telemetry as mt + + +class Clock: + def __init__(self, t=1000.0): + self.t = t + + def __call__(self): + return self.t + + +def test_heartbeats_sum_capacity_per_backend(): + reg = mt.MeshRegistry(ttl_seconds=30, clock=Clock()) + reg.heartbeat("n1", "k8s", 4) + reg.heartbeat("n2", "k8s", 6) + reg.heartbeat("n3", "hpc-slurm", 100) + assert reg.availability() == {"k8s": 10, "hpc-slurm": 100} + + +def test_stale_node_stops_counting_after_ttl(): + clk = Clock(1000.0) + reg = mt.MeshRegistry(ttl_seconds=30, clock=clk) + reg.heartbeat("n1", "volunteer-boinc", 200) # beat at t=1000 + assert reg.availability() == {"volunteer-boinc": 200} + clk.t = 1031.0 # 31s later, past the 30s TTL + assert reg.availability() == {} # fail-closed: no longer available + + +def test_a_fresh_beat_revives_a_node(): + clk = Clock(1000.0) + reg = mt.MeshRegistry(ttl_seconds=30, clock=clk) + reg.heartbeat("n1", "k8s", 8) + clk.t = 1040.0 + assert reg.availability() == {} # expired + reg.heartbeat("n1", "k8s", 8) # beats again at t=1040 + assert reg.availability() == {"k8s": 8} # live again + + +def test_nodes_view_reports_liveness_and_age(): + clk = Clock(1000.0) + reg = mt.MeshRegistry(ttl_seconds=30, clock=clk) + reg.heartbeat("live1", "k8s", 4, ts=990.0) # 10s old -> live + reg.heartbeat("dead1", "p2p-mesh", 9, ts=900.0) # 100s old -> dead + rows = {r["node_id"]: r for r in reg.nodes()} + assert rows["live1"]["live"] is True and rows["live1"]["age_s"] == 10.0 + assert rows["dead1"]["live"] is False + + +def test_summary_counts_live_vs_total(): + clk = Clock(1000.0) + reg = mt.MeshRegistry(ttl_seconds=30, clock=clk) + reg.heartbeat("a", "k8s", 1, ts=1000.0) + reg.heartbeat("b", "hpc-slurm", 1, ts=800.0) # stale + s = reg.summary() + assert s["total_nodes"] == 2 and s["live_nodes"] == 1 and s["backends_up"] == ["k8s"] + + +def test_from_dir_loads_heartbeat_files_and_skips_malformed(tmp_path): + mt.write_heartbeat(tmp_path, "good", "wasm-edge", 20, ts=1000.0) + (tmp_path / "broken.json").write_text("{not json") + (tmp_path / "incomplete.json").write_text(json.dumps({"node_id": "x"})) # missing keys + reg = mt.MeshRegistry.from_dir(tmp_path, ttl_seconds=30, clock=Clock(1005.0)) + assert reg.availability() == {"wasm-edge": 20} # only the good one; malformed = dead, no crash + + +def test_empty_dir_is_fail_closed_not_error(tmp_path): + reg = mt.MeshRegistry.from_dir(tmp_path, ttl_seconds=30, clock=Clock()) + assert reg.availability() == {} and reg.summary()["live_nodes"] == 0 + + +def test_telemetry_feeds_the_compute_plane_end_to_end(): + import compute_plane as cp + clk = Clock(1000.0) + reg = mt.MeshRegistry(ttl_seconds=30, clock=clk) + reg.heartbeat("gpu1", "hpc-slurm", 50) # both beat at t=1000 + reg.heartbeat("vol1", "volunteer-boinc", 500) + # a normal scalable workload scales out to the biggest live backend... + d = cp.place({"sensitivity": "normal", "scalable": True}, {}, reg.availability()) + assert d["backend"] == "volunteer-boinc" + # the volunteer node goes silent; the gpu node keeps beating. + clk.t = 1031.0 + reg.heartbeat("gpu1", "hpc-slurm", 50) # fresh beat at t=1031 + d2 = cp.place({"sensitivity": "normal", "scalable": True}, {}, reg.availability()) + assert d2["backend"] == "hpc-slurm" # volunteer expired -> re-placed onto live HPC + + +if __name__ == "__main__": + import sys + import tempfile + fns = [(k, v) for k, v in sorted(globals().items()) if k.startswith("test_")] + for name, fn in fns: + if "tmp_path" in fn.__code__.co_varnames: + import pathlib + with tempfile.TemporaryDirectory() as td: + fn(pathlib.Path(td)) + else: + fn() + print(f"ok: {len(fns)} mesh-telemetry tests passed") + sys.exit(0) diff --git a/tools/test_portal_server.py b/tools/test_portal_server.py new file mode 100644 index 0000000..a01250b --- /dev/null +++ b/tools/test_portal_server.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +"""Tests for the developer portal's pure router — every endpoint, the 404, and that the two views +the console renders (capabilities incl. DevSpace, and the compute mesh incl. the untrusted volunteer +grid) are actually served.""" +import json +import pathlib +import tempfile + +import mesh_telemetry as mt +import portal_server as ps + + +def _seed_fresh(dirpath): + """A fresh, isolated live mesh so telemetry-backed endpoints are deterministic.""" + mt.write_heartbeat(dirpath, "k8s-x", "k8s", 8) + mt.write_heartbeat(dirpath, "boinc-x", "volunteer-boinc", 300) + mt.write_heartbeat(dirpath, "slurm-x", "hpc-slurm", 100) + + +def test_root_serves_the_console_html(): + status, ctype, body = ps.route("/") + assert status == 200 + assert "text/html" in ctype + assert "Developer Console" in body and "Compute mesh" in body + + +def test_healthz_is_ok(): + assert ps.route("/healthz") == (200, "text/plain", "ok") + + +def test_api_endpoints_return_json(): + for path, key in (("/api/capabilities", "capabilities"), + ("/api/lifecycle", "lifecycle"), + ("/api/evidence", "evidence"), + ("/api/compute", "backends")): + status, ctype, body = ps.route(path) + assert status == 200 and ctype == "application/json" + assert key in json.loads(body) + + +def test_query_string_is_ignored_by_router(): + assert ps.route("/healthz?x=1")[0] == 200 + + +def test_unknown_path_is_404(): + status, _, _ = ps.route("/nope") + assert status == 404 + + +def test_devspace_capability_is_surfaced(): + caps = json.loads(ps.route("/api/capabilities")[2])["capabilities"] + ids = {c.get("capability_id") for c in caps} + assert any(str(i).startswith("caps.dev.devspace-inner-loop") for i in ids), ids + + +def test_compute_mesh_reflects_live_telemetry(): + with tempfile.TemporaryDirectory() as td: + old, ps._HEARTBEATS = ps._HEARTBEATS, pathlib.Path(td) + try: + _seed_fresh(td) + backends = json.loads(ps.route("/api/compute")[2])["backends"] + by_id = {b["id"]: b for b in backends} + assert by_id["volunteer-boinc"]["trust"] == "untrusted" + assert by_id["volunteer-boinc"]["available"] == 300 # summed from the live heartbeat + assert by_id["hpc-slurm"]["available"] == 100 + finally: + ps._HEARTBEATS = old + + +def test_mesh_endpoint_lists_live_nodes(): + with tempfile.TemporaryDirectory() as td: + old, ps._HEARTBEATS = ps._HEARTBEATS, pathlib.Path(td) + try: + _seed_fresh(td) + mesh = json.loads(ps.route("/api/mesh")[2]) + assert mesh["summary"]["live_nodes"] == 3 + assert {n["node_id"] for n in mesh["nodes"]} == {"k8s-x", "boinc-x", "slurm-x"} + finally: + ps._HEARTBEATS = old + + +def test_placements_govern_the_suite_over_live_availability(): + with tempfile.TemporaryDirectory() as td: + old, ps._HEARTBEATS = ps._HEARTBEATS, pathlib.Path(td) + try: + _seed_fresh(td) # k8s + volunteer-boinc + hpc-slurm live + pl = {p["id"]: p for p in json.loads(ps.route("/api/placements")[2])["placements"]} + # offensive tooling must never ride the volunteer grid, whatever the scale pressure + assert pl["bearbrowser.scan"]["backend"] != "volunteer-boinc" + # sensitive reasoning lands on trusted infra + assert pl["noetica.reasoning"]["backend_trust"] == "trusted" + finally: + ps._HEARTBEATS = old + + +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)} portal tests passed") + sys.exit(0) diff --git a/tools/validate.py b/tools/validate.py index 5ee93f3..af0ffa4 100644 --- a/tools/validate.py +++ b/tools/validate.py @@ -18,9 +18,25 @@ "docs/LIFECYCLE.md", "docs/CONTINUUM_SCOPE.md", "capd/continuum.local-paas.capd.json", + "capd/compute-plane.mesh.capd.json", + "capd/devspace.local-dev.capd.json", + "capd/cloudshell-fog.capd.json", + "capd/knowledge-commons.mesh.capd.json", "tools/promotion_gate.py", + "tools/portal_server.py", + "tools/compute_plane.py", + "tools/mesh_telemetry.py", + "tools/mcp_a2a_grant.py", + "tools/commons.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. +EXTRA_CAPD_IDS = { + "capd/compute-plane.mesh.capd.json": "caps.compute.mesh-plane", + "capd/devspace.local-dev.capd.json": "caps.dev.devspace-inner-loop", + "capd/cloudshell-fog.capd.json": "caps.compute.cloudshell-fog", + "capd/knowledge-commons.mesh.capd.json": "caps.knowledge.commons", +} errors: list[str] = [] @@ -42,6 +58,20 @@ except json.JSONDecodeError as exc: errors.append(f"capd invalid json: {exc}") +for rel, want_id in EXTRA_CAPD_IDS.items(): + path = ROOT / rel + if not path.exists(): + continue + try: + data = json.loads(path.read_text()) + for key in CAPD_KEYS: + if key not in data: + errors.append(f"{rel} missing key: {key}") + if data.get("capability_id", "").split("@")[0] != want_id: + errors.append(f"{rel} capability_id drift (want {want_id})") + except json.JSONDecodeError as exc: + errors.append(f"{rel} invalid json: {exc}") + for rel in ("README.md", "docs/LIFECYCLE.md", "docs/CONTINUUM_SCOPE.md"): path = ROOT / rel if path.exists() and "…" in path.read_text():