WeReadBoard/tests/test_auth.py

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)
        self.assertIn("版本 v0.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()