6fa931aa36
Testing / remote-protocol-compat (0.9.5) (push) Successful in 56s
Testing / remote-protocol-compat (0.9.3) (push) Successful in 59s
Testing / test (push) Successful in 1m1s
Build & Publish Package / publish (push) Successful in 33s
Package Extension / package-extension (push) Successful in 36s
- Gate TCP serve commands with safe-by-default policies, per-key allow tokens, per-key rate limiting, and audit labels. - Reuse authenticated encrypted remote sessions and parallelize/caches multi-browser fanout to reduce repeated handshake roundtrips. - Increase paged native-host batch size with extension-side byte budgeting to speed large tab listings safely. - Point install output at public Chrome Web Store / Firefox AMO listings by default, with --dev preserving unpacked workflows. - Share search-engine metadata between CLI and SDK and bump the package/extension version to 0.16.0. - Cover the new security, pooling, paging, install, and fanout behavior with expanded Python and extension tests.
92 lines
3.6 KiB
Python
92 lines
3.6 KiB
Python
"""File-based Ed25519 keys and authorized_keys helpers."""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
from cryptography.hazmat.primitives.serialization import (
|
|
Encoding,
|
|
NoEncryption,
|
|
PrivateFormat,
|
|
PublicFormat,
|
|
load_pem_private_key,
|
|
)
|
|
|
|
from browser_cli.auth.agent import AgentKey
|
|
|
|
def generate_keypair() -> tuple[bytes, str]:
|
|
"""Return (private_key_pem_bytes, public_key_hex)."""
|
|
private_key = Ed25519PrivateKey.generate()
|
|
pem = private_key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption())
|
|
public_hex = private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw).hex()
|
|
return pem, public_hex
|
|
|
|
def load_private_key(path: Path) -> Ed25519PrivateKey:
|
|
return load_pem_private_key(path.read_bytes(), password=None)
|
|
|
|
def public_key_hex(key: Ed25519PrivateKey | AgentKey) -> str:
|
|
if isinstance(key, AgentKey):
|
|
return key.pubkey_bytes.hex()
|
|
return key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw).hex()
|
|
|
|
def _parse_authorized_line(line: str) -> tuple[str, str, list[str] | None] | None:
|
|
"""Parse one authorized_keys line into (pubkey, name, categories).
|
|
|
|
Line format: ``<pubkey> [name words...] [allow:cat,cat,...]``. The optional
|
|
``allow:`` token may appear anywhere after the pubkey (conventionally last);
|
|
the remaining words form the name. ``categories`` is None when no ``allow:``
|
|
token is present (the key falls back to the server-wide policy), or a list of
|
|
category strings (possibly empty) otherwise.
|
|
"""
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
return None
|
|
tokens = line.split()
|
|
pubkey = tokens[0]
|
|
categories: list[str] | None = None
|
|
name_tokens: list[str] = []
|
|
for tok in tokens[1:]:
|
|
if tok.startswith("allow:"):
|
|
categories = [c for c in tok[len("allow:"):].split(",") if c]
|
|
else:
|
|
name_tokens.append(tok)
|
|
return pubkey, " ".join(name_tokens), categories
|
|
|
|
def format_authorized_line(pub_hex: str, name: str = "", categories: list[str] | None = None) -> str:
|
|
"""Render an authorized_keys line. Inverse of :func:`_parse_authorized_line`."""
|
|
parts = [pub_hex]
|
|
if name:
|
|
parts.append(name)
|
|
if categories is not None:
|
|
parts.append("allow:" + ",".join(categories))
|
|
return " ".join(parts)
|
|
|
|
def load_authorized_keys_with_names(path: Path) -> list[tuple[str, str]]:
|
|
"""Return list of (pubkey_hex, name) pairs. Name is empty string if not set."""
|
|
return [(pubkey, name) for pubkey, name, _cats in load_authorized_keys_with_policies(path)]
|
|
|
|
def load_authorized_keys_with_policies(path: Path) -> list[tuple[str, str, list[str] | None]]:
|
|
"""Return list of (pubkey_hex, name, categories) triples. categories is None when unset."""
|
|
if not path.exists():
|
|
return []
|
|
result = []
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
parsed = _parse_authorized_line(line)
|
|
if parsed is not None:
|
|
result.append(parsed)
|
|
return result
|
|
|
|
def load_authorized_keys(path: Path) -> list[str]:
|
|
return [pubkey for pubkey, _name in load_authorized_keys_with_names(path)]
|
|
|
|
def add_authorized_key(path: Path, pub_hex: str, name: str = "", categories: list[str] | None = None) -> bool:
|
|
"""Append pub_hex to authorized_keys. Returns False if already present."""
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
existing = {pubkey for pubkey, _name in load_authorized_keys_with_names(path)}
|
|
if pub_hex in existing:
|
|
return False
|
|
line = format_authorized_line(pub_hex, name, categories) + "\n"
|
|
with open(path, "a", encoding="utf-8") as file:
|
|
file.write(line)
|
|
return True
|