adam
d41e3838d5
- enclosure: fetch SES element descriptor page (0x07) and label temp/fan/ psu/voltage elements with human-readable names - temps service + /api/temps: per-enclosure hotspot (max across SES sensors and housed drive temps) plus named SES sensor list - mqtt_publisher: paho-mqtt with HA discovery (device per enclosure, hotspot + per-sensor entities), LWT availability, opt-in via MQTT_HOST - secrets: OpenBao KV v2 reader; MQTT creds sourced from secret/home_assistant with env fallback - compose/requirements/README updated
59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
"""Minimal OpenBao (Vault) KV v2 reader for runtime secret injection.
|
|
|
|
Mirrors the homelab pattern used by other non-k8s services: read
|
|
``<addr>/v1/<mount>/data/<path>`` with an ``X-Vault-Token``. Stdlib only —
|
|
no extra dependency. Token comes from ``OPENBAO_TOKEN`` or, preferably, a
|
|
mounted token file via ``OPENBAO_TOKEN_FILE`` (Docker/host secret).
|
|
"""
|
|
import json
|
|
import logging
|
|
import os
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_ADDR = "https://vault.adamksmith.xyz"
|
|
DEFAULT_MOUNT = "secret"
|
|
|
|
|
|
def _resolve_token() -> str | None:
|
|
token = os.environ.get("OPENBAO_TOKEN")
|
|
if token:
|
|
return token.strip()
|
|
token_file = os.environ.get("OPENBAO_TOKEN_FILE")
|
|
if token_file:
|
|
try:
|
|
return open(token_file).read().strip()
|
|
except OSError as e:
|
|
logger.warning("Could not read OPENBAO_TOKEN_FILE %s: %s", token_file, e)
|
|
return None
|
|
|
|
|
|
def read_kv2(path: str) -> dict | None:
|
|
"""Read a KV v2 secret. ``path`` is relative to the mount (e.g.
|
|
"jbod-monitor/mqtt"). Returns the inner data dict, or None on failure.
|
|
"""
|
|
token = _resolve_token()
|
|
if not token:
|
|
logger.warning("OpenBao read requested for %s but no token configured", path)
|
|
return None
|
|
|
|
addr = os.environ.get("OPENBAO_ADDR", DEFAULT_ADDR).rstrip("/")
|
|
mount = os.environ.get("OPENBAO_MOUNT", DEFAULT_MOUNT).strip("/")
|
|
url = f"{addr}/v1/{mount}/data/{path.strip('/')}"
|
|
|
|
req = urllib.request.Request(url, headers={"X-Vault-Token": token})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
body = json.loads(resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
logger.warning("OpenBao read %s failed: HTTP %s", path, e.code)
|
|
return None
|
|
except Exception as e:
|
|
logger.warning("OpenBao read %s failed: %s", path, e)
|
|
return None
|
|
|
|
# KV v2 wraps the secret in data.data.
|
|
return body.get("data", {}).get("data")
|