Reisacher-Forderung/tests/test_auth.py

import tests  # Initialize temporary logging before importing the application.
import unittest
from datetime import timedelta
from unittest.mock import patch

from fastapi.testclient import TestClient

from app.config import settings
from app.ldap import DirectoryUser, normalize_username
from app.main import app
from app.oauth_token import InvalidSessionToken, create_access_token, decode_access_token, token_needs_renewal, utcnow
from app.permissions import Entitlement, highest_role


def rights(role: str = "Mitarbeiter", priority: int = 100):
    return (
        Entitlement("1", "10", None, role, priority),
        Entitlement("1", "20", "Service", "Mitarbeiter", 100),
    )


class UsernameTests(unittest.TestCase):
    def test_supported_ad_name_formats_are_normalized(self):
        self.assertEqual(normalize_username(r"DOMAIN\Max.Mustermann"), "max.mustermann")
        self.assertEqual(normalize_username("Max.Mustermann@example.local"), "max.mustermann")
        self.assertEqual(normalize_username("Max.Mustermann"), "max.mustermann")

    def test_invalid_username_is_rejected(self):
        with self.assertRaises(ValueError):
            normalize_username("../../admin")


class TokenTests(unittest.TestCase):
    def test_token_contains_required_identity_and_session_claims(self):
        token = create_access_token(
            username="max.mustermann",
            name="Max Mustermann",
            email="max@example.local",
            role="Mitarbeiter",
            auth_source="ldap",
        )
        payload = decode_access_token(token)
        self.assertEqual(payload["sub"], "max.mustermann")
        self.assertEqual(payload["role"], "Mitarbeiter")
        self.assertEqual(payload["auth_source"], "ldap")
        self.assertLessEqual(payload["exp"], payload["session_exp"])
        self.assertTrue(payload["csrf"])

    def test_role_change_requires_immediate_renewal(self):
        token = create_access_token(
            username="max.mustermann",
            name="Max Mustermann",
            email="",
            role="Mitarbeiter",
            auth_source="iis",
        )
        self.assertTrue(token_needs_renewal(decode_access_token(token), "Buchhaltung"))

    def test_active_token_is_not_renewed_too_early(self):
        now = utcnow()
        token = create_access_token(
            username="max.mustermann",
            name="Max Mustermann",
            email="",
            role="Mitarbeiter",
            auth_source="iis",
            now=now,
        )
        self.assertFalse(token_needs_renewal(decode_access_token(token), "Mitarbeiter", now + timedelta(minutes=1)))

    def test_absolute_session_expiry_is_enforced(self):
        token = create_access_token(
            username="max.mustermann",
            name="Max Mustermann",
            email="",
            role="Mitarbeiter",
            auth_source="iis",
            now=utcnow() - timedelta(hours=9),
        )
        with self.assertRaises(InvalidSessionToken):
            decode_access_token(token)


class PermissionTests(unittest.TestCase):
    def test_highest_role_uses_configured_priority(self):
        entitlements = (
            Entitlement("1", "10", None, "Mitarbeiter", 100),
            Entitlement("1", "20", None, "Buchhaltung", 400),
            Entitlement("1", "30", None, "Standortleitung", 300),
        )
        self.assertEqual(highest_role(entitlements), "Buchhaltung")

    def test_no_explicit_rights_are_rejected(self):
        with self.assertRaises(Exception) as context:
            highest_role(())
        self.assertEqual(context.exception.status_code, 403)


class AuthenticationFlowTests(unittest.TestCase):
    def setUp(self):
        debug_config = patch.object(settings, "DEBUG_AUTH", False)
        debug_config.start()
        self.addCleanup(debug_config.stop)
        self.old_proxies = list(settings.TRUSTED_PROXY_IPS)
        self.old_secure = settings.COOKIE_SECURE
        settings.TRUSTED_PROXY_IPS = ["testclient"]
        settings.COOKIE_SECURE = True
        self.directory_user = DirectoryUser(
            username="max.mustermann",
            distinguished_name="CN=Max Mustermann,DC=example,DC=local",
            name="Max Mustermann",
            email="max@example.local",
        )

    def tearDown(self):
        settings.TRUSTED_PROXY_IPS = self.old_proxies
        settings.COOKIE_SECURE = self.old_secure

    def test_missing_identity_redirects_to_login(self):
        with TestClient(app, base_url="https://testserver") as client:
            response = client.get("/select?from=test", follow_redirects=False)
        self.assertEqual(response.status_code, 303)
        self.assertIn("/login?next=", response.headers["location"])

    def test_htmx_request_receives_full_page_redirect_header(self):
        with TestClient(app, base_url="https://testserver") as client:
            response = client.get("/select", headers={"HX-Request": "true"}, follow_redirects=False)
        self.assertEqual(response.status_code, 200)
        self.assertIn("/login?next=", response.headers["HX-Redirect"])

    def test_untrusted_identity_header_is_ignored(self):
        settings.TRUSTED_PROXY_IPS = ["127.0.0.1"]
        with TestClient(app, base_url="https://testserver") as client:
            response = client.get(
                "/select", headers={"X-Remote-User": r"DOMAIN\Max.Mustermann"}, follow_redirects=False
            )
        self.assertEqual(response.status_code, 303)

    @patch("app.auth.load_entitlements", return_value=rights())
    @patch("app.auth.lookup_directory_user")
    def test_trusted_iis_identity_creates_secure_session(self, lookup, _load):
        lookup.return_value = self.directory_user
        with TestClient(app, base_url="https://testserver") as client:
            response = client.get("/select", headers={"X-Remote-User": r"DOMAIN\Max.Mustermann"})
        self.assertEqual(response.status_code, 200)
        cookie = response.headers.get("set-cookie", "")
        self.assertIn("gcepic_session=", cookie)
        self.assertIn("HttpOnly", cookie)
        self.assertIn("Secure", cookie)
        self.assertIn("SameSite=lax", cookie)

    @patch("app.auth.load_entitlements", return_value=())
    @patch("app.auth.lookup_directory_user")
    def test_iis_user_without_explicit_rights_is_forbidden(self, lookup, _load):
        lookup.return_value = self.directory_user
        with TestClient(app, base_url="https://testserver") as client:
            response = client.get("/select", headers={"X-Remote-User": r"DOMAIN\Max.Mustermann"})
        self.assertEqual(response.status_code, 403)

    @patch("app.routes.load_entitlements", return_value=rights())
    @patch("app.routes.ldap_authenticate")
    def test_ldap_fallback_creates_session(self, authenticate, _load):
        authenticate.return_value = self.directory_user
        with TestClient(app, base_url="https://testserver") as client:
            login_page = client.get("/login")
            csrf = client.cookies[settings.LOGIN_CSRF_COOKIE_NAME]
            response = client.post(
                "/login",
                data={"username": "max.mustermann", "password": "secret", "_csrf": csrf, "next": "/select"},
                follow_redirects=False,
            )
        self.assertEqual(login_page.status_code, 200)
        self.assertEqual(response.status_code, 303)
        self.assertEqual(response.headers["location"], "/select")
        self.assertIn("gcepic_session=", response.headers.get("set-cookie", ""))

    @patch("app.auth.load_entitlements", return_value=rights())
    @patch("app.auth.lookup_directory_user")
    def test_logout_requires_csrf_and_deletes_session(self, lookup, _load):
        lookup.return_value = self.directory_user
        with TestClient(app, base_url="https://testserver") as client:
            login = client.get("/select", headers={"X-Remote-User": r"DOMAIN\Max.Mustermann"})
            token = login.cookies[settings.SESSION_COOKIE_NAME]
            csrf = decode_access_token(token)["csrf"]
            rejected = client.post("/logout")
            accepted = client.post("/logout", data={"_csrf": csrf})
        self.assertEqual(rejected.status_code, 403)
        self.assertEqual(accepted.status_code, 200)
        self.assertIn("Max-Age=0", accepted.headers.get("set-cookie", ""))


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