feat(auth): 增加登录验证和五天会话

tests/test_auth.py

@@ -0,0 +1,159 @@
+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_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()