152 lines
5.5 KiB
Python
152 lines
5.5 KiB
Python
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
from typing import Callable, TypeVar
|
|
|
|
|
|
ALIAS_REGISTRY_VERSION = 1
|
|
MAX_ALIAS_LENGTH = 48
|
|
UpdateResult = TypeVar("UpdateResult")
|
|
|
|
|
|
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:
|
|
def __init__(self, path: Path | None = None) -> None:
|
|
self.path = path or default_alias_registry_path()
|
|
|
|
@property
|
|
def lock_path(self) -> Path:
|
|
return self.path.with_name(f".{self.path.name}.lock")
|
|
|
|
def load(self) -> dict[str, str]:
|
|
if not self.path.exists():
|
|
return {}
|
|
try:
|
|
payload = json.loads(self.path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as error:
|
|
raise AliasRegistryError(f"cannot read {self.path}: {error}") from error
|
|
return _parse_aliases(payload, self.path)
|
|
|
|
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 _update(
|
|
self,
|
|
callback: Callable[[dict[str, str]], tuple[dict[str, str], UpdateResult]],
|
|
) -> UpdateResult:
|
|
try:
|
|
self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
with self.lock_path.open("a+", encoding="utf-8") as lock:
|
|
fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
|
|
aliases = self.load()
|
|
updated, outcome = callback(aliases)
|
|
if updated != aliases:
|
|
self._write(updated)
|
|
return outcome
|
|
except AliasRegistryError:
|
|
raise
|
|
except OSError as error:
|
|
raise AliasRegistryError(f"cannot update {self.path}: {error}") from error
|
|
|
|
def _write(self, aliases: dict[str, str]) -> None:
|
|
payload = {
|
|
"version": ALIAS_REGISTRY_VERSION,
|
|
"aliases": dict(sorted(aliases.items())),
|
|
}
|
|
temporary_name: str | None = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w",
|
|
encoding="utf-8",
|
|
dir=self.path.parent,
|
|
prefix=f".{self.path.name}.",
|
|
delete=False,
|
|
) as temporary:
|
|
temporary_name = temporary.name
|
|
json.dump(payload, temporary, ensure_ascii=False, indent=2)
|
|
temporary.write("\n")
|
|
temporary.flush()
|
|
os.fsync(temporary.fileno())
|
|
os.chmod(temporary_name, 0o600)
|
|
os.replace(temporary_name, self.path)
|
|
directory_fd = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(directory_fd)
|
|
finally:
|
|
os.close(directory_fd)
|
|
finally:
|
|
if temporary_name is not None:
|
|
try:
|
|
Path(temporary_name).unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
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")
|