From 9b4c325ff91db40cfad24c6788d2f989ca7642d0 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Wed, 19 Aug 2026 21:29:01 -0400 Subject: [PATCH] feat: add authenticated recorder control channel Add an authenticated cross-process status and stop contract for an active Capture recorder, so a launcher can stop a recording and prove that the recording finished. - bind every request and response to the exact session, PID, process start identity, and request ID, with HMAC-SHA256 over an unguessable per-session capability - store the capability only in a current-user-only runtime descriptor, using POSIX owner modes, macOS extended-ACL clearing, and a protected current-user Windows DACL, each verified after it is applied - refuse to start the control channel, and therefore the recording, when that boundary cannot be established or verified - make stop idempotent and report success only after database and event integrity verification; a timeout or a failed check never returns a success-shaped status - retain atomic incomplete terminal metadata after timeout, failure, or a discovered crash, and remove exact runtime descriptors after normal exit, abort, direct stop, and authenticated stale-process discovery - shut the event writers down with a separate terminate_writers event, so producers stop before writers and the committed rows can be reconciled - expose status_recording, stop_recording, discover_recorders, and RecorderStatus, plus capture status and capture stop commands - add a focused macOS and Windows pull-request workflow for the real subprocess contract The endpoint binds to IPv4 loopback on an ephemeral port. The capability does not enter process arguments, logs, responses, or capture artifacts. Stale discovery verifies the owner-protected descriptor and the exact process identity, checks the Windows kernel process signal before fallible metadata calls, rejects PID reuse, and keeps an ambiguous or access-denied process conservatively live. Closes #77 Co-Authored-By: Claude Opus 5 --- .github/workflows/control-contract.yml | 45 + README.md | 36 +- openadapt_capture/__init__.py | 16 + openadapt_capture/cli.py | 67 ++ openadapt_capture/control.py | 1275 ++++++++++++++++++++++++ openadapt_capture/recorder.py | 393 +++++++- tests/control_recorder_process.py | 71 ++ tests/test_cli_record.py | 40 +- tests/test_control.py | 821 +++++++++++++++ 9 files changed, 2734 insertions(+), 30 deletions(-) create mode 100644 .github/workflows/control-contract.yml create mode 100644 openadapt_capture/control.py create mode 100644 tests/control_recorder_process.py create mode 100644 tests/test_control.py diff --git a/.github/workflows/control-contract.yml b/.github/workflows/control-contract.yml new file mode 100644 index 0000000..0bd6533 --- /dev/null +++ b/.github/workflows/control-contract.yml @@ -0,0 +1,45 @@ +name: Cross-platform control contract + +on: + pull_request: + branches: [main] + paths: + - "openadapt_capture/control.py" + - "openadapt_capture/recorder.py" + - "openadapt_capture/cli.py" + - "tests/test_control.py" + - "tests/control_recorder_process.py" + - ".github/workflows/control-contract.yml" + +concurrency: + group: capture-control-${{ github.ref }} + cancel-in-progress: true + +jobs: + control-contract: + # The normal PR matrix proves Linux. This focused matrix proves the same + # subprocess ready -> status -> stop -> verified-complete contract on the + # two other supported operating systems without running the costly native + # video and input-injection suites on every PR. + strategy: + fail-fast: false + matrix: + os: [macos-latest, windows-latest] + runs-on: ${{ matrix.os }} + timeout-minutes: 10 + steps: + - uses: actions/checkout@v7 + + - name: Install uv + uses: astral-sh/setup-uv@v7 + with: + version: "latest" + + - name: Set up Python + run: uv python install 3.12 + + - name: Install dependencies + run: uv sync --extra dev + + - name: Run the authenticated control contract + run: uv run pytest tests/test_control.py -v --timeout=120 diff --git a/README.md b/README.md index fb2c06c..5072ee0 100644 --- a/README.md +++ b/README.md @@ -119,11 +119,44 @@ Record from the command line: ```bash capture record ./my-capture --description "Describe the workflow" -# Press Ctrl-C to stop. +# The ready message prints the exact session ID. + +# From another terminal: +capture status --session-id SESSION_ID +capture stop --session-id SESSION_ID capture info ./my-capture ``` +The recorder creates its control endpoint before it reports ready. The endpoint +listens on IPv4 loopback only. Each request and response uses an authenticated +session capability. Capture stores that capability only in an owner-only +runtime file (`0700` directory and `0600` file on macOS/Linux; a protected +current-user owner and DACL on Windows). Capture removes and verifies the +absence of macOS extended ACL entries. The capability does not enter command +arguments, logs, or the capture directory. + +`capture stop` binds the request to the exact session ID, process ID, and +process start identity. It waits for producer and writer shutdown, reconciles +producer counts with committed rows, checks database and relationship integrity, +validates replay-relevant events, and writes atomic terminal metadata before it +returns success. Repeated stop requests share one finalization result. A timeout, +worker failure, invalid capability, stale process, or ambiguous set of active +sessions returns non-success. A crash leaves `capture-state.json` incomplete; +the next authenticated discovery removes the stale runtime descriptor only after +it proves that the bound process instance is no longer live. + +Launchers and embedded clients use the public Python contract instead of +reading recorder internals: + +```python +from openadapt_capture import status_recording, stop_recording + +current = status_recording(session_id) +completed = stop_recording(session_id, timeout=60) +assert completed.complete and completed.integrity_verified +``` + Or inspect processed actions in Python: ```python @@ -140,6 +173,7 @@ A capture normally contains: ```text my-capture/ +├── capture-state.json ├── recording.db ├── oa_recording-*.mp4 └── profiling.json diff --git a/openadapt_capture/__init__.py b/openadapt_capture/__init__.py index 97365a2..3a8203d 100644 --- a/openadapt_capture/__init__.py +++ b/openadapt_capture/__init__.py @@ -42,6 +42,15 @@ plot_comparison, ) from openadapt_capture.config import RecordingConfig +from openadapt_capture.control import ( + CaptureControlAuthenticationError, + CaptureControlError, + CaptureControlUnavailable, + RecorderStatus, + discover_recorders, + status_recording, + stop_recording, +) from openadapt_capture.db.models import ( ActionEvent as DBActionEvent, ) @@ -137,6 +146,13 @@ # High-level APIs "Recorder", "RecordingConfig", + "RecorderStatus", + "CaptureControlError", + "CaptureControlUnavailable", + "CaptureControlAuthenticationError", + "discover_recorders", + "status_recording", + "stop_recording", "Capture", "CaptureSession", "Action", diff --git a/openadapt_capture/cli.py b/openadapt_capture/cli.py index 861129b..59d7f1a 100644 --- a/openadapt_capture/cli.py +++ b/openadapt_capture/cli.py @@ -108,6 +108,11 @@ def record( if not recorder.wait_for_ready(): print("Recording did not become ready. No successful capture was saved.") raise SystemExit(1) + print(f"Capture session: {recorder.control_session_id}") + print( + "Stop from another terminal: " + f"capture stop --session-id {recorder.control_session_id}" + ) try: while recorder.is_recording: time.sleep(1) @@ -119,6 +124,66 @@ def record( print(f"Saved to: {output_dir}") +def status( + session_id: str | None = None, + timeout: float = 5.0, + runtime_dir: str | None = None, +) -> None: + """Show the authenticated status of an active recorder. + + Args: + session_id: Exact Capture session ID. It can be omitted only when one + recorder is active. + timeout: Maximum seconds to wait for the recorder. + runtime_dir: Owner-only runtime directory override for an embedded + launcher or a test environment. + """ + import json + + from openadapt_capture.control import CaptureControlError, status_recording + + try: + current = status_recording( + session_id, + timeout=timeout, + runtime_dir=runtime_dir, + ) + except CaptureControlError as exc: + print(str(exc)) + raise SystemExit(1) from exc + print(json.dumps(current.__dict__, sort_keys=True)) + + +def stop( + session_id: str | None = None, + timeout: float = 60.0, + runtime_dir: str | None = None, +) -> None: + """Stop one recorder and confirm its finalized Capture session. + + Args: + session_id: Exact Capture session ID. It can be omitted only when one + recorder is active. + timeout: Maximum seconds to wait for finalization and integrity checks. + runtime_dir: Owner-only runtime directory override for an embedded + launcher or a test environment. + """ + import json + + from openadapt_capture.control import CaptureControlError, stop_recording + + try: + completed = stop_recording( + session_id, + timeout=timeout, + runtime_dir=runtime_dir, + ) + except CaptureControlError as exc: + print(str(exc)) + raise SystemExit(1) from exc + print(json.dumps(completed.__dict__, sort_keys=True)) + + def visualize( capture_dir: str, output: str | None = None, @@ -394,6 +459,8 @@ def main() -> None: import fire fire.Fire({ "record": record, + "status": status, + "stop": stop, "visualize": visualize, "info": info, "transcribe": transcribe, diff --git a/openadapt_capture/control.py b/openadapt_capture/control.py new file mode 100644 index 0000000..0cf84f3 --- /dev/null +++ b/openadapt_capture/control.py @@ -0,0 +1,1275 @@ +"""Authenticated local control for an active Capture recorder. + +The control endpoint is deliberately small. It listens only on the IPv4 +loopback interface, and every request and response is authenticated with a +per-session capability that exists only in an owner-only runtime file. The +capability never appears in command-line arguments, logs, or recording +artifacts. + +``status_recording`` and ``stop_recording`` are the public client contract used +by the OpenAdapt launcher, Flow, and Desktop. They do not inspect Capture's +database or private recorder state. +""" + +from __future__ import annotations + +import ctypes +import errno +import hashlib +import hmac +import json +import os +import secrets +import socket +import socketserver +import stat +import sys +import tempfile +import threading +import time +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable + +import psutil + +CONTROL_SCHEMA_VERSION = "openadapt.capture-control.v1" +TERMINAL_STATE_SCHEMA_VERSION = "openadapt.capture-terminal.v1" +TERMINAL_STATE_FILENAME = "capture-state.json" +_MAX_MESSAGE_BYTES = 64 * 1024 +_REQUEST_CLOCK_SKEW_SECONDS = 60.0 +_DEFAULT_TIMEOUT_SECONDS = 60.0 +_MAX_TIMEOUT_SECONDS = 15 * 60.0 +_MAX_CONTROL_REQUEST_THREADS = 16 + + +class CaptureControlError(RuntimeError): + """The requested recorder control operation did not complete safely.""" + + +class CaptureControlUnavailable(CaptureControlError): + """No unambiguous live recorder session is available.""" + + +class CaptureControlAuthenticationError(CaptureControlError): + """The control peer or runtime descriptor could not be authenticated.""" + + +@dataclass(frozen=True) +class RecorderStatus: + """A privacy-bounded status snapshot returned by the recorder.""" + + session_id: str + pid: int + process_started_at: float + capture_dir: str + phase: str + ready: bool + complete: bool + integrity_verified: bool + event_counts: dict[str, int] + error_code: str | None = None + + @classmethod + def from_payload(cls, payload: dict[str, Any]) -> "RecorderStatus": + try: + counts_raw = payload["event_counts"] + if not isinstance(counts_raw, dict): + raise TypeError("event_counts must be an object") + counts = { + str(name): int(value) + for name, value in counts_raw.items() + if isinstance(value, int) and not isinstance(value, bool) and value >= 0 + } + return cls( + session_id=str(payload["session_id"]), + pid=int(payload["pid"]), + process_started_at=float(payload["process_started_at"]), + capture_dir=str(payload["capture_dir"]), + phase=str(payload["phase"]), + ready=payload["ready"] is True, + complete=payload["complete"] is True, + integrity_verified=payload["integrity_verified"] is True, + event_counts=counts, + error_code=( + str(payload["error_code"]) if payload.get("error_code") is not None else None + ), + ) + except (KeyError, TypeError, ValueError) as exc: + raise CaptureControlAuthenticationError( + "The recorder returned an invalid status payload." + ) from exc + + +@dataclass(frozen=True) +class _ControlDescriptor: + session_id: str + pid: int + process_started_at: float + capture_dir: str + host: str + port: int + created_at: float + path: Path + token: str = field(repr=False) + + def public_fields(self) -> dict[str, Any]: + return { + "schema_version": CONTROL_SCHEMA_VERSION, + "session_id": self.session_id, + "pid": self.pid, + "process_started_at": self.process_started_at, + "capture_dir": self.capture_dir, + "endpoint": {"host": self.host, "port": self.port}, + "created_at": self.created_at, + } + + def serialized(self) -> dict[str, Any]: + fields = self.public_fields() + fields["token"] = self.token + fields["descriptor_mac"] = _descriptor_mac(self.token, fields) + return fields + + +def _canonical_json(value: dict[str, Any]) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + allow_nan=False, + ).encode("utf-8") + + +def _message_mac(token: str, value: dict[str, Any]) -> str: + unsigned = {key: item for key, item in value.items() if key != "mac"} + return hmac.new( + token.encode("ascii"), + _canonical_json(unsigned), + hashlib.sha256, + ).hexdigest() + + +def _descriptor_mac(token: str, value: dict[str, Any]) -> str: + unsigned = {key: item for key, item in value.items() if key != "descriptor_mac"} + return hmac.new( + token.encode("ascii"), + _canonical_json(unsigned), + hashlib.sha256, + ).hexdigest() + + +def _default_runtime_dir() -> Path: + configured = os.environ.get("OPENADAPT_CAPTURE_RUNTIME_DIR") + if configured: + return Path(configured).expanduser() + if sys.platform == "win32": + local_app_data = os.environ.get("LOCALAPPDATA") + if not local_app_data: + raise CaptureControlUnavailable( + "LOCALAPPDATA is unavailable. Capture cannot create an owner-only " + "control directory." + ) + return Path(local_app_data) / "OpenAdapt" / "Capture" / "control" + runtime = os.environ.get("XDG_RUNTIME_DIR") + if runtime: + return Path(runtime) / "openadapt" / "capture" + try: + uid = os.getuid() + except AttributeError as exc: # pragma: no cover - defensive platform guard + raise CaptureControlUnavailable("The operating-system user is unavailable.") from exc + return Path(tempfile.gettempdir()) / f"openadapt-capture-{uid}" + + +def _is_windows_reparse_point(path: Path) -> bool: + if sys.platform != "win32": + return False + get_attributes = ctypes.windll.kernel32.GetFileAttributesW + get_attributes.argtypes = [ctypes.c_wchar_p] + get_attributes.restype = ctypes.c_uint32 + attributes = get_attributes(str(path)) + if attributes == 0xFFFFFFFF: + raise OSError(ctypes.get_last_error(), f"Cannot inspect {path}") + return bool(attributes & 0x400) # FILE_ATTRIBUTE_REPARSE_POINT + + +def _windows_current_user_sid() -> str: + """Return the current process token's SID without invoking a shell.""" + + from ctypes import wintypes + + advapi32 = ctypes.WinDLL("advapi32", use_last_error=True) + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + + class _SidAndAttributes(ctypes.Structure): + _fields_ = [("sid", ctypes.c_void_p), ("attributes", wintypes.DWORD)] + + class _TokenUser(ctypes.Structure): + _fields_ = [("user", _SidAndAttributes)] + + advapi32.OpenProcessToken.argtypes = [ + wintypes.HANDLE, + wintypes.DWORD, + ctypes.POINTER(wintypes.HANDLE), + ] + advapi32.OpenProcessToken.restype = wintypes.BOOL + advapi32.GetTokenInformation.argtypes = [ + wintypes.HANDLE, + ctypes.c_int, + ctypes.c_void_p, + wintypes.DWORD, + ctypes.POINTER(wintypes.DWORD), + ] + advapi32.GetTokenInformation.restype = wintypes.BOOL + advapi32.ConvertSidToStringSidW.argtypes = [ + ctypes.c_void_p, + ctypes.POINTER(wintypes.LPWSTR), + ] + advapi32.ConvertSidToStringSidW.restype = wintypes.BOOL + kernel32.GetCurrentProcess.restype = wintypes.HANDLE + kernel32.CloseHandle.argtypes = [wintypes.HANDLE] + kernel32.CloseHandle.restype = wintypes.BOOL + kernel32.LocalFree.argtypes = [ctypes.c_void_p] + kernel32.LocalFree.restype = ctypes.c_void_p + + token_query = 0x0008 + token_user = 1 + token = wintypes.HANDLE() + if not advapi32.OpenProcessToken( + kernel32.GetCurrentProcess(), token_query, ctypes.byref(token) + ): + raise OSError(ctypes.get_last_error(), "OpenProcessToken failed") + try: + required = wintypes.DWORD() + advapi32.GetTokenInformation(token, token_user, None, 0, ctypes.byref(required)) + if not required.value: + raise OSError(ctypes.get_last_error(), "GetTokenInformation failed") + buffer = ctypes.create_string_buffer(required.value) + if not advapi32.GetTokenInformation( + token, + token_user, + buffer, + required, + ctypes.byref(required), + ): + raise OSError(ctypes.get_last_error(), "GetTokenInformation failed") + sid_pointer = ctypes.cast(buffer, ctypes.POINTER(_TokenUser)).contents.user.sid + if not sid_pointer: + raise OSError("The current process token has no user SID") + sid_string = wintypes.LPWSTR() + if not advapi32.ConvertSidToStringSidW(sid_pointer, ctypes.byref(sid_string)): + raise OSError(ctypes.get_last_error(), "ConvertSidToStringSidW failed") + try: + return sid_string.value + finally: + kernel32.LocalFree(ctypes.cast(sid_string, ctypes.c_void_p)) + finally: + kernel32.CloseHandle(token) + + +def _set_and_verify_windows_owner_acl(path: Path, *, _apply: bool = True) -> None: + """Apply and verify a protected DACL containing only the current user. + + A best-effort ``chmod`` is not an authentication boundary on Windows. This + function uses a protected DACL and refuses the control channel if Windows + cannot establish or verify it. + """ + + from ctypes import wintypes + + class _AclSizeInformation(ctypes.Structure): + _fields_ = [ + ("ace_count", wintypes.DWORD), + ("acl_bytes_in_use", wintypes.DWORD), + ("acl_bytes_free", wintypes.DWORD), + ] + + class _AceHeader(ctypes.Structure): + _fields_ = [ + ("ace_type", wintypes.BYTE), + ("ace_flags", wintypes.BYTE), + ("ace_size", wintypes.WORD), + ] + + class _AccessAllowedAce(ctypes.Structure): + _fields_ = [ + ("header", _AceHeader), + ("mask", wintypes.DWORD), + ("sid_start", wintypes.DWORD), + ] + + advapi32 = ctypes.WinDLL("advapi32", use_last_error=True) + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + advapi32.ConvertStringSecurityDescriptorToSecurityDescriptorW.argtypes = [ + wintypes.LPCWSTR, + wintypes.DWORD, + ctypes.POINTER(ctypes.c_void_p), + ctypes.POINTER(wintypes.ULONG), + ] + advapi32.ConvertStringSecurityDescriptorToSecurityDescriptorW.restype = wintypes.BOOL + advapi32.SetFileSecurityW.argtypes = [ + wintypes.LPCWSTR, + wintypes.DWORD, + ctypes.c_void_p, + ] + advapi32.SetFileSecurityW.restype = wintypes.BOOL + advapi32.GetFileSecurityW.argtypes = [ + wintypes.LPCWSTR, + wintypes.DWORD, + ctypes.c_void_p, + wintypes.DWORD, + ctypes.POINTER(wintypes.DWORD), + ] + advapi32.GetFileSecurityW.restype = wintypes.BOOL + advapi32.ConvertStringSidToSidW.argtypes = [ + wintypes.LPCWSTR, + ctypes.POINTER(ctypes.c_void_p), + ] + advapi32.ConvertStringSidToSidW.restype = wintypes.BOOL + advapi32.GetSecurityDescriptorOwner.argtypes = [ + ctypes.c_void_p, + ctypes.POINTER(ctypes.c_void_p), + ctypes.POINTER(wintypes.BOOL), + ] + advapi32.GetSecurityDescriptorOwner.restype = wintypes.BOOL + advapi32.GetSecurityDescriptorDacl.argtypes = [ + ctypes.c_void_p, + ctypes.POINTER(wintypes.BOOL), + ctypes.POINTER(ctypes.c_void_p), + ctypes.POINTER(wintypes.BOOL), + ] + advapi32.GetSecurityDescriptorDacl.restype = wintypes.BOOL + advapi32.GetSecurityDescriptorControl.argtypes = [ + ctypes.c_void_p, + ctypes.POINTER(ctypes.c_ushort), + ctypes.POINTER(wintypes.DWORD), + ] + advapi32.GetSecurityDescriptorControl.restype = wintypes.BOOL + advapi32.GetAclInformation.argtypes = [ + ctypes.c_void_p, + ctypes.c_void_p, + wintypes.DWORD, + ctypes.c_int, + ] + advapi32.GetAclInformation.restype = wintypes.BOOL + advapi32.GetAce.argtypes = [ + ctypes.c_void_p, + wintypes.DWORD, + ctypes.POINTER(ctypes.c_void_p), + ] + advapi32.GetAce.restype = wintypes.BOOL + advapi32.EqualSid.argtypes = [ctypes.c_void_p, ctypes.c_void_p] + advapi32.EqualSid.restype = wintypes.BOOL + kernel32.LocalFree.argtypes = [ctypes.c_void_p] + kernel32.LocalFree.restype = ctypes.c_void_p + + sid = _windows_current_user_sid() + sddl_revision_1 = 1 + owner_security_information = 0x00000001 + dacl_security_information = 0x00000004 + protected_dacl_security_information = 0x80000000 + if _apply: + security_descriptor = ctypes.c_void_p() + descriptor_size = wintypes.ULONG() + sddl = f"O:{sid}D:P(A;;GA;;;{sid})" + if not advapi32.ConvertStringSecurityDescriptorToSecurityDescriptorW( + sddl, + sddl_revision_1, + ctypes.byref(security_descriptor), + ctypes.byref(descriptor_size), + ): + raise OSError(ctypes.get_last_error(), "Cannot build the owner-only DACL") + try: + if not advapi32.SetFileSecurityW( + str(path), + owner_security_information + | dacl_security_information + | protected_dacl_security_information, + security_descriptor, + ): + raise OSError(ctypes.get_last_error(), f"Cannot protect {path}") + finally: + kernel32.LocalFree(security_descriptor) + + current_sid = ctypes.c_void_p() + if not advapi32.ConvertStringSidToSidW(sid, ctypes.byref(current_sid)): + raise OSError(ctypes.get_last_error(), "Cannot parse the current-user SID") + try: + security_information = owner_security_information | dacl_security_information + needed = wintypes.DWORD() + advapi32.GetFileSecurityW( + str(path), + security_information, + None, + 0, + ctypes.byref(needed), + ) + if not needed.value: + raise OSError(ctypes.get_last_error(), f"Cannot inspect the DACL for {path}") + current = ctypes.create_string_buffer(needed.value) + if not advapi32.GetFileSecurityW( + str(path), + security_information, + current, + needed, + ctypes.byref(needed), + ): + raise OSError(ctypes.get_last_error(), f"Cannot inspect the DACL for {path}") + + owner_sid = ctypes.c_void_p() + owner_defaulted = wintypes.BOOL() + if not advapi32.GetSecurityDescriptorOwner( + current, + ctypes.byref(owner_sid), + ctypes.byref(owner_defaulted), + ) or not owner_sid.value or not advapi32.EqualSid(owner_sid, current_sid): + raise PermissionError(f"The current user does not own {path}") + + dacl_present = wintypes.BOOL() + dacl_defaulted = wintypes.BOOL() + dacl = ctypes.c_void_p() + if not advapi32.GetSecurityDescriptorDacl( + current, + ctypes.byref(dacl_present), + ctypes.byref(dacl), + ctypes.byref(dacl_defaulted), + ): + raise OSError(ctypes.get_last_error(), f"Cannot read the DACL for {path}") + if not dacl_present.value or not dacl.value: + raise PermissionError(f"The DACL for {path} is absent or unrestricted") + + control = ctypes.c_ushort() + revision = wintypes.DWORD() + if not advapi32.GetSecurityDescriptorControl( + current, + ctypes.byref(control), + ctypes.byref(revision), + ): + raise OSError(ctypes.get_last_error(), f"Cannot inspect DACL control for {path}") + if not control.value & 0x1000: # SE_DACL_PROTECTED + raise PermissionError(f"The DACL for {path} permits inherited access") + + acl_size = _AclSizeInformation() + if not advapi32.GetAclInformation( + dacl, + ctypes.byref(acl_size), + ctypes.sizeof(acl_size), + 2, # AclSizeInformation + ): + raise OSError(ctypes.get_last_error(), f"Cannot inspect DACL entries for {path}") + if acl_size.ace_count != 1: + raise PermissionError(f"The DACL for {path} is not current-user-only") + + ace_pointer = ctypes.c_void_p() + if not advapi32.GetAce(dacl, 0, ctypes.byref(ace_pointer)): + raise OSError(ctypes.get_last_error(), f"Cannot inspect the DACL entry for {path}") + ace = ctypes.cast(ace_pointer, ctypes.POINTER(_AccessAllowedAce)).contents + if ace.header.ace_type != 0 or ace.header.ace_flags != 0: # ACCESS_ALLOWED_ACE_TYPE + raise PermissionError(f"The DACL for {path} has an invalid access entry") + if ace.mask not in {0x10000000, 0x001F01FF}: # GENERIC_ALL or FILE_ALL_ACCESS + raise PermissionError(f"The DACL for {path} does not grant exact full access") + ace_sid = ctypes.c_void_p(ace_pointer.value + _AccessAllowedAce.sid_start.offset) + if not advapi32.EqualSid(ace_sid, current_sid): + raise PermissionError(f"The DACL for {path} grants access to another identity") + finally: + kernel32.LocalFree(current_sid) + + +def _macos_extended_acl_present(descriptor: int, path: Path) -> bool: + """Return whether a file descriptor has a valid non-empty macOS ACL.""" + + libc = ctypes.CDLL(None, use_errno=True) + libc.acl_get_fd_np.argtypes = [ctypes.c_int, ctypes.c_int] + libc.acl_get_fd_np.restype = ctypes.c_void_p + libc.acl_get_entry.argtypes = [ + ctypes.c_void_p, + ctypes.c_int, + ctypes.POINTER(ctypes.c_void_p), + ] + libc.acl_get_entry.restype = ctypes.c_int + libc.acl_valid.argtypes = [ctypes.c_void_p] + libc.acl_valid.restype = ctypes.c_int + libc.acl_free.argtypes = [ctypes.c_void_p] + libc.acl_free.restype = ctypes.c_int + + ctypes.set_errno(0) + acl = libc.acl_get_fd_np(descriptor, 0x00000100) # ACL_TYPE_EXTENDED + if not acl: + error = ctypes.get_errno() + if error == errno.ENOENT: + return False + raise OSError(error, f"Cannot inspect the extended ACL for {path}") + try: + if libc.acl_valid(acl) != 0: + raise OSError(ctypes.get_errno(), f"The extended ACL for {path} is invalid") + entry = ctypes.c_void_p() + ctypes.set_errno(0) + result = libc.acl_get_entry(acl, 0, ctypes.byref(entry)) # ACL_FIRST_ENTRY + if result == 0: + return True + error = ctypes.get_errno() + if result == -1 and error == errno.EINVAL: + return False + raise OSError(error, f"Cannot inspect extended ACL entries for {path}") + finally: + libc.acl_free(acl) + + +def _clear_and_verify_macos_acl(descriptor: int, path: Path) -> None: + """Remove all macOS extended ACL entries and verify their absence.""" + + libc = ctypes.CDLL(None, use_errno=True) + libc.acl_init.argtypes = [ctypes.c_int] + libc.acl_init.restype = ctypes.c_void_p + libc.acl_set_fd_np.argtypes = [ctypes.c_int, ctypes.c_void_p, ctypes.c_int] + libc.acl_set_fd_np.restype = ctypes.c_int + libc.acl_free.argtypes = [ctypes.c_void_p] + libc.acl_free.restype = ctypes.c_int + + empty_acl = libc.acl_init(0) + if not empty_acl: + raise OSError(ctypes.get_errno(), f"Cannot allocate an empty ACL for {path}") + try: + if libc.acl_set_fd_np(descriptor, empty_acl, 0x00000100) != 0: + raise OSError(ctypes.get_errno(), f"Cannot clear the extended ACL for {path}") + finally: + libc.acl_free(empty_acl) + if _macos_extended_acl_present(descriptor, path): + raise PermissionError(f"The extended ACL for {path} still grants access") + + +def _protect_path(path: Path, *, directory: bool) -> None: + if sys.platform == "win32": + if _is_windows_reparse_point(path): + raise PermissionError(f"Refusing a reparse-point control path: {path}") + _set_and_verify_windows_owner_acl(path) + return + + mode = 0o700 if directory else 0o600 + flags = os.O_RDONLY + if hasattr(os, "O_CLOEXEC"): + flags |= os.O_CLOEXEC + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + if directory and hasattr(os, "O_DIRECTORY"): + flags |= os.O_DIRECTORY + descriptor = os.open(path, flags) + try: + details = os.fstat(descriptor) + expected_kind = stat.S_ISDIR if directory else stat.S_ISREG + if not expected_kind(details.st_mode): + raise PermissionError(f"The control path has the wrong file type: {path}") + if details.st_uid != os.getuid(): + raise PermissionError(f"The current user does not own {path}") + os.fchmod(descriptor, mode) + if sys.platform == "darwin": + _clear_and_verify_macos_acl(descriptor, path) + protected = os.fstat(descriptor) + if stat.S_IMODE(protected.st_mode) != mode: + raise PermissionError( + f"Owner-only permissions could not be established for {path}" + ) + finally: + os.close(descriptor) + + +def _secure_runtime_dir(runtime_dir: str | os.PathLike[str] | None = None) -> Path: + path = Path(runtime_dir).expanduser() if runtime_dir is not None else _default_runtime_dir() + path = path.absolute() + if path.is_symlink() or (path.exists() and _is_windows_reparse_point(path)): + raise PermissionError(f"Refusing an indirect control directory: {path}") + path.mkdir(mode=0o700, parents=True, exist_ok=True) + _protect_path(path, directory=True) + return path + + +def _write_json_atomic(path: Path, payload: dict[str, Any], *, owner_only: bool) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + if path.is_symlink() or (path.exists() and not path.is_file()): + raise PermissionError(f"Refusing an unsafe metadata path: {path}") + temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + descriptor = os.open(temporary, flags, 0o600) + try: + encoded = _canonical_json(payload) + b"\n" + stream = os.fdopen(descriptor, "wb") + descriptor = -1 + with stream: + stream.write(encoded) + stream.flush() + os.fsync(stream.fileno()) + if owner_only: + _protect_path(temporary, directory=False) + os.replace(temporary, path) + if owner_only: + _protect_path(path, directory=False) + finally: + if descriptor >= 0: + try: + os.close(descriptor) + except OSError: + pass + try: + temporary.unlink() + except FileNotFoundError: + pass + + +def write_terminal_state(capture_dir: str | os.PathLike[str], payload: dict[str, Any]) -> Path: + """Atomically persist non-secret terminal metadata in the capture.""" + + path = Path(capture_dir) / TERMINAL_STATE_FILENAME + _write_json_atomic(path, payload, owner_only=True) + return path + + +def _read_json_bounded(path: Path, *, require_owner_only: bool) -> dict[str, Any]: + if require_owner_only: + _protect_path(path, directory=False) + flags = os.O_RDONLY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + descriptor = os.open(path, flags) + try: + details = os.fstat(descriptor) + if details.st_size > _MAX_MESSAGE_BYTES: + raise CaptureControlAuthenticationError("The control descriptor is too large.") + raw = os.read(descriptor, _MAX_MESSAGE_BYTES + 1) + finally: + os.close(descriptor) + if len(raw) > _MAX_MESSAGE_BYTES: + raise CaptureControlAuthenticationError("The control descriptor is too large.") + try: + payload = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise CaptureControlAuthenticationError( + "The control descriptor is not valid JSON." + ) from exc + if not isinstance(payload, dict): + raise CaptureControlAuthenticationError("The control descriptor is invalid.") + return payload + + +def _parse_descriptor(path: Path) -> _ControlDescriptor: + payload = _read_json_bounded(path, require_owner_only=True) + try: + token = payload["token"] + descriptor_mac = payload["descriptor_mac"] + endpoint = payload["endpoint"] + if not isinstance(token, str) or len(token) < 48: + raise ValueError("invalid capability") + if not isinstance(descriptor_mac, str) or not isinstance(endpoint, dict): + raise ValueError("invalid authentication fields") + if payload["schema_version"] != CONTROL_SCHEMA_VERSION: + raise ValueError("unsupported schema") + if not hmac.compare_digest(descriptor_mac, _descriptor_mac(token, payload)): + raise ValueError("descriptor authentication failed") + session_id = str(uuid.UUID(str(payload["session_id"]))) + if path.name != f"{session_id}.json": + raise ValueError("descriptor filename does not match the session") + host = str(endpoint["host"]) + if host != "127.0.0.1": + raise ValueError("non-loopback endpoint") + port = int(endpoint["port"]) + if not 1 <= port <= 65535: + raise ValueError("invalid port") + return _ControlDescriptor( + session_id=session_id, + pid=int(payload["pid"]), + process_started_at=float(payload["process_started_at"]), + capture_dir=str(payload["capture_dir"]), + host=host, + port=port, + created_at=float(payload["created_at"]), + path=path, + token=token, + ) + except (KeyError, TypeError, ValueError) as exc: + raise CaptureControlAuthenticationError( + f"The control descriptor {path.name!r} failed authentication." + ) from exc + + +def _windows_process_live(pid: int, *, _kernel32: Any | None = None) -> bool | None: + """Return exact Windows process signal state, or ``None`` if unknown.""" + + from ctypes import wintypes + + kernel32 = _kernel32 + if kernel32 is None: + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + kernel32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] + kernel32.OpenProcess.restype = wintypes.HANDLE + kernel32.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD] + kernel32.WaitForSingleObject.restype = wintypes.DWORD + kernel32.CloseHandle.argtypes = [wintypes.HANDLE] + kernel32.CloseHandle.restype = wintypes.BOOL + + synchronize = 0x00100000 + wait_object_0 = 0x00000000 + wait_timeout = 0x00000102 + error_invalid_parameter = 87 + + handle = kernel32.OpenProcess( + synchronize, + False, + pid, + ) + if not handle: + error = ctypes.get_last_error() + return False if error == error_invalid_parameter else None + try: + result = int(kernel32.WaitForSingleObject(handle, 0)) + if result == wait_object_0: + return False + if result == wait_timeout: + return True + return None + finally: + kernel32.CloseHandle(handle) + + +def _process_instance_live(pid: int, process_started_at: float) -> bool: + if pid <= 0: + return False + windows_live: bool | None = None + if sys.platform == "win32": + try: + windows_live = _windows_process_live(pid) + except OSError: + windows_live = None + if windows_live is False: + # A signaled process object proves that no recorder is live at PID. + # This check must precede psutil because Windows can retain the dead + # object while denying later metadata queries for it. + return False + try: + process = psutil.Process(pid) + actual = process.create_time() + if actual != process_started_at: + return False + if sys.platform == "win32": + # A live or unknown kernel result plus the exact creation identity is + # conservatively live. Only a signaled object authorizes cleanup. + return True + if not process.is_running(): + return False + if process.status() in {psutil.STATUS_DEAD, psutil.STATUS_ZOMBIE}: + return False + try: + process.wait(timeout=0) + except psutil.TimeoutExpired: + return True + return False + except (psutil.NoSuchProcess, psutil.ZombieProcess): + return False + except (psutil.AccessDenied, OSError): + # An uninspectable process is not proof of a stale endpoint. + return True + + +def _mark_crashed_if_bound(descriptor: _ControlDescriptor) -> None: + path = Path(descriptor.capture_dir) / TERMINAL_STATE_FILENAME + try: + payload = _read_json_bounded(path, require_owner_only=True) + except (FileNotFoundError, OSError, CaptureControlError): + return + if ( + payload.get("schema_version") != TERMINAL_STATE_SCHEMA_VERSION + or payload.get("session_id") != descriptor.session_id + or payload.get("pid") != descriptor.pid + or payload.get("process_started_at") != descriptor.process_started_at + or payload.get("complete") is True + ): + return + payload.update( + { + "phase": "crashed", + "complete": False, + "integrity_verified": False, + "error_code": "recorder_process_exited", + "finalized_at": time.time(), + } + ) + try: + write_terminal_state(descriptor.capture_dir, payload) + except OSError: + return + + +def _remove_descriptor_if_exact(descriptor: _ControlDescriptor) -> bool: + try: + current = _parse_descriptor(descriptor.path) + except (FileNotFoundError, CaptureControlError, OSError): + return False + if ( + current.session_id != descriptor.session_id + or current.pid != descriptor.pid + or current.process_started_at != descriptor.process_started_at + or current.host != descriptor.host + or current.port != descriptor.port + or not hmac.compare_digest(current.token, descriptor.token) + ): + return False + try: + descriptor.path.unlink() + return True + except FileNotFoundError: + return False + + +def discover_recorders( + runtime_dir: str | os.PathLike[str] | None = None, +) -> list[str]: + """Return live session IDs and remove only proven-stale descriptors.""" + + root = _secure_runtime_dir(runtime_dir) + live: list[str] = [] + for path in sorted(root.glob("*.json")): + try: + descriptor = _parse_descriptor(path) + except CaptureControlAuthenticationError: + # An unauthenticated file cannot authorize deletion or discovery. + continue + if _process_instance_live(descriptor.pid, descriptor.process_started_at): + live.append(descriptor.session_id) + continue + _mark_crashed_if_bound(descriptor) + _remove_descriptor_if_exact(descriptor) + return live + + +def _select_descriptor( + session_id: str | None, + runtime_dir: str | os.PathLike[str] | None, +) -> _ControlDescriptor: + root = _secure_runtime_dir(runtime_dir) + descriptors: list[_ControlDescriptor] = [] + for path in sorted(root.glob("*.json")): + try: + candidate = _parse_descriptor(path) + except CaptureControlAuthenticationError: + continue + if not _process_instance_live(candidate.pid, candidate.process_started_at): + _mark_crashed_if_bound(candidate) + _remove_descriptor_if_exact(candidate) + continue + descriptors.append(candidate) + + if session_id is not None: + try: + normalized = str(uuid.UUID(session_id)) + except ValueError as exc: + raise CaptureControlUnavailable("The Capture session ID is invalid.") from exc + matches = [item for item in descriptors if item.session_id == normalized] + if len(matches) != 1: + raise CaptureControlUnavailable( + f"No live Capture recorder has session ID {normalized}." + ) + return matches[0] + if not descriptors: + raise CaptureControlUnavailable("No live Capture recorder was found.") + if len(descriptors) > 1: + ids = ", ".join(item.session_id for item in descriptors) + raise CaptureControlUnavailable( + f"More than one Capture recorder is active. Select a session ID: {ids}" + ) + return descriptors[0] + + +def _recv_line(connection: socket.socket) -> bytes: + chunks = bytearray() + while len(chunks) <= _MAX_MESSAGE_BYTES: + block = connection.recv(min(4096, _MAX_MESSAGE_BYTES + 1 - len(chunks))) + if not block: + break + chunks.extend(block) + if b"\n" in block: + break + if len(chunks) > _MAX_MESSAGE_BYTES: + raise CaptureControlAuthenticationError("The control message is too large.") + line, separator, remainder = bytes(chunks).partition(b"\n") + if not separator or remainder: + raise CaptureControlAuthenticationError("The control message framing is invalid.") + return line + + +def _request( + descriptor: _ControlDescriptor, + command: str, + *, + timeout: float, +) -> RecorderStatus: + timeout = float(timeout) + if not 0 < timeout <= _MAX_TIMEOUT_SECONDS: + raise ValueError(f"timeout must be between 0 and {_MAX_TIMEOUT_SECONDS} seconds") + request_id = str(uuid.uuid4()) + request: dict[str, Any] = { + "schema_version": CONTROL_SCHEMA_VERSION, + "command": command, + "session_id": descriptor.session_id, + "pid": descriptor.pid, + "process_started_at": descriptor.process_started_at, + "request_id": request_id, + "issued_at": time.time(), + "timeout_seconds": timeout, + } + request["mac"] = _message_mac(descriptor.token, request) + try: + with socket.create_connection( + (descriptor.host, descriptor.port), timeout=min(timeout + 2.0, _MAX_TIMEOUT_SECONDS) + ) as connection: + connection.settimeout(min(timeout + 2.0, _MAX_TIMEOUT_SECONDS)) + connection.sendall(_canonical_json(request) + b"\n") + raw = _recv_line(connection) + except (OSError, TimeoutError) as exc: + if not _process_instance_live(descriptor.pid, descriptor.process_started_at): + _mark_crashed_if_bound(descriptor) + _remove_descriptor_if_exact(descriptor) + raise CaptureControlUnavailable( + "The Capture recorder control endpoint did not respond." + ) from exc + try: + response = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise CaptureControlAuthenticationError( + "The recorder returned an invalid control response." + ) from exc + if not isinstance(response, dict): + raise CaptureControlAuthenticationError("The control response is invalid.") + response_mac = response.get("mac") + if not isinstance(response_mac, str) or not hmac.compare_digest( + response_mac, _message_mac(descriptor.token, response) + ): + raise CaptureControlAuthenticationError( + "The recorder control response failed authentication." + ) + if ( + response.get("schema_version") != CONTROL_SCHEMA_VERSION + or response.get("request_id") != request_id + or response.get("session_id") != descriptor.session_id + or response.get("pid") != descriptor.pid + or response.get("process_started_at") != descriptor.process_started_at + ): + raise CaptureControlAuthenticationError( + "The recorder response does not match the requested process instance." + ) + if response.get("ok") is not True: + error_code = str(response.get("error_code") or "control_request_failed") + raise CaptureControlError(f"Capture control failed: {error_code}") + status = RecorderStatus.from_payload(response) + if status.session_id != descriptor.session_id: + raise CaptureControlAuthenticationError( + "The recorder status belongs to a different session." + ) + return status + + +def status_recording( + session_id: str | None = None, + *, + runtime_dir: str | os.PathLike[str] | None = None, + timeout: float = 5.0, +) -> RecorderStatus: + """Return the status of one exact active Capture session.""" + + descriptor = _select_descriptor(session_id, runtime_dir) + return _request(descriptor, "status", timeout=timeout) + + +def stop_recording( + session_id: str | None = None, + *, + runtime_dir: str | os.PathLike[str] | None = None, + timeout: float = _DEFAULT_TIMEOUT_SECONDS, +) -> RecorderStatus: + """Stop one exact recorder and return only after verified finalization. + + The stop operation is idempotent inside the recorder. Concurrent or + repeated requests share the same termination event and finalization result. + A timeout or failed integrity check raises ``CaptureControlError`` and never + returns a success-shaped status. + """ + + descriptor = _select_descriptor(session_id, runtime_dir) + status = _request(descriptor, "stop", timeout=timeout) + if not status.complete or not status.integrity_verified or status.phase != "complete": + raise CaptureControlError( + f"Capture stop did not produce a verified complete session ({status.phase})." + ) + return status + + +class _LoopbackServer(socketserver.ThreadingTCPServer): + allow_reuse_address = False + daemon_threads = False + block_on_close = True + + def __init__(self, *args: Any, **kwargs: Any) -> None: + self._request_slots = threading.BoundedSemaphore(_MAX_CONTROL_REQUEST_THREADS) + super().__init__(*args, **kwargs) + + def verify_request(self, request: socket.socket, client_address: tuple[str, int]) -> bool: + return client_address[0] == "127.0.0.1" + + def process_request(self, request: socket.socket, client_address: tuple[str, int]) -> None: + if not self._request_slots.acquire(blocking=False): + request.close() + return + try: + super().process_request(request, client_address) + except BaseException: + self._request_slots.release() + raise + + def process_request_thread( + self, + request: socket.socket, + client_address: tuple[str, int], + ) -> None: + try: + super().process_request_thread(request, client_address) + finally: + self._request_slots.release() + + +class _ControlRequestHandler(socketserver.BaseRequestHandler): + def handle(self) -> None: + owner: RecorderControlServer = self.server.owner # type: ignore[attr-defined] + owner._handle(self.request) + + +class RecorderControlServer: + """Recorder-owned loopback server. This is not the public client API.""" + + def __init__( + self, + *, + capture_dir: str, + snapshot: Callable[[], dict[str, Any]], + stop: Callable[[float], dict[str, Any]], + session_id: str | None = None, + runtime_dir: str | os.PathLike[str] | None = None, + ) -> None: + self.capture_dir = str(Path(capture_dir).resolve()) + self.session_id = str(uuid.UUID(session_id)) if session_id else str(uuid.uuid4()) + self.pid = os.getpid() + self.process_started_at = psutil.Process(self.pid).create_time() + self._snapshot = snapshot + self._stop = stop + self._runtime_dir_arg = runtime_dir + self._token = secrets.token_urlsafe(48) + self._server: _LoopbackServer | None = None + self._thread: threading.Thread | None = None + self._descriptor: _ControlDescriptor | None = None + self._closed = False + self._close_lock = threading.Lock() + self._seen_requests: dict[str, float] = {} + self._seen_lock = threading.Lock() + + @property + def descriptor_path(self) -> Path | None: + return self._descriptor.path if self._descriptor is not None else None + + def start(self) -> "RecorderControlServer": + runtime_dir = _secure_runtime_dir(self._runtime_dir_arg) + server = _LoopbackServer(("127.0.0.1", 0), _ControlRequestHandler) + server.owner = self # type: ignore[attr-defined] + host, port = server.server_address + descriptor = _ControlDescriptor( + session_id=self.session_id, + pid=self.pid, + process_started_at=self.process_started_at, + capture_dir=self.capture_dir, + host=str(host), + port=int(port), + created_at=time.time(), + path=runtime_dir / f"{self.session_id}.json", + token=self._token, + ) + try: + _write_json_atomic(descriptor.path, descriptor.serialized(), owner_only=True) + except BaseException: + server.server_close() + raise + self._server = server + self._descriptor = descriptor + self._thread = threading.Thread( + target=server.serve_forever, + name=f"capture-control-{self.session_id[:8]}", + daemon=True, + ) + try: + self._thread.start() + except BaseException: + _remove_descriptor_if_exact(descriptor) + server.server_close() + self._server = None + self._descriptor = None + self._thread = None + self._closed = True + self._token = "" + raise + return self + + def _authenticate_request(self, request: dict[str, Any]) -> tuple[str, float]: + supplied_mac = request.get("mac") + if not isinstance(supplied_mac, str) or not hmac.compare_digest( + supplied_mac, _message_mac(self._token, request) + ): + raise CaptureControlAuthenticationError("authentication_failed") + try: + issued_at = float(request["issued_at"]) + request_id = str(uuid.UUID(str(request["request_id"]))) + timeout = float(request["timeout_seconds"]) + except (KeyError, TypeError, ValueError) as exc: + raise CaptureControlAuthenticationError("invalid_request") from exc + if abs(time.time() - issued_at) > _REQUEST_CLOCK_SKEW_SECONDS: + raise CaptureControlAuthenticationError("expired_request") + if not 0 < timeout <= _MAX_TIMEOUT_SECONDS: + raise CaptureControlAuthenticationError("invalid_timeout") + if ( + request.get("schema_version") != CONTROL_SCHEMA_VERSION + or request.get("session_id") != self.session_id + or request.get("pid") != self.pid + or request.get("process_started_at") != self.process_started_at + ): + raise CaptureControlAuthenticationError("recorder_instance_mismatch") + with self._seen_lock: + if request_id in self._seen_requests: + raise CaptureControlAuthenticationError("replayed_request") + oldest_valid = time.time() - _REQUEST_CLOCK_SKEW_SECONDS + self._seen_requests = { + seen_id: seen_at + for seen_id, seen_at in self._seen_requests.items() + if seen_at >= oldest_valid + } + if len(self._seen_requests) >= 4096: + # Never clear still-valid request IDs. Saturation must fail + # closed instead of reopening the replay window. + raise CaptureControlAuthenticationError("request_limit") + self._seen_requests[request_id] = issued_at + return request_id, timeout + + def _response( + self, + request_id: str, + *, + ok: bool, + status: dict[str, Any] | None = None, + error_code: str | None = None, + ) -> dict[str, Any]: + response: dict[str, Any] = { + "schema_version": CONTROL_SCHEMA_VERSION, + "request_id": request_id, + "session_id": self.session_id, + "pid": self.pid, + "process_started_at": self.process_started_at, + "ok": ok, + } + if status: + for key in ( + "capture_dir", + "phase", + "ready", + "complete", + "integrity_verified", + "event_counts", + "error_code", + ): + if key in status: + response[key] = status[key] + if error_code: + response["error_code"] = error_code + response["mac"] = _message_mac(self._token, response) + return response + + def _handle(self, connection: socket.socket) -> None: + connection.settimeout(5.0) + request_id = str(uuid.uuid4()) + try: + raw = _recv_line(connection) + request = json.loads(raw.decode("utf-8")) + if not isinstance(request, dict): + raise CaptureControlAuthenticationError("invalid_request") + request_id, timeout = self._authenticate_request(request) + command = request.get("command") + if command == "status": + status = self._snapshot() + elif command == "stop": + status = self._stop(timeout) + else: + raise CaptureControlAuthenticationError("unsupported_command") + complete = status.get("complete") is True + verified = status.get("integrity_verified") is True + clean_completion = ( + complete + and verified + and status.get("phase") == "complete" + and status.get("error_code") is None + ) + if command == "stop" and not clean_completion: + response = self._response( + request_id, + ok=False, + status=status, + error_code=str(status.get("error_code") or "finalization_incomplete"), + ) + else: + response = self._response(request_id, ok=True, status=status) + except CaptureControlAuthenticationError: + # Do not reveal whether a token, session, or process field was wrong. + response = self._response( + request_id, + ok=False, + error_code="authentication_failed", + ) + except (UnicodeDecodeError, json.JSONDecodeError, OSError, ValueError): + response = self._response( + request_id, + ok=False, + error_code="invalid_request", + ) + try: + connection.sendall(_canonical_json(response) + b"\n") + except OSError: + pass + + def close(self) -> None: + with self._close_lock: + if self._closed: + return + self._closed = True + server = self._server + descriptor = self._descriptor + if server is not None: + server.shutdown() + server.server_close() + if self._thread is not None and self._thread is not threading.current_thread(): + self._thread.join(timeout=5.0) + if descriptor is not None: + _remove_descriptor_if_exact(descriptor) + self._token = "" + + def __enter__(self) -> "RecorderControlServer": + return self.start() + + def __exit__(self, exc_type, exc_value, traceback) -> None: + self.close() + + +__all__ = [ + "CONTROL_SCHEMA_VERSION", + "TERMINAL_STATE_FILENAME", + "TERMINAL_STATE_SCHEMA_VERSION", + "CaptureControlAuthenticationError", + "CaptureControlError", + "CaptureControlUnavailable", + "RecorderStatus", + "discover_recorders", + "status_recording", + "stop_recording", +] diff --git a/openadapt_capture/recorder.py b/openadapt_capture/recorder.py index 02ad1eb..ead42c3 100644 --- a/openadapt_capture/recorder.py +++ b/openadapt_capture/recorder.py @@ -24,10 +24,12 @@ import os import queue import signal +import sqlite3 import sys import threading import time import tracemalloc +import uuid from collections import namedtuple from functools import partial from typing import Any, Callable @@ -430,6 +432,7 @@ def process_events( recording, perf_q, ) + num_browser_events.value += 1 elif event.type == "action": if prev_screen_event is None: logger.warning("Discarding action that came before screen") @@ -1712,6 +1715,7 @@ def record( window_write_q = sq.SynchronizedQueue() browser_write_q = sq.SynchronizedQueue() video_write_q = sq.SynchronizedQueue() + terminate_writers = multiprocessing.Event() # TODO: save write times to DB; display performance plot in visualize.py perf_q = sq.SynchronizedQueue() if terminate_processing is None: @@ -1799,25 +1803,32 @@ def record( if num_video_events is None: num_video_events = multiprocessing.Value("i", 0) + event_processor_args = ( + event_q, + screen_write_q, + action_write_q, + window_write_q, + browser_write_q, + video_write_q, + perf_q, + recording, + terminate_processing, + task_started_events.setdefault("event_processor", threading.Event()), + num_screen_events, + num_action_events, + num_window_events, + num_browser_events, + num_video_events, + ) event_processor = threading.Thread( - target=process_events, + target=_run_task_fail_loud, daemon=True, args=( - event_q, - screen_write_q, - action_write_q, - window_write_q, - browser_write_q, - video_write_q, - perf_q, - recording, + "event_processor", + process_events, + event_processor_args, terminate_processing, - task_started_events.setdefault("event_processor", threading.Event()), - num_screen_events, - num_action_events, - num_window_events, - num_browser_events, - num_video_events, + task_errors, ), ) event_processor.start() @@ -1833,7 +1844,7 @@ def record( perf_q, recording, db_path, - terminate_processing, + terminate_writers, task_started_events.setdefault("screen_event_writer", multiprocessing.Event()), ), ) @@ -1850,7 +1861,7 @@ def record( perf_q, recording, db_path, - terminate_processing, + terminate_writers, task_started_events.setdefault("action_event_writer", multiprocessing.Event()), ), ) @@ -1868,7 +1879,7 @@ def record( perf_q, recording, db_path, - terminate_processing, + terminate_writers, task_started_events.setdefault("window_event_writer", multiprocessing.Event()), ), ) @@ -1886,7 +1897,7 @@ def record( perf_q, recording, db_path, - terminate_processing, + terminate_writers, task_started_events.setdefault("video_writer", multiprocessing.Event()), partial( video_pre_callback, @@ -1994,11 +2005,21 @@ def record( "screen_event_reader", "input_event_reader", "event_processor", + "audio_recorder", + ], + timeout=pre_ready_timeout, + ) + + # No writer can stop while the event processor can still enqueue work. + # Signal writer completion only after all producers have exited. + terminate_writers.set() + _join_tasks( + task_by_name, + [ "screen_event_writer", "action_event_writer", "window_event_writer", "video_writer", - "audio_recorder", ], timeout=pre_ready_timeout, ) @@ -2169,6 +2190,8 @@ def __init__( send_profile: bool = False, window: dict | None = None, structural_observer: StructuralObserver | None = None, + control_enabled: bool = True, + control_runtime_dir: str | None = None, ) -> None: from pathlib import Path @@ -2178,6 +2201,8 @@ def __init__( self.capture_dir = str(Path(capture_dir).resolve()) self.task_description = task_description self._send_profile = send_profile + self._control_enabled = control_enabled + self._control_runtime_dir = control_runtime_dir if capture_browser_events: # Preserve a clear error for callers of the former keyword while @@ -2224,6 +2249,7 @@ def __init__( self._ready_event = threading.Event() self._stopped_event = threading.Event() self._ready_or_stopped_event = threading.Event() + self._finalized_event = threading.Event() # Internal self._record_thread: threading.Thread | None = None @@ -2232,6 +2258,229 @@ def __init__( self._worker_error: BaseException | None = None self._worker_error_lock = threading.Lock() self._structural_observer = structural_observer + self._control_server = None + self._control_state_lock = threading.RLock() + self._control_stop_lock = threading.Lock() + self._control_session_id = str(uuid.uuid4()) + self._process_started_at = psutil.Process(os.getpid()).create_time() + self._control_phase = "starting" + self._control_complete = False + self._control_integrity_verified = False + self._control_error_code: str | None = None + self._control_started_at = time.time() + self._control_finalized_at: float | None = None + + def _control_payload(self) -> dict[str, Any]: + """Return the secret-free state shared with the client and state file.""" + with self._control_state_lock: + return { + "schema_version": "openadapt.capture-terminal.v1", + "session_id": self._control_session_id, + "pid": os.getpid(), + "process_started_at": self._process_started_at, + "capture_dir": self.capture_dir, + "phase": self._control_phase, + "ready": self._ready_event.is_set(), + "complete": self._control_complete, + "integrity_verified": self._control_integrity_verified, + "error_code": self._control_error_code, + "started_at": self._control_started_at, + "finalized_at": self._control_finalized_at, + "event_counts": { + "action": self._num_action_events.value, + "screen": self._num_screen_events.value, + "window": self._num_window_events.value, + "browser": self._num_browser_events.value, + "video": self._num_video_events.value, + }, + } + + def _persist_control_state(self) -> None: + if not self._control_enabled: + return + from openadapt_capture.control import write_terminal_state + + terminal = self._control_payload() + # The file location already binds the capture. Do not retain an + # absolute local path (which can disclose a user/profile name) in an + # artifact that may later enter sanitization and review. + terminal.pop("capture_dir", None) + write_terminal_state(self.capture_dir, terminal) + + def _transition_control( + self, + phase: str, + *, + complete: bool = False, + integrity_verified: bool = False, + error_code: str | None = None, + finalized: bool = False, + ) -> None: + with self._control_state_lock: + if self._control_phase in {"complete", "failed", "crashed"}: + return + previous = ( + self._control_phase, + self._control_complete, + self._control_integrity_verified, + self._control_error_code, + self._control_finalized_at, + ) + if ( + self._control_error_code == "finalization_timeout" + and error_code is None + and phase not in {"complete", "failed", "crashed"} + ): + error_code = self._control_error_code + self._control_phase = phase + self._control_complete = complete + self._control_integrity_verified = integrity_verified + self._control_error_code = error_code + if finalized: + self._control_finalized_at = time.time() + try: + self._persist_control_state() + except BaseException: + ( + self._control_phase, + self._control_complete, + self._control_integrity_verified, + self._control_error_code, + self._control_finalized_at, + ) = previous + raise + + def _set_worker_error(self, exc: BaseException) -> None: + with self._worker_error_lock: + if self._worker_error is None: + self._worker_error = exc + + def _verify_completed_capture(self) -> None: + """Verify the finalized database before control reports completion.""" + from pathlib import Path + + from openadapt_capture.capture import ( + CaptureSession, + _convert_action_event, + _convert_browser_event, + ) + + db_path = Path(self.capture_dir) / "recording.db" + details = db_path.lstat() + if not db_path.is_file() or db_path.is_symlink(): + raise RuntimeError("The finalized Capture database is not a regular file.") + if details.st_size <= 0: + raise RuntimeError("The finalized Capture database is empty.") + database = sqlite3.connect(f"{db_path.resolve().as_uri()}?mode=ro", uri=True) + try: + quick_check = database.execute("PRAGMA quick_check").fetchall() + if quick_check != [("ok",)]: + raise RuntimeError("The finalized Capture database failed integrity check.") + foreign_key_errors = database.execute("PRAGMA foreign_key_check").fetchall() + if foreign_key_errors: + raise RuntimeError("The finalized Capture database has broken relationships.") + recordings = database.execute("SELECT id FROM recording").fetchall() + if len(recordings) != 1: + raise RuntimeError( + "The finalized Capture database does not contain one session." + ) + recording_id = recordings[0][0] + expected_counts = { + "action_event": self._num_action_events.value, + "screenshot": self._num_screen_events.value, + "window_event": self._num_window_events.value, + "browser_event": self._num_browser_events.value, + } + for table, expected in expected_counts.items(): + committed = database.execute( + f"SELECT COUNT(*) FROM {table}", + ).fetchone()[0] + if committed != expected: + raise RuntimeError( + f"The finalized Capture database lost {table} events " + f"(expected {expected}, committed {committed})." + ) + wrong_recording = database.execute( + f"SELECT COUNT(*) FROM {table} " + "WHERE recording_id IS NULL OR recording_id != ?", + (recording_id,), + ).fetchone()[0] + if wrong_recording: + raise RuntimeError( + f"The finalized Capture database has unbound {table} events." + ) + actions_without_screenshots = database.execute( + "SELECT COUNT(*) FROM action_event " + "WHERE recording_id = ? AND screenshot_id IS NULL", + (recording_id,), + ).fetchone()[0] + if actions_without_screenshots: + raise RuntimeError( + "The finalized Capture database has actions without screenshots." + ) + finally: + database.close() + capture = CaptureSession.load(self.capture_dir) + try: + for event in capture._recording.action_events: + _convert_action_event(event) + for event in capture._recording.browser_events: + if _convert_browser_event(event) is None: + raise RuntimeError( + "The finalized Capture database has an invalid browser event." + ) + finally: + capture.close() + + def _start_control_server(self) -> None: + from pathlib import Path + + from openadapt_capture.control import RecorderControlServer + + Path(self.capture_dir).mkdir(parents=True, exist_ok=True) + self._persist_control_state() + server = RecorderControlServer( + capture_dir=self.capture_dir, + snapshot=self._control_payload, + stop=self._control_stop, + session_id=self._control_session_id, + runtime_dir=self._control_runtime_dir, + ) + self._control_server = server.start() + + def _control_stop(self, timeout: float) -> dict[str, Any]: + """Idempotently request stop and wait for the one finalization result.""" + with self._control_stop_lock: + if ( + not self._finalized_event.is_set() + and not self._terminate_processing.is_set() + ): + try: + self._transition_control("stopping") + except BaseException as exc: + self._set_worker_error(exc) + self._terminate_processing.set() + return { + **self._control_payload(), + "error_code": "terminal_metadata_failed", + } + self._terminate_processing.set() + if not self._finalized_event.wait(timeout=timeout): + try: + self._transition_control( + "finalizing", + error_code="finalization_timeout", + ) + except BaseException as exc: + self._set_worker_error(exc) + return { + **self._control_payload(), + "phase": "finalizing", + "complete": False, + "integrity_verified": False, + "error_code": "finalization_timeout", + } + return self._control_payload() def _drain_status_pipe(self) -> None: """Background thread that reads status messages from record().""" @@ -2243,11 +2492,18 @@ def _drain_status_pipe(self) -> None: if msg.get("type") == "record.started": self._ready_event.set() self._ready_or_stopped_event.set() + self._transition_control("recording") + elif msg.get("type") == "record.stopping": + self._transition_control("finalizing") elif msg.get("type") == "record.stopped": self._stopped_event.set() self._ready_or_stopped_event.set() except (EOFError, OSError): pass + except BaseException as exc: + self._set_worker_error(exc) + self._terminate_processing.set() + self._ready_or_stopped_event.set() def _run_record(self) -> None: """Thread target: apply config overrides, then call record().""" @@ -2269,39 +2525,106 @@ def _run_record(self) -> None: send_profile=self._send_profile, structural_observer=self._structural_observer, ) + self.check_health() + if self._ready_event.is_set(): + self._verify_completed_capture() + self._transition_control( + "complete", + complete=True, + integrity_verified=True, + finalized=True, + ) + else: + self._transition_control( + "failed", + error_code="startup_incomplete", + finalized=True, + ) except BaseException as exc: # A setup exception must wake wait_for_ready() and let context-manager # teardown finish instead of leaving callers blocked for its timeout. - with self._worker_error_lock: - if self._worker_error is None: - self._worker_error = exc + self._set_worker_error(exc) self._terminate_processing.set() + try: + self._transition_control( + "failed", + error_code="recording_or_finalization_failed", + finalized=True, + ) + except BaseException as state_exc: + add_exception_note( + exc, + f"terminal state persistence also failed: {state_exc!r}", + ) try: self._status_send.send({"type": "record.stopped"}) except (BrokenPipeError, EOFError, OSError): self._stopped_event.set() self._ready_or_stopped_event.set() + finally: + self._stopped_event.set() + self._ready_or_stopped_event.set() + self._finalized_event.set() def __enter__(self) -> "Recorder": + if self._control_enabled: + try: + self._start_control_server() + except BaseException: + try: + self._transition_control( + "failed", + error_code="control_channel_failed", + finalized=True, + ) + except BaseException: + pass + raise + # Start status drain thread self._status_thread = threading.Thread( target=self._drain_status_pipe, daemon=True, ) - self._status_thread.start() - - # Start recording thread self._record_thread = threading.Thread(target=self._run_record) - self._record_thread.start() + try: + self._status_thread.start() + self._record_thread.start() + except BaseException as exc: + self._terminate_processing.set() + self._stopped_event.set() + self._ready_or_stopped_event.set() + self._finalized_event.set() + try: + self._transition_control( + "failed", + error_code="recorder_thread_start_failed", + finalized=True, + ) + except BaseException as state_exc: + add_exception_note( + exc, + f"terminal state persistence also failed: {state_exc!r}", + ) + if self._control_server is not None: + self._control_server.close() + raise return self def __exit__(self, exc_type, exc_val, exc_tb) -> None: + if not self._finalized_event.is_set(): + try: + self._transition_control("stopping") + except BaseException as exc: + self._set_worker_error(exc) self._terminate_processing.set() if self._record_thread is not None: self._record_thread.join() self._stopped_event.set() # ensure status thread exits if self._status_thread is not None: self._status_thread.join(timeout=5) + if self._control_server is not None: + self._control_server.close() if self._worker_error is not None: if exc_val is not None: add_exception_note( @@ -2312,10 +2635,19 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> None: def stop(self) -> None: """Stop, join, and surface any recording-worker failure.""" + if not self._finalized_event.is_set(): + try: + self._transition_control("stopping") + except BaseException as exc: + self._set_worker_error(exc) self._terminate_processing.set() if self._record_thread is not None: self._record_thread.join() - self.check_health() + try: + self.check_health() + finally: + if self._control_server is not None: + self._control_server.close() def check_health(self) -> None: """Raise the first recording-worker error observed by the owner thread.""" @@ -2369,6 +2701,11 @@ def stats(self) -> dict: "is_recording": self.is_recording, } + @property + def control_session_id(self) -> str: + """Stable identifier used by authenticated cross-process clients.""" + return self._control_session_id + @property def capture(self): """Load the CaptureSession after recording completes. diff --git a/tests/control_recorder_process.py b/tests/control_recorder_process.py new file mode 100644 index 0000000..5bea07b --- /dev/null +++ b/tests/control_recorder_process.py @@ -0,0 +1,71 @@ +"""Cross-process helper for the Capture control contract tests.""" + +from __future__ import annotations + +import os +import sys +import time +from pathlib import Path + +from openadapt_capture import recorder as recorder_module +from openadapt_capture.db import create_db, crud + + +def _fake_record( + *, + capture_dir, + terminate_processing, + terminate_recording, + status_pipe, + **_kwargs, +) -> None: + Path(capture_dir).mkdir(parents=True, exist_ok=True) + db_path = Path(capture_dir) / "recording.db" + engine, session_factory = create_db(str(db_path)) + session = session_factory() + try: + crud.insert_recording( + session, + { + "timestamp": time.time(), + "monitor_width": 1280, + "monitor_height": 720, + "pixel_ratio": 1.0, + "double_click_interval_seconds": 0.5, + "double_click_distance_pixels": 5.0, + "platform": sys.platform, + "task_description": "control contract subprocess", + }, + ) + finally: + session.close() + engine.dispose() + status_pipe.send({"type": "record.started"}) + terminate_processing.wait() + status_pipe.send({"type": "record.stopping"}) + if os.environ.get("OPENADAPT_CONTROL_TEST_STALL") == "1": + time.sleep(2.0) + terminate_recording.set() + status_pipe.send({"type": "record.stopped"}) + + +def main() -> None: + capture_dir, runtime_dir = sys.argv[1:3] + recorder_module.record = _fake_record + recorder = recorder_module.Recorder( + capture_dir, + capture_video=False, + capture_images=True, + plot_performance=False, + control_runtime_dir=runtime_dir, + ) + with recorder: + if not recorder.wait_for_ready(timeout=15): + raise RuntimeError("test recorder did not become ready") + print(recorder.control_session_id, flush=True) + while recorder.is_recording: + time.sleep(0.05) + + +if __name__ == "__main__": + main() diff --git a/tests/test_cli_record.py b/tests/test_cli_record.py index 3dee62e..9a4b92d 100644 --- a/tests/test_cli_record.py +++ b/tests/test_cli_record.py @@ -4,7 +4,8 @@ import pytest -from openadapt_capture.cli import record +from openadapt_capture.cli import record, status, stop +from openadapt_capture.control import CaptureControlUnavailable, RecorderStatus class _RecorderThatNeverBecomesReady: @@ -46,3 +47,40 @@ def recorder_factory(*args, **kwargs): assert "did not become ready" in output assert "Recorded 0 events" not in output assert "Saved to:" not in output + + +def _status_result() -> RecorderStatus: + return RecorderStatus( + session_id="7fce2c55-5391-47d4-96bf-b3c90feaa69f", + pid=42, + process_started_at=100.0, + capture_dir="/capture", + phase="complete", + ready=True, + complete=True, + integrity_verified=True, + event_counts={"action": 1}, + ) + + +def test_status_cli_calls_the_public_control_api(monkeypatch, capsys): + import openadapt_capture.control as control_module + + monkeypatch.setattr(control_module, "status_recording", lambda *args, **kwargs: _status_result()) + status("7fce2c55-5391-47d4-96bf-b3c90feaa69f") + output = capsys.readouterr().out + assert '"phase": "complete"' in output + assert '"integrity_verified": true' in output + + +def test_stop_cli_returns_non_success_when_no_recorder_exists(monkeypatch, capsys): + import openadapt_capture.control as control_module + + def unavailable(*args, **kwargs): + raise CaptureControlUnavailable("No live Capture recorder was found.") + + monkeypatch.setattr(control_module, "stop_recording", unavailable) + with pytest.raises(SystemExit) as raised: + stop() + assert raised.value.code == 1 + assert "No live Capture recorder" in capsys.readouterr().out diff --git a/tests/test_control.py b/tests/test_control.py new file mode 100644 index 0000000..3d0911e --- /dev/null +++ b/tests/test_control.py @@ -0,0 +1,821 @@ +"""Security and lifecycle tests for cross-process Capture control.""" + +from __future__ import annotations + +import json +import multiprocessing +import os +import queue +import socket +import stat +import subprocess +import sys +import threading +import time +import uuid +from pathlib import Path +from types import SimpleNamespace + +import psutil +import pytest + +from openadapt_capture import control +from openadapt_capture import recorder as recorder_module +from openadapt_capture.config import RecordingConfig, config_override +from openadapt_capture.control import ( + CaptureControlError, + CaptureControlUnavailable, + RecorderControlServer, + discover_recorders, + status_recording, + stop_recording, +) +from openadapt_capture.db import create_db, crud + + +def _terminal_payload(capture_dir: Path, session_id: str) -> dict: + return { + "schema_version": control.TERMINAL_STATE_SCHEMA_VERSION, + "session_id": session_id, + "pid": os.getpid(), + "process_started_at": psutil.Process().create_time(), + "capture_dir": str(capture_dir), + "phase": "recording", + "ready": True, + "complete": False, + "integrity_verified": False, + "error_code": None, + "started_at": time.time(), + "finalized_at": None, + "event_counts": { + "action": 0, + "screen": 0, + "window": 0, + "browser": 0, + "video": 0, + }, + } + + +def _create_minimal_recording( + capture_dir: Path, + *, + browser_messages: list[object] | None = None, +) -> None: + capture_dir.mkdir(parents=True, exist_ok=True) + engine, session_factory = create_db(str(capture_dir / "recording.db")) + session = session_factory() + try: + recording = crud.insert_recording( + session, + { + "timestamp": time.time(), + "monitor_width": 1280, + "monitor_height": 720, + "pixel_ratio": 1.0, + "double_click_interval_seconds": 0.5, + "double_click_distance_pixels": 5.0, + "platform": sys.platform, + "task_description": "control verification test", + }, + ) + for offset, message in enumerate(browser_messages or []): + crud.insert_browser_event( + session, + recording, + time.time() + offset, + {"message": message}, + ) + finally: + session.close() + engine.dispose() + + +def _wait_for_session(runtime_dir: Path, timeout: float = 15.0) -> str: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + sessions = discover_recorders(runtime_dir) + if len(sessions) == 1: + try: + if status_recording(sessions[0], runtime_dir=runtime_dir, timeout=1).ready: + return sessions[0] + except CaptureControlError: + pass + time.sleep(0.05) + raise AssertionError("the subprocess recorder did not publish its control session") + + +def _start_child( + tmp_path: Path, + *, + stall: bool = False, +) -> tuple[subprocess.Popen[str], Path, Path]: + capture_dir = tmp_path / "capture" + runtime_dir = tmp_path / "runtime" + env = os.environ.copy() + if stall: + env["OPENADAPT_CONTROL_TEST_STALL"] = "1" + child = subprocess.Popen( + [ + sys.executable, + str(Path(__file__).with_name("control_recorder_process.py")), + str(capture_dir), + str(runtime_dir), + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + env=env, + ) + return child, capture_dir, runtime_dir + + +def _finish_child(child: subprocess.Popen[str]) -> tuple[str, str]: + try: + stdout, stderr = child.communicate(timeout=15) + except subprocess.TimeoutExpired: + child.kill() + stdout, stderr = child.communicate(timeout=5) + raise AssertionError(f"recorder child did not exit\nstdout={stdout}\nstderr={stderr}") + return stdout, stderr + + +def test_subprocess_ready_status_stop_and_complete(tmp_path: Path) -> None: + """The same public contract runs on the macOS, Windows, and Linux matrix.""" + child, capture_dir, runtime_dir = _start_child(tmp_path) + try: + session_id = _wait_for_session(runtime_dir) + current = status_recording(session_id, runtime_dir=runtime_dir) + assert current.phase == "recording" + assert current.ready + assert not current.complete + + completed = stop_recording(session_id, runtime_dir=runtime_dir, timeout=10) + assert completed.phase == "complete" + assert completed.complete + assert completed.integrity_verified + stdout, stderr = _finish_child(child) + assert child.returncode == 0, f"stdout={stdout}\nstderr={stderr}" + terminal = json.loads( + (capture_dir / control.TERMINAL_STATE_FILENAME).read_text(encoding="utf-8") + ) + assert terminal["session_id"] == session_id + assert terminal["complete"] is True + assert terminal["integrity_verified"] is True + assert "token" not in terminal + assert "capture_dir" not in terminal + assert discover_recorders(runtime_dir) == [] + finally: + if child.poll() is None: + child.kill() + child.wait(timeout=5) + + +def test_concurrent_repeated_stop_is_idempotent(tmp_path: Path) -> None: + child, _capture_dir, runtime_dir = _start_child(tmp_path) + try: + session_id = _wait_for_session(runtime_dir) + results: list[object] = [] + + def request_stop() -> None: + try: + results.append(stop_recording(session_id, runtime_dir=runtime_dir, timeout=10)) + except BaseException as exc: # retained for an exact assertion below + results.append(exc) + + callers = [threading.Thread(target=request_stop) for _ in range(2)] + for caller in callers: + caller.start() + for caller in callers: + caller.join(timeout=15) + assert len(results) == 2 + assert all(not isinstance(result, BaseException) for result in results), repr(results) + assert all(result.complete for result in results) # type: ignore[union-attr] + _finish_child(child) + assert child.returncode == 0 + finally: + if child.poll() is None: + child.kill() + child.wait(timeout=5) + + +def test_crash_recovery_removes_only_proven_stale_endpoint(tmp_path: Path) -> None: + child, capture_dir, runtime_dir = _start_child(tmp_path) + session_id = _wait_for_session(runtime_dir) + descriptor = control._parse_descriptor(runtime_dir / f"{session_id}.json") + recorder_process = psutil.Process(descriptor.pid) + recorder_process.kill() + recorder_process.wait(timeout=15) + _finish_child(child) + + kernel_live = ( + control._windows_process_live(descriptor.pid) + if sys.platform == "win32" + else None + ) + instance_live = control._process_instance_live( + descriptor.pid, + descriptor.process_started_at, + ) + assert instance_live is False, { + "child_pid": child.pid, + "descriptor_pid": descriptor.pid, + "child_returncode": child.returncode, + "kernel_live": kernel_live, + "pid_exists": psutil.pid_exists(descriptor.pid), + } + assert discover_recorders(runtime_dir) == [] + assert not (runtime_dir / f"{session_id}.json").exists() + terminal = json.loads( + (capture_dir / control.TERMINAL_STATE_FILENAME).read_text(encoding="utf-8") + ) + assert terminal["phase"] == "crashed" + assert terminal["complete"] is False + assert terminal["error_code"] == "recorder_process_exited" + + +def test_finalization_timeout_is_not_success(tmp_path: Path) -> None: + child, capture_dir, runtime_dir = _start_child(tmp_path, stall=True) + try: + session_id = _wait_for_session(runtime_dir) + with pytest.raises(CaptureControlError, match="finalization_timeout"): + stop_recording(session_id, runtime_dir=runtime_dir, timeout=0.05) + terminal = json.loads( + (capture_dir / control.TERMINAL_STATE_FILENAME).read_text(encoding="utf-8") + ) + assert terminal["complete"] is False + assert terminal["integrity_verified"] is False + assert terminal["error_code"] == "finalization_timeout" + stdout, stderr = _finish_child(child) + assert child.returncode == 0, f"stdout={stdout}\nstderr={stderr}" + terminal = json.loads( + (capture_dir / control.TERMINAL_STATE_FILENAME).read_text(encoding="utf-8") + ) + assert terminal["complete"] is True + finally: + if child.poll() is None: + child.kill() + child.wait(timeout=5) + + +def test_wrong_token_and_replaced_instance_fail_closed(tmp_path: Path) -> None: + state = _terminal_payload(tmp_path / "capture", str(uuid.uuid4())) + stop_calls = 0 + + def snapshot() -> dict: + return dict(state) + + def stop(_timeout: float) -> dict: + nonlocal stop_calls + stop_calls += 1 + return dict(state) + + server = RecorderControlServer( + capture_dir=state["capture_dir"], + snapshot=snapshot, + stop=stop, + session_id=state["session_id"], + runtime_dir=tmp_path / "runtime", + ).start() + try: + descriptor = control._parse_descriptor(server.descriptor_path) # type: ignore[arg-type] + request = { + "schema_version": control.CONTROL_SCHEMA_VERSION, + "command": "status", + "session_id": descriptor.session_id, + "pid": descriptor.pid, + "process_started_at": descriptor.process_started_at, + "request_id": str(uuid.uuid4()), + "issued_at": time.time(), + "timeout_seconds": 1.0, + } + request["mac"] = control._message_mac("x" * 64, request) + with socket.create_connection((descriptor.host, descriptor.port), timeout=2) as peer: + peer.sendall(control._canonical_json(request) + b"\n") + response = json.loads(control._recv_line(peer)) + assert response["ok"] is False + assert response["error_code"] == "authentication_failed" + assert "token" not in response + + request["request_id"] = str(uuid.uuid4()) + request["mac"] = control._message_mac(descriptor.token, request) + with socket.create_connection((descriptor.host, descriptor.port), timeout=2) as peer: + peer.sendall(control._canonical_json(request) + b"\n") + first_response = json.loads(control._recv_line(peer)) + assert first_response["ok"] is True + with socket.create_connection((descriptor.host, descriptor.port), timeout=2) as peer: + peer.sendall(control._canonical_json(request) + b"\n") + replay_response = json.loads(control._recv_line(peer)) + assert replay_response["ok"] is False + assert replay_response["error_code"] == "authentication_failed" + + request["request_id"] = str(uuid.uuid4()) + request["process_started_at"] += 10 + request["mac"] = control._message_mac(descriptor.token, request) + with socket.create_connection((descriptor.host, descriptor.port), timeout=2) as peer: + peer.sendall(control._canonical_json(request) + b"\n") + response = json.loads(control._recv_line(peer)) + assert response["ok"] is False + assert response["error_code"] == "authentication_failed" + assert stop_calls == 0 + finally: + server.close() + + +def test_control_thread_start_failure_removes_descriptor( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + state = _terminal_payload(tmp_path / "capture", str(uuid.uuid4())) + + def fail_start(_thread: threading.Thread) -> None: + raise RuntimeError("synthetic thread start failure") + + monkeypatch.setattr(threading.Thread, "start", fail_start) + server = RecorderControlServer( + capture_dir=state["capture_dir"], + snapshot=lambda: dict(state), + stop=lambda _timeout: dict(state), + session_id=state["session_id"], + runtime_dir=tmp_path / "runtime", + ) + with pytest.raises(RuntimeError, match="thread start failure"): + server.start() + assert list((tmp_path / "runtime").glob("*.json")) == [] + + +def test_unauthenticated_connections_have_a_hard_thread_limit(tmp_path: Path) -> None: + state = _terminal_payload(tmp_path / "capture", str(uuid.uuid4())) + server = RecorderControlServer( + capture_dir=state["capture_dir"], + snapshot=lambda: dict(state), + stop=lambda _timeout: dict(state), + session_id=state["session_id"], + runtime_dir=tmp_path / "runtime", + ).start() + peers: list[socket.socket] = [] + try: + descriptor = control._parse_descriptor(server.descriptor_path) # type: ignore[arg-type] + for _ in range(control._MAX_CONTROL_REQUEST_THREADS): + peers.append(socket.create_connection((descriptor.host, descriptor.port), timeout=2)) + overflow = socket.create_connection((descriptor.host, descriptor.port), timeout=2) + try: + overflow.settimeout(2) + assert overflow.recv(1) == b"" + finally: + overflow.close() + finally: + for peer in peers: + peer.close() + server.close() + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX mode assertion") +def test_runtime_secret_is_owner_only_and_wrong_owner_is_rejected( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + state = _terminal_payload(tmp_path / "capture", str(uuid.uuid4())) + server = RecorderControlServer( + capture_dir=state["capture_dir"], + snapshot=lambda: dict(state), + stop=lambda _timeout: dict(state), + session_id=state["session_id"], + runtime_dir=tmp_path / "runtime", + ).start() + try: + assert stat.S_IMODE((tmp_path / "runtime").stat().st_mode) == 0o700 + assert server.descriptor_path is not None + assert stat.S_IMODE(server.descriptor_path.stat().st_mode) == 0o600 + raw = server.descriptor_path.read_text(encoding="utf-8") + token = json.loads(raw)["token"] + assert token not in repr(control._parse_descriptor(server.descriptor_path)) + + actual_uid = os.getuid() + monkeypatch.setattr(os, "getuid", lambda: actual_uid + 1) + with pytest.raises(PermissionError, match="does not own"): + control._parse_descriptor(server.descriptor_path) + finally: + server.close() + + +@pytest.mark.skipif(sys.platform != "darwin", reason="macOS extended ACL contract") +def test_macos_extended_acl_is_removed(tmp_path: Path) -> None: + runtime_dir = tmp_path / "runtime" + runtime_dir.mkdir(mode=0o700) + subprocess.run( + ["chmod", "+a", "everyone allow read", str(runtime_dir)], + check=True, + capture_output=True, + text=True, + ) + descriptor = os.open(runtime_dir, os.O_RDONLY | os.O_NOFOLLOW) + try: + assert control._macos_extended_acl_present(descriptor, runtime_dir) + finally: + os.close(descriptor) + + control._protect_path(runtime_dir, directory=True) + descriptor = os.open(runtime_dir, os.O_RDONLY | os.O_NOFOLLOW) + try: + assert not control._macos_extended_acl_present(descriptor, runtime_dir) + finally: + os.close(descriptor) + + +@pytest.mark.skipif(sys.platform != "win32", reason="Windows owner and DACL contract") +def test_windows_owner_check_rejects_a_foreign_sid( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runtime_dir = control._secure_runtime_dir(tmp_path / "runtime") + control._set_and_verify_windows_owner_acl(runtime_dir, _apply=False) + + monkeypatch.setattr(control, "_windows_current_user_sid", lambda: "S-1-5-18") + with pytest.raises(PermissionError, match="current user does not own"): + control._set_and_verify_windows_owner_acl(runtime_dir, _apply=False) + + +def test_multiple_sessions_require_exact_selection(tmp_path: Path) -> None: + state_one = _terminal_payload(tmp_path / "one", str(uuid.uuid4())) + state_two = _terminal_payload(tmp_path / "two", str(uuid.uuid4())) + servers = [ + RecorderControlServer( + capture_dir=state["capture_dir"], + snapshot=lambda state=state: dict(state), + stop=lambda _timeout, state=state: dict(state), + session_id=state["session_id"], + runtime_dir=tmp_path / "runtime", + ).start() + for state in (state_one, state_two) + ] + try: + with pytest.raises(CaptureControlUnavailable, match="More than one"): + status_recording(runtime_dir=tmp_path / "runtime") + assert ( + status_recording(state_one["session_id"], runtime_dir=tmp_path / "runtime").session_id + == state_one["session_id"] + ) + finally: + for server in servers: + server.close() + + +def test_unauthenticated_descriptor_is_never_deleted(tmp_path: Path) -> None: + runtime_dir = tmp_path / "runtime" + runtime_dir.mkdir(mode=0o700) + invalid = runtime_dir / f"{uuid.uuid4()}.json" + invalid.write_text('{"pid": 999999}\n', encoding="utf-8") + if sys.platform != "win32": + invalid.chmod(0o600) + assert discover_recorders(runtime_dir) == [] + assert invalid.exists() + + +def test_pid_reuse_descriptor_is_marked_crashed_and_removed(tmp_path: Path) -> None: + runtime_dir = control._secure_runtime_dir(tmp_path / "runtime") + capture_dir = tmp_path / "capture" + session_id = str(uuid.uuid4()) + stale_start = psutil.Process().create_time() - 100.0 + terminal = _terminal_payload(capture_dir, session_id) + terminal["process_started_at"] = stale_start + control.write_terminal_state(capture_dir, terminal) + descriptor = control._ControlDescriptor( + session_id=session_id, + pid=os.getpid(), + process_started_at=stale_start, + capture_dir=str(capture_dir), + host="127.0.0.1", + port=65534, + created_at=time.time(), + path=runtime_dir / f"{session_id}.json", + token="x" * 64, + ) + control._write_json_atomic( + descriptor.path, + descriptor.serialized(), + owner_only=True, + ) + + assert discover_recorders(runtime_dir) == [] + assert not descriptor.path.exists() + recovered = json.loads( + (capture_dir / control.TERMINAL_STATE_FILENAME).read_text(encoding="utf-8") + ) + assert recovered["phase"] == "crashed" + assert recovered["complete"] is False + + +def test_exited_but_inspectable_process_is_not_live( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Windows can retain an exited process object while a handle stays open.""" + started_at = 1234.5 + + class ExitedProcess: + def create_time(self) -> float: + return started_at + + def is_running(self) -> bool: + raise AssertionError("Windows must use the kernel process signal") + + def status(self) -> str: + raise AssertionError("Windows must use the kernel process signal") + + def fail_psutil(_pid: int) -> ExitedProcess: + raise AssertionError("a signaled Windows process must bypass psutil") + + monkeypatch.setattr(control.psutil, "Process", fail_psutil) + monkeypatch.setattr(control.sys, "platform", "win32") + monkeypatch.setattr(control, "_windows_process_live", lambda _pid: False) + + assert control._process_instance_live(123, started_at) is False + + +def test_exact_running_process_instance_is_live( + monkeypatch: pytest.MonkeyPatch, +) -> None: + started_at = 1234.5 + + class RunningProcess: + def create_time(self) -> float: + return started_at + + def is_running(self) -> bool: + raise AssertionError("Windows must use the kernel process signal") + + def status(self) -> str: + raise AssertionError("Windows must use the kernel process signal") + + monkeypatch.setattr(control.psutil, "Process", lambda _pid: RunningProcess()) + monkeypatch.setattr(control.sys, "platform", "win32") + monkeypatch.setattr(control, "_windows_process_live", lambda _pid: True) + + assert control._process_instance_live(123, started_at) is True + + +def test_windows_signaled_process_object_is_terminal_and_handle_is_closed() -> None: + class Kernel32: + closed: list[int] = [] + + @staticmethod + def OpenProcess(access: int, inherit: bool, pid: int) -> int: + assert access == 0x00100000 + assert inherit is False + assert pid == 123 + return 456 + + @staticmethod + def WaitForSingleObject(handle: int, timeout: int) -> int: + assert handle == 456 + assert timeout == 0 + return 0x00000000 + + @classmethod + def CloseHandle(cls, handle: int) -> bool: + cls.closed.append(handle) + return True + + kernel32 = Kernel32() + + assert control._windows_process_live(123, _kernel32=kernel32) is False + assert kernel32.closed == [456] + + +@pytest.mark.parametrize("wait_result", [0x00000102, 0xFFFFFFFF]) +def test_windows_live_and_unknown_wait_results_fail_closed(wait_result: int) -> None: + class Kernel32: + closed: list[int] = [] + + @staticmethod + def OpenProcess(_access: int, _inherit: bool, _pid: int) -> int: + return 456 + + @staticmethod + def WaitForSingleObject(_handle: int, _timeout: int) -> int: + return wait_result + + @classmethod + def CloseHandle(cls, handle: int) -> bool: + cls.closed.append(handle) + return True + + kernel32 = Kernel32() + + expected = True if wait_result == 0x00000102 else None + assert control._windows_process_live(123, _kernel32=kernel32) is expected + assert kernel32.closed == [456] + + +def test_windows_access_denied_is_unknown( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class Kernel32: + @staticmethod + def OpenProcess(_access: int, _inherit: bool, _pid: int) -> int: + return 0 + + monkeypatch.setattr(control.ctypes, "get_last_error", lambda: 5, raising=False) + + assert control._windows_process_live(123, _kernel32=Kernel32()) is None + + +def test_recorder_failure_keeps_incomplete_state_and_removes_endpoint( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + capture_dir = tmp_path / "capture" + runtime_dir = tmp_path / "runtime" + + def fail_after_ready(*, status_pipe, **_kwargs) -> None: + status_pipe.send({"type": "record.started"}) + raise RuntimeError("synthetic recorder failure") + + monkeypatch.setattr(recorder_module, "record", fail_after_ready) + recorder = recorder_module.Recorder( + str(capture_dir), + capture_video=False, + capture_images=True, + control_runtime_dir=str(runtime_dir), + ) + with pytest.raises(RuntimeError, match="synthetic recorder failure"): + with recorder: + recorder.wait_for_ready(timeout=2) + + assert discover_recorders(runtime_dir) == [] + terminal = json.loads( + (capture_dir / control.TERMINAL_STATE_FILENAME).read_text(encoding="utf-8") + ) + assert terminal["phase"] == "failed" + assert terminal["complete"] is False + assert terminal["integrity_verified"] is False + assert terminal["error_code"] == "recording_or_finalization_failed" + + +def test_complete_state_write_failure_cannot_return_success( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + recorder = recorder_module.Recorder( + str(tmp_path / "capture"), + capture_video=False, + capture_images=True, + ) + persisted: list[dict] = [] + + def fail_complete_write(_capture_dir: str, payload: dict) -> Path: + if payload.get("complete") is True: + raise OSError("synthetic terminal metadata failure") + persisted.append(dict(payload)) + return tmp_path / "capture" / control.TERMINAL_STATE_FILENAME + + monkeypatch.setattr(control, "write_terminal_state", fail_complete_write) + with pytest.raises(OSError, match="terminal metadata failure"): + recorder._transition_control( + "complete", + complete=True, + integrity_verified=True, + finalized=True, + ) + + rolled_back = recorder._control_payload() + assert rolled_back["phase"] == "starting" + assert rolled_back["complete"] is False + assert rolled_back["integrity_verified"] is False + + recorder._transition_control( + "failed", + error_code="recording_or_finalization_failed", + finalized=True, + ) + recorder._finalized_event.set() + returned = recorder._control_stop(0.1) + assert returned["phase"] == "failed" + assert returned["complete"] is False + assert returned["integrity_verified"] is False + assert persisted[-1]["phase"] == "failed" + + +def test_delayed_finalizing_message_cannot_erase_timeout( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + recorder = recorder_module.Recorder( + str(tmp_path / "capture"), + capture_video=False, + capture_images=True, + ) + persisted: list[dict] = [] + + def retain_state(_capture_dir: str, payload: dict) -> Path: + persisted.append(dict(payload)) + return tmp_path / "capture" / control.TERMINAL_STATE_FILENAME + + monkeypatch.setattr(control, "write_terminal_state", retain_state) + recorder._transition_control("finalizing", error_code="finalization_timeout") + recorder._transition_control("finalizing") + + assert recorder._control_payload()["error_code"] == "finalization_timeout" + assert persisted[-1]["error_code"] == "finalization_timeout" + + +def test_completion_race_after_stop_timeout_cannot_return_success( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + recorder = recorder_module.Recorder( + str(tmp_path / "capture"), + capture_video=False, + capture_images=True, + ) + monkeypatch.setattr( + control, + "write_terminal_state", + lambda _capture_dir, _payload: tmp_path / "capture" / "capture-state.json", + ) + + class CompletionRaceEvent: + def is_set(self) -> bool: + return False + + def wait(self, timeout: float) -> bool: + del timeout + recorder._transition_control( + "complete", + complete=True, + integrity_verified=True, + finalized=True, + ) + return False + + recorder._finalized_event = CompletionRaceEvent() # type: ignore[assignment] + + returned = recorder._control_stop(0.01) + + assert returned["phase"] == "finalizing" + assert returned["complete"] is False + assert returned["integrity_verified"] is False + assert returned["error_code"] == "finalization_timeout" + assert recorder._control_payload()["phase"] == "complete" + assert recorder._control_payload()["integrity_verified"] is True + + +def test_integrity_verification_rejects_missing_committed_events(tmp_path: Path) -> None: + capture_dir = tmp_path / "capture" + _create_minimal_recording(capture_dir) + recorder = recorder_module.Recorder( + str(capture_dir), + capture_video=False, + capture_images=True, + ) + recorder._num_action_events.value = 1 + + with pytest.raises(RuntimeError, match="lost action_event events"): + recorder._verify_completed_capture() + + +def test_integrity_verification_rejects_malformed_browser_event(tmp_path: Path) -> None: + capture_dir = tmp_path / "capture" + _create_minimal_recording(capture_dir, browser_messages=["not-an-object"]) + recorder = recorder_module.Recorder( + str(capture_dir), + capture_video=False, + capture_images=True, + ) + recorder._num_browser_events.value = 1 + + with pytest.raises(RuntimeError, match="invalid browser event"): + recorder._verify_completed_capture() + + +def test_browser_events_increment_the_persisted_count() -> None: + event_q: queue.Queue = queue.Queue() + event_q.put( + recorder_module.Event( + timestamp=time.time(), + type="browser", + data={"message": {"eventType": "navigate", "url": "https://example.test"}}, + ) + ) + queues = [queue.Queue() for _ in range(6)] + counters = [multiprocessing.Value("i", 0) for _ in range(5)] + terminate = multiprocessing.Event() + terminate.set() + + with config_override(RecordingConfig(capture_browser_events=True)): + recorder_module.process_events( + event_q, + queues[0], + queues[1], + queues[2], + queues[3], + queues[4], + queues[5], + SimpleNamespace(timestamp=time.time()), + terminate, + threading.Event(), + *counters, + ) + + assert counters[3].value == 1 + assert queues[3].qsize() == 1