diff --git a/backend/app/api/routes/_oidc_helpers.py b/backend/app/api/routes/_oidc_helpers.py index 184f621eb..d959ca04e 100644 --- a/backend/app/api/routes/_oidc_helpers.py +++ b/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") diff --git a/backend/app/api/routes/_url_safety.py b/backend/app/api/routes/_url_safety.py index cccf2ff5a..732b2cdf3 100644 --- a/backend/app/api/routes/_url_safety.py +++ b/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") diff --git a/backend/app/api/routes/github_backup.py b/backend/app/api/routes/github_backup.py index 450d7e08b..c58d04107 100644 --- a/backend/app/api/routes/github_backup.py +++ b/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" diff --git a/backend/app/services/homeassistant.py b/backend/app/services/homeassistant.py index 57ccfe5d6..24d3925c3 100644 --- a/backend/app/services/homeassistant.py +++ b/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: + # 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 - 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."): - 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. diff --git a/backend/app/services/obico_detection.py b/backend/app/services/obico_detection.py index 70c8da3b5..476b15aae 100644 --- a/backend/app/services/obico_detection.py +++ b/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: diff --git a/backend/app/services/rest_smart_plug.py b/backend/app/services/rest_smart_plug.py index 480971bbf..ffc4e8e88 100644 --- a/backend/app/services/rest_smart_plug.py +++ b/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) diff --git a/backend/app/services/tasmota.py b/backend/app/services/tasmota.py index 188cdd58b..2e2886488 100644 --- a/backend/app/services/tasmota.py +++ b/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, diff --git a/backend/tests/unit/services/test_rest_smart_plug.py b/backend/tests/unit/services/test_rest_smart_plug.py index c3837153a..83e02165b 100644 --- a/backend/tests/unit/services/test_rest_smart_plug.py +++ b/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() diff --git a/backend/tests/unit/test_outbound_url_ssrf_guards.py b/backend/tests/unit/test_outbound_url_ssrf_guards.py index 4bb7b1e57..2d716234c 100644 --- a/backend/tests/unit/test_outbound_url_ssrf_guards.py +++ b/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)}"