iris_x_hermes/gateway-plugin/tools/iris_devices.py

#!/usr/bin/env python3
"""Iris device administration (docs/09 §9.3): list / revoke / re-pair devices.

Per-device tokens are minted automatically at pairing (the gateway returns
them in ``hello.ack.device_token``); this tool is the operator's control
surface for the registry under ``<hermes-home>/iris/devices.db``:

  iris_devices.py list                 show paired devices + revoked ids
  iris_devices.py revoke <device_id>   revoke ONE device (its token stops
                                       working AND the shared token no longer
                                       authenticates it; other devices are
                                       unaffected)
  iris_devices.py unrevoke <device_id> allow the device to pair again
  iris_devices.py reissue <device_id>  rotate the device's token (the old
                                       one stops working; the app picks up
                                       the new one on its next (re)connect)

The hermes home is resolved like the gateway: ``HERMES_HOME`` env var, else
``~/.hermes`` (``hermes_constants.get_hermes_home`` when importable, so an
active profile is honored). Run it on the gateway host — the registry is
local state.

Zero dependencies (stdlib only).
"""

from __future__ import annotations

import sys
import time
from pathlib import Path

_USAGE = """\
usage: iris_devices.py <command> [device_id]

commands:
  list                 show paired devices + revoked ids
  revoke <device_id>   revoke ONE device (its token stops working AND the
                       shared token no longer authenticates it; other
                       devices are unaffected)
  unrevoke <device_id> allow the device to pair again
  reissue <device_id>  rotate the device's token (the old one stops working;
                       the app picks up the new one on its next (re)connect)
"""


def _plugin_dir() -> Path:
    return Path(__file__).resolve().parents[1]


def _hermes_home() -> Path:
    import os

    env = os.environ.get("HERMES_HOME", "").strip()
    if env:
        return Path(env)
    try:
        from hermes_constants import get_hermes_home

        return Path(get_hermes_home())
    except ImportError:
        return Path.home() / ".hermes"


def _registry():
    sys.path.insert(0, str(_plugin_dir()))
    from pairing import DeviceRegistry

    return DeviceRegistry(_hermes_home() / "iris" / "devices.db")


def _fmt_ts(ts: float) -> str:
    try:
        return time.strftime("%Y-%m-%d %H:%M", time.localtime(float(ts)))
    except (TypeError, ValueError, OSError):
        return "?"


def cmd_list(reg) -> int:
    devices = reg.list()
    revoked = reg.list_revoked()
    if not devices and not revoked:
        print("No paired devices.")
        return 0
    if devices:
        print(f"{'DEVICE ID':<24} {'NAME':<24} {'TOKEN':<6} {'LAST SEEN':<17} CREATED")
        for d in devices:
            has_token = "yes" if reg.token_for(d["device_id"]) else "no"
            print(
                f"{d['device_id']:<24} {d['name'][:23]:<24} {has_token:<6} "
                f"{_fmt_ts(d['last_seen']):<17} {_fmt_ts(d['created'])}"
            )
    if revoked:
        print("\nRevoked (rejected even with the shared token):")
        for r in revoked:
            print(f"  {r['device_id']}  (revoked {_fmt_ts(r['revoked_at'])})")
    return 0


def cmd_revoke(reg, device_id: str) -> int:
    if not reg.is_revoked(device_id) and reg.get(device_id) is None:
        print(f"unknown device: {device_id}")
        return 1
    reg.revoke(device_id)
    print(f"revoked {device_id} — it can no longer connect (shared token included).")
    print("Re-pairing requires: unrevoke <device_id> (or the app gets a fresh device id).")
    return 0


def cmd_unrevoke(reg, device_id: str) -> int:
    if not reg.is_revoked(device_id):
        print(f"not revoked: {device_id}")
        return 1
    reg.unrevoke(device_id)
    print(f"unrevoked {device_id} — it can pair again (a fresh token is minted).")
    return 0


def cmd_reissue(reg, device_id: str) -> int:
    if reg.get(device_id) is None:
        print(f"unknown device: {device_id}")
        return 1
    reg.reissue_token(device_id)
    print(f"reissued the token for {device_id} — the old one is dead.")
    print("The app picks up the new token on its next (re)connect (hello.ack).")
    return 0


def main(argv: list[str]) -> int:
    args = argv[1:]
    if not args or args[0] in ("-h", "--help", "help"):
        print(_USAGE.strip())
        return 0 if args else 2
    reg = _registry()
    try:
        cmd, rest = args[0], args[1:]
        if cmd == "list":
            return cmd_list(reg)
        if cmd in ("revoke", "unrevoke", "reissue"):
            if not rest or rest[1:]:
                print(f"usage: {Path(sys.argv[0]).name} {cmd} <device_id>")
                return 2
            return {"revoke": cmd_revoke, "unrevoke": cmd_unrevoke, "reissue": cmd_reissue}[cmd](
                reg, rest[0]
            )
        print(f"unknown command: {cmd}")
        print(_USAGE.strip())
        return 2
    finally:
        reg.close()


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