mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-10-28 07:37:56 +09:00
Juan Denis
d67e5d07d0
Centralize backup schedule calculations and shared request/form flows, encode query values, and move repository finalization into its service. Add browser CI coverage, repair stale test fixtures, and provide opt-in request and build measurement tools.
242 lines
11 KiB
Python
242 lines
11 KiB
Python
"""Regression coverage for cross-authentication credential and revocation chains."""
|
|
from datetime import timedelta
|
|
|
|
import pyotp
|
|
import pytest
|
|
from flask_jwt_extended import decode_token
|
|
|
|
from factories import make_user, headers_for, access_token_for
|
|
from app.services.api_key_service import ApiKeyService
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolated_throttles(app):
|
|
from app import limiter
|
|
limiter.reset()
|
|
yield
|
|
limiter.reset()
|
|
|
|
|
|
def _headers(token):
|
|
return {'Authorization': f'Bearer {token}'}
|
|
|
|
|
|
def _login(client, user):
|
|
response = client.post('/api/v1/auth/login', json={
|
|
'email': user.email, 'password': 'password123',
|
|
})
|
|
assert response.status_code == 200, response.get_json()
|
|
return response.get_json()
|
|
|
|
|
|
def _refresh(client, token):
|
|
return client.post('/api/v1/auth/refresh', headers=_headers(token))
|
|
|
|
|
|
def test_pending_mfa_cannot_mint_link_and_has_five_minute_expiry(client, db_session):
|
|
user = make_user(db_session, role='admin', password='password123',
|
|
totp_enabled=True, totp_secret=pyotp.random_base32())
|
|
pending = _login(client, user)
|
|
claims = decode_token(pending['temp_token'])
|
|
assert 0 < claims['exp'] - claims['iat'] <= 300
|
|
for path in ('/api/v1/auth/login-links', '/api/v1/auth/logout'):
|
|
response = client.post(path, headers=_headers(pending['temp_token']), json={})
|
|
assert response.status_code in (401, 403)
|
|
assert client.get('/api/v1/auth/me', headers=_headers(pending['temp_token'])).status_code == 401
|
|
|
|
verified = client.post('/api/v1/auth/2fa/verify', json={
|
|
'temp_token': pending['temp_token'], 'code': pyotp.TOTP(user.totp_secret).now(),
|
|
})
|
|
assert verified.status_code == 200
|
|
assert client.get('/api/v1/auth/me', headers=_headers(verified.json['access_token'])).status_code == 200
|
|
|
|
|
|
@pytest.mark.parametrize('scopes', [['apps:read'], ['*']])
|
|
def test_api_keys_cannot_mint_browser_sessions(client, db_session, scopes):
|
|
user = make_user(db_session, role='admin')
|
|
_, raw_key = ApiKeyService.create_key(user.id, name='session-probe', scopes=scopes)
|
|
# Even combining a valid JWT with a key must not silently ignore the key.
|
|
for headers in ({'X-API-Key': raw_key}, {**headers_for(user), 'X-API-Key': raw_key}):
|
|
response = client.post('/api/v1/auth/login-links', json={}, headers=headers)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_login_link_redemption_respects_target_mfa(client, db_session):
|
|
admin = make_user(db_session, role='admin')
|
|
target = make_user(db_session, totp_enabled=True, totp_secret=pyotp.random_base32())
|
|
created = client.post('/api/v1/auth/login-links', headers=headers_for(admin),
|
|
json={'user_id': target.id})
|
|
assert created.status_code == 201
|
|
redeemed = client.post('/api/v1/auth/login-links/redeem', json={'token': created.json['token']})
|
|
assert redeemed.status_code == 200
|
|
assert redeemed.json['requires_2fa'] is True
|
|
assert 'access_token' not in redeemed.json and 'refresh_token' not in redeemed.json
|
|
result = client.post('/api/v1/auth/2fa/verify', json={
|
|
'temp_token': redeemed.json['temp_token'], 'code': pyotp.TOTP(target.totp_secret).now(),
|
|
})
|
|
assert result.status_code == 200 and result.json['user']['id'] == target.id
|
|
|
|
|
|
def test_password_change_requires_current_password_and_revokes_previous_tokens(client, db_session):
|
|
user = make_user(db_session, password='password123')
|
|
original = _login(client, user)
|
|
headers = _headers(original['access_token'])
|
|
for body in ({'password': 'replacement123'},
|
|
{'password': 'replacement123', 'current_password': 'incorrect'}):
|
|
assert client.put('/api/v1/auth/me', headers=headers, json=body).status_code == 403
|
|
updated = client.put('/api/v1/auth/me', headers=headers, json={
|
|
'password': 'replacement123', 'current_password': 'password123',
|
|
})
|
|
assert updated.status_code == 200
|
|
assert _refresh(client, original['refresh_token']).status_code == 401
|
|
assert client.get('/api/v1/auth/me', headers=headers).status_code == 401
|
|
assert client.get('/api/v1/auth/me', headers=_headers(updated.json['access_token'])).status_code == 200
|
|
assert _refresh(client, updated.json['refresh_token']).status_code == 200
|
|
|
|
|
|
def test_disable_reenable_and_admin_password_reset_revoke_existing_tokens(client, db_session):
|
|
user = make_user(db_session, password='password123')
|
|
tokens = _login(client, user)
|
|
user.is_active = False
|
|
db_session.session.commit()
|
|
assert client.get('/api/v1/auth/me', headers=_headers(tokens['access_token'])).status_code == 401
|
|
user.is_active = True
|
|
db_session.session.commit()
|
|
assert _refresh(client, tokens['refresh_token']).status_code == 401
|
|
fresh = _login(client, user)
|
|
admin = make_user(db_session, role='admin')
|
|
response = client.put(f'/api/v1/admin/users/{user.id}', headers=headers_for(admin),
|
|
json={'password': 'adminreset123'})
|
|
assert response.status_code == 200
|
|
assert _refresh(client, fresh['refresh_token']).status_code == 401
|
|
|
|
|
|
def test_logout_revokes_entire_browser_family_but_preserves_other_browser(client, db_session):
|
|
user = make_user(db_session, role='viewer', password='password123')
|
|
first = _login(client, user)
|
|
second = _login(client, user)
|
|
refreshed = _refresh(client, first['refresh_token'])
|
|
assert refreshed.status_code == 200
|
|
assert decode_token(first['access_token'])['session_id'] == decode_token(first['refresh_token'])['session_id']
|
|
assert client.post('/api/v1/auth/logout', headers=_headers(first['access_token'])).status_code == 200
|
|
for token in (first['access_token'], refreshed.json['access_token']):
|
|
assert client.get('/api/v1/auth/me', headers=_headers(token)).status_code == 401
|
|
assert _refresh(client, first['refresh_token']).status_code == 401
|
|
assert client.get('/api/v1/auth/me', headers=_headers(second['access_token'])).status_code == 200
|
|
assert _refresh(client, second['refresh_token']).status_code == 200
|
|
|
|
|
|
@pytest.mark.parametrize('change', ['disable', 'password'])
|
|
def test_revoked_pending_mfa_cannot_finish_login(client, db_session, change):
|
|
user = make_user(db_session, password='password123', totp_enabled=True,
|
|
totp_secret=pyotp.random_base32())
|
|
pending = _login(client, user)
|
|
if change == 'disable':
|
|
user.is_active = False
|
|
else:
|
|
user.set_password('resetpassword123')
|
|
db_session.session.commit()
|
|
response = client.post('/api/v1/auth/2fa/verify', json={
|
|
'temp_token': pending['temp_token'], 'code': pyotp.TOTP(user.totp_secret).now(),
|
|
})
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_unexpired_legacy_or_expired_mfa_tokens_are_rejected(client, db_session):
|
|
user = make_user(db_session, totp_enabled=True, totp_secret=pyotp.random_base32())
|
|
legacy = access_token_for(user, additional_claims={'auth_version': None})
|
|
assert client.get('/api/v1/auth/me', headers=_headers(legacy)).status_code == 401
|
|
for expiration in (False, timedelta(seconds=-1)):
|
|
pending = access_token_for(user, additional_claims={'2fa_pending': True},
|
|
expires_delta=expiration)
|
|
response = client.post('/api/v1/auth/2fa/verify', json={
|
|
'temp_token': pending, 'code': pyotp.TOTP(user.totp_secret).now(),
|
|
})
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_password_reset_invalidates_outstanding_login_links(client, db_session):
|
|
admin = make_user(db_session, role='admin')
|
|
target = make_user(db_session)
|
|
created = client.post('/api/v1/auth/login-links', headers=headers_for(admin),
|
|
json={'user_id': target.id})
|
|
assert created.status_code == 201
|
|
target.set_password('resetpassword123')
|
|
db_session.session.commit()
|
|
assert client.post('/api/v1/auth/login-links/redeem',
|
|
json={'token': created.json['token']}).status_code == 401
|
|
|
|
|
|
def test_sso_tokens_share_session_and_pending_tokens_expire(app, db_session):
|
|
from app.api.sso import _complete_sso_login
|
|
user = make_user(db_session, auth_provider='oidc', password_hash=None)
|
|
with app.test_request_context('/api/v1/sso/callback/oidc'):
|
|
response, status = _complete_sso_login(user, 'oidc', False)
|
|
assert status == 200
|
|
pair = response.get_json()
|
|
assert decode_token(pair['access_token'])['session_id'] == decode_token(pair['refresh_token'])['session_id']
|
|
user.totp_enabled = True
|
|
db_session.session.commit()
|
|
response, status = _complete_sso_login(user, 'oidc', False)
|
|
assert status == 200
|
|
claims = decode_token(response.get_json()['temp_token'])
|
|
assert claims['2fa_pending'] is True
|
|
assert 0 < claims['exp'] - claims['iat'] <= 300
|
|
|
|
|
|
def test_refresh_cannot_renew_recent_authentication(client, db_session):
|
|
import time
|
|
from flask_jwt_extended import create_refresh_token
|
|
user = make_user(db_session, auth_provider='oidc', password_hash=None)
|
|
stale_auth = int(time.time()) - 600
|
|
refresh_token = create_refresh_token(identity=user.id, additional_claims={'auth_time': stale_auth})
|
|
refreshed = _refresh(client, refresh_token)
|
|
assert refreshed.status_code == 200
|
|
token = refreshed.json['access_token']
|
|
assert decode_token(token)['auth_time'] == stale_auth
|
|
assert client.put('/api/v1/auth/me', headers=_headers(token),
|
|
json={'password': 'replacement123'}).status_code == 403
|
|
|
|
|
|
def test_session_migration_upgrades_existing_users_and_is_repeatable(tmp_path, monkeypatch):
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import sqlalchemy as sa
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
|
|
path = Path(__file__).parents[1] / 'migrations/versions/097_user_auth_version.py'
|
|
spec = importlib.util.spec_from_file_location('session_migration', path)
|
|
migration = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(migration)
|
|
engine = sa.create_engine('sqlite:///' + (tmp_path / 'previous.db').as_posix())
|
|
with engine.begin() as connection:
|
|
connection.exec_driver_sql('CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT)')
|
|
connection.exec_driver_sql("INSERT INTO users VALUES (1, 'existing')")
|
|
monkeypatch.setattr(migration, 'op', Operations(MigrationContext.configure(connection)))
|
|
migration.upgrade()
|
|
migration.upgrade()
|
|
assert connection.exec_driver_sql('SELECT username, auth_version FROM users').one() == ('existing', '0')
|
|
assert 'revoked_sessions' in sa.inspect(connection).get_table_names()
|
|
migration.downgrade()
|
|
assert 'auth_version' not in {c['name'] for c in sa.inspect(connection).get_columns('users')}
|
|
engine.dispose()
|
|
|
|
|
|
def test_revocation_rows_cascade_on_user_deletion(db_session):
|
|
from app.models import RevokedSession
|
|
user = make_user(db_session)
|
|
db_session.session.add(RevokedSession(session_id='a' * 32, user_id=user.id))
|
|
db_session.session.commit()
|
|
assert RevokedSession.query.filter_by(user_id=user.id).count() == 1
|
|
connection = db_session.session.connection()
|
|
connection.exec_driver_sql('PRAGMA foreign_keys=ON')
|
|
assert connection.exec_driver_sql('PRAGMA foreign_keys').scalar() == 1
|
|
try:
|
|
db_session.session.delete(user)
|
|
db_session.session.commit()
|
|
assert RevokedSession.query.count() == 0
|
|
finally:
|
|
db_session.session.rollback()
|
|
db_session.session.connection().exec_driver_sql('PRAGMA foreign_keys=OFF')
|