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