homelab-codex-ws/scripts/ha/lib/ha_api.py

#!/usr/bin/env python3
"""Minimal REST client for the HA "api" adapter (see DESIGN.md, "Deploy path:
adapter per instance").

Every entrypoint here takes a *token_path*, never a token value, and reads
the token from disk itself. The bearer token must never appear as a process
argv (visible to any local user via `ps`) or in a log line — see DESIGN.md,
"Tokens", and the api-adapter requirements in kb/phases/backlog.md. Callers
needing the token value in-process (e.g. the WebSocket client in
ha_ws.py) should call read_token() directly rather than shelling out.

Read-only by construction: the only method this module exposes is GET.
"""
import os
import sys

import requests


class HaApiError(RuntimeError):
    """Raised for adapter-level failures (missing/empty token, etc.)."""


def read_token(token_path):
    """Read and strip the bearer token from `token_path` (`~` expanded).

    Hard error (raises HaApiError) if the file is absent or empty — the api
    adapter has no filesystem fallback, so a missing token means the import
    cannot do anything at all, unlike the fixtures fetch, which soft-skips.
    """
    path = os.path.expanduser(token_path) if token_path else ""
    if not path or not os.path.isfile(path):
        raise HaApiError(
            f"no HA token at '{path or '<unset>'}' — the api adapter requires a "
            "deploy_agent long-lived access token (see DESIGN.md, 'Tokens'); "
            "this is a hard error, not a soft-skip, because nothing in this "
            "adapter works without one"
        )
    token = open(path, "r", encoding="utf-8").read().strip()
    if not token:
        raise HaApiError(f"token file '{path}' is empty")
    return token


class Client:
    """Thin wrapper around one HA instance's REST API (GET only)."""

    def __init__(self, base_url, token, timeout=30):
        self._base_url = base_url.rstrip("/")
        self._token = token
        self._timeout = timeout

    def _headers(self):
        return {"Authorization": f"Bearer {self._token}"}

    def get(self, path):
        """GET `path` (e.g. "/api/states"), return parsed JSON.

        Returns None on 404 so callers can treat "config not found" (e.g. an
        automation removed between the states poll and the config fetch) as
        a skip-and-report condition instead of a hard failure. Any other
        non-2xx status raises requests.HTTPError.
        """
        resp = requests.get(
            self._base_url + path, headers=self._headers(), timeout=self._timeout
        )
        if resp.status_code == 404:
            return None
        resp.raise_for_status()
        return resp.json()

    def get_raw_text(self, path):
        """GET `path`, return the raw response body (used for fixtures)."""
        resp = requests.get(
            self._base_url + path, headers=self._headers(), timeout=self._timeout
        )
        resp.raise_for_status()
        return resp.text


def _cli_get_raw(args):
    if len(args) != 3:
        print("usage: ha_api.py get-raw <base_url> <token_path> <api_path>", file=sys.stderr)
        return 2
    base_url, token_path, api_path = args
    token = read_token(token_path)
    client = Client(base_url, token)
    sys.stdout.write(client.get_raw_text(api_path))
    return 0


def main(argv):
    """CLI entrypoint. Only sub-command: `get-raw` — print a raw GET body.

    Used by import.sh's fixtures fetch for the api adapter so the token
    never touches a subprocess argv (unlike `curl -H "Authorization: ..."`,
    which would put it in `ps` output for anyone on the same host).
    """
    if len(argv) < 2:
        print("usage: ha_api.py get-raw <base_url> <token_path> <api_path>", file=sys.stderr)
        return 2
    command, rest = argv[1], argv[2:]
    try:
        if command == "get-raw":
            return _cli_get_raw(rest)
        print(f"ha_api.py: unknown command '{command}'", file=sys.stderr)
        return 2
    except (HaApiError, requests.RequestException) as exc:
        print(f"ha_api.py: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))