368 lines
19 KiB
Python
368 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
import unittest
|
|
from datetime import timedelta
|
|
from tempfile import TemporaryDirectory
|
|
from unittest.mock import patch
|
|
from urllib.parse import urlparse
|
|
|
|
from flask import session
|
|
|
|
from app import create_app
|
|
from app.config import Config
|
|
from app.email import EmailDeliveryError
|
|
from app.extensions import db
|
|
from app.models import EmailVerificationCode, PasswordResetToken, RegistrationInvitation, 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)
|
|
self.assertIn("al****ce@example.com", html)
|
|
self.assertNotIn("alice@example.com", 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 test_account_security_uses_custom_email_validation_and_password_autofill(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
|
|
|
|
client = app.test_client()
|
|
with app.app_context():
|
|
self.login_as(client, db.session.get(User, user_id))
|
|
response = client.get("/account/security")
|
|
html = response.get_data(as_text=True)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('novalidate data-change-email-form', html)
|
|
self.assertIn('autocomplete="current-password"', html)
|
|
self.assertIn('autocomplete="email"', html)
|
|
self.assertIn('请输入有效的新登录邮箱。', html)
|
|
|
|
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)
|
|
|
|
def login_as(self, client, user: User) -> None:
|
|
with client.session_transaction() as state:
|
|
state["_user_id"] = str(user.id)
|
|
state["_fresh"] = True
|
|
state["auth_version"] = user.auth_version
|
|
|
|
def test_admin_user_list_is_paginated_and_searchable(self):
|
|
with TemporaryDirectory() as directory:
|
|
app = self.create_app(directory)
|
|
with app.app_context():
|
|
admin = User(
|
|
username="Admin",
|
|
email="admin@example.com",
|
|
is_admin=True,
|
|
is_active_account=True,
|
|
)
|
|
admin.set_password("Password-1234!")
|
|
db.session.add(admin)
|
|
for index in range(21):
|
|
user = User(
|
|
username=f"User{index:02d}",
|
|
email=f"user{index:02d}@example.com",
|
|
is_active_account=True,
|
|
)
|
|
user.set_password("Password-1234!")
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
admin_id = admin.id
|
|
|
|
client = app.test_client()
|
|
with app.app_context():
|
|
self.login_as(client, db.session.get(User, admin_id))
|
|
|
|
response = client.get("/admin?user_page=2")
|
|
html = response.get_data(as_text=True)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("共 21 位用户，第 2 / 2 页", html)
|
|
self.assertIn("User00", html)
|
|
self.assertNotIn("User20", html)
|
|
|
|
response = client.get("/admin?q=user20@example.com")
|
|
html = response.get_data(as_text=True)
|
|
self.assertIn("共 1 位用户，第 1 / 1 页", html)
|
|
self.assertIn("User20", html)
|
|
|
|
@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)
|
|
self.assertEqual(response.location, "/home")
|
|
self.assertEqual(client.get("/home").status_code, 200)
|
|
with app.app_context():
|
|
self.assertTrue(User.query.filter_by(email="alice@example.com").one().is_active_account)
|
|
self.assertEqual(TrustedDevice.query.count(), 0)
|
|
|
|
@patch("app.routes.send_transactional_email")
|
|
@patch("app.routes.secrets.randbelow", return_value=123456)
|
|
def test_login_accepts_username_or_email(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()
|
|
|
|
for identifier in ("Alice", "alice@example.com"):
|
|
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), "identifier": identifier,
|
|
"password": "Password-1234!", "captcha": captcha,
|
|
})
|
|
self.assertEqual(response.location, "/verify/login")
|
|
with app.app_context():
|
|
EmailVerificationCode.query.delete()
|
|
db.session.commit()
|
|
|
|
def test_login_reports_verification_code_cooldown(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.flush()
|
|
db.session.add(EmailVerificationCode(
|
|
email=user.email,
|
|
purpose="login",
|
|
code_hash="a" * 64,
|
|
expires_at=utc_now() + timedelta(minutes=10),
|
|
requested_ip="127.0.0.1",
|
|
))
|
|
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), "identifier": "Alice",
|
|
"password": "Password-1234!", "captcha": captcha,
|
|
})
|
|
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertEqual(response.location, "/verify/login")
|
|
with app.app_context():
|
|
self.assertEqual(EmailVerificationCode.query.count(), 1)
|
|
|
|
verify_page = client.get("/verify/login")
|
|
html = verify_page.get_data(as_text=True)
|
|
self.assertIn("邮箱二次认证", html)
|
|
self.assertIn("邮箱验证码已发送，请输入验证码完成二次认证。", html)
|
|
self.assertIn('name="trust_device"', html)
|
|
|
|
def test_invalid_graphic_captcha_does_not_start_email_verification(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()
|
|
|
|
client = app.test_client()
|
|
page = client.get("/login")
|
|
response = client.post("/login", data={
|
|
"csrf_token": self.csrf(page), "identifier": "Alice",
|
|
"password": "Password-1234!", "captcha": "WRONG",
|
|
})
|
|
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("图形验证码错误", response.get_data(as_text=True))
|
|
self.assertIn('window.__authErrorField = "captcha"', response.get_data(as_text=True))
|
|
with app.app_context():
|
|
self.assertEqual(EmailVerificationCode.query.count(), 0)
|
|
|
|
@patch("app.routes.send_transactional_email", side_effect=EmailDeliveryError("delivery failed"))
|
|
def test_email_delivery_failure_does_not_mark_graphic_captcha(self, _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), "identifier": "Alice",
|
|
"password": "Password-1234!", "captcha": captcha,
|
|
})
|
|
|
|
self.assertEqual(response.status_code, 503)
|
|
self.assertIn("验证码发送失败，请稍后重试。", response.get_data(as_text=True))
|
|
self.assertIn("window.__authErrorField = null", response.get_data(as_text=True))
|
|
|
|
@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), "identifier": "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", "trust_device": "on"})
|
|
self.assertEqual(response.status_code, 302)
|
|
with app.app_context(): self.assertEqual(TrustedDevice.query.count(), 1)
|
|
|
|
@patch("app.routes.send_transactional_email")
|
|
def test_password_reset_link_revokes_trusted_devices(self, 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)
|
|
self.assertEqual(response.location, "/forgot-password")
|
|
self.assertEqual(send.call_count, 1)
|
|
html = send.call_args.kwargs["html"]
|
|
reset_path = urlparse(re.search(r'href="([^"]+)"', html).group(1)).path
|
|
page = client.get(reset_path)
|
|
response = client.post(reset_path, 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); self.assertIsNotNone(PasswordResetToken.query.one().used_at)
|
|
|
|
@patch("app.routes.send_transactional_email")
|
|
def test_admin_can_invite_user_while_self_registration_is_disabled(self, send):
|
|
with TemporaryDirectory() as directory:
|
|
app = self.create_app(directory)
|
|
with app.app_context():
|
|
admin = User(username="Admin", email="admin@example.com", is_admin=True, is_active_account=True)
|
|
admin.set_password("Password-1234!")
|
|
db.session.add(admin)
|
|
db.session.commit()
|
|
admin_id = admin.id
|
|
client = app.test_client()
|
|
with app.app_context():
|
|
self.login_as(client, db.session.get(User, admin_id))
|
|
admin_page = client.get("/admin")
|
|
response = client.post(
|
|
"/admin/invitations",
|
|
data={"csrf_token": self.csrf(admin_page), "email": "invitee@example.com"},
|
|
)
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertEqual(send.call_count, 1)
|
|
with app.app_context():
|
|
self.assertEqual(RegistrationInvitation.query.count(), 1)
|
|
invite_html = send.call_args.kwargs["html"]
|
|
invite_path = urlparse(re.search(r'href="([^"]+)"', invite_html).group(1)).path
|
|
page = client.get(invite_path)
|
|
response = client.post(invite_path, data={"csrf_token": self.csrf(page), "username": "Invited User", "password": "Invited-password-1234!", "password_confirm": "Invited-password-1234!"})
|
|
self.assertEqual(response.status_code, 302)
|
|
with app.app_context():
|
|
invited = User.query.filter_by(email="invitee@example.com").one()
|
|
self.assertTrue(invited.is_active_account)
|
|
self.assertIsNotNone(invited.email_verified_at)
|
|
self.assertIsNotNone(RegistrationInvitation.query.one().used_at)
|
|
|
|
@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), "identifier": "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)
|