token_service.py

"""Passcode check, LiveKit room tokens, and a login rate limit.

Pure functions, no network: signing a LiveKit token is local HMAC with the API
secret, so this service never talks to the SFU or the Pi. The browser and the
Pi meet at the SFU; this only hands the browser a ticket that (a) joins one
fresh room and (b) asks the SFU to dispatch exactly one agent -- the Pi.
"""
from __future__ import annotations

import base64
import hashlib
import hmac
import secrets
import time
from collections import defaultdict, deque
from dataclasses import dataclass
from datetime import timedelta

from livekit import api

# scrypt cost: ~50 ms on a small CPU. The hash lives in a secret, not a public
# database, so this is defence in depth rather than the only barrier.
_SCRYPT = {"n": 2**14, "r": 8, "p": 1}
TOKEN_TTL_S = 600


def _b64(b: bytes) -> str:
    return base64.urlsafe_b64encode(b).decode().rstrip("=")


def _unb64(s: str) -> bytes:
    return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))


def hash_passcode(passcode: str) -> str:
    """'scrypt$n$r$p$salt$hash' -- stored as the NOVA_PASSCODE_HASH secret."""
    salt = secrets.token_bytes(16)
    digest = hashlib.scrypt(passcode.encode(), salt=salt, dklen=32, **_SCRYPT)
    return f"scrypt${_SCRYPT['n']}${_SCRYPT['r']}${_SCRYPT['p']}${_b64(salt)}${_b64(digest)}"


def verify_passcode(passcode: str, stored: str | None) -> bool:
    """Constant-time check. A malformed stored hash never verifies."""
    try:
        scheme, n, r, p, salt, want = (stored or "").split("$")
        if scheme != "scrypt":
            return False
        want_b = _unb64(want)
        got = hashlib.scrypt(
            passcode.encode(), salt=_unb64(salt), dklen=len(want_b), n=int(n), r=int(r), p=int(p)
        )
    except (ValueError, TypeError):
        return False
    return hmac.compare_digest(got, want_b)


@dataclass(frozen=True)
class Session:
    room: str
    identity: str
    token: str
    expires_in: int


def mint_session(*, api_key: str, api_secret: str, agent_name: str) -> Session:
    room = f"nova-{secrets.token_hex(4)}"
    identity = f"web-{secrets.token_hex(4)}"
    token = (
        api.AccessToken(api_key, api_secret)
        .with_identity(identity)
        .with_ttl(timedelta(seconds=TOKEN_TTL_S))
        .with_grants(
            api.VideoGrants(
                room_join=True,
                room=room,
                can_subscribe=True,
                can_publish=True,
                # a voice session: the mic only -- no camera, no screen share
                can_publish_sources=["microphone"],
                can_publish_data=True,  # nova.control settings messages
            )
        )
        .with_room_config(
            api.RoomConfiguration(agents=[api.RoomAgentDispatch(agent_name=agent_name)])
        )
        .to_jwt()
    )
    return Session(room=room, identity=identity, token=token, expires_in=TOKEN_TTL_S)


class RateLimiter:
    """Sliding window per caller. In memory: one process, a demo-sized budget."""

    def __init__(self, *, max_attempts: int = 5, window_s: float = 300.0) -> None:
        self._max = max_attempts
        self._window = window_s
        self._hits: dict[str, deque[float]] = defaultdict(deque)

    def allow(self, key: str, *, now: float | None = None) -> bool:
        now = time.monotonic() if now is None else now
        hits = self._hits[key]
        while hits and now - hits[0] >= self._window:
            hits.popleft()
        if len(hits) >= self._max:
            return False
        hits.append(now)
        return True