105 lines
6.3 KiB
Python
105 lines
6.3 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
import unittest
|
|
from tempfile import TemporaryDirectory
|
|
from unittest.mock import patch
|
|
|
|
from app import create_app
|
|
from app.config import Config
|
|
from app.extensions import db
|
|
from app.models import EmailVerificationCode, TrustedDevice, User
|
|
|
|
|
|
class EmailAuthenticationTest(unittest.TestCase):
|
|
def create_app(self, directory: str):
|
|
class TestConfig(Config):
|
|
TESTING = True
|
|
SECRET_KEY = "test-secret"
|
|
SECRET_KEY_GENERATED = False
|
|
SQLALCHEMY_DATABASE_URI = f"sqlite:///{directory}/test.db"
|
|
ALLOW_REGISTRATION = True
|
|
ADMIN_PASSWORD = None
|
|
ADMIN_EMAIL = ""
|
|
RESEND_API_KEY = "test"
|
|
RESEND_FROM_EMAIL = "no-reply@waternetwork.cn"
|
|
SESSION_COOKIE_SECURE = False
|
|
return create_app(TestConfig, load_model_on_start=False)
|
|
|
|
def csrf(self, response) -> str:
|
|
return re.search(rb'name="csrf_token" value="([^"]+)"', response.data).group(1).decode()
|
|
|
|
def form(self, client, path: str, **data):
|
|
page = client.get(path)
|
|
data["csrf_token"] = self.csrf(page)
|
|
return client.post(path, data=data)
|
|
|
|
@patch("app.routes.send_transactional_email")
|
|
@patch("app.routes.secrets.randbelow", return_value=123456)
|
|
def test_registration_requires_and_consumes_email_code(self, _random, _send):
|
|
with TemporaryDirectory() as directory:
|
|
app = self.create_app(directory); client = app.test_client()
|
|
page = client.get("/register")
|
|
with client.session_transaction() as state: captcha = state["captcha"]
|
|
response = client.post("/register", data={"csrf_token": self.csrf(page), "username": "Alice", "email": "Alice@example.com", "password": "password-1234", "captcha": captcha})
|
|
self.assertEqual(response.status_code, 302)
|
|
with app.app_context():
|
|
user = User.query.filter_by(email="alice@example.com").one()
|
|
self.assertFalse(user.is_active_account)
|
|
self.assertEqual(EmailVerificationCode.query.count(), 1)
|
|
verify = client.get("/verify/register")
|
|
response = client.post("/verify/register", data={"csrf_token": self.csrf(verify), "code": "123456"})
|
|
self.assertEqual(response.status_code, 302)
|
|
with app.app_context(): self.assertTrue(User.query.filter_by(email="alice@example.com").one().is_active_account)
|
|
|
|
@patch("app.routes.send_transactional_email")
|
|
@patch("app.routes.secrets.randbelow", return_value=123456)
|
|
def test_unknown_device_requires_email_mfa_and_creates_trusted_device(self, _random, _send):
|
|
with TemporaryDirectory() as directory:
|
|
app = self.create_app(directory)
|
|
with app.app_context():
|
|
user = User(username="Alice", email="alice@example.com", is_active_account=True); user.set_password("password-1234"); db.session.add(user); db.session.commit()
|
|
client = app.test_client(); page = client.get("/login")
|
|
with client.session_transaction() as state: captcha = state["captcha"]
|
|
response = client.post("/login", data={"csrf_token": self.csrf(page), "email": "alice@example.com", "password": "password-1234", "captcha": captcha})
|
|
self.assertEqual(response.location, "/verify/login")
|
|
verify = client.get("/verify/login")
|
|
response = client.post("/verify/login", data={"csrf_token": self.csrf(verify), "code": "123456"})
|
|
self.assertEqual(response.status_code, 302)
|
|
with app.app_context(): self.assertEqual(TrustedDevice.query.count(), 1)
|
|
|
|
@patch("app.routes.send_transactional_email")
|
|
@patch("app.routes.secrets.randbelow", return_value=123456)
|
|
def test_password_reset_revokes_trusted_devices(self, _random, _send):
|
|
with TemporaryDirectory() as directory:
|
|
app = self.create_app(directory)
|
|
with app.app_context():
|
|
user = User(username="Alice", email="alice@example.com", is_active_account=True); user.set_password("password-1234"); db.session.add(user); db.session.commit(); db.session.add(TrustedDevice(user_id=user.id, token_hash="a" * 64, auth_version=1, expires_at=__import__('app.time_utils', fromlist=['utc_now']).utc_now())); db.session.commit()
|
|
client = app.test_client()
|
|
response = self.form(client, "/forgot-password", email="alice@example.com")
|
|
self.assertEqual(response.status_code, 302)
|
|
verify = client.get("/verify/reset")
|
|
response = client.post("/verify/reset", data={"csrf_token": self.csrf(verify), "code": "123456"})
|
|
self.assertEqual(response.location, "/set-password")
|
|
page = client.get("/set-password")
|
|
response = client.post("/set-password", data={"csrf_token": self.csrf(page), "password": "new-password-1234", "password_confirm": "new-password-1234"})
|
|
self.assertEqual(response.status_code, 302)
|
|
with app.app_context():
|
|
user = User.query.filter_by(email="alice@example.com").one()
|
|
self.assertTrue(user.check_password("new-password-1234")); self.assertEqual(TrustedDevice.query.count(), 0); self.assertEqual(user.auth_version, 2)
|
|
|
|
@patch("app.routes.send_transactional_email")
|
|
@patch("app.routes.secrets.randbelow", return_value=123456)
|
|
def test_invalid_code_attempts_are_limited(self, _random, _send):
|
|
with TemporaryDirectory() as directory:
|
|
app = self.create_app(directory); client = app.test_client()
|
|
with app.app_context():
|
|
user = User(username="Alice", email="alice@example.com", is_active_account=True); user.set_password("password-1234"); db.session.add(user); db.session.commit()
|
|
page = client.get("/login")
|
|
with client.session_transaction() as state: captcha = state["captcha"]
|
|
client.post("/login", data={"csrf_token": self.csrf(page), "email": "alice@example.com", "password": "password-1234", "captcha": captcha})
|
|
for _ in range(5):
|
|
page = client.get("/verify/login"); client.post("/verify/login", data={"csrf_token": self.csrf(page), "code": "000000"})
|
|
page = client.get("/verify/login"); response = client.post("/verify/login", data={"csrf_token": self.csrf(page), "code": "123456"})
|
|
self.assertEqual(response.status_code, 400)
|