Spaces:
Running
Running
File size: 3,585 Bytes
d65a9ae | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | """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
|