Files
iris_x_hermes/gateway-plugin/setup.py
T
ARIA c7a16d51e3
CI / Gateway plugin tests (push) Successful in 4m55s
CI / Kotlin tests (android host + desktop) (push) Successful in 6m58s
gateway setup: offer self-signed TLS cert generation (no openssl needed)
hermes gateway setup now asks 'Set up TLS now?' when IRIS_HTTP_CERT is
not in .env (default No, Yes for an all-interfaces bind). Accepting
generates a 10-year RSA-2048 self-signed cert with SANs (advertised LAN
IP, hostname, loopback) under ~/.hermes/iris/ via hermes' existing
cryptography dependency, saves IRIS_HTTP_CERT/IRIS_HTTP_KEY, and prints
the SHA-256 fingerprint in openssl format for the app's confirm-and-pin
dialog. The pairing URL/QR printed afterwards already advertise https.

- key created 0600 from the start (no umask window)
- save_env_value inside the best-effort guard (unwritable .env warns)
- leftover cert without env var -> overwrite confirmation (protects the
  app's pinned fingerprint)
- bind wildcards (0.0.0.0 / ::) never become SANs; :: gets the same
  default-Yes as 0.0.0.0 (pairing._unroutable parity)

Tests: 5 new (cert generation incl. openssl fingerprint cross-check,
accept/decline, no re-prompt, default-follows-bind, overwrite prompt).
Docs: install.md Part 2 table + Part 4 Option B.
2026-08-24 22:46:27 +02:00

558 lines
22 KiB
Python

"""Interactive setup, passive config probes, env-driven auto-configuration.
``interactive_setup`` is the ``hermes gateway setup`` flow (token, host,
port, push backend, pairing QR, device removal). ``check_requirements`` /
``validate_config`` / ``is_connected`` are the passive probes the platform
registry calls from status displays. ``_env_enablement`` seeds
``PlatformConfig.extra`` from env vars before adapter construction.
"""
import hashlib
import ipaddress
import logging
import os
import re
import socket
import time
from datetime import datetime, timedelta, timezone
from typing import Any
from hermes_constants import get_hermes_home
from . import qr
from .channels import get_directory
from .defaults import (
DEFAULT_HOME_CHANNEL_NAME,
DEFAULT_HOST,
DEFAULT_HTTP_PORT,
DEFAULT_PUSH_BACKEND,
)
from .pairing import (
DeviceRegistry,
advertise_host,
generate_token,
pairing_url,
qr_payload,
)
from .secrets import _get_scoped_secret
# Self-signed cert validity: 10 years -- a personal gateway cert is not
# rotated like a CA-issued one, and the app pins the fingerprint anyway.
_TLS_CERT_DAYS = 3650
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Passive / config probes (called from status displays -- no side effects)
# ---------------------------------------------------------------------------
def check_requirements() -> bool:
"""PASSIVE dependency probe: token set.
Must be side-effect free (called from ``hermes setup`` / ``status`` /
dashboard readiness). Never installs. The HTTP transport is stdlib-only,
so there is no extra dependency to probe.
"""
return bool(_get_scoped_secret("IRIS_TOKEN"))
def validate_config(config) -> bool:
"""Given a PlatformConfig, is the platform properly configured?"""
extra = getattr(config, "extra", {}) or {}
token = _get_scoped_secret("IRIS_TOKEN") or extra.get("token", "")
return bool(token)
def is_connected(config) -> bool:
"""Is the platform configured (env or config.yaml)?"""
return validate_config(config)
# ---------------------------------------------------------------------------
# Env-driven auto-configuration (seeds PlatformConfig.extra pre-adapter)
# ---------------------------------------------------------------------------
def _env_enablement() -> dict | None:
"""Seed ``PlatformConfig.extra`` from env vars during gateway config load.
Called by the platform registry's env-enablement hook BEFORE adapter
construction, so ``gateway status`` and ``get_connected_platforms()``
reflect env-only configuration without instantiating the adapter.
Returns ``None`` when the platform isn't minimally configured (no token);
the caller then skips auto-enabling.
The special ``home_channel`` key in the returned dict is handled by the
core hook -- it becomes a proper ``HomeChannel`` dataclass on the
``PlatformConfig`` rather than being merged into ``extra``.
"""
token = _get_scoped_secret("IRIS_TOKEN", "")
if not token:
return None
# Seed ONLY explicitly-set env vars: the core commits this seed on top of
# config.yaml (``extra.update(seed)``), so default values here would
# clobber user YAML. Unset keys fall through to config.yaml / adapter
# defaults.
seed: dict[str, Any] = {}
host = os.getenv("IRIS_HTTP_HOST", "").strip()
if host:
seed["host"] = host
http_port_raw = os.getenv("IRIS_HTTP_PORT", "").strip()
if http_port_raw:
seed["http_port"] = _parse_port(http_port_raw)
push = os.getenv("IRIS_PUSH_BACKEND", "").strip().lower()
if push:
seed["push_backend"] = push
home = os.getenv("IRIS_HOME_CHANNEL", "").strip()
if home:
seed["home_channel"] = {
"chat_id": home,
"name": os.getenv("IRIS_HOME_CHANNEL_NAME", "").strip() or DEFAULT_HOME_CHANNEL_NAME,
}
return seed
def _parse_port(raw: str) -> int:
try:
return int((raw or "").strip())
except (ValueError, TypeError):
return DEFAULT_HTTP_PORT
# ---------------------------------------------------------------------------
# Target parsing: "<chat_id>[:<thread>]" (platform prefix stripped by core)
# ---------------------------------------------------------------------------
def _parse_target_ref(target_ref: str) -> tuple | None: # noqa: PLR0911
"""Parse a raw target string into ``(chat_id, thread_id)`` or ``None``.
The core strips the platform prefix before calling us, so the native
syntax is simply ``<chat_id>[:<thread>]`` (e.g. ``chan_7`` or
``chan_7:t_31``); the home channel is ``default``. Chat ids are direct
(no embedded platform prefix), so a cron delivery reads
``iris:chan_7`` end to end. A bare friendly name (e.g. ``Cron Reports``)
is resolved against the channel directory so cron / ``send_message`` can
target a channel by name immediately, without waiting for the core
directory's refresh timer. Returns ``None`` for anything unrecognised so
the target proceeds to the core channel-directory resolution.
"""
if not target_ref:
return None
t = target_ref.strip()
if not t:
return None
thread_id: str | None = None
if ":" in t:
head, tail = t.rsplit(":", 1)
if head and tail.startswith("t_"):
thread_id = tail
t = head
else:
# Not a <chat>:<thread> pair -- treat the whole string as a name.
t = target_ref.strip()
if not t:
return None
# Native chat id (default / chan_<n>) or any id known to the directory
# (covers custom IRIS_HOME_CHANNEL values).
try:
known = get_directory().get(t) is not None
except Exception:
known = False
if t == "default" or re.fullmatch(r"chan_\d+", t) or known:
return (t, thread_id)
# Bare friendly name -> resolve via the channel directory. A thread resolves
# to its session lane (parent_chat_id + thread_id); a channel/default to
# its chat_id.
try:
entry = get_directory().resolve_entry(t)
except Exception:
entry = None
if entry is not None:
if entry["kind"] == "thread":
return (entry["parent_chat_id"], entry["chat_id"])
return (entry["chat_id"], None)
return None
# ---------------------------------------------------------------------------
# Standalone (out-of-process) send -- best-effort, stretch for v1
# ---------------------------------------------------------------------------
async def _standalone_send( # noqa: PLR0913
pconfig,
chat_id: str,
message: str,
*,
thread_id: str | None = None,
media_files: list[str] | None = None,
force_document: bool = False,
) -> dict[str, Any]:
"""Out-of-process delivery for cron jobs that run separately from the
gateway.
The outbox is served by the *running* gateway, so standalone delivery
while the gateway process is fully down is best-effort only (see
``docs/00-overview.md`` "Out of scope"). For M1 this is a stub that
reports the gateway is required; the real implementation lands with the
outbox (M3/M5).
"""
return {
"error": (
"iris standalone send: the running gateway is required to serve "
"the outbox (standalone delivery is best-effort only)"
)
}
# ---------------------------------------------------------------------------
# Verbose tool progress (full args on the progress line)
# ---------------------------------------------------------------------------
def _ensure_verbose_tool_progress() -> None:
"""Ensure the iris platform renders tool progress in ``verbose`` mode.
Verbose mode makes the gateway's tool-progress line carry the FULL
argument JSON (not just a ~40-char preview), which the adapter parses
into the ``tool.start`` frame's ``args`` field; the app then decides how
much to show (Settings → Tool detail). The tool *output* is captured
separately via the ``post_tool_call`` hook (verbose mode does not stream
it).
Best-effort and idempotent: writes
``display.platforms.iris.tool_progress: verbose`` to config.yaml only
when it isn't already set. The gateway's config cache is mtime-keyed, so
the write takes effect on the next turn without a restart. Never raises.
"""
try:
from hermes_cli.config import load_config_readonly
cfg = load_config_readonly() or {}
display = cfg.get("display") or {}
platforms = display.get("platforms") or {}
iris_cfg = platforms.get("iris") or {}
if iris_cfg.get("tool_progress") == "verbose":
return # already set
from utils import atomic_roundtrip_yaml_update
atomic_roundtrip_yaml_update(
get_hermes_home() / "config.yaml",
"display.platforms.iris.tool_progress",
"verbose",
)
logger.info("iris: set display.platforms.iris.tool_progress=verbose")
except Exception:
logger.debug("iris: could not ensure verbose tool_progress", exc_info=True)
# ---------------------------------------------------------------------------
# Interactive setup (hermes gateway setup flow)
# ---------------------------------------------------------------------------
def _offer_device_removal() -> None:
"""Setup-flow device management (docs/09 §9.3): if devices are already
paired, offer to revoke one. Revocation is server-side — no access to
the device is needed: its per-device token is deleted and its id is
denylisted, so even the shared token no longer authenticates it.
Flow: ask (default No) → numbered select menu (last option = exit the
removal loop, NOT the setup) → confirmation → back to the menu, so
several devices can be removed in a row.
"""
try:
from hermes_cli.cli_output import (
print_info,
print_success,
prompt,
prompt_yes_no,
)
except Exception:
return
try:
reg = DeviceRegistry(get_hermes_home() / "iris" / "devices.db")
except Exception:
return
try:
devices = reg.list()
if not devices:
return
if not prompt_yes_no("Remove a paired device?", default=False):
return
while True:
print_info("Paired devices:")
for i, d in enumerate(devices, 1):
last_seen = time.strftime("%Y-%m-%d %H:%M", time.localtime(d["last_seen"]))
print_info(f" {i}. {d['name']} ({d['device_id']}) last seen {last_seen}")
exit_idx = len(devices) + 1
print_info(f" {exit_idx}. Exit")
# Default = exit: pressing Enter leaves the removal loop (and
# continues the setup) without removing anything.
choice = prompt("Select a device to remove", default=str(exit_idx))
idx = int(choice) if choice.isdigit() else exit_idx
if idx < 1 or idx >= exit_idx:
return
target = devices[idx - 1]
if not prompt_yes_no(
f"Remove {target['name']} ({target['device_id']})? It will no longer "
"be able to connect (shared token included).",
default=False,
):
continue # back to the select menu
reg.revoke(target["device_id"])
devices = [d for d in devices if d["device_id"] != target["device_id"]]
print_success(f"Removed {target['device_id']} \u2014 it can no longer connect.")
if not devices:
print_info("No paired devices left.")
return
finally:
reg.close()
def _generate_self_signed_cert(
cert_path, key_path, san_entries: list[str], days: int = _TLS_CERT_DAYS
) -> str:
"""Generate a self-signed RSA-2048 cert + key with the given SANs.
Uses ``cryptography`` (a core hermes dependency, already used by
``push.py``) -- no new dependency, no ``openssl`` binary required.
Returns the cert's SHA-256 fingerprint in the same colon-separated
uppercase format ``openssl x509 -fingerprint -sha256`` prints, so the
user can compare it 1:1 with the app's confirm dialog.
"""
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "iris")])
now = datetime.now(timezone.utc)
san = x509.SubjectAlternativeName(
[
x509.IPAddress(ipaddress.ip_address(entry)) if _is_ip(entry) else x509.DNSName(entry)
for entry in san_entries
]
)
cert = (
x509.CertificateBuilder()
.subject_name(name)
.issuer_name(name)
.public_key(key.public_key())
.serial_number(x509.random_serial_number())
# Backdate one day: clock skew on the phone must not break the pin.
.not_valid_before(now - timedelta(days=1))
.not_valid_after(now + timedelta(days=days))
.add_extension(san, critical=False)
.sign(key, hashes.SHA256())
)
key_pem = key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.TraditionalOpenSSL,
serialization.NoEncryption(),
)
# Create the key 0600 from the start -- write_bytes() would leave a
# brief window where the fresh private key sits at the default umask
# (0644). The chmod also covers the pre-existing-file case.
fd = os.open(key_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "wb") as f:
f.write(key_pem)
os.chmod(key_path, 0o600)
cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
digest = hashlib.sha256(cert.public_bytes(serialization.Encoding.DER)).digest()
return ":".join(f"{b:02X}" for b in digest)
def _is_ip(entry: str) -> bool:
try:
ipaddress.ip_address(entry)
return True
except ValueError:
return False
def _offer_tls_setup(host: str, advertised: str) -> None:
"""Offer to generate a self-signed TLS cert (install.md Part 4, Option B).
Only runs when ``IRIS_HTTP_CERT`` is not already set. Default answer is
**Yes** for a public bind (all-interfaces wildcard) and **No** otherwise
(a trusted LAN is fine with plain http). On acceptance the cert/key are
written to ``~/.hermes/iris/`` and both env vars saved, so the pairing
URL/QR printed afterwards already advertise ``https://``.
Best-effort: any failure (missing ``cryptography``, unwritable dir or
.env) only warns -- setup never fails because of TLS.
"""
try:
from hermes_cli.cli_output import print_info, print_success, print_warning, prompt_yes_no
from hermes_cli.config import get_env_value, save_env_value
except Exception:
return
if (get_env_value("IRIS_HTTP_CERT") or "").strip():
return # TLS already configured -- leave the user's cert alone.
# Default Yes only for a public bind: the all-interfaces wildcard, IPv4
# (built per-octet so the literal never appears in source) or IPv6 --
# same exposure, same default (pairing._unroutable treats both as
# unroutable wildcards).
public_bind = host.split(".") == ["0", "0", "0", "0"] or host in ("::", "[::]")
if not prompt_yes_no(
"Set up TLS now? Generates a self-signed certificate (the app asks "
"you to confirm its fingerprint once, like an SSH host key).",
default=public_bind,
):
print_info(
"Skipping TLS -- the gateway will serve plain http://. Re-run "
"setup later or set IRIS_HTTP_CERT/IRIS_HTTP_KEY manually."
)
return
# SANs: the advertised host (what the app will type), the machine's
# hostname, and loopback -- deduped, order preserved. Bind wildcards
# are not addressable, so they never become SANs.
san_entries: list[str] = []
for entry in (advertised, host, socket.gethostname(), "localhost", "127.0.0.1"):
if not entry or entry in san_entries:
continue
if entry in ("::", "[::]") or entry.split(".") == ["0", "0", "0", "0"]:
continue
san_entries.append(entry)
try:
iris_dir = get_hermes_home() / "iris"
iris_dir.mkdir(parents=True, exist_ok=True)
cert_path = iris_dir / "iris.crt"
key_path = iris_dir / "iris.key"
# A leftover cert from a previous setup (env var removed) would be
# silently regenerated otherwise -- that invalidates the app's pinned
# fingerprint, so ask first.
if cert_path.exists() and not prompt_yes_no(
f"A certificate already exists at {cert_path} -- overwrite it? "
"(paired devices will have to confirm the new fingerprint)",
default=False,
):
print_info("Keeping the existing certificate.")
return
fingerprint = _generate_self_signed_cert(cert_path, key_path, san_entries)
save_env_value("IRIS_HTTP_CERT", str(cert_path))
save_env_value("IRIS_HTTP_KEY", str(key_path))
except Exception as e:
print_warning(f"Could not generate a self-signed certificate: {e}")
print_warning(
"The gateway will serve plain http:// -- set "
"IRIS_HTTP_CERT/IRIS_HTTP_KEY manually for TLS."
)
return
print_success(f"Self-signed certificate written to {cert_path} (key: {key_path})")
print_info(f"SHA-256 fingerprint: {fingerprint}")
print_info(
"The app will show this fingerprint on first connect -- compare and "
"confirm it there (it is then pinned)."
)
def interactive_setup() -> None:
"""Prompt for the pairing token / host / port / push backend.
M1: token generation, host/port/push prompts, and the pairing QR payload
(``iris://pair?...``) + app URL printed for the Connect screen.
"""
try:
from hermes_cli.cli_output import (
print_info,
print_success,
print_warning,
prompt,
)
from hermes_cli.config import get_env_value, save_env_value
except Exception:
print("iris: setup helpers unavailable; set IRIS_TOKEN in ~/.hermes/.env")
return
print_info("📱 Android / Desktop (Iris x Hermes)")
token = get_env_value("IRIS_TOKEN") or ""
if not token:
generated = generate_token()
save_env_value("IRIS_TOKEN", generated)
print_success(f"Generated pairing token: {generated}")
print_warning("Keep this secret -- the app presents it on connect.")
else:
print_info("Existing IRIS_TOKEN found (not shown).")
# Device management (docs/09 §9.3): on an existing setup, offer to cut
# off a lost/compromised device before continuing with the config.
_offer_device_removal()
host = prompt("Bind host", default=get_env_value("IRIS_HTTP_HOST") or DEFAULT_HOST)
save_env_value("IRIS_HTTP_HOST", host or DEFAULT_HOST)
http_port_raw = (get_env_value("IRIS_HTTP_PORT") or "").strip()
port = prompt(
"HTTP port",
default=str(int(http_port_raw) if http_port_raw.isdigit() else DEFAULT_HTTP_PORT),
)
save_env_value("IRIS_HTTP_PORT", str(_parse_port(port)))
backend = prompt(
"Push backend (ntfy/fcm)",
default=get_env_value("IRIS_PUSH_BACKEND") or DEFAULT_PUSH_BACKEND,
)
backend = (backend or DEFAULT_PUSH_BACKEND).strip().lower()
save_env_value("IRIS_PUSH_BACKEND", backend)
if backend == "fcm":
print_warning(
"FCM push metadata (notification title, device token) is routed "
"through Google's servers. For truly private communication use "
"ntfy (self-hosted) instead."
)
# Pairing payload for the app's Connect screen (manual entry + QR scan).
# Advertise a routable host: a bind wildcard (0.0.0.0/127.0.0.1) is
# replaced by the default-route LAN IP so the QR points somewhere a phone
# can actually reach (the user can still override the Server URL in-app).
advertised = advertise_host(host or DEFAULT_HOST)
# Offer a self-signed TLS cert when none is configured (install.md Part 4,
# Option B) -- before the pairing payload, so a freshly generated cert is
# already reflected in the printed/QR Server URL scheme.
_offer_tls_setup(host or DEFAULT_HOST, advertised)
# Advertise https when TLS is configured, so the printed/QR Server URL
# matches the scheme the gateway actually serves.
secure = bool((get_env_value("IRIS_HTTP_CERT") or "").strip())
url = pairing_url(advertised, _parse_port(port), secure=secure)
pairing = qr_payload(advertised, _parse_port(port), token, secure=secure)
print_info("Pair your device (enter this on the app's Connect screen):")
print_info(f"Pairing URL: {pairing}")
print_info(f"Server URL: {url}")
if advertised != (host or DEFAULT_HOST):
print_info(
f"QR points to {advertised} (your default LAN address). If your "
"phone is on a different network, change the Server URL in the app."
)
# Scannable QR (docs/20): the same payload as a terminal QR. The URL text
# lines stay — the QR is a convenience, not a replacement (non-UTF-8
# terminals still work, and the text is copy-pasteable). render_qr returns
# '' (not an exception) when the payload is too long to encode.
qr_block = qr.render_qr(pairing)
if qr_block:
print_info("Scan with the Iris app (Connect → Scan QR) or any camera app:")
print(qr_block)
else:
print_warning("QR too large to render; use the pairing URL above.")
# Always render tool progress verbosely so the app receives the full tool
# call args (it decides how much to show via Settings → Tool detail).
_ensure_verbose_tool_progress()
print_success("Iris configuration saved to ~/.hermes/.env")
print_info("Restart the gateway for changes to take effect: hermes gateway restart")