"""Contracts for autonomous cross-gateway hosted-room members.""" from __future__ import annotations import hashlib import json import stat from concurrent.futures import ThreadPoolExecutor from pathlib import Path import pytest from gateway.hosted_room_execution_policy import execution_policy_mapping from gateway.hosted_room_peer import ( GatewayRoomCatalog, HostedMemberDispatch, HostedRoomGrantError, HostedRoomPeerError, PROTOCOL_VERSION, catalog_mapping, derive_room_grant_secret, gateway_room_grant_secret, issue_room_grant, local_room_link_endpoint, verify_room_grant, ) from hermes_constants import reset_hermes_home_override, set_hermes_home_override SECRET = b"s" * 32 EXECUTION_POLICY = execution_policy_mapping(target_profile="reviewer", config={"approvals": {"mode": "manual"}}) def test_gateway_room_grant_secret_is_private_persistent_and_not_an_api_key( tmp_path, monkeypatch ): home = tmp_path / ".hermes" profile_home = home / "profiles" / "reviewer" profile_home.mkdir(parents=True) monkeypatch.setenv("HERMES_HOME", str(home)) first = gateway_room_grant_secret() token = set_hermes_home_override(str(profile_home)) try: second = gateway_room_grant_secret() finally: reset_hermes_home_override(token) secret_path = home / ".room-link-grant-secret" assert first == second assert len(first) == 32 assert secret_path.read_bytes() != first assert first != derive_room_grant_secret("gateway-api-key-1234567890") @pytest.mark.platforms("posix") def test_gateway_room_grant_secret_has_owner_only_mode(tmp_path, monkeypatch): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) gateway_room_grant_secret() assert stat.S_IMODE((tmp_path / ".room-link-grant-secret").stat().st_mode) == 0o600 def test_gateway_room_grant_secret_is_atomic_across_concurrent_workers( tmp_path, monkeypatch ): home = tmp_path / ".hermes" monkeypatch.setenv("HERMES_HOME", str(home)) with ThreadPoolExecutor(max_workers=8) as pool: secrets = list(pool.map(lambda _index: gateway_room_grant_secret(), range(8))) assert len(set(secrets)) == 1 assert (home / ".room-link-grant-secret").stat().st_size == 32 def test_room_link_protocol_fixture_matches_backend_contract(): fixture = json.loads( (Path(__file__).parents[1] / "fixtures" / "room_link_protocol_v2.json").read_text( encoding="utf-8" ) ) assert fixture["protocol_version"] == PROTOCOL_VERSION assert fixture["catalog"]["protocol_versions"] == [PROTOCOL_VERSION] def test_room_link_endpoint_reads_supported_config_with_env_override( tmp_path, monkeypatch ): home = tmp_path / ".hermes" home.mkdir() (home / "config.yaml").write_text( "gateway:\n room_link_url: https://configured.example.test/hermes\n", encoding="utf-8", ) monkeypatch.setenv("HERMES_HOME", str(home)) monkeypatch.delenv("HERMES_ROOM_LINK_URL", raising=False) assert local_room_link_endpoint() == { "available": True, "url": "https://configured.example.test/hermes", "transport_security": "tls", } monkeypatch.setenv( "HERMES_ROOM_LINK_URL", "https://override.example.test/hermes" ) assert local_room_link_endpoint()["url"] == ( "https://override.example.test/hermes" ) def test_named_profile_inherits_gateway_room_link_endpoint(tmp_path, monkeypatch): root = tmp_path / "hermes" profile = root / "profiles" / "reviewer" profile.mkdir(parents=True) (root / "config.yaml").write_text( "gateway:\n room_link_url: https://gateway.example.test/hermes\n", encoding="utf-8", ) (profile / "config.yaml").write_text("gateway: {}\n", encoding="utf-8") monkeypatch.setenv("HERMES_HOME", str(root)) monkeypatch.delenv("HERMES_ROOM_LINK_URL", raising=False) token = set_hermes_home_override(profile) try: assert local_room_link_endpoint() == { "available": True, "url": "https://gateway.example.test/hermes", "transport_security": "tls", } finally: reset_hermes_home_override(token) def test_named_profile_room_link_override_wins_over_gateway_root( tmp_path, monkeypatch ): root = tmp_path / "hermes" profile = root / "profiles" / "reviewer" profile.mkdir(parents=True) (root / "config.yaml").write_text( "gateway:\n room_link_url: https://gateway.example.test/hermes\n", encoding="utf-8", ) (profile / "config.yaml").write_text( "gateway:\n room_link_url: https://profile.example.test/hermes\n", encoding="utf-8", ) monkeypatch.setenv("HERMES_HOME", str(root)) monkeypatch.delenv("HERMES_ROOM_LINK_URL", raising=False) token = set_hermes_home_override(profile) try: assert local_room_link_endpoint()["url"] == ( "https://profile.example.test/hermes" ) finally: reset_hermes_home_override(token) def _dispatch(**overrides): prompt = overrides.pop("prompt", "Review the current room state.") value = { "protocol_version": 2, "room_id": "room-1", "home_install_id": "install-home", "authority_gateway_id": "gateway-home", "authority_epoch": 2, "member_id": "member-reviewer", "target_install_id": "install-peer", "target_profile": "reviewer", "task_id": "task-1", "execution_generation": 1, "source_event_seq": 9, "cancellation_scope_id": "cancel-1", "prompt": prompt, "prompt_digest": hashlib.sha256(prompt.encode()).hexdigest(), "capability_digest": "a" * 64, "execution_policy_digest": EXECUTION_POLICY["policy_digest"], "trace_id": "trace-1", **overrides, } return HostedMemberDispatch.from_mapping(value) def test_catalog_digest_is_canonical_and_tamper_evident(): value = catalog_mapping( target_profile="default", installation_id="install-peer", protocol_versions=(2,), link_modes=("direct", "pull"), persistent_process=True, ) catalog = GatewayRoomCatalog.from_mapping(value) assert catalog.text is True assert catalog.attachments is False value["attachments"] = True with pytest.raises(HostedRoomPeerError, match="catalog_digest"): GatewayRoomCatalog.from_mapping(value) def test_dispatch_rejects_unknown_fields_and_prompt_digest_mismatch(): with pytest.raises(HostedRoomPeerError, match="unknown fields"): _dispatch(extra=True) with pytest.raises(HostedRoomPeerError, match="prompt_digest"): _dispatch(prompt_digest="0" * 64) def test_room_grant_is_scoped_to_exact_room_home_target_and_profile(): dispatch = _dispatch() token = issue_room_grant( SECRET, grant_id="grant-1", room_id=dispatch.room_id, home_install_id=dispatch.home_install_id, authority_gateway_id=dispatch.authority_gateway_id, authority_epoch=dispatch.authority_epoch, member_id=dispatch.member_id, target_install_id=dispatch.target_install_id, target_profile=dispatch.target_profile, execution_policy_digest=dispatch.execution_policy_digest, issued_at=100, ttl_seconds=60, ) claims = verify_room_grant(SECRET, token, dispatch, now=120) assert claims["grant_id"] == "grant-1" wrong_target = _dispatch(target_profile="other") with pytest.raises(HostedRoomGrantError, match="scope"): verify_room_grant(SECRET, token, wrong_target, now=120) with pytest.raises(HostedRoomGrantError, match="scope"): verify_room_grant( SECRET, token, _dispatch(member_id="member-other"), now=120 ) with pytest.raises(HostedRoomGrantError, match="scope"): verify_room_grant( SECRET, token, _dispatch(authority_epoch=999), now=120 ) def test_room_grant_fails_closed_for_tamper_expiry_and_permission(): dispatch = _dispatch() token = issue_room_grant( SECRET, grant_id="grant-1", room_id=dispatch.room_id, home_install_id=dispatch.home_install_id, authority_gateway_id=dispatch.authority_gateway_id, authority_epoch=dispatch.authority_epoch, member_id=dispatch.member_id, target_install_id=dispatch.target_install_id, target_profile=dispatch.target_profile, execution_policy_digest=dispatch.execution_policy_digest, permissions=("status",), issued_at=100, ttl_seconds=10, status_expires_at=120, ) with pytest.raises(HostedRoomGrantError, match="allow"): verify_room_grant(SECRET, token, dispatch, now=105) # Dispatch expires quickly, while observation/stop remains available for # bounded headless recovery after the Desktop has closed. assert ( verify_room_grant(SECRET, token, dispatch, permission="status", now=111)[ "grant_id" ] == "grant-1" ) with pytest.raises(HostedRoomGrantError, match="expired"): verify_room_grant( SECRET, token, dispatch, permission="status", now=100 + 30 * 24 * 60 * 60, ) with pytest.raises(HostedRoomGrantError, match="signature"): verify_room_grant(SECRET, token[:-1] + "A", dispatch, now=105) def test_local_catalog_is_honest_for_app_managed_process(monkeypatch): from gateway.hosted_room_peer import local_catalog_mapping monkeypatch.setenv("HERMES_DESKTOP", "1") catalog = local_catalog_mapping(target_profile="default", installation_id="install-desktop") assert catalog["persistent_process"] is False assert catalog["link_modes"] == ["direct"] @pytest.mark.parametrize( ("configured", "available", "reason", "security"), [ (None, False, "not_configured", None), ("http://peer.example.test:8000", False, "invalid_configuration", None), ("http://127.0.0.1:8000", True, None, "loopback"), ("https://peer.example.test", True, None, "tls"), ], ) def test_self_advertised_endpoint_is_explicit_and_validated( monkeypatch, configured, available, reason, security ): from gateway.hosted_room_peer import local_catalog_mapping if configured is None: monkeypatch.delenv("HERMES_ROOM_LINK_URL", raising=False) else: monkeypatch.setenv("HERMES_ROOM_LINK_URL", configured) endpoint = local_catalog_mapping(target_profile="default", installation_id="install-peer")["endpoint"] assert endpoint["available"] is available if reason is not None: assert endpoint == {"available": False, "reason": reason} else: assert endpoint["transport_security"] == security