refactor(server): extract the diagnostics routes into routers/diagnostics.py (R3) (#857)

The three /api/diagnostics/* routes (export, preview, hardware) plus their
exclusive payload-cap helpers and the `_diag_*` normalisers. Bodies verbatim
except @app->@router, CONFIG_DIR->appstate.config_dir, _running_version()->
appstate.running_version() (a new seam slot; the impl stays in server.py where
the settings region also calls it), and the builtin-plugins lookup in
_diag_plugins_roots: Path(__file__).parent -> Path(__file__).resolve().parents[2]
(routers -> lib -> app root; plugins/ ships at the app root in every packaging
path).

The pure caps/normalisers (_diag_cap_console/_dict/_contributions,
_diag_coerce_bool, _diag_normalize_include, _DIAG_MAX_*) are re-exported from
server.py so the existing `server._diag_*` / `server._DIAG_*` tests keep
resolving — none of them monkeypatch these, so no test retargets.

server.py: 7,216 -> 6,960.

Verified: pyflakes clean (bar the intentional re-export lines); route table
IDENTICAL (143); full pytest 2399 passed (77 diag/packaging + 122 diagnostic-
matched cases incl the cap/coerce/normalize suites); eslint 0; Codex pending.
Boot smoke: /hardware, /preview, and POST /export (200 application/zip) all
serve from the new router location.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Byron Gamatos
2026-07-11 01:27:52 +02:00
committed by GitHub
co-authored by Claude Opus 4.8
parent 32ebc7671e
commit cce95cbd1e
5 changed files with 316 additions and 275 deletions
+1 -1
View File
@@ -27,7 +27,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added ### Added
- **Perf harness now measures 2D-highway frame time (R3c gate).** `scripts/perf-baseline.mjs` gains a `--song` mode that reports per-frame draw-cost p50/p95/p99 (draw-tagged via `highway.addDrawHook`), the metric that gates the `highway.js` split. Maintainer/CI-only; baseline recorded in `docs/perf-baseline.md`. - **Perf harness now measures 2D-highway frame time (R3c gate).** `scripts/perf-baseline.mjs` gains a `--song` mode that reports per-frame draw-cost p50/p95/p99 (draw-tagged via `highway.addDrawHook`), the metric that gates the `highway.js` split. Maintainer/CI-only; baseline recorded in `docs/perf-baseline.md`.
- **`routers/` — extracting `server.py`'s route layer, cheapest-first (R3).** Each PR moves a cohesive route group into a `fastapi.APIRouter` under `lib/routers/`, mounted with `app.include_router(...)` at its original site (FastAPI matches in registration order; the full route table stays byte-identical). Bodies are verbatim — only the decorator receiver (`@app` → `@router`) and singleton reads (`meta_db` → `appstate.meta_db`, resolved at call time) change. So far: `audio_effects` (5), `artist_aliases` (5), `loops` (3), `playlists` (12 + covers), `ws_highway` (the 902-line highway chart WebSocket), `chart` (split/unsplit/work/fileinfo — unblocked by the DLC-path substrate). The DLC library-path resolution (`_get_dlc_dir`, pure `_resolve_dlc_path`) moved to `lib/dlc_paths.py`, reading paths through the seam; `config_dir`/`dlc_dir`/`dlc_dir_env` now ride the `appstate` seam (env-derived, so the pop-and-reimport fixtures reconfigure it for free), and the shared request-field sanitizer `_clean_str` moved to `lib/reqfields.py`. The next cut is picked by a dependency-closure scan that ranks groups by how many `monkeypatch.setattr(server, …)` targets they'd drag along. - **`routers/` — extracting `server.py`'s route layer, cheapest-first (R3).** Each PR moves a cohesive route group into a `fastapi.APIRouter` under `lib/routers/`, mounted with `app.include_router(...)` at its original site (FastAPI matches in registration order; the full route table stays byte-identical). Bodies are verbatim — only the decorator receiver (`@app` → `@router`) and singleton reads (`meta_db` → `appstate.meta_db`, resolved at call time) change. So far: `audio_effects` (5), `artist_aliases` (5), `loops` (3), `playlists` (12 + covers), `ws_highway` (the 902-line highway chart WebSocket), `chart` (split/unsplit/work/fileinfo — unblocked by the DLC-path substrate), `library_extras`, `wanted`, `shop`, `progression`, `profile`, `stats` (the `/api/stats/{path}` catch-all stays registered last so it can't shadow `/recent` `/best` `/top`), `version` (`/api/version`; VERSION-file lookup adjusted for the router subdir depth), and `diagnostics` (`/api/diagnostics/export|preview|hardware`; the plugins-root lookup adjusted for the router subdir depth, `_running_version` reached through the `appstate` seam, pure payload-cap helpers re-exported for the `server._diag_*` tests). The DLC library-path resolution (`_get_dlc_dir`, pure `_resolve_dlc_path`) moved to `lib/dlc_paths.py`, reading paths through the seam; `config_dir`/`dlc_dir`/`dlc_dir_env` now ride the `appstate` seam (env-derived, so the pop-and-reimport fixtures reconfigure it for free), and the shared request-field sanitizer `_clean_str` moved to `lib/reqfields.py`. The next cut is picked by a dependency-closure scan that ranks groups by how many `monkeypatch.setattr(server, …)` targets they'd drag along.
- **`routers/` — the first extracted route module (R3).** The five audio-effects mapping - **`routers/` — the first extracted route module (R3).** The five audio-effects mapping
endpoints move out of `server.py` into `lib/routers/audio_effects.py` as a endpoints move out of `server.py` into `lib/routers/audio_effects.py` as a
`fastapi.APIRouter`, mounted with `app.include_router(...)` **at the point in the file `fastapi.APIRouter`, mounted with `app.include_router(...)` **at the point in the file
+2 -2
View File
@@ -55,8 +55,8 @@ without a *signed* exemption" is unenforceable.
## Planned, NOT exempt (owned by split plans — listed so nothing falls between states) ## Planned, NOT exempt (owned by split plans — listed so nothing falls between states)
core `static/app.js` (11,852) · `static/highway.js` (4,168, whole file) · `server.py` core `static/app.js` (11,852) · `static/highway.js` (4,168, whole file) · `server.py`
(7,216 — was 14,037; ratcheted by the R3 `MetadataDB` + `AudioEffectsMappingDB` (6,960 — was 14,037; ratcheted by the R3 `MetadataDB` + `AudioEffectsMappingDB`
extractions and thirteen `routers/` modules) · extractions and fourteen `routers/` modules) ·
`lib/metadata_db.py` (4,373 — new in R3; the `MetadataDB` class alone is 4,018 lines `lib/metadata_db.py` (4,373 — new in R3; the `MetadataDB` class alone is 4,018 lines
and is a monolith in its own right, to be split per-table once the router train and is a monolith in its own right, to be split per-table once the router train
lands) · `static/v3/songs.js` (4,134) · `static/capabilities/audio-session.js` lands) · `static/v3/songs.js` (4,134) · `static/capabilities/audio-session.js`
+2
View File
@@ -87,12 +87,14 @@ audio_cache_dir = None
# in server.py (its `setattr(server, "_progression_content")` test is untouched). # in server.py (its `setattr(server, "_progression_content")` test is untouched).
get_progression_content = None get_progression_content = None
builtin_diagnostic_filename = None builtin_diagnostic_filename = None
running_version = None
_SLOTS = frozenset({ _SLOTS = frozenset({
"meta_db", "audio_effect_mappings", "meta_db", "audio_effect_mappings",
"config_dir", "dlc_dir", "dlc_dir_env", "config_dir", "dlc_dir", "dlc_dir_env",
"static_dir", "sloppak_cache_dir", "audio_cache_dir", "static_dir", "sloppak_cache_dir", "audio_cache_dir",
"get_progression_content", "builtin_diagnostic_filename", "get_progression_content", "builtin_diagnostic_filename",
"running_version",
}) })
+295
View File
@@ -0,0 +1,295 @@
"""Diagnostic bundle export + hardware probe (/api/diagnostics/*).
One-click "Export Diagnostics" in Settings produces a redacted zip combining
server logs, system info, hardware (CPU/GPU/RAM), plugin inventory, and the
browser-side console transcript + hardware probe. Bundle format is specified in
docs/diagnostics-bundle-spec.md.
Extracted verbatim from server.py (R3) except:
- the decorators (@app -> @router),
- CONFIG_DIR -> appstate.config_dir and _running_version() ->
appstate.running_version() (both read through the appstate seam),
- the builtin-plugins lookup in _diag_plugins_roots: Path(__file__).parent
(the app root when this lived at the top level) ->
Path(__file__).resolve().parents[2] (routers -> lib -> app root). The
plugins/ dir ships at the app root in every packaging path.
The pure helpers + caps here are re-exported from server.py so the existing
`server._diag_*` / `server._DIAG_*` tests keep resolving (none monkeypatch them).
"""
import json
import logging
import os
from pathlib import Path
from fastapi import APIRouter, Body, Response
import appstate
from dlc_paths import _get_dlc_dir
from diagnostics_bundle import build_bundle as _diag_build, preview_bundle as _diag_preview
from diagnostics_hardware import collect as _diag_hardware
from env_compat import getenv_compat
log = logging.getLogger("feedBack.server")
router = APIRouter()
def _diag_log_file() -> Path | None:
raw = os.environ.get("LOG_FILE", "").strip()
if not raw:
return None
return Path(raw)
def _diag_plugins_roots() -> list[Path]:
"""Return all plugin root directories for orphan scanning.
Includes both the built-in ``plugins/`` directory and
``FEEDBACK_PLUGINS_DIR`` when set, so user-installed plugins and
orphans in the external dir are reflected in the bundle.
"""
roots: list[Path] = []
user_dir = getenv_compat("FEEDBACK_PLUGINS_DIR", "").strip()
if user_dir:
p = Path(user_dir)
if p.is_dir():
roots.append(p)
builtin = Path(__file__).resolve().parents[2] / "plugins" # R3: app root from lib/routers/
if builtin not in roots:
roots.append(builtin)
return roots
def _diag_coerce_bool(v, *, default: bool = True) -> bool:
"""Coerce a request-side value to bool, accepting both JSON booleans and
string representations.
- Falsy strings: ``"false"``, ``"0"``, ``"no"``, ``""`` → ``False``
- ``None`` → *default*
- Everything else (including ``"true"``, ``"1"``) → ``True``
"""
if v is None:
return default
if isinstance(v, bool):
return v
if isinstance(v, str):
return v.strip().lower() not in ("false", "0", "no", "")
return bool(v)
def _diag_normalize_include(include: dict | None) -> dict:
"""Coerce request-side flags to the booleans build_bundle expects.
Missing keys default to True so a bare {} request still produces
the full bundle.
Accepts both JSON booleans (``true``/``false``) and string
representations so callers that serialize flags as strings behave
consistently with the preview endpoint:
- Falsy strings: ``"false"``, ``"0"``, ``"no"``, ``""`` → ``False``
- Everything else (including ``"true"``, ``"1"``, ``"yes"``) → ``True``
"""
keys = ("system", "hardware", "logs", "console", "plugins")
if not isinstance(include, dict):
return {k: True for k in keys}
return {k: _diag_coerce_bool(include.get(k), default=True) for k in keys}
# Server-side caps on client-supplied payload sections. diagnostics.js
# enforces a 500-entry / ~250 KB ring buffer on the browser side; these
# bounds give generous headroom while still preventing a crafted POST from
# forcing the server to allocate arbitrarily large in-memory bundles.
_DIAG_MAX_CONSOLE_ENTRIES = 1000 # hard cap: truncate silently
_DIAG_MAX_CONSOLE_BYTES = 2 * 1024 * 1024 # 2 MB hard cap on total console list
_DIAG_MAX_CLIENT_PAYLOAD_BYTES = 2 * 1024 * 1024 # 2 MB per dict section
_DIAG_MAX_CONTRIBUTIONS_BYTES = 4 * 1024 * 1024 # 4 MB aggregate cap for contributions
def _diag_cap_console(v) -> list | None:
"""Return *v* if it is a list, truncated to _DIAG_MAX_CONSOLE_ENTRIES entries
and _DIAG_MAX_CONSOLE_BYTES total. Entries are accumulated until either cap
is reached; no partial-entry splitting occurs."""
if not isinstance(v, list):
return None
result = v[:_DIAG_MAX_CONSOLE_ENTRIES]
# Also enforce a byte cap — the count cap alone does not bound memory when
# entries contain arbitrarily large strings.
try:
out = []
total = 0
for entry in result:
encoded = json.dumps(entry, separators=(",", ":")).encode("utf-8", errors="replace")
if total + len(encoded) > _DIAG_MAX_CONSOLE_BYTES:
break
out.append(entry)
total += len(encoded)
return out
except (TypeError, ValueError):
return None
def _diag_cap_dict(v) -> dict | None:
"""Return *v* if it is a dict whose JSON serialisation fits within
_DIAG_MAX_CLIENT_PAYLOAD_BYTES, otherwise return None."""
if not isinstance(v, dict):
return None
try:
encoded = json.dumps(v, separators=(",", ":")).encode("utf-8", errors="replace")
except (TypeError, ValueError) as e:
log.warning("diagnostics client payload is not JSON-serialisable, dropping: %s", e)
return None
if len(encoded) > _DIAG_MAX_CLIENT_PAYLOAD_BYTES:
return None
return v
def _diag_cap_contributions(v, known_ids=None) -> dict | None:
"""Apply per-plugin and aggregate size caps on client_contributions.
Unlike _diag_cap_dict(), which drops the whole dict when any plugin
exceeds the limit, this function caps each plugin independently so
one noisy plugin does not silence every other plugin's contribution.
Parameters
----------
v:
The raw contributions dict from the POST payload.
known_ids:
When provided, contributions from plugins not in this set are
skipped *before* serialisation, preventing a malicious caller
from forcing the server to JSON-encode hundreds of near-limit
payloads that ``build_bundle()`` would later discard anyway.
``None`` means "accept all plugin ids" (used in tests / preview).
"""
if not isinstance(v, dict):
return None
result = {}
total_bytes = 0
for pid, contribution in v.items():
if not isinstance(pid, str):
continue
# Filter unknown plugin ids early — before serialising — so a
# crafted request cannot force large allocations for plugins that
# build_bundle() would drop.
if known_ids is not None and pid not in known_ids:
continue
try:
encoded = json.dumps(contribution, separators=(",", ":")).encode("utf-8", errors="replace")
except (TypeError, ValueError) as e:
log.warning(
"client_contributions[%r] is not JSON-serialisable, dropping: %s", pid, e
)
continue
if len(encoded) > _DIAG_MAX_CLIENT_PAYLOAD_BYTES:
log.warning(
"client_contributions[%r] exceeds %d bytes, dropping",
pid, _DIAG_MAX_CLIENT_PAYLOAD_BYTES,
)
continue
if total_bytes + len(encoded) > _DIAG_MAX_CONTRIBUTIONS_BYTES:
log.warning(
"client_contributions aggregate size limit (%d bytes) reached, "
"dropping remaining entries",
_DIAG_MAX_CONTRIBUTIONS_BYTES,
)
break
result[pid] = contribution
total_bytes += len(encoded)
return result or None
@router.post("/api/diagnostics/export")
def export_diagnostics(payload: dict = Body(default_factory=dict)):
"""Build a diagnostic bundle and stream it back as a zip download.
The browser layers in `client_console`, `client_hardware`,
`client_ua`, and `local_storage` before posting; the server adds
server logs, hardware, plugin inventory, and packages everything
into a single zip.
Errors during plugin diagnostics callables are caught and logged
to the bundle's manifest `notes` rather than failing the export.
"""
from plugins import LOADED_PLUGINS, PLUGINS_LOCK
redact = _diag_coerce_bool(payload.get("redact", True), default=True)
include = _diag_normalize_include(payload.get("include"))
client_console = _diag_cap_console(payload.get("client_console"))
client_hardware = _diag_cap_dict(payload.get("client_hardware"))
client_ua = _diag_cap_dict(payload.get("client_ua"))
local_storage = _diag_cap_dict(payload.get("local_storage"))
# Fetch the plugin list first so we can filter contributions to known
# plugin ids before serialising — prevents a crafted request from
# forcing large allocations for plugins build_bundle() would drop.
with PLUGINS_LOCK:
plugins_snapshot = list(LOADED_PLUGINS)
known_ids = {p.get("id") for p in plugins_snapshot if isinstance(p.get("id"), str)}
client_contributions = _diag_cap_contributions(
payload.get("client_contributions"), known_ids=known_ids
)
zip_bytes, filename, _manifest = _diag_build(
feedBack_version=appstate.running_version(),
config_dir=appstate.config_dir,
dlc_dir=_get_dlc_dir(),
log_file=_diag_log_file(),
loaded_plugins=plugins_snapshot,
include=include,
redact=redact,
client_console=client_console,
client_hardware=client_hardware,
client_ua=client_ua,
local_storage=local_storage,
client_contributions=client_contributions,
log=log,
plugins_root=_diag_plugins_roots(),
)
return Response(
content=zip_bytes,
media_type="application/zip",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
@router.get("/api/diagnostics/preview")
def preview_diagnostics(
redact: bool = True,
system: bool = True,
hardware: bool = True,
logs: bool = True,
console: bool = True,
plugins: bool = True,
):
"""Return what `/api/diagnostics/export` would produce, minus the
actual file contents — file tree, sizes, schemas, redaction counts.
Lets the Settings UI show the user what's about to be sent."""
from plugins import LOADED_PLUGINS, PLUGINS_LOCK
include = {
"system": system,
"hardware": hardware,
"logs": logs,
"console": console,
"plugins": plugins,
}
with PLUGINS_LOCK:
plugins_snapshot = list(LOADED_PLUGINS)
return _diag_preview(
feedBack_version=appstate.running_version(),
config_dir=appstate.config_dir,
dlc_dir=_get_dlc_dir(),
log_file=_diag_log_file(),
loaded_plugins=plugins_snapshot,
include=include,
redact=redact,
log=log,
plugins_root=_diag_plugins_roots(),
)
@router.get("/api/diagnostics/hardware")
def diagnostics_hardware():
"""Backend hardware probe (cross-platform). Reusable independently
of the bundle export — handy for "what's my GPU" plugin queries."""
return _diag_hardware()
+16 -272
View File
@@ -55,7 +55,7 @@ from dlc_paths import _get_dlc_dir, _resolve_dlc_path
# Lives in lib/ because that is the one core dir every packaging path copies. # Lives in lib/ because that is the one core dir every packaging path copies.
import appstate import appstate
# Extracted route modules. They import `appstate`, never `server` — one-way graph. # Extracted route modules. They import `appstate`, never `server` — one-way graph.
from routers import audio_effects, artist_aliases, loops, playlists, ws_highway, chart, wanted, library_extras, shop, progression, profile, stats, version from routers import audio_effects, artist_aliases, loops, playlists, ws_highway, chart, wanted, library_extras, shop, progression, profile, stats, version, diagnostics
import sloppak as sloppak_mod import sloppak as sloppak_mod
import loosefolder as loosefolder_mod import loosefolder as loosefolder_mod
# Pure text-matching engine for MusicBrainz enrichment (P8): denoise/score/ # Pure text-matching engine for MusicBrainz enrichment (P8): denoise/score/
@@ -5257,6 +5257,11 @@ def _running_version() -> str:
return "unknown" return "unknown"
# _running_version is defined below the import-top configure() calls, so publish
# it here (configure is idempotent/additive) for routers/diagnostics.py.
appstate.configure(running_version=_running_version)
def _validate_server_config_types(cfg: dict) -> str | None: def _validate_server_config_types(cfg: dict) -> str | None:
"""Type-and-range gate for the server_config block of an import """Type-and-range gate for the server_config block of an import
bundle, mirroring the per-key checks in `POST /api/settings`. The bundle, mirroring the per-key checks in `POST /api/settings`. The
@@ -5983,277 +5988,16 @@ def import_settings(bundle: dict):
} }
# ── Diagnostic bundle export (feedBack#166) ────────────────────────── # ── Diagnostic bundle export (feedBack#166) → routers/diagnostics.py (R3) ────
# # The pure caps/normalisers are re-exported so existing server._diag_* /
# One-click "Export Diagnostics" in Settings produces a redacted zip # server._DIAG_* tests keep resolving (none of them monkeypatch these).
# combining server logs, system info, hardware (CPU/GPU/RAM), plugin from routers.diagnostics import ( # noqa: E402 (re-export for test compatibility)
# inventory, and the browser-side console transcript + hardware probe. _diag_cap_console, _diag_cap_contributions, _diag_cap_dict,
# The bundle format is specified in docs/diagnostics-bundle-spec.md. _diag_coerce_bool, _diag_normalize_include,
_DIAG_MAX_CLIENT_PAYLOAD_BYTES, _DIAG_MAX_CONSOLE_BYTES,
from fastapi import Body _DIAG_MAX_CONSOLE_ENTRIES, _DIAG_MAX_CONTRIBUTIONS_BYTES,
)
from diagnostics_bundle import build_bundle as _diag_build, preview_bundle as _diag_preview app.include_router(diagnostics.router)
from diagnostics_hardware import collect as _diag_hardware
def _diag_log_file() -> Path | None:
raw = os.environ.get("LOG_FILE", "").strip()
if not raw:
return None
return Path(raw)
def _diag_plugins_roots() -> list[Path]:
"""Return all plugin root directories for orphan scanning.
Includes both the built-in ``plugins/`` directory and
``FEEDBACK_PLUGINS_DIR`` when set, so user-installed plugins and
orphans in the external dir are reflected in the bundle.
"""
roots: list[Path] = []
user_dir = getenv_compat("FEEDBACK_PLUGINS_DIR", "").strip()
if user_dir:
p = Path(user_dir)
if p.is_dir():
roots.append(p)
builtin = Path(__file__).parent / "plugins"
if builtin not in roots:
roots.append(builtin)
return roots
def _diag_coerce_bool(v, *, default: bool = True) -> bool:
"""Coerce a request-side value to bool, accepting both JSON booleans and
string representations.
- Falsy strings: ``"false"``, ``"0"``, ``"no"``, ``""`` ``False``
- ``None`` *default*
- Everything else (including ``"true"``, ``"1"``) ``True``
"""
if v is None:
return default
if isinstance(v, bool):
return v
if isinstance(v, str):
return v.strip().lower() not in ("false", "0", "no", "")
return bool(v)
def _diag_normalize_include(include: dict | None) -> dict:
"""Coerce request-side flags to the booleans build_bundle expects.
Missing keys default to True so a bare {} request still produces
the full bundle.
Accepts both JSON booleans (``true``/``false``) and string
representations so callers that serialize flags as strings behave
consistently with the preview endpoint:
- Falsy strings: ``"false"``, ``"0"``, ``"no"``, ``""`` ``False``
- Everything else (including ``"true"``, ``"1"``, ``"yes"``) ``True``
"""
keys = ("system", "hardware", "logs", "console", "plugins")
if not isinstance(include, dict):
return {k: True for k in keys}
return {k: _diag_coerce_bool(include.get(k), default=True) for k in keys}
# Server-side caps on client-supplied payload sections. diagnostics.js
# enforces a 500-entry / ~250 KB ring buffer on the browser side; these
# bounds give generous headroom while still preventing a crafted POST from
# forcing the server to allocate arbitrarily large in-memory bundles.
_DIAG_MAX_CONSOLE_ENTRIES = 1000 # hard cap: truncate silently
_DIAG_MAX_CONSOLE_BYTES = 2 * 1024 * 1024 # 2 MB hard cap on total console list
_DIAG_MAX_CLIENT_PAYLOAD_BYTES = 2 * 1024 * 1024 # 2 MB per dict section
_DIAG_MAX_CONTRIBUTIONS_BYTES = 4 * 1024 * 1024 # 4 MB aggregate cap for contributions
def _diag_cap_console(v) -> list | None:
"""Return *v* if it is a list, truncated to _DIAG_MAX_CONSOLE_ENTRIES entries
and _DIAG_MAX_CONSOLE_BYTES total. Entries are accumulated until either cap
is reached; no partial-entry splitting occurs."""
if not isinstance(v, list):
return None
result = v[:_DIAG_MAX_CONSOLE_ENTRIES]
# Also enforce a byte cap — the count cap alone does not bound memory when
# entries contain arbitrarily large strings.
try:
out = []
total = 0
for entry in result:
encoded = json.dumps(entry, separators=(",", ":")).encode("utf-8", errors="replace")
if total + len(encoded) > _DIAG_MAX_CONSOLE_BYTES:
break
out.append(entry)
total += len(encoded)
return out
except (TypeError, ValueError):
return None
def _diag_cap_dict(v) -> dict | None:
"""Return *v* if it is a dict whose JSON serialisation fits within
_DIAG_MAX_CLIENT_PAYLOAD_BYTES, otherwise return None."""
if not isinstance(v, dict):
return None
try:
encoded = json.dumps(v, separators=(",", ":")).encode("utf-8", errors="replace")
except (TypeError, ValueError) as e:
log.warning("diagnostics client payload is not JSON-serialisable, dropping: %s", e)
return None
if len(encoded) > _DIAG_MAX_CLIENT_PAYLOAD_BYTES:
return None
return v
def _diag_cap_contributions(v, known_ids=None) -> dict | None:
"""Apply per-plugin and aggregate size caps on client_contributions.
Unlike _diag_cap_dict(), which drops the whole dict when any plugin
exceeds the limit, this function caps each plugin independently so
one noisy plugin does not silence every other plugin's contribution.
Parameters
----------
v:
The raw contributions dict from the POST payload.
known_ids:
When provided, contributions from plugins not in this set are
skipped *before* serialisation, preventing a malicious caller
from forcing the server to JSON-encode hundreds of near-limit
payloads that ``build_bundle()`` would later discard anyway.
``None`` means "accept all plugin ids" (used in tests / preview).
"""
if not isinstance(v, dict):
return None
result = {}
total_bytes = 0
for pid, contribution in v.items():
if not isinstance(pid, str):
continue
# Filter unknown plugin ids early — before serialising — so a
# crafted request cannot force large allocations for plugins that
# build_bundle() would drop.
if known_ids is not None and pid not in known_ids:
continue
try:
encoded = json.dumps(contribution, separators=(",", ":")).encode("utf-8", errors="replace")
except (TypeError, ValueError) as e:
log.warning(
"client_contributions[%r] is not JSON-serialisable, dropping: %s", pid, e
)
continue
if len(encoded) > _DIAG_MAX_CLIENT_PAYLOAD_BYTES:
log.warning(
"client_contributions[%r] exceeds %d bytes, dropping",
pid, _DIAG_MAX_CLIENT_PAYLOAD_BYTES,
)
continue
if total_bytes + len(encoded) > _DIAG_MAX_CONTRIBUTIONS_BYTES:
log.warning(
"client_contributions aggregate size limit (%d bytes) reached, "
"dropping remaining entries",
_DIAG_MAX_CONTRIBUTIONS_BYTES,
)
break
result[pid] = contribution
total_bytes += len(encoded)
return result or None
@app.post("/api/diagnostics/export")
def export_diagnostics(payload: dict = Body(default_factory=dict)):
"""Build a diagnostic bundle and stream it back as a zip download.
The browser layers in `client_console`, `client_hardware`,
`client_ua`, and `local_storage` before posting; the server adds
server logs, hardware, plugin inventory, and packages everything
into a single zip.
Errors during plugin diagnostics callables are caught and logged
to the bundle's manifest `notes` rather than failing the export.
"""
from plugins import LOADED_PLUGINS, PLUGINS_LOCK
redact = _diag_coerce_bool(payload.get("redact", True), default=True)
include = _diag_normalize_include(payload.get("include"))
client_console = _diag_cap_console(payload.get("client_console"))
client_hardware = _diag_cap_dict(payload.get("client_hardware"))
client_ua = _diag_cap_dict(payload.get("client_ua"))
local_storage = _diag_cap_dict(payload.get("local_storage"))
# Fetch the plugin list first so we can filter contributions to known
# plugin ids before serialising — prevents a crafted request from
# forcing large allocations for plugins build_bundle() would drop.
with PLUGINS_LOCK:
plugins_snapshot = list(LOADED_PLUGINS)
known_ids = {p.get("id") for p in plugins_snapshot if isinstance(p.get("id"), str)}
client_contributions = _diag_cap_contributions(
payload.get("client_contributions"), known_ids=known_ids
)
zip_bytes, filename, _manifest = _diag_build(
feedBack_version=_running_version(),
config_dir=CONFIG_DIR,
dlc_dir=_get_dlc_dir(),
log_file=_diag_log_file(),
loaded_plugins=plugins_snapshot,
include=include,
redact=redact,
client_console=client_console,
client_hardware=client_hardware,
client_ua=client_ua,
local_storage=local_storage,
client_contributions=client_contributions,
log=log,
plugins_root=_diag_plugins_roots(),
)
return Response(
content=zip_bytes,
media_type="application/zip",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
@app.get("/api/diagnostics/preview")
def preview_diagnostics(
redact: bool = True,
system: bool = True,
hardware: bool = True,
logs: bool = True,
console: bool = True,
plugins: bool = True,
):
"""Return what `/api/diagnostics/export` would produce, minus the
actual file contents file tree, sizes, schemas, redaction counts.
Lets the Settings UI show the user what's about to be sent."""
from plugins import LOADED_PLUGINS, PLUGINS_LOCK
include = {
"system": system,
"hardware": hardware,
"logs": logs,
"console": console,
"plugins": plugins,
}
with PLUGINS_LOCK:
plugins_snapshot = list(LOADED_PLUGINS)
return _diag_preview(
feedBack_version=_running_version(),
config_dir=CONFIG_DIR,
dlc_dir=_get_dlc_dir(),
log_file=_diag_log_file(),
loaded_plugins=plugins_snapshot,
include=include,
redact=redact,
log=log,
plugins_root=_diag_plugins_roots(),
)
@app.get("/api/diagnostics/hardware")
def diagnostics_hardware():
"""Backend hardware probe (cross-platform). Reusable independently
of the bundle export handy for "what's my GPU" plugin queries."""
return _diag_hardware()
# ── Plugin-provided routes are registered at startup via plugins/__init__.py ─ # ── Plugin-provided routes are registered at startup via plugins/__init__.py ─