diff --git a/docs/CLAIM_EVIDENCE_MATRIX.md b/docs/CLAIM_EVIDENCE_MATRIX.md index d006a7b..89c52c2 100644 --- a/docs/CLAIM_EVIDENCE_MATRIX.md +++ b/docs/CLAIM_EVIDENCE_MATRIX.md @@ -1,6 +1,6 @@ # SelfConnect Claim Evidence Matrix -Last updated: 2026-07-18 +Last updated: 2026-07-20 This matrix ties current SelfConnect positioning to concrete proof artifacts on `test/win32-hardening-v1`. It is intentionally conservative: if a capability has @@ -29,10 +29,10 @@ not been live-tested or committed as a probe, it is marked as pending. | Pipe-authenticated role leases/generations | Proven as isolated control-plane proof | `sc_mesh_lease.py`, `experiments/win32_probe/pipe_role_lease_probe.py`, redacted PASS artifact | | Governed lease gate on guarded send/read path | Proven as optional in-process runtime gate | optional runtime governed enforcement on the guarded send/read path (role+birth_id+generation+hwnd+owner_sid_hash) layered over the explore-mode target guard; `sc_mesh_lease.evaluate_lease_gate`, `sc_mesh_lease.current_owner_sid`, `sc_cli.send_text_to_window`/`read_window`, `sc_mcp` tools, `tests/test_mesh_lease.py`, `tests/test_package_adapters.py`, `experiments/win32_probe/runtime_sid_probe.py`, `experiments/win32_probe/results/runtime_sid_probe_PASS_redacted.json`. Boundary: OPTIONAL, in-process, NOT a full daemon; explore mode unchanged (no-op); birth_id optional in gate (checked when provided, skipped when omitted for backward compat); runtime OS SID lookup now uses `OpenProcessToken` -> `GetTokenInformation(TokenUser)` -> `ConvertSidToStringSidW` and fails closed on `` | | Channel-router composition proof | Proven as redacted model proof plus live throwaway/local proof | `experiments/win32_probe/channel_router_composition_probe.py`, `experiments/win32_probe/results/channel_router_composition_PASS_redacted.json`, `experiments/win32_probe/results/channel_router_composition_LIVE_PASS_redacted.json`, `tests/test_channel_router_composition.py`, `docs/CHANNEL_ROUTER_COMPOSITION_PROOF.md`. Boundary: the recorded model selected terminal `WM_CHAR` for its CASCADIA fixture; current production auto routing is class-aware and uses `WriteConsoleInputW` for `ConsoleWindowClass`. The browser route remains UIA Value/Invoke | -| TPM/CNG key use | Proven in experiment/enterprise lane | `experiments/win32_probe/CAPABILITY_BACKLOG.md`; full attestation pending | -| TPM platform attestation | Pending | `NCryptCreateClaim` descriptor fix still required | +| TPM/CNG key use | Proven in package path | `sc_tpm_attestation.py`, `tests/test_tpm_attestation.py`; the installed identity key is machine-scoped, hardware-backed, PCP-identity marked, and non-exportable | +| TPM platform attestation | Proven locally with an explicit remote-trust boundary | `sc_tpm_attestation.py`, `tests/test_tpm_attestation.py`, `docs/TPM_PLATFORM_ATTESTATION.md`, and the 2026-07-20 redacted live/selftest artifacts prove nonce-bound `NCRYPT_CLAIM_PLATFORM`, PCR 0-23 digest verification, pinned-key signature verification, and durable replay rejection. Manufacturer/EK chain trust, fleet enrollment/revocation, and an external relying-party policy remain deployment gates | | ETW provider smoke | Proven as isolated probe | `experiments/win32_probe/etw_provider.py`, `CAPABILITY_BACKLOG.md` | -| Windows SCM service wrapper | Implemented and test-covered; live installation evidence pending | `sc_fabric_windows_svc.py`, `tests/test_fabric_windows_svc.py`; SelfConnectFabricV2 service name and install/remove/start/stop/query code exist. Boundary: this is not evidence that the service is installed, running, or production-ready on a release machine. | +| Windows SCM service wrapper | Proven live on the release workstation | `sc_fabric_windows_svc.py`, `tests/test_fabric_windows_svc.py`, `experiments/fabric_v2/results/fabric_v2_scm_live_20260720_redacted.json`; auto-start LocalSystem installation, real named-pipe ACK, graceful restart replay persistence, forced-process SCM recovery, and post-recovery traffic passed. Boundary: one-host proof is not fleet deployment evidence | | Job Object containment | Proven in experiment/enterprise lane | `CAPABILITY_BACKLOG.md`; runtime adapter pending | | MCP/package distribution | Proven as optional adapter | `sc_mcp.py`, `pyproject.toml`, package tests, built wheel inspection. Boundary: MCP is not required for the proven WM_CHAR terminal path, UIA browser path, echo-filtered readback, target guard, or lease-gate model | | Fabric V0 logical scale harness | Proven as benchmark/evidence harness | `sc_fabric_benchmark.py`, `docs/FABRIC_V2_BENCHMARK_RESULTS.md`; 5/10/15/20 logical agents passed with flat sub-ms transport/governance p99 and `0.0` model calls per known task | @@ -59,13 +59,12 @@ The current non-claims are: - first AI-to-AI communication through desktop automation; - unrestricted browser automation across public sites; - CAPTCHA bypass or anti-bot evasion; -- production TPM attestation; +- manufacturer/EK-chain-approved remote TPM attestation or fleet enrollment; - production named-pipe control-plane replacement for terminal-visible routing. -Note: the Windows SCM wrapper is implemented and test-covered -(`sc_fabric_windows_svc.py`, `sc_fabric_service.py`). Live installation, -startup/restart behavior under SCM, and full governed daemon coverage remain -release evidence gaps. +The Windows SCM wrapper is installed and live-tested on the release workstation. +Fleet rollout, signed installer policy, and independent deployment evidence are +separate gates. ## Next Highest-Value Evidence @@ -74,8 +73,10 @@ release evidence gaps. as an opt-in gate. Runtime OS owner SID lookup is proven via `OpenProcessToken` -> `GetTokenInformation(TokenUser)` -> `ConvertSidToStringSidW`. -2. Capture live installed-service evidence for the existing Fabric V2 SCM wrapper. -3. Finish TPM platform attestation with correct `NCryptBufferDesc`. +2. Done on the release workstation: capture live installed-service evidence for + the Fabric V2 SCM wrapper. +3. Done for the local mechanism: provision and verify a nonce-bound TPM platform + quote with replay rejection. Manufacturer/EK trust remains external policy. 4. Add browser multi-tab/stale-tab proof. 5. Add governed audit event for protected checkpoint pause. 6. Wrap job-object containment as an optional runtime adapter. diff --git a/docs/TPM_PLATFORM_ATTESTATION.md b/docs/TPM_PLATFORM_ATTESTATION.md new file mode 100644 index 0000000..3b6b3c4 --- /dev/null +++ b/docs/TPM_PLATFORM_ATTESTATION.md @@ -0,0 +1,56 @@ +# TPM Platform Attestation + +SelfConnect can provision a machine-scoped, non-exportable RSA identity key in +the Windows Microsoft Platform Crypto Provider and use it to issue a +nonce-bound `NCRYPT_CLAIM_PLATFORM` quote. + +## Provision + +Run from an elevated PowerShell prompt: + +```powershell +python -m pip install ".[trust]" +selfconnect-tpm-attestation provision +``` + +The default key is `SelfConnectPlatformAIK-v1`. Its private material remains in +the platform provider. The command reports only the public-key digest and key +properties. + +## Verify The Hardware Path + +```powershell +selfconnect-tpm-attestation selftest --output tpm-selftest-redacted.json +``` + +The selftest provisions an ephemeral identity key, issues a quote covering PCR +0 through PCR 23, verifies its RSA signature and PCR digest, rejects a wrong +nonce, rejects modified PCR evidence, and deletes the ephemeral key. + +Application verifiers should generate a fresh 16-64 byte nonce, pin the +provisioned public-key digest, call `verify_and_consume`, and keep the +`NonceReplayStore` on durable local storage. The replay store uses a +cross-process file lock and atomic replacement. + +## Proven Boundary + +The local verifier proves all of the following: + +- the quote is signed by the pinned, non-exportable PCP identity key; +- the TPM quote contains the verifier nonce; +- the selected PCR inventory recomputes to the quoted digest; +- the expected PCR 0-23 selection is bound to the quote; +- the verifier nonce has not already been consumed by the durable replay store. + +This is platform-attestation mechanism evidence on the tested Windows TPM 2.0 +host. It does not establish manufacturer or endorsement-key certificate-chain +trust, remote fleet enrollment, verifier policy distribution, certificate +revocation, independent assessment, or an authorization to operate. Those are +relying-party and deployment controls. + +## Evidence + +- `sc_tpm_attestation.py` +- `tests/test_tpm_attestation.py` +- `experiments/win32_probe/results/tpm_platform_attestation_live_20260720_redacted.json` +- `experiments/win32_probe/results/tpm_platform_attestation_selftest_20260720_redacted.json` diff --git a/experiments/fabric_v2/results/fabric_v2_scm_live_20260720_redacted.json b/experiments/fabric_v2/results/fabric_v2_scm_live_20260720_redacted.json new file mode 100644 index 0000000..286b69b --- /dev/null +++ b/experiments/fabric_v2/results/fabric_v2_scm_live_20260720_redacted.json @@ -0,0 +1,18 @@ +{ + "account": "LocalSystem", + "auto_start": true, + "failure_restart_delays_ms": [5000, 15000, 60000], + "failure_reset_period_s": 86400, + "forced_process_termination": true, + "installed": true, + "live_roundtrip": true, + "ok": true, + "post_crash_roundtrip": true, + "raw_payload_included": false, + "replay_rejected_after_graceful_restart": true, + "schema_version": 1, + "scm_restarted_with_new_pid": true, + "service_name": "SelfConnectFabricV2", + "source_parent_commit": "569f456a81b22905c0f96827471cb2cdcb65bf01", + "system_scope_dependencies": true +} diff --git a/experiments/win32_probe/results/tpm_platform_attestation_live_20260720_redacted.json b/experiments/win32_probe/results/tpm_platform_attestation_live_20260720_redacted.json new file mode 100644 index 0000000..ed75e8f --- /dev/null +++ b/experiments/win32_probe/results/tpm_platform_attestation_live_20260720_redacted.json @@ -0,0 +1,16 @@ +{ + "claim_sha256": "d45817c097ff944dfc1762359db127161962feaad23a857c4d9b245654022285", + "key_name": "SelfConnectPlatformAIK-v1", + "manufacturer_chain_verified": false, + "nonce_sha256": "ffc471d499db1085f953919a9d9d3724c5f9ae283805d239b6a86684929185c2", + "ok": true, + "pcr_algorithm": 11, + "pcr_mask": 16777215, + "pcr_values_sha256": "cfa6d2a34942a42a3e96ce4ed26037f138d6695ad8d127824a13ccba6d22540d", + "provider": "Microsoft Platform Crypto Provider", + "public_key_sha256": "e5954da7514dcc6574cd9d429d7cf36fa3fa0ca8a57da348fb6e99d48402ba96", + "raw_claim_included": false, + "replay_checked": true, + "replay_rejected": true, + "schema_version": 1 +} diff --git a/experiments/win32_probe/results/tpm_platform_attestation_selftest_20260720_redacted.json b/experiments/win32_probe/results/tpm_platform_attestation_selftest_20260720_redacted.json new file mode 100644 index 0000000..fb4b2dc --- /dev/null +++ b/experiments/win32_probe/results/tpm_platform_attestation_selftest_20260720_redacted.json @@ -0,0 +1,12 @@ +{ + "hardware_backed": true, + "manufacturer_chain_verified": false, + "nonce_mismatch_rejected": true, + "ok": true, + "private_key_exportable": false, + "provider": "Microsoft Platform Crypto Provider", + "quote_verified": true, + "raw_claim_included": false, + "schema_version": 1, + "tampered_pcr_rejected": true +} diff --git a/pyproject.toml b/pyproject.toml index 4613037..c8950fc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -89,6 +89,7 @@ artifacts = [ "experiments/fabric_v2/results/baseline_5agent_fabric_v2_service_transport.json", "experiments/fabric_v2/results/SC_FABRIC_SERVICE_20260621_1135_redacted.json", "experiments/fabric_v2/results/fabric_v2_service_selftest_20260621_113419_redacted.json", + "experiments/fabric_v2/results/fabric_v2_scm_live_20260720_redacted.json", "experiments/fabric_v2/results/baseline_5agent_fabric_v2_frame_mailbox.json", "experiments/fabric_v2/results/fabric_v2_5agent_baseline_redacted.json", "experiments/fabric_v2/results/fabric_v2_host_selftest_20260621_074925_redacted.json", @@ -97,6 +98,8 @@ artifacts = [ "experiments/fabric_v2/results/fabric_v2_router_state_20260621_081434_redacted.json", "experiments/fabric_v2/results/fabric_v2_selftest_20260621_073951_redacted.json", "experiments/win32_probe/results/runtime_sid_probe_PASS_redacted.json", + "experiments/win32_probe/results/tpm_platform_attestation_live_20260720_redacted.json", + "experiments/win32_probe/results/tpm_platform_attestation_selftest_20260720_redacted.json", ] include = [ "_win32_abi.py", @@ -131,6 +134,7 @@ include = [ "sc_nats_bridge.py", "sc_spawn.py", "sc_tasks.py", + "sc_tpm_attestation.py", "sc_transcript.py", "claudego/**", "experiments/win32_probe/CAPABILITY_BACKLOG.md", @@ -151,11 +155,14 @@ include = [ "experiments/win32_probe/results/channel_router_composition_PASS_redacted.json", "experiments/win32_probe/results/pipe_role_lease_PASS_redacted.json", "experiments/win32_probe/results/runtime_sid_probe_PASS_redacted.json", + "experiments/win32_probe/results/tpm_platform_attestation_live_20260720_redacted.json", + "experiments/win32_probe/results/tpm_platform_attestation_selftest_20260720_redacted.json", "experiments/win32_probe/results/terminal_active_tab_guard_LIVE_PASS_redacted.json", "experiments/win32_probe/results/uia_echo_filter_PASS_redacted.json", "experiments/fabric_v2/results/baseline_5agent_fabric_v2_service_transport.json", "experiments/fabric_v2/results/SC_FABRIC_SERVICE_20260621_1135_redacted.json", "experiments/fabric_v2/results/fabric_v2_service_selftest_20260621_113419_redacted.json", + "experiments/fabric_v2/results/fabric_v2_scm_live_20260720_redacted.json", "experiments/fabric_v2/results/baseline_5agent_fabric_v2_frame_mailbox.json", "experiments/fabric_v2/results/fabric_v2_5agent_baseline_redacted.json", "experiments/fabric_v2/results/fabric_v2_host_selftest_20260621_074925_redacted.json", @@ -186,6 +193,7 @@ include = [ "docs/UIA_ECHO_FILTER_TERMCONTROL.md", "docs/PROVEN_VS_UNTESTED.md", "docs/SELFCONNECT_PRODUCT_BOUNDARIES.md", + "docs/TPM_PLATFORM_ATTESTATION.md", "skills/selfconnect-win32/**", ] @@ -194,6 +202,7 @@ artifacts = [ "experiments/fabric_v2/results/baseline_5agent_fabric_v2_service_transport.json", "experiments/fabric_v2/results/SC_FABRIC_SERVICE_20260621_1135_redacted.json", "experiments/fabric_v2/results/fabric_v2_service_selftest_20260621_113419_redacted.json", + "experiments/fabric_v2/results/fabric_v2_scm_live_20260720_redacted.json", "experiments/fabric_v2/results/baseline_5agent_fabric_v2_frame_mailbox.json", "experiments/fabric_v2/results/fabric_v2_5agent_baseline_redacted.json", "experiments/fabric_v2/results/fabric_v2_host_selftest_20260621_074925_redacted.json", @@ -202,6 +211,8 @@ artifacts = [ "experiments/fabric_v2/results/fabric_v2_router_state_20260621_081434_redacted.json", "experiments/fabric_v2/results/fabric_v2_selftest_20260621_073951_redacted.json", "experiments/win32_probe/results/runtime_sid_probe_PASS_redacted.json", + "experiments/win32_probe/results/tpm_platform_attestation_live_20260720_redacted.json", + "experiments/win32_probe/results/tpm_platform_attestation_selftest_20260720_redacted.json", ] [project.scripts] @@ -215,6 +226,7 @@ selfconnect-fabric-host = "sc_fabric_host:main" selfconnect-fabric-router = "sc_fabric_router:main" selfconnect-fabric-service = "sc_fabric_service:main" selfconnect-local-model = "sc_local_model_role:main" +selfconnect-tpm-attestation = "sc_tpm_attestation:main" [tool.pytest.ini_options] testpaths = ["tests", "test_self_connect.py"] diff --git a/release/claims.json b/release/claims.json index 2d2b61c..a635210 100644 --- a/release/claims.json +++ b/release/claims.json @@ -94,7 +94,7 @@ }, { "id": "package.installation_extras", - "statement": "The package declares optional UIA, MCP, Windows service, Telegram, ClaudeGo, and full dependency groups.", + "statement": "The package declares optional UIA, trust, MCP, Windows service, Telegram, ClaudeGo, and full dependency groups.", "status": "proven", "release": true, "scope": "Dependency declarations in pyproject.toml.", @@ -108,7 +108,7 @@ "evidence": [ { "path": "pyproject.toml", - "sha256_text": "3421c7c0961457d891f207d27c683536a3f661a1ee10389aeb50318b1fa2c9de" + "sha256_text": "ce2a1f486016b1a3fff5f1962e834f541cd375a5b0d276a7c4c2c6c45f6d5c49" } ] }, @@ -703,6 +703,37 @@ } ] }, + { + "id": "fabric_v2.scm_live_service", + "statement": "The SelfConnect Fabric V2 service is installed as an auto-start LocalSystem Windows service on the release workstation and passed live ACK, graceful-restart replay, forced-crash recovery, and post-recovery traffic checks.", + "status": "proven", + "release": true, + "scope": "The SelfConnectFabricV2 SCM service on the tested Windows release workstation.", + "boundary": "This one-host result does not establish fleet deployment, signed-installer policy, every target machine, or default routing of every SelfConnect workload through Fabric V2.", + "verified_on": "2026-07-20", + "evidence": [ + { + "path": "experiments/fabric_v2/results/fabric_v2_scm_live_20260720_redacted.json", + "sha256_text": "24f7ca09dce2c0aadc855949014ce1867abf1a02ec450daf6a793effd708505b", + "assertions": [ + {"path": "ok", "equals": true}, + {"path": "installed", "equals": true}, + {"path": "live_roundtrip", "equals": true}, + {"path": "replay_rejected_after_graceful_restart", "equals": true}, + {"path": "scm_restarted_with_new_pid", "equals": true}, + {"path": "post_crash_roundtrip", "equals": true} + ] + }, + { + "path": "sc_fabric_host.py", + "sha256_text": "e4817749e2d05f8db69dda7a476cc5f8a18f0bb75fdb4b4e9057ce960bb75fb5" + }, + { + "path": "sc_fabric_windows_svc.py", + "sha256_text": "4ab1d65ba24160b7089a116556bc8ffa988991bf87b6be0fd0ab0bec175bd885" + } + ] + }, { "id": "real_agent.twenty_agent_portable_release_evidence", "statement": "A portable release claim for the twenty-real-agent ladder is not yet approved because the raw local evidence is intentionally ignored and no tracked redacted run artifact is attached to this ledger.", @@ -719,15 +750,42 @@ }, { "id": "identity.tpm_platform_attestation", - "statement": "TPM platform attestation is pending.", - "status": "pending", - "release": false, - "scope": "No production attestation claim.", - "boundary": "A TPM capability probe or TPM-backed signing must not be described as remote-verifiable platform attestation.", - "verified_on": "2026-07-09", + "statement": "SelfConnect provisions a machine-scoped non-exportable PCP identity key and locally issues and verifies nonce-bound TPM platform quotes covering PCR 0 through PCR 23 with durable replay rejection.", + "status": "proven", + "release": true, + "scope": "The packaged Windows TPM 2.0 mechanism and the tested release workstation's SelfConnectPlatformAIK-v1 key.", + "boundary": "This does not establish manufacturer or endorsement-key certificate-chain trust, remote fleet enrollment, verifier-policy distribution, certificate revocation, independent assessment, or authorization to operate.", + "verified_on": "2026-07-20", "evidence": [ { - "path": "experiments/win32_probe/CAPABILITY_BACKLOG.md" + "path": "experiments/win32_probe/results/tpm_platform_attestation_live_20260720_redacted.json", + "sha256_text": "788945911da23576ea262a77d05c7f0f455a7637ab7333bcc1e22d5f66e6ab35", + "assertions": [ + {"path": "ok", "equals": true}, + {"path": "pcr_mask", "equals": 16777215}, + {"path": "replay_checked", "equals": true}, + {"path": "replay_rejected", "equals": true}, + {"path": "manufacturer_chain_verified", "equals": false}, + {"path": "raw_claim_included", "equals": false} + ] + }, + { + "path": "experiments/win32_probe/results/tpm_platform_attestation_selftest_20260720_redacted.json", + "sha256_text": "700f8db598d302a5fa2e03ac1500fb596af52146d58ad4b8cc0a05c9aad4149b", + "assertions": [ + {"path": "quote_verified", "equals": true}, + {"path": "nonce_mismatch_rejected", "equals": true}, + {"path": "tampered_pcr_rejected", "equals": true}, + {"path": "private_key_exportable", "equals": false} + ] + }, + { + "path": "sc_tpm_attestation.py", + "sha256_text": "9ca6abbb1b4b704f82413a600d221051a7900f5e10e428a063a35f20446cff7e" + }, + { + "path": "tests/test_tpm_attestation.py", + "sha256_text": "ecfd175db3e009cc126933e0098dcb603b8ae67255e53d6c43354602dd0efda2" } ] }, diff --git a/runbooks/fabric_v2_windows_service.md b/runbooks/fabric_v2_windows_service.md index cc70e49..79f634f 100644 --- a/runbooks/fabric_v2_windows_service.md +++ b/runbooks/fabric_v2_windows_service.md @@ -24,13 +24,15 @@ Key files: ## Prerequisites -1. **Python with pywin32 installed:** +1. **The service interpreter with SelfConnect and pywin32 installed system-wide:** ``` - pip install pywin32 - python Scripts/pywin32_postinstall.py -install + python -m pip install --no-user ".[service]" + python -m pywin32_postinstall -install ``` - The post-install step registers the pywin32 service DLLs with Windows. - Skip this and the service will fail to start with a DLL-not-found error. + `LocalSystem` cannot see packages installed only in a user's site directory. + The post-install step registers the pywin32 service DLLs with Windows. Skip + either system-scope installation or the DLL registration and the service can + fail during import before it reaches SelfConnect. 2. **Admin rights** — install, remove, start, and stop all require an elevated prompt. Query and status checks do not require elevation. @@ -189,6 +191,12 @@ Then restart the elevated prompt and retry. ## Verified +- 2026-07-20 — Installed as the auto-start `SelfConnectFabricV2` LocalSystem + service on the release workstation. A real named-pipe ACK completed, replay + rejection survived graceful restart, forced process termination triggered SCM + recovery with a new PID, and a post-recovery roundtrip completed. Recovery + delays are 5/15/60 seconds with a one-day reset period. Evidence: + `experiments/fabric_v2/results/fabric_v2_scm_live_20260720_redacted.json`. - Session 15 (2026-06-21) — `SelfConnectFabricV2` SCM service install/start/stop/remove proved; pywin32 `ServiceFramework` wrapper confirmed; state restore and watchdog checks passed; DACL hardening (owner SID + SYSTEM, deny-all fallback) confirmed in pipe layer. diff --git a/sc_fabric_host.py b/sc_fabric_host.py index e7a7ebd..12d632b 100644 --- a/sc_fabric_host.py +++ b/sc_fabric_host.py @@ -9,6 +9,7 @@ from __future__ import annotations import argparse +import ctypes import hashlib import json import sys @@ -60,22 +61,57 @@ class IocpCompletionQueue: """Small wrapper over Win32 IOCP post/get used by the host worker.""" def __init__(self) -> None: - if win32file is None or sys.platform != "win32": + if sys.platform != "win32": raise IocpUnavailableError("Windows IOCP APIs are unavailable") - self.handle = win32file.CreateIoCompletionPort(win32file.INVALID_HANDLE_VALUE, None, 0, 0) + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + self._create = kernel32.CreateIoCompletionPort + self._create.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_size_t, ctypes.c_uint32] + self._create.restype = ctypes.c_void_p + self._post = kernel32.PostQueuedCompletionStatus + self._post.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.c_size_t, ctypes.c_void_p] + self._post.restype = ctypes.c_int + self._get = kernel32.GetQueuedCompletionStatus + self._get.argtypes = [ + ctypes.c_void_p, + ctypes.POINTER(ctypes.c_uint32), + ctypes.POINTER(ctypes.c_size_t), + ctypes.POINTER(ctypes.c_void_p), + ctypes.c_uint32, + ] + self._get.restype = ctypes.c_int + self._close = kernel32.CloseHandle + self._close.argtypes = [ctypes.c_void_p] + self._close.restype = ctypes.c_int + self.handle = self._create(ctypes.c_void_p(-1), None, 0, 0) + if not self.handle: + raise ctypes.WinError(ctypes.get_last_error()) def post(self, *, completion_id: int, byte_count: int = 0) -> None: - win32file.PostQueuedCompletionStatus(self.handle, int(byte_count), int(completion_id), None) + if not self._post(self.handle, int(byte_count), int(completion_id), None): + raise ctypes.WinError(ctypes.get_last_error()) def get(self, timeout_ms: int) -> tuple[int, int]: - _rc, byte_count, completion_id, _overlapped = win32file.GetQueuedCompletionStatus( + byte_count = ctypes.c_uint32() + completion_id = ctypes.c_size_t() + overlapped = ctypes.c_void_p() + ok = self._get( self.handle, + ctypes.byref(byte_count), + ctypes.byref(completion_id), + ctypes.byref(overlapped), int(timeout_ms), ) - return int(completion_id), int(byte_count) + if not ok: + error = ctypes.get_last_error() + if error == 258: # WAIT_TIMEOUT + raise TimeoutError("IOCP completion wait timed out") + raise ctypes.WinError(error) + return int(completion_id.value), int(byte_count.value) def close(self) -> None: - self.handle.Close() + if self.handle: + self._close(self.handle) + self.handle = None def _require_overlapped_win32() -> None: @@ -285,7 +321,12 @@ def _completion_worker(self) -> None: while not self._stop.is_set(): try: completion_id, _byte_count = self._iocp.get(500) - except Exception: + except Exception as exc: + message = f"iocp_wait:{exc.__class__.__name__}:{exc}" + with self._lock: + if not self._errors or self._errors[-1] != message: + self._errors.append(message) + del self._errors[:-20] continue if completion_id == STOP_KEY: break diff --git a/sc_fabric_service.py b/sc_fabric_service.py index b779dfb..170e927 100644 --- a/sc_fabric_service.py +++ b/sc_fabric_service.py @@ -161,6 +161,8 @@ def save_state(self) -> Path: state = self._router.snapshot_state() if self._session is not None: state["accepted_sequences"] = self._session.export_replay_state() + if self._host is not None: + state["host_stats"] = self._host.stats() state["service_pid"] = os.getpid() state["schema_version"] = SCHEMA_VERSION try: diff --git a/sc_fabric_windows_svc.py b/sc_fabric_windows_svc.py index 60d56c9..98ce01e 100644 --- a/sc_fabric_windows_svc.py +++ b/sc_fabric_windows_svc.py @@ -27,6 +27,16 @@ ) +def _wait_for_stop(stop_handle: Any, *, poll_ms: int = 250) -> None: + """Poll the SCM stop event without pinning the service's Python thread.""" + while True: + result = win32event.WaitForSingleObject(stop_handle, int(poll_ms)) + if result == win32event.WAIT_OBJECT_0: + return + if result != win32event.WAIT_TIMEOUT: + raise RuntimeError(f"unexpected service stop wait result: {result}") + + if _WIN32_SVC: class SelfConnectFabricWinSvc(win32serviceutil.ServiceFramework): @@ -55,7 +65,7 @@ def SvcDoRun(self) -> None: self._fabric = svc_mod.FabricService(config) try: self._fabric.start() - win32event.WaitForSingleObject(self._hWaitStop, win32event.INFINITE) + _wait_for_stop(self._hWaitStop) finally: if self._fabric is not None: self._fabric.stop() @@ -97,11 +107,6 @@ def install_service() -> dict[str, Any]: description=SVC_DESCRIPTION, startType=win32service.SERVICE_AUTO_START, ) - win32serviceutil.ChangeServiceConfig( - pythonClassString=None, - serviceName=SVC_NAME, - serviceName2=SVC_NAME, - ) return {"ok": True, "action": "installed", "service_name": SVC_NAME} except Exception as exc: return {"ok": False, "error": str(exc)} diff --git a/sc_tpm_attestation.py b/sc_tpm_attestation.py new file mode 100644 index 0000000..69be7df --- /dev/null +++ b/sc_tpm_attestation.py @@ -0,0 +1,743 @@ +"""Nonce-bound TPM 2.0 platform attestation for SelfConnect on Windows. + +This module provisions a non-exportable RSA identity key in the Microsoft +Platform Crypto Provider, asks Windows for an NCRYPT_CLAIM_PLATFORM quote, and +verifies the returned TPM quote without trusting caller-supplied metadata. + +The local verifier proves possession of the provisioned TPM key, freshness, +the selected PCR values, and the quote signature. Manufacturer/EK certificate +chain approval remains a relying-party deployment policy, not an implicit +claim made by this module. +""" + +from __future__ import annotations + +import argparse +import base64 +import ctypes +import hashlib +import json +import os +import struct +import sys +import tempfile +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any, ClassVar + +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.asymmetric import padding, rsa +from sc_tasks import FileLock + +SCHEMA_VERSION = 1 +PROVIDER = "Microsoft Platform Crypto Provider" +DEFAULT_KEY_NAME = "SelfConnectPlatformAIK-v1" +DEFAULT_PCR_MASK = 0xFFFFFF +MAX_CLAIM_BYTES = 64 * 1024 +NCRYPT_MACHINE_KEY_FLAG = 0x20 +NCRYPT_OVERWRITE_KEY_FLAG = 0x80 +NCRYPT_CLAIM_PLATFORM = 0x00010000 +NCRYPTBUFFER_TPM_PLATFORM_CLAIM_PCR_MASK = 80 +NCRYPTBUFFER_TPM_PLATFORM_CLAIM_NONCE = 81 +NCRYPT_PCP_IDENTITY_KEY = 0x00000008 +NCRYPT_TPM12_PROVIDER = 0x00010000 +NCRYPT_IMPL_HARDWARE_FLAG = 0x00000001 +TPM_GENERATED_VALUE = 0xFF544347 +TPM_ST_ATTEST_QUOTE = 0x8018 +TPM_ALG_SHA1 = 0x0004 +TPM_ALG_SHA256 = 0x000B +TPM_ALG_SHA384 = 0x000C +TPM_ALG_SHA512 = 0x000D +TPM_ALG_RSASSA = 0x0014 +_PLATFORM_MAGIC = b"ALPT" + + +class TpmAttestationError(RuntimeError): + """Base error for provisioning, issuance, or verification failures.""" + + +class TpmAttestationVerificationError(TpmAttestationError): + """The claim did not satisfy cryptographic or policy checks.""" + + +class TpmAttestationReplayError(TpmAttestationVerificationError): + """The verifier nonce was already consumed.""" + + +class _NCryptBuffer(ctypes.Structure): + _fields_: ClassVar[list[tuple[str, Any]]] = [ + ("cbBuffer", ctypes.c_ulong), + ("BufferType", ctypes.c_ulong), + ("pvBuffer", ctypes.c_void_p), + ] + + +class _NCryptBufferDesc(ctypes.Structure): + _fields_: ClassVar[list[tuple[str, Any]]] = [ + ("ulVersion", ctypes.c_ulong), + ("cBuffers", ctypes.c_ulong), + ("pBuffers", ctypes.POINTER(_NCryptBuffer)), + ] + + +def _b64u(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def _unb64u(value: str) -> bytes: + if not isinstance(value, str) or not value or len(value) > MAX_CLAIM_BYTES * 2: + raise TpmAttestationVerificationError("invalid base64url field") + try: + return base64.b64decode(value + "=" * (-len(value) % 4), altchars=b"-_", validate=True) + except Exception as exc: + raise TpmAttestationVerificationError("invalid base64url field") from exc + + +def _sha256(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _status(label: str, code: int) -> None: + if code != 0: + raise TpmAttestationError(f"{label} failed: 0x{code & 0xFFFFFFFF:08x}") + + +class _NCrypt: + def __init__(self) -> None: + if sys.platform != "win32": + raise TpmAttestationError("TPM attestation requires Windows") + self.dll = ctypes.WinDLL("ncrypt", use_last_error=True) + p = ctypes.c_void_p + d = ctypes.c_ulong + self.open_provider = self._bind( + "NCryptOpenStorageProvider", [ctypes.POINTER(p), ctypes.c_wchar_p, d] + ) + self.create_key = self._bind( + "NCryptCreatePersistedKey", + [p, ctypes.POINTER(p), ctypes.c_wchar_p, ctypes.c_wchar_p, d, d], + ) + self.open_key = self._bind( + "NCryptOpenKey", [p, ctypes.POINTER(p), ctypes.c_wchar_p, d, d] + ) + self.set_property = self._bind( + "NCryptSetProperty", [p, ctypes.c_wchar_p, p, d, d] + ) + self.get_property = self._bind( + "NCryptGetProperty", [p, ctypes.c_wchar_p, p, d, ctypes.POINTER(d), d] + ) + self.finalize_key = self._bind("NCryptFinalizeKey", [p, d]) + self.delete_key = self._bind("NCryptDeleteKey", [p, d]) + self.free_object = self._bind("NCryptFreeObject", [p]) + self.export_key = self._bind( + "NCryptExportKey", [p, p, ctypes.c_wchar_p, p, p, d, ctypes.POINTER(d), d] + ) + self.create_claim = self._bind( + "NCryptCreateClaim", + [p, p, d, ctypes.POINTER(_NCryptBufferDesc), p, d, ctypes.POINTER(d), d], + ) + + def _bind(self, name: str, args: list[Any]) -> Any: + fn = getattr(self.dll, name) + fn.argtypes = args + fn.restype = ctypes.c_long + return fn + + +class _Handles: + def __init__(self, api: _NCrypt, provider: int, key: int = 0) -> None: + self.api = api + self.provider = ctypes.c_void_p(provider) + self.key = ctypes.c_void_p(key) + + def close(self) -> None: + if self.key: + self.api.free_object(self.key) + self.key = ctypes.c_void_p() + if self.provider: + self.api.free_object(self.provider) + self.provider = ctypes.c_void_p() + + def __enter__(self) -> _Handles: + return self + + def __exit__(self, *_args: Any) -> None: + self.close() + + +def _open_provider(api: _NCrypt) -> ctypes.c_void_p: + provider = ctypes.c_void_p() + _status("NCryptOpenStorageProvider", api.open_provider(ctypes.byref(provider), PROVIDER, 0)) + return provider + + +def _open_key(api: _NCrypt, key_name: str) -> _Handles: + if not key_name or len(key_name) > 128 or any(ord(ch) < 32 for ch in key_name): + raise ValueError("invalid TPM key name") + provider = _open_provider(api) + key = ctypes.c_void_p() + try: + _status( + "NCryptOpenKey", + api.open_key(provider, ctypes.byref(key), key_name, 0, NCRYPT_MACHINE_KEY_FLAG), + ) + except Exception: + api.free_object(provider) + raise + return _Handles(api, provider.value or 0, key.value or 0) + + +def _get_dword(api: _NCrypt, key: ctypes.c_void_p, name: str) -> int: + value = ctypes.c_ulong() + size = ctypes.c_ulong() + _status( + f"NCryptGetProperty({name})", + api.get_property(key, name, ctypes.byref(value), ctypes.sizeof(value), ctypes.byref(size), 0), + ) + if size.value != ctypes.sizeof(value): + raise TpmAttestationError(f"unexpected {name} property size") + return int(value.value) + + +def _export_public_blob(api: _NCrypt, key: ctypes.c_void_p) -> bytes: + size = ctypes.c_ulong() + _status( + "NCryptExportKey(size)", + api.export_key(key, None, "RSAPUBLICBLOB", None, None, 0, ctypes.byref(size), 0), + ) + if size.value <= 24 or size.value > 16 * 1024: + raise TpmAttestationError("unexpected public-key blob size") + out = (ctypes.c_ubyte * size.value)() + got = ctypes.c_ulong() + _status( + "NCryptExportKey", + api.export_key( + key, None, "RSAPUBLICBLOB", None, out, size.value, ctypes.byref(got), 0 + ), + ) + return bytes(out[: got.value]) + + +def _assert_identity_key( + api: _NCrypt, provider: ctypes.c_void_p, key: ctypes.c_void_p +) -> dict[str, Any]: + impl = _get_dword(api, provider, "Impl Type") + export_policy = _get_dword(api, key, "Export Policy") + usage = _get_dword(api, key, "PCP_KEY_USAGE_POLICY") + if not impl & NCRYPT_IMPL_HARDWARE_FLAG: + raise TpmAttestationError("identity key is not hardware-backed") + if export_policy != 0: + raise TpmAttestationError("identity key is exportable") + if not usage & NCRYPT_PCP_IDENTITY_KEY: + raise TpmAttestationError("key is not a PCP identity key") + return { + "hardware_backed": True, + "private_key_exportable": False, + "pcp_identity_key": True, + } + + +def provision_identity_key( + key_name: str = DEFAULT_KEY_NAME, *, overwrite: bool = False +) -> dict[str, Any]: + """Create and validate a machine-scoped, non-exportable TPM identity key.""" + api = _NCrypt() + provider = _open_provider(api) + key = ctypes.c_void_p() + flags = NCRYPT_MACHINE_KEY_FLAG | (NCRYPT_OVERWRITE_KEY_FLAG if overwrite else 0) + try: + _status( + "NCryptCreatePersistedKey", + api.create_key(provider, ctypes.byref(key), "RSA", key_name, 0, flags), + ) + bits = ctypes.c_ulong(2048) + _status( + "NCryptSetProperty(Length)", + api.set_property(key, "Length", ctypes.byref(bits), ctypes.sizeof(bits), 0), + ) + usage = ctypes.c_ulong(NCRYPT_TPM12_PROVIDER | NCRYPT_PCP_IDENTITY_KEY) + _status( + "NCryptSetProperty(PCP_KEY_USAGE_POLICY)", + api.set_property( + key, "PCP_KEY_USAGE_POLICY", ctypes.byref(usage), ctypes.sizeof(usage), 0 + ), + ) + _status("NCryptFinalizeKey", api.finalize_key(key, 0)) + properties = _assert_identity_key(api, provider, key) + public_blob = _export_public_blob(api, key) + return { + "schema_version": SCHEMA_VERSION, + "provider": PROVIDER, + "key_name": key_name, + "public_key_sha256": _sha256(public_blob), + **properties, + } + except Exception: + if key: + api.delete_key(key, 0) + key = ctypes.c_void_p() + raise + finally: + if key: + api.free_object(key) + api.free_object(provider) + + +def _claim_parameters(nonce: bytes, pcr_mask: int) -> tuple[Any, _NCryptBufferDesc]: + if not isinstance(nonce, bytes) or not 16 <= len(nonce) <= 64: + raise ValueError("nonce must be 16..64 bytes") + if not 0 < pcr_mask <= DEFAULT_PCR_MASK: + raise ValueError("PCR mask must select PCR 0..23") + nonce_buffer = ctypes.create_string_buffer(nonce) + mask = ctypes.c_ulong(pcr_mask) + buffers = (_NCryptBuffer * 2)( + _NCryptBuffer( + ctypes.sizeof(mask), + NCRYPTBUFFER_TPM_PLATFORM_CLAIM_PCR_MASK, + ctypes.cast(ctypes.byref(mask), ctypes.c_void_p), + ), + _NCryptBuffer( + len(nonce), + NCRYPTBUFFER_TPM_PLATFORM_CLAIM_NONCE, + ctypes.cast(nonce_buffer, ctypes.c_void_p), + ), + ) + desc = _NCryptBufferDesc(0, len(buffers), buffers) + return (nonce_buffer, mask, buffers), desc + + +def issue_platform_attestation( + expected_nonce: bytes, + *, + key_name: str = DEFAULT_KEY_NAME, + pcr_mask: int = DEFAULT_PCR_MASK, +) -> dict[str, Any]: + """Issue a TPM quote bound to a verifier-provided nonce and PCR mask.""" + api = _NCrypt() + with _open_key(api, key_name) as handles: + properties = _assert_identity_key(api, handles.provider, handles.key) + public_blob = _export_public_blob(api, handles.key) + keepalive, params = _claim_parameters(expected_nonce, pcr_mask) + _ = keepalive + size = ctypes.c_ulong() + _status( + "NCryptCreateClaim(size)", + api.create_claim( + None, + handles.key, + NCRYPT_CLAIM_PLATFORM, + ctypes.byref(params), + None, + 0, + ctypes.byref(size), + 0, + ), + ) + if size.value <= 24 or size.value > MAX_CLAIM_BYTES: + raise TpmAttestationError("unexpected platform claim size") + claim = (ctypes.c_ubyte * size.value)() + got = ctypes.c_ulong() + _status( + "NCryptCreateClaim", + api.create_claim( + None, + handles.key, + NCRYPT_CLAIM_PLATFORM, + ctypes.byref(params), + claim, + size.value, + ctypes.byref(got), + 0, + ), + ) + claim_bytes = bytes(claim[: got.value]) + return { + "schema_version": SCHEMA_VERSION, + "claim_type": "tpm-platform", + "provider": PROVIDER, + "key_name": key_name, + "public_key_blob_b64u": _b64u(public_blob), + "public_key_sha256": _sha256(public_blob), + "claim_b64u": _b64u(claim_bytes), + "claim_sha256": _sha256(claim_bytes), + "nonce_sha256": _sha256(expected_nonce), + "pcr_mask": pcr_mask, + "issued_at": int(time.time()), + **properties, + } + + +class _Reader: + def __init__(self, data: bytes) -> None: + self.data = data + self.pos = 0 + + def take(self, size: int) -> bytes: + if size < 0 or self.pos + size > len(self.data): + raise TpmAttestationVerificationError("truncated TPM structure") + out = self.data[self.pos : self.pos + size] + self.pos += size + return out + + def u8(self) -> int: + return self.take(1)[0] + + def u16(self) -> int: + return int.from_bytes(self.take(2), "big") + + def u32(self) -> int: + return int.from_bytes(self.take(4), "big") + + def tpm2b(self) -> bytes: + return self.take(self.u16()) + + def done(self) -> None: + if self.pos != len(self.data): + raise TpmAttestationVerificationError("trailing TPM structure bytes") + + +@dataclass(frozen=True) +class _Quote: + nonce: bytes + selections: tuple[tuple[int, bytes], ...] + pcr_digest: bytes + + +def _parse_quote(quote: bytes) -> _Quote: + r = _Reader(quote) + if r.u32() != TPM_GENERATED_VALUE or r.u16() != TPM_ST_ATTEST_QUOTE: + raise TpmAttestationVerificationError("not a TPM quote") + r.tpm2b() # qualified signer + nonce = r.tpm2b() + r.take(8 + 4 + 4 + 1 + 8) # clockInfo + firmwareVersion + count = r.u32() + if not 1 <= count <= 8: + raise TpmAttestationVerificationError("invalid PCR selection count") + selections: list[tuple[int, bytes]] = [] + for _ in range(count): + alg = r.u16() + select_size = r.u8() + if not 1 <= select_size <= 4: + raise TpmAttestationVerificationError("invalid PCR selection size") + selections.append((alg, r.take(select_size))) + pcr_digest = r.tpm2b() + r.done() + return _Quote(nonce, tuple(selections), pcr_digest) + + +def _parse_signature(signature: bytes) -> tuple[int, int, bytes]: + r = _Reader(signature) + scheme = r.u16() + hash_alg = r.u16() + sig = r.tpm2b() + r.done() + return scheme, hash_alg, sig + + +def _rsa_public_key(blob: bytes) -> rsa.RSAPublicKey: + if len(blob) < 24: + raise TpmAttestationVerificationError("truncated RSA public blob") + magic, bit_length, cb_exp, cb_modulus, cb_p1, cb_p2 = struct.unpack("<6I", blob[:24]) + if magic != int.from_bytes(b"RSA1", "little") or cb_p1 or cb_p2: + raise TpmAttestationVerificationError("invalid RSA public blob") + if bit_length != cb_modulus * 8 or bit_length < 2048: + raise TpmAttestationVerificationError("invalid RSA public-key size") + if len(blob) != 24 + cb_exp + cb_modulus: + raise TpmAttestationVerificationError("invalid RSA public-blob length") + exponent = int.from_bytes(blob[24 : 24 + cb_exp], "big") + modulus = int.from_bytes(blob[24 + cb_exp :], "big") + try: + return rsa.RSAPublicNumbers(exponent, modulus).public_key() + except ValueError as exc: + raise TpmAttestationVerificationError("invalid RSA public key") from exc + + +_HASHES: dict[int, type[hashes.HashAlgorithm]] = { + TPM_ALG_SHA1: hashes.SHA1, + TPM_ALG_SHA256: hashes.SHA256, + TPM_ALG_SHA384: hashes.SHA384, + TPM_ALG_SHA512: hashes.SHA512, +} + + +def _mask_bytes(pcr_mask: int, size: int) -> bytes: + return int(pcr_mask).to_bytes(size, "little") + + +def _selected_pcr_digest( + pcrs: bytes, *, pcr_alg: int, digest_alg: int, mask: bytes +) -> bytes: + pcr_hash_type = _HASHES.get(pcr_alg) + digest_hash_type = _HASHES.get(digest_alg) + if pcr_hash_type is None or digest_hash_type is None: + raise TpmAttestationVerificationError("unsupported PCR hash algorithm") + digest_size = pcr_hash_type().digest_size + if len(pcrs) != 24 * digest_size: + raise TpmAttestationVerificationError("unexpected PCR value inventory") + selected = b"".join( + pcrs[index * digest_size : (index + 1) * digest_size] + for index in range(24) + if mask[index // 8] & (1 << (index % 8)) + ) + digest = hashes.Hash(digest_hash_type()) + digest.update(selected) + return digest.finalize() + + +def _strict_artifact(artifact: dict[str, Any]) -> dict[str, Any]: + required = { + "schema_version", + "claim_type", + "provider", + "key_name", + "public_key_blob_b64u", + "public_key_sha256", + "claim_b64u", + "claim_sha256", + "nonce_sha256", + "pcr_mask", + "issued_at", + "hardware_backed", + "private_key_exportable", + "pcp_identity_key", + } + if type(artifact) is not dict or set(artifact) != required: + raise TpmAttestationVerificationError("invalid attestation artifact shape") + if artifact["schema_version"] != SCHEMA_VERSION or artifact["claim_type"] != "tpm-platform": + raise TpmAttestationVerificationError("unsupported attestation artifact") + if artifact["provider"] != PROVIDER: + raise TpmAttestationVerificationError("unexpected key provider") + if artifact["hardware_backed"] is not True or artifact["private_key_exportable"] is not False: + raise TpmAttestationVerificationError("artifact does not assert a hardware non-exportable key") + if artifact["pcp_identity_key"] is not True: + raise TpmAttestationVerificationError("artifact does not assert a PCP identity key") + if type(artifact["pcr_mask"]) is not int or not 0 < artifact["pcr_mask"] <= DEFAULT_PCR_MASK: + raise TpmAttestationVerificationError("invalid PCR mask") + if type(artifact["issued_at"]) is not int or artifact["issued_at"] <= 0: + raise TpmAttestationVerificationError("invalid issuance time") + return dict(artifact) + + +def verify_platform_attestation( + artifact: dict[str, Any], + expected_nonce: bytes, + *, + expected_public_key_sha256: str, + expected_pcr_mask: int = DEFAULT_PCR_MASK, +) -> dict[str, Any]: + """Cryptographically verify a platform quote against verifier policy.""" + item = _strict_artifact(artifact) + if not isinstance(expected_nonce, bytes) or not 16 <= len(expected_nonce) <= 64: + raise ValueError("expected nonce must be 16..64 bytes") + if item["nonce_sha256"] != _sha256(expected_nonce): + raise TpmAttestationVerificationError("nonce metadata mismatch") + if item["pcr_mask"] != expected_pcr_mask: + raise TpmAttestationVerificationError("PCR mask policy mismatch") + public_blob = _unb64u(item["public_key_blob_b64u"]) + claim = _unb64u(item["claim_b64u"]) + if len(claim) > MAX_CLAIM_BYTES or _sha256(claim) != item["claim_sha256"]: + raise TpmAttestationVerificationError("claim digest mismatch") + public_digest = _sha256(public_blob) + if public_digest != item["public_key_sha256"] or public_digest != expected_public_key_sha256: + raise TpmAttestationVerificationError("attestation key is not pinned") + if len(claim) < 24 or claim[:4] != _PLATFORM_MAGIC: + raise TpmAttestationVerificationError("invalid platform claim header") + _magic, version, pcr_alg, cb_signature, cb_quote, cb_pcrs = struct.unpack( + "<6I", claim[:24] + ) + if version != 0 or any(size <= 0 for size in (cb_signature, cb_quote, cb_pcrs)): + raise TpmAttestationVerificationError("invalid platform claim version or sizes") + if 24 + cb_signature + cb_quote + cb_pcrs != len(claim): + raise TpmAttestationVerificationError("platform claim length mismatch") + signature = claim[24 : 24 + cb_signature] + quote = claim[24 + cb_signature : 24 + cb_signature + cb_quote] + pcrs = claim[-cb_pcrs:] + parsed = _parse_quote(quote) + scheme, signature_hash, signature_bytes = _parse_signature(signature) + hash_type = _HASHES.get(signature_hash) + if scheme != TPM_ALG_RSASSA or hash_type is None: + raise TpmAttestationVerificationError("unsupported TPM quote signature") + if parsed.nonce != expected_nonce: + raise TpmAttestationVerificationError("TPM quote nonce mismatch") + expected_mask = _mask_bytes(expected_pcr_mask, 3) + if parsed.selections != ((pcr_alg, expected_mask),): + raise TpmAttestationVerificationError("TPM quote PCR selection mismatch") + if parsed.pcr_digest != _selected_pcr_digest( + pcrs, pcr_alg=pcr_alg, digest_alg=signature_hash, mask=expected_mask + ): + raise TpmAttestationVerificationError("TPM quote PCR digest mismatch") + try: + _rsa_public_key(public_blob).verify( + signature_bytes, quote, padding.PKCS1v15(), hash_type() + ) + except Exception as exc: + raise TpmAttestationVerificationError("TPM quote signature verification failed") from exc + return { + "ok": True, + "schema_version": SCHEMA_VERSION, + "claim_sha256": item["claim_sha256"], + "public_key_sha256": public_digest, + "nonce_sha256": item["nonce_sha256"], + "pcr_mask": expected_pcr_mask, + "pcr_algorithm": pcr_alg, + "pcr_values_sha256": _sha256(pcrs), + "manufacturer_chain_verified": False, + "raw_claim_included": False, + } + + +class NonceReplayStore: + """Durable single-use verifier nonce store with a cross-process file lock.""" + + def __init__(self, path: str | Path, *, retention_s: int = 86400) -> None: + self.path = Path(path) + self.lock_path = self.path.with_suffix(self.path.suffix + ".lock") + if retention_s < 60: + raise ValueError("retention_s must be at least 60") + self.retention_s = int(retention_s) + + def consume(self, nonce: bytes, *, now: int | None = None) -> str: + if not isinstance(nonce, bytes) or not 16 <= len(nonce) <= 64: + raise ValueError("nonce must be 16..64 bytes") + moment = int(time.time()) if now is None else int(now) + digest = _sha256(nonce) + self.path.parent.mkdir(parents=True, exist_ok=True) + with FileLock(self.lock_path, timeout=5.0): + data: dict[str, int] = {} + if self.path.exists(): + try: + loaded = json.loads(self.path.read_text(encoding="utf-8")) + if type(loaded) is not dict or any( + type(k) is not str or type(v) is not int for k, v in loaded.items() + ): + raise ValueError + data = loaded + except Exception as exc: + raise TpmAttestationReplayError("nonce replay store is corrupt") from exc + cutoff = moment - self.retention_s + data = {key: seen for key, seen in data.items() if seen >= cutoff} + if digest in data: + raise TpmAttestationReplayError("attestation nonce was already consumed") + data[digest] = moment + fd, temp_name = tempfile.mkstemp(prefix=self.path.name + ".", dir=self.path.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle: + json.dump(data, handle, sort_keys=True, separators=(",", ":")) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp_name, self.path) + finally: + try: + os.unlink(temp_name) + except FileNotFoundError: + pass + return digest + + +def verify_and_consume( + artifact: dict[str, Any], + expected_nonce: bytes, + *, + expected_public_key_sha256: str, + replay_store: NonceReplayStore, + expected_pcr_mask: int = DEFAULT_PCR_MASK, +) -> dict[str, Any]: + result = verify_platform_attestation( + artifact, + expected_nonce, + expected_public_key_sha256=expected_public_key_sha256, + expected_pcr_mask=expected_pcr_mask, + ) + replay_store.consume(expected_nonce) + result["replay_checked"] = True + return result + + +def hardware_selftest(*, key_name: str | None = None) -> dict[str, Any]: + """Provision an ephemeral TPM identity key and exercise positive/negative paths.""" + name = key_name or f"SelfConnectPlatformAIK-Selftest-{os.getpid()}-{time.time_ns()}" + api = _NCrypt() + created = False + try: + provisioned = provision_identity_key(name) + created = True + nonce = os.urandom(32) + artifact = issue_platform_attestation(nonce, key_name=name) + verified = verify_platform_attestation( + artifact, + nonce, + expected_public_key_sha256=provisioned["public_key_sha256"], + ) + nonce_rejected = False + try: + verify_platform_attestation( + artifact, + bytes([nonce[0] ^ 1]) + nonce[1:], + expected_public_key_sha256=provisioned["public_key_sha256"], + ) + except TpmAttestationVerificationError: + nonce_rejected = True + tampered = dict(artifact) + raw = bytearray(_unb64u(tampered["claim_b64u"])) + raw[-1] ^= 1 + tampered["claim_b64u"] = _b64u(bytes(raw)) + tampered["claim_sha256"] = _sha256(bytes(raw)) + tamper_rejected = False + try: + verify_platform_attestation( + tampered, + nonce, + expected_public_key_sha256=provisioned["public_key_sha256"], + ) + except TpmAttestationVerificationError: + tamper_rejected = True + return { + "ok": bool(verified["ok"] and nonce_rejected and tamper_rejected), + "schema_version": SCHEMA_VERSION, + "provider": PROVIDER, + "hardware_backed": provisioned["hardware_backed"], + "private_key_exportable": provisioned["private_key_exportable"], + "quote_verified": verified["ok"], + "nonce_mismatch_rejected": nonce_rejected, + "tampered_pcr_rejected": tamper_rejected, + "public_key_sha256": provisioned["public_key_sha256"], + "manufacturer_chain_verified": False, + "raw_claim_included": False, + } + finally: + if created: + try: + with _open_key(api, name) as handles: + _status("NCryptDeleteKey", api.delete_key(handles.key, 0)) + handles.key = ctypes.c_void_p() + except Exception: + pass + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="SelfConnect TPM platform attestation") + sub = parser.add_subparsers(dest="command", required=True) + p_provision = sub.add_parser("provision") + p_provision.add_argument("--key-name", default=DEFAULT_KEY_NAME) + p_provision.add_argument("--overwrite", action="store_true") + p_selftest = sub.add_parser("selftest") + p_selftest.add_argument("--output", default="") + return parser + + +def main(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + if args.command == "provision": + result = provision_identity_key(args.key_name, overwrite=args.overwrite) + else: + result = hardware_selftest() + if args.output: + path = Path(args.output) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 if result.get("ok", True) else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_fabric_windows_svc.py b/tests/test_fabric_windows_svc.py index 78f98e1..1e6585d 100644 --- a/tests/test_fabric_windows_svc.py +++ b/tests/test_fabric_windows_svc.py @@ -51,6 +51,44 @@ def test_install_remove_noop_when_not_available() -> None: assert remove_result == {"ok": False, "error": "win32service not available"} +def test_install_uses_supported_pywin32_arguments() -> None: + import sc_fabric_windows_svc as winsvc + + with ( + mock.patch.object(winsvc, "_WIN32_SVC", True), + mock.patch.object(winsvc.sys, "platform", "win32"), + mock.patch.object(winsvc.win32serviceutil, "InstallService") as install, + ): + result = winsvc.install_service() + + assert result == { + "ok": True, + "action": "installed", + "service_name": winsvc.SVC_NAME, + } + install.assert_called_once_with( + pythonClassString="sc_fabric_windows_svc.SelfConnectFabricWinSvc", + serviceName=winsvc.SVC_NAME, + displayName=winsvc.SVC_DISPLAY, + description=winsvc.SVC_DESCRIPTION, + startType=winsvc.win32service.SERVICE_AUTO_START, + ) + + +def test_service_stop_wait_is_bounded() -> None: + import sc_fabric_windows_svc as winsvc + + with mock.patch.object( + winsvc.win32event, + "WaitForSingleObject", + side_effect=[winsvc.win32event.WAIT_TIMEOUT, winsvc.win32event.WAIT_OBJECT_0], + ) as wait: + winsvc._wait_for_stop(object(), poll_ms=125) + + assert wait.call_count == 2 + assert all(call.args[1] == 125 for call in wait.call_args_list) + + # --------------------------------------------------------------------------- # 4. Module-level constants are present and correct # --------------------------------------------------------------------------- diff --git a/tests/test_tpm_attestation.py b/tests/test_tpm_attestation.py new file mode 100644 index 0000000..70cef35 --- /dev/null +++ b/tests/test_tpm_attestation.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import json + +import pytest +import sc_tpm_attestation as tpm + + +def test_nonce_replay_store_is_durable(tmp_path) -> None: + nonce = b"n" * 32 + path = tmp_path / "nonces.json" + tpm.NonceReplayStore(path).consume(nonce, now=1000) + + with pytest.raises(tpm.TpmAttestationReplayError): + tpm.NonceReplayStore(path).consume(nonce, now=1001) + + persisted = json.loads(path.read_text(encoding="utf-8")) + assert persisted == {tpm._sha256(nonce): 1000} + + +def test_nonce_replay_store_prunes_expired_entries(tmp_path) -> None: + path = tmp_path / "nonces.json" + store = tpm.NonceReplayStore(path, retention_s=60) + store.consume(b"a" * 32, now=1000) + store.consume(b"b" * 32, now=1061) + assert json.loads(path.read_text(encoding="utf-8")) == {tpm._sha256(b"b" * 32): 1061} + + +def test_strict_artifact_rejects_extra_fields() -> None: + artifact = { + "schema_version": 1, + "claim_type": "tpm-platform", + "provider": tpm.PROVIDER, + "key_name": "key", + "public_key_blob_b64u": "AA", + "public_key_sha256": "0" * 64, + "claim_b64u": "AA", + "claim_sha256": "0" * 64, + "nonce_sha256": "0" * 64, + "pcr_mask": tpm.DEFAULT_PCR_MASK, + "issued_at": 1, + "hardware_backed": True, + "private_key_exportable": False, + "pcp_identity_key": True, + "extra": True, + } + with pytest.raises(tpm.TpmAttestationVerificationError, match="shape"): + tpm._strict_artifact(artifact) + + +def test_platform_claim_parser_rejects_bad_magic() -> None: + with pytest.raises(tpm.TpmAttestationVerificationError, match="TPM quote"): + tpm._parse_quote(b"\x00" * 80) + + +@pytest.mark.skipif(tpm.sys.platform != "win32", reason="requires Windows TPM") +def test_hardware_selftest_when_explicitly_enabled(monkeypatch) -> None: + if not tpm.os.environ.get("SELFCONNECT_TEST_TPM_ATTESTATION"): + pytest.skip("set SELFCONNECT_TEST_TPM_ATTESTATION=1 for destructive TPM selftest") + result = tpm.hardware_selftest() + assert result["ok"] is True + assert result["quote_verified"] is True + assert result["nonce_mismatch_rejected"] is True + assert result["tampered_pcr_rejected"] is True + assert result["manufacturer_chain_verified"] is False