R15D: bind MCP product runtime to existing Windows TPM anchor

continuityos/gate/windows_tpm_product_runtime.py

@@ -0,0 +1,144 @@
+"""Read-only R15D product binding for an already provisioned Windows TPM anchor.
+
+This module never provisions, defines, primes, clears, undefines, or extends TPM
+state. It reconstructs the reviewed R15B profile from explicit external
+configuration, verifies the controller pin and DPAPI custody envelope, performs
+one read-only hardware snapshot, and returns the R15 monotonic anchor adapter.
+"""
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+from pathlib import Path
+from typing import Any, Callable
+
+from .monotonic_anchor import MonotonicExecutionAnchor
+from .tpm2_provider_binding import BoundTpm2NvExtendProvider
+from .windows_tpm_dpapi import DpapiIndexAuthStore
+from .windows_tpm_provisioning import build_reviewed_plan
+from .windows_tpm_runtime import WindowsTbsNvExtendBackend
+
+_HEX = frozenset("0123456789abcdef")
+_CUSTODY_SCHEMA = "continuityos.r15b.windows-dpapi-index-auth/v1"
+
+
+class WindowsTpmProductRuntimeError(RuntimeError):
+    """Explicit R15 product runtime configuration or binding is invalid."""
+
+
+def _sha256_bytes(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def _require_sha256(value: Any, label: str) -> str:
+    if (
+        not isinstance(value, str)
+        or len(value) != 64
+        or set(value) - _HEX
+    ):
+        raise WindowsTpmProductRuntimeError(f"{label} is not lowercase SHA-256")
+    return value
+
+
+def _read_absolute_file(path: str, label: str) -> tuple[Path, bytes]:
+    if not isinstance(path, str) or not path.strip():
+        raise WindowsTpmProductRuntimeError(f"{label} path is required")
+    target = Path(path).expanduser()
+    if not target.is_absolute():
+        raise WindowsTpmProductRuntimeError(f"{label} path must be absolute")
+    try:
+        data = target.read_bytes()
+    except OSError as exc:
+        raise WindowsTpmProductRuntimeError(f"{label} is unreadable") from exc
+    if not data:
+        raise WindowsTpmProductRuntimeError(f"{label} is empty")
+    return target, data
+
+
+def _json_object(data: bytes, label: str) -> dict[str, Any]:
+    try:
+        value = json.loads(data.decode("ascii"))
+    except (UnicodeError, json.JSONDecodeError) as exc:
+        raise WindowsTpmProductRuntimeError(f"{label} is not canonical ASCII JSON") from exc
+    if type(value) is not dict:
+        raise WindowsTpmProductRuntimeError(f"{label} must be a JSON object")
+    return value
+
+
+def _custody_metadata(path: str) -> tuple[Path, dict[str, Any]]:
+    target, raw = _read_absolute_file(path, "R15 DPAPI custody")
+    payload = _json_object(raw, "R15 DPAPI custody")
+    expected = {
+        "schema", "provider", "nv_index", "ek_public_sha256", "ciphertext_b64"
+    }
+    if set(payload) != expected:
+        raise WindowsTpmProductRuntimeError("R15 DPAPI custody envelope schema is invalid")
+    if (
+        payload["schema"] != _CUSTODY_SCHEMA
+        or payload["provider"] != "WINDOWS_DPAPI_CURRENT_USER"
+        or type(payload["nv_index"]) is not int
+    ):
+        raise WindowsTpmProductRuntimeError("R15 DPAPI custody envelope identity is invalid")
+    _require_sha256(payload["ek_public_sha256"], "R15 custody EK identity")
+    try:
+        ciphertext = base64.b64decode(payload["ciphertext_b64"], validate=True)
+    except (TypeError, ValueError) as exc:
+        raise WindowsTpmProductRuntimeError("R15 DPAPI custody ciphertext is invalid") from exc
+    if not ciphertext:
+        raise WindowsTpmProductRuntimeError("R15 DPAPI custody ciphertext is empty")
+    return target, payload
+
+
+def build_windows_r15_monotonic_anchor(
+    *,
+    controller_profile_path: str,
+    controller_profile_sha256: str,
+    custody_path: str,
+    backend_factory: Callable[[DpapiIndexAuthStore], Any] | None = None,
+) -> MonotonicExecutionAnchor:
+    """Bind product runtime to existing R15 hardware without any TPM mutation."""
+    expected_controller_sha = _require_sha256(
+        controller_profile_sha256, "R15 controller profile pin"
+    )
+    controller_path, controller_raw = _read_absolute_file(
+        controller_profile_path, "R15 controller profile"
+    )
+    observed_controller_sha = _sha256_bytes(controller_raw)
+    if observed_controller_sha != expected_controller_sha:
+        raise WindowsTpmProductRuntimeError(
+            "R15 controller profile SHA-256 differs from explicit pin"
+        )
+    controller = _json_object(controller_raw, "R15 controller profile")
+
+    custody_target, custody = _custody_metadata(custody_path)
+    nv_index = custody["nv_index"]
+    if controller.get("nv_index") != nv_index:
+        raise WindowsTpmProductRuntimeError(
+            "R15 controller profile NV index differs from custody"
+        )
+
+    plan = build_reviewed_plan(
+        ek_public_sha256=custody["ek_public_sha256"],
+        nv_index=nv_index,
+    )
+    profile = plan.active_profile
+    if controller != profile.binding_document():
+        raise WindowsTpmProductRuntimeError(
+            "R15 controller profile differs from reviewed Windows TPM binding"
+        )
+
+    secret_store = DpapiIndexAuthStore(
+        custody_target,
+        nv_index=nv_index,
+        ek_public_sha256=custody["ek_public_sha256"],
+    )
+    factory = backend_factory or WindowsTbsNvExtendBackend
+    backend = factory(secret_store)
+    provider = BoundTpm2NvExtendProvider(backend, profile=profile)
+
+    # Mandatory startup proof is read-only: NV_Read + public identity read.
+    provider.read_snapshot()
+    return MonotonicExecutionAnchor(
+        provider, profile=profile.anchor_profile()
+    )

continuityos/mcp_server.py

@@ -27,6 +27,15 @@
 PROTOCOL = "2024-11-05"
 _PRODUCT_LEDGER = _Ledger
 
+
+def _build_windows_r15_monotonic_anchor(**kwargs):
+    # Lazy import keeps ordinary R14/non-Windows product startup unchanged.
+    from .gate.windows_tpm_product_runtime import (
+        build_windows_r15_monotonic_anchor,
+    )
+    return build_windows_r15_monotonic_anchor(**kwargs)
+
+
 TOOLS = [
  {"name":"remember","description":"Store a durable memory. Use for facts about the user, projects, rules, decisions you should recall later.",
   "inputSchema":{"type":"object","properties":{
@@ -96,7 +105,13 @@
 ]
 
 class Server:
-    def __init__(self, db=None, policy_path: str = "", db_source: str = ""):
+    def __init__(
+        self, db=None, policy_path: str = "", db_source: str = "",
+        *, governance_witness_path: str = "",
+        r15_controller_profile_path: str = "",
+        r15_controller_profile_sha256: str = "",
+        r15_custody_path: str = "",
+    ):
         resolved = resolve_memory_db(db)
         db = resolved["path"]
         self.db_path = db
@@ -123,13 +138,47 @@ def __init__(self, db=None, policy_path: str = "", db_source: str = ""):
         runtime_policy = policy_path or _discover_policy(os.path.expanduser("~/.continuityos"))
         self.policy = _load_policy(runtime_policy)
         governance_root = os.path.expanduser("~/.continuityos")
+        witness_path = (
+            os.path.abspath(os.path.expanduser(governance_witness_path))
+            if governance_witness_path
+            else os.path.join(governance_root, "governance.witness.json")
+        )
         self._governance_paths = {
             "registry_path": os.path.join(governance_root, "gate_broker.db"),
             "ledger_path": os.path.join(governance_root, "ledger.db"),
-            "witness_path": os.path.join(
-                governance_root, "governance.witness.json"
-            ),
+            "witness_path": witness_path,
         }
+        self._monotonic_anchor = None
+        self._r15_runtime_error = ""
+        r15_values = {
+            "controller_profile_path": r15_controller_profile_path,
+            "controller_profile_sha256": r15_controller_profile_sha256,
+            "custody_path": r15_custody_path,
+        }
+        supplied = {key for key, value in r15_values.items() if value}
+        if supplied:
+            missing = sorted(set(r15_values) - supplied)
+            if missing:
+                self._r15_runtime_error = (
+                    "explicit R15 runtime configuration is incomplete; missing "
+                    + ", ".join(missing)
+                )
+            elif not governance_witness_path:
+                self._r15_runtime_error = (
+                    "explicit R15 runtime requires an explicit external governance witness path"
+                )
+            else:
+                try:
+                    self._monotonic_anchor = _build_windows_r15_monotonic_anchor(
+                        controller_profile_path=r15_controller_profile_path,
+                        controller_profile_sha256=r15_controller_profile_sha256,
+                        custody_path=r15_custody_path,
+                    )
+                except Exception as exc:
+                    self._r15_runtime_error = (
+                        "R15 runtime binding failed: "
+                        f"{type(exc).__name__}: {exc}"
+                    )
         self._broker = None
         self.turns = 0
         self.std = 10  # Safe Turn Depth: re-inject canon before omission-rules ("never do X") decay (long-session SRD research)
@@ -296,11 +345,15 @@ def _gate_broker(self):
             paths = getattr(self, "_governance_paths", None)
             if paths is None:
                 raise RuntimeError("product witness configuration unavailable")
+            runtime_error = getattr(self, "_r15_runtime_error", "")
+            if runtime_error:
+                raise RuntimeError(runtime_error)
             self._broker = _GateBroker(
                 **paths,
                 db=self.db_path,
                 policy_snapshot=self.policy,
                 context_error=self._governance_context_error,
+                monotonic_anchor=getattr(self, "_monotonic_anchor", None),
             )
         return self._broker
 
@@ -311,8 +364,30 @@ def main():
     ap = argparse.ArgumentParser()
     ap.add_argument("--db", default=None)
     ap.add_argument("--policy", default="", help="Path to one JSON policy, or YAML when PyYAML is installed")
+    ap.add_argument(
+        "--governance-witness", default="",
+        help="Explicit external R14 witness path for product governance",
+    )
+    ap.add_argument(
+        "--r15-controller-profile", default="",
+        help="Explicit external R15 controller binding profile path",
+    )
+    ap.add_argument(
+        "--r15-controller-profile-sha256", default="",
+        help="Pinned lowercase SHA-256 of the R15 controller profile",
+    )
+    ap.add_argument(
+        "--r15-custody", default="",
+        help="Explicit DPAPI custody envelope path for the provisioned R15 NV index",
+    )
     a = ap.parse_args()
-    srv = Server(a.db, a.policy)
+    srv = Server(
+        a.db, a.policy,
+        governance_witness_path=a.governance_witness,
+        r15_controller_profile_path=a.r15_controller_profile,
+        r15_controller_profile_sha256=a.r15_controller_profile_sha256,
+        r15_custody_path=a.r15_custody,
+    )
     for line in sys.stdin:
         line = line.strip()
         if not line: continue