import re import importlib import os import unittest from datetime import timedelta from unittest.mock import patch from flask import Flask, jsonify import auth class AuthTests(unittest.TestCase): def setUp(self): self.app = Flask(__name__, template_folder="../templates") self.app.config.update( SECRET_KEY="test-secret", AUTH_SETUP_TOKEN="one-time-token", SESSION_COOKIE_SECURE=False, ) self.accounts = {} @self.app.get("/private") def private_page(): return "private" @self.app.get("/api/private") def private_api(): return jsonify(ok=True) self.create_patcher = patch.object( auth, "create_account", side_effect=self.create_account ) self.account_patcher = patch.object( auth, "get_account", side_effect=lambda: self.accounts.get("account") ) self.create_patcher.start() self.account_patcher.start() self.addCleanup(self.create_patcher.stop) self.addCleanup(self.account_patcher.stop) auth.register_auth(self.app) self.app.config.update( SECRET_KEY="test-secret", AUTH_SETUP_TOKEN="one-time-token", SESSION_COOKIE_SECURE=False, ) self.client = self.app.test_client() def create_account(self, username, password_hash): if "account" in self.accounts: return False self.accounts["account"] = { "username": username, "password_hash": password_hash, } return True @staticmethod def csrf(response): match = re.search(rb'name="csrf_token" value="([^"]+)"', response.data) if not match: raise AssertionError("login page did not contain a CSRF token") return match.group(1).decode() def test_anonymous_page_and_api_are_blocked(self): page = self.client.get("/private") api = self.client.get("/api/private") self.assertEqual(page.status_code, 302) self.assertEqual(page.location, "/login") self.assertEqual(api.status_code, 401) self.assertEqual(api.json["error"], "需要登录") def test_current_and_legacy_reading_apis_are_blocked_by_real_app_guard(self): with patch.dict(os.environ, {"WEREAD_SESSION_SECRET": "test-secret"}): web_app = importlib.import_module("app").app web_app.config.update(SECRET_KEY="test-secret", SESSION_COOKIE_SECURE=False) with patch("app.pymysql.connect", side_effect=RuntimeError("test avoids database access")): client = web_app.test_client() self.assertEqual(client.get("/").status_code, 302) self.assertEqual(client.get("/api/reading/status").status_code, 401) self.assertEqual(client.get("/api/v2/status").status_code, 401) def test_home_footer_displays_icp_record_link(self): with patch.dict(os.environ, {"WEREAD_SESSION_SECRET": "test-secret"}): web_app = importlib.import_module("app").app web_app.config.update(SECRET_KEY="test-secret", SESSION_COOKIE_SECURE=False) client = web_app.test_client() with client.session_transaction() as session: session["auth_user"] = "reader" response = client.get("/") self.assertEqual(response.status_code, 200) self.assertIn(b"https://beian.miit.gov.cn/", response.data) self.assertIn("浙ICP备202607916号-1".encode(), response.data) def test_first_account_requires_setup_token_and_is_stored_as_hash(self): page = self.client.get("/login") csrf = self.csrf(page) denied = self.client.post("/auth/initialize", data={ "csrf_token": csrf, "setup_token": "wrong-token", "username": "reader", "password": "long-enough-password", }) self.assertEqual(denied.status_code, 403) self.assertNotIn("account", self.accounts) created = self.client.post("/auth/initialize", data={ "csrf_token": csrf, "setup_token": "one-time-token", "username": "reader", "password": "long-enough-password", }) self.assertEqual(created.status_code, 302) saved = self.accounts["account"] self.assertEqual(saved["username"], "reader") self.assertNotEqual(saved["password_hash"], "long-enough-password") self.assertTrue(auth.check_password_hash(saved["password_hash"], "long-enough-password")) with self.client.session_transaction() as session: self.assertTrue(session.permanent) self.assertEqual(session["auth_user"], "reader") self.assertEqual(self.app.permanent_session_lifetime, timedelta(days=5)) def test_login_rejects_wrong_password_and_accepts_matching_hash(self): saved_hash = auth.generate_password_hash("long-enough-password") self.accounts["account"] = {"username": "阅读者", "password_hash": saved_hash} page = self.client.get("/login") csrf = self.csrf(page) denied = self.client.post("/auth/login", data={ "csrf_token": csrf, "username": "阅读者", "password": "wrong-password", }) self.assertEqual(denied.status_code, 401) accepted = self.client.post("/auth/login", data={ "csrf_token": csrf, "username": "阅读者", "password": "long-enough-password", }) self.assertEqual(accepted.status_code, 302) self.assertEqual(self.client.get("/private").data, b"private") def test_missing_session_secret_fails_closed(self): self.app.config["SECRET_KEY"] = None response = self.client.get("/login") self.assertEqual(response.status_code, 503) self.assertIn("服务未配置会话密钥".encode(), response.data) def test_mutating_api_requires_csrf_token(self): @self.app.post("/api/write") def write_api(): return jsonify(ok=True) self.client.get("/login") self.accounts["account"] = { "username": "reader", "password_hash": auth.generate_password_hash("long-enough-password"), } # Simulate a session established through the regular authenticated login flow. with self.client.session_transaction() as session: session.permanent = True session["auth_user"] = "reader" session["csrf_token"] = "test-csrf" self.assertEqual(self.client.post("/api/write").status_code, 400) self.assertEqual(self.client.post("/api/write", headers={"X-CSRF-Token": "test-csrf"}).status_code, 200) if __name__ == "__main__": unittest.main()