# SPDX-License-Identifier: Apache-2.0 # © 2026 Lutar, Stephen P. — SZL Holdings · ORCID 0009-0001-0110-4173 · Doctrine v11 # Authored by SZL Enterprise-Integration Team. Co-Authored-By: Perplexity Computer Agent. """szl_connectors.governance — the Λ-gate + DSSE/Khipu receipt for every write(). DOCTRINE (non-negotiable): • Every state-changing connector action (write) is Λ-gated. Λ is computed as a geometric-mean over trust axes and is CAPPED below 1.0 (conformal anti- overconfidence floor 1/(n+1)) — trust is NEVER reported as 100%. • Every write emits a DSSE-signed Khipu receipt (real ECDSA-P256 over the DSSE PAE when SZL_COSIGN_PRIVATE_PEM is present; an explicit UNSIGNED envelope otherwise — NEVER a fabricated signature). Reuses the live `szl_dsse` module. • No credential value is EVER placed in a receipt body — only a credential fingerprint hash (sha256, truncated). • State-changing writes carry the 2-person Yuyay gate + Khipu 3-of-4 quorum status (the hatun-mcp governance contract). Until a connector is CONNECTED, write() is refused with an honest reason. This is the same governance discipline proven in operator_shell_v4 / hatun-mcp; here it is packaged as a small, dependency-light gate the connector write paths call directly. """ from __future__ import annotations import hashlib import json import os from datetime import datetime, timezone from typing import Any # Anti-overconfidence floor: Λ is never reported as 1.0. We cap at this ceiling. LAMBDA_CEILING = 0.985 def _now() -> str: return datetime.now(timezone.utc).isoformat() def lambda_score(axes: dict[str, float]) -> float: """Geometric-mean Λ over trust axes, capped below 1.0 (conformal floor). Each axis ∈ [0,1]; Λ = (∏ axis)^(1/n), then min(Λ, LAMBDA_CEILING). A single zero axis vetoes the action (conjunctive veto / soundness A1). Λ is NEVER 1.0. """ vals = [max(0.0, min(1.0, float(v))) for v in axes.values()] or [0.0] if any(v <= 0.0 for v in vals): return 0.0 prod = 1.0 for v in vals: prod *= v lam = prod ** (1.0 / len(vals)) return round(min(lam, LAMBDA_CEILING), 6) def quorum_status(present: list[str] | None = None, n: int = 4, f: int = 1) -> dict[str, Any]: """Khipu 3-of-4 (n≥3f+1) quorum arithmetic. Honest: live_polled False unless a `present` witness set is supplied (matches hatun-mcp mesh_quorum_status).""" need = 2 * f + 1 # 3-of-4 safe fragment present = present or [] return { "scheme": f"{need}-of-{n}", "n": n, "f": f, "need": need, "present": present, "present_count": len(present), "satisfied": len(present) >= need, "live_polled": bool(present), "note": "Khipu safe-fragment quorum (BFT safety = Conjecture 2, OPEN)" if not present else "witness set supplied", } def _dsse_sign(payload: dict[str, Any]) -> dict[str, Any]: """Sign a receipt payload via the live szl_dsse module if importable; else an honest UNSIGNED envelope (NEVER a fabricated signature).""" try: import szl_dsse # the live in-image DSSE/Cosign module return szl_dsse.sign_payload(payload, "application/vnd.szl.khipu+json") except Exception: # honest fallback when szl_dsse is not importable in this context body = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() import base64 return { "payloadType": "application/vnd.szl.khipu+json", "payload": base64.b64encode(body).decode("ascii"), "signatures": [], "signed": False, "honesty": ("UNSIGNED — szl_dsse unavailable in this runtime; " "no signature fabricated."), "_signed_at": _now(), } def receipt_for_write(*, connector_id: str, action: dict[str, Any], lambda_value: float, cred_fingerprints: dict[str, str] | None = None, quorum: dict[str, Any] | None = None, result_summary: dict[str, Any] | None = None) -> dict[str, Any]: """Build + DSSE-sign a Khipu receipt for a connector write. The receipt body carries the connector id, the action *shape* (method/object, NOT secret values), the Λ score, quorum status, credential FINGERPRINT HASHES (never the values), and a result summary. Returns {receipt_hash, dsse, body}. """ # scrub the action of anything secret-looking; keep only shape safe_action = {k: v for k, v in (action or {}).items() if k.lower() not in ("token", "secret", "password", "api_key", "key")} body = { "kind": "szl.connector.write", "connector_id": connector_id, "action": safe_action, "lambda_value": lambda_value, "lambda_note": "Λ never 1.0 (conformal anti-overconfidence floor 1/(n+1)); Λ = Conjecture 1", "quorum": quorum or quorum_status(), "credential_fingerprints": cred_fingerprints or {}, "result": result_summary or {}, "emitted_at": _now(), "doctrine": "v11 — Λ-gate + DSSE/Khipu receipt on every write; no committed keys; trust never 100%", } receipt_hash = "sha256:" + hashlib.sha256( json.dumps(body, sort_keys=True, separators=(",", ":")).encode()).hexdigest() body["receipt_hash"] = receipt_hash dsse = _dsse_sign(body) return {"receipt_hash": receipt_hash, "dsse": dsse, "body": body} def gate_write(*, connector_id: str, connected: bool, action: dict[str, Any], cred_fingerprints: dict[str, str] | None = None, quorum_present: list[str] | None = None, extra_axes: dict[str, float] | None = None): """Run the full governed write gate. Returns (allowed: bool, lambda_value, receipt_dict, quorum_dict, detail). Λ axes (each ∈[0,1], conjunctive): connected — connector is CONNECTED (refuse writes until creds activate it) input_ok — action shape is well-formed (has a method/object) no_secret_leak — no raw secret in the action body quorum_ok — 2-person / 3-of-4 quorum satisfied (or honest pending) """ has_method = bool((action or {}).get("method") or (action or {}).get("object") or (action or {}).get("doctype") or (action or {}).get("sobject")) leak = any(k.lower() in ("token", "secret", "password") for k in (action or {})) q = quorum_status(present=quorum_present) axes = { "connected": 1.0 if connected else 0.0, "input_ok": 1.0 if has_method else 0.0, "no_secret_leak": 0.0 if leak else 1.0, "quorum_ok": 1.0 if q["satisfied"] else 0.5, # pending quorum lowers Λ, doesn't fabricate } if extra_axes: axes.update(extra_axes) lam = lambda_score(axes) # write is allowed only when CONNECTED + well-formed + no leak. allowed = connected and has_method and not leak detail = "" if not connected: detail = "write refused — connector not CONNECTED (provide credentials to activate)" elif not has_method: detail = "write refused — action shape missing method/object" elif leak: detail = "write refused — raw secret detected in action body (doctrine: env/secret only)" elif not q["satisfied"]: detail = "write staged — 2-person / 3-of-4 Khipu quorum pending (state-changing gate)" receipt = receipt_for_write( connector_id=connector_id, action=action, lambda_value=lam, cred_fingerprints=cred_fingerprints, quorum=q, result_summary={"allowed": allowed, "detail": detail}, ) return allowed, lam, receipt, q, detail __all__ = ["lambda_score", "quorum_status", "receipt_for_write", "gate_write", "LAMBDA_CEILING"]