89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
"""Authentication fails closed and validates users against Supabase."""
|
|
import asyncio
|
|
import io
|
|
import json
|
|
from unittest.mock import AsyncMock
|
|
import urllib.error
|
|
|
|
import pytest
|
|
import socketio
|
|
|
|
from defense_sim.server.auth import AuthenticationError, verify_token
|
|
from defense_sim.server import socketio_server as server
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_environment(monkeypatch):
|
|
monkeypatch.setenv('SUPABASE_URL', 'https://auth.example.test')
|
|
monkeypatch.setenv('SUPABASE_ANON_KEY', 'public-key')
|
|
|
|
|
|
def test_missing_auth_configuration_rejects_connections(monkeypatch):
|
|
monkeypatch.delenv('SUPABASE_URL', raising=False)
|
|
monkeypatch.delenv('SUPABASE_ANON_KEY', raising=False)
|
|
with pytest.raises(AuthenticationError, match='not configured'):
|
|
verify_token('token')
|
|
|
|
|
|
def test_missing_token_does_not_contact_auth_server(auth_environment, monkeypatch):
|
|
def unexpected_request(*args, **kwargs):
|
|
pytest.fail('Missing tokens must be rejected locally')
|
|
monkeypatch.setattr('urllib.request.urlopen', unexpected_request)
|
|
with pytest.raises(AuthenticationError, match='Sign in'):
|
|
verify_token('')
|
|
|
|
|
|
def test_token_is_verified_with_auth_server(auth_environment, monkeypatch):
|
|
def response(request, timeout):
|
|
assert request.full_url == 'https://auth.example.test/auth/v1/user'
|
|
assert request.get_header('Authorization') == 'Bearer signed-token'
|
|
assert request.get_header('Apikey') == 'public-key'
|
|
return io.BytesIO(json.dumps({'id': 'verified-user', 'is_anonymous': False}).encode())
|
|
monkeypatch.setattr('urllib.request.urlopen', response)
|
|
assert verify_token('signed-token') == 'verified-user'
|
|
|
|
|
|
@pytest.mark.parametrize('payload', [{'id': 'guest', 'is_anonymous': True}, {}, [], {'id': None}])
|
|
def test_anonymous_or_invalid_users_rejected(auth_environment, monkeypatch, payload):
|
|
monkeypatch.setattr('urllib.request.urlopen', lambda *args, **kwargs: io.BytesIO(json.dumps(payload).encode()))
|
|
with pytest.raises(AuthenticationError):
|
|
verify_token('token')
|
|
|
|
|
|
def test_expired_token_rejected(auth_environment, monkeypatch):
|
|
def response(*args, **kwargs):
|
|
raise urllib.error.HTTPError('https://auth.example.test', 401, 'Expired token', {}, None)
|
|
monkeypatch.setattr('urllib.request.urlopen', response)
|
|
with pytest.raises(AuthenticationError):
|
|
verify_token('expired-token')
|
|
|
|
|
|
def test_socket_rejects_unauthenticated_client_before_sending_state(monkeypatch):
|
|
def rejected(token):
|
|
raise AuthenticationError('Sign in')
|
|
monkeypatch.setattr(server, 'verify_token', rejected)
|
|
emit = AsyncMock()
|
|
monkeypatch.setattr(server.sio, 'emit', emit)
|
|
with pytest.raises(socketio.exceptions.ConnectionRefusedError):
|
|
asyncio.run(server.connect('client', {}, None))
|
|
emit.assert_not_called()
|
|
|
|
|
|
def test_expired_session_cannot_stop_running_simulation(monkeypatch):
|
|
async def rejected(sid, token=None):
|
|
return False
|
|
running_task = AsyncMock()
|
|
monkeypatch.setattr(server, 'authenticated', rejected)
|
|
monkeypatch.setattr(server, 'task', running_task)
|
|
asyncio.run(server.stop_simulation('client'))
|
|
running_task.cancel.assert_not_called()
|
|
|
|
|
|
def test_expired_session_cannot_start_simulation(monkeypatch):
|
|
async def rejected(sid, token=None):
|
|
return False
|
|
monkeypatch.setattr(server, 'authenticated', rejected)
|
|
original_engine = server.engine
|
|
asyncio.run(server.start_simulation('client', {}))
|
|
assert server.engine is original_engine
|