"""Browser target discovery and local socket resolution.""" import os import socket from dataclasses import dataclass from pathlib import Path from browser_cli.endpoints import display_browser_name from browser_cli.errors import BrowserNotConnected from browser_cli.platform import endpoint_for_alias, is_windows, registry_path from browser_cli.registry import load_registry REGISTRY_PATH = registry_path() @dataclass(frozen=True) class BrowserTarget: profile: str display_name: str socket_path: str remote: str | None = None browser_name: str | None = None display_group: str | None = None # Populated from a remote ``browser-cli.targets`` response when the remote is # new enough to advertise them, letting ``clients`` skip a redundant # ``clients.list`` roundtrip. None means "unknown — fall back to a query". version: str | None = None extension_version: str | None = None def is_reachable_unix_endpoint(endpoint: str) -> bool: """Return True when a Unix socket path exists and accepts connections.""" path = Path(endpoint) if not path.exists(): return False try: with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: sock.settimeout(0.2) sock.connect(endpoint) return True except OSError: return False def active_endpoints(reg: dict) -> dict: """Return only entries whose endpoint appears reachable.""" if is_windows(): return dict(reg) return {k: v for k, v in reg.items() if is_reachable_unix_endpoint(v)} def active_local_browser_targets() -> list[BrowserTarget]: if not REGISTRY_PATH.exists(): return [] reg = load_registry(REGISTRY_PATH) return [ BrowserTarget(profile=profile, display_name=display_browser_name(profile, sock_path), socket_path=sock_path) for profile, sock_path in active_endpoints(reg).items() ] def is_active_local_profile(profile: str | None) -> bool: """Return True when profile names a reachable local browser endpoint.""" if not profile: return False if REGISTRY_PATH.exists(): reg = load_registry(REGISTRY_PATH) if profile in active_endpoints(reg): return True if not is_windows(): try: return Path(endpoint_for_alias(profile)).exists() except Exception: return False return False def resolve_socket(profile: str | None = None) -> str: """Return the socket path for the given profile (or auto-detect).""" target = profile or os.environ.get("BROWSER_CLI_PROFILE") if target: if REGISTRY_PATH.exists(): reg = load_registry(REGISTRY_PATH) if target in reg: return reg[target] return endpoint_for_alias(target) try: active = active_local_browser_targets() if len(active) > 1: aliases = [target.profile for target in active] examples = "\n".join(f" browser-cli --browser {a} ..." for a in aliases) raise BrowserNotConnected( f"Multiple browser instances are active: {', '.join(aliases)}\n" f"Use --browser to select one:\n{examples}" ) if active: return active[0].socket_path except BrowserNotConnected: raise except Exception: pass raise BrowserNotConnected( "Cannot resolve a browser socket automatically.\n" "Make sure the browser is running with the browser-cli extension enabled,\n" "or pass --browser / set BROWSER_CLI_PROFILE to a known alias." )