learningcircuit--local-deep-research/tests/security/test_csrf_e2e_flow.py

# allow: no-sut-import — black-box HTTP test; drives real routes through the Flask test client
"""
End-to-end CSRF flow test for browser-facing API endpoints.

Verifies the full flow: login → CSRF token → API call with CSRF enabled,
ensuring the narrowed CSRF exemptions (only api_v1 exempt) work correctly
for browser-facing endpoints like /api/start_research.
"""

import re
import uuid

import pytest


class TestCSRFEndToEndFlow:
    """Test the complete CSRF flow from login through API usage."""

    @pytest.fixture
    def csrf_app(self, app):
        """Wrap the root conftest app fixture with CSRF enabled."""
        app.config["WTF_CSRF_ENABLED"] = True
        app.config["WTF_CSRF_CHECK_DEFAULT"] = True
        return app

    def test_full_csrf_flow_login_through_api_call(self, csrf_app):
        """Test complete browser-like flow: register → get CSRF token → API call."""
        client = csrf_app.test_client()

        with client:
            # Step 1: GET /auth/login to get a CSRF token for registration
            response = client.get("/auth/login")
            assert response.status_code == 200

            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"
            )
            form_csrf_token = csrf_match.group(1)

            # Step 2: Register (auto-logs in the user)
            username = f"csrf_e2e_{uuid.uuid4().hex[:12]}"
            password = "TestPass123!"

            register_response = client.post(
                "/auth/register",
                data={
                    "username": username,
                    "password": password,
                    "confirm_password": password,
                    "acknowledge": "true",
                    "csrf_token": form_csrf_token,
                },
                follow_redirects=False,
            )
            assert register_response.status_code == 302, (
                f"Registration should redirect, got {register_response.status_code}"
            )

            # Step 3: Get API CSRF token from /auth/csrf-token
            csrf_response = client.get("/auth/csrf-token")
            assert csrf_response.status_code == 200
            csrf_data = csrf_response.get_json()
            assert csrf_data is not None and "csrf_token" in csrf_data
            api_csrf_token = csrf_data["csrf_token"]

            # Step 4: POST /api/start_research WITH CSRF token — should pass CSRF
            response = client.post(
                "/api/start_research",
                json={"query": "test csrf flow"},
                headers={"X-CSRFToken": api_csrf_token},
            )
            # Expect 200 (success) or 400 from business logic — but not a CSRF rejection
            assert response.status_code in (200, 400), (
                f"Expected 200 or business-logic 400, got {response.status_code}"
            )
            if response.status_code == 400:
                data = response.get_json()
                error_msg = data.get("error", "") if data else ""
                assert "csrf" not in error_msg.lower(), (
                    f"CSRF should have passed but got: {error_msg}"
                )
                assert "session cookie" not in error_msg.lower(), (
                    f"CSRF should have passed but got: {error_msg}"
                )

            # Step 5: POST /api/start_research WITHOUT CSRF token — should be rejected
            response = client.post(
                "/api/start_research",
                json={"query": "test csrf flow"},
            )
            assert response.status_code == 400
            data = response.get_json()
            assert data is not None and "error" in data
            error_lower = data["error"].lower()
            assert "csrf" in error_lower or "session cookie" in error_lower, (
                f"Expected CSRF error but got: {data['error']}"
            )