iris_x_hermes/gateway-plugin/pairing.py

"""Pairing: token generation/verification + device registry.

Token generation (64-hex) and constant-time verification. Device registry
(SQLite) tracks ``device_id``, name, caps, fcm_token, ntfy_topic, last_seen,
created. QR payload for the pairing flow (``interactive_setup``).

Storage: ``get_hermes_home()/"android"/devices.db``.

Milestone M1.
"""

import hmac
import json
import logging
import secrets
import sqlite3
import threading
import time
from pathlib import Path
from typing import Any, Dict, List, Optional
from urllib.parse import quote

logger = logging.getLogger(__name__)

# 32 random bytes -> 64 hex chars (docs/09-pairing-security.md)
TOKEN_BYTES = 32


def generate_token() -> str:
    """Mint a fresh high-entropy pairing token (64 hex chars)."""
    return secrets.token_hex(TOKEN_BYTES)


def verify_token(provided: Optional[str], expected: Optional[str]) -> bool:
    """Constant-time token comparison (never time-leaks the token)."""
    if not provided or not expected:
        return False
    return hmac.compare_digest(
        provided.encode("utf-8", "replace"),
        expected.encode("utf-8", "replace"),
    )


def qr_payload(host: str, port: int, token: str, secure: bool = False) -> str:
    """Pairing URL encoded into the QR / pre-filled into the app.

    ``iris://pair?host=<lan-ip>&port=8790&token=<token>`` — the app's
    Connect screen parses this to pre-fill settings (docs/09 §9.2).
    """
    return (
        f"iris://pair?host={quote(host, safe='')}"
        f"&port={int(port)}"
        f"&secure={'1' if secure else '0'}"
        f"&token={quote(token, safe='')}"
    )


def pairing_url(host: str, port: int, secure: bool = False) -> str:
    """Plain ws(s) URL the app connects to (shown next to the QR)."""
    scheme = "wss" if secure else "ws"
    return f"{scheme}://{host}:{int(port)}/ws"


# ---------------------------------------------------------------------------
# Device registry (SQLite)
# ---------------------------------------------------------------------------

class DeviceRegistry:
    """Persistent device registry under ``get_hermes_home()/"android"``.

    Thread-safe (single connection + lock); all operations are small and
    fast enough to run inline on the gateway's asyncio loop.
    """

    def __init__(self, db_path: Path):
        self._db_path = Path(db_path)
        self._db_path.parent.mkdir(parents=True, exist_ok=True)
        self._lock = threading.Lock()
        self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False)
        self._conn.row_factory = sqlite3.Row
        with self._lock:
            self._conn.execute("PRAGMA journal_mode=WAL")
            self._conn.execute(
                """
                CREATE TABLE IF NOT EXISTS devices (
                    device_id  TEXT PRIMARY KEY,
                    name       TEXT NOT NULL DEFAULT '',
                    caps       TEXT NOT NULL DEFAULT '{}',
                    fcm_token  TEXT,
                    ntfy_topic TEXT,
                    last_seen  REAL NOT NULL DEFAULT 0,
                    created    REAL NOT NULL DEFAULT 0
                )
                """
            )
            self._conn.commit()

    def upsert(
        self,
        device_id: str,
        name: str,
        caps: Optional[Dict[str, Any]] = None,
        fcm_token: Optional[str] = None,
        ntfy_topic: Optional[str] = None,
    ) -> None:
        now = time.time()
        caps_json = json.dumps(caps or {}, separators=(",", ":"))
        with self._lock:
            self._conn.execute(
                """
                INSERT INTO devices (device_id, name, caps, fcm_token, ntfy_topic,
                                     last_seen, created)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(device_id) DO UPDATE SET
                    name = excluded.name,
                    caps = excluded.caps,
                    fcm_token = COALESCE(excluded.fcm_token, devices.fcm_token),
                    ntfy_topic = COALESCE(excluded.ntfy_topic, devices.ntfy_topic),
                    last_seen = excluded.last_seen
                """,
                (device_id, name or "", caps_json, fcm_token, ntfy_topic, now, now),
            )
            self._conn.commit()

    def update_push_tokens(
        self,
        device_id: str,
        fcm_token: Optional[str] = None,
        ntfy_topic: Optional[str] = None,
    ) -> None:
        with self._lock:
            self._conn.execute(
                """
                UPDATE devices SET
                    fcm_token = COALESCE(?, fcm_token),
                    ntfy_topic = COALESCE(?, ntfy_topic),
                    last_seen = ?
                WHERE device_id = ?
                """,
                (fcm_token, ntfy_topic, time.time(), device_id),
            )
            self._conn.commit()

    def touch(self, device_id: str) -> None:
        with self._lock:
            self._conn.execute(
                "UPDATE devices SET last_seen = ? WHERE device_id = ?",
                (time.time(), device_id),
            )
            self._conn.commit()

    def get(self, device_id: str) -> Optional[Dict[str, Any]]:
        with self._lock:
            row = self._conn.execute(
                "SELECT * FROM devices WHERE device_id = ?", (device_id,)
            ).fetchone()
        return _row_to_device(row) if row else None

    def list(self) -> List[Dict[str, Any]]:
        with self._lock:
            rows = self._conn.execute(
                "SELECT * FROM devices ORDER BY last_seen DESC"
            ).fetchall()
        return [_row_to_device(r) for r in rows]

    def close(self) -> None:
        with self._lock:
            try:
                self._conn.close()
            except Exception:
                pass


def _row_to_device(row: sqlite3.Row) -> Dict[str, Any]:
    try:
        caps = json.loads(row["caps"] or "{}")
        if not isinstance(caps, dict):
            caps = {}
    except (json.JSONDecodeError, TypeError):
        caps = {}
    return {
        "device_id": row["device_id"],
        "name": row["name"],
        "caps": caps,
        "fcm_token": row["fcm_token"],
        "ntfy_topic": row["ntfy_topic"],
        "last_seen": row["last_seen"],
        "created": row["created"],
    }