diff --git a/backend/tests/test_permission_service.py b/backend/tests/test_permission_service.py new file mode 100644 index 00000000..c3e99165 --- /dev/null +++ b/backend/tests/test_permission_service.py @@ -0,0 +1,107 @@ +"""PermissionService + User permission resolution per role.""" +import pytest +from werkzeug.security import generate_password_hash + +from app import db +from app.models import User +from app.services.permission_service import PermissionService + + +def _user(role, name=None): + u = User(email=f'{name or role}@example.com', username=name or role, + password_hash=generate_password_hash('x'), role=role, is_active=True) + db.session.add(u) + db.session.commit() + return u + + +def test_role_templates_cover_every_feature_for_admin(): + admin = PermissionService.get_role_template('admin') + assert set(admin) == set(User.PERMISSION_FEATURES) + assert all(v == {'read': True, 'write': True} for v in admin.values()) + + +def test_unknown_role_has_no_permissions(app): + assert PermissionService.get_role_template('wizard') == {} + u = _user('viewer', 'nobody') + u.role = 'wizard' + assert not u.has_permission('files') + + +def test_developer_defaults_follow_template(app): + dev = _user('developer') + assert dev.has_permission('applications', 'write') + assert dev.has_permission('monitoring', 'read') + assert not dev.has_permission('monitoring', 'write') + assert not dev.has_permission('users') # not in the developer template + + +def test_admin_always_allowed(app): + admin = _user('admin') + assert all(admin.has_permission(f, 'write') for f in User.PERMISSION_FEATURES) + assert admin.has_permission('anything-at-all') + + +def test_viewer_cannot_write_anything(app): + viewer = _user('viewer') + assert not any(viewer.has_permission(f, 'write') for f in User.PERMISSION_FEATURES) + + +def test_get_user_permissions_missing_user(app): + assert PermissionService.get_user_permissions(999999) is None + + +def test_update_custom_permissions_overrides_role(app): + dev = _user('developer') + res = PermissionService.update_user_permissions( + dev.id, {'files': {'read': True, 'write': False}}) + assert res['success'] + assert res['permissions'] == {'files': {'read': True, 'write': False}} + assert not dev.has_permission('files', 'write') + assert not dev.has_permission('applications') # custom set replaces the template wholesale + + +def test_reset_restores_role_template(app): + dev = _user('developer') + PermissionService.update_user_permissions(dev.id, {'files': {'read': True}}) + res = PermissionService.reset_to_role_defaults(dev.id) + assert res['success'] and res['permissions'] == User.ROLE_PERMISSION_TEMPLATES['developer'] + assert dev.permissions is None + + +def test_admin_permissions_cannot_be_customized(app): + admin = _user('admin') + res = PermissionService.update_user_permissions(admin.id, {'files': {'read': False}}) + assert res == {'success': False, 'error': 'Admin permissions cannot be customized'} + + +def test_unknown_user_is_reported(app): + assert PermissionService.update_user_permissions(424242, {}) == { + 'success': False, 'error': 'User not found'} + assert PermissionService.reset_to_role_defaults(424242) == { + 'success': False, 'error': 'User not found'} + + +@pytest.mark.parametrize('perms, fragment', [ + ('nope', 'must be an object'), + ({'bogus': {'read': True}}, 'Unknown feature'), + ({'files': True}, 'read/write object'), + ({'files': {'execute': True}}, 'Invalid permission key'), + ({'files': {'read': 'yes'}}, 'must be boolean'), + ({'files': {'read': False, 'write': True}}, 'without read'), +]) +def test_validate_permissions_rejects(perms, fragment): + assert fragment in PermissionService.validate_permissions(perms) + + +def test_validate_permissions_accepts_good_input(): + assert PermissionService.validate_permissions({}) is None + assert PermissionService.validate_permissions( + {'files': {'read': True, 'write': True}, 'cron': {'read': True}}) is None + + +def test_invalid_update_does_not_persist(app): + dev = _user('developer') + res = PermissionService.update_user_permissions(dev.id, {'bogus': {}}) + assert not res['success'] + assert dev.permissions is None diff --git a/backend/tests/test_totp_service.py b/backend/tests/test_totp_service.py new file mode 100644 index 00000000..ee248b04 --- /dev/null +++ b/backend/tests/test_totp_service.py @@ -0,0 +1,174 @@ +"""TOTPService / TwoFactorSetup behaviour (RFC 6238 via pyotp).""" +import time +from urllib.parse import parse_qs, urlparse + +import pyotp +import pytest +from werkzeug.security import generate_password_hash + +from app import db +from app.models import User +from app.services.totp_service import TOTPService, TwoFactorSetup + + +@pytest.fixture +def secret(): + return TOTPService.generate_secret() + + +@pytest.fixture +def user(app): + u = User(email='t@example.com', username='totp-user', + password_hash=generate_password_hash('x'), role='developer', is_active=True) + db.session.add(u) + db.session.commit() + return u + + +def test_generate_secret_is_random_valid_base32(): + a, b = TOTPService.generate_secret(), TOTPService.generate_secret() + assert a != b + assert len(a) == 32 + pyotp.TOTP(a).now() # decodes as base32 + + +def test_current_code_verifies(secret): + code = TOTPService.get_current_code(secret) + assert len(code) == 6 and code.isdigit() + assert TOTPService.verify_code(secret, code) + + +def test_code_with_spaces_and_dashes_is_normalised(secret): + code = TOTPService.get_current_code(secret) + assert TOTPService.verify_code(secret, f'{code[:3]} {code[3:]}') + assert TOTPService.verify_code(secret, f'{code[:3]}-{code[3:]}') + + +@pytest.mark.parametrize('bad', ['', None, 'abcdef', '12345', '1234567', '000000x']) +def test_malformed_or_missing_code_is_rejected(secret, bad): + # A wrong 6-digit code could collide with the real one 1-in-a-million; use a shape check only. + assert not TOTPService.verify_code(secret, bad) + + +def test_missing_secret_is_rejected(): + assert not TOTPService.verify_code(None, '123456') + assert not TOTPService.verify_code('', '123456') + + +def test_wrong_code_is_rejected(secret): + real = TOTPService.get_current_code(secret) + wrong = f'{(int(real) + 1) % 1000000:06d}' + # Neighbouring-window codes are also valid, so make sure we truly picked a stranger. + totp = TOTPService.get_totp(secret) + now = time.time() + valid = {totp.at(now + d * 30) for d in (-1, 0, 1)} + if wrong in valid: + pytest.skip('collided with a neighbouring window') + assert not TOTPService.verify_code(secret, wrong) + + +def test_window_drift(secret): + totp = TOTPService.get_totp(secret) + prev = totp.at(time.time() - 30) + far = totp.at(time.time() - 120) + assert TOTPService.verify_code(secret, prev, valid_window=1) + assert not TOTPService.verify_code(secret, far, valid_window=1) + assert not TOTPService.verify_code(secret, far, valid_window=0) + + +def test_provisioning_uri(secret): + uri = TOTPService.get_provisioning_uri(secret, 'me@example.com') + parsed = urlparse(uri) + assert parsed.scheme == 'otpauth' and parsed.netloc == 'totp' + qs = parse_qs(parsed.query) + assert qs['secret'] == [secret] + assert qs['issuer'] == ['ServerKit'] + assert 'me%40example.com' in parsed.path or 'me@example.com' in parsed.path + + +def test_qr_code_is_png_data_uri(secret): + qr = TOTPService.generate_qr_code_base64(secret, 'me@example.com') + if qr is None: + pytest.skip('qrcode not installed') + assert qr.startswith('data:image/png;base64,') + + +def test_backup_codes_roundtrip(): + plain, hashed = TOTPService.generate_backup_codes() + assert len(plain) == len(hashed) == TOTPService.BACKUP_CODE_COUNT + assert len(set(plain)) == len(plain) + assert all(len(c) == TOTPService.BACKUP_CODE_LENGTH for c in plain) + assert TOTPService.verify_backup_code(plain[0], hashed) == hashed[0] + assert TOTPService.verify_backup_code(plain[0].upper(), hashed) == hashed[0] + formatted = TOTPService.format_backup_codes([plain[0]])[0] + assert formatted == f'{plain[0][:4]}-{plain[0][4:]}' + assert TOTPService.verify_backup_code(formatted, hashed) == hashed[0] + + +def test_backup_code_rejects_unknown_and_empty(): + _, hashed = TOTPService.generate_backup_codes() + assert TOTPService.verify_backup_code('deadbeef', hashed) is None + assert TOTPService.verify_backup_code('', hashed) is None + assert TOTPService.verify_backup_code('abcd1234', []) is None + + +def test_setup_flow_enables_2fa_and_returns_backup_codes(user): + setup = TwoFactorSetup(user) + assert setup.confirm_setup('123456') == (False, '2FA setup not initiated') + + data = setup.initiate_setup() + assert data['issuer'] == 'ServerKit' and data['secret'] == user.totp_secret + assert not user.totp_enabled + + ok, codes = setup.confirm_setup(TOTPService.get_current_code(user.totp_secret)) + assert ok and len(codes) == 10 + assert user.totp_enabled and user.totp_confirmed_at + assert setup.get_remaining_backup_codes_count() == 10 + + +def test_confirm_with_wrong_code_does_not_enable(user): + setup = TwoFactorSetup(user) + setup.initiate_setup() + totp = TOTPService.get_totp(user.totp_secret) + now = time.time() + wrong = next(f'{n:06d}' for n in range(1000000) + if f'{n:06d}' not in {totp.at(now + d * 30) for d in (-1, 0, 1)}) + assert setup.confirm_setup(wrong) == (False, 'Invalid verification code') + assert not user.totp_enabled + + +def _enabled(user): + setup = TwoFactorSetup(user) + setup.initiate_setup() + _, codes = setup.confirm_setup(TOTPService.get_current_code(user.totp_secret)) + return setup, codes + + +def test_disable_with_totp_code(user): + setup, _ = _enabled(user) + ok, err = setup.disable(TOTPService.get_current_code(user.totp_secret)) + assert ok and err is None + assert not user.totp_enabled and user.totp_secret is None and user.backup_codes is None + + +def test_disable_with_backup_code(user): + setup, codes = _enabled(user) + ok, err = setup.disable(codes[0]) + assert ok and err is None and not user.totp_enabled + + +def test_disable_rejects_garbage_and_when_not_enabled(user): + setup = TwoFactorSetup(user) + assert setup.disable('123456') == (False, '2FA is not enabled') + setup, _ = _enabled(user) + assert setup.disable('zzzz-zzzz') == (False, 'Invalid verification code') + assert user.totp_enabled + + +def test_regenerate_backup_codes_replaces_old_set(user): + setup, old = _enabled(user) + old_hashes = set(user.get_backup_codes()) + ok, new = setup.regenerate_backup_codes(TOTPService.get_current_code(user.totp_secret)) + assert ok and len(new) == 10 + assert not old_hashes & set(user.get_backup_codes()) + assert setup.regenerate_backup_codes('abc') == (False, 'Invalid verification code')