from __future__ import annotations import re import unittest from datetime import timedelta from tempfile import TemporaryDirectory from unittest.mock import patch from flask import session from app import create_app from app.config import Config from app.extensions import db from app.models import EmailVerificationCode, TrustedDevice, User from app.routes import grant_fresh_authorization, has_fresh_authorization, valid_password from app.time_utils import utc_now class EmailAuthenticationTest(unittest.TestCase): def test_password_policy_requires_all_character_categories(self): self.assertTrue(valid_password("Strong-password-123!")) self.assertFalse(valid_password("lowercase-password-123!")) self.assertFalse(valid_password("UPPERCASE-PASSWORD-123!")) self.assertFalse(valid_password("NoSpecialPassword123")) self.assertFalse(valid_password("NoWhitespace-123 !")) def test_verification_page_uses_six_code_inputs_and_initial_resend_delay(self): with TemporaryDirectory() as directory: app = self.create_app(directory) client = app.test_client() with client.session_transaction() as state: state["pending_email"] = "alice@example.com" state["pending_purpose"] = "login" response = client.get("/verify/login") html = response.get_data(as_text=True) self.assertEqual(response.status_code, 200) self.assertEqual(html.count("data-code-digit\n"), 6) self.assertIn("60 秒后可重新发送", html) self.assertIn('id="resendButton"', html) def test_fresh_authorization_is_bound_to_user_and_expires(self): 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() user_id = user.id with app.test_request_context(): user = db.session.get(User, user_id) grant_fresh_authorization(user, "password_change") self.assertTrue(has_fresh_authorization(user, "password_change")) session["fresh_auth_password_change"]["expires_at"] = ( utc_now() - timedelta(seconds=1) ).isoformat() self.assertFalse(has_fresh_authorization(user, "password_change")) 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)