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

"""Integration tests for the settings-save egress validators.

These drive the REAL cross-field policy validators in
``security/egress/validators.py`` exactly as the settings write routes invoke
them (``web/routes/settings_routes.py``: ``save_all_settings`` /
``save_settings`` / ``api_update_setting``):

- ``validate_allowed_local_hostnames``: a PUBLIC hostname may not be smuggled
  into ``llm.allowed_local_hostnames`` (the host classifier would then trust
  external hosts as "local"); private / loopback hosts are accepted.

Each guarded property is asserted with an allow+deny pair so the test fails if
the rule were reverted. Direct-validator tests use realistic
``form_data`` / ``all_db_settings`` inputs; route-level tests drive the real
``api_update_setting`` PUT endpoint with only the settings/DB backend mocked,
proving the validation error is surfaced (not silently dropped).

All host classification uses literal IPs (8.8.8.8 public, 10.0.0.1 / 127.0.0.1
private), which the real classifier resolves OFFLINE via getaddrinfo on a
literal — no network round-trip, fully deterministic.
"""

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

import pytest
from flask import Flask

from local_deep_research.security.egress import validators
from local_deep_research.security.egress.validators import (
    validate_allowed_local_hostnames,
)
from local_deep_research.web.routes.settings_routes import settings_bp

HOSTS_KEY = "llm.allowed_local_hostnames"
SCOPE_KEY = "policy.egress_scope"
ENGINE_KEY = "search.tool"
MODULE = "local_deep_research.web.routes.settings_routes"
DECORATOR_MODULE = "local_deep_research.web.utils.route_decorators"


# ---------------------------------------------------------------------------
# Removed validator regression guard
# ---------------------------------------------------------------------------


def test_validate_strict_meta_combo_is_gone():
    """The meta-picker engines were removed, so the STRICT+meta-picker
    save-time validator must no longer exist (stray meta names are denied at
    runtime by ``evaluate_engine`` as ``engine_unknown`` instead)."""
    assert not hasattr(validators, "validate_strict_meta_combo")


# ---------------------------------------------------------------------------
# validate_allowed_local_hostnames  (no PUBLIC host in the local allowlist)
# ---------------------------------------------------------------------------


def test_public_host_in_local_allowlist_is_rejected():
    """A public IP would let the policy treat an external host as local."""
    err = validate_allowed_local_hostnames({HOSTS_KEY: ["8.8.8.8"]}, {})
    assert err is not None
    assert err["key"] == HOSTS_KEY
    assert "8.8.8.8" in err["error"]


def test_private_host_in_local_allowlist_is_accepted():
    """Allow side: an RFC1918 private address is a legitimate local host."""
    assert (
        validate_allowed_local_hostnames({HOSTS_KEY: ["10.0.0.1"]}, {}) is None
    )


def test_loopback_host_in_local_allowlist_is_accepted():
    """Loopback is local and must be accepted."""
    assert (
        validate_allowed_local_hostnames({HOSTS_KEY: ["127.0.0.1"]}, {}) is None
    )


def test_mixed_list_rejects_only_the_public_entries():
    """A list mixing private + public is rejected, naming only the public host."""
    err = validate_allowed_local_hostnames(
        {HOSTS_KEY: ["10.0.0.1", "8.8.8.8"]}, {}
    )
    assert err is not None
    assert "8.8.8.8" in err["error"]
    assert "10.0.0.1" not in err["error"]


def test_unresolvable_host_is_accepted_fail_open():
    """A name that does not resolve (DNS down / split-horizon) is accepted so a
    flaky-network user can still save; runtime classification still gates it.
    Resolution is stubbed to None to keep this deterministic and offline."""
    with patch.object(validators, "_resolve_with_timeout", return_value=None):
        assert (
            validate_allowed_local_hostnames(
                {HOSTS_KEY: ["intranet.corp.example"]}, {}
            )
            is None
        )


def test_public_host_still_rejected_when_resolution_succeeds():
    """Deny counterpart to the fail-open test: when resolution returns an
    address, a public host is rejected (the stub must not blanket-accept)."""
    addrinfo = [(2, 1, 6, "", ("8.8.8.8", 0))]
    with patch.object(
        validators, "_resolve_with_timeout", return_value=addrinfo
    ):
        err = validate_allowed_local_hostnames(
            {HOSTS_KEY: ["resolves-public.example"]}, {}
        )
    assert err is not None
    assert "resolves-public.example" in err["error"]


def test_json_string_list_of_private_host_is_accepted():
    """The save pipeline may hand the JSON-typed value as a JSON string."""
    assert (
        validate_allowed_local_hostnames({HOSTS_KEY: '["10.0.0.1"]'}, {})
        is None
    )


def test_json_string_list_of_public_host_is_rejected():
    """Deny counterpart for the JSON-string decode path."""
    err = validate_allowed_local_hostnames({HOSTS_KEY: '["8.8.8.8"]'}, {})
    assert err is not None
    assert "8.8.8.8" in err["error"]


def test_malformed_json_string_is_rejected():
    """A non-JSON string for a JSON-typed setting is a hard validation error."""
    err = validate_allowed_local_hostnames({HOSTS_KEY: "not json"}, {})
    assert err is not None
    assert err["key"] == HOSTS_KEY


def test_non_list_value_is_rejected():
    """A scalar (non-list) value is rejected — must be a list of hostnames."""
    err = validate_allowed_local_hostnames({HOSTS_KEY: 5}, {})
    assert err is not None


def test_hosts_key_absent_returns_none():
    """The guard is inert when its key is not part of the save."""
    assert validate_allowed_local_hostnames({"other.key": "x"}, {}) is None


# ---------------------------------------------------------------------------
# Route-level wiring: api_update_setting PUT surfaces the validator errors
# ---------------------------------------------------------------------------


def _make_setting(key, value, ui_element="text"):
    s = MagicMock()
    s.key = key
    s.value = value
    s.ui_element = ui_element
    s.editable = True
    return s


@contextmanager
def _routed_client(existing_settings):
    """Drive api_update_setting with the auth + DB backend mocked.

    ``existing_settings`` is a list of mock Setting rows returned by
    ``db_session.query(Setting).all()``; the per-key lookup
    ``.filter(...).first()`` returns the row whose key matches the request.
    """
    app = Flask(__name__)
    app.config["SECRET_KEY"] = "test-secret"
    app.config["WTF_CSRF_ENABLED"] = False
    app.register_blueprint(settings_bp)

    by_key = {s.key: s for s in existing_settings}

    db_session = MagicMock()

    def _query(_model):
        q = MagicMock()
        q.all.return_value = list(existing_settings)

        def _filter(*_a, **_k):
            fq = MagicMock()
            # api_update_setting filters on Setting.key == <key>; the key is
            # bound in the route, so resolve it from the live request.
            from flask import request as _req

            requested = _req.view_args.get("key")
            fq.first.return_value = by_key.get(requested)
            return fq

        q.filter.side_effect = _filter
        return q

    db_session.query.side_effect = _query

    @contextmanager
    def _fake_user_session(_username):
        yield db_session

    fake_db_manager = MagicMock()
    fake_db_manager.is_user_connected.return_value = True

    patches = [
        patch(
            "local_deep_research.web.auth.decorators.db_manager",
            fake_db_manager,
        ),
        patch(
            f"{DECORATOR_MODULE}.get_user_db_session",
            side_effect=_fake_user_session,
        ),
        patch(f"{MODULE}.settings_limit", lambda f: f),
        # Isolate the egress decision from adjacent type/coercion concerns.
        patch(f"{MODULE}.validate_setting", return_value=(True, None)),
        patch(
            f"{MODULE}.coerce_setting_for_write",
            side_effect=lambda **kw: kw["value"],
        ),
        patch(f"{MODULE}.set_setting", return_value=True),
        patch(f"{MODULE}.calculate_warnings", return_value=[]),
        patch(f"{MODULE}.invalidate_settings_caches", return_value=None),
    ]
    for p in patches:
        p.start()
    try:
        with app.test_client() as client:
            with client.session_transaction() as sess:
                sess["username"] = "testuser"
                sess["session_id"] = "sid"
            yield client
    finally:
        for p in reversed(patches):
            p.stop()


def test_route_rejects_public_local_hostname():
    """PUT llm.allowed_local_hostnames=[public IP] -> 400 with the policy error."""
    settings = [_make_setting(HOSTS_KEY, [], ui_element="json")]
    with _routed_client(settings) as client:
        resp = client.put(
            f"/settings/api/{HOSTS_KEY}", json={"value": ["8.8.8.8"]}
        )
    assert resp.status_code == 400
    assert "8.8.8.8" in resp.get_json()["error"]


def test_route_accepts_private_local_hostname():
    """Allow counterpart: a private IP passes the validator and is saved (200)."""
    settings = [_make_setting(HOSTS_KEY, [], ui_element="json")]
    with _routed_client(settings) as client:
        resp = client.put(
            f"/settings/api/{HOSTS_KEY}", json={"value": ["10.0.0.1"]}
        )
    assert resp.status_code == 200
    assert "error" not in resp.get_json()


def test_route_accepts_strict_scope_with_db_concrete_engine():
    """Allow counterpart: STRICT scope with a concrete DB engine saves (200)."""
    settings = [
        _make_setting(SCOPE_KEY, "both"),
        _make_setting(ENGINE_KEY, "arxiv"),
    ]
    with _routed_client(settings) as client:
        resp = client.put(
            f"/settings/api/{SCOPE_KEY}", json={"value": "strict"}
        )
    assert resp.status_code == 200
    assert "error" not in resp.get_json()


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(pytest.main([__file__, "-q"]))