Add private Home Assistant credentials settings

2026-08-18 06:14:07 -04:00
parent 39a26adcc2
commit de1b8a7673
6 changed files with 780 additions and 2 deletions
@@ -0,0 +1,255 @@
+#!/usr/bin/env python3
+
+"""Read and atomically update Panama's private Home Assistant settings.
+
+Secret values are accepted only as a single JSON object on stdin and are never
+returned. The command line therefore remains safe to inspect with ps(1).
+"""
+
+from __future__ import annotations
+
+import fcntl
+import json
+import os
+import pathlib
+import re
+import shlex
+import sys
+import tempfile
+import urllib.parse
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+
+KEY_URL = "PANAMA_HOME_ASSISTANT_URL"
+KEY_TOKEN = "PANAMA_HOME_ASSISTANT_TOKEN"
+KEY_ENTITIES = "PANAMA_HOME_ASSISTANT_ENTITIES"
+TARGET_KEYS = (KEY_URL, KEY_TOKEN, KEY_ENTITIES)
+ASSIGNMENT = re.compile(
+    r"^(?P<prefix>\s*(?:export\s+)?)(?P<key>[A-Za-z_][A-Za-z0-9_]*)\s*=(?P<value>.*)$"
+)
+ENTITY_ID = re.compile(r"^[a-z_]+\.[a-z0-9_]+$")
+DEFAULT_ENV = pathlib.Path(__file__).resolve().parents[3] / "bash/env"
+
+
+class ConfigError(RuntimeError):
+    """An error code safe to display without including submitted values."""
+
+
+def env_path() -> pathlib.Path:
+    override = os.environ.get("PANAMA_HOME_ASSISTANT_ENV_FILE", "")
+    return pathlib.Path(override) if override else DEFAULT_ENV
+
+
+def parse_assignment(raw: str) -> str | None:
+    try:
+        parsed = shlex.split(raw, comments=True, posix=True)
+    except ValueError:
+        return None
+    return parsed[0] if len(parsed) == 1 else None
+
+
+def read_values(path: pathlib.Path) -> dict[str, str]:
+    try:
+        lines = path.read_text(encoding="utf-8").splitlines()
+    except FileNotFoundError:
+        return {}
+    except OSError as error:
+        raise ConfigError("read-failed") from error
+
+    values: dict[str, str] = {}
+    for line in lines:
+        match = ASSIGNMENT.match(line)
+        if not match or match.group("key") not in TARGET_KEYS:
+            continue
+        value = parse_assignment(match.group("value"))
+        if value is not None:
+            values[match.group("key")] = value
+    return values
+
+
+def normalize_url(value: Any) -> str:
+    if not isinstance(value, str):
+        raise ConfigError("invalid-url")
+    normalized = value.strip().rstrip("/")
+    if not normalized:
+        return ""
+    parsed = urllib.parse.urlsplit(normalized)
+    if (
+        parsed.scheme not in {"http", "https"}
+        or not parsed.hostname
+        or parsed.username
+        or parsed.password
+    ):
+        raise ConfigError("invalid-url")
+    return normalized
+
+
+def normalize_token(value: Any) -> str:
+    if not isinstance(value, str) or "\x00" in value or "\n" in value or "\r" in value:
+        raise ConfigError("invalid-token")
+    return value.strip()
+
+
+def normalize_entities(value: Any) -> tuple[str, ...]:
+    if isinstance(value, str):
+        candidates: Sequence[Any] = re.split(r"[,\n]", value)
+    elif isinstance(value, list):
+        candidates = value
+    else:
+        raise ConfigError("invalid-entities")
+
+    entities: list[str] = []
+    for candidate in candidates:
+        if not isinstance(candidate, str):
+            raise ConfigError("invalid-entities")
+        entity_id = candidate.strip()
+        if not entity_id:
+            continue
+        if not ENTITY_ID.fullmatch(entity_id):
+            raise ConfigError("invalid-entities")
+        if entity_id not in entities:
+            entities.append(entity_id)
+    return tuple(entities)
+
+
+def public_state(values: Mapping[str, str]) -> dict[str, object]:
+    try:
+        url = normalize_url(values.get(KEY_URL, ""))
+        entities = list(normalize_entities(values.get(KEY_ENTITIES, "")))
+        error = ""
+    except ConfigError as config_error:
+        url = ""
+        entities = []
+        error = str(config_error)
+    token_configured = bool(values.get(KEY_TOKEN, "").strip())
+    return {
+        "ok": error == "",
+        "configured": bool(url and token_configured),
+        "tokenConfigured": token_configured,
+        "url": url,
+        "entities": entities,
+        "error": error,
+    }
+
+
+def render_updated(original: str, updates: Mapping[str, str]) -> str:
+    lines = original.splitlines(keepends=True)
+    rendered: list[str] = []
+    replaced: set[str] = set()
+
+    for line in lines:
+        content = line.rstrip("\r\n")
+        ending = line[len(content) :]
+        match = ASSIGNMENT.match(content)
+        key = match.group("key") if match else ""
+        if key not in updates:
+            rendered.append(line)
+            continue
+        if key in replaced:
+            continue
+        rendered.append(f"export {key}={shlex.quote(updates[key])}{ending or os.linesep}")
+        replaced.add(key)
+
+    if rendered and not rendered[-1].endswith(("\n", "\r")):
+        rendered[-1] += os.linesep
+    for key in TARGET_KEYS:
+        if key in updates and key not in replaced:
+            rendered.append(f"export {key}={shlex.quote(updates[key])}{os.linesep}")
+    return "".join(rendered)
+
+
+def atomic_write(path: pathlib.Path, text: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    descriptor, temporary_name = tempfile.mkstemp(prefix=".env.", dir=path.parent)
+    temporary = pathlib.Path(temporary_name)
+    try:
+        os.fchmod(descriptor, 0o600)
+        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
+            stream.write(text)
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, path)
+        os.chmod(path, 0o600)
+        directory_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(directory_fd)
+        finally:
+            os.close(directory_fd)
+    finally:
+        try:
+            temporary.unlink()
+        except FileNotFoundError:
+            pass
+
+
+def read_payload() -> dict[str, Any]:
+    line = sys.stdin.buffer.readline(1_048_577)
+    if not line or len(line) > 1_048_576:
+        raise ConfigError("invalid-payload")
+    try:
+        payload = json.loads(line)
+    except (UnicodeDecodeError, json.JSONDecodeError) as error:
+        raise ConfigError("invalid-payload") from error
+    if not isinstance(payload, dict):
+        raise ConfigError("invalid-payload")
+    return payload
+
+
+def write_payload(path: pathlib.Path, payload: Mapping[str, Any]) -> dict[str, object]:
+    allowed = {"url", "token", "entities"}
+    if not set(payload).issubset(allowed) or not payload:
+        raise ConfigError("invalid-payload")
+
+    updates: dict[str, str] = {}
+    if "url" in payload:
+        updates[KEY_URL] = normalize_url(payload["url"])
+    if "token" in payload:
+        updates[KEY_TOKEN] = normalize_token(payload["token"])
+    if "entities" in payload:
+        updates[KEY_ENTITIES] = ",".join(normalize_entities(payload["entities"]))
+
+    lock_path = path.with_name("." + path.name + ".lock")
+    path.parent.mkdir(parents=True, exist_ok=True)
+    lock_fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
+    try:
+        os.fchmod(lock_fd, 0o600)
+        fcntl.flock(lock_fd, fcntl.LOCK_EX)
+        try:
+            original = path.read_text(encoding="utf-8")
+        except FileNotFoundError:
+            original = ""
+        except OSError as error:
+            raise ConfigError("read-failed") from error
+        atomic_write(path, render_updated(original, updates))
+    finally:
+        os.close(lock_fd)
+
+    result = public_state(read_values(path))
+    result["ok"] = True
+    result["error"] = ""
+    return result
+
+
+def compact_json(value: Mapping[str, object]) -> str:
+    return json.dumps(value, separators=(",", ":"))
+
+
+def main() -> None:
+    command = sys.argv[1] if len(sys.argv) > 1 else "status"
+    if len(sys.argv) > 2 or command not in {"status", "write"}:
+        raise ConfigError("usage")
+    path = env_path()
+    result = public_state(read_values(path)) if command == "status" else write_payload(path, read_payload())
+    print(compact_json(result))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except ConfigError as error:
+        print(compact_json({"ok": False, "error": str(error)}))
+        raise SystemExit(1) from error
+    except OSError:
+        print(compact_json({"ok": False, "error": "write-failed"}))
+        raise SystemExit(1)