from hashlib import sha256 import json from pathlib import Path import tempfile from desktop.windows.config import CompanionConfig from desktop.windows.node import AuditLog from desktop.windows.security import GrantAuthority, GrantError, RevocationRegistry, TrustPolicy TOKEN = "unit-test-token-123456" GRANT_KEY = "grant-test-key-0123456789-abcdef-XYZ" def test_remote_bind_requires_explicit_listener_mtls_and_trust() -> None: try: CompanionConfig(host="192.168.50.21", token=TOKEN) except ValueError as exc: assert "explicit" in str(exc) else: raise AssertionError("LAN bind without explicit remote listener must fail") try: CompanionConfig( host="192.168.50.21", token=TOKEN, remote_listener_enabled=True, grant_hmac_key=GRANT_KEY, ) except ValueError as exc: assert "mtls_ca_path" in str(exc) else: raise AssertionError("remote listener without mTLS files must fail") def test_unspecified_bind_is_rejected_even_when_remote_is_explicit() -> None: try: CompanionConfig( host="0.0.0.0", token=TOKEN, remote_listener_enabled=True, grant_hmac_key=GRANT_KEY, ) except ValueError as exc: assert "forbidden" in str(exc) else: raise AssertionError("unspecified bind must never be accepted") def test_explicit_remote_configuration_requires_every_security_artifact() -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) ca = root / "ca.pem" cert = root / "node.pem" key = root / "node-key.pem" trust = root / "nodes.json" for path in (ca, cert, key): path.write_text("offline placeholder; listener is never started", encoding="utf-8") trust.write_text( json.dumps( { "nodes": [ { "node_id": "controller", "certificate_fingerprint_sha256": "a" * 64, "allowed_roles": ["desktop.node.client"], } ] } ), encoding="utf-8", ) config = CompanionConfig( host="192.168.50.21", token=TOKEN, remote_listener_enabled=True, mtls_ca_path=ca, mtls_cert_path=cert, mtls_key_path=key, trust_store_path=trust, grant_hmac_key=GRANT_KEY, audit_path=root / "audit.jsonl", ) assert not config.is_loopback assert config.public_dict()["mtls_required"] assert GRANT_KEY not in repr(config) def test_trust_policy_matches_certificate_fingerprint_and_can_revoke_peer() -> None: certificate = b"offline synthetic certificate bytes" fingerprint = sha256(certificate).hexdigest() with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "nodes.json" path.write_text( json.dumps( { "version": 1, "nodes": [ { "node_id": "paired-controller", "name": "Paired controller", "certificate_fingerprint_sha256": fingerprint, "allowed_roles": ["desktop.node.client"], "enabled": True, } ], } ), encoding="utf-8", ) trust = TrustPolicy.from_file(path) identity = trust.identity_for_certificate(certificate) assert identity is not None assert identity.node_id == "paired-controller" assert "desktop.node.client" in identity.roles assert trust.revoke(fingerprint) assert trust.identity_for_certificate(certificate) is None persisted = json.loads(path.read_text(encoding="utf-8")) assert persisted["nodes"] == [] def test_signed_grants_enforce_target_expiry_revocation_and_device_scope() -> None: revocations = RevocationRegistry() authority = GrantAuthority(GRANT_KEY, revocations=revocations) unsigned = { "grant_id": "grant_camera_1", "node_id": "windows-local", "capability": "webcam.capture", "session_id": "session_1", "path_or_device_scope": {"device_ids": ["camera-front"]}, "issued_at": 100.0, "expires_at": 200.0, } grant = authority.sign(unsigned) verified = authority.verify( grant, node_id="windows-local", session_id="session_1", capability="webcam.capture", arguments={"device_id": "camera-front"}, now=150.0, ) assert verified.grant_id == "grant_camera_1" try: authority.verify( grant, node_id="windows-local", session_id="session_1", capability="webcam.capture", arguments={"device_id": "camera-back"}, now=150.0, ) except GrantError as exc: assert exc.code == "device_outside_grant" else: raise AssertionError("ungranted camera device must fail") revocations.revoke_grant("grant_camera_1") try: authority.verify( grant, node_id="windows-local", session_id="session_1", capability="webcam.capture", arguments={"device_id": "camera-front"}, now=150.0, ) except GrantError as exc: assert exc.code == "grant_revoked" else: raise AssertionError("revoked grant must fail") def test_file_grant_prevents_path_escape() -> None: revocations = RevocationRegistry() authority = GrantAuthority(GRANT_KEY, revocations=revocations) with tempfile.TemporaryDirectory() as directory: allowed = Path(directory) / "allowed" outside = Path(directory) / "outside" / "secret.txt" grant = authority.sign( { "grant_id": "grant_files_1", "node_id": "windows-local", "capability": "files.read", "session_id": "session_files", "path_or_device_scope": {"path_prefixes": [str(allowed)]}, "issued_at": 100.0, "expires_at": 200.0, } ) try: authority.verify( grant, node_id="windows-local", session_id="session_files", capability="files.read", arguments={"path": str(outside)}, now=150.0, ) except GrantError as exc: assert exc.code == "path_outside_grant" else: raise AssertionError("path traversal outside grant must fail") def test_primary_screen_grant_cannot_be_reused_for_all_screens() -> None: authority = GrantAuthority(GRANT_KEY, revocations=RevocationRegistry()) grant = authority.sign( { "grant_id": "grant_screen_1", "node_id": "windows-local", "capability": "screen.capture", "session_id": "session_screen", "path_or_device_scope": {"screen_ids": ["primary"]}, "issued_at": 100.0, "expires_at": 200.0, } ) try: authority.verify( grant, node_id="windows-local", session_id="session_screen", capability="screen.capture", arguments={"all_screens": True}, now=150.0, ) except GrantError as exc: assert exc.code == "screen_outside_grant" else: raise AssertionError("primary-screen grant must not capture every monitor") def test_audit_events_redact_credentials_and_raw_media() -> None: audit = AuditLog() audit.record( "tool_started", node_id="windows-local", payload={"token": "secret-token", "frame": b"raw pixels", "safe": "desktop.control"}, ) event = audit.replay()[0] assert set(event) == { "protocol_version", "type", "event_id", "seq", "timestamp", "session_id", "turn_id", "operation_id", "trace_id", "causation_id", "source", "payload", } payload = event["payload"] assert payload["node_id"] == "windows-local" assert payload["token"] == "[redacted]" assert payload["frame"] == "[redacted]" assert payload["safe"] == "desktop.control"