diff --git a/continuityos/gate/windows_tpm_dpapi.py b/continuityos/gate/windows_tpm_dpapi.py new file mode 100644 index 00000000..91f7d6c9 --- /dev/null +++ b/continuityos/gate/windows_tpm_dpapi.py @@ -0,0 +1,240 @@ +"""Windows DPAPI custody for one R15B TPM NV index authorization secret. + +This module never provisions or talks to the TPM. It stores only a DPAPI- +protected 32-byte index authValue plus public binding metadata. Plaintext +secret bytes are returned only as a mutable bytearray so callers can zero them. +""" +from __future__ import annotations + +import base64 +import ctypes +import hashlib +import json +import os +from pathlib import Path +import secrets +import sys +from typing import Any, Protocol + +_SCHEMA = "continuityos.r15b.windows-dpapi-index-auth/v1" +_ENTROPY_DOMAIN = b"continuityos.r15b.windows-dpapi-index-auth/v1\0" +_CRYPTPROTECT_UI_FORBIDDEN = 0x1 + + +class WindowsDpapiCustodyError(RuntimeError): + """DPAPI custody is unavailable, corrupt, or bound to another profile.""" + + +class _Protector(Protocol): + def protect(self, plaintext: bytearray, *, entropy: bytes) -> bytes: ... + def unprotect(self, ciphertext: bytes, *, entropy: bytes) -> bytearray: ... + + +class _DataBlob(ctypes.Structure): + _fields_ = [ + ("cbData", ctypes.c_uint32), + ("pbData", ctypes.POINTER(ctypes.c_ubyte)), + ] + + +def zeroize(value: bytearray) -> None: + """Best-effort in-place zeroization for mutable secret buffers.""" + if type(value) is not bytearray: + raise TypeError("zeroize requires bytearray") + for index in range(len(value)): + value[index] = 0 + + +def _require_windows() -> None: + if sys.platform != "win32": + raise WindowsDpapiCustodyError("Windows DPAPI requires win32") + + +def _entropy(*, nv_index: int, ek_public_sha256: str) -> bytes: + if type(nv_index) is not int or not 0x01000000 <= nv_index <= 0x01FFFFFF: + raise WindowsDpapiCustodyError("DPAPI NV index is invalid") + if ( + type(ek_public_sha256) is not str + or len(ek_public_sha256) != 64 + or any(ch not in "0123456789abcdef" for ch in ek_public_sha256) + ): + raise WindowsDpapiCustodyError("DPAPI EK identity is invalid") + return hashlib.sha256( + _ENTROPY_DOMAIN + + nv_index.to_bytes(4, "big") + + bytes.fromhex(ek_public_sha256) + ).digest() + + +def _blob_from_bytes(value: bytes | bytearray) -> tuple[_DataBlob, Any]: + if not value: + return _DataBlob(0, None), None + if type(value) is bytearray: + buffer = (ctypes.c_ubyte * len(value)).from_buffer(value) + else: + buffer = (ctypes.c_ubyte * len(value)).from_buffer_copy(value) + return _DataBlob(len(value), buffer), buffer + + +def _copy_blob(value: _DataBlob) -> bytes: + if value.cbData == 0: + return b"" + if not value.pbData: + raise WindowsDpapiCustodyError("DPAPI returned an invalid blob") + return bytes(ctypes.string_at(value.pbData, value.cbData)) + + +class _WindowsDpapiProtector: + def _crypt32(self): + _require_windows() + try: + dll = ctypes.WinDLL("crypt32.dll") + kernel = ctypes.WinDLL("kernel32.dll") + except Exception as exc: + raise WindowsDpapiCustodyError("Windows DPAPI DLLs are unavailable") from exc + kernel.LocalFree.argtypes = [ctypes.c_void_p] + kernel.LocalFree.restype = ctypes.c_void_p + return dll, kernel + + def protect(self, plaintext: bytearray, *, entropy: bytes) -> bytes: + if type(plaintext) is not bytearray or len(plaintext) != 32: + raise WindowsDpapiCustodyError("index authValue must be exactly 32 bytes") + dll, kernel = self._crypt32() + dll.CryptProtectData.argtypes = [ + ctypes.POINTER(_DataBlob), ctypes.c_wchar_p, + ctypes.POINTER(_DataBlob), ctypes.c_void_p, ctypes.c_void_p, + ctypes.c_uint32, ctypes.POINTER(_DataBlob), + ] + dll.CryptProtectData.restype = ctypes.c_bool + input_blob, input_keepalive = _blob_from_bytes(plaintext) + entropy_blob, entropy_keepalive = _blob_from_bytes(entropy) + output = _DataBlob() + try: + ok = dll.CryptProtectData( + ctypes.byref(input_blob), None, ctypes.byref(entropy_blob), + None, None, _CRYPTPROTECT_UI_FORBIDDEN, ctypes.byref(output), + ) + if not ok: + raise WindowsDpapiCustodyError( + f"CryptProtectData failed with WinError {ctypes.get_last_error()}" + ) + return _copy_blob(output) + finally: + _ = input_keepalive, entropy_keepalive + if output.pbData: + kernel.LocalFree(output.pbData) + + def unprotect(self, ciphertext: bytes, *, entropy: bytes) -> bytearray: + if type(ciphertext) is not bytes or not ciphertext: + raise WindowsDpapiCustodyError("DPAPI ciphertext is invalid") + dll, kernel = self._crypt32() + dll.CryptUnprotectData.argtypes = [ + ctypes.POINTER(_DataBlob), ctypes.c_void_p, + ctypes.POINTER(_DataBlob), ctypes.c_void_p, ctypes.c_void_p, + ctypes.c_uint32, ctypes.POINTER(_DataBlob), + ] + dll.CryptUnprotectData.restype = ctypes.c_bool + input_blob, input_keepalive = _blob_from_bytes(ciphertext) + entropy_blob, entropy_keepalive = _blob_from_bytes(entropy) + output = _DataBlob() + try: + ok = dll.CryptUnprotectData( + ctypes.byref(input_blob), None, ctypes.byref(entropy_blob), + None, None, _CRYPTPROTECT_UI_FORBIDDEN, ctypes.byref(output), + ) + if not ok: + raise WindowsDpapiCustodyError( + f"CryptUnprotectData failed with WinError {ctypes.get_last_error()}" + ) + plaintext = bytearray(_copy_blob(output)) + if len(plaintext) != 32: + zeroize(plaintext) + raise WindowsDpapiCustodyError("DPAPI plaintext length is invalid") + return plaintext + finally: + _ = input_keepalive, entropy_keepalive + if output.pbData: + kernel.LocalFree(output.pbData) + + +class DpapiIndexAuthStore: + """Persist one DPAPI-protected index authValue with exact public binding.""" + + def __init__( + self, + path: str | os.PathLike[str], + *, + nv_index: int, + ek_public_sha256: str, + protector: _Protector | None = None, + ) -> None: + self.path = Path(path) + self.nv_index = nv_index + self.ek_public_sha256 = ek_public_sha256 + self._entropy = _entropy( + nv_index=nv_index, ek_public_sha256=ek_public_sha256 + ) + self._protector = protector or _WindowsDpapiProtector() + + def exists(self) -> bool: + return self.path.is_file() + + def _envelope(self, ciphertext: bytes) -> dict[str, Any]: + return { + "schema": _SCHEMA, + "provider": "WINDOWS_DPAPI_CURRENT_USER", + "nv_index": self.nv_index, + "ek_public_sha256": self.ek_public_sha256, + "ciphertext_b64": base64.b64encode(ciphertext).decode("ascii"), + } + + def store_once(self, secret: bytearray) -> None: + if type(secret) is not bytearray or len(secret) != 32: + raise WindowsDpapiCustodyError("index authValue must be exactly 32 bytes") + if self.exists(): + raise WindowsDpapiCustodyError("DPAPI custody file already exists") + ciphertext = self._protector.protect(secret, entropy=self._entropy) + payload = json.dumps( + self._envelope(ciphertext), + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ).encode("ascii") + self.path.parent.mkdir(parents=True, exist_ok=True) + temporary = self.path.with_name( + f".{self.path.name}.{secrets.token_hex(8)}.tmp" + ) + try: + with open(temporary, "xb") as handle: + handle.write(payload) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, self.path) + finally: + if temporary.exists(): + temporary.unlink() + + def load(self) -> bytearray: + try: + payload = json.loads(self.path.read_text(encoding="ascii")) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise WindowsDpapiCustodyError("DPAPI custody envelope is unreadable") from exc + expected = { + "schema", "provider", "nv_index", "ek_public_sha256", "ciphertext_b64" + } + if type(payload) is not dict or set(payload) != expected: + raise WindowsDpapiCustodyError("DPAPI custody envelope schema is invalid") + if ( + payload["schema"] != _SCHEMA + or payload["provider"] != "WINDOWS_DPAPI_CURRENT_USER" + or payload["nv_index"] != self.nv_index + or payload["ek_public_sha256"] != self.ek_public_sha256 + ): + raise WindowsDpapiCustodyError("DPAPI custody binding differs") + try: + ciphertext = base64.b64decode( + payload["ciphertext_b64"], validate=True + ) + except (TypeError, ValueError) as exc: + raise WindowsDpapiCustodyError("DPAPI ciphertext encoding is invalid") from exc + return self._protector.unprotect(ciphertext, entropy=self._entropy) diff --git a/continuityos/gate/windows_tpm_provisioning.py b/continuityos/gate/windows_tpm_provisioning.py new file mode 100644 index 00000000..42df849a --- /dev/null +++ b/continuityos/gate/windows_tpm_provisioning.py @@ -0,0 +1,696 @@ +"""Offline-only R15B Windows TPM provisioning ceremony. + +This module is intentionally not imported by runtime, MCP, broker, or CLI +surfaces. It can define exactly one reviewed NV_EXTEND index and perform one +domain-separated primer extend, but only when called with its exact generated +hardware-write authorization token. +""" +from __future__ import annotations + +from dataclasses import dataclass +import ctypes +import hashlib +import json +import struct +import sys +from typing import Any, Callable + +from .tpm2_provider_binding import ( + BoundTpm2NvExtendProfile, + Tpm2NvBindingConstraints, +) +from .windows_tpm_dpapi import DpapiIndexAuthStore, zeroize +from .windows_tpm_readonly import ( + TBS_COMMAND_LOCALITY_ZERO, + TBS_COMMAND_PRIORITY_LOW, + TBS_CONTEXT_FLAGS, + TBS_SUCCESS, + TPM_VERSION_20, + _MAX_TPM_RESPONSE, + _TbsContextParams2, + _load_tbs, + enumerate_nv_handles, + read_nv_public, +) +from .windows_tpm_runtime import ( + TPM_CC_NV_EXTEND, + TPM_RS_PW, + TPM_ST_SESSIONS, + WindowsTbsNvExtendBackend, + _require_pw_response, +) + +TPM_CC_NV_DEFINE_SPACE = 0x0000012A +TPM_RH_OWNER = 0x40000001 +TBS_OWNERAUTH_TYPE_STORAGE_20 = 13 + +TPMA_NV_AUTHWRITE = 0x00000004 +TPMA_NV_NT_EXTEND = 0x00000040 +TPMA_NV_AUTHREAD = 0x00040000 +TPMA_NV_NO_DA = 0x02000000 +TPMA_NV_WRITTEN = 0x20000000 + +DEFAULT_DEFINITION_MASK = ( + TPMA_NV_AUTHWRITE | TPMA_NV_NT_EXTEND | TPMA_NV_AUTHREAD | TPMA_NV_NO_DA +) +DEFAULT_NV_INDEX = 0x01500020 +_BACKEND_KIND = "WINDOWS_TBS_TPM2_NV_V1" +_PRIMER_DOMAIN = "continuityos.r15b.nv-extend-primer.v1" + +_ALLOWED_PROVISIONING_COMMANDS = frozenset({ + TPM_CC_NV_DEFINE_SPACE, TPM_CC_NV_EXTEND +}) + + +class WindowsTpmProvisioningError(RuntimeError): + """Offline provisioning failed closed.""" + + +STATE_EMPTY = "EMPTY" +STATE_CUSTODY_ONLY = "CUSTODY_ONLY" +STATE_DEFINED_UNPRIMED = "DEFINED_UNPRIMED" +STATE_PRIMED_UNVERIFIED = "PRIMED_UNVERIFIED" +STATE_PRIMED_VERIFIED = "PRIMED_VERIFIED" + + +@dataclass(frozen=True) +class ProvisioningState: + state: str + custody_present: bool + handle_present: bool + public_phase: str | None + digest_verified: bool + + def public_receipt(self) -> dict[str, Any]: + return { + "state": self.state, + "custody_present": self.custody_present, + "handle_present": self.handle_present, + "public_phase": self.public_phase, + "digest_verified": self.digest_verified, + } + + +def _canonical(value: Any) -> bytes: + return json.dumps( + value, sort_keys=True, ensure_ascii=True, separators=(",", ":") + ).encode("ascii") + + +def _sha256_hex(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _require_sha256(value: str, label: str) -> str: + if ( + type(value) is not str + or len(value) != 64 + or any(ch not in "0123456789abcdef" for ch in value) + ): + raise WindowsTpmProvisioningError(f"{label} is invalid") + return value + + +def _marshal_nv_public(*, nv_index: int, attributes: int) -> bytes: + return b"".join(( + nv_index.to_bytes(4, "big"), + (0x000B).to_bytes(2, "big"), + attributes.to_bytes(4, "big"), + (0).to_bytes(2, "big"), + (32).to_bytes(2, "big"), + )) + + +def _raw_name(public: bytes) -> bytes: + return b"\x00\x0b" + hashlib.sha256(public).digest() + + +@dataclass(frozen=True) +class WindowsTpmNvProvisioningPlan: + """Immutable reviewed inputs and predicted post-primer profile.""" + + nv_index: int + ek_public_sha256: str + definition_mask: int = DEFAULT_DEFINITION_MASK + backend_kind: str = _BACKEND_KIND + + def __post_init__(self) -> None: + if type(self.nv_index) is not int or not 0x01000000 <= self.nv_index <= 0x01FFFFFF: + raise WindowsTpmProvisioningError("provisioning NV index is invalid") + _require_sha256(self.ek_public_sha256, "provisioning EK identity") + if self.definition_mask != DEFAULT_DEFINITION_MASK: + raise WindowsTpmProvisioningError("provisioning TPMA_NV mask is unreviewed") + if self.backend_kind != _BACKEND_KIND: + raise WindowsTpmProvisioningError("provisioning backend kind is unreviewed") + + @property + def active_mask(self) -> int: + return self.definition_mask | TPMA_NV_WRITTEN + + @property + def transport_identity(self) -> str: + return ( + "windows-tbs:v1;locality=0;" + f"ekpub-sha256={self.ek_public_sha256};" + "vendor=INTC;fw=403.1.0.0" + ) + + @property + def definition_public(self) -> bytes: + return _marshal_nv_public( + nv_index=self.nv_index, attributes=self.definition_mask + ) + + @property + def active_public(self) -> bytes: + return _marshal_nv_public( + nv_index=self.nv_index, attributes=self.active_mask + ) + + @property + def primer_document(self) -> dict[str, str]: + return { + "domain": _PRIMER_DOMAIN, + "device_ek_sha256": self.ek_public_sha256, + "nv_index": f"0x{self.nv_index:08x}", + "definition_nv_public_sha256": _sha256_hex(self.definition_public), + } + + @property + def primer_commitment(self) -> bytes: + return hashlib.sha256(_canonical(self.primer_document)).digest() + + @property + def primed_genesis_digest(self) -> str: + return hashlib.sha256(bytes(32) + self.primer_commitment).hexdigest() + + @property + def active_profile(self) -> BoundTpm2NvExtendProfile: + constraints = Tpm2NvBindingConstraints( + nv_index=self.nv_index, + tpma_nv_mask=self.active_mask, + auth_policy_sha256=None, + backend_kind=self.backend_kind, + transport_identity=self.transport_identity, + ) + return BoundTpm2NvExtendProfile( + constraints=constraints, + nv_public_sha256=_sha256_hex(self.active_public), + nv_name_sha256=_sha256_hex(_raw_name(self.active_public)), + genesis_digest=self.primed_genesis_digest, + ) + + def public_document(self) -> dict[str, Any]: + profile = self.active_profile + return { + "schema": "continuityos.r15b.windows-tpm-provisioning-plan/v1", + "nv_index": f"0x{self.nv_index:08x}", + "define_hierarchy": "TPM_RH_OWNER", + "definition_mask": f"0x{self.definition_mask:08x}", + "active_mask": f"0x{self.active_mask:08x}", + "definition_public_sha256": _sha256_hex(self.definition_public), + "definition_name_sha256": _sha256_hex(_raw_name(self.definition_public)), + "active_public_sha256": profile.nv_public_sha256, + "active_name_sha256": profile.nv_name_sha256, + "primer_sha256": self.primer_commitment.hex(), + "primed_genesis_digest": self.primed_genesis_digest, + "backend_kind": self.backend_kind, + "transport_identity": self.transport_identity, + "active_profile_binding_sha256": profile.binding_sha256(), + } + + @property + def fingerprint(self) -> str: + return hashlib.sha256(_canonical(self.public_document())).hexdigest() + + @property + def hardware_write_authorization_token(self) -> str: + """Return the reviewed-plan confirmation token. + + This deterministic value is not a secret or an authentication + credential. Human execution authority remains external to this module. + """ + return ( + "APPROVE_CONTINUITYOS_R15B_HARDWARE_PROVISION_" + + self.fingerprint[:16].upper() + ) + + +def build_reviewed_plan( + *, + ek_public_sha256: str, + nv_index: int = DEFAULT_NV_INDEX, +) -> WindowsTpmNvProvisioningPlan: + return WindowsTpmNvProvisioningPlan( + nv_index=nv_index, ek_public_sha256=ek_public_sha256 + ) + + +def _pw_area(auth_value: bytearray) -> bytes: + if type(auth_value) is not bytearray or not 1 <= len(auth_value) <= 64: + raise WindowsTpmProvisioningError("provisioning authorization is invalid") + return b"".join(( + struct.pack(">I", TPM_RS_PW), + struct.pack(">H", 0), + b"\x00", + struct.pack(">H", len(auth_value)), + bytes(auth_value), + )) + + +def _offline_session_command( + *, + command_code: int, + handles: tuple[int, ...], + auth_value: bytearray, + parameters: bytes, +) -> bytes: + if command_code not in _ALLOWED_PROVISIONING_COMMANDS: + raise WindowsTpmProvisioningError( + "TPM command is outside offline provisioning allowlist" + ) + handle_bytes = b"".join(struct.pack(">I", item) for item in handles) + auth = _pw_area(auth_value) + body = handle_bytes + struct.pack(">I", len(auth)) + auth + parameters + return struct.pack( + ">HII", TPM_ST_SESSIONS, 10 + len(body), command_code + ) + body + + +class _OfflineTbsProvisioningTransport: + """Private transport with only NV_DefineSpace and one NV_Extend primitive.""" + + def _submit(self, command: bytes) -> bytes: + if type(command) is not bytes or len(command) < 10: + raise WindowsTpmProvisioningError("provisioning TPM command is invalid") + _, declared_size, command_code = struct.unpack(">HII", command[:10]) + if ( + declared_size != len(command) + or command_code not in _ALLOWED_PROVISIONING_COMMANDS + ): + raise WindowsTpmProvisioningError( + "provisioning TPM command is outside allowlist" + ) + dll = _load_tbs() + params = _TbsContextParams2(TPM_VERSION_20, TBS_CONTEXT_FLAGS) + context = ctypes.c_void_p() + result = dll.Tbsi_Context_Create( + ctypes.byref(params), ctypes.byref(context) + ) + if result != TBS_SUCCESS or not context.value: + raise WindowsTpmProvisioningError( + f"TBS provisioning context failed with 0x{result:08x}" + ) + try: + in_buffer = (ctypes.c_ubyte * len(command)).from_buffer_copy(command) + out_buffer = (ctypes.c_ubyte * _MAX_TPM_RESPONSE)() + out_size = ctypes.c_uint32(len(out_buffer)) + result = dll.Tbsip_Submit_Command( + context, + TBS_COMMAND_LOCALITY_ZERO, + TBS_COMMAND_PRIORITY_LOW, + in_buffer, + len(command), + out_buffer, + ctypes.byref(out_size), + ) + if result != TBS_SUCCESS: + raise WindowsTpmProvisioningError( + f"TBS provisioning submit failed with 0x{result:08x}" + ) + return _require_pw_response(bytes(out_buffer[: out_size.value])) + finally: + close_result = dll.Tbsip_Context_Close(context) + if close_result != TBS_SUCCESS: + raise WindowsTpmProvisioningError( + f"TBS provisioning close failed with 0x{close_result:08x}" + ) + + def define_space( + self, + *, + owner_auth: bytearray, + index_auth: bytearray, + plan: WindowsTpmNvProvisioningPlan, + ) -> None: + parameters = ( + struct.pack(">H", len(index_auth)) + + bytes(index_auth) + + struct.pack(">H", len(plan.definition_public)) + + plan.definition_public + ) + command = _offline_session_command( + command_code=TPM_CC_NV_DEFINE_SPACE, + handles=(TPM_RH_OWNER,), + auth_value=owner_auth, + parameters=parameters, + ) + if self._submit(command) != b"": + raise WindowsTpmProvisioningError( + "NV_DefineSpace returned unexpected parameters" + ) + + def primer_extend( + self, + *, + index_auth: bytearray, + plan: WindowsTpmNvProvisioningPlan, + ) -> None: + command = _offline_session_command( + command_code=TPM_CC_NV_EXTEND, + handles=(plan.nv_index, plan.nv_index), + auth_value=index_auth, + parameters=( + struct.pack(">H", len(plan.primer_commitment)) + + plan.primer_commitment + ), + ) + if self._submit(command) != b"": + raise WindowsTpmProvisioningError( + "primer NV_Extend returned unexpected parameters" + ) + + +def _generate_index_auth_windows() -> bytearray: + if sys.platform != "win32": + raise WindowsTpmProvisioningError("Windows CSPRNG requires win32") + bcrypt = ctypes.WinDLL("bcrypt.dll") + bcrypt.BCryptGenRandom.argtypes = [ + ctypes.c_void_p, ctypes.c_void_p, ctypes.c_ulong, ctypes.c_ulong + ] + bcrypt.BCryptGenRandom.restype = ctypes.c_long + buffer = (ctypes.c_ubyte * 32)() + status = bcrypt.BCryptGenRandom(None, buffer, 32, 0x00000002) + if status != 0: + raise WindowsTpmProvisioningError( + f"BCryptGenRandom failed with NTSTATUS 0x{status & 0xffffffff:08x}" + ) + value = bytearray(buffer[:]) + for index in range(len(buffer)): + buffer[index] = 0 + return value + + +def _load_storage_owner_auth() -> bytearray: + dll = _load_tbs() + dll.Tbsi_Get_OwnerAuth.argtypes = [ + ctypes.c_void_p, ctypes.c_uint32, ctypes.c_void_p, + ctypes.POINTER(ctypes.c_uint32), + ] + dll.Tbsi_Get_OwnerAuth.restype = ctypes.c_uint32 + params = _TbsContextParams2(TPM_VERSION_20, TBS_CONTEXT_FLAGS) + context = ctypes.c_void_p() + result = dll.Tbsi_Context_Create( + ctypes.byref(params), ctypes.byref(context) + ) + if result != TBS_SUCCESS or not context.value: + raise WindowsTpmProvisioningError( + f"TBS ownerAuth context failed with 0x{result:08x}" + ) + try: + length = ctypes.c_uint32(0) + result = dll.Tbsi_Get_OwnerAuth( + context, TBS_OWNERAUTH_TYPE_STORAGE_20, None, ctypes.byref(length) + ) + if result != TBS_SUCCESS or not 1 <= length.value <= 64: + raise WindowsTpmProvisioningError( + f"TBS storage ownerAuth unavailable: 0x{result:08x}" + ) + buffer = (ctypes.c_ubyte * length.value)() + result = dll.Tbsi_Get_OwnerAuth( + context, + TBS_OWNERAUTH_TYPE_STORAGE_20, + buffer, + ctypes.byref(length), + ) + if result != TBS_SUCCESS: + raise WindowsTpmProvisioningError( + f"TBS storage ownerAuth read failed: 0x{result:08x}" + ) + value = bytearray(buffer[: length.value]) + for index in range(len(buffer)): + buffer[index] = 0 + return value + finally: + close_result = dll.Tbsip_Context_Close(context) + if close_result != TBS_SUCCESS: + raise WindowsTpmProvisioningError( + f"TBS ownerAuth close failed with 0x{close_result:08x}" + ) + + +def _require_public( + value: dict[str, Any], + *, + plan: WindowsTpmNvProvisioningPlan, + active: bool, +) -> None: + expected_public = plan.active_public if active else plan.definition_public + expected_mask = plan.active_mask if active else plan.definition_mask + if ( + value.get("nv_index") != plan.nv_index + or value.get("tpma_nv_mask") != expected_mask + or value.get("nv_type") != "TPM_NT_EXTEND" + or value.get("data_size") != 32 + or value.get("orderly") is not False + or value.get("auth_policy_sha256") is not None + or value.get("tpms_nv_public_marshaled") != expected_public + or value.get("raw_tpm_name") != _raw_name(expected_public) + or value.get("name_verified") is not True + ): + label = "active" if active else "definition" + raise WindowsTpmProvisioningError( + f"{label} NV public identity differs from reviewed plan" + ) + + +class OfflineWindowsTpmNvProvisioner: + """Crash-resumable forward-only offline provisioning ceremony.""" + + def __init__( + self, + *, + transport: Any | None = None, + handle_reader: Callable[[], list[int]] | None = None, + public_reader: Callable[[int], dict[str, Any]] | None = None, + owner_auth_loader: Callable[[], bytearray] | None = None, + runtime_backend_factory: Callable[..., Any] | None = None, + secret_generator: Callable[[], bytearray] | None = None, + ) -> None: + self._transport = transport or _OfflineTbsProvisioningTransport() + self._handle_reader = handle_reader or enumerate_nv_handles + self._public_reader = public_reader or read_nv_public + self._owner_auth_loader = owner_auth_loader or _load_storage_owner_auth + self._runtime_backend_factory = ( + runtime_backend_factory or WindowsTbsNvExtendBackend + ) + self._secret_generator = secret_generator or _generate_index_auth_windows + + def _active_digest( + self, + *, + plan: WindowsTpmNvProvisioningPlan, + secret_store: DpapiIndexAuthStore, + ) -> str: + backend = self._runtime_backend_factory(secret_store) + snapshot = backend.read_snapshot(profile=plan.active_profile) + digest = snapshot.get("observed_nv_extend_digest") + if type(digest) is not str: + raise WindowsTpmProvisioningError( + "active NV digest verification returned an invalid snapshot" + ) + return digest + + def inspect_state( + self, + *, + plan: WindowsTpmNvProvisioningPlan, + secret_store: DpapiIndexAuthStore, + ) -> ProvisioningState: + """Classify durable custody plus TPM public state without mutation.""" + custody = secret_store.exists() + handles = self._handle_reader() + if type(handles) is not list or any(type(item) is not int for item in handles): + raise WindowsTpmProvisioningError( + "provisioning handle inventory is invalid" + ) + present = plan.nv_index in handles + if not present: + return ProvisioningState( + state=STATE_CUSTODY_ONLY if custody else STATE_EMPTY, + custody_present=custody, + handle_present=False, + public_phase=None, + digest_verified=False, + ) + if not custody: + raise WindowsTpmProvisioningError( + "reviewed NV index exists without matching DPAPI custody" + ) + + public = self._public_reader(plan.nv_index) + try: + _require_public(public, plan=plan, active=False) + except WindowsTpmProvisioningError: + pass + else: + return ProvisioningState( + state=STATE_DEFINED_UNPRIMED, + custody_present=True, + handle_present=True, + public_phase="DEFINITION", + digest_verified=False, + ) + + try: + _require_public(public, plan=plan, active=True) + except WindowsTpmProvisioningError as exc: + raise WindowsTpmProvisioningError( + "existing NV public identity is outside reviewed provisioning states" + ) from exc + + try: + digest = self._active_digest(plan=plan, secret_store=secret_store) + except Exception: + return ProvisioningState( + state=STATE_PRIMED_UNVERIFIED, + custody_present=True, + handle_present=True, + public_phase="ACTIVE", + digest_verified=False, + ) + if digest != plan.primed_genesis_digest: + raise WindowsTpmProvisioningError( + "active NV digest differs from reviewed primed genesis" + ) + return ProvisioningState( + state=STATE_PRIMED_VERIFIED, + custody_present=True, + handle_present=True, + public_phase="ACTIVE", + digest_verified=True, + ) + + def _create_custody( + self, *, secret_store: DpapiIndexAuthStore + ) -> None: + index_auth = self._secret_generator() + if type(index_auth) is not bytearray or len(index_auth) != 32: + raise WindowsTpmProvisioningError( + "index auth generator must return exactly 32 mutable bytes" + ) + try: + secret_store.store_once(index_auth) + finally: + zeroize(index_auth) + + def _define_from_custody( + self, + *, + plan: WindowsTpmNvProvisioningPlan, + secret_store: DpapiIndexAuthStore, + ) -> None: + index_auth = secret_store.load() + try: + owner_auth = self._owner_auth_loader() + try: + self._transport.define_space( + owner_auth=owner_auth, + index_auth=index_auth, + plan=plan, + ) + finally: + zeroize(owner_auth) + finally: + zeroize(index_auth) + + def _primer_from_custody( + self, + *, + plan: WindowsTpmNvProvisioningPlan, + secret_store: DpapiIndexAuthStore, + ) -> None: + index_auth = secret_store.load() + try: + self._transport.primer_extend( + index_auth=index_auth, plan=plan + ) + finally: + zeroize(index_auth) + + def _verified_receipt( + self, *, plan: WindowsTpmNvProvisioningPlan + ) -> dict[str, Any]: + return { + **plan.public_document(), + "plan_fingerprint": plan.fingerprint, + "provisioning_state": STATE_PRIMED_VERIFIED, + "result": "PRIMED_PROFILE_READY", + } + + def resume_provisioning( + self, + *, + plan: WindowsTpmNvProvisioningPlan, + secret_store: DpapiIndexAuthStore, + authorization_token: str, + ) -> dict[str, Any]: + """Resume only forward from durable facts; never rollback or repeat a step.""" + if authorization_token != plan.hardware_write_authorization_token: + raise WindowsTpmProvisioningError( + "hardware provisioning authorization token mismatch" + ) + + for _ in range(4): + state = self.inspect_state(plan=plan, secret_store=secret_store) + if state.state == STATE_PRIMED_VERIFIED: + return self._verified_receipt(plan=plan) + if state.state == STATE_PRIMED_UNVERIFIED: + digest = self._active_digest( + plan=plan, secret_store=secret_store + ) + if digest != plan.primed_genesis_digest: + raise WindowsTpmProvisioningError( + "active NV digest differs from reviewed primed genesis" + ) + return self._verified_receipt(plan=plan) + if state.state == STATE_EMPTY: + # Custody-only is intentionally not resumable across + # invocations: after an external TPM Clear/deletion it is + # indistinguishable from a crash immediately after custody + # creation. Create custody and attempt Define only within the + # same explicitly authorized invocation. + self._create_custody(secret_store=secret_store) + self._define_from_custody( + plan=plan, secret_store=secret_store + ) + continue + if state.state == STATE_CUSTODY_ONLY: + raise WindowsTpmProvisioningError( + "custody-only state requires fresh explicit recovery authority" + ) + if state.state == STATE_DEFINED_UNPRIMED: + self._primer_from_custody( + plan=plan, secret_store=secret_store + ) + continue + raise WindowsTpmProvisioningError( + "provisioning state is not resumable" + ) + raise WindowsTpmProvisioningError( + "provisioning did not converge within bounded transitions" + ) + + def provision_once( + self, + *, + plan: WindowsTpmNvProvisioningPlan, + secret_store: DpapiIndexAuthStore, + authorization_token: str, + ) -> dict[str, Any]: + """Compatibility alias for the crash-resumable forward-only ceremony.""" + return self.resume_provisioning( + plan=plan, + secret_store=secret_store, + authorization_token=authorization_token, + ) diff --git a/continuityos/gate/windows_tpm_runtime.py b/continuityos/gate/windows_tpm_runtime.py new file mode 100644 index 00000000..9dfd8555 --- /dev/null +++ b/continuityos/gate/windows_tpm_runtime.py @@ -0,0 +1,290 @@ +"""Narrow Windows TBS runtime backend for one provisioned R15B NV_EXTEND index. + +The runtime surface contains only authorized NV_Read and NV_Extend plus public +NV_ReadPublic verification. It has no define, undefine, clear, reset, owner +authorization retrieval, provisioning, or generic TPM command API. +""" +from __future__ import annotations + +import ctypes +import struct +from typing import Any, Callable + +from .tpm2_provider_binding import ( + BoundTpm2NvExtendProfile, + Tpm2ProviderBindingError, +) +from .windows_tpm_dpapi import DpapiIndexAuthStore, zeroize +from .windows_tpm_readonly import ( + TBS_COMMAND_LOCALITY_ZERO, + TBS_COMMAND_PRIORITY_LOW, + TBS_CONTEXT_FLAGS, + TBS_SUCCESS, + TPM_VERSION_20, + _MAX_TPM_RESPONSE, + _TbsContextParams2, + _load_tbs, + read_nv_public, +) + +TPM_ST_SESSIONS = 0x8002 +TPM_RS_PW = 0x40000009 +TPM_CC_NV_EXTEND = 0x00000136 +TPM_CC_NV_READ = 0x0000014E + +_ALLOWED_RUNTIME_COMMANDS = frozenset({TPM_CC_NV_READ, TPM_CC_NV_EXTEND}) + + +class WindowsTpmRuntimeError(Tpm2ProviderBindingError): + """The narrow Windows runtime TPM boundary failed closed.""" + + +def _is_sha256(value: Any) -> bool: + return ( + type(value) is str + and len(value) == 64 + and all(ch in "0123456789abcdef" for ch in value) + ) + + +def _pw_authorization_area(auth_value: bytearray) -> bytes: + if type(auth_value) is not bytearray or len(auth_value) != 32: + raise WindowsTpmRuntimeError("runtime index authValue must be 32 bytes") + return b"".join(( + struct.pack(">I", TPM_RS_PW), + struct.pack(">H", 0), + b"\x00", + struct.pack(">H", len(auth_value)), + bytes(auth_value), + )) + + +def _session_command( + *, + command_code: int, + handles: tuple[int, ...], + auth_value: bytearray, + parameters: bytes, +) -> bytes: + if command_code not in _ALLOWED_RUNTIME_COMMANDS: + raise WindowsTpmRuntimeError("TPM command is outside runtime allowlist") + if not handles or any(type(item) is not int for item in handles): + raise WindowsTpmRuntimeError("runtime TPM handles are invalid") + if type(parameters) is not bytes: + raise WindowsTpmRuntimeError("runtime TPM parameters must be bytes") + handle_bytes = b"".join(struct.pack(">I", item) for item in handles) + auth_area = _pw_authorization_area(auth_value) + body = ( + handle_bytes + + struct.pack(">I", len(auth_area)) + + auth_area + + parameters + ) + return struct.pack( + ">HII", TPM_ST_SESSIONS, 10 + len(body), command_code + ) + body + + +def _require_pw_response(data: bytes) -> bytes: + if type(data) is not bytes or len(data) < 14: + raise WindowsTpmRuntimeError("TPM runtime response is truncated") + tag, declared_size, response_code = struct.unpack(">HII", data[:10]) + if declared_size != len(data): + raise WindowsTpmRuntimeError("TPM runtime response size mismatch") + if response_code != 0: + raise WindowsTpmRuntimeError( + f"TPM runtime command failed with response code 0x{response_code:08x}" + ) + if tag != TPM_ST_SESSIONS: + raise WindowsTpmRuntimeError("TPM runtime response session tag is invalid") + parameter_size = int.from_bytes(data[10:14], "big") + parameter_end = 14 + parameter_size + if parameter_end > len(data): + raise WindowsTpmRuntimeError("TPM runtime parameter area is truncated") + parameters = data[14:parameter_end] + auth = data[parameter_end:] + if len(auth) != 5: + raise WindowsTpmRuntimeError("TPM password response auth area is invalid") + nonce_size = int.from_bytes(auth[0:2], "big") + attrs = auth[2] + hmac_size = int.from_bytes(auth[3:5], "big") + # TPM 2.0 requires continueSession (bit 0) SET in the response for + # password authorization, even though the bit has no password-session + # lifetime semantics. No other session attribute is valid here. + if nonce_size != 0 or attrs != 0x01 or hmac_size != 0: + raise WindowsTpmRuntimeError( + "TPM password response auth area is invalid" + ) + return parameters + + +class _WindowsTbsNvTransport: + """Private transport with exactly two effectful TPM command codes.""" + + def _submit(self, command: bytes) -> bytes: + if type(command) is not bytes or len(command) < 10: + raise WindowsTpmRuntimeError("runtime TPM command is invalid") + _, declared_size, command_code = struct.unpack(">HII", command[:10]) + if declared_size != len(command) or command_code not in _ALLOWED_RUNTIME_COMMANDS: + raise WindowsTpmRuntimeError("runtime TPM command is outside allowlist") + dll = _load_tbs() + params = _TbsContextParams2(TPM_VERSION_20, TBS_CONTEXT_FLAGS) + context = ctypes.c_void_p() + result = dll.Tbsi_Context_Create( + ctypes.byref(params), ctypes.byref(context) + ) + if result != TBS_SUCCESS or not context.value: + raise WindowsTpmRuntimeError( + f"TBS runtime context creation failed with 0x{result:08x}" + ) + try: + in_buffer = (ctypes.c_ubyte * len(command)).from_buffer_copy(command) + out_buffer = (ctypes.c_ubyte * _MAX_TPM_RESPONSE)() + out_size = ctypes.c_uint32(len(out_buffer)) + result = dll.Tbsip_Submit_Command( + context, + TBS_COMMAND_LOCALITY_ZERO, + TBS_COMMAND_PRIORITY_LOW, + in_buffer, + len(command), + out_buffer, + ctypes.byref(out_size), + ) + if result != TBS_SUCCESS: + raise WindowsTpmRuntimeError( + f"TBS runtime submit failed with 0x{result:08x}" + ) + if out_size.value > _MAX_TPM_RESPONSE: + raise WindowsTpmRuntimeError("runtime TPM response exceeded buffer") + return _require_pw_response(bytes(out_buffer[: out_size.value])) + finally: + close_result = dll.Tbsip_Context_Close(context) + if close_result != TBS_SUCCESS: + raise WindowsTpmRuntimeError( + f"TBS runtime close failed with 0x{close_result:08x}" + ) + + def nv_read(self, *, nv_index: int, auth_value: bytearray) -> bytes: + command = _session_command( + command_code=TPM_CC_NV_READ, + handles=(nv_index, nv_index), + auth_value=auth_value, + parameters=struct.pack(">HH", 32, 0), + ) + parameters = self._submit(command) + if len(parameters) != 34: + raise WindowsTpmRuntimeError("NV_Read response size is invalid") + size = int.from_bytes(parameters[:2], "big") + digest = parameters[2:] + if size != 32 or len(digest) != 32: + raise WindowsTpmRuntimeError("NV_Read digest size is invalid") + return digest + + def nv_extend( + self, + *, + nv_index: int, + auth_value: bytearray, + commitment: bytes, + ) -> None: + if type(commitment) is not bytes or len(commitment) != 32: + raise WindowsTpmRuntimeError("NV_Extend commitment must be 32 bytes") + command = _session_command( + command_code=TPM_CC_NV_EXTEND, + handles=(nv_index, nv_index), + auth_value=auth_value, + parameters=struct.pack(">H", len(commitment)) + commitment, + ) + parameters = self._submit(command) + if parameters != b"": + raise WindowsTpmRuntimeError("NV_Extend returned unexpected parameters") + + +class WindowsTbsNvExtendBackend: + """Bound runtime backend using DPAPI index auth and a stable WRITTEN profile.""" + + def __init__( + self, + secret_store: DpapiIndexAuthStore, + *, + transport: Any | None = None, + public_reader: Callable[[int], dict[str, Any]] | None = None, + ) -> None: + if secret_store is None: + raise WindowsTpmRuntimeError("runtime DPAPI secret store is required") + self._secret_store = secret_store + self._transport = transport or _WindowsTbsNvTransport() + self._public_reader = public_reader or read_nv_public + + @staticmethod + def _commitment_bytes(value: str) -> bytes: + if not _is_sha256(value) or value == "0" * 64: + raise WindowsTpmRuntimeError("runtime commitment digest is invalid") + return bytes.fromhex(value) + + def _public_snapshot( + self, *, profile: BoundTpm2NvExtendProfile + ) -> dict[str, Any]: + public = self._public_reader(profile.constraints.nv_index) + required = { + "nv_index", "name_alg", "tpma_nv_mask", "nv_type_bits", "nv_type", + "data_size", "orderly", "auth_policy_hex", "auth_policy_sha256", + "tpms_nv_public_marshaled", "raw_tpm_name", "nv_public_sha256", + "nv_name_sha256", "name_verified", + } + if type(public) is not dict or set(public) != required: + raise WindowsTpmRuntimeError("runtime NV public snapshot schema is invalid") + if public["name_verified"] is not True: + raise WindowsTpmRuntimeError("runtime NV Name is not verified") + return public + + def read_snapshot( + self, *, profile: BoundTpm2NvExtendProfile + ) -> dict[str, Any]: + public = self._public_snapshot(profile=profile) + secret = self._secret_store.load() + try: + digest = self._transport.nv_read( + nv_index=profile.constraints.nv_index, + auth_value=secret, + ) + finally: + zeroize(secret) + if type(digest) is not bytes or len(digest) != 32: + raise WindowsTpmRuntimeError("runtime NV digest is invalid") + return { + "nv_index": public["nv_index"], + "tpma_nv_mask": public["tpma_nv_mask"], + "auth_policy_sha256": public["auth_policy_sha256"], + "backend_kind": profile.constraints.backend_kind, + "transport_identity": profile.constraints.transport_identity, + "nv_type": public["nv_type"], + "name_alg": "SHA256" if public["name_alg"] == 0x000B else "OTHER", + "data_size": public["data_size"], + "orderly": public["orderly"], + "tpms_nv_public_marshaled": public["tpms_nv_public_marshaled"], + "raw_tpm_name": public["raw_tpm_name"], + "observed_nv_extend_digest": digest.hex(), + } + + def extend_once( + self, + *, + profile: BoundTpm2NvExtendProfile, + expected_previous_digest: str, + commitment_sha256: str, + ) -> dict[str, Any]: + before = self.read_snapshot(profile=profile) + if before["observed_nv_extend_digest"] != expected_previous_digest: + raise WindowsTpmRuntimeError("runtime TPM frontier moved before extend") + commitment = self._commitment_bytes(commitment_sha256) + secret = self._secret_store.load() + try: + self._transport.nv_extend( + nv_index=profile.constraints.nv_index, + auth_value=secret, + commitment=commitment, + ) + finally: + zeroize(secret) + return self.read_snapshot(profile=profile) diff --git a/docs/r15b_windows_tbs_backend_dpapi_primer.md b/docs/r15b_windows_tbs_backend_dpapi_primer.md new file mode 100644 index 00000000..5b0c7f19 --- /dev/null +++ b/docs/r15b_windows_tbs_backend_dpapi_primer.md @@ -0,0 +1,162 @@ +# R15B Windows TBS Runtime Backend, DPAPI Custody, and Primer + +## Scope + +This patch prepares a real Windows TPM 2.0 runtime backend without performing +any TPM mutation. It separates three responsibilities: + +- windows_tpm_runtime.py: normal runtime read/extend only; +- windows_tpm_dpapi.py: DPAPI custody for the 32-byte NV index authValue; +- windows_tpm_provisioning.py: offline-only define plus one primer extend. + +The provisioner is not imported by MCP, broker, CLI, or runtime entrypoints. + +## Reviewed host/profile + +EK public-key SHA-256: +d4493341ea776e196234af9547d2a22118767eea3a0312d00e445fa497f6469c + +Reviewed candidate handle: 0x01500020. +Owner-created definition mask: 0x02040044. + +Flags are AUTHWRITE, TPM_NT_EXTEND, AUTHREAD, and NO_DA. +OWNERREAD, OWNERWRITE, PLATFORMCREATE, and ORDERLY are absent. + +Windows TBS storage ownerAuth is used only by the offline define operation. +Runtime never retrieves ownerAuth and authenticates only with index authValue. + +## Why a primer is mandatory + +TPMA_NV_WRITTEN is clear immediately after NV_DefineSpace and becomes set after +the first successful write/extend. TPMS_NV_PUBLIC includes that attributes +field, so the transition changes both the public hash and TPM Name. + +R15B pins exact public bytes and Name. Normal governance therefore activates +only after one domain-separated provisioning primer. Stable active mask: +0x22040044. + +Reviewed hashes: + +- definition public: 27b26b0e2f7a5f11d63615367817fbbd52410c778e78bfd0c66adc88f9d6d103 +- active public: 4e6f663d09b9af433074e55b967ba1f201c5e34c8db8eee7a7ae372855642063 +- active Name: 5f350a71c3e0946639146d182d5f345d1e7203824a77fcd29cbf654fcc110fc5 +- primer: db5a5fa1867e616716b0c3d34e94a87a58c53875f9d5e11b9782756545074ec0 +- primed genesis: 54d69bcdee19684366afcc77cf894482b81fc0864e9b43d769adc020632147e4 +- active binding: d732c991382a2edb99b9dcb39063224c2a62df2c56813668e76a74444c2965d4 + +## Secret custody + +The index authValue is exactly 32 random bytes. The offline provisioner +generates it using the Windows system CSPRNG and stores it through current-user +DPAPI before attempting NV_DefineSpace. + +The persisted envelope contains ciphertext plus public binding metadata only. +DPAPI optional entropy binds ciphertext to the exact NV handle and EK hash. +Runtime loads plaintext into mutable bytearray memory and zeroizes it after each +authorization operation. + +Storage ownerAuth is obtained from TBS in memory only, is never persisted or +returned in evidence, and is zeroized after NV_DefineSpace. + +## Runtime command boundary + +The normal runtime private TBS transport accepts only TPM2_NV_Read and +TPM2_NV_Extend. Public identity remains verified through the existing +TPM2_NV_ReadPublic boundary. + +Runtime has no define, undefine, clear, reset, ownerAuth retrieval, hierarchy +management, provisioning, or generic command-submit API. + +## Offline ceremony + +A later separately authorized hardware ceremony must: + +1. require PPI=0, RestartPending=False, TPM ready, and lockout count 0; +2. re-enumerate NV handles and require 0x01500020 absent; +3. require the reviewed EK public-key hash; +4. persist a fresh DPAPI-protected index authValue; +5. retrieve storage ownerAuth in memory only; +6. execute exactly one TPM2_NV_DefineSpace; +7. re-read and verify the unprimed public identity; +8. execute exactly one primer TPM2_NV_Extend; +9. re-read the stable WRITTEN=1 public identity; +10. authorized-read the digest and require reviewed primed genesis; +11. emit only public evidence and stop. + +The effectful method additionally requires its generated exact hardware-write +authorization token. This patch does not call that method. + +## Failure model + +If DPAPI custody creation fails, no TPM mutation is attempted. If a mutation +occurs and later verification fails, execution stops. It does not auto-undefine, +clear, recreate, or re-enroll the index. + +External TPM clear or deletion makes the bound profile unavailable and must fail +closed. Recovery requires a new explicit enrollment/provisioning authority. + +## Current status + +This branch is preparation only. No real NV_DefineSpace, NV_Read, NV_Extend, +NV_UndefineSpace, TPM clear/reset, reboot, bind_genesis, deploy, signing, +trading, or capital action is authorized by this patch gate. + +Reviewed plan fingerprint: +5ecae1fc79cf17f30862aa6f4678439e76d4e1c12a85ad47f3d7347f8b83aad1 + +Generated future plan-bound hardware-write confirmation token: +APPROVE_CONTINUITYOS_R15B_HARDWARE_PROVISION_5ECAE1FC79CF17F3 + +This deterministic token is not a secret and is not an authentication +credential. It binds a call to the reviewed plan. Human execution authority is +external and must be granted separately for the exact operation/state. + + +## Crash-resumable forward-only state machine + +Provisioning recovery is derived from durable DPAPI custody plus current TPM +public state. It does not trust a local progress marker. + +Observed states are: + +- EMPTY: no custody and reviewed handle absent; +- CUSTODY_ONLY: custody exists and handle absent; this is a hard HOLD across + invocations because it is indistinguishable from external TPM Clear/deletion; +- DEFINED_UNPRIMED: custody exists and exact definition public area is present; +- PRIMED_UNVERIFIED: exact active WRITTEN public area is present but authorized + digest verification did not complete; +- PRIMED_VERIFIED: exact active public area and reviewed primed genesis digest + are both verified. + +The initial explicitly authorized invocation may advance: +EMPTY -> create custody -> Define -> DEFINED_UNPRIMED -> PRIMED_UNVERIFIED/ +PRIMED_VERIFIED. + +CUSTODY_ONLY is deliberately not an automatic resume edge. If execution stops +after custody creation but before a verifiable Define, a later invocation must +HOLD for fresh explicit recovery/enrollment authority. This is the same +fail-closed treatment used when an external TPM Clear/deletion leaves stale +custody behind. + +A lost response after NV_DefineSpace is recovered by observing the exact +definition public area and continuing with primer; NV_DefineSpace is not +repeated. A lost response after primer is recovered by observing the exact +WRITTEN public area and performing verification only; primer is not repeated. + +A custody-only state, a handle without custody, an unreviewed public area, or an +active digest that differs from the reviewed primed genesis is a hard HOLD. +Recovery never invokes NV_UndefineSpace, TPM Clear, deletion, rollback, or +automatic re-enrollment. + + +## Password authorization response parsing + +TPM password authorization responses are expected to contain an empty nonce, +sessionAttributes equal to 0x01 (continueSession), and an empty HMAC. The TPM +2.0 specification requires continueSession to be SET in a response associated +with password authorization even though it has no password-session lifetime +semantics. + +The parser therefore accepts exactly 0x01 for the response attributes and +rejects 0x00 or any additional session-attribute bits. This matches the +observed hardware response auth area 0000010000. diff --git a/tests/test_r15b_windows_tpm_runtime_backend.py b/tests/test_r15b_windows_tpm_runtime_backend.py new file mode 100644 index 00000000..8e9b2e7a --- /dev/null +++ b/tests/test_r15b_windows_tpm_runtime_backend.py @@ -0,0 +1,659 @@ +"""Hardware-free tests for the R15B Windows TBS backend and provisioning ceremony.""" +from __future__ import annotations + +import base64 +import hashlib +import inspect +import json + +import pytest + +from continuityos.gate import windows_tpm_runtime as runtime +from continuityos.gate.windows_tpm_dpapi import ( + DpapiIndexAuthStore, + WindowsDpapiCustodyError, +) +from continuityos.gate.windows_tpm_provisioning import ( + OfflineWindowsTpmNvProvisioner, + WindowsTpmProvisioningError, + build_reviewed_plan, +) + + +EK_SHA256 = "d4493341ea776e196234af9547d2a22118767eea3a0312d00e445fa497f6469c" + + +def _password_response(parameters: bytes, *, attrs: int = 0x01) -> bytes: + auth = b"\x00\x00" + bytes([attrs]) + b"\x00\x00" + size = 14 + len(parameters) + len(auth) + return ( + b"\x80\x02" + + size.to_bytes(4, "big") + + b"\x00\x00\x00\x00" + + len(parameters).to_bytes(4, "big") + + parameters + + auth + ) + + +def test_password_response_accepts_continue_session_from_real_tpm_shape() -> None: + parameters = b"\x00\x20" + bytes.fromhex("54" * 32) + response = _password_response(parameters, attrs=0x01) + assert runtime._require_pw_response(response) == parameters + + +@pytest.mark.parametrize("attrs", [0x00, 0x02, 0x20, 0x41, 0xFF]) +def test_password_response_rejects_noncanonical_session_attributes(attrs: int) -> None: + response = _password_response(b"", attrs=attrs) + with pytest.raises(runtime.WindowsTpmRuntimeError, match="auth area is invalid"): + runtime._require_pw_response(response) + + +def test_password_response_rejects_nonce_or_hmac_material() -> None: + nonce_auth = b"\x00\x01X\x01\x00\x00" + nonce_size = 14 + len(nonce_auth) + nonce_response = ( + b"\x80\x02" + nonce_size.to_bytes(4, "big") + + b"\x00\x00\x00\x00" + b"\x00\x00\x00\x00" + nonce_auth + ) + with pytest.raises(runtime.WindowsTpmRuntimeError, match="auth area is invalid"): + runtime._require_pw_response(nonce_response) + + hmac_auth = b"\x00\x00\x01\x00\x01X" + hmac_size = 14 + len(hmac_auth) + hmac_response = ( + b"\x80\x02" + hmac_size.to_bytes(4, "big") + + b"\x00\x00\x00\x00" + b"\x00\x00\x00\x00" + hmac_auth + ) + with pytest.raises(runtime.WindowsTpmRuntimeError, match="auth area is invalid"): + runtime._require_pw_response(hmac_response) + + +class FakeProtector: + def protect(self, plaintext: bytearray, *, entropy: bytes) -> bytes: + assert len(entropy) == 32 + return bytes(item ^ entropy[index % len(entropy)] for index, item in enumerate(plaintext)) + + def unprotect(self, ciphertext: bytes, *, entropy: bytes) -> bytearray: + assert len(entropy) == 32 + return bytearray( + item ^ entropy[index % len(entropy)] + for index, item in enumerate(ciphertext) + ) + + +def test_dpapi_store_persists_ciphertext_only_and_binds_context(tmp_path) -> None: + path = tmp_path / "r15b.dpapi.json" + secret = bytearray(range(32)) + store = DpapiIndexAuthStore( + path, + nv_index=0x01500020, + ek_public_sha256=EK_SHA256, + protector=FakeProtector(), + ) + store.store_once(secret) + text = path.read_text(encoding="ascii") + assert secret.hex() not in text + assert base64.b64encode(bytes(secret)).decode("ascii") not in text + loaded = store.load() + assert loaded == secret + + other = DpapiIndexAuthStore( + path, + nv_index=0x01500021, + ek_public_sha256=EK_SHA256, + protector=FakeProtector(), + ) + with pytest.raises(WindowsDpapiCustodyError, match="binding differs"): + other.load() + with pytest.raises(WindowsDpapiCustodyError, match="already exists"): + store.store_once(secret) + + +def public_record(plan, *, active: bool) -> dict: + public = plan.active_public if active else plan.definition_public + attrs = plan.active_mask if active else plan.definition_mask + raw_name = b"\x00\x0b" + hashlib.sha256(public).digest() + return { + "nv_index": plan.nv_index, + "name_alg": 0x000B, + "tpma_nv_mask": attrs, + "nv_type_bits": 0x40, + "nv_type": "TPM_NT_EXTEND", + "data_size": 32, + "orderly": False, + "auth_policy_hex": "", + "auth_policy_sha256": None, + "tpms_nv_public_marshaled": public, + "raw_tpm_name": raw_name, + "nv_public_sha256": hashlib.sha256(public).hexdigest(), + "nv_name_sha256": hashlib.sha256(raw_name).hexdigest(), + "name_verified": True, + } + + +class FakeSecretStore: + def __init__(self, value: bytes = b"s" * 32): + self.value = bytes(value) + self.loads: list[bytearray] = [] + + def load(self) -> bytearray: + result = bytearray(self.value) + self.loads.append(result) + return result + + +class FakeRuntimeTransport: + def __init__(self, digest: str): + self.digest = digest + self.read_calls = 0 + self.extend_calls = 0 + + def nv_read(self, *, nv_index: int, auth_value: bytearray) -> bytes: + assert nv_index == 0x01500020 + assert bytes(auth_value) == b"s" * 32 + self.read_calls += 1 + return bytes.fromhex(self.digest) + + def nv_extend( + self, *, nv_index: int, auth_value: bytearray, commitment: bytes + ) -> None: + assert nv_index == 0x01500020 + assert bytes(auth_value) == b"s" * 32 + self.extend_calls += 1 + self.digest = hashlib.sha256( + bytes.fromhex(self.digest) + commitment + ).hexdigest() + + +def test_runtime_backend_reads_extends_and_zeroizes_secret_buffers() -> None: + plan = build_reviewed_plan(ek_public_sha256=EK_SHA256) + store = FakeSecretStore() + transport = FakeRuntimeTransport(plan.primed_genesis_digest) + backend = runtime.WindowsTbsNvExtendBackend( + store, + transport=transport, + public_reader=lambda handle: public_record(plan, active=True), + ) + profile = plan.active_profile + before = backend.read_snapshot(profile=profile) + assert before["observed_nv_extend_digest"] == plan.primed_genesis_digest + assert all(set(secret) == {0} for secret in store.loads) + + commitment = "ab" * 32 + after = backend.extend_once( + profile=profile, + expected_previous_digest=plan.primed_genesis_digest, + commitment_sha256=commitment, + ) + expected = hashlib.sha256( + bytes.fromhex(plan.primed_genesis_digest) + + bytes.fromhex(commitment) + ).hexdigest() + assert after["observed_nv_extend_digest"] == expected + assert after["tpma_nv_mask"] == plan.active_mask + assert transport.extend_calls == 1 + assert all(set(secret) == {0} for secret in store.loads) + + +def test_runtime_module_exposes_no_provisioning_or_owner_auth_surface() -> None: + public_names = {name for name in dir(runtime) if not name.startswith("_")} + forbidden = { + "OfflineWindowsTpmNvProvisioner", + "TPM_CC_NV_DEFINE_SPACE", + "TPM_CC_CLEAR", + "TPM_CC_NV_UNDEFINE_SPACE", + "load_storage_owner_auth", + } + assert public_names.isdisjoint(forbidden) + source = inspect.getsource(runtime) + assert "windows_tpm_provisioning" not in source + assert "Tbsi_Get_OwnerAuth" not in source + + +class FakeProvisioningSecretStore: + def __init__(self): + self.stored: bytes | None = None + self.loads: list[bytearray] = [] + + def exists(self) -> bool: + return self.stored is not None + + def store_once(self, secret: bytearray) -> None: + assert self.stored is None + self.stored = bytes(secret) + + def load(self) -> bytearray: + assert self.stored is not None + value = bytearray(self.stored) + self.loads.append(value) + return value + + +class FakeProvisioningTransport: + def __init__(self, store: FakeProvisioningSecretStore): + self.store = store + self.defined = False + self.primed = False + self.calls: list[str] = [] + + def define_space(self, *, owner_auth, index_auth, plan) -> None: + assert self.store.stored == bytes(index_auth) + assert bytes(owner_auth) == b"o" * 20 + assert len(index_auth) == 32 + self.calls.append("define") + self.defined = True + + def primer_extend(self, *, index_auth, plan) -> None: + assert self.defined + assert bytes(index_auth) == self.store.stored + assert len(plan.primer_commitment) == 32 + self.calls.append("primer") + self.primed = True + + +class FakePostPrimerBackend: + def __init__(self, plan, store): + self.plan = plan + self.store = store + + def read_snapshot(self, *, profile): + assert profile.binding_sha256() == self.plan.active_profile.binding_sha256() + return { + "observed_nv_extend_digest": self.plan.primed_genesis_digest + } + + +def test_offline_provisioner_requires_exact_token_and_orders_custody_before_define() -> None: + plan = build_reviewed_plan(ek_public_sha256=EK_SHA256) + store = FakeProvisioningSecretStore() + transport = FakeProvisioningTransport(store) + + def public_reader(_handle): + if transport.primed: + return public_record(plan, active=True) + if transport.defined: + return public_record(plan, active=False) + raise AssertionError("public read before definition") + + provisioner = OfflineWindowsTpmNvProvisioner( + transport=transport, + handle_reader=lambda: [plan.nv_index] if transport.defined else [], + public_reader=public_reader, + owner_auth_loader=lambda: bytearray(b"o" * 20), + runtime_backend_factory=lambda secret_store: FakePostPrimerBackend( + plan, secret_store + ), + secret_generator=lambda: bytearray(b"s" * 32), + ) + with pytest.raises( + WindowsTpmProvisioningError, match="authorization token mismatch" + ): + provisioner.provision_once( + plan=plan, secret_store=store, authorization_token="NO" + ) + assert transport.calls == [] + assert store.stored is None + + result = provisioner.provision_once( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert transport.calls == ["define", "primer"] + assert result["result"] == "PRIMED_PROFILE_READY" + assert result["active_profile_binding_sha256"] == ( + plan.active_profile.binding_sha256() + ) + + +def test_plan_pins_reviewed_owner_created_profile_and_stable_written_identity() -> None: + plan = build_reviewed_plan(ek_public_sha256=EK_SHA256) + assert plan.nv_index == 0x01500020 + assert plan.definition_mask == 0x02040044 + assert plan.active_mask == 0x22040044 + assert hashlib.sha256(plan.definition_public).hexdigest() == ( + "27b26b0e2f7a5f11d63615367817fbbd52410c778e78bfd0c66adc88f9d6d103" + ) + assert plan.active_profile.nv_public_sha256 == ( + "4e6f663d09b9af433074e55b967ba1f201c5e34c8db8eee7a7ae372855642063" + ) + assert plan.active_profile.nv_name_sha256 == ( + "5f350a71c3e0946639146d182d5f345d1e7203824a77fcd29cbf654fcc110fc5" + ) + assert plan.primer_commitment.hex() == ( + "db5a5fa1867e616716b0c3d34e94a87a58c53875f9d5e11b9782756545074ec0" + ) + assert plan.primed_genesis_digest == ( + "54d69bcdee19684366afcc77cf894482b81fc0864e9b43d769adc020632147e4" + ) + assert plan.active_profile.binding_sha256() == ( + "d732c991382a2edb99b9dcb39063224c2a62df2c56813668e76a74444c2965d4" + ) + + +def test_plan_public_document_contains_no_secret_material() -> None: + plan = build_reviewed_plan(ek_public_sha256=EK_SHA256) + document = plan.public_document() + encoded = json.dumps(document, sort_keys=True) + assert "authValue" not in encoded + assert "ownerAuth" not in encoded + assert "ciphertext" not in encoded + assert plan.hardware_write_authorization_token.startswith( + "APPROVE_CONTINUITYOS_R15B_HARDWARE_PROVISION_" + ) + + +class CrashAwareTransport(FakeProvisioningTransport): + def __init__(self, store, *, fail_after_define=False, fail_after_primer=False): + super().__init__(store) + self.fail_after_define = fail_after_define + self.fail_after_primer = fail_after_primer + self.define_attempts = 0 + self.primer_attempts = 0 + + def define_space(self, *, owner_auth, index_auth, plan) -> None: + self.define_attempts += 1 + super().define_space( + owner_auth=owner_auth, index_auth=index_auth, plan=plan + ) + if self.fail_after_define: + self.fail_after_define = False + raise RuntimeError("lost response after define") + + def primer_extend(self, *, index_auth, plan) -> None: + self.primer_attempts += 1 + super().primer_extend(index_auth=index_auth, plan=plan) + if self.fail_after_primer: + self.fail_after_primer = False + raise RuntimeError("lost response after primer") + + +def recovery_fixture(*, state: str, fail_after_define=False, fail_after_primer=False): + from continuityos.gate.windows_tpm_provisioning import ( + STATE_CUSTODY_ONLY, + STATE_DEFINED_UNPRIMED, + STATE_EMPTY, + STATE_PRIMED_VERIFIED, + ) + + plan = build_reviewed_plan(ek_public_sha256=EK_SHA256) + store = FakeProvisioningSecretStore() + transport = CrashAwareTransport( + store, + fail_after_define=fail_after_define, + fail_after_primer=fail_after_primer, + ) + + if state != STATE_EMPTY: + store.stored = b"s" * 32 + if state in {STATE_DEFINED_UNPRIMED, STATE_PRIMED_VERIFIED}: + transport.defined = True + if state == STATE_PRIMED_VERIFIED: + transport.primed = True + + def handles(): + return [plan.nv_index] if transport.defined else [] + + def public_reader(_handle): + assert transport.defined + return public_record(plan, active=transport.primed) + + provisioner = OfflineWindowsTpmNvProvisioner( + transport=transport, + handle_reader=handles, + public_reader=public_reader, + owner_auth_loader=lambda: bytearray(b"o" * 20), + runtime_backend_factory=lambda secret_store: FakePostPrimerBackend( + plan, secret_store + ), + secret_generator=lambda: bytearray(b"s" * 32), + ) + return plan, store, transport, provisioner + + +def test_preexisting_custody_only_is_hard_hold_without_tpm_mutation() -> None: + from continuityos.gate.windows_tpm_provisioning import STATE_CUSTODY_ONLY + + plan, store, transport, provisioner = recovery_fixture( + state=STATE_CUSTODY_ONLY + ) + original = store.stored + with pytest.raises( + WindowsTpmProvisioningError, + match="custody-only state requires fresh explicit recovery authority", + ): + provisioner.resume_provisioning( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert store.stored == original + assert transport.define_attempts == 0 + assert transport.primer_attempts == 0 + + +def test_external_clear_or_deletion_with_stale_custody_is_hard_hold() -> None: + from continuityos.gate.windows_tpm_provisioning import STATE_PRIMED_VERIFIED + + plan, store, transport, provisioner = recovery_fixture( + state=STATE_PRIMED_VERIFIED + ) + # Simulate an external Clear/deletion after a previously verified state: + # durable DPAPI custody remains but the reviewed NV handle is gone. + transport.defined = False + transport.primed = False + + with pytest.raises( + WindowsTpmProvisioningError, + match="custody-only state requires fresh explicit recovery authority", + ): + provisioner.resume_provisioning( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert transport.define_attempts == 0 + assert transport.primer_attempts == 0 + + +def test_resume_from_defined_unprimed_never_redefines() -> None: + from continuityos.gate.windows_tpm_provisioning import STATE_DEFINED_UNPRIMED + + plan, store, transport, provisioner = recovery_fixture( + state=STATE_DEFINED_UNPRIMED + ) + result = provisioner.resume_provisioning( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert result["provisioning_state"] == "PRIMED_VERIFIED" + assert transport.define_attempts == 0 + assert transport.primer_attempts == 1 + + +def test_resume_from_primed_verified_performs_no_tpm_mutation() -> None: + from continuityos.gate.windows_tpm_provisioning import STATE_PRIMED_VERIFIED + + plan, store, transport, provisioner = recovery_fixture( + state=STATE_PRIMED_VERIFIED + ) + result = provisioner.resume_provisioning( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert result["result"] == "PRIMED_PROFILE_READY" + assert transport.define_attempts == 0 + assert transport.primer_attempts == 0 + + +def test_lost_define_response_resumes_without_second_define() -> None: + from continuityos.gate.windows_tpm_provisioning import STATE_EMPTY + + plan, store, transport, provisioner = recovery_fixture( + state=STATE_EMPTY, fail_after_define=True + ) + with pytest.raises(RuntimeError, match="lost response after define"): + provisioner.resume_provisioning( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert store.exists() + assert transport.defined + assert transport.define_attempts == 1 + assert transport.primer_attempts == 0 + + result = provisioner.resume_provisioning( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert result["provisioning_state"] == "PRIMED_VERIFIED" + assert transport.define_attempts == 1 + assert transport.primer_attempts == 1 + + +def test_lost_primer_response_resumes_without_second_primer() -> None: + from continuityos.gate.windows_tpm_provisioning import STATE_EMPTY + + plan, store, transport, provisioner = recovery_fixture( + state=STATE_EMPTY, fail_after_primer=True + ) + with pytest.raises(RuntimeError, match="lost response after primer"): + provisioner.resume_provisioning( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert transport.defined + assert transport.primed + assert transport.define_attempts == 1 + assert transport.primer_attempts == 1 + + result = provisioner.resume_provisioning( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert result["provisioning_state"] == "PRIMED_VERIFIED" + assert transport.define_attempts == 1 + assert transport.primer_attempts == 1 + + +def test_existing_handle_without_custody_is_hard_hold() -> None: + plan = build_reviewed_plan(ek_public_sha256=EK_SHA256) + store = FakeProvisioningSecretStore() + transport = FakeProvisioningTransport(store) + transport.defined = True + provisioner = OfflineWindowsTpmNvProvisioner( + transport=transport, + handle_reader=lambda: [plan.nv_index], + public_reader=lambda handle: public_record(plan, active=False), + owner_auth_loader=lambda: bytearray(b"o" * 20), + runtime_backend_factory=lambda secret_store: FakePostPrimerBackend( + plan, secret_store + ), + secret_generator=lambda: bytearray(b"s" * 32), + ) + with pytest.raises( + WindowsTpmProvisioningError, match="exists without matching DPAPI custody" + ): + provisioner.resume_provisioning( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert transport.calls == [] + + +def test_unreviewed_existing_public_identity_is_hard_hold() -> None: + plan, store, transport, _ = recovery_fixture(state="DEFINED_UNPRIMED") + bad = public_record(plan, active=False) + bad["tpma_nv_mask"] ^= 0x1 + provisioner = OfflineWindowsTpmNvProvisioner( + transport=transport, + handle_reader=lambda: [plan.nv_index], + public_reader=lambda handle: bad, + owner_auth_loader=lambda: bytearray(b"o" * 20), + runtime_backend_factory=lambda secret_store: FakePostPrimerBackend( + plan, secret_store + ), + secret_generator=lambda: bytearray(b"s" * 32), + ) + with pytest.raises( + WindowsTpmProvisioningError, + match="outside reviewed provisioning states", + ): + provisioner.resume_provisioning( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert transport.calls == [] + + +def test_active_wrong_digest_is_hard_hold_without_repeat_primer() -> None: + from continuityos.gate.windows_tpm_provisioning import STATE_PRIMED_VERIFIED + + plan, store, transport, _ = recovery_fixture( + state=STATE_PRIMED_VERIFIED + ) + + class WrongDigestBackend: + def read_snapshot(self, *, profile): + return {"observed_nv_extend_digest": "11" * 32} + + provisioner = OfflineWindowsTpmNvProvisioner( + transport=transport, + handle_reader=lambda: [plan.nv_index], + public_reader=lambda handle: public_record(plan, active=True), + owner_auth_loader=lambda: bytearray(b"o" * 20), + runtime_backend_factory=lambda secret_store: WrongDigestBackend(), + secret_generator=lambda: bytearray(b"s" * 32), + ) + with pytest.raises( + WindowsTpmProvisioningError, match="differs from reviewed primed genesis" + ): + provisioner.resume_provisioning( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert transport.define_attempts == 0 + assert transport.primer_attempts == 0 + + +def test_primed_unverified_retries_verification_only() -> None: + from continuityos.gate.windows_tpm_provisioning import STATE_PRIMED_VERIFIED + + plan, store, transport, _ = recovery_fixture( + state=STATE_PRIMED_VERIFIED + ) + calls = {"count": 0} + class FlakyBackend: + def read_snapshot(self, *, profile): + calls["count"] += 1 + if calls["count"] == 1: + raise RuntimeError("transient read failure") + return {"observed_nv_extend_digest": plan.primed_genesis_digest} + + provisioner = OfflineWindowsTpmNvProvisioner( + transport=transport, + handle_reader=lambda: [plan.nv_index], + public_reader=lambda handle: public_record(plan, active=True), + owner_auth_loader=lambda: bytearray(b"o" * 20), + runtime_backend_factory=lambda secret_store: FlakyBackend(), + secret_generator=lambda: bytearray(b"s" * 32), + ) + result = provisioner.resume_provisioning( + plan=plan, + secret_store=store, + authorization_token=plan.hardware_write_authorization_token, + ) + assert result["provisioning_state"] == "PRIMED_VERIFIED" + assert calls["count"] == 2 + assert transport.define_attempts == 0 + assert transport.primer_attempts == 0