daniel156161
076914e5b7
- Split client, native, remote, serve, markdown, and SDK internals into focused packages with direct imports. - Move local and remote transport framing/protocol helpers behind clearer module boundaries. - Break up the extension injected DOM logic into a separate content dispatch bundle and dedicated content modules. - Add explicit client handling for passive remote discovery without noisy PQ warnings. - Keep behavior covered with updated unit, integration, and extension tests.
54 lines
1.9 KiB
Python
54 lines
1.9 KiB
Python
"""Persistence for remembered remote browser endpoints and key specs."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from browser_cli.constants import CONFIG_DIR
|
|
from browser_cli.endpoints import _normalize_endpoint
|
|
|
|
REMOTE_REGISTRY_PATH = CONFIG_DIR / "remotes.json"
|
|
|
|
def load_remotes() -> dict[str, dict[str, str]]:
|
|
if not REMOTE_REGISTRY_PATH.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(REMOTE_REGISTRY_PATH.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return {}
|
|
if not isinstance(data, dict):
|
|
return {}
|
|
# Normalize keys so old entries stored as "domain:443" match current lookups.
|
|
return {_normalize_endpoint(str(endpoint)): cfg for endpoint, cfg in data.items() if isinstance(cfg, dict)}
|
|
|
|
def is_valid_key_spec(value: str) -> bool:
|
|
"""Return True for 'agent', 'agent:<selector>', or a plausible key file path."""
|
|
return value == "agent" or value.startswith("agent:") or (
|
|
not value.startswith("<") and ("/" in value or Path(value).suffix in {".pem", ".key"})
|
|
)
|
|
|
|
def save_remote_key(endpoint: str, key_spec: str) -> None:
|
|
"""Persist the key spec (e.g. 'agent' or a file path) for a remote endpoint."""
|
|
if not endpoint or not key_spec or not is_valid_key_spec(key_spec):
|
|
return
|
|
remotes = load_remotes()
|
|
current = remotes.get(endpoint, {})
|
|
current["key"] = key_spec
|
|
remotes[endpoint] = current
|
|
REMOTE_REGISTRY_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
fd = os.open(str(REMOTE_REGISTRY_PATH), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
|
f.write(json.dumps(remotes, indent=2, sort_keys=True))
|
|
|
|
def key_for_remote(endpoint: str | None) -> str | None:
|
|
if not endpoint:
|
|
return None
|
|
cfg = load_remotes().get(endpoint) or {}
|
|
key = cfg.get("key")
|
|
if not key:
|
|
return None
|
|
key_str = str(key)
|
|
# Reject corrupted values (e.g. str(AgentKey(...)) saved by an older bug).
|
|
return key_str if is_valid_key_spec(key_str) else None
|