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

Merge branch 'dev' into fix/external-camera-capture-coalescing

Browse Source
This commit is contained in:
MartinNYHC
2026-07-31 08:32:46 +02:00
committed by GitHub
parent 68f5651f97 3daae22f3d
commit 6ad5bf30f7
40 changed files with 1397 additions and 132 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/api/routes/printers.py
+39 -5
View File
@@ -64,6 +64,7 @@ from backend.app.services.printer_manager import (
)
from backend.app.utils.filament_ids import filament_id_to_setting_id
from backend.app.utils.http import build_content_disposition
from backend.app.utils.printer_models import uses_exhaust_fan_label
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/printers", tags=["printers"])
@@ -798,6 +799,8 @@ async def get_printer_status(
big_fan1_speed=state.big_fan1_speed,
big_fan2_speed=state.big_fan2_speed,
heatbreak_fan_speed=state.heatbreak_fan_speed,
left_aux_fan_speed=state.left_aux_fan_speed,
exhaust_fan_present=state.exhaust_fan_present,
firmware_version=state.firmware_version,
developer_mode=state.developer_mode if state else None,
ams_filament_backup=state.ams_filament_backup if state else None,
@@ -3193,16 +3196,28 @@ async def set_chamber_temperature(
@router.post("/{printer_id}/fan-speed")
async def set_fan_speed(
printer_id: int,
fan: str = Query(..., description="Fan to control: part, aux, or chamber"),
fan: str = Query(..., description="Fan to control: part, aux, aux2 (left aux), or chamber"),
speed: int = Query(..., ge=0, le=100, description="Fan speed percentage"),
_=RequirePermissionIfAuthEnabled(Permission.PRINTERS_CONTROL),
db: AsyncSession = Depends(get_db),
):
"""Set a fan speed by percentage."""
fan_ids = {"part": 1, "aux": 2, "chamber": 3}
"""Set a fan speed by percentage.
Fan index 10 ("aux2") is the optional left auxiliary part cooling fan on
P2S/X2D — driven with "M106 P10" exactly like Bambu's official machine
profile gcode does. It only exists when the printer reports airduct part 10,
so the request is rejected rather than sending M106 P10 into the void on a
machine that has no such fan.
That gate also rejects for the short window between connecting and the
first airduct push, when nothing is known about the fan yet. The card hides
the badge over the same window, so there is no control to click; a direct
API caller gets a 400 and should retry once the status reports the fan.
"""
fan_ids = {"part": 1, "aux": 2, "chamber": 3, "aux2": 10}
fan_id = fan_ids.get(fan)
if fan_id is None:
raise HTTPException(400, "fan must be 'part', 'aux', or 'chamber'")
raise HTTPException(400, "fan must be 'part', 'aux', 'aux2', or 'chamber'")
result = await db.execute(select(Printer).where(Printer.id == printer_id))
printer = result.scalar_one_or_none()
@@ -3213,12 +3228,31 @@ async def set_fan_speed(
if not client:
raise HTTPException(400, "Printer not connected")
# Presence gate for the accessory fan. Without this, aux2 is accepted for
# every model and an A1 would be sent M106 P10 for a fan it does not have.
# The UI already hides the badge; this closes the same hole on the API.
if fan == "aux2" and getattr(client.state, "left_aux_fan_speed", None) is None:
raise HTTPException(
400,
"This printer does not report a left auxiliary fan "
"(no airduct part 10). The fan is an accessory kit on the P2S "
"and factory-fitted on the X2D.",
)
pwm_speed = round(speed * 255 / 100)
success = client.set_fan_speed(fan_id, pwm_speed)
if not success:
raise HTTPException(500, "Failed to set fan speed")
fan_names = {"part": "Part cooling fan", "aux": "Auxiliary fan", "chamber": "Chamber fan"}
# The enclosure fan is called "Exhaust" on P2S/X2D and "Chamber" elsewhere;
# match whatever the printer card badge shows so the toast agrees with the
# control the user just clicked.
fan_names = {
"part": "Part cooling fan",
"aux": "Auxiliary fan",
"aux2": "Left auxiliary fan",
"chamber": "Exhaust fan" if uses_exhaust_fan_label(printer.model) else "Chamber fan",
}
return {"success": True, "message": f"{fan_names[fan]} set to {speed}%"}
backend/app/schemas/printer.py
+5
View File
@@ -360,6 +360,11 @@ class PrinterStatus(BaseModel):
big_fan1_speed: int | None = None # Auxiliary fan
big_fan2_speed: int | None = None # Chamber/exhaust fan
heatbreak_fan_speed: int | None = None # Hotend heatbreak fan
# Left auxiliary part cooling fan (optional P2S/X2D accessory, airduct part id 10).
# None = not installed / not reported by this model.
left_aux_fan_speed: int | None = None
# Chamber exhaust fan present (P2S/X2D External Exhaust Fan kit; airduct part id 3).
exhaust_fan_present: bool = False
# Firmware version (from info.module[name="ota"].sw_ver)
firmware_version: str | None = None
# Developer LAN mode: True = enabled, False = disabled (MQTT encryption), None = unknown
backend/app/services/bambu_mqtt.py
+101 -2
View File
@@ -472,6 +472,21 @@ class PrinterState:
big_fan1_speed: int | None = None # Auxiliary fan
big_fan2_speed: int | None = None # Chamber/exhaust fan
heatbreak_fan_speed: int | None = None # Hotend heatbreak fan
# Left auxiliary part cooling fan (optional accessory on P2S/X2D). Reported ONLY
# via device.airduct.parts (decoded part id 10 = FAN_REMOTE_COOLING_1 in Bambu
# Studio's AIR_FUN enum) — the firmware does NOT mirror it into any flat
# big_fanX_speed field, which is why it was previously dropped. 0-100 percent.
left_aux_fan_speed: int | None = None
# Chamber exhaust fan, derived from the airduct parts list containing decoded
# id 3. On the P2S this is the External Exhaust Fan kit and a base machine
# omits it, which is the case this flag exists to detect.
#
# NOTE: the flag is not P2S/X2D-specific despite the name. The H2 series
# (H2C/H2D/H2S) also reports part 3, so this goes True there too. That is
# harmless because only the P2S/X2D badge consults it — those models keep
# their unconditional "Chamber Fan" badge — but do not read this as
# "an exhaust kit is fitted" without also checking the model.
exhaust_fan_present: bool = False
# Tray change history during current print: [(global_tray_id, layer_num), ...]
# Used by usage tracker to split filament weight on mid-print tray switch
tray_change_log: list = field(default_factory=list)
@@ -3522,6 +3537,83 @@ class BambuMQTTClient:
f"[{self.serial_number}] airduct_mode changed: {self.state.airduct_mode} -> {new_mode}"
)
self.state.airduct_mode = new_mode
# Parse individual airduct fan parts (new-protocol models: P2S/X2D/H2*).
# Raw part ids are bit-packed — decoded id = raw_id >> 4 (bits 4-11),
# mirroring Bambu Studio DevFan::ParseV3_0. Decoded ids follow the
# AIR_FUN enum: 1=part cooling, 2=right aux, 3=chamber/exhaust,
# 10=left aux (FAN_REMOTE_COOLING_1). The airduct `parts` list only
# contains the fans that physically exist, so it doubles as a
# presence signal for the two P2S/X2D add-on kits:
# - id 10 (left auxiliary part cooling fan) — reported ONLY here,
# never mirrored into a flat big_fanX_speed field.
# - id 3 (chamber exhaust fan) — its speed is mirrored into
# big_fan2_speed, but the part is only listed when the External
# Exhaust Fan kit (get_version module "eef") is installed.
# `state` is already a 0-100 percentage.
parts = airduct_data.get("parts")
if isinstance(parts, list):
speeds: dict[int, int] = {}
for part in parts:
if not isinstance(part, dict):
continue
try:
# Studio reads the id with get_flag_bits(id, 4, 8),
# so mask after shifting for the same reason `state`
# is masked below. Every id seen in the wild
# (16/32/48/160) decodes identically either way —
# this is consistency, not a live bug.
part_id = (int(part["id"]) >> 4) & 0xFF
# `state` is bit-packed like its sibling `range`
# (end << 16 | start), so take only the low 8 bits —
# the same decode Bambu Studio does with
# get_flag_bits(state, 0, 8). Without the mask a
# packed value would clamp to 100 instead of
# decoding to the real percentage.
part_state = int(part["state"]) & 0xFF
except (KeyError, ValueError, TypeError):
continue
# Ids seen across the support-package archive:
# 1 part cooling, 2 aux, 3 chamber/exhaust,
# 6 (H2 series, unmapped), 10 left aux.
speeds[part_id] = max(0, min(100, part_state))
# Absence in this list is what tells us a kit is NOT fitted,
# so it may only be trusted when the list is a full
# inventory rather than a diff frame. `device.airduct` is
# pushed field by field — the `modeCur` handler above exists
# for exactly that reason — and a truncated `parts` read as
# gospel would retract both accessory badges mid-print and
# start rejecting `aux2` on a printer that has the fan.
#
# Every airduct layout in the support-package archive
# (P2S base 1,2 / P2S+kit 1,2,3 / X2D 1,2,3,10 /
# H2C,H2D,H2S 1,2,3,6 — 37 of 37 bundles) contains both the
# part cooling fan and the aux fan, neither of which is
# optional on any machine that reports an airduct at all.
# A list carrying both is therefore a complete inventory; a
# list missing either is a partial frame, and we take its
# speeds without touching presence.
is_full_inventory = 1 in speeds and 2 in speeds
left_aux_speed = speeds.get(10)
if left_aux_speed is None and not is_full_inventory:
# Partial frame that didn't mention the left aux fan —
# keep whatever we already knew about it.
left_aux_speed = self.state.left_aux_fan_speed
if left_aux_speed != self.state.left_aux_fan_speed:
logger.debug(
f"[{self.serial_number}] left_aux_fan_speed changed: "
f"{self.state.left_aux_fan_speed} -> {left_aux_speed}"
)
# A FULL parts list without id 10 means the left aux fan is
# not installed — report None so the UI can hide the widget.
self.state.left_aux_fan_speed = left_aux_speed
# id 3 present == chamber exhaust fan installed (base P2S
# omits it). Only ever retracted on a full inventory.
if 3 in speeds:
self.state.exhaust_fan_present = True
elif is_full_inventory:
self.state.exhaust_fan_present = False
# Parse chamber temp - may be encoded as (target*65536+current) when > 500
# Check if we recently set the target locally (within 5 seconds)
local_set_time = self.state.temperatures.get("_chamber_target_set_time", 0)
@@ -5793,13 +5885,16 @@ class BambuMQTTClient:
"""Set fan speed.
Args:
fan: Fan index (1=part cooling, 2=auxiliary, 3=chamber)
fan: Fan index (1=part cooling, 2=auxiliary, 3=chamber, 10=left auxiliary).
Index 10 is the optional left auxiliary part cooling fan on P2S/X2D
(airduct part id 10); Bambu's official machine profiles drive it with
"M106 P10" in start/layer-change gcode.
speed: Speed 0-255 (0=off, 255=full)
Returns:
True if command was sent, False otherwise
"""
if fan not in (1, 2, 3):
if fan not in (1, 2, 3, 10):
logger.warning("[%s] Invalid fan index: %s", self.serial_number, fan)
return False
@@ -5818,6 +5913,10 @@ class BambuMQTTClient:
"""Set chamber fan speed (0-255)."""
return self.set_fan_speed(3, speed)
def set_left_aux_fan(self, speed: int) -> bool:
"""Set left auxiliary part cooling fan speed (0-255). P2S/X2D accessory."""
return self.set_fan_speed(10, speed)
def set_airduct_mode(self, mode: str) -> bool:
"""Set air conditioning mode (cooling or heating).
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/mqtt_relay.py
+2
View File
@@ -282,6 +282,8 @@ class MQTTRelayService:
"big_fan1_speed": state.big_fan1_speed,
"big_fan2_speed": state.big_fan2_speed,
"heatbreak_fan_speed": state.heatbreak_fan_speed,
"left_aux_fan_speed": state.left_aux_fan_speed,
"exhaust_fan_present": state.exhaust_fan_present,
# Bambuddy-side gate, not printer telemetry (#2525). Mirrors what the
# Web UI already receives via printer_state_to_dict, so an external
# automation can tell "finished" from "finished and still waiting for
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/printer_manager.py
+2
View File
@@ -1418,6 +1418,8 @@ def printer_state_to_dict(
"big_fan1_speed": state.big_fan1_speed,
"big_fan2_speed": state.big_fan2_speed,
"heatbreak_fan_speed": state.heatbreak_fan_speed,
"left_aux_fan_speed": state.left_aux_fan_speed,
"exhaust_fan_present": state.exhaust_fan_present,
# Chamber light state
"chamber_light": state.chamber_light,
# Active extruder for dual-nozzle printers (0=right, 1=left)
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/app/utils/printer_models.py
+30
View File
@@ -212,6 +212,36 @@ DUAL_NOZZLE_MODELS = frozenset(
)
# Models where Bambu's own firmware/UI names the enclosure fan (big_fan2 /
# airduct part id 3) "Exhaust" rather than "Chamber". On these the printer's
# touchscreen and Bambu Studio both call it the exhaust fan, and on the P2S it
# is an add-on kit rather than built-in hardware. Other enclosed models
# (X1 / P1S / H2 series) keep the "Chamber" naming.
EXHAUST_FAN_LABEL_MODELS = frozenset(
[
# Display names (uppercase, no spaces)
"P2S",
"X2D",
# Internal codes
"N7", # P2S
"N6", # X2D
]
)
def uses_exhaust_fan_label(model: str | None) -> bool:
"""Return True if this model calls the big_fan2 enclosure fan "Exhaust".
P2S/X2D name that fan "Exhaust" in Bambu's firmware/UI; everything else
enclosed calls it the chamber fan. Used so the UI badge and the API
response message agree on what the user sees.
"""
if not model:
return False
normalized = model.strip().upper().replace(" ", "").replace("-", "")
return normalized in EXHAUST_FAN_LABEL_MODELS
def has_ethernet(model: str | None) -> bool:
"""Return True if the printer model has an ethernet port."""
if not model:
backend/tests/integration/test_printers_api.py
+72 -3
View File
@@ -3855,8 +3855,10 @@ class TestSetChamberTemperatureAPI:
class TestSetFanSpeedAPI:
"""Integration tests for POST /printers/{id}/fan-speed (#1661).
The fan-id mapping (part->1, aux->2, chamber->3) is the critical
correctness gate — wrong mapping would target the wrong physical fan.
The fan-id mapping (part->1, aux->2, chamber->3, aux2->10) is the
critical correctness gate — wrong mapping would target the wrong
physical fan. "aux2" (M106 P10) is the optional left auxiliary part
cooling fan on P2S/X2D.
"""
@pytest.mark.asyncio
@@ -3879,13 +3881,17 @@ class TestSetFanSpeedAPI:
@pytest.mark.integration
@pytest.mark.parametrize(
"fan_name,expected_fan_id",
[("part", 1), ("aux", 2), ("chamber", 3)],
[("part", 1), ("aux", 2), ("chamber", 3), ("aux2", 10)],
)
async def test_fan_id_mapping(self, async_client: AsyncClient, printer_factory, fan_name, expected_fan_id):
"""Verify each fan name maps to the correct hardware fan-id."""
printer = await printer_factory(name="P", model="X1C")
mock_client = MagicMock()
mock_client.set_fan_speed.return_value = True
# aux2 is presence-gated on the printer reporting airduct part 10, so
# give the mock a reported speed. Set explicitly rather than leaning on
# MagicMock's auto-attribute, which would satisfy the gate by accident.
mock_client.state.left_aux_fan_speed = 0
with patch("backend.app.api.routes.printers.printer_manager") as mock_pm:
mock_pm.get_client.return_value = mock_client
response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan={fan_name}&speed=100")
@@ -3893,6 +3899,39 @@ class TestSetFanSpeedAPI:
called_fan_id, called_pwm = mock_client.set_fan_speed.call_args.args
assert called_fan_id == expected_fan_id
@pytest.mark.asyncio
@pytest.mark.integration
async def test_aux2_rejected_when_printer_has_no_left_aux_fan(self, async_client: AsyncClient, printer_factory):
"""A printer that never reports airduct part 10 must not be sent M106 P10.
Without the gate the endpoint accepted aux2 for every model, so a POST
against an A1 would fire a command for hardware that does not exist.
"""
printer = await printer_factory(name="P", model="A1")
mock_client = MagicMock()
mock_client.set_fan_speed.return_value = True
mock_client.state.left_aux_fan_speed = None
with patch("backend.app.api.routes.printers.printer_manager") as mock_pm:
mock_pm.get_client.return_value = mock_client
response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan=aux2&speed=50")
assert response.status_code == 400
assert "left auxiliary fan" in response.json()["detail"]
mock_client.set_fan_speed.assert_not_called()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_other_fans_unaffected_by_the_aux2_gate(self, async_client: AsyncClient, printer_factory):
"""The gate is aux2-only — a base P2S can still drive its built-in fans."""
printer = await printer_factory(name="P", model="P2S")
mock_client = MagicMock()
mock_client.set_fan_speed.return_value = True
mock_client.state.left_aux_fan_speed = None
with patch("backend.app.api.routes.printers.printer_manager") as mock_pm:
mock_pm.get_client.return_value = mock_client
for fan_name in ("part", "aux", "chamber"):
response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan={fan_name}&speed=50")
assert response.status_code == 200, fan_name
@pytest.mark.asyncio
@pytest.mark.integration
@pytest.mark.parametrize(
@@ -3911,6 +3950,36 @@ class TestSetFanSpeedAPI:
_called_fan_id, called_pwm = mock_client.set_fan_speed.call_args.args
assert called_pwm == expected_pwm
@pytest.mark.asyncio
@pytest.mark.integration
@pytest.mark.parametrize(
"model,expected_label",
[
("P2S", "Exhaust fan"),
("X2D", "Exhaust fan"),
("X1C", "Chamber fan"),
("P1S", "Chamber fan"),
("H2D", "Chamber fan"),
],
)
async def test_chamber_fan_message_matches_model_label(
self, async_client: AsyncClient, printer_factory, model, expected_label
):
"""The success toast must use the same name as the printer card badge.
On P2S/X2D the big_fan2 fan is labelled "Exhaust"; everywhere else it
stays "Chamber". A mismatch means the user clicks "Exhaust" and gets
told "Chamber fan set to N%".
"""
printer = await printer_factory(name="P", model=model)
mock_client = MagicMock()
mock_client.set_fan_speed.return_value = True
with patch("backend.app.api.routes.printers.printer_manager") as mock_pm:
mock_pm.get_client.return_value = mock_client
response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan=chamber&speed=50")
assert response.status_code == 200
assert response.json()["message"] == f"{expected_label} set to 50%"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_speed_out_of_range_rejected(self, async_client: AsyncClient, printer_factory):
backend/tests/unit/services/test_p2s_accessory_fans.py
+394
View File
@@ -0,0 +1,394 @@
"""Tests for the P2S/X2D left auxiliary part cooling fan (#2576).
The "Auxiliary Part Cooling Fan - Left" (also fits X2D) is reported ONLY as
device.airduct part with raw id 160 (decoded id = 160 >> 4 = 10,
AIR_FUN.FAN_REMOTE_COOLING_1 in Bambu Studio) — the firmware does NOT mirror
it into any flat big_fanX_speed field, which is why it was previously dropped.
It is controlled with "M106 P10", exactly like Bambu's official P2S machine-
profile gcode does.
The airduct payloads below are verbatim captures from a live P2S
(fw 01.02.00.00) with the accessory installed.
"""
import pytest
@pytest.fixture
def mqtt_client():
from backend.app.services.bambu_mqtt import BambuMQTTClient
return BambuMQTTClient(
ip_address="192.168.1.100",
serial_number="TESTP2S",
access_code="12345678",
)
def _airduct_device(parts):
"""Wrap airduct parts in the device envelope as pushed by a P2S."""
return {
"device": {
"airduct": {
"modeCur": 0,
"modeFunc": 0,
"modeList": [
{"ctrl": [16, 32, 160, 48], "modeId": 0, "off": []},
{"ctrl": [16, 32, 48], "modeId": 1, "off": [160]},
],
"modeVisable": 7,
"parts": parts,
"subFunc": 0,
"subMode": 0,
"subVisable": 7,
"version": 1,
},
"type": 1,
}
}
# Verbatim parts list from a live P2S: part cooling ramping (state 30,
# target 90), right aux at 40%, left aux OFF, chamber at 70%.
P2S_PARTS_LEFT_AUX_OFF = [
{"func": 0, "id": 16, "range": 6553600, "state": 30, "tar_state": 90},
{"func": 6, "id": 32, "range": 6553600, "state": 40, "tar_state": 40},
{"func": 5, "id": 160, "range": 6553600, "state": 0, "tar_state": 0},
{"func": 2, "id": 48, "range": 6553600, "state": 70, "tar_state": 70},
]
# Same printer later in the print: left aux running at 80%.
P2S_PARTS_LEFT_AUX_80 = [
{"func": 0, "id": 16, "range": 6553600, "state": 60, "tar_state": 60},
{"func": 6, "id": 32, "range": 6553600, "state": 100, "tar_state": 100},
{"func": 5, "id": 160, "range": 6553600, "state": 80, "tar_state": 80},
{"func": 2, "id": 48, "range": 6553600, "state": 80, "tar_state": 80},
]
class TestLeftAuxFanParsing:
"""device.airduct part id 10 (raw 160) -> state.left_aux_fan_speed."""
def test_defaults_to_none(self, mqtt_client):
assert mqtt_client.state.left_aux_fan_speed is None
def test_parses_left_aux_running(self, mqtt_client):
mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
assert mqtt_client.state.left_aux_fan_speed == 80
def test_parses_left_aux_off(self, mqtt_client):
mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_OFF))
assert mqtt_client.state.left_aux_fan_speed == 0
def test_raw_id_is_bit_unpacked(self, mqtt_client):
"""Raw id 160 must decode to part id 10 (id >> 4), NOT match on 160."""
# A hypothetical raw id of 10 would decode to part id 0 — must not match.
parts = [{"func": 5, "id": 10, "range": 6553600, "state": 50, "tar_state": 50}]
mqtt_client._update_state(_airduct_device(parts))
assert mqtt_client.state.left_aux_fan_speed is None
def test_parts_without_left_aux_reports_none(self, mqtt_client):
"""A full parts list without id 10 means the fan is not installed."""
mqtt_client.state.left_aux_fan_speed = 80 # previously seen
parts = [p for p in P2S_PARTS_LEFT_AUX_80 if p["id"] != 160]
mqtt_client._update_state(_airduct_device(parts))
assert mqtt_client.state.left_aux_fan_speed is None
def test_diff_push_without_device_preserves_value(self, mqtt_client):
"""P-series diff pushes omit device.airduct — value must survive."""
mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
mqtt_client._update_state({"nozzle_temper": 250.0})
assert mqtt_client.state.left_aux_fan_speed == 80
def test_state_clamped_to_0_100(self, mqtt_client):
parts = [{"func": 5, "id": 160, "range": 6553600, "state": 250, "tar_state": 0}]
mqtt_client._update_state(_airduct_device(parts))
assert mqtt_client.state.left_aux_fan_speed == 100
def test_packed_state_decodes_from_low_8_bits(self, mqtt_client):
"""`state` is bit-packed like its sibling `range` (end << 16 | start).
Bambu Studio decodes it with get_flag_bits(state, 0, 8), so only the low
byte carries the percentage. Without the mask a packed value would clamp
to 100 instead of decoding to the real speed.
"""
packed = (60 << 16) | 45 # sibling field in the high bits, 45% in the low byte
parts = [{"func": 5, "id": 160, "range": 6553600, "state": packed, "tar_state": 0}]
mqtt_client._update_state(_airduct_device(parts))
assert mqtt_client.state.left_aux_fan_speed == 45
def test_unpacked_state_is_unaffected_by_the_mask(self, mqtt_client):
# Plain 0-100 values (what a P2S actually sends) must round-trip exactly.
for speed in (0, 30, 80, 100):
parts = [{"func": 5, "id": 160, "range": 6553600, "state": speed, "tar_state": speed}]
mqtt_client._update_state(_airduct_device(parts))
assert mqtt_client.state.left_aux_fan_speed == speed
def test_malformed_part_entries_ignored(self, mqtt_client):
parts = [
"not-a-dict",
{"func": 5}, # no id/state
{"id": "garbage", "state": 10},
{"func": 5, "id": 160, "range": 6553600, "state": 30, "tar_state": 30},
]
mqtt_client._update_state(_airduct_device(parts))
assert mqtt_client.state.left_aux_fan_speed == 30
def test_flat_fan_fields_unaffected(self, mqtt_client):
"""Regression: flat fields keep coming from the flat MQTT keys."""
payload = {
"cooling_fan_speed": "4",
"big_fan1_speed": "6",
"big_fan2_speed": "10",
"heatbreak_fan_speed": "14",
**_airduct_device(P2S_PARTS_LEFT_AUX_OFF),
}
mqtt_client._update_state(payload)
assert mqtt_client.state.cooling_fan_speed == 27 # 4/15
assert mqtt_client.state.big_fan1_speed == 40 # 6/15
assert mqtt_client.state.big_fan2_speed == 67 # 10/15
assert mqtt_client.state.heatbreak_fan_speed == 93 # 14/15
assert mqtt_client.state.left_aux_fan_speed == 0
class TestExhaustFanPresence:
"""device.airduct part id 3 (raw 48) presence -> state.exhaust_fan_present.
The chamber exhaust fan is a P2S/X2D add-on kit (get_version module "eef").
Its speed rides on the flat big_fan2_speed field, but the airduct only lists
part id 3 when the kit is physically installed — so part-3 presence is the
signal the UI uses to show/hide the Exhaust tile.
"""
def test_defaults_to_false(self, mqtt_client):
assert mqtt_client.state.exhaust_fan_present is False
def test_present_when_part_3_reported(self, mqtt_client):
# Full P2S parts list includes id 48 (>>4 = 3).
mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
assert mqtt_client.state.exhaust_fan_present is True
def test_absent_when_part_3_missing(self, mqtt_client):
mqtt_client.state.exhaust_fan_present = True # previously seen
parts = [p for p in P2S_PARTS_LEFT_AUX_80 if p["id"] != 48]
mqtt_client._update_state(_airduct_device(parts))
assert mqtt_client.state.exhaust_fan_present is False
def test_base_p2s_only_part_cooling_and_aux(self, mqtt_client):
# A base P2S (no exhaust kit, no left aux kit) lists only ids 1 and 2.
parts = [
{"func": 0, "id": 16, "range": 6553600, "state": 0, "tar_state": 0},
{"func": 6, "id": 32, "range": 6553600, "state": 0, "tar_state": 0},
]
mqtt_client._update_state(_airduct_device(parts))
assert mqtt_client.state.exhaust_fan_present is False
assert mqtt_client.state.left_aux_fan_speed is None
def test_diff_push_without_device_preserves_value(self, mqtt_client):
mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
mqtt_client._update_state({"nozzle_temper": 250.0})
assert mqtt_client.state.exhaust_fan_present is True
class TestPartialPartsFrames:
"""A `parts` list that is not a full inventory must not retract presence.
`device.airduct` is pushed field by field — the `modeCur` handler reads it
with an `in` check for exactly that reason — so a frame can carry `parts`
without carrying every fan. Absence is what tells us a kit is not fitted, so
it is only trustworthy on a complete list. Read as gospel, a truncated frame
would make both accessory badges vanish mid-print and start rejecting
``fan=aux2`` on a printer that does have the fan.
Completeness is judged on ids 1 (part cooling) and 2 (aux) being present:
neither is optional on any machine that reports an airduct at all, and both
appear in every layout in the support-package archive (P2S base 1,2 /
P2S+kit 1,2,3 / X2D 1,2,3,10 / H2C,H2D,H2S 1,2,3,6).
"""
def test_partial_frame_does_not_retract_the_left_aux_fan(self, mqtt_client):
mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
assert mqtt_client.state.left_aux_fan_speed == 80
# Only the part cooling fan changed — the frame says nothing about the
# left aux fan, which is not the same as saying it is gone.
mqtt_client._update_state(
_airduct_device([{"func": 0, "id": 16, "range": 6553600, "state": 70, "tar_state": 70}])
)
assert mqtt_client.state.left_aux_fan_speed == 80
def test_partial_frame_does_not_retract_the_exhaust_fan(self, mqtt_client):
mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
assert mqtt_client.state.exhaust_fan_present is True
mqtt_client._update_state(
_airduct_device([{"func": 0, "id": 16, "range": 6553600, "state": 70, "tar_state": 70}])
)
assert mqtt_client.state.exhaust_fan_present is True
def test_a_partial_frame_still_applies_the_speed_it_carries(self, mqtt_client):
"""Not-authoritative-for-absence is not the same as ignored."""
mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
assert mqtt_client.state.left_aux_fan_speed == 80
mqtt_client._update_state(
_airduct_device([{"func": 5, "id": 160, "range": 6553600, "state": 25, "tar_state": 25}])
)
assert mqtt_client.state.left_aux_fan_speed == 25
def test_a_partial_frame_can_still_reveal_a_fan(self, mqtt_client):
"""Presence may always be added — only retraction needs a full list."""
mqtt_client._update_state(
_airduct_device([{"func": 2, "id": 48, "range": 6553600, "state": 70, "tar_state": 70}])
)
assert mqtt_client.state.exhaust_fan_present is True
def test_a_full_frame_still_retracts_both(self, mqtt_client):
"""The kits really can be removed, and a complete list must say so —
this is the behaviour the presence gate exists for."""
mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
assert mqtt_client.state.left_aux_fan_speed == 80
assert mqtt_client.state.exhaust_fan_present is True
# Base P2S layout: part cooling + aux only.
mqtt_client._update_state(
_airduct_device(
[
{"func": 0, "id": 16, "range": 6553600, "state": 0, "tar_state": 0},
{"func": 6, "id": 32, "range": 6553600, "state": 0, "tar_state": 0},
]
)
)
assert mqtt_client.state.left_aux_fan_speed is None
assert mqtt_client.state.exhaust_fan_present is False
def test_an_empty_parts_list_changes_nothing(self, mqtt_client):
mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
mqtt_client._update_state(_airduct_device([]))
assert mqtt_client.state.left_aux_fan_speed == 80
assert mqtt_client.state.exhaust_fan_present is True
class TestLeftAuxFanCommand:
"""set_fan_speed must accept index 10 and emit M106 P10."""
def test_set_fan_speed_10_sends_m106_p10(self, mqtt_client, monkeypatch):
sent = []
monkeypatch.setattr(mqtt_client, "send_gcode", lambda g: sent.append(g) or True)
assert mqtt_client.set_fan_speed(10, 204) is True
assert sent == ["M106 P10 S204"]
def test_set_left_aux_fan_helper(self, mqtt_client, monkeypatch):
sent = []
monkeypatch.setattr(mqtt_client, "send_gcode", lambda g: sent.append(g) or True)
assert mqtt_client.set_left_aux_fan(255) is True
assert sent == ["M106 P10 S255"]
def test_speed_clamped_to_255(self, mqtt_client, monkeypatch):
sent = []
monkeypatch.setattr(mqtt_client, "send_gcode", lambda g: sent.append(g) or True)
mqtt_client.set_left_aux_fan(999)
assert sent == ["M106 P10 S255"]
def test_invalid_fan_index_rejected(self, mqtt_client, monkeypatch):
sent = []
monkeypatch.setattr(mqtt_client, "send_gcode", lambda g: sent.append(g) or True)
assert mqtt_client.set_fan_speed(4, 100) is False
assert mqtt_client.set_fan_speed(11, 100) is False
assert sent == []
def test_existing_fan_indexes_still_accepted(self, mqtt_client, monkeypatch):
sent = []
monkeypatch.setattr(mqtt_client, "send_gcode", lambda g: sent.append(g) or True)
for idx in (1, 2, 3):
assert mqtt_client.set_fan_speed(idx, 128) is True
assert sent == ["M106 P1 S128", "M106 P2 S128", "M106 P3 S128"]
class TestExhaustFanLabelModels:
"""P2S/X2D call the big_fan2 enclosure fan "Exhaust"; others say "Chamber"."""
def test_p2s_and_x2d_use_exhaust_label(self):
from backend.app.utils.printer_models import uses_exhaust_fan_label
for model in ("P2S", "X2D", "p2s", " P2S ", "N7", "N6"):
assert uses_exhaust_fan_label(model) is True, model
def test_other_enclosed_models_keep_chamber_label(self):
from backend.app.utils.printer_models import uses_exhaust_fan_label
for model in ("X1C", "X1", "X1E", "P1S", "H2D", "H2C", "H2S", "A1"):
assert uses_exhaust_fan_label(model) is False, model
def test_unknown_or_missing_model_defaults_to_chamber(self):
from backend.app.utils.printer_models import uses_exhaust_fan_label
assert uses_exhaust_fan_label(None) is False
assert uses_exhaust_fan_label("") is False
assert uses_exhaust_fan_label("SomeFutureModel") is False
class TestExhaustLabelModelListsAgree:
"""The exhaust-label model list is duplicated across the stack.
The backend keeps ``EXHAUST_FAN_LABEL_MODELS`` (display names plus the N7/N6
internal codes, since the API can be handed either) and the frontend keeps
``MODELS_WITH_EXHAUST_LABEL`` in PrintersPage.tsx (display names only —
``printer.model`` is always a display name by the time it reaches the card).
Both are correct as written, but nothing stopped them drifting apart: adding
a model to one and forgetting the other silently produces a card labelled
"Exhaust" whose control toast says "Chamber fan", or vice versa.
"""
def _frontend_models(self) -> set[str]:
import re
from pathlib import Path
import pytest
# Walk up rather than hard-coding a parent depth, so the test survives
# the file being moved and works whatever directory pytest runs from.
relative = Path("frontend") / "src" / "pages" / "PrintersPage.tsx"
source = next(
(candidate for parent in Path(__file__).resolve().parents if (candidate := parent / relative).is_file()),
None,
)
if source is None:
pytest.skip("frontend sources not present in this checkout")
text = source.read_text(encoding="utf-8")
match = re.search(
r"const MODELS_WITH_EXHAUST_LABEL:[^=]*=\s*new Set\(\[(.*?)\]\)",
text,
re.DOTALL,
)
assert match, "MODELS_WITH_EXHAUST_LABEL not found in PrintersPage.tsx"
return set(re.findall(r"['\"]([^'\"]+)['\"]", match.group(1)))
def test_frontend_list_is_the_display_name_subset_of_the_backend_list(self):
from backend.app.utils.printer_models import EXHAUST_FAN_LABEL_MODELS
frontend = self._frontend_models()
assert frontend, "frontend list parsed as empty"
missing = frontend - set(EXHAUST_FAN_LABEL_MODELS)
assert not missing, (
f"models {sorted(missing)} label the fan 'Exhaust' in the UI but the backend "
f"would report 'Chamber fan' — add them to EXHAUST_FAN_LABEL_MODELS"
)
def test_every_backend_display_name_is_handled_by_the_frontend(self):
from backend.app.utils.printer_models import EXHAUST_FAN_LABEL_MODELS
# N7/N6 are internal codes that never reach the card, so exclude them.
internal_codes = {"N7", "N6"}
backend_display = set(EXHAUST_FAN_LABEL_MODELS) - internal_codes
missing = backend_display - self._frontend_models()
assert not missing, (
f"models {sorted(missing)} say 'Exhaust fan' in the API response but the card "
f"would still show 'Chamber Fan' — add them to MODELS_WITH_EXHAUST_LABEL"
)
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)}"
backend/tests/unit/test_plate_clear_mqtt_notification.py
+2
View File
@@ -44,6 +44,8 @@ def _state() -> SimpleNamespace:
big_fan1_speed=0,
big_fan2_speed=0,
heatbreak_fan_speed=0,
left_aux_fan_speed=None,
exhaust_fan_present=False,
)
backend/tests/unit/test_printer_manager_status_broadcast.py
+2
View File
@@ -90,6 +90,8 @@ def _fake_state(**overrides):
"firmware_version": None,
"gcode_file": None,
"heatbreak_fan_speed": None,
"left_aux_fan_speed": None,
"exhaust_fan_present": False,
"layer_num": None,
"remaining_time": None,
"speed_level": None,