Security hardening (maziggy/bambuddy-security #8)

backend/app/api/routes/_oidc_helpers.py

@@ -13,7 +13,12 @@ from __future__ import annotations
 import ipaddress
 from urllib.parse import urlparse
 
-from backend.app.api.routes._url_safety import CLOUD_METADATA_IPS, NUMERIC_IP_RE, unwrap_ipv4_mapped
+from backend.app.api.routes._url_safety import (
+    CLOUD_METADATA_HOSTNAMES,
+    CLOUD_METADATA_IPS,
+    NUMERIC_IP_RE,
+    unwrap_ipv4_mapped,
+)
 
 
 def assert_safe_public_https_url(url: str) -> None:
@@ -38,10 +43,12 @@ def assert_safe_public_https_url(url: str) -> None:
     - IPv4-mapped IPv6 (``::ffff:127.0.0.1``) — unwrapped before the IP-class
       check so an attacker can't bypass via IPv6 encoding.
 
-    Hostname-based addresses are accepted without DNS resolution — the
-    operator is trusted to configure a sensible IdP host, and resolving here
-    would both add a TOCTOU gap (DNS can change between validation and
-    request) and make the validator issue network requests of its own.
+    Hostname-based addresses are otherwise accepted without DNS resolution —
+    the operator is trusted to configure a sensible IdP host, and resolving
+    here would both add a TOCTOU gap (DNS can change between validation and
+    request) and make the validator issue network requests of its own. The
+    fixed cloud-metadata hostnames are the exception: matching them is a
+    literal string comparison, not a resolution.
     """
     parsed = urlparse(url)
     if parsed.scheme.lower() != "https":
@@ -49,6 +56,14 @@ def assert_safe_public_https_url(url: str) -> None:
 
     hostname = (parsed.hostname or "").lower()
 
+    # "https:///path" parses to an empty hostname; without this it reaches the
+    # ip_address() ValueError branch and is accepted as a symbolic hostname.
+    if not hostname:
+        raise ValueError("icon URL must include a hostname")
+
+    if hostname in CLOUD_METADATA_HOSTNAMES:
+        raise ValueError("icon URL must not point to a cloud metadata endpoint")
+
     if NUMERIC_IP_RE.match(hostname):
         raise ValueError("icon URL must not use numeric-encoded IP addresses")
 

backend/app/api/routes/_url_safety.py

@@ -40,6 +40,18 @@ CLOUD_METADATA_IPS = frozenset(
     }
 )
 
+# The DNS-name form of the same targets. Neither guard resolves hostnames (see
+# the TOCTOU note on each), so an IP blocklist alone cannot catch these — but a
+# literal-string match needs no resolution and costs nothing. These names only
+# resolve inside the respective cloud, so there is no legitimate reason for any
+# Bambuddy integration to point at one.
+CLOUD_METADATA_HOSTNAMES = frozenset(
+    {
+        "metadata.google.internal",  # GCP
+        "metadata.goog",  # GCP short form
+    }
+)
+
 
 # libc and browsers parse numeric-encoded IP forms (decimal ``2130706433``
 # for 127.0.0.1, hex ``0x7f000001``) but Python's ``ipaddress.ip_address``
@@ -90,10 +102,11 @@ def assert_safe_lan_service_url(url: str, *, label: str) -> None:
       indicative of misuse.
     - IPv4-mapped IPv6 encodings of any of the above.
 
-    Symbolic hostnames are accepted without DNS resolution, matching the
-    public-internet guard: resolution here would be both a TOCTOU (DNS can
+    Symbolic hostnames are otherwise accepted without DNS resolution, matching
+    the public-internet guard: resolution here would be both a TOCTOU (DNS can
     change between validation and request) and a request the validator
-    shouldn't be making.
+    shouldn't be making. The one exception is the fixed set of cloud-metadata
+    hostnames, which is a literal-string match and needs no resolution.
     """
     parsed = urlparse(url)
     if parsed.scheme.lower() not in ("http", "https"):
@@ -101,6 +114,15 @@ def assert_safe_lan_service_url(url: str, *, label: str) -> None:
 
     hostname = (parsed.hostname or "").lower()
 
+    # "http:///path" parses to an empty hostname. Never a valid destination,
+    # and without this it falls through the ip_address() ValueError branch
+    # below and is accepted as if it were a symbolic hostname.
+    if not hostname:
+        raise ValueError(f"{label} must include a hostname")
+
+    if hostname in CLOUD_METADATA_HOSTNAMES:
+        raise ValueError(f"{label} must not point to a cloud metadata endpoint")
+
     if NUMERIC_IP_RE.match(hostname):
         raise ValueError(f"{label} must not use numeric-encoded IP addresses; use standard dotted-decimal notation")
 

backend/app/api/routes/github_backup.py

@@ -49,7 +49,21 @@ async def _enforce_private_repo(repo_url: str, token: str, provider: str) -> Non
 
     Used by POST and PATCH /config so a backup configuration can never be
     saved against a public repository.
+
+    The URL is policy-checked first: the Gitea and Forgejo backends derive
+    their API base from this value (``get_api_base``) and then request it with
+    the supplied token, so an unchecked repository_url is an outbound fetch to
+    an operator-supplied host. A self-hosted Gitea on the LAN is the normal
+    case, so the LAN-service tier applies — this only rules out the targets
+    that are wrong under any topology.
     """
+    from backend.app.api.routes._url_safety import assert_safe_lan_service_url
+
+    try:
+        assert_safe_lan_service_url(repo_url, label="Repository URL")
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
     result = await github_backup_service.test_connection(repo_url, token, provider=provider)
     if not result.get("success"):
         message = result.get("message") or "Connection test failed"

backend/app/services/homeassistant.py

@@ -187,17 +187,40 @@ class HomeAssistantService:
 
     @staticmethod
     def _validate_url(url: str) -> str | None:
-        """Validate HA URL scheme and block dangerous destinations."""
+        """Normalise a caller-supplied HA URL, or return None if it is unsafe.
+
+        The stored ``ha_url`` setting is already validated at the schema layer
+        (``LAN_SERVICE_URL_SETTINGS`` in schemas/settings.py), but
+        ``test_connection`` takes its URL straight from the request body, so
+        the same policy has to be applied here.
+
+        Delegates to ``_url_safety.assert_safe_lan_service_url`` rather than
+        the string blocklist this replaces. That blocklist only knew three
+        literal hostnames plus a ``169.254.`` prefix and never parsed the
+        hostname as an IP, so it let through the Alibaba (100.100.100.200)
+        and AWS-IPv6 (fd00:ec2::254) metadata endpoints, numeric-encoded
+        loopback, multicast, and IPv4-mapped IPv6 encodings of the IMDS
+        address it did know about.
+
+        Loopback and RFC-1918 remain permitted — Home Assistant is a
+        LAN-resident service by design, and the shared guard is documented
+        that way.
+        """
+        from backend.app.api.routes._url_safety import assert_safe_lan_service_url
+
         try:
-            parsed = urlparse(url)
+            assert_safe_lan_service_url(url, label="Home Assistant URL")
         except ValueError:
             return None
-        if parsed.scheme not in ("http", "https") or not parsed.hostname:
-            return None
-        blocked = ("169.254.169.254", "metadata.google.internal", "0.0.0.0")  # nosec B104
-        if parsed.hostname.lower() in blocked or (parsed.hostname or "").startswith("169.254."):
+        # Guard passed, so the scheme is http/https and a hostname is present;
+        # re-parse only to drop query/fragment and normalise the authority.
+        parsed = urlparse(url)
+        if not parsed.hostname:
             return None
-        return f"{parsed.scheme}://{parsed.hostname}" + (f":{parsed.port}" if parsed.port else "") + (parsed.path or "")
+        # urlparse strips the brackets off an IPv6 literal, so they have to go
+        # back on or the rebuilt URL is unparseable ("http://fd00::1:8123").
+        host = f"[{parsed.hostname}]" if ":" in parsed.hostname else parsed.hostname
+        return f"{parsed.scheme.lower()}://{host}" + (f":{parsed.port}" if parsed.port else "") + (parsed.path or "")
 
     async def test_connection(self, url: str, token: str) -> dict:
         """Test connection to Home Assistant.

backend/app/services/obico_detection.py

@@ -365,7 +365,23 @@ class ObicoDetectionService:
         }
 
     async def test_connection(self, url: str) -> dict:
-        """Ping the ML API health endpoint. Returns {ok, status_code, body, error}."""
+        """Ping the ML API health endpoint. Returns {ok, status_code, body, error}.
+
+        The stored ``obico_ml_url`` setting is validated at the schema layer,
+        but this route takes its URL from the request body, so the same
+        LAN-service policy has to be applied here or the guard is trivially
+        sidestepped by testing a URL instead of saving it. The response body
+        is returned to the caller (it is the health signal — the endpoint
+        answers "ok"), which is exactly why the destination must be inside
+        policy before the request is made.
+        """
+        from backend.app.api.routes._url_safety import assert_safe_lan_service_url
+
+        try:
+            assert_safe_lan_service_url(url, label="Obico ML URL")
+        except ValueError as exc:
+            return {"ok": False, "status_code": None, "body": None, "error": str(exc)}
+
         target = f"{url.rstrip('/')}/hc/"
         try:
             async with httpx.AsyncClient(timeout=HEALTH_TIMEOUT) as client:

backend/app/services/rest_smart_plug.py

@@ -1,10 +1,8 @@
 """Service for controlling smart plugs via generic REST/HTTP API."""
 
-import ipaddress
 import json
 import logging
 from typing import TYPE_CHECKING, Any
-from urllib.parse import urlparse
 
 import httpx
 
@@ -24,18 +22,39 @@ class RESTSmartPlugService:
         self.timeout = timeout
 
     @staticmethod
-    def _validate_url(url: str) -> bool:
-        """Block cloud metadata and link-local IPs."""
+    def _url_error(url: str) -> str | None:
+        """Return why *url* is rejected by the LAN-service policy, else None.
+
+        Split out from ``_validate_url`` so ``test_connection`` can tell the
+        user which rule the URL broke instead of a single fixed sentence.
+        """
+        from backend.app.api.routes._url_safety import assert_safe_lan_service_url
+
         try:
-            parsed = urlparse(url)
-            hostname = parsed.hostname
-            if not hostname:
-                return False
-            addr = ipaddress.ip_address(hostname)
-            return not addr.is_loopback and not addr.is_link_local
-        except ValueError:
-            # Hostname is not an IP (e.g., "openhab.local") — allow it
-            return True
+            assert_safe_lan_service_url(url, label="REST plug URL")
+        except ValueError as exc:
+            return str(exc)
+        return None
+
+    @staticmethod
+    def _validate_url(url: str) -> bool:
+        """Apply the shared LAN-service SSRF policy to a REST plug URL.
+
+        Delegates to ``_url_safety.assert_safe_lan_service_url`` — the same
+        guard Spoolman, the notification providers and the LAN-service
+        settings use — rather than reimplementing a narrower check. The
+        hand-rolled version this replaces got the policy wrong in both
+        directions: it rejected a literal ``127.0.0.1`` (so an openHAB or
+        Node-RED instance on the same host could only be reached by spelling
+        it ``localhost``), while allowing every target the shared policy
+        rejects unconditionally — Alibaba/AWS-IPv6 metadata endpoints,
+        numeric-encoded IPs, multicast and the unspecified address — because
+        anything that wasn't a bare IP literal fell through to ``True``.
+
+        Loopback and RFC-1918 stay permitted on purpose: a REST-controlled
+        plug bridge running next to Bambuddy is the normal topology.
+        """
+        return RESTSmartPlugService._url_error(url) is None
 
     def _parse_headers(self, headers_json: str | None) -> dict[str, str]:
         """Parse JSON string to dict of headers."""
@@ -273,8 +292,9 @@ class RESTSmartPlugService:
             - success: bool
             - error: error message if failed
         """
-        if not self._validate_url(url):
-            return {"success": False, "error": "Invalid URL (loopback/link-local addresses are blocked)"}
+        url_error = self._url_error(url)
+        if url_error:
+            return {"success": False, "error": url_error}
 
         parsed_headers = self._parse_headers(headers)
 

backend/app/services/tasmota.py

@@ -26,12 +26,32 @@ class TasmotaService:
 
     @staticmethod
     def _validate_ip(ip: str) -> bool:
-        """Block cloud metadata and link-local IPs."""
+        """Block cloud metadata, loopback and link-local destinations.
+
+        Deliberately stricter than the shared LAN-service guard, and kept that
+        way: a Tasmota plug is always a separate device on the LAN, so a bare
+        IP literal is the only sensible value. Anything that is not one —
+        including a symbolic hostname — still fails closed here, which is why
+        this does not simply delegate to ``assert_safe_lan_service_url``.
+
+        What it borrows from the shared guard is the destination set that is
+        dangerous under any topology: cloud-metadata endpoints beyond the AWS
+        IPv4 address (Alibaba's 100.100.100.200, AWS's fd00:ec2::254),
+        multicast and unspecified addresses, and IPv4-mapped IPv6 encodings
+        used to smuggle any of the above past the per-class checks.
+        """
+        from backend.app.api.routes._url_safety import CLOUD_METADATA_IPS, unwrap_ipv4_mapped
+
         try:
             addr = ipaddress.ip_address(ip)
         except ValueError:
             return False  # Not a valid IP
-        return not addr.is_loopback and not addr.is_link_local
+        effective = unwrap_ipv4_mapped(addr)
+        if effective in CLOUD_METADATA_IPS:
+            return False
+        if effective.is_multicast or effective.is_unspecified:
+            return False
+        return not effective.is_loopback and not effective.is_link_local
 
     async def _send_command(
         self,

backend/tests/unit/services/test_rest_smart_plug.py

@@ -49,11 +49,39 @@ class TestURLValidation:
     def test_hostname_url(self, service):
         assert service._validate_url("http://openhab.local:8080/api") is True
 
-    def test_loopback_blocked(self, service):
-        assert service._validate_url("http://127.0.0.1/api") is False
+    def test_loopback_allowed(self, service):
+        """Deliberate change: the LAN-service policy permits loopback, because
+        an openHAB/Node-RED bridge on the same host is the normal topology.
 
-    def test_link_local_blocked(self, service):
-        assert service._validate_url("http://169.254.1.1/api") is False
+        The previous check rejected a literal 127.0.0.1 while accepting the
+        equivalent "localhost", so the same target was configurable one way and
+        not the other. See test_outbound_url_ssrf_guards.py for the policy.
+        """
+        assert service._validate_url("http://127.0.0.1/api") is True
+
+    def test_link_local_allowed(self, service):
+        """Also deliberate: a generic APIPA address is a LAN host like any
+        other. The cloud-metadata address inside that range is blocked by
+        name, not by rejecting the whole /16 — see test_metadata_blocked."""
+        assert service._validate_url("http://169.254.1.1/api") is True
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "http://169.254.169.254/latest/meta-data/",
+            "http://100.100.100.200/",
+            "http://[fd00:ec2::254]/",
+            "http://metadata.google.internal/",
+            "http://[::ffff:169.254.169.254]/",
+            "http://2130706433/",
+            "http://0.0.0.0/",
+        ],
+    )
+    def test_metadata_and_encoded_targets_blocked(self, service, url):
+        """The gap the previous hand-rolled check left: anything that was not a
+        bare IP literal fell through to True, and the literals it did parse were
+        only tested for loopback/link-local."""
+        assert service._validate_url(url) is False
 
     def test_empty_hostname(self, service):
         assert service._validate_url("http:///api") is False
@@ -405,6 +433,10 @@ class TestTestConnection:
 
     @pytest.mark.asyncio
     async def test_connection_invalid_url(self, service):
-        result = await service.test_connection("http://127.0.0.1/api")
+        """127.0.0.1 is now permitted (see TestURLValidation), so the rejection
+        case here is a target that is out of policy under any topology. The
+        error is the guard's own message rather than a fixed sentence, so the
+        user learns which rule the URL broke."""
+        result = await service.test_connection("http://169.254.169.254/latest/meta-data/")
         assert result["success"] is False
-        assert "blocked" in result["error"].lower()
+        assert "cloud metadata" in result["error"].lower()

backend/tests/unit/test_outbound_url_ssrf_guards.py

@@ -40,6 +40,9 @@ 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
+from backend.app.services.homeassistant import HomeAssistantService
+from backend.app.services.rest_smart_plug import RESTSmartPlugService
+from backend.app.services.tasmota import TasmotaService
 
 # Dangerous under any topology — both tiers must reject all of these.
 UNIVERSALLY_BLOCKED = [
@@ -54,6 +57,11 @@ UNIVERSALLY_BLOCKED = [
     "http://[::ffff:169.254.169.254]/",
     "http://0.0.0.0/",
     "http://239.255.255.250/",
+    # The DNS-name form of the same target. Neither tier resolves hostnames,
+    # but these are a fixed literal set, so matching them costs no lookup.
+    "http://metadata.google.internal/",
+    "http://METADATA.GOOGLE.INTERNAL/computeMetadata/v1/",
+    "http://metadata.goog/",
 ]
 
 # The normal self-hosted topology — the LAN tier must permit all of these.
@@ -374,3 +382,269 @@ async def test_test_config_refuses_metadata_targets_without_a_request(provider_t
     assert success is False
     assert called is False
     assert "cloud metadata" in message
+
+
+# ---------------------------------------------------------------------------
+# Smart plugs: the same request-body-URL shape as the notification test endpoint
+# ---------------------------------------------------------------------------
+#
+# POST /smart-plugs/{ha,rest}/test-connection take their URL from the request
+# body and never persist it, so the schema-layer validator on ``ha_url`` does
+# not apply. Both are reachable with only ``SMART_PLUGS_CONTROL``, which the
+# default Operators group carries and which does NOT imply ``SETTINGS_UPDATE``
+# — identical to the notification case above.
+#
+# Both previously used hand-rolled checks that got the policy wrong in both
+# directions: the REST one rejected a literal ``127.0.0.1`` while allowing
+# every non-literal hostname, and the HA one matched three literal strings and
+# never parsed the hostname as an IP at all.
+
+
+@pytest.mark.parametrize("url", UNIVERSALLY_BLOCKED)
+def test_rest_plug_guard_rejects_dangerous_targets(url: str):
+    assert RESTSmartPlugService._validate_url(url) is False
+
+
+@pytest.mark.parametrize("url", LAN_ALLOWED)
+def test_rest_plug_guard_permits_the_normal_self_hosted_topology(url: str):
+    """Includes literal 127.0.0.1, which the previous implementation rejected
+    while accepting the equivalent "localhost" — a plug bridge on the same
+    host could only be configured by spelling it one particular way."""
+    assert RESTSmartPlugService._validate_url(url) is True
+
+
+@pytest.mark.parametrize("url", UNIVERSALLY_BLOCKED)
+def test_ha_guard_rejects_dangerous_targets(url: str):
+    assert HomeAssistantService._validate_url(url) is None
+
+
+@pytest.mark.parametrize("url", LAN_ALLOWED)
+def test_ha_guard_permits_the_normal_self_hosted_topology(url: str):
+    assert HomeAssistantService._validate_url(url) is not None
+
+
+def test_ha_guard_still_normalises_the_url_it_returns():
+    """Delegating the policy must not change what the caller gets back:
+    scheme+host+port+path, with query and fragment dropped."""
+    assert HomeAssistantService._validate_url("http://192.168.1.5:8123/base?x=1#f") == "http://192.168.1.5:8123/base"
+    assert HomeAssistantService._validate_url("http://ha.lan") == "http://ha.lan"
+
+
+def test_ha_guard_keeps_ipv6_literals_bracketed():
+    """urlparse strips the brackets off an IPv6 host; re-emitting it without
+    them yields an unparseable URL that httpx cannot dial."""
+    assert HomeAssistantService._validate_url("http://[fd00::1]:8123/api") == "http://[fd00::1]:8123/api"
+
+
+@pytest.mark.parametrize("ip", ["169.254.169.254", "100.100.100.200", "fd00:ec2::254", "0.0.0.0", "239.255.255.250"])
+def test_tasmota_guard_rejects_metadata_and_misuse_addresses(ip: str):
+    """Tasmota keeps its own stricter rule (bare IP literals only, loopback
+    rejected — a plug is always a separate LAN device), but must not miss the
+    destinations that are dangerous regardless of topology."""
+    assert TasmotaService._validate_ip(ip) is False
+
+
+@pytest.mark.parametrize("ip", ["::ffff:169.254.169.254", "::ffff:100.100.100.200"])
+def test_tasmota_guard_unwraps_ipv4_mapped_ipv6(ip: str):
+    assert TasmotaService._validate_ip(ip) is False
+
+
+@pytest.mark.parametrize("ip", ["192.168.1.50", "10.0.0.7", "172.16.4.9"])
+def test_tasmota_guard_still_permits_a_normal_lan_plug(ip: str):
+    assert TasmotaService._validate_ip(ip) is True
+
+
+@pytest.mark.parametrize("ip", ["127.0.0.1", "tasmota.local", "not-an-ip"])
+def test_tasmota_guard_keeps_failing_closed_on_non_lan_device_values(ip: str):
+    """Deliberately stricter than the shared LAN guard, and unchanged here."""
+    assert TasmotaService._validate_ip(ip) is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "target",
+    ["http://169.254.169.254/", "http://100.100.100.200/", "http://metadata.google.internal/"],
+)
+async def test_rest_test_connection_refuses_metadata_without_a_request(target: str, monkeypatch):
+    """End-to-end shape of the reported attack, mirroring the notification
+    test above: an unsaved URL aimed at IMDS via the test endpoint must be
+    refused before any HTTP call is made."""
+
+    def _fail_if_called(*_a, **_kw):
+        raise AssertionError("outbound request should not have been attempted")
+
+    monkeypatch.setattr(httpx, "AsyncClient", _fail_if_called)
+    result = await RESTSmartPlugService().test_connection(target, "GET", None)
+
+    assert result["success"] is False
+    assert "cloud metadata" in result["error"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "target",
+    ["http://169.254.169.254", "http://100.100.100.200", "http://metadata.google.internal"],
+)
+async def test_ha_test_connection_refuses_metadata_without_a_request(target: str, monkeypatch):
+    def _fail_if_called(*_a, **_kw):
+        raise AssertionError("outbound request should not have been attempted")
+
+    monkeypatch.setattr(httpx, "AsyncClient", _fail_if_called)
+    result = await HomeAssistantService().test_connection(target, "token")
+
+    assert result["success"] is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("target", ["http://169.254.169.254", "http://metadata.google.internal"])
+async def test_obico_test_connection_refuses_metadata_without_a_request(target: str, monkeypatch):
+    """Same shape again: obico_ml_url is guarded when saved via settings, but
+    this route takes the URL from the request body and echoes the response."""
+    from backend.app.services.obico_detection import ObicoDetectionService
+
+    def _fail_if_called(*_a, **_kw):
+        raise AssertionError("outbound request should not have been attempted")
+
+    monkeypatch.setattr(httpx, "AsyncClient", _fail_if_called)
+    result = await ObicoDetectionService().test_connection(target)
+
+    assert result["ok"] is False
+    assert result["body"] is None
+    assert "cloud metadata" in result["error"]
+
+
+# ---------------------------------------------------------------------------
+# Drift backstop, part 2: URLs that arrive in a request body
+# ---------------------------------------------------------------------------
+#
+# `test_every_url_setting_is_either_guarded_or_explicitly_exempt` above only
+# walks `AppSettingsUpdate`. That is why the notification test endpoint, and
+# then the two smart-plug test endpoints, each had to be found by hand: a URL
+# that arrives in a request body and is never persisted is not a settings
+# field, so nothing enumerated it. This walks the live route table instead.
+
+
+def _request_body_url_fields() -> set[tuple[str, str]]:
+    """Every (model, field) pair on a mutating route whose body carries a URL."""
+    from fastapi.routing import APIRoute
+    from pydantic import BaseModel
+
+    from backend.app.main import app
+
+    found: set[tuple[str, str]] = set()
+    for route in app.routes:
+        if not isinstance(route, APIRoute) or not ({"POST", "PUT", "PATCH"} & set(route.methods or ())):
+            continue
+        for param in route.dependant.body_params:
+            # FastAPI moved the resolved annotation from `type_` onto
+            # `field_info.annotation`; read both so this can't silently
+            # enumerate nothing (which would make the assertions vacuous).
+            annotation = getattr(param, "type_", None)
+            if annotation is None:
+                annotation = getattr(getattr(param, "field_info", None), "annotation", None)
+            if not (isinstance(annotation, type) and issubclass(annotation, BaseModel)):
+                continue
+            for field in annotation.model_fields:
+                if field == "url" or field.endswith("_url"):
+                    found.add((annotation.__name__, field))
+    return found
+
+
+# Guarded: the handler (or the service it calls) puts the value through one of
+# the two tiers before any request is issued.
+GUARDED_BODY_URLS = {
+    ("AppSettingsUpdate", "bambu_studio_api_url"),
+    ("AppSettingsUpdate", "ha_url"),
+    ("AppSettingsUpdate", "obico_ml_url"),
+    ("AppSettingsUpdate", "orcaslicer_api_url"),
+    ("AppSettingsUpdate", "spoolman_url"),  # assert_safe_spoolman_url at each consumer
+    ("HATestConnectionRequest", "url"),  # homeassistant._validate_url
+    ("RESTTestConnectionRequest", "url"),  # rest_smart_plug._validate_url
+    ("TestConnectionRequest", "url"),  # obico_detection.test_connection
+    ("OIDCProviderCreate", "issuer_url"),  # public tier, via schemas.auth
+    ("OIDCProviderCreate", "icon_url"),
+    ("OIDCProviderUpdate", "issuer_url"),
+    ("OIDCProviderUpdate", "icon_url"),
+    # Gitea/Forgejo derive their API base from this and request it with the
+    # stored token, so it is a real fetch target — guarded in
+    # github_backup._enforce_private_repo, which both POST and PATCH funnel through.
+    ("GitHubBackupConfigCreate", "repository_url"),
+    ("GitHubBackupConfigUpdate", "repository_url"),
+    # SmartPlug{Create,Update} persist these; every read goes back out through
+    # RESTSmartPlugService._send_request, which applies the same guard.
+    ("SmartPlugCreate", "rest_on_url"),
+    ("SmartPlugCreate", "rest_off_url"),
+    ("SmartPlugCreate", "rest_status_url"),
+    ("SmartPlugCreate", "rest_power_url"),
+    ("SmartPlugCreate", "rest_energy_url"),
+    ("SmartPlugUpdate", "rest_on_url"),
+    ("SmartPlugUpdate", "rest_off_url"),
+    ("SmartPlugUpdate", "rest_status_url"),
+    ("SmartPlugUpdate", "rest_power_url"),
+    ("SmartPlugUpdate", "rest_energy_url"),
+}
+
+# Not a destination Bambuddy requests — no guard applies.
+NOT_A_FETCH_TARGET = {
+    ("AppSettingsUpdate", "external_url"),  # Bambuddy's own address (see exempt list above)
+    ("AppSettingsUpdate", "ldap_server_url"),  # ldap://, handed to an LDAP client
+    ("ProjectCreate", "url"),  # stored link, rendered in the UI, never fetched
+    ("ProjectUpdate", "url"),
+    ("BOMItemCreate", "sourcing_url"),  # stored supplier link, never fetched
+    ("BOMItemUpdate", "sourcing_url"),
+    ("MakerWorldResolveRequest", "url"),  # parsed for a model id; fetches go to a pinned CDN allowlist
+    ("DeviceRegisterRequest", "backend_url"),  # the device's view of Bambuddy's own address
+    ("HeartbeatRequest", "backend_url"),
+    ("SystemConfigRequest", "backend_url"),
+    ("ExternalLinkCreate", "url"),  # sidebar link, rendered in the UI, never requested
+    ("ExternalLinkUpdate", "url"),
+    ("MaintenanceTypeCreate", "wiki_url"),  # documentation link surfaced in the UI/notifications
+    ("MaintenanceTypeUpdate", "wiki_url"),
+    ("ArchiveUpdate", "external_url"),  # stored source link for the model, never fetched
+}
+
+# Genuinely unguarded, and deliberately recorded rather than quietly exempted.
+# These reach `external_camera.capture_frame`, which dials rtsp:// as well as
+# http(s):// — the LAN-service guard rejects any non-HTTP scheme, so wiring it
+# up as-is would break every RTSP camera. Closing these needs a scheme-aware
+# variant of the guard, not a one-line delegation.
+KNOWN_UNGUARDED_NEEDS_SCHEME_AWARE_GUARD = {
+    ("PrinterCreate", "external_camera_url"),
+    ("PrinterCreate", "external_camera_snapshot_url"),
+    ("PrinterUpdate", "external_camera_url"),
+    ("PrinterUpdate", "external_camera_snapshot_url"),
+}
+
+
+def test_the_route_walk_actually_finds_something():
+    """Guards the guard. If FastAPI's internals move again and the walk starts
+    returning nothing, both assertions below pass vacuously and the backstop
+    silently stops working — which is the exact failure it exists to prevent."""
+    found = _request_body_url_fields()
+    assert ("RESTTestConnectionRequest", "url") in found
+    assert ("HATestConnectionRequest", "url") in found
+    assert len(found) > 20
+
+
+def test_every_request_body_url_is_classified():
+    """A new URL-bearing request field can't land without a decision.
+
+    Add it to GUARDED_BODY_URLS once the handler runs it through a guard, or
+    to NOT_A_FETCH_TARGET with the reason it is never requested. Do not add
+    anything to KNOWN_UNGUARDED_* without also raising it.
+    """
+    classified = GUARDED_BODY_URLS | NOT_A_FETCH_TARGET | KNOWN_UNGUARDED_NEEDS_SCHEME_AWARE_GUARD
+    unclassified = _request_body_url_fields() - classified
+    assert not unclassified, (
+        f"Unclassified request-body URL field(s): {sorted(unclassified)}. Route the value "
+        f"through a guard and list it in GUARDED_BODY_URLS, or list it in NOT_A_FETCH_TARGET "
+        f"with the reason it is never fetched."
+    )
+
+
+def test_classification_lists_do_not_drift_from_the_routes():
+    """The reverse direction: a stale entry means a route was renamed or
+    removed and the list was not updated, which would hide the next one."""
+    actual = _request_body_url_fields()
+    stale = (GUARDED_BODY_URLS | NOT_A_FETCH_TARGET | KNOWN_UNGUARDED_NEEDS_SCHEME_AWARE_GUARD) - actual
+    assert not stale, f"Classification entries no longer match any route: {sorted(stale)}"