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.
60 lines
2.4 KiB
Python
60 lines
2.4 KiB
Python
"""Built-in control commands handled directly by ``browser-cli serve``."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from browser_cli.serve.logging import log_request
|
|
|
|
class ServeControlMixin:
|
|
addr: tuple
|
|
command: str
|
|
auth_keys_path: Path | None
|
|
|
|
async def send_error(self, msg: str, msg_id=None) -> None: ...
|
|
async def send_ok(self, payload, command: str | None = None) -> None: ...
|
|
|
|
async def handle_control_command(self, msg: dict) -> bool:
|
|
if self.command == "browser-cli.targets":
|
|
from browser_cli.client import active_browser_targets
|
|
targets = [
|
|
{"profile": target.profile, "displayName": target.display_name}
|
|
for target in active_browser_targets(include_remotes=False)
|
|
]
|
|
await self.send_ok(targets, self.command)
|
|
log_request(self.addr, self.command, None, "OK")
|
|
return True
|
|
|
|
if self.command == "browser-cli.auth.keys":
|
|
if self.auth_keys_path is None:
|
|
await self.send_error("no authorized keys file configured on this server")
|
|
log_request(self.addr, self.command, None, "ERROR", "no authorized keys file")
|
|
return True
|
|
from browser_cli.auth import load_authorized_keys_with_names
|
|
entries = [{"pubkey": pk, "name": name} for pk, name in load_authorized_keys_with_names(self.auth_keys_path)]
|
|
await self.send_ok(entries, self.command)
|
|
log_request(self.addr, self.command, None, "OK")
|
|
return True
|
|
|
|
if self.command == "browser-cli.auth.trust":
|
|
return await self._handle_trust(msg)
|
|
return False
|
|
|
|
async def _handle_trust(self, msg: dict) -> bool:
|
|
if self.auth_keys_path is None:
|
|
await self.send_error("no authorized keys file configured on this server")
|
|
log_request(self.addr, self.command, None, "ERROR", "no authorized keys file")
|
|
return True
|
|
from browser_cli.auth import add_authorized_key
|
|
args = msg.get("args") or {}
|
|
pubkey = str(args.get("pubkey") or "")
|
|
name = str(args.get("name") or "")
|
|
if not re.fullmatch(r"[0-9a-f]{64}", pubkey):
|
|
await self.send_error("invalid pubkey: expected 64 lowercase hex characters")
|
|
log_request(self.addr, self.command, None, "ERROR", "invalid pubkey")
|
|
return True
|
|
added = add_authorized_key(self.auth_keys_path, pubkey, name)
|
|
await self.send_ok({"added": added}, self.command)
|
|
log_request(self.addr, self.command, None, "OK" if added else "ALREADY_TRUSTED")
|
|
return True
|