learningcircuit--local-deep-research/tests/auth_tests/test_auth_rate_limiting.py

"""
Tests for authentication rate limiting (login and registration endpoints).
Tests that brute force protection is working correctly.
"""

import shutil
import tempfile
from pathlib import Path

import pytest


class TestAuthRateLimiting:
    """Test rate limiting on authentication endpoints."""

    @pytest.fixture
    def temp_data_dir(self):
        """Create a temporary data directory for testing."""
        temp_dir = tempfile.mkdtemp()
        yield Path(temp_dir)
        shutil.rmtree(temp_dir, ignore_errors=True)

    @pytest.fixture
    def app(self, temp_data_dir, monkeypatch):
        """Create a test Flask app with rate limiting."""
        monkeypatch.setenv("LDR_DATA_DIR", str(temp_data_dir))

        from local_deep_research.database.auth_db import init_auth_database
        from local_deep_research.database.encrypted_db import db_manager
        from local_deep_research.web.app_factory import create_app
        from local_deep_research.security.rate_limiter import limiter

        # Reset db_manager state
        db_manager.close_all_databases()
        db_manager.data_dir = temp_data_dir / "encrypted_databases"
        db_manager.data_dir.mkdir(parents=True, exist_ok=True)

        # Initialize auth database
        init_auth_database()

        app, _ = create_app()
        app.config["TESTING"] = True
        app.config["WTF_CSRF_ENABLED"] = False  # Disable CSRF for testing

        # In CI, rate limiting is disabled at startup so init_app() returns
        # early without initializing storage or registering error handlers.
        # Re-call init_app() with RATELIMIT_ENABLED=True to fully initialize.
        app.config["RATELIMIT_ENABLED"] = True
        limiter.enabled = True
        limiter.init_app(app)

        # Reset limiter storage between tests to prevent cross-test pollution
        limiter.reset()

        yield app

        # Restore original state
        limiter.enabled = True
        db_manager.close_all_databases()

    @pytest.fixture
    def client(self, app):
        """Create a test client."""
        return app.test_client()

    def test_login_rate_limit_allows_5_attempts(self, client):
        """Test that login allows 5 attempts before rate limiting."""
        # Make 5 login attempts - should all be allowed
        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: tighten to assert status_code == 401 (invalid creds), not membership of 3 codes).
        for i in range(5):
            response = client.post(
                "/auth/login",
                data={"username": f"testuser{i}", "password": "wrongpassword"},
                follow_redirects=False,
            )
            # Should get 401 (invalid credentials), NOT 429 (rate limit).
            # Tightened from `status_code in [200, 401, 400]` (PUNCHLIST
            # H8_STATUS_OR) — that broad list masked bugs where the auth
            # path silently returned 400 / 200 for wrong credentials.
            assert response.status_code == 401, (
                f"Attempt {i + 1}: wrong-credential login must return 401, "
                f"got {response.status_code}"
            )

    def test_login_rate_limit_blocks_6th_attempt(self, client):
        """Test that login blocks the 6th attempt within 15 minutes."""
        # Make 5 login attempts
        for i in range(5):
            client.post(
                "/auth/login",
                data={"username": f"testuser{i}", "password": "wrongpassword"},
            )

        # 6th attempt should be rate limited
        response = client.post(
            "/auth/login",
            data={"username": "testuser6", "password": "wrongpassword"},
        )
        assert response.status_code == 429, "6th attempt should be rate limited"

    def test_login_rate_limit_returns_proper_error(self, client):
        """Test that rate limit returns proper JSON error response."""
        # Trigger rate limit
        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: remove conditional — assert response.status_code == 429 unconditionally before checking body).
        for i in range(6):
            response = client.post(
                "/auth/login",
                data={"username": f"testuser{i}", "password": "wrongpassword"},
            )

        # Confirm rate limit actually fired before inspecting the body.
        # Previously the body checks were gated by `if response.status_code == 429`,
        # which silently passed when rate limiting didn't fire — masking real bugs.
        assert response.status_code == 429, (
            "6th attempt must be rate-limited; "
            f"got status {response.status_code}"
        )
        data = response.get_json()
        assert "error" in data
        assert "message" in data
        assert "Too many" in data["message"] or "Too many" in data["error"]

    def test_login_rate_limit_includes_retry_after_header(self, client):
        """Test that 429 response includes Retry-After header."""
        # Trigger rate limit
        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: remove conditional and assert 429 first).
        for i in range(6):
            response = client.post(
                "/auth/login",
                data={"username": f"testuser{i}", "password": "wrongpassword"},
            )

        # Confirm rate limit actually fired before inspecting headers.
        # Previously the header check was gated by `if response.status_code == 429`,
        # which silently passed when rate limiting didn't fire.
        assert response.status_code == 429, (
            "6th attempt must be rate-limited; "
            f"got status {response.status_code}"
        )
        assert (
            "Retry-After" in response.headers
            or "X-RateLimit-Reset" in response.headers
        ), "Rate limit response should include retry timing header"

    def test_registration_rate_limit_allows_3_attempts(self, client):
        """Test that registration allows 3 attempts before rate limiting."""
        # Make 3 registration attempts - should all be allowed
        for i in range(3):
            response = client.post(
                "/auth/register",
                data={
                    "username": f"newuser{i}",
                    "password": "TestPass123",
                    "confirm_password": "TestPass123",
                    "acknowledge": "true",
                },
                follow_redirects=False,
            )
            # Should get 200/302 (success/redirect) or 400 (validation error)
            # but not 429 (rate limit)
            assert response.status_code in [
                200,
                302,
                400,
            ], f"Attempt {i + 1} should not be rate limited"

    def test_registration_rate_limit_blocks_4th_attempt(self, client):
        """Test that registration blocks the 4th attempt within 1 hour."""
        # Make 3 registration attempts
        for i in range(3):
            client.post(
                "/auth/register",
                data={
                    "username": f"newuser{i}",
                    "password": "TestPass123",
                    "confirm_password": "TestPass123",
                    "acknowledge": "true",
                },
            )

        # 4th attempt should be rate limited
        response = client.post(
            "/auth/register",
            data={
                "username": "newuser4",
                "password": "TestPass123",
                "confirm_password": "TestPass123",
                "acknowledge": "true",
            },
        )
        assert response.status_code == 429, "4th attempt should be rate limited"

    def test_password_change_rate_limit_blocks_6th_attempt(self, client, app):
        """Test that password change blocks the 6th attempt."""
        # Disable exception propagation so template rendering errors
        # (e.g. missing 'research.index' endpoint) return 500 instead
        # of crashing the test.  Rate limiting fires before the handler,
        # so non-429 responses still count toward the limit.
        app.config["PROPAGATE_EXCEPTIONS"] = False
        app.config["TESTING"] = False

        with client.session_transaction() as sess:
            sess["username"] = "testuser"

        for i in range(5):
            response = client.post(
                "/auth/change-password",
                data={
                    "current_password": "",
                    "new_password": "NewStrongP4ss!",
                    "confirm_password": "NewStrongP4ss!",
                },
            )
            assert response.status_code != 429, (
                f"Attempt {i + 1} should not be rate limited"
            )

        # 6th attempt should be rate limited
        response = client.post(
            "/auth/change-password",
            data={
                "current_password": "",
                "new_password": "NewStrongP4ss!",
                "confirm_password": "NewStrongP4ss!",
            },
        )
        assert response.status_code == 429, (
            "6th password change attempt should be rate limited"
        )

    def test_password_change_has_separate_limit_from_login(self, client, app):
        """Test that login and password change have independent rate limits."""
        app.config["PROPAGATE_EXCEPTIONS"] = False
        app.config["TESTING"] = False

        # Exhaust login limit (5 attempts)
        for i in range(5):
            client.post(
                "/auth/login",
                data={"username": f"testuser{i}", "password": "wrongpassword"},
            )

        # Verify login is rate limited
        response = client.post(
            "/auth/login",
            data={"username": "testuser6", "password": "wrongpassword"},
        )
        assert response.status_code == 429, "Login should be rate limited"

        # Password change should still work (separate scope)
        with client.session_transaction() as sess:
            sess["username"] = "testuser"

        response = client.post(
            "/auth/change-password",
            data={
                "current_password": "",
                "new_password": "NewStrongP4ss!",
                "confirm_password": "NewStrongP4ss!",
            },
        )
        assert response.status_code != 429, (
            "Password change should have separate rate limit from login"
        )

    def test_different_endpoints_have_separate_limits(self, client):
        """Test that login and registration have independent rate limits."""
        # Exhaust login limit (5 attempts)
        for i in range(5):
            client.post(
                "/auth/login",
                data={"username": f"testuser{i}", "password": "wrongpassword"},
            )

        # Registration should still work (separate limit)
        response = client.post(
            "/auth/register",
            data={
                "username": "newuser1",
                "password": "TestPass123",
                "confirm_password": "TestPass123",
                "acknowledge": "true",
            },
        )
        assert response.status_code != 429, (
            "Registration should have separate rate limit from login"
        )

    def test_rate_limit_is_per_ip(self, client, app):
        """Test that rate limiting is applied per IP address."""
        # Make 5 requests from "IP 1"
        for i in range(5):
            with app.test_request_context(
                "/auth/login",
                method="POST",
                environ_base={"REMOTE_ADDR": "192.168.1.1"},
            ):
                client.post(
                    "/auth/login",
                    data={
                        "username": f"testuser{i}",
                        "password": "wrongpassword",
                    },
                    environ_base={"REMOTE_ADDR": "192.168.1.1"},
                )

        # 6th request from "IP 1" should be rate limited
        response1 = client.post(
            "/auth/login",
            data={"username": "testuser6", "password": "wrongpassword"},
            environ_base={"REMOTE_ADDR": "192.168.1.1"},
        )

        # Request from "IP 2" should still work (different IP)
        response2 = client.post(
            "/auth/login",
            data={"username": "testuser7", "password": "wrongpassword"},
            environ_base={"REMOTE_ADDR": "192.168.1.2"},
        )

        assert response1.status_code == 429, "IP 1 should be rate limited"
        assert response2.status_code != 429, (
            "IP 2 should not be rate limited (different IP)"
        )

    def test_proxy_headers_are_respected(self, client):
        """Test that X-Forwarded-For headers are used for rate limiting."""
        # Make 5 requests with same X-Forwarded-For header
        for i in range(5):
            client.post(
                "/auth/login",
                data={"username": f"testuser{i}", "password": "wrongpassword"},
                headers={"X-Forwarded-For": "10.0.0.1"},
            )

        # 6th request with same X-Forwarded-For should be rate limited
        response = client.post(
            "/auth/login",
            data={"username": "testuser6", "password": "wrongpassword"},
            headers={"X-Forwarded-For": "10.0.0.1"},
        )

        assert response.status_code == 429, (
            "Requests from same X-Forwarded-For IP should be rate limited"
        )

    def test_successful_login_still_counts_toward_limit(self, client):
        """Test that successful logins also count toward rate limit."""
        # This prevents attackers from resetting the limit with valid credentials
        # Create a test user first (this uses the programmatic API, not the web endpoint)
        from local_deep_research.database.encrypted_db import db_manager

        test_username = "ratelimituser"
        test_password = "TestPass123"

        # Create user if doesn't exist
        if not db_manager.user_exists(test_username):
            db_manager.create_user_database(test_username, test_password)

        # Make 5 successful login attempts
        for i in range(5):
            client.post(
                "/auth/login",
                data={"username": test_username, "password": test_password},
            )

        # 6th attempt should still be rate limited, even with valid credentials
        response = client.post(
            "/auth/login",
            data={"username": test_username, "password": test_password},
        )

        assert response.status_code == 429, (
            "Even successful logins should count toward rate limit"
        )

        # Clean up: delete test user
        db_manager.close_user_database(test_username)

    def test_account_enumeration_prevented(self, client):
        """Test that registration errors don't reveal username existence."""
        # First, register a real user so the second attempt below collides.
        # The `app` fixture scope is `function`, so the DB is fresh; the user
        # this test relies on must be created here rather than depending on
        # any other test's side effects.
        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: unconditionally assert both responses == 400 before content checks).
        existing_username = "enum_target_user"
        setup_response = client.post(
            "/auth/register",
            data={
                "username": existing_username,
                "password": "TestPass123",
                "confirm_password": "TestPass123",
                "acknowledge": "true",
            },
        )
        # Registration must succeed for the rest of the test to be meaningful.
        # Accept 200 or 302 (some flows redirect after success).
        assert setup_response.status_code in (200, 201, 302), (
            "Setup user registration must succeed; "
            f"got {setup_response.status_code}"
        )

        # Try to register with a username that definitely doesn't exist
        # but with an invalid password so validation rejects it.
        response1 = client.post(
            "/auth/register",
            data={
                "username": "definitelynonexistentuser12345",
                "password": "short",  # Will fail validation
                "confirm_password": "short",
                "acknowledge": "true",
            },
        )

        # Try to register with the username that now exists (collision)
        response2 = client.post(
            "/auth/register",
            data={
                "username": existing_username,
                "password": "TestPass123",
                "confirm_password": "TestPass123",
                "acknowledge": "true",
            },
        )

        # Both should return generic errors, not revealing if username exists.
        # Previously the body checks were gated by
        # `if response1.status_code == 400 and response2.status_code == 400`,
        # which silently passed if either response was anything else — masking
        # account-enumeration regressions (the whole point of this test). Worse,
        # the test relied on a leftover user from a sibling test, which never
        # existed because the fixture is function-scoped, so response2 was
        # always a successful registration and the body checks were always
        # skipped.
        assert response1.status_code == 400, (
            "Short-password registration must return 400; "
            f"got {response1.status_code}"
        )
        assert response2.status_code == 400, (
            "Duplicate-username registration must return 400; "
            f"got {response2.status_code}"
        )

        # Check that error messages are generic
        data2 = response2.get_data(as_text=True)

        # Should NOT contain "Username already exists"
        assert "Username already exists" not in data2, (
            "Error should not reveal username existence"
        )
        # Should contain generic message
        assert (
            "Registration failed" in data2
            or "try a different username" in data2
        ), "Should use generic error message"


class TestRateLimitReset:
    """Test that rate limits reset after the time window."""

    @pytest.fixture
    def temp_data_dir(self):
        """Create a temporary data directory for testing."""
        temp_dir = tempfile.mkdtemp()
        yield Path(temp_dir)
        shutil.rmtree(temp_dir, ignore_errors=True)

    @pytest.fixture
    def app(self, temp_data_dir, monkeypatch):
        """Create a test Flask app with rate limiting."""
        monkeypatch.setenv("LDR_DATA_DIR", str(temp_data_dir))

        from local_deep_research.database.auth_db import init_auth_database
        from local_deep_research.database.encrypted_db import db_manager
        from local_deep_research.web.app_factory import create_app
        from local_deep_research.security.rate_limiter import limiter

        db_manager.close_all_databases()
        db_manager.data_dir = temp_data_dir / "encrypted_databases"
        db_manager.data_dir.mkdir(parents=True, exist_ok=True)

        init_auth_database()

        app, _ = create_app()
        app.config["TESTING"] = True
        app.config["WTF_CSRF_ENABLED"] = False

        # Enable rate limiting AFTER init_app (CI sets LDR_DISABLE_RATE_LIMITING=true)
        app.config["RATELIMIT_ENABLED"] = True
        limiter.enabled = True
        limiter.init_app(app)
        limiter.reset()

        yield app

        db_manager.close_all_databases()

    @pytest.fixture
    def client(self, app):
        """Create a test client."""
        return app.test_client()

    @pytest.mark.slow
    def test_rate_limit_resets_after_time_window(self, client):
        """Test that rate limit resets after 15 minutes (for login)."""
        # Note: This test would take 15 minutes to run in real time
        # In practice, you'd mock time or use a shorter limit for testing
        pytest.skip(
            "This test requires mocking time or waiting 15 minutes - "
            "implement with time mocking if needed"
        )

        # Implementation would look like:
        # 1. Make 5 login attempts
        # 2. Verify 6th attempt is blocked
        # 3. Fast-forward time by 15 minutes (using time mocking)
        # 4. Verify new attempt is allowed