""" core/capability.py — what "this capability works" is actually allowed to mean. A registry that lists a tool as *available* and finds out at call time is guessing. This is the vocabulary that replaces the guess, and it is a ladder: each rung is a separate fact, established separately, and no rung implies the one above it. cataloged the mesh knows it exists — a description, an install line wrapped it is represented as coordinates that can be addressed simulatable mock and dry_run produce a real envelope for it reachable the edge that would run it is paired and online authenticated the session it needs is currently valid at the far end callable an executor for this specific coordinate resolved authorized this identity is permitted to invoke it executed it ran, and there is a receipt `authenticated` is separate from `reachable` on purpose. A seller's laptop can be online and paired while Amazon has expired the session or is asking for MFA. Collapsing the two produces the worst failure this system can have: a district that looks live, and silently returns yesterday's numbers. The rule that keeps it honest: SEC-Ω describes transport and policy protection. This describes whether work can actually be performed. Neither implies the other. A SEC-Ω-protected request can be perfectly secure and still correctly answer edge_unreachable, executor_missing, unsupported_platform, authorization_denied, approval_required or execution_failed. A protected channel is not a working destination, and a catalog entry must never be able to present itself as a live tool because it happened to be wrapped. """ from __future__ import annotations from typing import Any, Dict, List, Optional SCHEMA = "signalmesh.capability/v1" # Ordered. Index is the rung; nothing here is a set of independent flags. LADDER: List[str] = [ "cataloged", "wrapped", "simulatable", "reachable", "authenticated", "callable", "authorized", "executed", ] MEANING: Dict[str, str] = { "cataloged": "known to the mesh — described, not represented", "wrapped": "represented as coordinates that can be addressed", "simulatable": "mock and dry_run produce a real envelope", "reachable": "the edge that would run it is paired and online", "authenticated": "the session it needs is currently valid at the far end", "callable": "an executor resolved for this coordinate", "authorized": "this identity is permitted to invoke it", "executed": "it ran, and there is a receipt", } # Why a rung was not reached. These are answers, not errors — a capability that # honestly reports why it cannot run is working correctly. BLOCKED: Dict[str, str] = { "reachable": "edge_unreachable", "authenticated": "session_expired", "callable": "executor_missing", "authorized": "authorization_denied", "executed": "not_yet_invoked", } def rung(status: str) -> int: try: return LADDER.index(status) except ValueError: return -1 def at_least(status: str, required: str) -> bool: """Is this capability at or above the rung an action needs?""" a, b = rung(status), rung(required) return a >= 0 and b >= 0 and a >= b def assess(*, wrapped: bool = False, has_coords: bool = False, simulatable: bool = False, edge_required: bool = False, edge_online: bool = False, session_required: bool = False, session_valid: bool = False, executor_resolved: Optional[bool] = None, authorized: Optional[bool] = None, receipt: Optional[str] = None) -> Dict[str, Any]: """Climb as far as the evidence goes, then stop and say why. Deliberately conservative at every step. `executor_resolved=None` means nobody has checked, which is not the same as an executor being present, and is therefore not enough to call something callable. """ if receipt: return _at("executed", receipt=receipt) if not wrapped or not has_coords: return _at("cataloged", why="not wrapped — it is described, not represented") if not simulatable: return _at("wrapped", why="no mock or dry_run behaviour to exercise") # An origin that needs no edge is reachable by definition; one that does is # reachable only while a paired device is actually connected. if edge_required and not edge_online: return _at("simulatable", blocked_at="reachable", why=BLOCKED["reachable"], detail="requires a paired meshd edge; none is connected") # A live machine is not a live session. This is the rung that stops a # district reporting healthy while the far end is asking for MFA. if session_required and not session_valid: return _at("reachable", blocked_at="authenticated", why=BLOCKED["authenticated"], detail=("the edge is online but the session it needs has " "expired or is being challenged — the owner has to " "sign in again; no stored password can do it for them")) if executor_resolved is not True: return _at("authenticated" if session_required else "reachable", blocked_at="callable", why=BLOCKED["callable"], detail=("no executor has been resolved for this coordinate. " "A reachable edge is not proof that the thing it " "would run is installed, supported or complete.")) if authorized is not True: return _at("callable", blocked_at="authorized", why=BLOCKED["authorized"], detail="this identity has not been permitted to invoke it") return _at("authorized", blocked_at="executed", why=BLOCKED["executed"], detail="permitted, but it has not been run") def _at(status: str, *, blocked_at: str = "", why: str = "", detail: str = "", receipt: str = "") -> Dict[str, Any]: out: Dict[str, Any] = { "schema": SCHEMA, "status": status, "rung": rung(status), "means": MEANING[status], "ladder": LADDER, # said out loud on every response, because it is the thing people get wrong "note": ("SEC-Ω protects the channel; this describes whether work can be " "performed. Neither implies the other."), } if blocked_at: out["blocked_at"] = blocked_at out["reason"] = why out["detail"] = detail elif why: out["why"] = why if receipt: out["receipt"] = receipt return out if __name__ == "__main__": assert LADDER[0] == "cataloged" and LADDER[-1] == "executed" assert rung("callable") > rung("reachable") > rung("simulatable") assert rung("nonsense") == -1 assert at_least("executed", "callable") assert at_least("callable", "callable") assert not at_least("simulatable", "callable") assert not at_least("callable", "nonsense") # a catalog entry is not a tool, however much it describes itself c = assess() assert c["status"] == "cataloged", c assert "not wrapped" in c["why"], "a plain status must still say why it stopped" # wrapped, but nothing to exercise w = assess(wrapped=True, has_coords=True) assert w["status"] == "wrapped" and "no mock" in w["why"], w # the nicoli case: wrapped and simulatable, edge not paired n = assess(wrapped=True, has_coords=True, simulatable=True, edge_required=True) assert n["status"] == "simulatable" and n["blocked_at"] == "reachable" assert n["reason"] == "edge_unreachable" # pairing the edge does NOT make it callable — this is the overclaim guard paired = assess(wrapped=True, has_coords=True, simulatable=True, edge_required=True, edge_online=True) assert paired["status"] == "reachable", paired assert paired["blocked_at"] == "callable" and paired["reason"] == "executor_missing" assert "not proof" in paired["detail"] # an executor that nobody checked is not an executor that exists unchecked = assess(wrapped=True, has_coords=True, simulatable=True, edge_required=True, edge_online=True, executor_resolved=None) assert unchecked["status"] == "reachable" # ── the seller's laptop is on; Amazon is asking for MFA ───────────────── # The failure this rung exists to prevent: a district that looks healthy # and quietly serves yesterday's numbers. stale = assess(wrapped=True, has_coords=True, simulatable=True, edge_required=True, edge_online=True, session_required=True, session_valid=False, executor_resolved=True, authorized=True) assert stale["status"] == "reachable", stale assert stale["blocked_at"] == "authenticated" assert stale["reason"] == "session_expired" assert "sign in again" in stale["detail"] assert "no stored password" in stale["detail"] signed_in = assess(wrapped=True, has_coords=True, simulatable=True, edge_required=True, edge_online=True, session_required=True, session_valid=True, executor_resolved=True, authorized=True) assert signed_in["status"] == "authorized", signed_in # a session that expires between two calls demotes the capability — being # authorized a minute ago is not evidence about now assert rung(stale["status"]) < rung(signed_in["status"]) # and something needing no session is not held at that rung nosession = assess(wrapped=True, has_coords=True, simulatable=True, edge_required=True, edge_online=True, session_required=False, executor_resolved=True, authorized=True) assert nosession["status"] == "authorized" resolved = assess(wrapped=True, has_coords=True, simulatable=True, edge_required=True, edge_online=True, executor_resolved=True) assert resolved["status"] == "callable" and resolved["blocked_at"] == "authorized" ok = assess(wrapped=True, has_coords=True, simulatable=True, edge_required=True, edge_online=True, executor_resolved=True, authorized=True) assert ok["status"] == "authorized" and ok["reason"] == "not_yet_invoked" done = assess(wrapped=True, has_coords=True, simulatable=True, edge_required=True, edge_online=True, executor_resolved=True, authorized=True, receipt="call_abc123") assert done["status"] == "executed" and done["receipt"] == "call_abc123" # an http origin needs no edge, so reachability is not gated on one http = assess(wrapped=True, has_coords=True, simulatable=True, edge_required=False, executor_resolved=True, authorized=True) assert http["status"] == "authorized", http # every rung says the thing people get wrong for state in (c, w, n, paired, resolved, ok, done, http): assert "SEC-Ω protects the channel" in state["note"] print("capability self-check OK")