286 lines
15 KiB
Python
286 lines
15 KiB
Python
"""Offline control-plane tests: no Docker, provider, LAN, or deployment access."""
|
|
import asyncio
|
|
import sqlite3
|
|
import time
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from control.accounts import Accounts, password_matches, public_user
|
|
from control.app import Settings, create_app
|
|
|
|
PASSWORD = 'correct-horse-password'
|
|
SECRET = 'internal-secret-' + 'x' * 32
|
|
|
|
|
|
class FakeProvisioner:
|
|
def __init__(self):
|
|
self.ensured = []
|
|
self.removed = []
|
|
self.fail_ensure = False
|
|
self.fail_remove = False
|
|
self.healthy = True
|
|
|
|
async def ensure(self, user):
|
|
self.ensured.append(user['id'])
|
|
if self.fail_ensure:
|
|
raise RuntimeError('sensitive-provider-key-must-not-leak')
|
|
return {'container_name': user['container_name'], 'hermes_url': 'http://127.0.0.1:1234', 'hermes_api_key': 'private-hermes-key'}
|
|
|
|
async def ready(self, user):
|
|
return self.healthy
|
|
|
|
async def remove(self, user):
|
|
self.removed.append(user['id'])
|
|
if self.fail_remove:
|
|
raise RuntimeError('container ownership verification failed')
|
|
|
|
|
|
@pytest.fixture
|
|
def environment(tmp_path):
|
|
fake = FakeProvisioner()
|
|
settings = Settings(db_path=str(tmp_path / 'accounts.db'), admin_password=PASSWORD, service_secret=SECRET, allowed_origins=('https://testserver',), reconcile_interval=3600)
|
|
app = create_app(settings, fake)
|
|
with TestClient(app, base_url='https://testserver') as client:
|
|
yield client, app, fake, settings
|
|
|
|
|
|
def login(client, username='admin', password=PASSWORD):
|
|
result = client.post('/api/auth/login', headers={'Origin': 'https://testserver'}, json={'username': username, 'password': password})
|
|
assert result.status_code == 200, result.text
|
|
return result.json()
|
|
|
|
|
|
def auth(result):
|
|
return {'Authorization': 'Bearer ' + result['token']}
|
|
|
|
|
|
def wait_state(client, headers, user_id, status):
|
|
# Requests yield to the ASGI lifecycle loop; bounded wait, no network.
|
|
deadline = time.monotonic() + 3
|
|
while time.monotonic() < deadline:
|
|
users = client.get('/api/admin/users', headers=headers).json()['users']
|
|
found = next((u for u in users if u['id'] == user_id), None)
|
|
if (status is None and found is None) or (found and found['status'] == status):
|
|
return found
|
|
raise AssertionError(f'Account did not reach {status}: {found}')
|
|
|
|
|
|
def create(client, headers, username='john-doe'):
|
|
response = client.post('/api/admin/users', headers=headers, json={'username': username, 'display_name': 'John Doe', 'password': PASSWORD})
|
|
assert response.status_code == 201, response.text
|
|
return response.json()['user']
|
|
|
|
|
|
def test_bootstrap_required(tmp_path):
|
|
app = create_app(Settings(db_path=str(tmp_path / 'missing.db'), admin_password=None), FakeProvisioner())
|
|
with pytest.raises(RuntimeError, match='CONTROL_ADMIN_PASSWORD'):
|
|
with TestClient(app):
|
|
pass
|
|
|
|
|
|
def test_password_sessions_cookie_schema(environment):
|
|
client, app, fake, settings = environment
|
|
assert client.get('/api/auth/me').status_code == 401
|
|
session = login(client)
|
|
assert set(session['user']) == {'id', 'username', 'display_name', 'role', 'status', 'container_name', 'error'}
|
|
assert len(session['user']['id']) == 32
|
|
assert session['expires_at'] > time.time()
|
|
cookie = client.cookies.get('hope_session')
|
|
assert cookie == session['token']
|
|
raw = app.state.accounts.get(session['user']['id'])
|
|
assert raw['password_hash'] != PASSWORD
|
|
assert password_matches(PASSWORD, raw['password_hash'])
|
|
row = app.state.accounts.db.execute('SELECT hash FROM sessions').fetchone()
|
|
assert row['hash'] != session['token']
|
|
assert client.get('/api/auth/me').status_code == 200
|
|
assert client.post('/api/auth/logout').status_code == 403
|
|
assert client.post('/api/auth/logout', headers={'Origin': 'https://evil.example'}).status_code == 403
|
|
assert client.post('/api/auth/logout', headers=auth(session)).status_code == 200
|
|
assert client.get('/api/auth/me', headers=auth(session)).status_code == 401
|
|
again = client.post('/api/auth/login', headers={'Origin': 'https://testserver'}, json={'username': 'admin', 'password': PASSWORD})
|
|
cookie_header = again.headers['set-cookie'].lower()
|
|
assert all(x in cookie_header for x in ('httponly', 'secure', 'samesite=strict'))
|
|
assert client.post('/api/auth/logout', headers={'Origin': 'https://testserver'}).status_code == 200
|
|
|
|
|
|
def test_create_auth_reset_and_validation(environment):
|
|
client, app, fake, settings = environment
|
|
admin = auth(login(client))
|
|
user = create(client, admin)
|
|
ready = wait_state(client, admin, user['id'], 'ready')
|
|
assert ready['container_name'] == f"hope-hermes-john-doe-{user['id'][:8]}"
|
|
assert client.post('/api/admin/users', headers=admin, json={'username': 'Bad/Name', 'display_name': 'Bad', 'password': PASSWORD}).status_code == 400
|
|
assert client.post('/api/admin/users', headers=admin, json={'username': 'john-doe', 'display_name': 'John', 'password': PASSWORD}).status_code == 409
|
|
ordinary = auth(login(client, 'john-doe'))
|
|
assert client.get('/api/admin/users', headers=ordinary).status_code == 403
|
|
assert client.get('/internal/users/' + user['id'], headers=ordinary).status_code == 401
|
|
assert client.post('/api/admin/users/' + user['id'] + '/password', headers=admin, json={'password': 'new-safe-password'}).status_code == 200
|
|
assert client.get('/api/auth/me', headers=ordinary).status_code == 401
|
|
assert client.post('/api/auth/login', headers={'Origin': 'https://testserver'}, json={'username': 'john-doe', 'password': PASSWORD}).status_code == 401
|
|
login(client, 'john-doe', 'new-safe-password')
|
|
|
|
|
|
def test_last_admin_and_delete_revokes_sessions_rooms(environment):
|
|
client, app, fake, settings = environment
|
|
session = login(client)
|
|
admin = auth(session)
|
|
assert client.delete('/api/admin/users/' + session['user']['id'], headers=admin).status_code == 409
|
|
user = create(client, admin)
|
|
wait_state(client, admin, user['id'], 'ready')
|
|
ordinary = login(client, 'john-doe')
|
|
internal = {'X-Control-Secret': SECRET}
|
|
resolve = client.post('/internal/resolve-session', headers=internal, json={'token': ordinary['token']})
|
|
assert resolve.status_code == 200
|
|
assert resolve.json()['hermes_api_key'] == 'private-hermes-key'
|
|
assert 'password_hash' not in resolve.text
|
|
assert client.post('/internal/rooms', headers=internal, json={'user_id': user['id'], 'room_name': 'voice-test', 'session_token': ordinary['token']}).status_code == 201
|
|
assert client.get('/internal/rooms/voice-test', headers=internal).status_code == 200
|
|
fake.fail_remove = True
|
|
assert client.delete('/api/admin/users/' + user['id'], headers=admin).status_code == 202
|
|
assert client.get('/api/auth/me', headers=auth(ordinary)).status_code == 401
|
|
assert client.get('/internal/rooms/voice-test', headers=internal).status_code == 404
|
|
assert client.post('/internal/resolve-session', headers=internal, json={'token': ordinary['token']}).status_code == 401
|
|
wait_state(client, admin, user['id'], 'deleting')
|
|
fake.fail_remove = False
|
|
# Wait for failed first operation to finish before retry scheduling.
|
|
async def settled():
|
|
await asyncio.gather(*app.state.lifecycle.tasks.values())
|
|
client.portal.call(settled)
|
|
client.post('/api/admin/users/' + user['id'] + '/retry', headers=admin)
|
|
wait_state(client, admin, user['id'], None)
|
|
assert user['id'] in fake.removed
|
|
|
|
|
|
def test_failure_visible_and_restart_reconcile(tmp_path):
|
|
fake = FakeProvisioner()
|
|
fake.fail_ensure = True
|
|
settings = Settings(db_path=str(tmp_path / 'restart.db'), admin_password=PASSWORD, service_secret=SECRET, reconcile_interval=3600)
|
|
with TestClient(create_app(settings, fake), base_url='https://testserver') as client:
|
|
session = login(client)
|
|
header = auth(session)
|
|
user = wait_state(client, header, session['user']['id'], 'error')
|
|
assert 'sensitive' not in user['error']
|
|
fake.fail_ensure = False
|
|
settings.admin_password = None
|
|
with TestClient(create_app(settings, fake), base_url='https://testserver') as client:
|
|
wait_state(client, header, session['user']['id'], 'ready')
|
|
assert client.get('/api/auth/me', headers=header).status_code == 200
|
|
|
|
|
|
def test_expiry_throttle_and_bounded_input(environment):
|
|
client, app, fake, settings = environment
|
|
session = login(client)
|
|
with app.state.accounts.lock, app.state.accounts.db:
|
|
app.state.accounts.db.execute('UPDATE sessions SET expires_at=0')
|
|
assert client.get('/api/auth/me', headers=auth(session)).status_code == 401
|
|
client.cookies.clear()
|
|
assert client.post('/api/auth/login', json={'username': 'admin', 'password': 'p' * 257}).status_code == 400
|
|
assert client.post('/api/auth/login', content='x' * 9000, headers={'Content-Type': 'application/json'}).status_code == 413
|
|
assert client.post('/api/auth/login', content='{}', headers={'Content-Type': 'text/plain'}).status_code == 415
|
|
client.cookies.clear()
|
|
for _ in range(settings.login_limit):
|
|
result = client.post('/api/auth/login', json={'username': 'missing', 'password': PASSWORD})
|
|
assert result.status_code == 401
|
|
assert client.post('/api/auth/login', json={'username': 'missing', 'password': PASSWORD}).status_code == 429
|
|
|
|
|
|
def test_internal_fail_closed(environment):
|
|
client, app, fake, settings = environment
|
|
assert client.post('/internal/resolve-session', json={'token': 'x' * 32}).status_code == 401
|
|
assert client.get('/internal/users/' + 'a' * 32, headers={'X-Control-Secret': 'wrong'}).status_code == 401
|
|
settings.service_secret = None
|
|
assert client.get('/internal/rooms/voice-test', headers={'X-Control-Secret': SECRET}).status_code == 401
|
|
|
|
|
|
def test_room_session_binding_and_logout(environment):
|
|
client, app, fake, settings = environment
|
|
admin_session = login(client)
|
|
header = auth(admin_session)
|
|
wait_state(client, header, admin_session['user']['id'], 'ready')
|
|
user = create(client, header, 'alice')
|
|
wait_state(client, header, user['id'], 'ready')
|
|
session = login(client, 'alice')
|
|
internal = {'X-Control-Secret': SECRET}
|
|
mismatch = {'user_id': user['id'], 'room_name': 'wrong', 'session_token': admin_session['token']}
|
|
assert client.post('/internal/rooms', headers=internal, json=mismatch).status_code == 409
|
|
grant = {'user_id': user['id'], 'room_name': 'voice-alice', 'session_token': session['token']}
|
|
assert client.post('/internal/rooms', headers=internal, json=grant).status_code == 201
|
|
assert client.get('/internal/rooms/voice-alice', headers=internal).status_code == 200
|
|
assert client.post('/api/auth/logout', headers=auth(session)).status_code == 200
|
|
assert client.get('/internal/rooms/voice-alice', headers=internal).status_code == 404
|
|
|
|
|
|
def test_failed_removal_restart_retries(tmp_path):
|
|
path = str(tmp_path / 'removal.db')
|
|
store = Accounts(path)
|
|
store.create('admin', 'Administrator', 'admin', PASSWORD)
|
|
user = store.create('alice', 'Alice', 'user', PASSWORD)
|
|
store.begin_delete(user['id'])
|
|
store.close()
|
|
settings = Settings(db_path=path, admin_password=None, service_secret=SECRET, reconcile_interval=3600)
|
|
fake = FakeProvisioner()
|
|
fake.fail_remove = True
|
|
with TestClient(create_app(settings, fake), base_url='https://testserver') as client:
|
|
header = auth(login(client))
|
|
wait_state(client, header, user['id'], 'deleting')
|
|
assert client.post('/api/auth/login', headers={'Origin': 'https://testserver'}, json={'username': 'alice', 'password': PASSWORD}).status_code == 401
|
|
fake.fail_remove = False
|
|
with TestClient(create_app(settings, fake), base_url='https://testserver') as client:
|
|
header = auth(login(client))
|
|
wait_state(client, header, user['id'], None)
|
|
assert fake.removed.count(user['id']) >= 2
|
|
|
|
|
|
def test_reset_between_password_verification_and_session_issue(environment, monkeypatch):
|
|
client, app, fake, settings = environment
|
|
original = app.state.accounts.issue_session
|
|
|
|
def reset_then_issue(user_id, ttl, expected_hash):
|
|
app.state.accounts.reset_password(user_id, 'changed-safe-password')
|
|
return original(user_id, ttl, expected_hash)
|
|
|
|
monkeypatch.setattr(app.state.accounts, 'issue_session', reset_then_issue)
|
|
response = client.post('/api/auth/login', json={'username': 'admin', 'password': PASSWORD})
|
|
assert response.status_code == 401
|
|
assert app.state.accounts.db.execute('SELECT count(*) FROM sessions').fetchone()[0] == 0
|
|
monkeypatch.setattr(app.state.accounts, 'issue_session', original)
|
|
login(client, password='changed-safe-password')
|
|
|
|
|
|
def test_successes_and_other_accounts_do_not_share_failure_budget(environment):
|
|
client, app, fake, settings = environment
|
|
for _ in range(settings.login_limit + 2):
|
|
login(client)
|
|
for _ in range(settings.login_limit):
|
|
assert client.post('/api/auth/login', headers={'Origin': 'https://testserver'}, json={'username': 'missing', 'password': PASSWORD}).status_code == 401
|
|
assert client.post('/api/auth/login', headers={'Origin': 'https://testserver'}, json={'username': 'missing', 'password': PASSWORD}).status_code == 429
|
|
login(client) # Same IP, different account remains usable.
|
|
|
|
|
|
@pytest.mark.parametrize('peer,trusted', [('127.0.0.1', True), ('::1', True), ('203.0.113.10', False)])
|
|
def test_real_ip_only_trusted_from_loopback(tmp_path, peer, trusted):
|
|
settings = Settings(db_path=str(tmp_path / 'proxy.db'), admin_password=PASSWORD, login_limit=2)
|
|
with TestClient(create_app(settings, FakeProvisioner()), base_url='https://testserver', client=(peer, 1234)) as client:
|
|
body = {'username': 'missing', 'password': PASSWORD}
|
|
for _ in range(settings.login_limit):
|
|
assert client.post('/api/auth/login', headers={'X-Real-IP': '198.51.100.1'}, json=body).status_code == 401
|
|
assert client.post('/api/auth/login', headers={'X-Real-IP': '198.51.100.1'}, json=body).status_code == 429
|
|
other = client.post('/api/auth/login', headers={'X-Real-IP': '198.51.100.2', 'X-Forwarded-For': '198.51.100.3'}, json=body)
|
|
assert other.status_code == (401 if trusted else 429)
|
|
|
|
|
|
def test_database_delete_intent_persists(tmp_path):
|
|
db = str(tmp_path / 'durable.db')
|
|
store = Accounts(db)
|
|
admin = store.create('admin', 'Administrator', 'admin', PASSWORD)
|
|
user = store.create('alice', 'Alice', 'user', PASSWORD)
|
|
token, expiry = store.issue_session(user['id'], 60, user['password_hash'])
|
|
store.begin_delete(user['id'])
|
|
store.close()
|
|
store = Accounts(db)
|
|
assert store.get(user['id'])['desired'] == 'deleted'
|
|
assert store.resolve_session(token) is None
|
|
assert store.get(admin['id'])['role'] == 'admin'
|
|
store.close()
|