From 2189e488e4d4168537d93f96f8658f6fbebd0c32 Mon Sep 17 00:00:00 2001 From: bitmaster162 Date: Sun, 20 Sep 2026 20:56:10 +0700 Subject: [PATCH] R15D: bind MCP runtime to existing Windows TPM anchor --- .../gate/windows_tpm_product_runtime.py | 144 +++++++++ continuityos/mcp_server.py | 85 +++++- .../test_r15d_mcp_product_runtime_binding.py | 282 ++++++++++++++++++ 3 files changed, 506 insertions(+), 5 deletions(-) create mode 100644 continuityos/gate/windows_tpm_product_runtime.py create mode 100644 tests/test_r15d_mcp_product_runtime_binding.py diff --git a/continuityos/gate/windows_tpm_product_runtime.py b/continuityos/gate/windows_tpm_product_runtime.py new file mode 100644 index 0000000..8cde559 --- /dev/null +++ b/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() + ) diff --git a/continuityos/mcp_server.py b/continuityos/mcp_server.py index 481c727..bf25576 100644 --- a/continuityos/mcp_server.py +++ b/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 diff --git a/tests/test_r15d_mcp_product_runtime_binding.py b/tests/test_r15d_mcp_product_runtime_binding.py new file mode 100644 index 0000000..f5632b5 --- /dev/null +++ b/tests/test_r15d_mcp_product_runtime_binding.py @@ -0,0 +1,282 @@ +from __future__ import annotations + +import base64 +import hashlib +import json +from pathlib import Path + +import pytest + +import continuityos.mcp_server as mcp_server +from continuityos.gate.broker import GateBroker +from continuityos.gate.monotonic_anchor import ( + BoundMonotonicAnchorProfile, + MonotonicExecutionAnchor, +) +from continuityos.gate.windows_tpm_product_runtime import ( + WindowsTpmProductRuntimeError, + build_windows_r15_monotonic_anchor, +) +from continuityos.gate.windows_tpm_provisioning import build_reviewed_plan +from continuityos.memory import Memory + + +EK_SHA256 = "d4493341ea776e196234af9547d2a22118767eea3a0312d00e445fa497f6469c" + + +def _policy(path: Path) -> None: + path.write_text(json.dumps({"default_decision": "ALLOW"}), encoding="utf-8") + + +def _memory(path: Path) -> None: + memory = Memory(str(path)) + memory.store.con.close() + + +def _controller_and_custody(tmp_path: Path): + plan = build_reviewed_plan(ek_public_sha256=EK_SHA256) + controller = tmp_path / "controller.json" + controller_raw = json.dumps( + plan.active_profile.binding_document(), + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode("ascii") + controller.write_bytes(controller_raw) + custody = tmp_path / "custody.json" + custody.write_text( + json.dumps( + { + "schema": "continuityos.r15b.windows-dpapi-index-auth/v1", + "provider": "WINDOWS_DPAPI_CURRENT_USER", + "nv_index": plan.nv_index, + "ek_public_sha256": EK_SHA256, + "ciphertext_b64": base64.b64encode(b"encrypted").decode("ascii"), + }, + sort_keys=True, + separators=(",", ":"), + ), + encoding="ascii", + ) + return plan, controller, hashlib.sha256(controller_raw).hexdigest(), custody + + +class _ReadOnlyBackend: + def __init__(self, store, plan): + self.store = store + self.plan = plan + self.reads = 0 + + def read_snapshot(self, *, profile): + self.reads += 1 + public = self.plan.active_public + return { + "nv_index": self.plan.nv_index, + "tpma_nv_mask": self.plan.active_mask, + "auth_policy_sha256": None, + "backend_kind": profile.constraints.backend_kind, + "transport_identity": profile.constraints.transport_identity, + "nv_type": "TPM_NT_EXTEND", + "name_alg": "SHA256", + "data_size": 32, + "orderly": False, + "tpms_nv_public_marshaled": public, + "raw_tpm_name": b"\x00\x0b" + hashlib.sha256(public).digest(), + "observed_nv_extend_digest": "a" * 64, + } + + def extend_once(self, **_kwargs): + raise AssertionError("R15D startup must never extend TPM state") + + +def test_explicit_runtime_factory_is_readonly_and_pinned(tmp_path): + plan, controller, controller_sha, custody = _controller_and_custody(tmp_path) + backends = [] + + def factory(store): + backend = _ReadOnlyBackend(store, plan) + backends.append(backend) + return backend + + anchor = build_windows_r15_monotonic_anchor( + controller_profile_path=str(controller), + controller_profile_sha256=controller_sha, + custody_path=str(custody), + backend_factory=factory, + ) + + assert len(backends) == 1 + assert backends[0].reads == 1 + assert anchor.provider.read_snapshot()["observed_digest"] == "a" * 64 + assert backends[0].reads == 2 + + +def test_runtime_factory_rejects_controller_pin_mismatch_before_backend(tmp_path): + _plan, controller, _controller_sha, custody = _controller_and_custody(tmp_path) + called = False + + def factory(_store): + nonlocal called + called = True + raise AssertionError("backend must not be constructed after pin mismatch") + + with pytest.raises( + WindowsTpmProductRuntimeError, + match="controller profile SHA-256 differs", + ): + build_windows_r15_monotonic_anchor( + controller_profile_path=str(controller), + controller_profile_sha256="0" * 64, + custody_path=str(custody), + backend_factory=factory, + ) + assert called is False + + +class _FakeProvider: + def __init__(self): + self.digest = "0" * 64 + self.extend_calls = 0 + + def read_snapshot(self): + return { + "provider": "TPM2_NV_EXTEND", + "nv_public_sha256": "1" * 64, + "nv_name_sha256": "2" * 64, + "observed_digest": self.digest, + } + + def extend(self, *, expected_previous_digest, commitment_sha256): + assert self.digest == expected_previous_digest + self.extend_calls += 1 + self.digest = hashlib.sha256( + bytes.fromhex(self.digest) + bytes.fromhex(commitment_sha256) + ).hexdigest() + return self.read_snapshot() + + +def _activated_product_state(tmp_path: Path, monkeypatch): + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + root = home / ".continuityos" + root.mkdir() + paths = { + "registry_path": str(root / "gate_broker.db"), + "ledger_path": str(root / "ledger.db"), + "witness_path": str(tmp_path / "external" / "witness.json"), + } + r14 = GateBroker(**paths) + provider = _FakeProvider() + profile = BoundMonotonicAnchorProfile( + nv_public_sha256="1" * 64, + nv_name_sha256="2" * 64, + genesis_digest="0" * 64, + ) + anchor = MonotonicExecutionAnchor(provider, profile=profile) + with r14._ledger() as ledger: + anchor.bind_genesis(ledger) + before = ledger.frontier() + db = tmp_path / "memory.db" + policy = tmp_path / "policy.json" + _memory(db) + _policy(policy) + return home, paths, anchor, db, policy, before + + +def test_product_startup_and_restart_inject_existing_r15_anchor( + tmp_path, monkeypatch +): + _home, paths, anchor, db, policy, before = _activated_product_state( + tmp_path, monkeypatch + ) + calls = [] + + def build(**kwargs): + calls.append(kwargs) + return anchor + + monkeypatch.setattr(mcp_server, "_build_windows_r15_monotonic_anchor", build) + common = dict( + governance_witness_path=paths["witness_path"], + r15_controller_profile_path=str(tmp_path / "controller.json"), + r15_controller_profile_sha256="a" * 64, + r15_custody_path=str(tmp_path / "custody.json"), + ) + first = mcp_server.Server(str(db), str(policy), **common) + first_broker = first._gate_broker() + second = mcp_server.Server(str(db), str(policy), **common) + second_broker = second._gate_broker() + + assert len(calls) == 2 + assert first_broker.monotonic_anchor is anchor + assert second_broker.monotonic_anchor is anchor + assert first_broker.witness.path == __import__("os").path.normcase( + __import__("os").path.abspath(paths["witness_path"]) + ) + with second_broker._ledger() as ledger: + assert ledger.frontier() == before + assert anchor.validate_global(ledger)["anchor_generation"] == 1 + + +def test_product_binding_mismatch_holds_without_ledger_write( + tmp_path, monkeypatch +): + _home, paths, _anchor, db, policy, before = _activated_product_state( + tmp_path, monkeypatch + ) + + def fail(**_kwargs): + raise WindowsTpmProductRuntimeError("hardware binding mismatch") + + monkeypatch.setattr(mcp_server, "_build_windows_r15_monotonic_anchor", fail) + server = mcp_server.Server( + str(db), + str(policy), + governance_witness_path=paths["witness_path"], + r15_controller_profile_path=str(tmp_path / "controller.json"), + r15_controller_profile_sha256="a" * 64, + r15_custody_path=str(tmp_path / "custody.json"), + ) + result = json.loads(server.call( + "preflight_exec", + { + "request_id": "must-hold", + "argv": ["cmd.exe", "/c", "exit", "0"], + "cwd": str(tmp_path), + "paths": [], + }, + )) + assert result["state"] == "HELD" + assert any("hardware binding mismatch" in reason for reason in result["reasons"]) + with GateBroker(**paths, monotonic_anchor=_anchor)._ledger() as ledger: + assert ledger.frontier() == before + + +def test_incomplete_explicit_r15_configuration_holds(tmp_path, monkeypatch): + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + db = tmp_path / "memory.db" + policy = tmp_path / "policy.json" + _memory(db) + _policy(policy) + server = mcp_server.Server( + str(db), + str(policy), + governance_witness_path=str(tmp_path / "witness.json"), + r15_controller_profile_path=str(tmp_path / "controller.json"), + ) + result = json.loads(server.call( + "preflight_exec", + { + "request_id": "incomplete", + "argv": ["cmd.exe"], + "cwd": str(tmp_path), + "paths": [], + }, + )) + assert result["state"] == "HELD" + assert any("configuration is incomplete" in reason for reason in result["reasons"])