5b9111afa7
ken is HAOS (no SSH/docker exec path), so it needs a REST/WebSocket-only adapter: automations/scripts/scenes via one GET per /api/config/<domain>/config/<id>, dashboards/area+entity registries/ input_* helpers via the HA WebSocket API (read-only commands only). scripts/ha/lib/ha_api.py and ha_ws.py never take a token as a value — only a token_path, read from disk in-process — so the bearer token never touches a subprocess argv or a log line. ha_ws.py depends on the optional websocket-client package and raises a clear, actionable ImportError if it's missing rather than a raw traceback; import.sh still completes the REST-only part of the import in that case. Automations/scripts/scenes reuse split.write_split() so both adapters produce byte-identical config/<instance>/ layouts and the same idempotent stale-file cleanup on re-run. docker-exec adapter logic is untouched.
115 lines
4 KiB
Python
Executable file
115 lines
4 KiB
Python
Executable file
#!/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 docs/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))
|