daniel156161
6270d8c956
Testing / remote-protocol-compat (0.16.0) (push) Successful in 1m1s
Testing / remote-protocol-compat (0.15.0) (push) Successful in 1m3s
Testing / test (push) Failing after 1m15s
Build & Publish Package / publish (push) Successful in 51s
Package Extension / package-extension (push) Successful in 1m6s
- Add SSH-style server identity keys and known-host verification for remote serve endpoints. - Add remote add/list/remove commands for explicit endpoint persistence. - Fix remote clients listing to fan out through target discovery instead of ambiguous auto-routing. - Add URL glob matching for tabs filter and count with extension tests. - Add n8n credential pinning for server public keys or SHA256 fingerprints. - Remove obsolete compat shim behavior while keeping empty compat seams for future protocol changes. - Bump browser-cli to 0.16.4 and n8n node to 0.3.1. - Cover known-hosts, remote registry, compat seams, n8n protocol verification, and URL matching with tests.
75 lines
2.7 KiB
Python
75 lines
2.7 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(endpoint: str, key_spec: str | None = None) -> None:
|
|
"""Persist a remote endpoint, optionally with a key spec."""
|
|
if not endpoint:
|
|
return
|
|
normalized = _normalize_endpoint(endpoint)
|
|
remotes = load_remotes()
|
|
current = remotes.get(normalized, {})
|
|
if key_spec:
|
|
if not is_valid_key_spec(key_spec):
|
|
return
|
|
current["key"] = key_spec
|
|
remotes[normalized] = 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) + "\n")
|
|
|
|
def remove_remote(endpoint: str) -> bool:
|
|
"""Remove a remembered remote endpoint. Returns True when it existed."""
|
|
normalized = _normalize_endpoint(endpoint)
|
|
remotes = load_remotes()
|
|
if normalized not in remotes:
|
|
return False
|
|
del remotes[normalized]
|
|
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) + "\n")
|
|
return True
|
|
|
|
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."""
|
|
save_remote(endpoint, key_spec)
|
|
|
|
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
|