shawn
9cd8292acc
Lint & Validate / lint (push) Has been cancelled
P2-R1: ALSA + JACK2 low-latency config (scripts, quirks, tuning) P2-R2: Carla integration (build scripts, 8ch rack config, NAM LV2 support) P2-R3: Plugin manager, categories, blacklist, NAM model support P3-R1: Mixer DSP engine (channel strip, routing matrix, bus mgr, automation) P4-R1: MIDI engine (learn mode, clock sync, HID discovery, mapping store) P4-R2: Network API (OSC server, FastAPI REST, WebSocket, auth, rate limiter) P5-R1: Touchscreen UI evaluation + main entry point docs: Audio stack, Carla integration, MIDI support, UI evaluation tests: Full test suite (292 passing)
200 lines
6.8 KiB
Python
200 lines
6.8 KiB
Python
"""Tests for session-based WebSocket authentication."""
|
|
|
|
import time
|
|
import pytest
|
|
from src.network.session import Session, SessionManager, DEFAULT_SESSION_TTL
|
|
|
|
|
|
class TestSession:
|
|
"""Tests for the Session data object."""
|
|
|
|
def test_session_creation(self):
|
|
s = Session(session_id="abc123", client_id="browser-1")
|
|
assert s.session_id == "abc123"
|
|
assert s.client_id == "browser-1"
|
|
assert not s.expired
|
|
assert s.ttl_remaining > 0
|
|
assert s.created_at > 0
|
|
|
|
def test_session_metadata(self):
|
|
s = Session("abc", "client", metadata={"user": "admin"})
|
|
assert s.metadata["user"] == "admin"
|
|
|
|
def test_session_expiry(self):
|
|
s = Session("abc", "client", ttl=0.01)
|
|
time.sleep(0.02)
|
|
assert s.expired
|
|
assert s.ttl_remaining == 0.0
|
|
|
|
def test_session_not_expired(self):
|
|
s = Session("abc", "client", ttl=3600)
|
|
assert not s.expired
|
|
assert s.ttl_remaining > 3500 # generous margin
|
|
|
|
def test_session_custom_ttl(self):
|
|
s = Session("abc", "client", ttl=60)
|
|
assert 59 < s.ttl_remaining <= 60
|
|
|
|
|
|
class TestSessionManager:
|
|
"""Tests for SessionManager."""
|
|
|
|
@pytest.fixture
|
|
def mgr(self):
|
|
return SessionManager(ttl=3600, max_sessions=10)
|
|
|
|
def test_initial_state(self, mgr):
|
|
assert mgr.active_sessions == 0
|
|
stats = mgr.stats
|
|
assert stats["active_sessions"] == 0
|
|
assert stats["total_created"] == 0
|
|
assert stats["total_destroyed"] == 0
|
|
assert stats["max_sessions"] == 10
|
|
assert stats["default_ttl"] == 3600
|
|
|
|
def test_create_session(self, mgr):
|
|
token = mgr.create_session("browser-chrome")
|
|
assert token is not None
|
|
assert len(token) == 64 # HMAC-SHA256 hex digest
|
|
assert mgr.active_sessions == 1
|
|
assert mgr.stats["total_created"] == 1
|
|
|
|
def test_validate_valid_token(self, mgr):
|
|
token = mgr.create_session("browser-firefox")
|
|
session = mgr.validate_token(token)
|
|
assert session is not None
|
|
assert session.client_id == "browser-firefox"
|
|
assert not session.expired
|
|
|
|
def test_validate_invalid_token(self, mgr):
|
|
session = mgr.validate_token("invalid-token")
|
|
assert session is None
|
|
|
|
def test_validate_empty_token(self, mgr):
|
|
assert mgr.validate_token("") is None
|
|
assert mgr.validate_token(None) is None # type: ignore
|
|
|
|
def test_validate_expired_token(self, mgr):
|
|
token = mgr.create_session("browser", ttl=0.01)
|
|
time.sleep(0.02)
|
|
session = mgr.validate_token(token)
|
|
assert session is None
|
|
assert mgr.active_sessions == 0
|
|
|
|
def test_destroy_session(self, mgr):
|
|
token = mgr.create_session("browser")
|
|
assert mgr.active_sessions == 1
|
|
destroyed = mgr.destroy_session(token)
|
|
assert destroyed
|
|
assert mgr.active_sessions == 0
|
|
assert mgr.validate_token(token) is None
|
|
|
|
def test_destroy_invalid_token(self, mgr):
|
|
assert mgr.destroy_session("nope") is False
|
|
|
|
def test_destroy_by_id(self, mgr):
|
|
token = mgr.create_session("browser")
|
|
session = mgr.validate_token(token)
|
|
assert session is not None
|
|
sid = session.session_id
|
|
destroyed = mgr.destroy_by_id(sid)
|
|
assert destroyed
|
|
assert mgr.validate_token(token) is None
|
|
|
|
def test_destroy_by_invalid_id(self, mgr):
|
|
assert mgr.destroy_by_id("nonexistent") is False
|
|
|
|
def test_get_session(self, mgr):
|
|
token = mgr.create_session("browser")
|
|
session = mgr.get_session(token)
|
|
assert session is not None
|
|
assert session.client_id == "browser"
|
|
|
|
# Even expired sessions are returned by get_session
|
|
session.expires_at = time.time() - 1
|
|
assert mgr.get_session(token) is not None
|
|
|
|
def test_refresh_session(self, mgr):
|
|
token = mgr.create_session("browser", ttl=3600)
|
|
session = mgr.get_session(token)
|
|
original_expiry = session.expires_at
|
|
|
|
# Fast-forward expiry then refresh
|
|
session.expires_at = time.time() + 1 # 1 second left
|
|
assert mgr.refresh_session(token)
|
|
session = mgr.get_session(token)
|
|
assert session.expires_at > original_expiry
|
|
|
|
def test_refresh_expired_session(self, mgr):
|
|
token = mgr.create_session("browser", ttl=0.01)
|
|
time.sleep(0.02)
|
|
assert not mgr.refresh_session(token)
|
|
|
|
def test_multiple_sessions(self, mgr):
|
|
tokens = [mgr.create_session(f"client-{i}") for i in range(5)]
|
|
assert mgr.active_sessions == 5
|
|
assert mgr.stats["total_created"] == 5
|
|
|
|
# Validate all
|
|
for token in tokens:
|
|
assert mgr.validate_token(token) is not None
|
|
|
|
def test_max_sessions(self, mgr):
|
|
"""Test that exceeding max_sessions evicts oldest."""
|
|
# Fill up to max
|
|
for i in range(10):
|
|
mgr.create_session(f"client-{i}")
|
|
|
|
assert mgr.active_sessions == 10
|
|
|
|
# Create one more — should evict oldest
|
|
mgr.create_session("overflow")
|
|
assert mgr.active_sessions == 10
|
|
assert mgr.stats["total_created"] == 11
|
|
assert mgr.stats["total_destroyed"] == 1
|
|
|
|
def test_stale_eviction_on_validate(self, mgr):
|
|
"""Test that stale sessions are evicted when validate is called."""
|
|
token1 = mgr.create_session("client-a", ttl=0.01)
|
|
token2 = mgr.create_session("client-b", ttl=3600)
|
|
|
|
time.sleep(0.02)
|
|
|
|
# Token1 expired, token2 still valid
|
|
assert mgr.validate_token(token1) is None
|
|
assert mgr.active_sessions == 1 # token1 evicted
|
|
assert mgr.validate_token(token2) is not None
|
|
|
|
def test_stats_evicts_stale(self, mgr):
|
|
"""Test that stats evicts stale sessions."""
|
|
mgr.create_session("client-a", ttl=0.01)
|
|
mgr.create_session("client-b", ttl=3600)
|
|
|
|
time.sleep(0.02)
|
|
|
|
stats = mgr.stats
|
|
assert stats["active_sessions"] == 1 # Only client-b remains
|
|
|
|
def test_session_metadata_storage(self, mgr):
|
|
token = mgr.create_session("browser", metadata={"role": "admin", "permissions": ["read", "write"]})
|
|
session = mgr.validate_token(token)
|
|
assert session.metadata == {"role": "admin", "permissions": ["read", "write"]}
|
|
|
|
|
|
class TestSessionTokens:
|
|
"""Tests for token uniqueness and security properties."""
|
|
|
|
def test_tokens_are_unique(self):
|
|
mgr = SessionManager()
|
|
tokens = {mgr.create_session("client") for _ in range(100)}
|
|
assert len(tokens) == 100 # All unique
|
|
|
|
def test_tokens_are_opaque(self):
|
|
"""Tokens should not contain the session ID in plaintext."""
|
|
mgr = SessionManager()
|
|
token = mgr.create_session("client")
|
|
session = mgr.get_session(token)
|
|
assert session is not None
|
|
assert session.session_id not in token
|
|
assert "client" not in token
|