Gabriel Brown
1f694b00b6
.bashrc is back to its one job: export the Panama paths and source what it finds. The personal-env block moves into config/bash/shell — first, because the tmux guard below it reads that file — and the cargo source that was duplicated between the two files lives only in shell now. The real fix is behind that tidying: both Home Assistant helpers defaulted to the IN-REPO config/bash/env, and the writer rebuilt it with only its own three lines — which read as "my env vars vanished" to the person who thought that file was hand-maintained. Both now prefer ~/.config/panama/env, the migrated home outside the checkout, with the repo path kept only as a read fallback for unmigrated machines. The stale legacy copy on this machine is retired; the working credentials were merged into the migrated file first, after probing both sets against the live Home Assistant. Claude-Session: https://claude.ai/code/session_01Ms2FbjQy31TVf3CEvQhGM8
269 lines
8.5 KiB
Python
Executable File
269 lines
8.5 KiB
Python
Executable File
#!/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_]+$")
|
|
# Personal environment lives OUTSIDE the checkout (~/.config/panama/env),
|
|
# where agents, backup tools and `panama update` never walk. The in-repo
|
|
# config/bash/env is only the legacy location for unmigrated machines — and
|
|
# writing there once rebuilt a file the owner thought was hand-maintained.
|
|
LEGACY_ENV = pathlib.Path(__file__).resolve().parents[3] / "bash/env"
|
|
|
|
|
|
def default_env() -> pathlib.Path:
|
|
config_home = pathlib.Path(
|
|
os.environ.get("XDG_CONFIG_HOME", str(pathlib.Path.home() / ".config")))
|
|
migrated = config_home / "panama" / "env"
|
|
if migrated.exists() or not LEGACY_ENV.exists():
|
|
return migrated
|
|
return LEGACY_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)
|