bambuddy/backend/tests/unit/test_outbound_url_ssrf_guards.py

"""Outbound-URL SSRF policy: two tiers, applied consistently.

Bambuddy makes outbound HTTP requests to hosts the operator configures. Which
policy applies is a property of the *service*, not the caller:

- LAN-service (Spoolman, ntfy, Bark, webhooks, Home Assistant, Obico ML, the
  slicer sidecars) — loopback and RFC-1918 MUST stay reachable, because
  self-hosting those next to Bambuddy is the normal topology. Blocking them
  would break most installs, which is why a blanket private-IP blocklist is
  the wrong fix here.
- Public-internet (OIDC issuer and icon URLs) — a private address cannot be a
  real IdP, so it is a probe.

Both tiers reject what is dangerous under any topology: non-HTTP schemes,
numeric-encoded IPs, cloud-metadata endpoints, multicast/unspecified, and
IPv4-mapped IPv6 encodings of the above.

The separate concern covered here is *response-body echo*. Notification
provider URLs are writable by anyone holding ``NOTIFICATIONS_CREATE`` — which
the default Operators group carries and which does NOT imply
``SETTINGS_UPDATE`` — and ``POST /notifications/test-config`` takes the URL
from the request body without persisting it. Returning the upstream body there
made an intended reachability check into an authenticated read primitive
against anything the process can reach. Providers whose host Bambuddy pins
(Pushover, Telegram, CallMeBot, Discord) may still echo, since the caller
cannot influence the destination.
"""

from __future__ import annotations

import inspect
import re

import httpx
import pytest

from backend.app.api.routes._oidc_helpers import assert_safe_public_https_url
from backend.app.api.routes._spoolman_helpers import assert_safe_spoolman_url
from backend.app.api.routes._url_safety import assert_safe_lan_service_url
from backend.app.schemas.auth import OIDCProviderCreate, OIDCProviderUpdate
from backend.app.schemas.settings import LAN_SERVICE_URL_SETTINGS, AppSettingsUpdate
from backend.app.services import notification_service as ns

# Dangerous under any topology — both tiers must reject all of these.
UNIVERSALLY_BLOCKED = [
    "file:///etc/passwd",
    "gopher://127.0.0.1:6379/_INFO",
    "ftp://internal.example.com/",
    "http://169.254.169.254/latest/meta-data/",
    "http://100.100.100.200/",
    "http://[fd00:ec2::254]/",
    "http://2130706433/",
    "http://0x7f000001/",
    "http://[::ffff:169.254.169.254]/",
    "http://0.0.0.0/",
    "http://239.255.255.250/",
]

# The normal self-hosted topology — the LAN tier must permit all of these.
LAN_ALLOWED = [
    "http://127.0.0.1:7912/",
    "http://localhost:3003",
    "http://192.168.1.50:8123",
    "http://10.0.0.7:3333",
    "http://172.16.4.9:8080",
    "https://ntfy.example.com/",
    "http://spoolman.lan:7912",
]


# ---------------------------------------------------------------------------
# The LAN-service tier
# ---------------------------------------------------------------------------


@pytest.mark.parametrize("url", UNIVERSALLY_BLOCKED)
def test_lan_tier_rejects_universally_dangerous_targets(url: str):
    with pytest.raises(ValueError):
        assert_safe_lan_service_url(url, label="Test URL")


@pytest.mark.parametrize("url", LAN_ALLOWED)
def test_lan_tier_permits_the_normal_self_hosted_topology(url: str):
    """A blanket private-IP block here would break most real installs."""
    assert_safe_lan_service_url(url, label="Test URL")


def test_lan_tier_names_the_field_in_its_error():
    with pytest.raises(ValueError, match="ntfy server URL"):
        assert_safe_lan_service_url("file:///etc/passwd", label="ntfy server URL")


def test_spoolman_wrapper_keeps_its_user_facing_wording():
    """The wording is asserted by pre-existing tests; delegation must not change it."""
    with pytest.raises(ValueError, match="^Spoolman URL must use http or https$"):
        assert_safe_spoolman_url("file:///etc/passwd")
    with pytest.raises(ValueError, match="^Spoolman URL must not point to a cloud metadata endpoint$"):
        assert_safe_spoolman_url("http://169.254.169.254/")


# ---------------------------------------------------------------------------
# The public-internet tier
# ---------------------------------------------------------------------------


@pytest.mark.parametrize("url", UNIVERSALLY_BLOCKED)
def test_public_tier_rejects_universally_dangerous_targets(url: str):
    with pytest.raises(ValueError):
        assert_safe_public_https_url(url)


@pytest.mark.parametrize(
    "url",
    [
        "https://127.0.0.1/",
        "https://192.168.1.5/",
        "https://10.1.2.3/",
        "https://[fe80::1]/",
        "https://[::ffff:127.0.0.1]/",
        "http://accounts.google.com/",  # scheme must be https
    ],
)
def test_public_tier_additionally_rejects_private_and_plain_http(url: str):
    with pytest.raises(ValueError):
        assert_safe_public_https_url(url)


# ---------------------------------------------------------------------------
# OIDC issuer_url — the encoding bypasses the hand-rolled validator missed
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "url",
    [
        "https://2130706433/",  # decimal-encoded 127.0.0.1
        "https://0x7f000001/",  # hex-encoded 127.0.0.1
        "https://[::ffff:127.0.0.1]/",  # IPv4-mapped loopback
        "https://[::ffff:169.254.169.254]/",  # IPv4-mapped IMDS
        "https://0.0.0.0/",
        "https://239.255.255.250/",
        "https://169.254.169.254/",
        "https://127.0.0.1/",
        "https://192.168.1.5/",
        "http://idp.example.com/",
    ],
)
def test_issuer_url_rejects_encoded_and_private_targets(url: str):
    with pytest.raises(ValueError):
        OIDCProviderCreate(
            name="SSO",
            issuer_url=url,
            client_id="cid",
            client_secret="secret",
        )


def test_issuer_url_update_is_guarded_too():
    """The update path matters most: it can change the issuer while the stored
    client_secret stays, which is the shape that would exfiltrate a real secret."""
    with pytest.raises(ValueError):
        OIDCProviderUpdate(issuer_url="https://[::ffff:127.0.0.1]/")


def test_issuer_url_error_names_the_field_not_the_icon():
    with pytest.raises(ValueError, match="issuer_url"):
        OIDCProviderUpdate(issuer_url="https://127.0.0.1/")


def test_a_real_idp_still_validates():
    provider = OIDCProviderCreate(
        name="SSO",
        issuer_url="https://accounts.google.com",
        client_id="cid",
        client_secret="secret",
    )
    assert provider.issuer_url == "https://accounts.google.com"


# ---------------------------------------------------------------------------
# Settings URLs
# ---------------------------------------------------------------------------

# Imported from the schema rather than duplicated, so the backstop below cannot
# silently disagree with what is actually validated.
LAN_SERVICE_SETTINGS = LAN_SERVICE_URL_SETTINGS


@pytest.mark.parametrize("field", LAN_SERVICE_SETTINGS)
@pytest.mark.parametrize("url", UNIVERSALLY_BLOCKED)
def test_settings_urls_reject_dangerous_targets(field: str, url: str):
    with pytest.raises(ValueError):
        AppSettingsUpdate(**{field: url})


@pytest.mark.parametrize("field", LAN_SERVICE_SETTINGS)
@pytest.mark.parametrize("url", LAN_ALLOWED)
def test_settings_urls_permit_lan_hosts(field: str, url: str):
    assert AppSettingsUpdate(**{field: url})


@pytest.mark.parametrize("field", LAN_SERVICE_SETTINGS)
@pytest.mark.parametrize("empty", ["", "   "])
def test_settings_urls_accept_empty_meaning_not_configured(field: str, empty: str):
    """Empty is the documented "fall back to the env var" value for all four."""
    assert AppSettingsUpdate(**{field: empty})


@pytest.mark.parametrize("field", LAN_SERVICE_SETTINGS)
@pytest.mark.parametrize(
    "legacy",
    [
        "192.168.1.10:3333",  # urlparse: scheme='', netloc='', hostname=None
        "localhost:3003",  # urlparse: scheme='localhost' (!), hostname=None
        "obico.local:3333",  # same trap, with dots
        "192.168.1.10",
    ],
)
def test_settings_urls_do_not_newly_reject_scheme_less_legacy_values(field: str, legacy: str):
    """Compatibility guard, not an endorsement.

    The settings inputs are plain text with no scheme enforcement, so values
    like these are already in the wild. They are inert — httpx raises
    UnsupportedProtocol, so no request is issued — and they were storable
    before the validator existed. Rejecting them now would block saves of
    unrelated fields bundled in the same request (the Obico panel auto-saves
    obico_ml_url alongside every other Obico setting).
    """
    assert AppSettingsUpdate(**{field: legacy})


@pytest.mark.parametrize("field", LAN_SERVICE_SETTINGS)
def test_settings_urls_still_reject_a_real_non_http_scheme(field: str):
    """The leniency above is scoped to strings that are not URLs at all."""
    with pytest.raises(ValueError):
        AppSettingsUpdate(**{field: "file:///etc/passwd"})


def test_every_url_setting_is_either_guarded_or_explicitly_exempt():
    """CI backstop: a new outbound-URL setting can't land unvalidated.

    Any new ``*_url`` field on AppSettingsUpdate must be added to the
    validator's field tuple or listed as exempt here with a reason. This
    catches the failure mode the original report correctly identified — guards
    added per-incident rather than to the whole class of fields.
    """
    exempt = {
        # Bambuddy's own public address, not a destination it requests. It is
        # rendered into notification bodies and OIDC redirect URIs, and handed
        # to Obico's ML server as the `img` parameter for that server to fetch
        # (obico_detection.py builds `{external_url}/api/v1/obico/cached-frame/
        # {nonce}`). Pointing it at a private address only breaks Bambuddy's own
        # links; it cannot make Bambuddy request anything it otherwise wouldn't.
        "external_url",
        # Guarded by assert_safe_spoolman_url at each consumer (spoolman.py,
        # location_service.py, inventory.py, spoolbuddy.py,
        # spoolman_inventory.py) rather than in the schema, keeping its
        # established user-facing "Spoolman URL ..." error wording.
        "spoolman_url",
        # Not an HTTP URL: ldap:// or ldaps://, handed to an LDAP client, never
        # to httpx. The LAN-service guard requires http/https and would reject
        # every valid value. It also cannot reach a cloud-metadata endpoint,
        # since IMDS only speaks HTTP.
        "ldap_server_url",
    }
    url_fields = {name for name in AppSettingsUpdate.model_fields if name.endswith("_url")}
    unguarded = url_fields - set(LAN_SERVICE_SETTINGS) - exempt
    assert not unguarded, (
        f"New outbound URL setting(s) {sorted(unguarded)} are not covered by a "
        f"SSRF guard. Add them to AppSettingsUpdate._LAN_SERVICE_URL_FIELDS (or "
        f"the public-internet guard), or add them to `exempt` above with a reason."
    )


# ---------------------------------------------------------------------------
# Notification providers: URL guard + no response-body echo
# ---------------------------------------------------------------------------


def _response(status: int = 500, body: str = "root:x:0:0:root:/root:/bin/bash") -> httpx.Response:
    return httpx.Response(status_code=status, text=body, request=httpx.Request("POST", "http://10.0.0.1/"))


SECRET_BODY = "root:x:0:0:root:/root:/bin/bash"


def test_opaque_failure_does_not_return_the_response_body():
    message = ns._opaque_http_failure(_response(), label="webhook endpoint")

    assert SECRET_BODY not in message
    assert "500" in message, "the status code is still useful and is not sensitive"
    assert "webhook endpoint" in message


def test_opaque_failure_logs_the_body_for_the_operator(caplog):
    """The body stays available to whoever administers the host — via logs,
    not via the API response."""
    with caplog.at_level("DEBUG", logger=ns.__name__):
        ns._opaque_http_failure(_response(), label="ntfy server")

    assert SECRET_BODY in caplog.text


@pytest.mark.parametrize(
    "provider_label",
    ["ntfy server", "Bark server", "webhook endpoint", "Home Assistant endpoint"],
)
def test_user_supplied_host_providers_use_the_opaque_path(provider_label: str):
    """Guards the mapping itself: each user-supplied-host provider must route
    its HTTP failure through _opaque_http_failure rather than formatting the
    body inline."""
    src = inspect.getsource(ns)
    assert f'_opaque_http_failure(response, label="{provider_label}")' in src


def test_no_user_supplied_host_provider_formats_the_body_inline():
    """Any remaining ``response.text[:200]`` must belong to a host-pinned provider.

    Pushover/Telegram/CallMeBot/Discord all target hardcoded hosts (Discord via
    a webhook-prefix allowlist), so there is no trust boundary to cross.
    """
    src = inspect.getsource(ns).split("\n")
    host_pinned = {"_send_callmebot", "_send_pushover", "_send_telegram", "_send_discord"}

    current = None
    offenders = []
    for line in src:
        match = re.match(r"\s+async def (_send_\w+)", line)
        if match:
            current = match.group(1)
        if "response.text[:200]" in line and current not in host_pinned:
            offenders.append(current)

    assert not offenders, (
        f"{offenders} echo the upstream response body but do not target a "
        f"hardcoded host. Route the failure through _opaque_http_failure."
    )


@pytest.mark.parametrize("url", UNIVERSALLY_BLOCKED)
def test_provider_url_guard_rejects_dangerous_targets(url: str):
    assert ns._assert_safe_provider_url(url, label="Webhook URL") is not None


@pytest.mark.parametrize("url", LAN_ALLOWED)
def test_provider_url_guard_permits_self_hosted_servers(url: str):
    assert ns._assert_safe_provider_url(url, label="ntfy server URL") is None


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("provider_type", "config"),
    [
        ("ntfy", {"server": "http://169.254.169.254", "topic": "t"}),
        ("bark", {"server": "http://169.254.169.254", "device_key": "k"}),
        ("webhook", {"webhook_url": "http://169.254.169.254/latest/meta-data/"}),
    ],
)
async def test_test_config_refuses_metadata_targets_without_a_request(provider_type: str, config: dict, monkeypatch):
    """The end-to-end shape of the reported attack: an unsaved config aimed at
    IMDS via the test endpoint. It must be refused before any HTTP call."""
    called = False

    async def _fail_if_called(*_a, **_kw):
        nonlocal called
        called = True
        raise AssertionError("outbound request should not have been attempted")

    service = ns.NotificationService()
    monkeypatch.setattr(service, "_get_client", _fail_if_called)

    success, message = await service.send_test_notification(provider_type, config)

    assert success is False
    assert called is False
    assert "cloud metadata" in message