# 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 PBKDF2-HMAC-SHA256 credential fingerprint. • 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 json import re from datetime import datetime, timezone from typing import Any from szl_content_address import sha256_content_address # Anti-overconfidence floor: Λ is never reported as 1.0. We cap at this ceiling. LAMBDA_CEILING = 0.985 _FINGERPRINT_RE = re.compile(r"\Apbkdf2-sha256:[0-9a-f]{32}\Z") _FINGERPRINT_LABEL_RE = re.compile(r"\A[A-Za-z0-9_.:-]{1,80}\Z") _SENSITIVE_KEY_PARTS = frozenset({"password", "passwd", "secret", "token", "credential"}) _SENSITIVE_KEY_NAMES = frozenset({ "api_key", "private_key", "authorization", "proxy_authorization", "cookie", "set_cookie", }) 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 _is_sensitive_key(key: object) -> bool: normalized = re.sub(r"[^a-z0-9]+", "_", str(key).lower()).strip("_") parts = frozenset(part for part in normalized.split("_") if part) return normalized in _SENSITIVE_KEY_NAMES or bool(parts & _SENSITIVE_KEY_PARTS) def _contains_sensitive_field(value: Any) -> bool: if isinstance(value, dict): return any( _is_sensitive_key(key) or _contains_sensitive_field(child) for key, child in value.items() ) if isinstance(value, (list, tuple)): return any(_contains_sensitive_field(child) for child in value) return False def _scrub_sensitive_fields(value: Any) -> Any: """Recursively remove secret-bearing fields without copying their values.""" if isinstance(value, dict): return { key: _scrub_sensitive_fields(child) for key, child in value.items() if not _is_sensitive_key(key) } if isinstance(value, list): return [_scrub_sensitive_fields(child) for child in value] if isinstance(value, tuple): return tuple(_scrub_sensitive_fields(child) for child in value) return value def _validated_fingerprints(values: dict[str, str] | None) -> dict[str, str]: """Allow only the KDF output format emitted by ``cred_fingerprint``.""" safe: dict[str, str] = {} for label, fingerprint in (values or {}).items(): if not isinstance(label, str) or _FINGERPRINT_LABEL_RE.fullmatch(label) is None: continue if fingerprint == "absent": safe[label] = "absent" elif isinstance(fingerprint, str) and _FINGERPRINT_RE.fullmatch(fingerprint): safe[label] = fingerprint else: safe[label] = "invalid-fingerprint" return safe 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 at every nesting level. Direct callers receive the same protection # as gate_write callers, including secrets hidden in list/dict children. safe_action = _scrub_sensitive_fields(action or {}) safe_result = _scrub_sensitive_fields(result_summary or {}) 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": _validated_fingerprints(cred_fingerprints), "result": safe_result, "emitted_at": _now(), "doctrine": "v11 — Λ-gate + DSSE/Khipu receipt on every write; no committed keys; trust never 100%", } canonical_body = json.dumps(body, sort_keys=True, separators=(",", ":")).encode() receipt_hash = "sha256:" + sha256_content_address( canonical_body, purpose="khipu-receipt" ) 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 = _contains_sensitive_field(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"]