/tests/web/routes/test_settings_routes_security.py

"""
HTTP security tests for settings routes.

Existing unit tests cover corrupted value detection at the function level,
but not via HTTP endpoints. This file tests the save_all_settings endpoint
end-to-end.

Source: src/local_deep_research/web/routes/settings_routes.py
"""

from contextlib import contextmanager
from unittest.mock import Mock, patch

from ._settings_route_helpers import _create_test_app


# ---------------------------------------------------------------------------
# Test Infrastructure
# ---------------------------------------------------------------------------


@contextmanager
def _authenticated_client(app, mock_settings=None):
    """Provide test client with mocked auth and DB session for settings routes."""
    mock_db = Mock()
    mock_db.connections = {"testuser": True}
    mock_db.has_encryption = False

    # Build mock db session with Setting query
    mock_setting_obj = Mock()
    mock_setting_obj.key = "llm.temperature"
    mock_setting_obj.value = "0.7"
    mock_setting_obj.editable = True
    mock_setting_obj.ui_element = "number"

    _mock_query = Mock()
    _mock_query.all.return_value = mock_settings or [mock_setting_obj]
    _mock_query.first.return_value = None
    _mock_query.filter_by.return_value = _mock_query
    _mock_query.filter.return_value = _mock_query

    _mock_db_session = Mock()
    _mock_db_session.query.return_value = _mock_query

    @contextmanager
    def _fake_session(*args, **kwargs):
        yield _mock_db_session

    _routes_mod = "local_deep_research.web.routes.settings_routes"
    _decorator_mod = "local_deep_research.web.utils.route_decorators"

    patches = [
        patch("local_deep_research.web.auth.decorators.db_manager", mock_db),
        patch(
            f"{_decorator_mod}.get_user_db_session", side_effect=_fake_session
        ),
        patch(
            f"{_routes_mod}.settings_limit", lambda f: f
        ),  # disable rate limit
    ]

    try:
        for p in patches:
            p.start()
        with app.test_client() as client:
            with client.session_transaction() as sess:
                sess["username"] = "testuser"
                sess["session_id"] = "test-session-id"
            yield client, _mock_db_session
    finally:
        for p in reversed(patches):
            p.stop()


# ---------------------------------------------------------------------------
# Corrupted value auto-correction via HTTP
# ---------------------------------------------------------------------------


class TestSaveAllSettingsValidation:
    """Additional HTTP tests for save_all_settings endpoint."""

    def test_empty_json_body_returns_400(self):
        """POST with empty JSON object triggers 'No settings data provided'."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _):
            resp = client.post(
                "/settings/save_all_settings",
                json={},
                content_type="application/json",
            )
            # Empty dict is falsy, so save_all_settings returns 400
            assert resp.status_code == 400
            data = resp.get_json()
            assert data["status"] == "error"
            assert "No settings data" in data["message"]


# ---------------------------------------------------------------------------
# Namespace validation gate (new-key creation)
# ---------------------------------------------------------------------------


class TestNewSettingNamespaceGate:
    """The three write routes must reject new keys outside allowed namespaces."""

    def test_put_api_rejects_blocked_prefix_with_400(self):
        """PUT to a new key under a blocked prefix returns 400, not 403/201."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _):
            resp = client.put(
                "/settings/api/security.evil",
                json={"value": "bad"},
                content_type="application/json",
            )
            assert resp.status_code == 400
            data = resp.get_json()
            assert "not allowed" in data["error"].lower()

    def test_put_api_rejects_unknown_prefix_with_400(self):
        """Unknown prefixes (neither allow nor block) also return 400."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _):
            resp = client.put(
                "/settings/api/custom.foo",
                json={"value": "x"},
                content_type="application/json",
            )
            assert resp.status_code == 400

    def test_put_api_rejects_trailing_dot_under_allowed_prefix_with_400(self):
        """A trailing-dot key under an ALLOWED prefix (the #4840 shape) is
        rejected with 400 — the malformed-key guard runs ahead of the
        namespace allow-list, so ``local_search_chunk_size.`` never persists."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _):
            for key in ("local_search_chunk_size.", "llm."):
                resp = client.put(
                    f"/settings/api/{key}",
                    json={"value": 1000},
                    content_type="application/json",
                )
                assert resp.status_code == 400, key
                # The error must name the real problem (malformed key), not
                # misdirect to namespaces — the prefix here IS allowed.
                assert "malformed" in resp.get_json()["error"].lower(), key

    def test_save_all_settings_rejects_blocked_prefix(self):
        """save_all_settings rejects new keys in blocked namespaces via validation_errors."""
        app = _create_test_app()
        # Empty DB so the key is a create attempt, not an update.
        with _authenticated_client(app, mock_settings=[]) as (client, _):
            resp = client.post(
                "/settings/save_all_settings",
                json={"security.admin_override": True},
                content_type="application/json",
            )
            assert resp.status_code == 400
            data = resp.get_json()
            assert data["status"] == "error"
            assert any(
                e["key"] == "security.admin_override"
                and "not allowed" in e["error"]
                for e in data["errors"]
            )

    def test_save_all_settings_rejects_unknown_prefix(self):
        """save_all_settings rejects new keys with unknown prefixes."""
        app = _create_test_app()
        with _authenticated_client(app, mock_settings=[]) as (client, _):
            resp = client.post(
                "/settings/save_all_settings",
                json={"custom.injected": 1},
                content_type="application/json",
            )
            assert resp.status_code == 400
            data = resp.get_json()
            assert any(e["key"] == "custom.injected" for e in data["errors"])

    def test_save_settings_form_post_rejects_blocked_prefix(self):
        """save_settings (non-JS form-POST fallback) rejects blocked namespaces.

        This is the bypass path the original PR #3088 left unguarded — an
        attacker switching from AJAX to form POST must not be able to inject
        `security.*` / `auth.*` / `bootstrap.*` keys. settings_manager.set_setting
        must not be called for rejected keys.
        """
        app = _create_test_app()
        # The @with_user_session decorator constructs SettingsManager(db_session)
        # directly; patch the SettingsManager class at the decorator's import site.
        mock_sm_instance = Mock()
        mock_sm_instance.set_setting.return_value = True
        # Empty DB so the key is genuinely a create attempt, not an update.
        with _authenticated_client(app, mock_settings=[]) as (client, _):
            with patch(
                "local_deep_research.web.utils.route_decorators.SettingsManager",
                return_value=mock_sm_instance,
            ):
                resp = client.post(
                    "/settings/save_settings",
                    data={"security.admin_override": "true"},
                )
            # The route redirects; the rejection itself is signalled via flash.
            assert resp.status_code == 302
            # The rejected key must NOT have reached set_setting.
            for call in mock_sm_instance.set_setting.call_args_list:
                assert call.args[0] != "security.admin_override", (
                    f"Blocked key reached set_setting: {call}"
                )

    def test_save_settings_form_post_rejects_unknown_prefix(self):
        """save_settings rejects unknown (non-allow-listed) prefixes too."""
        app = _create_test_app()
        mock_sm_instance = Mock()
        mock_sm_instance.set_setting.return_value = True
        with _authenticated_client(app, mock_settings=[]) as (client, _):
            with patch(
                "local_deep_research.web.utils.route_decorators.SettingsManager",
                return_value=mock_sm_instance,
            ):
                resp = client.post(
                    "/settings/save_settings",
                    data={"custom.injected": "x"},
                )
            assert resp.status_code == 302
            for call in mock_sm_instance.set_setting.call_args_list:
                assert call.args[0] != "custom.injected", (
                    f"Unknown-prefix key reached set_setting: {call}"
                )

    def test_save_settings_form_post_allows_known_prefix(self):
        """save_settings still writes legitimate keys in the allow-list."""
        app = _create_test_app()
        mock_sm_instance = Mock()
        mock_sm_instance.set_setting.return_value = True
        with _authenticated_client(app, mock_settings=[]) as (client, _):
            with patch(
                "local_deep_research.web.utils.route_decorators.SettingsManager",
                return_value=mock_sm_instance,
            ):
                resp = client.post(
                    "/settings/save_settings",
                    data={"llm.new_temperature": "0.5"},
                )
            assert resp.status_code == 302
            # Legitimate key DID reach set_setting.
            assert any(
                call.args[0] == "llm.new_temperature"
                for call in mock_sm_instance.set_setting.call_args_list
            ), "Legitimate allow-listed key did not reach set_setting"


class TestSaveSettingsPasswordNoop:
    """save_settings (no-JS form POST) must not wipe a stored password when
    an empty value or the redaction sentinel is submitted — the same
    write-back guard the JSON save paths have."""

    def _password_setting(self):
        s = Mock()
        s.key = "llm.openai.api_key"
        s.value = "sk-existing"
        s.editable = True
        s.ui_element = "password"
        return s

    def _post(self, submitted_value):
        app = _create_test_app()
        pw = self._password_setting()
        mock_sm = Mock()
        mock_sm.set_setting.return_value = True
        with _authenticated_client(app, mock_settings=[pw]) as (client, _):
            with patch(
                "local_deep_research.web.utils.route_decorators.SettingsManager",
                return_value=mock_sm,
            ):
                resp = client.post(
                    "/settings/save_settings",
                    data={"llm.openai.api_key": submitted_value},
                )
        return resp, mock_sm

    def test_empty_password_form_post_is_noop(self):
        resp, mock_sm = self._post("")
        assert resp.status_code == 302
        assert all(
            call.args[0] != "llm.openai.api_key"
            for call in mock_sm.set_setting.call_args_list
        ), "Empty password reached set_setting via save_settings"

    def test_redacted_sentinel_form_post_is_noop(self):
        from local_deep_research.security.data_sanitizer import DataSanitizer

        resp, mock_sm = self._post(DataSanitizer.REDACTION_TEXT)
        assert resp.status_code == 302
        assert all(
            call.args[0] != "llm.openai.api_key"
            for call in mock_sm.set_setting.call_args_list
        ), "Redaction sentinel reached set_setting via save_settings"


# ---------------------------------------------------------------------------
# Unit tests for _is_allowed_new_setting_key guards
# ---------------------------------------------------------------------------


class TestIsAllowedNewSettingKey:
    """Helper guards: type, empty-string, double-dot, block-then-allow ordering."""

    def test_rejects_non_string(self):
        from local_deep_research.web.routes.settings_routes import (
            _is_allowed_new_setting_key,
        )

        assert _is_allowed_new_setting_key(None) is False
        assert _is_allowed_new_setting_key(42) is False
        assert _is_allowed_new_setting_key(["llm.model"]) is False

    def test_rejects_empty_and_whitespace_only(self):
        from local_deep_research.web.routes.settings_routes import (
            _is_allowed_new_setting_key,
        )

        assert _is_allowed_new_setting_key("") is False
        # Whitespace-only doesn't match any allowed prefix.
        assert _is_allowed_new_setting_key("   ") is False

    def test_rejects_double_dot(self):
        from local_deep_research.web.routes.settings_routes import (
            _is_allowed_new_setting_key,
        )

        # Even with a valid leading prefix, double-dot is malformed.
        assert _is_allowed_new_setting_key("llm..foo") is False
        assert _is_allowed_new_setting_key("search.engine..x") is False

    def test_rejects_trailing_and_leading_dot(self):
        """Trailing-/leading-dot keys are malformed even under an allowed
        prefix — these are the rows that corrupt prefix lookups (#4840)."""
        from local_deep_research.web.routes.settings_routes import (
            _is_allowed_new_setting_key,
        )

        # Trailing dot: passes the prefix allow-list but is malformed.
        assert _is_allowed_new_setting_key("llm.") is False
        assert _is_allowed_new_setting_key("search.max_results.") is False
        assert _is_allowed_new_setting_key("local_search_chunk_size.") is False
        # Leading dot and stray whitespace.
        assert _is_allowed_new_setting_key(".llm.model") is False
        assert _is_allowed_new_setting_key(" llm.model") is False
        assert _is_allowed_new_setting_key("llm.model ") is False

    def test_block_list_wins_over_allow_list(self):
        from local_deep_research.web.routes.settings_routes import (
            _is_allowed_new_setting_key,
        )

        # Even if a future allow-list contained "security.", the block-list
        # would still reject; check the current lists too.
        assert _is_allowed_new_setting_key("security.foo") is False
        assert _is_allowed_new_setting_key("auth.token") is False
        assert _is_allowed_new_setting_key("bootstrap.x") is False
        assert _is_allowed_new_setting_key("db_config.kdf_iterations") is False
        assert _is_allowed_new_setting_key("server.max_concurrent") is False
        assert _is_allowed_new_setting_key("testing.test_mode") is False

    def test_case_insensitive_prefix_match(self):
        """Uppercase keys are lowercased before prefix comparison."""
        from local_deep_research.web.routes.settings_routes import (
            _is_allowed_new_setting_key,
        )

        assert _is_allowed_new_setting_key("SECURITY.injected") is False
        assert _is_allowed_new_setting_key("LLM.custom_key") is True

    def test_allows_known_prefixes(self):
        from local_deep_research.web.routes.settings_routes import (
            _is_allowed_new_setting_key,
        )

        for key in (
            "app.flag",
            "backup.destination",
            "llm.model",
            "search.tool",
            "rag.chunk_size",
            "embeddings.ollama.url",
        ):
            assert _is_allowed_new_setting_key(key) is True, key