feat(system): log-health scanner + Add/Edit-Printer setup pre-flight

Adds a passive log-health check that complements the active Connection
  Diagnostic. Scans Bambuddy's recent app log against a curated allowlist
  catalog of known failure signatures (rejected access code, FTPS :990
  timeout, FTPS TLS failure, flapping MQTT, unreachable camera, SQLite
  "database is locked" contention), dedupes and classifies each finding
  as layer8/environment/bug, and deep-links to the troubleshooting wiki.
  Sample log lines are sanitized before they leave the process. Exposed
  via GET /system/health and surfaced on two surfaces sharing one
  SystemHealthPanel component: a System Health section on the System
  page, and inline in the bug reporter when the form opens.

  The Add-Printer and Edit-Printer dialogs gained a setup-time pre-flight:
  saving runs the connection diagnostic and, on a failed check, warns with
  a "save anyway" escape hatch instead of silently saving a printer that
  will immediately show offline.

  Log read/parse/sanitize primitives extracted from routes/support.py into
  a shared services/log_reader.py (behaviour-preserving); affected support
  tests repointed accordingly.

  Tests: test_log_health.py (11), test_system_api.py (2 new),
  SystemHealthPanel + BugReportBubble + AddPrinterPreflight +
  EditPrinterPreflight (8 frontend). All strings translated across the 9
  locales. Backend ruff clean, full unit suite green, frontend build +
  eslint clean, i18n parity green.
This commit is contained in:
maziggy
2026-05-22 16:31:16 +02:00
parent ab8e07618f
commit e222a0ef0e
31 changed files with 2325 additions and 690 deletions
+11 -195
View File
@@ -33,6 +33,12 @@ from backend.app.models.settings import Settings
from backend.app.models.smart_plug import SmartPlug
from backend.app.models.user import User
from backend.app.services.discovery import is_running_in_docker
from backend.app.services.log_reader import (
LogEntry,
collect_sensitive_strings,
read_log_entries,
sanitize_log_content,
)
from backend.app.services.network_utils import get_network_interfaces
from backend.app.services.printer_manager import printer_manager
@@ -156,15 +162,6 @@ async def toggle_debug_logging(
)
class LogEntry(BaseModel):
"""A single log entry."""
timestamp: str
level: str
logger_name: str
message: str
class LogsResponse(BaseModel):
"""Response containing log entries."""
@@ -173,107 +170,6 @@ class LogsResponse(BaseModel):
filtered_count: int
# Log line regex pattern: "2024-01-15 10:30:45,123 INFO [module.name] Message here"
LOG_LINE_PATTERN = re.compile(r"^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2},\d{3})\s+(\w+)\s+\[([^\]]+)\]\s+(.*)$")
def _parse_log_line(line: str) -> LogEntry | None:
"""Parse a single log line into a LogEntry."""
match = LOG_LINE_PATTERN.match(line.strip())
if match:
return LogEntry(
timestamp=match.group(1),
level=match.group(2),
logger_name=match.group(3),
message=match.group(4),
)
return None
def _read_log_entries(
limit: int = 200,
level_filter: str | None = None,
search: str | None = None,
) -> tuple[list[LogEntry], int]:
"""Read and parse log entries from file with optional filtering."""
log_file = settings.log_dir / "bambuddy.log"
if not log_file.exists():
return [], 0
entries: list[LogEntry] = []
total_lines = 0
try:
with open(log_file, encoding="utf-8", errors="replace") as f:
# Read all lines and process
lines = f.readlines()
total_lines = len(lines)
# Parse lines in reverse order (newest first)
current_entry: LogEntry | None = None
multi_line_buffer: list[str] = []
for line in reversed(lines):
parsed = _parse_log_line(line)
if parsed:
# Found a new log entry start
if current_entry:
# Apply filters and add previous entry (without multi_line_buffer - it belongs to new entry)
should_include = True
# Level filter
if level_filter and current_entry.level.upper() != level_filter.upper():
should_include = False
# Search filter (case-insensitive)
if search and should_include:
search_lower = search.lower()
if not (
search_lower in current_entry.message.lower()
or search_lower in current_entry.logger_name.lower()
):
should_include = False
if should_include:
entries.append(current_entry)
if len(entries) >= limit:
break
# Set new entry and attach any accumulated multi-line content to it
# (in reverse order, continuation lines come before their parent entry)
current_entry = parsed
if multi_line_buffer:
current_entry.message += "\n" + "\n".join(reversed(multi_line_buffer))
multi_line_buffer = []
elif line.strip():
# Continuation of multi-line log entry (will be attached to next parsed entry)
multi_line_buffer.append(line.rstrip())
# Don't forget the last (oldest) entry
# Note: any remaining multi_line_buffer would be orphaned lines before the first entry
if current_entry and len(entries) < limit:
should_include = True
if level_filter and current_entry.level.upper() != level_filter.upper():
should_include = False
if search and should_include:
search_lower = search.lower()
if not (
search_lower in current_entry.message.lower()
or search_lower in current_entry.logger_name.lower()
):
should_include = False
if should_include:
entries.append(current_entry)
except Exception as e:
logger.error("Error reading log file: %s", e)
return [], 0
# Entries are already in newest-first order
return entries, total_lines
@router.get("/logs", response_model=LogsResponse)
async def get_logs(
limit: int = Query(200, ge=1, le=1000, description="Maximum number of entries to return"),
@@ -282,7 +178,7 @@ async def get_logs(
_: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
):
"""Get recent application log entries with optional filtering."""
entries, total_lines = _read_log_entries(limit=limit, level_filter=level, search=search)
entries, total_lines = read_log_entries(limit=limit, level_filter=level, search=search)
return LogsResponse(
entries=entries,
@@ -1206,42 +1102,6 @@ async def _collect_support_info() -> dict:
return info
def _sanitize_log_content(content: str, sensitive_strings: dict[str, str] | None = None) -> str:
"""Remove sensitive data from log content."""
# First, replace known sensitive values (database-aware exact matching)
# This catches printer names, usernames, and other arbitrary user-chosen strings
# that regex patterns cannot detect
if sensitive_strings:
# Sort by length descending to avoid partial matches (e.g. "My Printer 1" before "My Printer")
for value, label in sorted(sensitive_strings.items(), key=lambda x: len(x[0]), reverse=True):
if len(value) < 3:
continue # Skip very short strings to prevent over-redaction
content = re.sub(re.escape(value), label, content)
# Replace credentials in URLs (e.g. http://user:pass@host, rtsps://bblp:code@host)
content = re.sub(r"((?:https?|rtsps?)://)[^/:@\s]+:[^/@\s]+@", r"\1[CREDENTIALS]@", content)
# Replace email addresses
content = re.sub(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "[EMAIL]", content)
# Replace Bambu Lab printer serial numbers (format: 00M/01D/01S/01P/03W + alphanumeric, 12-16 chars total)
content = re.sub(r"\b0[0-3][A-Z0-9][A-Z0-9]{9,13}\b", "[SERIAL]", content, flags=re.IGNORECASE)
# Replace IPv4 addresses (skip firmware versions like 01.09.01.00 which have leading zeros)
content = re.sub(
r"\b(?:(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]\d|\d)\.){3}(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]\d|\d)\b",
"[IP]",
content,
)
# Replace paths with usernames
content = re.sub(r"/home/[^/\s]+/", "/home/[user]/", content)
content = re.sub(r"/Users/[^/\s]+/", "/Users/[user]/", content)
content = re.sub(r"/opt/[^/\s]+/", "/opt/[user]/", content)
return content
def _get_log_content(max_bytes: int = 10 * 1024 * 1024, sensitive_strings: dict[str, str] | None = None) -> bytes:
"""Get log file content, limited to max_bytes from the end."""
log_file = settings.log_dir / "bambuddy.log"
@@ -1260,35 +1120,15 @@ def _get_log_content(max_bytes: int = 10 * 1024 * 1024, sensitive_strings: dict[
content = f.read().decode("utf-8", errors="replace")
# Sanitize sensitive data
content = _sanitize_log_content(content, sensitive_strings)
content = sanitize_log_content(content, sensitive_strings)
return content.encode("utf-8")
async def _get_recent_sanitized_logs(max_lines: int = 200) -> str:
"""Get recent log lines, sanitized for inclusion in bug reports."""
# Collect sensitive strings from DB for redaction
sensitive_strings: dict[str, str] = {}
async with async_session() as db:
result = await db.execute(select(Printer.name, Printer.serial_number, Printer.ip_address, Printer.access_code))
for name, serial, ip_address, access_code in result.all():
if name:
sensitive_strings[name] = "[PRINTER]"
if serial:
sensitive_strings[serial] = "[SERIAL]"
if ip_address:
sensitive_strings[ip_address] = "[IP]"
if access_code:
sensitive_strings[access_code] = "[ACCESS_CODE]"
result = await db.execute(select(User.username))
for (username,) in result.all():
if username:
sensitive_strings[username] = "[USER]"
result = await db.execute(select(Settings.value).where(Settings.key == "bambu_cloud_email"))
cloud_email = result.scalar_one_or_none()
if cloud_email:
sensitive_strings[cloud_email] = "[EMAIL]"
sensitive_strings = await collect_sensitive_strings(db)
log_file = settings.log_dir / "bambuddy.log"
if not log_file.exists():
@@ -1299,7 +1139,7 @@ async def _get_recent_sanitized_logs(max_lines: int = 200) -> str:
content = log_file.read_text(encoding="utf-8", errors="replace")
lines = content.splitlines()
recent = "\n".join(lines[-max_lines:])
return _sanitize_log_content(recent, sensitive_strings)
return sanitize_log_content(recent, sensitive_strings)
except Exception:
logger.debug("Failed to read logs for bug report", exc_info=True)
return ""
@@ -1322,31 +1162,7 @@ async def generate_support_bundle(
)
# Collect known sensitive values for log redaction
sensitive_strings: dict[str, str] = {}
# Printer names, serial numbers, IP addresses, and access codes
result = await db.execute(select(Printer.name, Printer.serial_number, Printer.ip_address, Printer.access_code))
for name, serial, ip_address, access_code in result.all():
if name:
sensitive_strings[name] = "[PRINTER]"
if serial:
sensitive_strings[serial] = "[SERIAL]"
if ip_address:
sensitive_strings[ip_address] = "[IP]"
if access_code:
sensitive_strings[access_code] = "[ACCESS_CODE]"
# Auth usernames
result = await db.execute(select(User.username))
for (username,) in result.all():
if username:
sensitive_strings[username] = "[USER]"
# Bambu Cloud email
result = await db.execute(select(Settings.value).where(Settings.key == "bambu_cloud_email"))
cloud_email = result.scalar_one_or_none()
if cloud_email:
sensitive_strings[cloud_email] = "[EMAIL]"
sensitive_strings = await collect_sensitive_strings(db)
# Collect support info
support_info = await _collect_support_info()
+16
View File
@@ -23,6 +23,8 @@ from backend.app.models.printer import Printer
from backend.app.models.project import Project
from backend.app.models.smart_plug import SmartPlug
from backend.app.models.user import User
from backend.app.services.log_health import ScanResult, scan_logs
from backend.app.services.log_reader import collect_sensitive_strings
from backend.app.services.printer_manager import printer_manager
router = APIRouter(prefix="/system", tags=["system"])
@@ -574,3 +576,17 @@ async def get_storage_usage(
"""Get storage usage breakdown for Bambuddy data directories."""
max_age_seconds = max(0, min(max_age_seconds, 3600))
return await _get_storage_usage_cached(refresh=refresh, max_age_seconds=max_age_seconds)
@router.get("/health", response_model=ScanResult)
async def get_system_health(
db: AsyncSession = Depends(get_db),
_: User | None = RequirePermissionIfAuthEnabled(Permission.SYSTEM_READ),
):
"""Scan the recent application log against the known-issue catalog.
Powers the self-service triage surfaces (System page + bug reporter).
Sample lines are sanitized before they leave the process.
"""
sensitive_strings = await collect_sensitive_strings(db)
return await asyncio.to_thread(scan_logs, sensitive_strings=sensitive_strings)
+256
View File
@@ -0,0 +1,256 @@
"""Log-health scanner.
Matches the recent Bambuddy app log against a curated catalog of known failure
signatures, so users can self-diagnose setup ("layer 8") issues before filing a
bug report.
The catalog is a deliberate *allowlist*: only known-bad, actionable signatures
are matched — a healthy install produces an empty finding list. Human-readable
cause and fix text is intentionally NOT stored here; the frontend renders it
from i18n keys ``systemHealth.signature.<id>.{name,cause,fix}`` so it stays
translatable across all locales. This module only carries the machine-facing
fields (pattern, severity, category, wiki anchor).
"""
import logging
import re
from dataclasses import dataclass
from pydantic import BaseModel
from backend.app.core.config import settings
from backend.app.services.log_reader import LogEntry, read_log_entries, sanitize_log_content
logger = logging.getLogger(__name__)
# How many recent log entries to scan by default.
DEFAULT_SCAN_LIMIT = 4000
# Log levels ranked so a signature can require "at least WARNING" etc.
_LEVEL_RANK = {"DEBUG": 10, "INFO": 20, "WARNING": 30, "ERROR": 40, "CRITICAL": 50}
# Findings are ordered layer8 first (the user can act on these), then
# environment, then bug (please report). Within a group: errors before warnings.
_CATEGORY_ORDER = {"layer8": 0, "environment": 1, "bug": 2}
_SEVERITY_ORDER = {"error": 0, "warning": 1}
# Cap the sample line length so a finding can never carry a huge folded traceback.
_SAMPLE_MAX_LEN = 400
@dataclass(frozen=True)
class LogSignature:
"""One curated known-issue signature.
``patterns`` are matched (``re.search``, case-insensitive) against the log
entry message. A signature only becomes a reported finding once it has
matched ``min_count`` times within the scan window — this gates noisy,
individually-benign symptoms (e.g. an occasional MQTT reconnect after a
Wi-Fi blip) from being surfaced as a problem.
"""
id: str
patterns: tuple[re.Pattern[str], ...]
severity: str # "error" | "warning"
category: str # "layer8" | "environment" | "bug"
wiki_anchor: str # slug appended to the troubleshooting wiki page URL
min_level: str = "WARNING"
logger_prefix: str | None = None # only match entries from this logger tree
min_count: int = 1
def _compile(*patterns: str) -> tuple[re.Pattern[str], ...]:
return tuple(re.compile(p, re.IGNORECASE) for p in patterns)
# --- The catalog -----------------------------------------------------------
# Seeded from the ranked "layer 8" root causes found in the closed-issue triage
# review. Each id MUST have matching i18n keys: systemHealth.signature.<id>.*
SIGNATURES: tuple[LogSignature, ...] = (
LogSignature(
# Wrong/mistyped access code — FTPS login is rejected (530).
id="ftp-auth-rejected",
patterns=_compile(r"FTP connection permission error"),
severity="error",
category="layer8",
wiki_anchor="wrong-access-code",
logger_prefix="backend.app.services.bambu_ftp",
),
LogSignature(
# FTPS :990 unreachable — port blocked by a firewall, or the printer is
# off / on a different subnet.
id="ftp-connection-timeout",
patterns=_compile(r"FTP connection timed out"),
severity="warning",
category="layer8",
wiki_anchor="ftps-port-990-blocked",
logger_prefix="backend.app.services.bambu_ftp",
min_count=3,
),
LogSignature(
# TLS negotiation to the printer's FTPS server failed.
id="ftp-ssl-error",
patterns=_compile(r"FTP SSL error connecting"),
severity="warning",
category="layer8",
wiki_anchor="ftps-tls-failure",
logger_prefix="backend.app.services.bambu_ftp",
min_count=3,
),
LogSignature(
# MQTT connection keeps dropping — typically MQTT :8883 partially
# blocked, LAN mode unstable, or a flaky network path to the printer.
id="mqtt-connection-flapping",
patterns=_compile(r"Forcing MQTT reconnect", r"Hard reset reconnect failed"),
severity="warning",
category="layer8",
wiki_anchor="mqtt-connection-unstable",
logger_prefix="backend.app.services.bambu_mqtt",
min_count=5,
),
LogSignature(
# Camera stream unreachable — RTSPS :322 blocked, or the printer
# camera / LAN liveview is disabled.
id="camera-connection-refused",
patterns=_compile(
r"Chamber image: connection refused",
r"Chamber image: connection timeout",
r"Camera connection test failed",
),
severity="warning",
category="layer8",
wiki_anchor="camera-rtsps-port-322",
logger_prefix="backend.app.services.camera",
min_count=3,
),
LogSignature(
# SQLite write contention. Surfaces inside exception tracebacks; folded
# continuation lines are part of the entry message, so this still
# matches. The fix is switching to PostgreSQL under multi-printer load.
id="database-locked",
patterns=_compile(r"database is locked"),
severity="error",
category="environment",
wiki_anchor="database-is-locked",
),
)
class LogFinding(BaseModel):
"""An aggregated, sanitized match of one signature against the log."""
signature_id: str
severity: str
category: str
wiki_anchor: str
count: int
first_seen: str
last_seen: str
sample: str
class ScanResult(BaseModel):
"""Result of a log-health scan."""
findings: list[LogFinding]
scanned_entries: int
log_available: bool
summary: dict[str, int]
def _level_ok(entry: LogEntry, min_level: str) -> bool:
return _LEVEL_RANK.get(entry.level.upper(), 0) >= _LEVEL_RANK.get(min_level, 30)
def _matches(sig: LogSignature, entry: LogEntry) -> bool:
if not _level_ok(entry, sig.min_level):
return False
if sig.logger_prefix and not entry.logger_name.startswith(sig.logger_prefix):
return False
return any(p.search(entry.message) for p in sig.patterns)
def _sample_line(message: str) -> str:
"""Take the first line of a (possibly multi-line) entry, length-capped."""
first_line = message.splitlines()[0] if message else ""
if len(first_line) > _SAMPLE_MAX_LEN:
return first_line[:_SAMPLE_MAX_LEN] + "…"
return first_line
def scan_logs(
limit: int = DEFAULT_SCAN_LIMIT,
sensitive_strings: dict[str, str] | None = None,
) -> ScanResult:
"""Scan the recent app log against the signature catalog.
``sensitive_strings`` (from :func:`log_reader.collect_sensitive_strings`) is
applied to every sample line so printer names, serials, IPs, and access
codes never leave the process. Even when it is ``None`` the regex-based
redaction passes still run.
"""
log_file = settings.log_dir / "bambuddy.log"
log_available = log_file.exists()
entries, _total = read_log_entries(limit=limit)
# entry_id -> accumulator. entries arrive newest-first.
agg: dict[str, dict] = {}
for entry in entries:
for sig in SIGNATURES:
if not _matches(sig, entry):
continue
acc = agg.get(sig.id)
if acc is None:
# First (== newest) occurrence encountered.
agg[sig.id] = {
"count": 1,
"sample": entry.message,
"last_seen": entry.timestamp,
"first_seen": entry.timestamp,
}
else:
acc["count"] += 1
# Iterating newest-first, so each later hit is older.
acc["first_seen"] = entry.timestamp
findings: list[LogFinding] = []
for sig in SIGNATURES:
acc = agg.get(sig.id)
if acc is None or acc["count"] < sig.min_count:
continue
sample = sanitize_log_content(_sample_line(acc["sample"]), sensitive_strings)
findings.append(
LogFinding(
signature_id=sig.id,
severity=sig.severity,
category=sig.category,
wiki_anchor=sig.wiki_anchor,
count=acc["count"],
first_seen=acc["first_seen"],
last_seen=acc["last_seen"],
sample=sample,
)
)
findings.sort(
key=lambda f: (
_CATEGORY_ORDER.get(f.category, 9),
_SEVERITY_ORDER.get(f.severity, 9),
-f.count,
)
)
summary = {
"total": len(findings),
"layer8": sum(1 for f in findings if f.category == "layer8"),
"environment": sum(1 for f in findings if f.category == "environment"),
"bug": sum(1 for f in findings if f.category == "bug"),
}
return ScanResult(
findings=findings,
scanned_entries=len(entries),
log_available=log_available,
summary=summary,
)
+213
View File
@@ -0,0 +1,213 @@
"""Shared primitives for reading, parsing, and sanitizing the Bambuddy app log.
Extracted from ``routes/support.py`` so service-layer code (e.g. the log-health
scanner in ``log_health.py``) can reuse log reading and redaction without
importing from the API layer. ``support.py`` re-imports these helpers and keeps
its own route handlers.
"""
import logging
import re
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from backend.app.core.config import settings
from backend.app.models.printer import Printer
from backend.app.models.settings import Settings
from backend.app.models.user import User
logger = logging.getLogger(__name__)
# Log line format: "2024-01-15 10:30:45,123 INFO [module.name] [trace_id] Message"
# The trace_id is left as part of the message group — callers that need it can
# parse it out; the log-health scanner does not.
LOG_LINE_PATTERN = re.compile(r"^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2},\d{3})\s+(\w+)\s+\[([^\]]+)\]\s+(.*)$")
class LogEntry(BaseModel):
"""A single parsed log entry."""
timestamp: str
level: str
logger_name: str
message: str
def parse_log_line(line: str) -> LogEntry | None:
"""Parse a single log line into a LogEntry, or None if it is not a line start."""
match = LOG_LINE_PATTERN.match(line.strip())
if match:
return LogEntry(
timestamp=match.group(1),
level=match.group(2),
logger_name=match.group(3),
message=match.group(4),
)
return None
def read_log_entries(
limit: int = 200,
level_filter: str | None = None,
search: str | None = None,
) -> tuple[list[LogEntry], int]:
"""Read and parse log entries from ``bambuddy.log``, newest first.
Continuation lines (tracebacks etc.) are folded into the message of the
entry they belong to. Returns ``(entries, total_lines_in_file)``.
"""
log_file = settings.log_dir / "bambuddy.log"
if not log_file.exists():
return [], 0
entries: list[LogEntry] = []
total_lines = 0
try:
with open(log_file, encoding="utf-8", errors="replace") as f:
lines = f.readlines()
total_lines = len(lines)
# Parse lines in reverse order (newest first)
current_entry: LogEntry | None = None
multi_line_buffer: list[str] = []
for line in reversed(lines):
parsed = parse_log_line(line)
if parsed:
# Found a new log entry start
if current_entry:
# Apply filters and add previous entry (without multi_line_buffer - it belongs to new entry)
should_include = True
# Level filter
if level_filter and current_entry.level.upper() != level_filter.upper():
should_include = False
# Search filter (case-insensitive)
if search and should_include:
search_lower = search.lower()
if not (
search_lower in current_entry.message.lower()
or search_lower in current_entry.logger_name.lower()
):
should_include = False
if should_include:
entries.append(current_entry)
if len(entries) >= limit:
break
# Set new entry and attach any accumulated multi-line content to it
# (in reverse order, continuation lines come before their parent entry)
current_entry = parsed
if multi_line_buffer:
current_entry.message += "\n" + "\n".join(reversed(multi_line_buffer))
multi_line_buffer = []
elif line.strip():
# Continuation of multi-line log entry (will be attached to next parsed entry)
multi_line_buffer.append(line.rstrip())
# Don't forget the last (oldest) entry
# Note: any remaining multi_line_buffer would be orphaned lines before the first entry
if current_entry and len(entries) < limit:
should_include = True
if level_filter and current_entry.level.upper() != level_filter.upper():
should_include = False
if search and should_include:
search_lower = search.lower()
if not (
search_lower in current_entry.message.lower()
or search_lower in current_entry.logger_name.lower()
):
should_include = False
if should_include:
entries.append(current_entry)
except Exception as e:
logger.error("Error reading log file: %s", e)
return [], 0
# Entries are already in newest-first order
return entries, total_lines
def sanitize_log_content(content: str, sensitive_strings: dict[str, str] | None = None) -> str:
"""Remove sensitive data from log content.
``sensitive_strings`` maps known exact values (printer names, serials, etc.)
to replacement labels; pass the result of :func:`collect_sensitive_strings`.
Regex passes additionally redact credentials in URLs, emails, serials, and
IP addresses that were not captured by exact matching.
"""
# First, replace known sensitive values (database-aware exact matching)
# This catches printer names, usernames, and other arbitrary user-chosen strings
# that regex patterns cannot detect
if sensitive_strings:
# Sort by length descending to avoid partial matches (e.g. "My Printer 1" before "My Printer")
for value, label in sorted(sensitive_strings.items(), key=lambda x: len(x[0]), reverse=True):
if len(value) < 3:
continue # Skip very short strings to prevent over-redaction
content = re.sub(re.escape(value), label, content)
# Replace credentials in URLs (e.g. http://user:pass@host, rtsps://bblp:code@host)
content = re.sub(r"((?:https?|rtsps?)://)[^/:@\s]+:[^/@\s]+@", r"\1[CREDENTIALS]@", content)
# Replace email addresses
content = re.sub(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "[EMAIL]", content)
# Replace Bambu Lab printer serial numbers (format: 00M/01D/01S/01P/03W + alphanumeric, 12-16 chars total)
content = re.sub(r"\b0[0-3][A-Z0-9][A-Z0-9]{9,13}\b", "[SERIAL]", content, flags=re.IGNORECASE)
# Replace IPv4 addresses (skip firmware versions like 01.09.01.00 which have leading zeros)
content = re.sub(
r"\b(?:(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]\d|\d)\.){3}(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]\d|\d)\b",
"[IP]",
content,
)
# Replace paths with usernames
content = re.sub(r"/home/[^/\s]+/", "/home/[user]/", content)
content = re.sub(r"/Users/[^/\s]+/", "/Users/[user]/", content)
content = re.sub(r"/opt/[^/\s]+/", "/opt/[user]/", content)
return content
async def collect_sensitive_strings(db: AsyncSession) -> dict[str, str]:
"""Collect known sensitive values from the database for log redaction.
Covers printer names, serial numbers, IP addresses, access codes, auth
usernames, and the Bambu Cloud email. Pass the result to
:func:`sanitize_log_content`.
"""
sensitive_strings: dict[str, str] = {}
# Printer names, serial numbers, IP addresses, and access codes
result = await db.execute(select(Printer.name, Printer.serial_number, Printer.ip_address, Printer.access_code))
for name, serial, ip_address, access_code in result.all():
if name:
sensitive_strings[name] = "[PRINTER]"
if serial:
sensitive_strings[serial] = "[SERIAL]"
if ip_address:
sensitive_strings[ip_address] = "[IP]"
if access_code:
sensitive_strings[access_code] = "[ACCESS_CODE]"
# Auth usernames
result = await db.execute(select(User.username))
for (username,) in result.all():
if username:
sensitive_strings[username] = "[USER]"
# Bambu Cloud email
result = await db.execute(select(Settings.value).where(Settings.key == "bambu_cloud_email"))
cloud_email = result.scalar_one_or_none()
if cloud_email:
sensitive_strings[cloud_email] = "[EMAIL]"
return sensitive_strings
+10 -10
View File
@@ -22,7 +22,7 @@ class TestSupportLogsAPI:
@pytest.mark.integration
async def test_get_logs_empty_file(self, async_client: AsyncClient):
"""Verify get logs returns empty list when log file doesn't exist."""
with patch("backend.app.api.routes.support.settings") as mock_settings:
with patch("backend.app.services.log_reader.settings") as mock_settings:
mock_settings.log_dir = Path("/nonexistent/path")
response = await async_client.get("/api/v1/support/logs")
@@ -46,7 +46,7 @@ class TestSupportLogsAPI:
log_file = Path(tmpdir) / "bambuddy.log"
log_file.write_text(log_content)
with patch("backend.app.api.routes.support.settings") as mock_settings:
with patch("backend.app.services.log_reader.settings") as mock_settings:
mock_settings.log_dir = Path(tmpdir)
response = await async_client.get("/api/v1/support/logs")
@@ -76,7 +76,7 @@ class TestSupportLogsAPI:
log_file = Path(tmpdir) / "bambuddy.log"
log_file.write_text(log_content)
with patch("backend.app.api.routes.support.settings") as mock_settings:
with patch("backend.app.services.log_reader.settings") as mock_settings:
mock_settings.log_dir = Path(tmpdir)
response = await async_client.get("/api/v1/support/logs?level=ERROR")
@@ -100,7 +100,7 @@ class TestSupportLogsAPI:
log_file = Path(tmpdir) / "bambuddy.log"
log_file.write_text(log_content)
with patch("backend.app.api.routes.support.settings") as mock_settings:
with patch("backend.app.services.log_reader.settings") as mock_settings:
mock_settings.log_dir = Path(tmpdir)
response = await async_client.get("/api/v1/support/logs?search=printer")
@@ -124,7 +124,7 @@ class TestSupportLogsAPI:
log_file = Path(tmpdir) / "bambuddy.log"
log_file.write_text(log_content)
with patch("backend.app.api.routes.support.settings") as mock_settings:
with patch("backend.app.services.log_reader.settings") as mock_settings:
mock_settings.log_dir = Path(tmpdir)
response = await async_client.get("/api/v1/support/logs?limit=2")
@@ -153,7 +153,7 @@ ValueError: test error
log_file = Path(tmpdir) / "bambuddy.log"
log_file.write_text(log_content)
with patch("backend.app.api.routes.support.settings") as mock_settings:
with patch("backend.app.services.log_reader.settings") as mock_settings:
mock_settings.log_dir = Path(tmpdir)
response = await async_client.get("/api/v1/support/logs")
@@ -214,7 +214,7 @@ class TestLogParsingHelpers:
def test_parse_log_line_valid(self):
"""Verify _parse_log_line handles valid log lines."""
from backend.app.api.routes.support import _parse_log_line
from backend.app.services.log_reader import parse_log_line as _parse_log_line
line = "2024-01-15 10:30:45,123 INFO [backend.app.main] Server started"
entry = _parse_log_line(line)
@@ -227,7 +227,7 @@ class TestLogParsingHelpers:
def test_parse_log_line_invalid(self):
"""Verify _parse_log_line returns None for invalid lines."""
from backend.app.api.routes.support import _parse_log_line
from backend.app.services.log_reader import parse_log_line as _parse_log_line
line = "This is not a valid log line"
entry = _parse_log_line(line)
@@ -236,7 +236,7 @@ class TestLogParsingHelpers:
def test_parse_log_line_with_brackets_in_message(self):
"""Verify _parse_log_line handles messages with brackets."""
from backend.app.api.routes.support import _parse_log_line
from backend.app.services.log_reader import parse_log_line as _parse_log_line
line = "2024-01-15 10:30:45,123 INFO [backend.app.main] Processing [item 1] and [item 2]"
entry = _parse_log_line(line)
@@ -246,7 +246,7 @@ class TestLogParsingHelpers:
def test_parse_log_line_all_levels(self):
"""Verify _parse_log_line handles all log levels."""
from backend.app.api.routes.support import _parse_log_line
from backend.app.services.log_reader import parse_log_line as _parse_log_line
levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
for level in levels:
@@ -308,3 +308,48 @@ class TestSystemHelperFunctions:
result = format_uptime(30) # 30 seconds
assert result == "< 1m"
class TestSystemHealthAPI:
"""Integration tests for GET /api/v1/system/health (log-health scan)."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_health_clean_log(self, async_client: AsyncClient, tmp_path, monkeypatch):
"""A log with no known issues returns an empty, healthy result."""
from backend.app.core.config import settings
(tmp_path / "bambuddy.log").write_text(
"2026-05-22 10:00:00,000 INFO [backend.app.main] Application startup complete\n",
encoding="utf-8",
)
monkeypatch.setattr(settings, "log_dir", tmp_path)
response = await async_client.get("/api/v1/system/health")
assert response.status_code == 200
result = response.json()
assert result["log_available"] is True
assert result["findings"] == []
assert result["summary"]["total"] == 0
@pytest.mark.asyncio
@pytest.mark.integration
async def test_health_detects_known_issue(self, async_client: AsyncClient, tmp_path, monkeypatch):
"""A known signature in the log surfaces as a finding."""
from backend.app.core.config import settings
(tmp_path / "bambuddy.log").write_text(
"2026-05-22 10:00:00,000 WARNING [backend.app.services.bambu_ftp] "
"FTP connection permission error to 10.0.0.9: 530\n",
encoding="utf-8",
)
monkeypatch.setattr(settings, "log_dir", tmp_path)
response = await async_client.get("/api/v1/system/health")
assert response.status_code == 200
result = response.json()
ids = [f["signature_id"] for f in result["findings"]]
assert "ftp-auth-rejected" in ids
assert result["summary"]["layer8"] >= 1
@@ -0,0 +1,169 @@
"""Tests for the log-health scanner (backend/app/services/log_health.py)."""
import pytest
from backend.app.core.config import settings
from backend.app.services.log_health import SIGNATURES, scan_logs
def _line(level, logger, msg, ts="2026-05-22 10:00:00,000"):
"""Build one log line in the app's log format: TS LEVEL [logger] message."""
return f"{ts} {level} [{logger}] {msg}"
def _write_log(tmp_path, monkeypatch, lines):
log_file = tmp_path / "bambuddy.log"
log_file.write_text("\n".join(lines) + "\n", encoding="utf-8")
monkeypatch.setattr(settings, "log_dir", tmp_path)
return log_file
FTP_LOGGER = "backend.app.services.bambu_ftp"
MQTT_LOGGER = "backend.app.services.bambu_mqtt"
CAM_LOGGER = "backend.app.services.camera"
def test_clean_log_has_no_findings(tmp_path, monkeypatch):
_write_log(
tmp_path,
monkeypatch,
[
_line("INFO", "backend.app.main", "Application startup complete"),
_line("INFO", FTP_LOGGER, "FTP connected, logging in as bblp"),
],
)
result = scan_logs()
assert result.findings == []
assert result.log_available is True
assert result.scanned_entries == 2
assert result.summary == {"total": 0, "layer8": 0, "environment": 0, "bug": 0}
def test_log_unavailable_when_file_missing(tmp_path, monkeypatch):
# No log file written.
monkeypatch.setattr(settings, "log_dir", tmp_path)
result = scan_logs()
assert result.log_available is False
assert result.findings == []
def test_ftp_auth_rejected_is_detected(tmp_path, monkeypatch):
_write_log(
tmp_path,
monkeypatch,
[_line("WARNING", FTP_LOGGER, "FTP connection permission error to 10.0.0.9: 530 Login incorrect")],
)
result = scan_logs()
assert len(result.findings) == 1
f = result.findings[0]
assert f.signature_id == "ftp-auth-rejected"
assert f.severity == "error"
assert f.category == "layer8"
assert f.count == 1
def test_min_count_gates_low_frequency_signals(tmp_path, monkeypatch):
# ftp-connection-timeout requires min_count=3 — two hits must not surface.
_write_log(
tmp_path,
monkeypatch,
[_line("WARNING", FTP_LOGGER, "FTP connection timed out to 10.0.0.9: timed out")] * 2,
)
assert scan_logs().findings == []
_write_log(
tmp_path,
monkeypatch,
[_line("WARNING", FTP_LOGGER, "FTP connection timed out to 10.0.0.9: timed out")] * 3,
)
findings = scan_logs().findings
assert len(findings) == 1
assert findings[0].signature_id == "ftp-connection-timeout"
assert findings[0].count == 3
def test_aggregation_tracks_count_and_seen_range(tmp_path, monkeypatch):
_write_log(
tmp_path,
monkeypatch,
[
_line("WARNING", FTP_LOGGER, "FTP connection permission error to 10.0.0.9", ts="2026-05-22 09:00:00,000"),
_line("WARNING", FTP_LOGGER, "FTP connection permission error to 10.0.0.9", ts="2026-05-22 10:30:00,000"),
],
)
f = scan_logs().findings[0]
assert f.count == 2
assert f.first_seen == "2026-05-22 09:00:00,000"
assert f.last_seen == "2026-05-22 10:30:00,000"
def test_logger_prefix_filters_unrelated_loggers(tmp_path, monkeypatch):
# Same text, but logged by an unrelated logger — must not match the
# bambu_ftp-scoped signature.
_write_log(
tmp_path,
monkeypatch,
[_line("WARNING", "backend.app.services.something_else", "FTP connection permission error to 10.0.0.9")],
)
assert scan_logs().findings == []
def test_min_level_filters_below_threshold(tmp_path, monkeypatch):
# ftp-auth-rejected requires at least WARNING — an INFO line must not match.
_write_log(
tmp_path,
monkeypatch,
[_line("INFO", FTP_LOGGER, "FTP connection permission error to 10.0.0.9")],
)
assert scan_logs().findings == []
def test_sample_is_sanitized(tmp_path, monkeypatch):
_write_log(
tmp_path,
monkeypatch,
[_line("WARNING", FTP_LOGGER, "FTP connection permission error to 192.168.1.50: 530")],
)
f = scan_logs().findings[0]
assert "192.168.1.50" not in f.sample
assert "[IP]" in f.sample
def test_database_locked_matches_inside_traceback(tmp_path, monkeypatch):
# The signature text appears on a continuation line of a multi-line entry;
# read_log_entries folds it into the parent message.
_write_log(
tmp_path,
monkeypatch,
[
_line("ERROR", "backend.app.core.database", "Unhandled DB error"),
"Traceback (most recent call last):",
"sqlite3.OperationalError: database is locked",
],
)
findings = scan_logs().findings
assert len(findings) == 1
assert findings[0].signature_id == "database-locked"
assert findings[0].category == "environment"
def test_findings_sorted_layer8_then_environment(tmp_path, monkeypatch):
_write_log(
tmp_path,
monkeypatch,
[
_line("ERROR", "backend.app.core.database", "x: database is locked"),
_line("WARNING", FTP_LOGGER, "FTP connection timed out to 10.0.0.9"),
_line("WARNING", FTP_LOGGER, "FTP connection timed out to 10.0.0.9"),
_line("WARNING", FTP_LOGGER, "FTP connection timed out to 10.0.0.9"),
_line("WARNING", FTP_LOGGER, "FTP connection permission error to 10.0.0.9"),
],
)
ids = [f.signature_id for f in scan_logs().findings]
# layer8 error, then layer8 warning, then environment.
assert ids == ["ftp-auth-rejected", "ftp-connection-timeout", "database-locked"]
def test_every_signature_id_is_unique():
ids = [s.id for s in SIGNATURES]
assert len(ids) == len(set(ids))
+11 -11
View File
@@ -235,7 +235,7 @@ class TestSanitizeLogContent:
def test_ipv4_addresses_redacted(self):
"""IPv4 addresses in log lines are replaced with [IP]."""
from backend.app.api.routes.support import _sanitize_log_content
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
content = "2024-01-15 Connected to printer at 192.168.1.100 on port 8883"
result = _sanitize_log_content(content)
@@ -245,7 +245,7 @@ class TestSanitizeLogContent:
def test_multiple_ipv4_addresses_redacted(self):
"""Multiple different IPs in the same line are all redacted."""
from backend.app.api.routes.support import _sanitize_log_content
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
content = "Proxy 10.0.0.1 -> 192.168.1.50"
result = _sanitize_log_content(content)
@@ -253,7 +253,7 @@ class TestSanitizeLogContent:
def test_firmware_versions_with_leading_zeros_preserved(self):
"""Firmware versions like 01.09.01.00 have leading zeros and should NOT be redacted."""
from backend.app.api.routes.support import _sanitize_log_content
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
content = "Firmware version: 01.09.01.00"
result = _sanitize_log_content(content)
@@ -261,7 +261,7 @@ class TestSanitizeLogContent:
def test_firmware_version_mixed_with_ip(self):
"""Firmware versions preserved while real IPs are redacted in the same line."""
from backend.app.api.routes.support import _sanitize_log_content
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
content = "Printer at 192.168.1.5 running firmware 01.07.02.00"
result = _sanitize_log_content(content)
@@ -271,7 +271,7 @@ class TestSanitizeLogContent:
def test_printer_ip_from_sensitive_strings(self):
"""Printer IPs in sensitive_strings are replaced before regex pass."""
from backend.app.api.routes.support import _sanitize_log_content
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
content = "Connecting to 192.168.1.100"
result = _sanitize_log_content(content, sensitive_strings={"192.168.1.100": "[IP]"})
@@ -279,7 +279,7 @@ class TestSanitizeLogContent:
def test_edge_case_zero_ip(self):
"""0.0.0.0 is a valid IP and should be redacted."""
from backend.app.api.routes.support import _sanitize_log_content
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
content = "Binding to 0.0.0.0"
result = _sanitize_log_content(content)
@@ -287,7 +287,7 @@ class TestSanitizeLogContent:
def test_edge_case_broadcast_ip(self):
"""255.255.255.255 is a valid IP and should be redacted."""
from backend.app.api.routes.support import _sanitize_log_content
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
content = "Broadcast to 255.255.255.255"
result = _sanitize_log_content(content)
@@ -295,7 +295,7 @@ class TestSanitizeLogContent:
def test_invalid_octet_not_redacted(self):
"""Octets >255 are not valid IPs and should not be redacted."""
from backend.app.api.routes.support import _sanitize_log_content
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
content = "Value 999.999.999.999"
result = _sanitize_log_content(content)
@@ -303,7 +303,7 @@ class TestSanitizeLogContent:
def test_existing_serial_redaction_still_works(self):
"""Serial number redaction still functions alongside IP redaction."""
from backend.app.api.routes.support import _sanitize_log_content
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
content = "Printer 01SABCDEF1234 at 10.0.0.5"
result = _sanitize_log_content(content)
@@ -314,7 +314,7 @@ class TestSanitizeLogContent:
def test_existing_email_redaction_still_works(self):
"""Email redaction still functions alongside IP redaction."""
from backend.app.api.routes.support import _sanitize_log_content
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
content = "User user@example.com from 172.16.0.1"
result = _sanitize_log_content(content)
@@ -323,7 +323,7 @@ class TestSanitizeLogContent:
def test_existing_path_redaction_still_works(self):
"""Path redaction still functions alongside IP redaction."""
from backend.app.api.routes.support import _sanitize_log_content
from backend.app.services.log_reader import sanitize_log_content as _sanitize_log_content
content = "Config at /home/john/config.yaml from 192.168.0.1"
result = _sanitize_log_content(content)