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

"""Integration tests for the egress-policy PEP inside the notification
dispatch path (``notifications/manager.py``).

These drive the REAL ``NotificationManager._filter_urls_by_egress_policy``
and the real ``send_notification`` / ``test_service`` call sites. They assert
the *policy decision* (vendor/cloud webhook filtered vs. allowed, fail-closed
on a corrupt scope, master-switch gating) which happens BEFORE any Apprise /
HTTP work. The only thing mocked is the dispatch boundary
(``service.send_event`` / ``service.test_service``) so nothing actually leaves
the box, plus the env master switches and — for http(s) hosts — the DNS-backed
host classifier, so the tests are in-process and deterministic.

Why filtering vendor webhooks matters: research results contain the user's
queries and retrieved local-corpus chunks. A ``discord://`` / ``slack://`` /
``https://hooks...`` notification URL would exfiltrate those to an external
vendor API. Under PRIVATE_ONLY ("nothing leaves the box") and, for public
http(s) hosts, STRICT, those URLs must be dropped before dispatch.
"""

from __future__ import annotations

from unittest.mock import MagicMock, patch

from local_deep_research.notifications.manager import NotificationManager
from local_deep_research.notifications.templates import EventType


# ---------------------------------------------------------------------------
# Construction helper. NotificationManager.__init__ reads two ENV-only
# master switches and builds a NotificationService (an in-process
# apprise.Apprise() — no network). We patch the env reads so the manager is
# built deterministically; everything else is the real code.
# ---------------------------------------------------------------------------


def _build_manager(snapshot, *, outbound=True, user_id="notif-egress-test"):
    def fake_env(key, default=None):
        if key == "notifications.allow_outbound":
            return outbound
        if key == "notifications.allow_private_ips":
            return False
        return default

    with patch(
        "local_deep_research.notifications.manager.get_env_setting",
        side_effect=fake_env,
    ):
        mgr = NotificationManager(snapshot, user_id=user_id)
    # Sever the real dispatch path: any send must go through this mock, never
    # apprise/HTTP. Tests assert on whether/with-what it is called.
    mgr.service.send_event = MagicMock(return_value=True)
    mgr.service.test_service = MagicMock(
        return_value={"success": True, "message": "ok"}
    )
    return mgr


def _snapshot(scope, *, tool="arxiv", service_url=None):
    snap = {
        "policy.egress_scope": scope,
        "search.tool": tool,
    }
    if service_url is not None:
        snap["notifications.service_url"] = service_url
    return snap


_CLASSIFY = "local_deep_research.security.egress.policy._classify_host"


# ---------------------------------------------------------------------------
# A. Vendor (non-http) scheme: the core exfiltration guard.
# ---------------------------------------------------------------------------


def test_private_only_filters_vendor_scheme_but_both_allows():
    """discord:// is a vendor API egress that cannot be verified local, so
    PRIVATE_ONLY must drop it; BOTH (the permissive scope) passes it through.
    """
    url = "discord://webhook_id/token"

    deny_mgr = _build_manager(_snapshot("private_only"))
    assert deny_mgr._filter_urls_by_egress_policy(url) == ""

    allow_mgr = _build_manager(_snapshot("both"))
    assert allow_mgr._filter_urls_by_egress_policy(url) == url


def test_private_only_mixed_urls_keeps_only_local_https():
    """A space-separated Apprise string mixing a vendor scheme, a public
    cloud webhook, and a self-hosted local webhook: under PRIVATE_ONLY only
    the local https survives. Drives both the non-http branch and the real
    evaluate_url http branch in one shot.
    """
    urls = (
        "slack://tokA/tokB/tokC "
        "https://hooks.bad.test/services/x "
        "https://intranet.local/hook"
    )

    def fake_classify(host, ctx, *a, **k):
        return host == "intranet.local"

    mgr = _build_manager(_snapshot("private_only"))
    with patch(_CLASSIFY, side_effect=fake_classify):
        result = mgr._filter_urls_by_egress_policy(urls)

    assert result == "https://intranet.local/hook"


# ---------------------------------------------------------------------------
# C. STRICT: refuses public http(s) hosts (provenance-spoofing risk).
# ---------------------------------------------------------------------------


def test_strict_filters_public_https_webhook_but_both_allows():
    url = "https://hooks.bad.test/services/x"

    # Public host classification (False), deterministic — no real DNS.
    with patch(_CLASSIFY, return_value=False):
        deny_mgr = _build_manager(_snapshot("strict"))
        assert deny_mgr._filter_urls_by_egress_policy(url) == ""

        allow_mgr = _build_manager(_snapshot("both"))
        assert allow_mgr._filter_urls_by_egress_policy(url) == url


# ---------------------------------------------------------------------------
# D. PUBLIC_ONLY: refuses a private/intranet webhook, allows the public one.
# ---------------------------------------------------------------------------


def test_public_only_filters_local_webhook_but_allows_public():
    local_url = "https://intranet.local/hook"
    public_url = "https://hooks.bad.test/services/x"

    def fake_classify(host, ctx, *a, **k):
        return host == "intranet.local"

    mgr = _build_manager(_snapshot("public_only"))
    with patch(_CLASSIFY, side_effect=fake_classify):
        assert mgr._filter_urls_by_egress_policy(local_url) == ""
        assert mgr._filter_urls_by_egress_policy(public_url) == public_url


# ---------------------------------------------------------------------------
# E. Corrupt / unevaluable scope must fail CLOSED (filter, not dispatch).
# ---------------------------------------------------------------------------


def test_corrupt_scope_fails_closed():
    url = "discord://webhook_id/token"

    # A garbage scope makes context_from_snapshot raise PolicyDeniedError;
    # the PEP must refuse ALL urls rather than fall open to unfiltered send.
    deny_mgr = _build_manager(_snapshot("totally-not-a-scope"))
    assert deny_mgr._filter_urls_by_egress_policy(url) == ""

    # Sanity pair: a valid permissive scope on the same input passes it.
    allow_mgr = _build_manager(_snapshot("both"))
    assert allow_mgr._filter_urls_by_egress_policy(url) == url


# ---------------------------------------------------------------------------
# I. Snapshot-less back-compat branch: no snapshot => unchanged passthrough.
# ---------------------------------------------------------------------------


def test_snapshotless_manager_passes_through_but_private_only_filters():
    url = "discord://webhook_id/token"

    passthrough = _build_manager({})  # empty snapshot is falsy
    assert passthrough._filter_urls_by_egress_policy(url) == url

    filtered = _build_manager(_snapshot("private_only"))
    assert filtered._filter_urls_by_egress_policy(url) == ""


# ---------------------------------------------------------------------------
# F. send_notification: the filter decision actually gates real dispatch.
# ---------------------------------------------------------------------------


def test_send_notification_blocks_dispatch_when_policy_filters_all():
    snap_deny = _snapshot("private_only", service_url="discord://id/token")
    deny_mgr = _build_manager(snap_deny, outbound=True)
    # force=True bypasses per-user toggles + rate limit, NOT the egress filter.
    sent = deny_mgr.send_notification(EventType.TEST, {}, force=True)
    assert sent is False
    deny_mgr.service.send_event.assert_not_called()

    snap_allow = _snapshot("both", service_url="discord://id/token")
    allow_mgr = _build_manager(snap_allow, outbound=True)
    sent = allow_mgr.send_notification(EventType.TEST, {}, force=True)
    assert sent is True
    allow_mgr.service.send_event.assert_called_once()
    # The allowed subset (the discord url) is what gets handed to dispatch.
    _, kwargs = allow_mgr.service.send_event.call_args
    assert kwargs["service_urls"] == "discord://id/token"


# ---------------------------------------------------------------------------
# G. Env master switch gates dispatch independently of the policy decision.
# ---------------------------------------------------------------------------


def test_outbound_master_switch_blocks_even_policy_allowed_url():
    snap = _snapshot("both", service_url="discord://id/token")

    # Outbound OFF: a BOTH-scope, policy-allowed url is still not dispatched.
    off_mgr = _build_manager(snap, outbound=False)
    assert off_mgr.send_notification(EventType.TEST, {}, force=True) is False
    off_mgr.service.send_event.assert_not_called()

    # Flipping ONLY the master switch on lets the same url through.
    on_mgr = _build_manager(snap, outbound=True)
    assert on_mgr.send_notification(EventType.TEST, {}, force=True) is True
    on_mgr.service.send_event.assert_called_once()


# ---------------------------------------------------------------------------
# H. test_service (the /api/notifications/test-url PEP) honors the policy.
# ---------------------------------------------------------------------------


def test_test_service_endpoint_respects_private_only():
    url = "discord://webhook_id/token"

    deny_mgr = _build_manager(_snapshot("private_only"))
    result = deny_mgr.test_service(url)
    assert result["status"] == "error"
    assert "egress policy" in result["message"].lower()
    deny_mgr.service.test_service.assert_not_called()

    allow_mgr = _build_manager(_snapshot("both"))
    allow_mgr.test_service(url)
    allow_mgr.service.test_service.assert_called_once_with(url)