"""Capability advertisement, targeted operation state and redacted audit events.""" from __future__ import annotations from dataclasses import dataclass, replace from datetime import datetime, timezone import hashlib import hmac import json import os from pathlib import Path import string import threading import time from typing import Any, Mapping from uuid import uuid4 ACTION_POLICY: dict[str, tuple[str, str]] = { "desktop.screenshot": ("screen.capture", "desktop.observe"), "desktop.control": ("hermes.computer_use", "desktop.control"), "files.read": ("files.read", "files.read"), "files.write": ("files.write", "files.write"), "webcam.start": ("webcam.capture", "webcam.read"), "webcam.frame": ("webcam.capture", "webcam.read"), "webcam.stop": ("webcam.capture", "webcam.read"), } @dataclass(frozen=True, slots=True) class NodeOperation: action_id: str action_type: str permission: str capability: str node_id: str source_node_id: str source_node_name: str source_fingerprint: str | None session_id: str operation_id: str trace_id: str grant_id: str | None resource_id: str | None status: str performed: bool simulated: bool created_at: float updated_at: float stop_reason: str | None = None def as_dict(self) -> dict[str, object]: return { "action_id": self.action_id, "type": self.action_type, "permission": self.permission, "capability": self.capability, "node_id": self.node_id, "source_node_id": self.source_node_id, "source_node_name": self.source_node_name, "session_id": self.session_id, "operation_id": self.operation_id, "trace_id": self.trace_id, "grant_id": self.grant_id, "resource_id": self.resource_id, "status": self.status, "performed": self.performed, "simulated": self.simulated, "created_at": _iso(self.created_at), "updated_at": _iso(self.updated_at), "stop_reason": self.stop_reason, } class OperationRegistry: def __init__(self) -> None: self._lock = threading.RLock() self._operations: dict[str, NodeOperation] = {} def start(self, operation: NodeOperation) -> tuple[NodeOperation, bool]: with self._lock: existing = self._operations.get(operation.action_id) if existing is not None: same = ( existing.action_type == operation.action_type and existing.operation_id == operation.operation_id and existing.session_id == operation.session_id ) if not same: raise ValueError("action_id already belongs to another operation") return existing, False self._operations[operation.action_id] = operation return operation, True def get(self, action_id: str) -> NodeOperation | None: with self._lock: return self._operations.get(action_id) def stop(self, action_id: str, reason: str) -> NodeOperation | None: with self._lock: operation = self._operations.get(action_id) if operation is None: return None if operation.status not in {"stopped", "completed", "failed", "revoked", "killed"}: operation = replace( operation, status="revoked" if reason.startswith("revoked") else "stopped", updated_at=time.time(), stop_reason=reason[:240], ) self._operations[action_id] = operation return operation def stop_matching( self, *, reason: str, grant_id: str | None = None, session_id: str | None = None, source_fingerprint: str | None = None, ) -> list[NodeOperation]: stopped: list[NodeOperation] = [] with self._lock: for action_id, operation in list(self._operations.items()): if grant_id is not None and operation.grant_id != grant_id: continue if session_id is not None and operation.session_id != session_id: continue if source_fingerprint is not None and operation.source_fingerprint != source_fingerprint: continue if operation.status == "running": updated = replace( operation, status="killed" if reason == "emergency_stop" else "revoked", updated_at=time.time(), stop_reason=reason[:240], ) self._operations[action_id] = updated stopped.append(updated) return stopped def active(self) -> tuple[NodeOperation, ...]: with self._lock: return tuple(item for item in self._operations.values() if item.status == "running") def active_webcams(self) -> tuple[NodeOperation, ...]: return tuple(item for item in self.active() if item.action_type == "webcam.start") class AuditLog: """Small append-only event trail that never accepts raw action arguments.""" _FORBIDDEN_KEYS = { "token", "secret", "password", "private_key", "credential", "content", "frame", "image", "audio", "video", "clipboard", "text", } def __init__(self, path: Path | None = None, *, max_memory_events: int = 10_000) -> None: self.path = path self._max_memory_events = max_memory_events self._lock = threading.RLock() self._events: list[dict[str, object]] = [] self._seq = 0 def record( self, event_type: str, *, node_id: str, session_id: str = "system", operation_id: str = "system", trace_id: str = "system", payload: Mapping[str, Any] | None = None, ) -> dict[str, object]: with self._lock: self._seq += 1 event: dict[str, object] = { "protocol_version": "1.0", "type": event_type, "event_id": f"evt_{uuid4().hex}", "seq": self._seq, "timestamp": _iso(time.time()), "session_id": _safe_id(session_id), "turn_id": "desktop", "operation_id": _safe_id(operation_id), "trace_id": _safe_id(trace_id), "causation_id": None, "source": "desktop.sidecar", "payload": self._redact({"node_id": _safe_id(node_id), **(payload or {})}), } self._events.append(event) if len(self._events) > self._max_memory_events: self._events = self._events[-self._max_memory_events :] if self.path is not None: self._append(event) return dict(event) def replay(self, *, after_seq: int = 0, limit: int = 1000) -> list[dict[str, object]]: with self._lock: return [dict(event) for event in self._events if int(event["seq"]) > after_seq][:limit] def _append(self, event: Mapping[str, object]) -> None: assert self.path is not None self.path.parent.mkdir(parents=True, exist_ok=True) encoded = (json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8") descriptor = os.open(self.path, os.O_APPEND | os.O_CREAT | os.O_WRONLY, 0o600) try: os.write(descriptor, encoded) finally: os.close(descriptor) @classmethod def _redact(cls, value: object) -> object: if isinstance(value, bytes): return "[binary omitted]" if isinstance(value, Mapping): result: dict[str, object] = {} for key, item in value.items(): normalized = str(key).lower() if any(forbidden in normalized for forbidden in cls._FORBIDDEN_KEYS): result[str(key)] = "[redacted]" else: result[str(key)] = cls._redact(item) return result if isinstance(value, (list, tuple)): return [cls._redact(item) for item in value] if isinstance(value, (str, int, float, bool)) or value is None: return value return str(type(value).__name__) class CapabilityManifest: def __init__(self, *, node_id: str, node_name: str) -> None: self.node_id = node_id self.node_name = node_name self.boot_id = f"boot_{uuid4().hex}" self._sequence = 0 self._lock = threading.Lock() def build( self, *, driver_kind: str, remote_listener: bool, healthy: bool, active_operations: int, signing_key: str | bytes | None = None, runtime_capabilities: Mapping[str, bool] | None = None, provider_capabilities: set[str] | frozenset[str] | None = None, hermes_proxy_enabled: bool = False, ) -> dict[str, object]: with self._lock: self._sequence += 1 sequence = self._sequence runtime = dict(runtime_capabilities or {}) side_effects_enabled = driver_kind == "windows" and any(runtime.values()) manifest: dict[str, object] = { "protocol_version": "1.0", "node_id": self.node_id, "name": self.node_name, "boot_id": self.boot_id, "sequence": sequence, "timestamp": _iso(time.time()), "max_age_seconds": 5, "service": "omni-desktop-companion", "healthy": healthy, "queue_depth": 0, "available_slots": max(0, 1 - active_operations), "capabilities": sorted(ACTION_POLICY), "grant_capabilities": sorted({value[0] for value in ACTION_POLICY.values()}), "permissions": sorted({value[1] for value in ACTION_POLICY.values()}), "provider_capabilities": sorted(provider_capabilities or ()), "tool_schema_versions": {"node_action": "1.0", "event_envelope": "1.0"}, "resource_locality": "resources-stay-on-this-node", "raw_media_forwarding": False, "transport": { "loopback_default": True, "remote_listener": remote_listener, "mtls_required_for_remote": True, "node_action_bearer_required": True, "hermes_mtls_client_auth": hermes_proxy_enabled, }, "gateway": { "pairing_claim": remote_listener, "hermes_run_proxy": hermes_proxy_enabled, "run_events": hermes_proxy_enabled, "run_approval": hermes_proxy_enabled, "run_stop": hermes_proxy_enabled, "qwen_routing": "through-co-located-hermes" if hermes_proxy_enabled else "disabled", }, "execution": { "driver": driver_kind, "side_effects_enabled": side_effects_enabled, "screen_capture_enabled": bool(runtime.get("screen")), "desktop_control_enabled": bool(runtime.get("computer")), "webcam_capture_enabled": bool(runtime.get("webcam")), "files_access_enabled": bool(runtime.get("files")), "mode": ( "real-opt-in" if driver_kind == "windows" else "contract-only" if driver_kind == "noop" else "mock" ), }, "indicators": { "remote_activity": True, "camera_before_first_frame": True, "camera_persistent_until_stop": True, "kill_switch": "Ctrl+Alt+Shift+Esc", }, } if signing_key: key = signing_key.encode("utf-8") if isinstance(signing_key, str) else signing_key canonical = json.dumps( manifest, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") manifest["signature"] = { "algorithm": "hmac-sha256", "value": hmac.new(key, canonical, hashlib.sha256).hexdigest(), } else: manifest["signature"] = None return manifest def make_operation( *, action_id: str, action_type: str, permission: str, capability: str, node_id: str, source_node_id: str, source_node_name: str, source_fingerprint: str | None, session_id: str, operation_id: str, trace_id: str, grant_id: str | None, resource_id: str | None, performed: bool, simulated: bool, initial_status: str = "running", ) -> NodeOperation: timestamp = time.time() return NodeOperation( action_id=action_id, action_type=action_type, permission=permission, capability=capability, node_id=node_id, source_node_id=source_node_id, source_node_name=source_node_name, source_fingerprint=source_fingerprint, session_id=session_id, operation_id=operation_id, trace_id=trace_id, grant_id=grant_id, resource_id=resource_id, status=initial_status, performed=performed, simulated=simulated, created_at=timestamp, updated_at=timestamp, ) def _iso(timestamp: float) -> str: return datetime.fromtimestamp(timestamp, timezone.utc).isoformat().replace("+00:00", "Z") def _safe_id(value: str) -> str: allowed = set(string.ascii_letters + string.digits + "_.:-") filtered = "".join(character for character in str(value) if character in allowed) return (filtered or "unknown")[:160]