public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-09-30 03:01:21 +02:00
Code Issues Packages Projects Releases Wiki Activity

Security hardening (maziggy/bambuddy-security #8)

Browse Source
This commit is contained in:
maziggy
2026-08-02 09:46:09 +02:00
parent 2e45f77a4b 3daae22f3d
commit 51f36a4d24
9 changed files with 475 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/routes/_oidc_helpers.py
+20 -5
View File
@@ -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
+25 -3
View File
@@ -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
+14
View File
@@ -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
+30 -7
View File
@@ -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.
backend/app/services/obico_detection.py
+17 -1
View File
@@ -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
+35 -15
View File
@@ -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
+22 -2
View File
@@ -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
+38 -6
View File
@@ -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
+274
View File
@@ -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)}"