from __future__ import annotations import os from pathlib import Path from .jsonstore import JsonRegistry ALIAS_REGISTRY_VERSION = 1 MAX_ALIAS_LENGTH = 48 class AliasRegistryError(RuntimeError): """A display-name registry error which is safe to show to the user.""" def normalize_alias(label: str) -> str: normalized = label.strip() if len(normalized) > MAX_ALIAS_LENGTH: raise ValueError(f"display name must be at most {MAX_ALIAS_LENGTH} characters") if any(ord(character) < 32 or ord(character) == 127 for character in normalized): raise ValueError("display name must not contain control characters") return normalized def default_alias_registry_path() -> Path: if override := os.environ.get("HERDR_HUB_ALIASES_FILE"): return Path(override).expanduser() config_home = os.environ.get("XDG_CONFIG_HOME") root = Path(config_home).expanduser() if config_home else Path.home() / ".config" return root / "herdr-hub" / "aliases.json" class AliasRegistry(JsonRegistry[dict[str, str]]): error_type = AliasRegistryError def __init__(self, path: Path | None = None) -> None: super().__init__(path or default_alias_registry_path()) def _empty(self) -> dict[str, str]: return {} def _parse(self, payload: object) -> dict[str, str]: return _parse_aliases(payload, self.path) def _serialize(self, aliases: dict[str, str]) -> object: return { "version": ALIAS_REGISTRY_VERSION, "aliases": dict(sorted(aliases.items())), } def set(self, endpoint_key: str, label: str) -> bool: _validate_endpoint_key(endpoint_key) normalized = normalize_alias(label) def update(aliases: dict[str, str]) -> tuple[dict[str, str], bool]: updated = dict(aliases) if normalized: changed = updated.get(endpoint_key) != normalized updated[endpoint_key] = normalized else: changed = endpoint_key in updated updated.pop(endpoint_key, None) return updated, changed return self._update(update) def _parse_aliases(payload: object, path: Path) -> dict[str, str]: if ( not isinstance(payload, dict) or payload.get("version") != ALIAS_REGISTRY_VERSION ): raise AliasRegistryError(f"{path} has an unsupported or missing version") raw_aliases = payload.get("aliases") if not isinstance(raw_aliases, dict): raise AliasRegistryError(f"{path} does not contain an aliases object") aliases: dict[str, str] = {} for key, raw_label in raw_aliases.items(): if not isinstance(key, str) or not isinstance(raw_label, str): raise AliasRegistryError(f"{path} contains an invalid alias entry") try: _validate_endpoint_key(key) label = normalize_alias(raw_label) except ValueError as error: raise AliasRegistryError(f"{path}: {error}") from error if not label: raise AliasRegistryError(f"{path}: display names must not be empty") aliases[key] = label return dict(sorted(aliases.items())) def _validate_endpoint_key(key: str) -> None: if not key or len(key) > 1024: raise ValueError("endpoint key has an invalid length") if any(ord(character) < 32 or ord(character) == 127 for character in key): raise ValueError("endpoint key must not contain control characters")