Kareem Horstink
331de4bc9e
- New settings page at /settings (localhost-only) accessible via tray icon - PIN manager with 6-digit codes, 5-min TTL, single-use - API token uses secrets.token_urlsafe(32) for 256-bit entropy - Config auto-creates config.yaml on first run with detected local IP - Default PIN and token generated randomly on first startup - Added pin_ttl setting (default 300s) - Updated build_exe.py for onefile exe with settings assets bundled - Added start.bat and start.sh for easy development startup
142 lines
3.7 KiB
Python
142 lines
3.7 KiB
Python
"""PIN manager for PaperDash.
|
|
|
|
Generates time-limited, single-use PINs for WebSocket authentication.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import secrets
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger("paperdash.pin_manager")
|
|
|
|
|
|
@dataclass
|
|
class PinEntry:
|
|
"""A generated PIN with expiration and usage tracking."""
|
|
|
|
pin: str
|
|
created_at: float
|
|
expires_at: float
|
|
used: bool = False
|
|
|
|
|
|
class PinManager:
|
|
"""Manages time-limited, single-use PINs."""
|
|
|
|
def __init__(self, ttl_seconds: int = 300) -> None:
|
|
"""
|
|
Args:
|
|
ttl_seconds: How long a PIN remains valid (default: 5 minutes)
|
|
"""
|
|
self.ttl_seconds = ttl_seconds
|
|
self.pins: dict[str, PinEntry] = {}
|
|
self._lock = threading.Lock()
|
|
self._cleanup_timer: Optional[threading.Timer] = None
|
|
|
|
def generate(self) -> str:
|
|
"""Generate a new 6-digit PIN.
|
|
|
|
Returns:
|
|
The generated PIN string.
|
|
"""
|
|
pin = secrets.token_hex(3)[:6] # 6 random hex digits
|
|
|
|
now = time.time()
|
|
entry = PinEntry(
|
|
pin=pin,
|
|
created_at=now,
|
|
expires_at=now + self.ttl_seconds,
|
|
used=False,
|
|
)
|
|
|
|
with self._lock:
|
|
self.pins[pin] = entry
|
|
|
|
logger.info(f"Generated PIN {pin} (expires in {self.ttl_seconds}s)")
|
|
return pin
|
|
|
|
def validate(self, pin: str) -> bool:
|
|
"""Validate a PIN (marks it as used if valid).
|
|
|
|
Args:
|
|
pin: The PIN to validate.
|
|
|
|
Returns:
|
|
True if the PIN is valid and not yet used.
|
|
"""
|
|
with self._lock:
|
|
entry = self.pins.get(pin)
|
|
if entry is None:
|
|
return False
|
|
|
|
# Check expiration
|
|
if time.time() > entry.expires_at:
|
|
del self.pins[pin]
|
|
logger.info(f"PIN {pin} expired")
|
|
return False
|
|
|
|
# Check if already used
|
|
if entry.used:
|
|
logger.info(f"PIN {pin} already used")
|
|
return False
|
|
|
|
# Mark as used
|
|
entry.used = True
|
|
return True
|
|
|
|
def cleanup_expired(self) -> int:
|
|
"""Remove expired PINs.
|
|
|
|
Returns:
|
|
Number of PINs removed.
|
|
"""
|
|
now = time.time()
|
|
expired = [pin for pin, entry in self.pins.items() if now > entry.expires_at]
|
|
|
|
with self._lock:
|
|
for pin in expired:
|
|
del self.pins[pin]
|
|
|
|
if expired:
|
|
logger.info(f"Cleaned up {len(expired)} expired PIN(s)")
|
|
|
|
return len(expired)
|
|
|
|
def start_cleanup_loop(self) -> None:
|
|
"""Start periodic cleanup of expired PINs (every 60 seconds)."""
|
|
self._run_cleanup()
|
|
|
|
def _run_cleanup(self) -> None:
|
|
"""Run one cleanup cycle and schedule the next."""
|
|
self.cleanup_expired()
|
|
|
|
with self._lock:
|
|
if self._cleanup_timer is not None:
|
|
self._cleanup_timer.cancel()
|
|
|
|
self._cleanup_timer = threading.Timer(60, self._run_cleanup)
|
|
self._cleanup_timer.daemon = True
|
|
self._cleanup_timer.start()
|
|
|
|
def stop(self) -> None:
|
|
"""Stop the cleanup loop."""
|
|
with self._lock:
|
|
if self._cleanup_timer is not None:
|
|
self._cleanup_timer.cancel()
|
|
self._cleanup_timer = None
|
|
|
|
def get_active_count(self) -> int:
|
|
"""Get the number of active (non-expired, unused) PINs."""
|
|
now = time.time()
|
|
with self._lock:
|
|
return sum(
|
|
1
|
|
for entry in self.pins.values()
|
|
if now <= entry.expires_at and not entry.used
|
|
)
|