Reisacher-Forderung/tests/test_debug_auth.py

import tests  # Initialize temporary logging before importing the application.
import unittest
from html.parser import HTMLParser
from unittest.mock import patch

from fastapi.testclient import TestClient

from app.config import Settings, settings
from app.main import app
from app.oauth_token import InvalidSessionToken, create_access_token, decode_access_token
from app.permissions import Entitlement


class LoginFormParser(HTMLParser):
    def handle_starttag(self, tag, attrs):
        attributes = dict(attrs)
        if tag == "input" and attributes.get("name") == "_csrf":
            self.csrf = attributes["value"]


def form_csrf(response):
    parser = LoginFormParser()
    parser.feed(response.text)
    return parser.csrf


class DebugLoginTests(unittest.TestCase):
    def setUp(self):
        self.config = patch.multiple(settings, ENVIRONMENT="development", DEBUG_AUTH=True, COOKIE_SECURE=False)
        self.config.start()
        self.addCleanup(self.config.stop)
        self.ldap = patch("app.routes.ldap_authenticate")
        self.ldap_mock = self.ldap.start()
        self.addCleanup(self.ldap.stop)
        self.client = TestClient(app, base_url="http://testserver")
        self.addCleanup(self.client.close)
        self.entitlements = (
            Entitlement("1", "10", None, "Buchhaltung", 400),
            Entitlement("1", "20", "Service", "Mitarbeiter", 100),
        )

    def login(self, username="winter", password="winter", csrf=True):
        self.client.get("/login")
        return self.client.post(
            "/login",
            data={
                "username": username,
                "password": password,
                "_csrf": self.client.cookies[settings.LOGIN_CSRF_COOKIE_NAME] if csrf else "incorrect",
            },
            follow_redirects=False,
        )

    @patch("app.routes.load_entitlements")
    @patch("app.auth.load_entitlements")
    def test_existing_user_gets_jwt_and_can_access_protected_page(self, request_rights, login_rights):
        login_rights.return_value = request_rights.return_value = self.entitlements
        response = self.login("Winter", "Winter")
        self.assertEqual(response.status_code, 303)
        payload = decode_access_token(response.cookies[settings.SESSION_COOKIE_NAME])
        self.assertEqual(payload["sub"], "winter")
        self.assertEqual(payload["role"], "Buchhaltung")
        self.assertEqual(payload["auth_source"], "debug")
        login_rights.assert_called_once()
        self.assertEqual(login_rights.call_args.args[1], "winter")
        self.assertEqual(self.client.get("/select").status_code, 200)
        request_rights.assert_called_once()
        self.ldap_mock.assert_not_called()

    @patch("app.routes.load_entitlements")
    def test_different_password_is_rejected_without_ldap_or_rights_lookup(self, login_rights):
        response = self.login("winter", "Winter")
        self.assertEqual(response.status_code, 401)
        self.assertNotIn(settings.SESSION_COOKIE_NAME, response.cookies)
        login_rights.assert_not_called()
        self.ldap_mock.assert_not_called()

    @patch("app.routes.load_entitlements", return_value=())
    def test_user_without_rights_is_rejected(self, login_rights):
        response = self.login("unknown", "unknown")
        self.assertEqual(response.status_code, 403)
        self.assertNotIn(settings.SESSION_COOKIE_NAME, response.cookies)
        self.ldap_mock.assert_not_called()

    def test_csrf_remains_required(self):
        self.assertEqual(self.login(csrf=False).status_code, 403)

    @patch("app.routes.load_entitlements")
    def test_opening_another_login_page_keeps_first_form_valid(self, login_rights):
        login_rights.return_value = self.entitlements
        first_page = self.client.get("/login")
        second_page = self.client.get("/login?next=/select")
        self.assertEqual(form_csrf(first_page), form_csrf(second_page))
        self.assertEqual(second_page.headers["cache-control"], "no-store")
        response = self.client.post(
            "/login",
            data={"username": "winter", "password": "winter", "_csrf": form_csrf(first_page)},
            follow_redirects=False,
        )
        self.assertEqual(response.status_code, 303)

    @patch("app.routes.load_entitlements")
    def test_missing_cookie_returns_fresh_form_and_allows_retry(self, login_rights):
        login_rights.return_value = self.entitlements
        page = self.client.get("/login")
        self.client.cookies.clear()
        rejected = self.client.post(
            "/login",
            data={"username": "winter", "password": "winter", "_csrf": form_csrf(page), "next": "/select"},
        )
        self.assertEqual(rejected.status_code, 403)
        self.assertIn("text/html", rejected.headers["content-type"])
        self.assertEqual(rejected.headers["cache-control"], "no-store")
        self.assertNotEqual(form_csrf(rejected), form_csrf(page))
        login_rights.assert_not_called()
        self.ldap_mock.assert_not_called()
        accepted = self.client.post(
            "/login",
            data={"username": "winter", "password": "winter", "_csrf": form_csrf(rejected), "next": "/select"},
            follow_redirects=False,
        )
        self.assertEqual(accepted.status_code, 303)
        self.assertEqual(accepted.headers["location"], "/select")

    @patch("app.routes.load_entitlements")
    def test_missing_form_token_is_rejected_before_authentication(self, login_rights):
        self.client.get("/login")
        response = self.client.post("/login", data={"username": "winter", "password": "winter"})
        self.assertEqual(response.status_code, 403)
        self.assertEqual(form_csrf(response), self.client.cookies[settings.LOGIN_CSRF_COOKIE_NAME])
        login_rights.assert_not_called()
        self.ldap_mock.assert_not_called()

    @patch("app.auth.lookup_directory_user")
    def test_proxy_header_does_not_bypass_debug_form(self, lookup):
        with patch.object(settings, "TRUSTED_PROXY_IPS", ["testclient"]):
            response = self.client.get("/login", headers={"X-Remote-User": r"DOMAIN\winter"})
        self.assertEqual(response.status_code, 200)
        lookup.assert_not_called()

    def test_debug_token_is_rejected_after_disabling_debug_mode(self):
        token = create_access_token(
            username="winter", name="winter", email="", role="Mitarbeiter", auth_source="debug"
        )
        with patch.object(settings, "DEBUG_AUTH", False):
            with self.assertRaises(InvalidSessionToken):
                decode_access_token(token)

    def test_production_configuration_rejects_debug_auth(self):
        with self.assertRaises(ValueError):
            Settings(_env_file=None, ENVIRONMENT="production", DEBUG_AUTH=True)

    def test_debug_auth_is_disabled_by_default(self):
        self.assertFalse(Settings(_env_file=None, DEBUG_AUTH=False).debug_auth_enabled)


if __name__ == "__main__":
    unittest.main()