Xiaozhi WebSocket endpoint with handshake/auth, mock STT/TTS/Hermes backends, Thai chunker, barge-in queue, latency logger, GPU planner, voice-profile guard (reasoning=none, session_search only). CON-002: no GPU/audio libs loaded at import. MUST-NOT-001: all model names/tokens from config.
37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
"""Device authentication (REQ-009).
|
|
|
|
Handshake carries Device-Id + Authorization. A device is allowed only if its
|
|
device_id is on the allowlist AND its token hashes match. No device = no
|
|
session. This is what keeps the bridge the *only* internet-exposed surface
|
|
(MUST-NOT-002): everything behind it is LAN/localhost.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
|
|
|
|
class AuthError(Exception):
|
|
pass
|
|
|
|
|
|
class DeviceAuth:
|
|
def __init__(self, devices: dict[str, str], *, enabled: bool = True) -> None:
|
|
# device_id -> token_hash
|
|
self._devices = devices
|
|
self.enabled = enabled
|
|
|
|
@staticmethod
|
|
def hash_token(token: str) -> str:
|
|
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
|
|
|
def authenticate(self, device_id: str, token: str) -> bool:
|
|
if not self.enabled:
|
|
return True
|
|
expected = self._devices.get(device_id)
|
|
if expected is None:
|
|
raise AuthError(f"device not allowed: {device_id}")
|
|
got = self.hash_token(token)
|
|
if got != expected:
|
|
raise AuthError(f"bad token for device: {device_id}")
|
|
return True
|