learningcircuit--local-deep-research/tests/api_tests/test_proper_csrf.py

"""Test with proper CSRF handling."""

import re
import time
from loguru import logger


class TestProperCSRF:
    """Test authentication with proper CSRF token handling."""

    def test_login_with_csrf(self, client):
        """Test login with proper CSRF token."""
        # Step 1: Get the login page to establish session and get CSRF token
        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: REFACTOR_SPLIT).
        logger.info("Step 1: Getting login page...")
        response = client.get("/auth/login")
        assert response.status_code == 200
        logger.info(f"GET /auth/login -> {response.status_code}")

        # Extract CSRF token from the form
        csrf_match = re.search(
            r'name="csrf_token" value="([^"]+)"', response.data.decode()
        )
        assert csrf_match is not None, "Could not find CSRF token in login page"

        csrf_token = csrf_match.group(1)
        logger.info(f"CSRF token extracted: {csrf_token[:20]}...")

        # Create unique test username to avoid conflicts
        test_username = f"testuser_csrf_{int(time.time() * 1000)}"
        test_password = "TestPass123"

        # Register user first
        register_data = {
            "username": test_username,
            "password": test_password,
            "confirm_password": test_password,
            "acknowledge": "true",
            "csrf_token": csrf_token,
        }
        register_response = client.post(
            "/auth/register", data=register_data, follow_redirects=False
        )
        logger.info(
            f"Register response status: {register_response.status_code}"
        )

        # After registration, check if we were redirected and handle accordingly
        if register_response.status_code == 302:
            # We might be auto-logged in after registration, so go directly to auth check
            response = client.get("/auth/check")
            if response.status_code == 200:
                data = response.get_json()
                if data.get("authenticated") is True:
                    logger.info(
                        "✅ User auto-authenticated after registration!"
                    )
                    logger.info(f"Auth check: {data}")
                    return

        # Get fresh CSRF token for login
        response = client.get("/auth/login")
        logger.info(
            f"Login page status after registration: {response.status_code}"
        )

        # If we're redirected, we might already be logged in
        if response.status_code == 302:
            response = client.get("/auth/check")
            if response.status_code == 200:
                data = response.get_json()
                if data.get("authenticated") is True:
                    logger.info("✅ User already authenticated!")
                    logger.info(f"Auth check: {data}")
                    return

        csrf_match = re.search(
            r'name="csrf_token" value="([^"]+)"', response.data.decode()
        )
        assert csrf_match is not None, (
            f"Could not find CSRF token in login page after registration. Status: {response.status_code}, Content preview: {response.data.decode()[:200]}"
        )
        csrf_token = csrf_match.group(1)

        # Step 2: Submit login form with CSRF token
        login_data = {
            "username": test_username,
            "password": test_password,
            "csrf_token": csrf_token,
        }

        logger.info("Step 2: Submitting login form...")
        response = client.post(
            "/auth/login",
            data=login_data,
            headers={
                "Referer": "http://localhost/auth/login"
            },  # Some CSRF checks require referer
            follow_redirects=False,
        )

        logger.info(f"POST /auth/login -> {response.status_code}")

        assert response.status_code == 302, "Login should redirect on success"
        logger.info("✅ Login successful!")

        # Step 3: Follow redirect and test authenticated access
        location = response.headers.get("Location", "/")
        response = client.get(location)
        logger.info(f"GET {location} -> {response.status_code}")
        assert response.status_code == 200

        # Test auth check
        response = client.get("/auth/check")
        logger.info(f"GET /auth/check -> {response.status_code}")
        assert response.status_code == 200

        data = response.get_json()
        logger.info(f"Auth check: {data}")
        assert data.get("authenticated") is True
        logger.info("✅ Authentication test passed!")

    def test_login_without_csrf(self, app):
        """Test that login fails without CSRF token."""
        # Enable CSRF for this test
        app.config["WTF_CSRF_ENABLED"] = True
        client = app.test_client()

        # Try to login without CSRF token
        login_data = {
            "username": "testuser",
            "password": "TestPass123",
            # No csrf_token
        }

        response = client.post(
            "/auth/login", data=login_data, follow_redirects=False
        )

        # Should fail due to missing CSRF token
        assert response.status_code != 302  # Should not redirect (success)
        logger.info("✅ Login correctly rejected without CSRF token")

    def test_login_with_invalid_csrf(self, app):
        """Test that login fails with invalid CSRF token."""
        # Enable CSRF for this test
        app.config["WTF_CSRF_ENABLED"] = True
        client = app.test_client()

        # Get login page
        response = client.get("/auth/login")
        assert response.status_code == 200

        # Use an invalid CSRF token
        login_data = {
            "username": "testuser",
            "password": "TestPass123",
            "csrf_token": "invalid_token_12345",
        }

        response = client.post(
            "/auth/login", data=login_data, follow_redirects=False
        )

        # Should fail due to invalid CSRF token
        assert response.status_code != 302  # Should not redirect (success)
        logger.info("✅ Login correctly rejected with invalid CSRF token")

    def test_csrf_token_rotation(self, client):
        """Test that CSRF tokens are rotated between requests."""
        # Get first CSRF token
        response = client.get("/auth/login")
        csrf_match1 = re.search(
            r'name="csrf_token" value="([^"]+)"', response.data.decode()
        )
        csrf_token1 = csrf_match1.group(1) if csrf_match1 else None

        # Get second CSRF token
        response = client.get("/auth/login")
        csrf_match2 = re.search(
            r'name="csrf_token" value="([^"]+)"', response.data.decode()
        )
        csrf_token2 = csrf_match2.group(1) if csrf_match2 else None

        # Tokens should be different (rotation) or at least valid
        assert csrf_token1 is not None
        assert csrf_token2 is not None
        logger.info("✅ CSRF tokens are properly generated")