Security hardening (maziggy/bambuddy-security #7)

fix(settings): accept JSON booleans on the Spoolman settings endpoint
This commit is contained in:
maziggy
2026-07-30 13:52:56 +02:00
parent d9da60dd8d
commit 88dc56d6e1
11 changed files with 931 additions and 123 deletions
@@ -1168,16 +1168,24 @@ class TestBarkProvider:
mock_client.post.assert_not_called()
@pytest.mark.asyncio
async def test_send_bark_error_in_200_body(self, service):
"""bark-server can wrap a failure in HTTP 200; the body code must win."""
async def test_send_bark_error_in_200_body(self, service, caplog):
"""bark-server can wrap a failure in HTTP 200; the body code must win.
Only the numeric code is returned — the server is caller-supplied
(bark is self-hostable), so its free-text message is the same read
channel the HTTP-failure path closes. The text goes to the debug log.
"""
mock_client = self._client_returning(200, {"code": 400, "message": "device token invalid"})
with patch.object(service, "_get_client", new_callable=AsyncMock) as mock_get_client:
mock_get_client.return_value = mock_client
success, message = await service._send_bark({"device_key": "bad"}, "Title", "Body")
with caplog.at_level("DEBUG", logger="backend.app.services.notification_service"):
success, message = await service._send_bark({"device_key": "bad"}, "Title", "Body")
assert success is False
assert "device token invalid" in message
assert "Bark error 400" in message
assert "device token invalid" not in message
assert "device token invalid" in caplog.text
@pytest.mark.asyncio
async def test_send_bark_http_error(self, service):
@@ -2476,10 +2484,15 @@ class TestNtfyOutbound:
assert "<!DOCTYPE" not in detail
@pytest.mark.asyncio
async def test_ntfy_normal_403_still_surfaces_body(self, service):
"""A non-Cloudflare 403 (e.g. ntfy auth fail) must keep showing
the original body so the user can debug the real error — we
only intercept the Cloudflare-challenge shape."""
async def test_ntfy_normal_403_is_not_misread_as_a_cloudflare_challenge(self, service, caplog):
"""A non-Cloudflare 403 (e.g. ntfy auth fail) must report the real
status rather than the Cloudflare-challenge advice — we only intercept
the challenge shape.
The origin's body is no longer returned to the API caller: the ntfy
server URL is caller-supplied, so echoing it made this an SSRF read
primitive. It goes to the debug log instead.
"""
import httpx
mock_response = httpx.Response(
@@ -2491,7 +2504,10 @@ class TestNtfyOutbound:
mock_client = AsyncMock()
mock_client.post = AsyncMock(return_value=mock_response)
with patch.object(service, "_get_client", AsyncMock(return_value=mock_client)):
with (
patch.object(service, "_get_client", AsyncMock(return_value=mock_client)),
caplog.at_level("DEBUG", logger="backend.app.services.notification_service"),
):
ok, detail = await service._send_ntfy(
{"server": "https://ntfy.sh", "topic": "alerts", "auth_token": "bad"},
title="t",
@@ -2500,16 +2516,19 @@ class TestNtfyOutbound:
assert ok is False
assert "Cloudflare" not in detail
assert "invalid auth token" in detail
assert detail.startswith("HTTP 403:")
assert detail.startswith("HTTP 403")
assert "invalid auth token" not in detail
assert "invalid auth token" in caplog.text
@pytest.mark.asyncio
async def test_ntfy_origin_error_through_cloudflare_is_not_misclassified(self, service):
async def test_ntfy_origin_error_through_cloudflare_is_not_misclassified(self, service, caplog):
"""Cloudflare adds Server: cloudflare to EVERY proxied response,
including legitimate origin errors. A real 401 "wrong token"
from an ntfy server that happens to sit behind Cloudflare must
still surface the origin's actual error body — we must not flip
still be reported as the origin's status — we must not flip
every CF-fronted 4xx into a "your Cloudflare is blocking" message.
As above, the origin body reaches the debug log rather than the caller.
"""
import httpx
@@ -2527,7 +2546,10 @@ class TestNtfyOutbound:
mock_client = AsyncMock()
mock_client.post = AsyncMock(return_value=mock_response)
with patch.object(service, "_get_client", AsyncMock(return_value=mock_client)):
with (
patch.object(service, "_get_client", AsyncMock(return_value=mock_client)),
caplog.at_level("DEBUG", logger="backend.app.services.notification_service"),
):
ok, detail = await service._send_ntfy(
{"server": "https://ntfy.example", "topic": "alerts", "auth_token": "wrong"},
title="t",
@@ -2536,8 +2558,9 @@ class TestNtfyOutbound:
assert ok is False
assert "Cloudflare" not in detail
assert "unauthorized" in detail
assert detail.startswith("HTTP 401:")
assert detail.startswith("HTTP 401")
assert "unauthorized" not in detail
assert "unauthorized" in caplog.text
@pytest.mark.asyncio
async def test_ntfy_cloudflare_cf_mitigated_header_alone_triggers(self, service):
@@ -0,0 +1,376 @@
"""Outbound-URL SSRF policy: two tiers, applied consistently.
Bambuddy makes outbound HTTP requests to hosts the operator configures. Which
policy applies is a property of the *service*, not the caller:
- LAN-service (Spoolman, ntfy, Bark, webhooks, Home Assistant, Obico ML, the
slicer sidecars) — loopback and RFC-1918 MUST stay reachable, because
self-hosting those next to Bambuddy is the normal topology. Blocking them
would break most installs, which is why a blanket private-IP blocklist is
the wrong fix here.
- Public-internet (OIDC issuer and icon URLs) — a private address cannot be a
real IdP, so it is a probe.
Both tiers reject what is dangerous under any topology: non-HTTP schemes,
numeric-encoded IPs, cloud-metadata endpoints, multicast/unspecified, and
IPv4-mapped IPv6 encodings of the above.
The separate concern covered here is *response-body echo*. Notification
provider URLs are writable by anyone holding ``NOTIFICATIONS_CREATE`` — which
the default Operators group carries and which does NOT imply
``SETTINGS_UPDATE`` — and ``POST /notifications/test-config`` takes the URL
from the request body without persisting it. Returning the upstream body there
made an intended reachability check into an authenticated read primitive
against anything the process can reach. Providers whose host Bambuddy pins
(Pushover, Telegram, CallMeBot, Discord) may still echo, since the caller
cannot influence the destination.
"""
from __future__ import annotations
import inspect
import re
import httpx
import pytest
from backend.app.api.routes._oidc_helpers import assert_safe_public_https_url
from backend.app.api.routes._spoolman_helpers import assert_safe_spoolman_url
from backend.app.api.routes._url_safety import assert_safe_lan_service_url
from backend.app.schemas.auth import OIDCProviderCreate, OIDCProviderUpdate
from backend.app.schemas.settings import LAN_SERVICE_URL_SETTINGS, AppSettingsUpdate
from backend.app.services import notification_service as ns
# Dangerous under any topology — both tiers must reject all of these.
UNIVERSALLY_BLOCKED = [
"file:///etc/passwd",
"gopher://127.0.0.1:6379/_INFO",
"ftp://internal.example.com/",
"http://169.254.169.254/latest/meta-data/",
"http://100.100.100.200/",
"http://[fd00:ec2::254]/",
"http://2130706433/",
"http://0x7f000001/",
"http://[::ffff:169.254.169.254]/",
"http://0.0.0.0/",
"http://239.255.255.250/",
]
# The normal self-hosted topology — the LAN tier must permit all of these.
LAN_ALLOWED = [
"http://127.0.0.1:7912/",
"http://localhost:3003",
"http://192.168.1.50:8123",
"http://10.0.0.7:3333",
"http://172.16.4.9:8080",
"https://ntfy.example.com/",
"http://spoolman.lan:7912",
]
# ---------------------------------------------------------------------------
# The LAN-service tier
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("url", UNIVERSALLY_BLOCKED)
def test_lan_tier_rejects_universally_dangerous_targets(url: str):
with pytest.raises(ValueError):
assert_safe_lan_service_url(url, label="Test URL")
@pytest.mark.parametrize("url", LAN_ALLOWED)
def test_lan_tier_permits_the_normal_self_hosted_topology(url: str):
"""A blanket private-IP block here would break most real installs."""
assert_safe_lan_service_url(url, label="Test URL")
def test_lan_tier_names_the_field_in_its_error():
with pytest.raises(ValueError, match="ntfy server URL"):
assert_safe_lan_service_url("file:///etc/passwd", label="ntfy server URL")
def test_spoolman_wrapper_keeps_its_user_facing_wording():
"""The wording is asserted by pre-existing tests; delegation must not change it."""
with pytest.raises(ValueError, match="^Spoolman URL must use http or https$"):
assert_safe_spoolman_url("file:///etc/passwd")
with pytest.raises(ValueError, match="^Spoolman URL must not point to a cloud metadata endpoint$"):
assert_safe_spoolman_url("http://169.254.169.254/")
# ---------------------------------------------------------------------------
# The public-internet tier
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("url", UNIVERSALLY_BLOCKED)
def test_public_tier_rejects_universally_dangerous_targets(url: str):
with pytest.raises(ValueError):
assert_safe_public_https_url(url)
@pytest.mark.parametrize(
"url",
[
"https://127.0.0.1/",
"https://192.168.1.5/",
"https://10.1.2.3/",
"https://[fe80::1]/",
"https://[::ffff:127.0.0.1]/",
"http://accounts.google.com/", # scheme must be https
],
)
def test_public_tier_additionally_rejects_private_and_plain_http(url: str):
with pytest.raises(ValueError):
assert_safe_public_https_url(url)
# ---------------------------------------------------------------------------
# OIDC issuer_url — the encoding bypasses the hand-rolled validator missed
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"url",
[
"https://2130706433/", # decimal-encoded 127.0.0.1
"https://0x7f000001/", # hex-encoded 127.0.0.1
"https://[::ffff:127.0.0.1]/", # IPv4-mapped loopback
"https://[::ffff:169.254.169.254]/", # IPv4-mapped IMDS
"https://0.0.0.0/",
"https://239.255.255.250/",
"https://169.254.169.254/",
"https://127.0.0.1/",
"https://192.168.1.5/",
"http://idp.example.com/",
],
)
def test_issuer_url_rejects_encoded_and_private_targets(url: str):
with pytest.raises(ValueError):
OIDCProviderCreate(
name="SSO",
issuer_url=url,
client_id="cid",
client_secret="secret",
)
def test_issuer_url_update_is_guarded_too():
"""The update path matters most: it can change the issuer while the stored
client_secret stays, which is the shape that would exfiltrate a real secret."""
with pytest.raises(ValueError):
OIDCProviderUpdate(issuer_url="https://[::ffff:127.0.0.1]/")
def test_issuer_url_error_names_the_field_not_the_icon():
with pytest.raises(ValueError, match="issuer_url"):
OIDCProviderUpdate(issuer_url="https://127.0.0.1/")
def test_a_real_idp_still_validates():
provider = OIDCProviderCreate(
name="SSO",
issuer_url="https://accounts.google.com",
client_id="cid",
client_secret="secret",
)
assert provider.issuer_url == "https://accounts.google.com"
# ---------------------------------------------------------------------------
# Settings URLs
# ---------------------------------------------------------------------------
# Imported from the schema rather than duplicated, so the backstop below cannot
# silently disagree with what is actually validated.
LAN_SERVICE_SETTINGS = LAN_SERVICE_URL_SETTINGS
@pytest.mark.parametrize("field", LAN_SERVICE_SETTINGS)
@pytest.mark.parametrize("url", UNIVERSALLY_BLOCKED)
def test_settings_urls_reject_dangerous_targets(field: str, url: str):
with pytest.raises(ValueError):
AppSettingsUpdate(**{field: url})
@pytest.mark.parametrize("field", LAN_SERVICE_SETTINGS)
@pytest.mark.parametrize("url", LAN_ALLOWED)
def test_settings_urls_permit_lan_hosts(field: str, url: str):
assert AppSettingsUpdate(**{field: url})
@pytest.mark.parametrize("field", LAN_SERVICE_SETTINGS)
@pytest.mark.parametrize("empty", ["", " "])
def test_settings_urls_accept_empty_meaning_not_configured(field: str, empty: str):
"""Empty is the documented "fall back to the env var" value for all four."""
assert AppSettingsUpdate(**{field: empty})
@pytest.mark.parametrize("field", LAN_SERVICE_SETTINGS)
@pytest.mark.parametrize(
"legacy",
[
"192.168.1.10:3333", # urlparse: scheme='', netloc='', hostname=None
"localhost:3003", # urlparse: scheme='localhost' (!), hostname=None
"obico.local:3333", # same trap, with dots
"192.168.1.10",
],
)
def test_settings_urls_do_not_newly_reject_scheme_less_legacy_values(field: str, legacy: str):
"""Compatibility guard, not an endorsement.
The settings inputs are plain text with no scheme enforcement, so values
like these are already in the wild. They are inert — httpx raises
UnsupportedProtocol, so no request is issued — and they were storable
before the validator existed. Rejecting them now would block saves of
unrelated fields bundled in the same request (the Obico panel auto-saves
obico_ml_url alongside every other Obico setting).
"""
assert AppSettingsUpdate(**{field: legacy})
@pytest.mark.parametrize("field", LAN_SERVICE_SETTINGS)
def test_settings_urls_still_reject_a_real_non_http_scheme(field: str):
"""The leniency above is scoped to strings that are not URLs at all."""
with pytest.raises(ValueError):
AppSettingsUpdate(**{field: "file:///etc/passwd"})
def test_every_url_setting_is_either_guarded_or_explicitly_exempt():
"""CI backstop: a new outbound-URL setting can't land unvalidated.
Any new ``*_url`` field on AppSettingsUpdate must be added to the
validator's field tuple or listed as exempt here with a reason. This
catches the failure mode the original report correctly identified — guards
added per-incident rather than to the whole class of fields.
"""
exempt = {
# Bambuddy's own public address, not a destination it requests. It is
# rendered into notification bodies and OIDC redirect URIs, and handed
# to Obico's ML server as the `img` parameter for that server to fetch
# (obico_detection.py builds `{external_url}/api/v1/obico/cached-frame/
# {nonce}`). Pointing it at a private address only breaks Bambuddy's own
# links; it cannot make Bambuddy request anything it otherwise wouldn't.
"external_url",
# Guarded by assert_safe_spoolman_url at each consumer (spoolman.py,
# location_service.py, inventory.py, spoolbuddy.py,
# spoolman_inventory.py) rather than in the schema, keeping its
# established user-facing "Spoolman URL ..." error wording.
"spoolman_url",
# Not an HTTP URL: ldap:// or ldaps://, handed to an LDAP client, never
# to httpx. The LAN-service guard requires http/https and would reject
# every valid value. It also cannot reach a cloud-metadata endpoint,
# since IMDS only speaks HTTP.
"ldap_server_url",
}
url_fields = {name for name in AppSettingsUpdate.model_fields if name.endswith("_url")}
unguarded = url_fields - set(LAN_SERVICE_SETTINGS) - exempt
assert not unguarded, (
f"New outbound URL setting(s) {sorted(unguarded)} are not covered by a "
f"SSRF guard. Add them to AppSettingsUpdate._LAN_SERVICE_URL_FIELDS (or "
f"the public-internet guard), or add them to `exempt` above with a reason."
)
# ---------------------------------------------------------------------------
# Notification providers: URL guard + no response-body echo
# ---------------------------------------------------------------------------
def _response(status: int = 500, body: str = "root:x:0:0:root:/root:/bin/bash") -> httpx.Response:
return httpx.Response(status_code=status, text=body, request=httpx.Request("POST", "http://10.0.0.1/"))
SECRET_BODY = "root:x:0:0:root:/root:/bin/bash"
def test_opaque_failure_does_not_return_the_response_body():
message = ns._opaque_http_failure(_response(), label="webhook endpoint")
assert SECRET_BODY not in message
assert "500" in message, "the status code is still useful and is not sensitive"
assert "webhook endpoint" in message
def test_opaque_failure_logs_the_body_for_the_operator(caplog):
"""The body stays available to whoever administers the host — via logs,
not via the API response."""
with caplog.at_level("DEBUG", logger=ns.__name__):
ns._opaque_http_failure(_response(), label="ntfy server")
assert SECRET_BODY in caplog.text
@pytest.mark.parametrize(
"provider_label",
["ntfy server", "Bark server", "webhook endpoint", "Home Assistant endpoint"],
)
def test_user_supplied_host_providers_use_the_opaque_path(provider_label: str):
"""Guards the mapping itself: each user-supplied-host provider must route
its HTTP failure through _opaque_http_failure rather than formatting the
body inline."""
src = inspect.getsource(ns)
assert f'_opaque_http_failure(response, label="{provider_label}")' in src
def test_no_user_supplied_host_provider_formats_the_body_inline():
"""Any remaining ``response.text[:200]`` must belong to a host-pinned provider.
Pushover/Telegram/CallMeBot/Discord all target hardcoded hosts (Discord via
a webhook-prefix allowlist), so there is no trust boundary to cross.
"""
src = inspect.getsource(ns).split("\n")
host_pinned = {"_send_callmebot", "_send_pushover", "_send_telegram", "_send_discord"}
current = None
offenders = []
for line in src:
match = re.match(r"\s+async def (_send_\w+)", line)
if match:
current = match.group(1)
if "response.text[:200]" in line and current not in host_pinned:
offenders.append(current)
assert not offenders, (
f"{offenders} echo the upstream response body but do not target a "
f"hardcoded host. Route the failure through _opaque_http_failure."
)
@pytest.mark.parametrize("url", UNIVERSALLY_BLOCKED)
def test_provider_url_guard_rejects_dangerous_targets(url: str):
assert ns._assert_safe_provider_url(url, label="Webhook URL") is not None
@pytest.mark.parametrize("url", LAN_ALLOWED)
def test_provider_url_guard_permits_self_hosted_servers(url: str):
assert ns._assert_safe_provider_url(url, label="ntfy server URL") is None
@pytest.mark.asyncio
@pytest.mark.parametrize(
("provider_type", "config"),
[
("ntfy", {"server": "http://169.254.169.254", "topic": "t"}),
("bark", {"server": "http://169.254.169.254", "device_key": "k"}),
("webhook", {"webhook_url": "http://169.254.169.254/latest/meta-data/"}),
],
)
async def test_test_config_refuses_metadata_targets_without_a_request(provider_type: str, config: dict, monkeypatch):
"""The end-to-end shape of the reported attack: an unsaved config aimed at
IMDS via the test endpoint. It must be refused before any HTTP call."""
called = False
async def _fail_if_called(*_a, **_kw):
nonlocal called
called = True
raise AssertionError("outbound request should not have been attempted")
service = ns.NotificationService()
monkeypatch.setattr(service, "_get_client", _fail_if_called)
success, message = await service.send_test_notification(provider_type, config)
assert success is False
assert called is False
assert "cloud metadata" in message
@@ -0,0 +1,163 @@
"""PUT /settings/spoolman must not 500 on a JSON boolean.
The endpoint takes a free-form ``dict`` body, and settings are persisted in a
VARCHAR column that every reader compares as a string. Sending the natural JSON
form — ``{"spoolman_enabled": true}`` — used to fail twice over:
- ``bool.lower()`` raised AttributeError while deciding whether the mode had
changed, surfacing as an opaque 500;
- the raw bool was handed to ``upsert_setting``, which SQLite silently coerces
to 1/0 while asyncpg rejects it — so the stored representation depended on
the deployment's database.
The shipped UI sends strings, so this was reachable only through the REST API
(scripts, Home Assistant ``rest_command``) — which is exactly where a JSON
boolean is the obvious thing to send.
These tests cover the normalisers directly. They are pure functions, so the
matrix stays readable and the endpoint keeps a single code path per field.
"""
from __future__ import annotations
import pytest
from fastapi import HTTPException
from backend.app.api.routes.settings import (
normalize_bool_setting,
normalize_str_setting,
setting_is_true,
)
# ---------------------------------------------------------------------------
# The reported crash
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(("value", "expected"), [(True, "true"), (False, "false")])
def test_json_booleans_are_accepted_and_canonicalised(value: bool, expected: str):
"""The exact input that used to 500."""
assert normalize_bool_setting("spoolman_enabled", value) == expected
@pytest.mark.parametrize(("value", "expected"), [(1, "true"), (0, "false")])
def test_json_numbers_one_and_zero_are_accepted(value: int, expected: str):
assert normalize_bool_setting("spoolman_enabled", value) == expected
# ---------------------------------------------------------------------------
# String spellings — generous on purpose, this is a documented REST surface
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("value", ["true", "TRUE", "True", " true ", "1", "yes", "on", "ON"])
def test_truthy_spellings(value: str):
assert normalize_bool_setting("auto_add_unknown_rfid", value) == "true"
@pytest.mark.parametrize("value", ["false", "FALSE", "False", " false ", "0", "no", "off"])
def test_falsy_spellings(value: str):
assert normalize_bool_setting("auto_add_unknown_rfid", value) == "false"
def test_python_style_capitalised_true_is_normalised_lowercase():
"""The frontend compares with a case-sensitive ``=== 'true'``.
A client sending "True" previously had it stored verbatim, so the UI
rendered the setting as OFF while every backend reader (which all use
``.lower()``) treated it as ON.
"""
assert normalize_bool_setting("spoolman_enabled", "True") == "true"
# ---------------------------------------------------------------------------
# Empty means "use the default" — deliberately NOT normalised to "false"
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("value", ["", " "])
def test_empty_is_preserved_not_turned_into_false(value: str):
"""get_spoolman_settings reads these with ``or "<default>"``.
spoolman_report_partial_usage and auto_add_unknown_rfid default to ON, so
coercing a blank submission to "false" would silently switch them off.
Whitespace-only collapses to "" so it takes the same path rather than
being stored as a truthy-but-meaningless " ".
"""
assert normalize_bool_setting("spoolman_report_partial_usage", value) == ""
# ---------------------------------------------------------------------------
# Values with no sensible reading get a 400 naming the field, not a 500
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("value", ["banana", "maybe", "2", "-1", None, [], {}, 3.5, 7])
def test_uninterpretable_values_raise_400_naming_the_field(value: object):
with pytest.raises(HTTPException) as exc:
normalize_bool_setting("spoolman_enabled", value)
assert exc.value.status_code == 400
assert "spoolman_enabled" in str(exc.value.detail)
# ---------------------------------------------------------------------------
# String settings
# ---------------------------------------------------------------------------
def test_str_setting_passes_strings_through_untouched():
assert normalize_str_setting("spoolman_url", "http://192.168.1.5:7912/") == "http://192.168.1.5:7912/"
def test_str_setting_stringifies_numbers():
"""An unquoted host or port is a plausible client slip, not a hard error."""
assert normalize_str_setting("spoolman_url", 7912) == "7912"
def test_str_setting_maps_null_to_empty():
assert normalize_str_setting("spoolman_url", None) == ""
@pytest.mark.parametrize("value", [{"a": 1}, ["x"]])
def test_str_setting_refuses_containers_rather_than_storing_a_repr(value: object):
with pytest.raises(HTTPException) as exc:
normalize_str_setting("spoolman_url", value)
assert exc.value.status_code == 400
# ---------------------------------------------------------------------------
# setting_is_true — used for the mode-switch comparison
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
("stored", "expected"),
[
("true", True),
("True", True),
("TRUE", True),
(" true ", True),
("false", False),
("", False),
("banana", False),
(None, False), # setting absent from the table
(True, True), # legacy row: SQLite coerced a raw bool into the column
(False, False),
],
)
def test_setting_is_true(stored: object, expected: bool):
assert setting_is_true(stored) is expected
@pytest.mark.parametrize("stored", ["1", "on", "yes"])
def test_setting_is_true_stays_narrower_than_the_write_path(stored: str):
"""Reading must agree with the rest of the codebase, which only accepts "true".
normalize_bool_setting is generous about what clients may *send*; every
reader (spoolman_tracking, filament_deficit, inventory, spoolbuddy, labels,
main) compares ``.lower() == "true"``. Accepting more here would make the
mode-switch check disagree with them about a legacy row.
"""
assert setting_is_true(stored) is False