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

"""Tests for security/rate_limiter.py."""

import importlib
import pytest
from unittest.mock import MagicMock, patch
from flask import Flask


def _reload_rate_limiter_with_limits(user_limit, ip_limit):
    """Reload security.rate_limiter with patched upload limits.

    Decorators bind the limit string at module-import time, so functional
    tests must reload the module after patching `load_server_config` to
    re-bind the upload decorators with the test-controlled values.
    """
    from local_deep_research.web import server_config as sc

    fake_config = {
        "host": "0.0.0.0",
        "port": 5000,
        "debug": False,
        "use_https": True,
        "allow_registrations": True,
        "rate_limit_default": "5000 per hour;50000 per day",
        "rate_limit_login": "5 per 15 minutes",
        "rate_limit_registration": "3 per hour",
        "rate_limit_settings": "30 per minute",
        "rate_limit_upload_user": user_limit,
        "rate_limit_upload_ip": ip_limit,
    }
    with patch.object(sc, "load_server_config", return_value=fake_config):
        from local_deep_research.security import rate_limiter

        importlib.reload(rate_limiter)
        return rate_limiter


@pytest.fixture
def app():
    """Create test Flask application."""
    app = Flask(__name__)
    app.config["TESTING"] = True
    return app


@pytest.fixture
def restore_rate_limiter_module():
    """Snapshot rate_limiter's module dict and restore it after the test.

    The functional tests reload the module to re-bind the upload decorators
    under test-controlled limits. Restoring the snapshot puts back the
    original objects (the module-level Limiter instance included), instead
    of reloading a third time under a hardcoded fake config — which would
    leave fake limits and a fresh Limiter in the module for later tests
    while earlier `from rate_limiter import ...` bindings kept the
    originals.
    """
    from local_deep_research.security import rate_limiter

    snapshot = dict(rate_limiter.__dict__)
    yield
    rate_limiter.__dict__.clear()
    rate_limiter.__dict__.update(snapshot)


class TestGetClientIp:
    """Tests for get_client_ip function."""

    def test_returns_first_ip_from_x_forwarded_for(self, app):
        """Test that first IP from X-Forwarded-For chain is returned."""
        with app.test_request_context(
            environ_base={
                "HTTP_X_FORWARDED_FOR": "192.168.1.100, 10.0.0.1, 172.16.0.1"
            }
        ):
            from local_deep_research.security.rate_limiter import get_client_ip

            result = get_client_ip()
            assert result == "192.168.1.100"

    def test_strips_whitespace_from_forwarded_ip(self, app):
        """Test that whitespace is stripped from forwarded IP."""
        with app.test_request_context(
            environ_base={"HTTP_X_FORWARDED_FOR": "  192.168.1.100  , 10.0.0.1"}
        ):
            from local_deep_research.security.rate_limiter import get_client_ip

            result = get_client_ip()
            assert result == "192.168.1.100"

    def test_returns_x_real_ip_when_no_forwarded_for(self, app):
        """Test that X-Real-IP is used when X-Forwarded-For is absent."""
        with app.test_request_context(
            environ_base={"HTTP_X_REAL_IP": "10.20.30.40"}
        ):
            from local_deep_research.security.rate_limiter import get_client_ip

            result = get_client_ip()
            assert result == "10.20.30.40"

    def test_strips_whitespace_from_real_ip(self, app):
        """Test that whitespace is stripped from X-Real-IP."""
        with app.test_request_context(
            environ_base={"HTTP_X_REAL_IP": "  10.20.30.40  "}
        ):
            from local_deep_research.security.rate_limiter import get_client_ip

            result = get_client_ip()
            assert result == "10.20.30.40"

    def test_falls_back_to_remote_address(self, app):
        """Test fallback to get_remote_address when no proxy headers."""
        with app.test_request_context():
            from local_deep_research.security.rate_limiter import get_client_ip

            # When no proxy headers are set, get_remote_address returns the remote addr
            result = get_client_ip()
            # Result should be some IP (default is typically 127.0.0.1)
            assert result is not None

    def test_prefers_x_forwarded_for_over_x_real_ip(self, app):
        """Test that X-Forwarded-For takes precedence over X-Real-IP."""
        with app.test_request_context(
            environ_base={
                "HTTP_X_FORWARDED_FOR": "192.168.1.1",
                "HTTP_X_REAL_IP": "10.0.0.1",
            }
        ):
            from local_deep_research.security.rate_limiter import get_client_ip

            result = get_client_ip()
            assert result == "192.168.1.1"


class TestRateLimiterConfiguration:
    """Tests for rate limiter configuration."""

    def test_limiter_uses_get_client_ip_as_key_func(self):
        """Test that limiter is configured with get_client_ip as key function."""
        from local_deep_research.security.rate_limiter import (
            limiter,
            get_client_ip,
        )

        assert limiter._key_func == get_client_ip

    def test_limiter_uses_memory_storage(self):
        """Test that limiter uses in-memory storage by default."""
        from local_deep_research.security.rate_limiter import limiter

        # The storage URI should be memory
        assert limiter._storage_uri == "memory://"

    def test_limiter_has_headers_enabled(self):
        """Test that rate limit headers are enabled."""
        from local_deep_research.security.rate_limiter import limiter

        assert limiter._headers_enabled is True


class TestRateLimitConstants:
    """Tests for rate limit configuration constants."""

    def test_default_rate_limit_loaded_from_config(self):
        """Test that DEFAULT_RATE_LIMIT is loaded from server config."""
        from local_deep_research.security.rate_limiter import (
            DEFAULT_RATE_LIMIT,
        )

        # Should be a string like "X per hour" or similar
        assert isinstance(DEFAULT_RATE_LIMIT, str)
        assert DEFAULT_RATE_LIMIT  # Not empty

    def test_login_rate_limit_loaded_from_config(self):
        """Test that LOGIN_RATE_LIMIT is loaded from server config."""
        from local_deep_research.security.rate_limiter import LOGIN_RATE_LIMIT

        assert isinstance(LOGIN_RATE_LIMIT, str)
        assert LOGIN_RATE_LIMIT  # Not empty

    def test_registration_rate_limit_loaded_from_config(self):
        """Test that REGISTRATION_RATE_LIMIT is loaded from server config."""
        from local_deep_research.security.rate_limiter import (
            REGISTRATION_RATE_LIMIT,
        )

        assert isinstance(REGISTRATION_RATE_LIMIT, str)
        assert REGISTRATION_RATE_LIMIT  # Not empty

    def test_settings_rate_limit_loaded_from_config(self):
        """Test that SETTINGS_RATE_LIMIT is loaded from config with default fallback."""
        from local_deep_research.security.rate_limiter import (
            SETTINGS_RATE_LIMIT,
        )

        assert isinstance(SETTINGS_RATE_LIMIT, str)
        assert SETTINGS_RATE_LIMIT  # Not empty
        # Should contain a rate expression like "30 per minute"
        assert "per" in SETTINGS_RATE_LIMIT


class TestSettingsLimit:
    """Tests for settings_limit shared rate limiter (PR #2021)."""

    def test_settings_limit_is_shared_limit(self):
        """Test that settings_limit is a SharedLimitItem from flask-limiter."""
        from local_deep_research.security.rate_limiter import settings_limit

        # SharedLimitItem has a __call__ method (it's a decorator)
        assert callable(settings_limit)

    def test_settings_limit_can_decorate_function(self):
        """Test that settings_limit can be used as a decorator."""
        from local_deep_research.security.rate_limiter import settings_limit

        @settings_limit
        def dummy_view():
            return "ok"

        # The decorated function should still be callable
        assert callable(dummy_view)

    def test_settings_limit_default_value(self):
        """Test that SETTINGS_RATE_LIMIT defaults to 30 per minute."""
        from local_deep_research.security.rate_limiter import (
            SETTINGS_RATE_LIMIT,
        )

        # The config.get uses "30 per minute" as default
        # If not set in config, it should be "30 per minute"
        assert isinstance(SETTINGS_RATE_LIMIT, str)
        assert SETTINGS_RATE_LIMIT  # Not empty


class TestApiRateLimit:
    """Tests for api_rate_limit shared limiter."""

    def test_api_rate_limit_is_callable(self):
        """Test that api_rate_limit can be used as a decorator."""
        from local_deep_research.security.rate_limiter import api_rate_limit

        assert callable(api_rate_limit)

    def test_get_user_api_rate_limit_default(self, app):
        """Test that _get_user_api_rate_limit returns default when no user."""

        with app.test_request_context():
            from local_deep_research.security.rate_limiter import (
                _get_user_api_rate_limit,
            )

            result = _get_user_api_rate_limit()
            assert result == 60

    def test_get_api_rate_limit_string_format(self, app):
        """Test that _get_api_rate_limit_string returns correct format."""
        from unittest.mock import patch

        with app.test_request_context():
            with patch(
                "local_deep_research.security.rate_limiter._get_user_api_rate_limit",
                return_value=30,
            ):
                from local_deep_research.security.rate_limiter import (
                    _get_api_rate_limit_string,
                )

                result = _get_api_rate_limit_string()
                assert result == "30 per minute"

    def test_is_api_rate_limit_exempt_no_user(self, app):
        """Test that unauthenticated requests are exempt."""
        with app.test_request_context():
            from local_deep_research.security.rate_limiter import (
                _is_api_rate_limit_exempt,
            )

            assert _is_api_rate_limit_exempt() is True

    def test_is_api_rate_limit_exempt_zero_limit(self, app):
        """Test that rate_limit=0 means exempt."""
        from unittest.mock import patch

        app.config["SECRET_KEY"] = "test-secret"
        with app.test_request_context():
            from flask import g

            g.current_user = "testuser"

            with patch(
                "local_deep_research.security.rate_limiter._get_user_api_rate_limit",
                return_value=0,
            ):
                from local_deep_research.security.rate_limiter import (
                    _is_api_rate_limit_exempt,
                )

                assert _is_api_rate_limit_exempt() is True

    def test_get_api_user_key_with_user(self, app):
        """Test that key function returns user-prefixed key."""
        with app.test_request_context():
            from flask import g

            g.current_user = "alice"

            from local_deep_research.security.rate_limiter import (
                _get_api_user_key,
            )

            result = _get_api_user_key()
            assert result == "api_user:alice"


class TestUploadRateLimit:
    """Tests for upload_rate_limit_user and upload_rate_limit_ip shared limiters."""

    def test_upload_rate_limit_user_exists_and_callable(self):
        """Test that upload_rate_limit_user is defined and callable."""
        from local_deep_research.security.rate_limiter import (
            upload_rate_limit_user,
        )

        assert upload_rate_limit_user is not None
        assert callable(upload_rate_limit_user)

    def test_upload_rate_limit_ip_exists_and_callable(self):
        """Test that upload_rate_limit_ip is defined and callable."""
        from local_deep_research.security.rate_limiter import (
            upload_rate_limit_ip,
        )

        assert upload_rate_limit_ip is not None
        assert callable(upload_rate_limit_ip)

    def test_get_upload_user_key_with_user(self, app):
        """Test that upload key function returns user-prefixed key."""
        with app.test_request_context():
            from flask import g

            g.current_user = "bob"

            from local_deep_research.security.rate_limiter import (
                _get_upload_user_key,
            )

            result = _get_upload_user_key()
            assert result == "upload_user:bob"

    def test_get_upload_user_key_no_user(self, app):
        """Test that upload key function falls back to IP when no user."""
        with app.test_request_context():
            from local_deep_research.security.rate_limiter import (
                _get_upload_user_key,
            )

            result = _get_upload_user_key()
            assert result.startswith("upload_ip:")


class TestUploadRateLimitFunctional:
    """Functional tests for dual-key upload rate limiting."""

    def test_upload_rate_limit_enforces_per_user(
        self, restore_rate_limiter_module
    ):
        """Per-user upload limit blocks after threshold."""
        rate_limiter = _reload_rate_limiter_with_limits(
            user_limit="3 per minute", ip_limit="100 per minute"
        )
        test_app = Flask(__name__)
        test_app.config["SECRET_KEY"] = "test"
        test_app.config["TESTING"] = True
        test_app.config["RATELIMIT_ENABLED"] = True
        test_app.config["RATELIMIT_STRATEGY"] = "moving-window"

        @test_app.route("/upload", methods=["POST"])
        @rate_limiter.upload_rate_limit_user
        @rate_limiter.upload_rate_limit_ip
        def upload():
            return "ok"

        rate_limiter.limiter.init_app(test_app)

        with test_app.test_client() as c:
            with c.session_transaction() as sess:
                sess["username"] = "uploader"
            # Per-user limit is "3 per minute" — first 3 pass, 4th is blocked
            for i in range(3):
                resp = c.post("/upload")
                assert resp.status_code == 200, f"Request {i + 1} should pass"
            resp = c.post("/upload")
            assert resp.status_code == 429

    def test_upload_per_user_limit_is_independent(
        self, restore_rate_limiter_module
    ):
        """Per-user upload bucket is keyed by username, not shared."""
        rate_limiter = _reload_rate_limiter_with_limits(
            user_limit="3 per minute", ip_limit="100 per minute"
        )
        test_app = Flask(__name__)
        test_app.config["SECRET_KEY"] = "test"
        test_app.config["TESTING"] = True
        test_app.config["RATELIMIT_ENABLED"] = True
        test_app.config["RATELIMIT_STRATEGY"] = "moving-window"

        # Only per-user limit (no per-IP) to isolate user-key behavior
        @test_app.route("/upload", methods=["POST"])
        @rate_limiter.upload_rate_limit_user
        def upload():
            return "ok"

        rate_limiter.limiter.init_app(test_app)

        # User A exhausts their per-user limit
        client_a = test_app.test_client()
        with client_a.session_transaction() as sess:
            sess["username"] = "user_a"
        for _ in range(3):
            client_a.post("/upload")
        assert client_a.post("/upload").status_code == 429

        # User B is unaffected (different user bucket)
        client_b = test_app.test_client()
        with client_b.session_transaction() as sess:
            sess["username"] = "user_b"
        assert client_b.post("/upload").status_code == 200

    def test_upload_rate_limit_respects_env_var_override(
        self, restore_rate_limiter_module
    ):
        """End-to-end: patched config flows through to decorator binding.

        Reloading rate_limiter after patching load_server_config simulates
        the env-var-set-at-process-start scenario without relying on a real
        env var (the module captures _config at import time).
        """
        rate_limiter = _reload_rate_limiter_with_limits(
            user_limit="2 per minute", ip_limit="2 per minute"
        )
        assert rate_limiter._UPLOAD_RATE_LIMIT_USER == "2 per minute"
        assert rate_limiter._UPLOAD_RATE_LIMIT_IP == "2 per minute"

        test_app = Flask(__name__)
        test_app.config["SECRET_KEY"] = "test"
        test_app.config["TESTING"] = True
        test_app.config["RATELIMIT_ENABLED"] = True
        test_app.config["RATELIMIT_STRATEGY"] = "moving-window"

        @test_app.route("/upload", methods=["POST"])
        @rate_limiter.upload_rate_limit_user
        @rate_limiter.upload_rate_limit_ip
        def upload():
            return "ok"

        rate_limiter.limiter.init_app(test_app)

        with test_app.test_client() as c:
            with c.session_transaction() as sess:
                sess["username"] = "envtest"
            assert c.post("/upload").status_code == 200
            assert c.post("/upload").status_code == 200
            assert c.post("/upload").status_code == 429


class TestApiRateLimitCaching:
    """Tests for g-caching and DB fallback in _get_user_api_rate_limit."""

    def test_g_cache_hit_skips_db(self, app):
        """Second call returns cached value without DB access."""
        with app.test_request_context():
            from flask import g
            from local_deep_research.security.rate_limiter import (
                _get_user_api_rate_limit,
            )

            g._api_rate_limit = 42
            result = _get_user_api_rate_limit()
            assert result == 42

    def test_db_exception_falls_back_to_default(self, app):
        """DB failure returns default 60 without raising."""
        app.config["SECRET_KEY"] = "test"
        with app.test_request_context():
            from flask import g
            from local_deep_research.security.rate_limiter import (
                _get_user_api_rate_limit,
            )

            g.current_user = "testuser"
            with patch(
                "local_deep_research.database.session_context.get_user_db_session",
                side_effect=RuntimeError("DB down"),
            ):
                result = _get_user_api_rate_limit()
                assert result == 60

    def test_custom_rate_limit_from_db(self, app):
        """User-configured rate limit is returned from DB."""
        app.config["SECRET_KEY"] = "test"
        with app.test_request_context():
            from flask import g
            from local_deep_research.security.rate_limiter import (
                _get_user_api_rate_limit,
            )

            g.current_user = "testuser"
            mock_sm = MagicMock()
            mock_sm.get_setting.return_value = 30

            mock_session = MagicMock()
            with patch(
                "local_deep_research.database.session_context.get_user_db_session"
            ) as mock_ctx:
                mock_ctx.return_value.__enter__ = MagicMock(
                    return_value=mock_session
                )
                mock_ctx.return_value.__exit__ = MagicMock(return_value=None)
                with patch(
                    "local_deep_research.utilities.db_utils.get_settings_manager",
                    return_value=mock_sm,
                ):
                    result = _get_user_api_rate_limit()
                    assert result == 30
                    assert g._api_rate_limit == 30


class TestSessionFallback:
    """Tests for session fallback when g.current_user is not set."""

    def test_api_user_key_uses_session_fallback(self, app):
        """Key function uses session username when g.current_user absent."""
        app.config["SECRET_KEY"] = "test"
        with app.test_request_context():
            from local_deep_research.security.rate_limiter import (
                _get_api_user_key,
            )

            with app.test_client() as c:
                with c.session_transaction() as sess:
                    sess["username"] = "session_user"
                with c.application.test_request_context():
                    from flask import session as s

                    s["username"] = "session_user"
                    result = _get_api_user_key()
                    assert result == "api_user:session_user"

    def test_upload_user_key_uses_session_fallback(self, app):
        """Upload key function uses session username when g.current_user absent."""
        app.config["SECRET_KEY"] = "test"
        with app.test_request_context():
            from local_deep_research.security.rate_limiter import (
                _get_upload_user_key,
            )

            with app.test_client() as c:
                with c.application.test_request_context():
                    from flask import session as s

                    s["username"] = "session_uploader"
                    result = _get_upload_user_key()
                    assert result == "upload_user:session_uploader"

    def test_g_current_user_takes_priority_over_session(self, app):
        """g.current_user is preferred over session username."""
        app.config["SECRET_KEY"] = "test"
        with app.test_request_context():
            from flask import g, session
            from local_deep_research.security.rate_limiter import (
                _get_api_user_key,
            )

            g.current_user = "g_user"
            session["username"] = "session_user"
            result = _get_api_user_key()
            assert result == "api_user:g_user"

    def test_exempt_uses_session_fallback(self, app):
        """Exempt check uses session username when g.current_user absent."""
        app.config["SECRET_KEY"] = "test"
        with app.test_request_context():
            from flask import session
            from local_deep_research.security.rate_limiter import (
                _is_api_rate_limit_exempt,
            )

            # No g.current_user, no session username → exempt
            assert _is_api_rate_limit_exempt() is True

            # Set session username → not exempt (has user, default limit > 0)
            session["username"] = "session_user"
            with patch(
                "local_deep_research.security.rate_limiter._get_user_api_rate_limit",
                return_value=60,
            ):
                assert _is_api_rate_limit_exempt() is False


class TestGetCurrentUsername:
    """Tests for get_current_username helper."""

    def test_returns_g_current_user(self, app):
        with app.test_request_context():
            from flask import g
            from local_deep_research.security.rate_limiter import (
                get_current_username,
            )

            g.current_user = "alice"
            assert get_current_username() == "alice"

    def test_returns_session_when_no_g_current_user(self, app):
        app.config["SECRET_KEY"] = "test"
        with app.test_request_context():
            from flask import session
            from local_deep_research.security.rate_limiter import (
                get_current_username,
            )

            session["username"] = "bob"
            assert get_current_username() == "bob"

    def test_returns_none_when_neither_set(self, app):
        with app.test_request_context():
            from local_deep_research.security.rate_limiter import (
                get_current_username,
            )

            assert get_current_username() is None

    def test_g_current_user_priority_over_session(self, app):
        app.config["SECRET_KEY"] = "test"
        with app.test_request_context():
            from flask import g, session
            from local_deep_research.security.rate_limiter import (
                get_current_username,
            )

            g.current_user = "g_user"
            session["username"] = "session_user"
            assert get_current_username() == "g_user"

    def test_empty_string_g_current_user_falls_to_session(self, app):
        """Empty string g.current_user is falsy — should fall through to session."""
        app.config["SECRET_KEY"] = "test"
        with app.test_request_context():
            from flask import g, session
            from local_deep_research.security.rate_limiter import (
                get_current_username,
            )

            g.current_user = ""
            session["username"] = "session_user"
            assert get_current_username() == "session_user"