music-assistant-server

1"""
2Always-on diagnostics capture and sanitization helpers.
3
4This module is intentionally stdlib-only and free of any Music Assistant imports so the
5capture handler can be installed at the earliest possible moment (before the server core
6is even constructed) and adopted later by the diagnostics controller. The always-on cost
7is limited to one logging handler doing bounded, in-memory ring buffer appends;
8sanitization and report building only happen when a report is explicitly requested.
9"""
10
11from __future__ import annotations
12
13import hashlib
14import ipaddress
15import logging
16import re
17import time
18import traceback
19from collections import OrderedDict, deque
20from dataclasses import dataclass, replace
21from typing import NamedTuple
22
23# bounds for the captured data (everything in memory, so keep it small)
24LOG_RING_MAXLEN = 300
25MAX_EXCEPTION_FINGERPRINTS = 100
26MAX_TRACEBACK_CHARS = 6000
27MAX_MESSAGE_CHARS = 500
28TRACEBACK_FRAME_LIMIT = 50
29
30REDACTION_NOTICE = (
31    "This report is automatically redacted: home directories, media file paths/names, "
32    "URL credentials and query strings, passwords/tokens/secrets, e-mail addresses, "
33    "non-local IP addresses and MAC addresses are replaced by placeholders. Absolute "
34    "code paths are shortened to be relative to the application root. Media paths and "
35    "MAC addresses are reduced to a stable hash so identical values can still be "
36    "correlated within the report."
37)
38
39# file extensions that reveal (music) library content when they appear in a path
40_MEDIA_FILE_EXTENSIONS = (
41    "aac|aif|aiff|alac|ape|asx|avi|cue|dff|dsf|flac|gif|jpeg|jpg|m3u|m3u8|m4a|m4b|m4v|"
42    "mka|mkv|mov|mp3|mp4|nfo|oga|ogg|opus|pls|png|wav|webm|webp|wma|wv|xspf"
43)
44
45# absolute code paths -> relative to site-packages / music_assistant root
46_RE_SITE_PACKAGES_PATH = re.compile(r"(?:[A-Za-z]:)?[^\s\"']*[/\\]site-packages[/\\]")
47_RE_MASS_ROOT_PATH = re.compile(r"(?:[A-Za-z]:)?[^\s\"']*[/\\]music_assistant[/\\]")
48# URL userinfo (user:pass@host) and query strings (tokens/signatures live there)
49_RE_URL_USERINFO = re.compile(r"(\w+://)[^/\s@\"']+@")
50_RE_URL_QUERY = re.compile(r"(\w+://[^\s\"'<>]*?)\?[^\s\"'<>]*")
51# query strings on path-style (relative) request URLs, e.g. OAuth callbacks
52_RE_PATH_QUERY = re.compile(r"(?<!\w)(/[^\s\"'<>?]*)\?[^\s\"'<>]+")
53# secrets: JWT's, Authorization header schemes, key=value assignments, long token blobs
54_RE_JWT = re.compile(r"\beyJ[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]*")
55_RE_AUTH_SCHEME = re.compile(r"(?i)\b(bearer|basic|digest)\s+[A-Za-z0-9\-._~+/=]{16,}")
56_RE_SECRET_ASSIGNMENT = re.compile(
57    r"(?i)\b([\w-]*(?:password|passwd|pwd|secret|token|api[_-]?key|apikey|auth|"
58    r"authorization|credentials?|access[_-]?key|private[_-]?key|cookie|csrf))\b"
59    r"(\"?'?\s*[=:]\s*)(\"[^\"\r\n]*\"|'[^'\r\n]*'|[^\s\"',;&]+)"
60)
61# long token blobs: 32+ chars of token-ish alphabet containing at least one digit
62# (the digit requirement spares long snake_case identifiers and module paths)
63_RE_LONG_TOKEN = re.compile(
64    r"(?<![A-Za-z0-9+_-])(?=[A-Za-z0-9+_-]*\d)[A-Za-z0-9+_-]{32,}(?![A-Za-z0-9+_-])"
65)
66# media library paths: full paths (at least one path separator) and bare filenames;
67# bare filenames may contain spaces, so unquoted/undelimited prose directly in front
68# of a media filename is redacted along with it (privacy beats message fidelity here)
69_RE_MEDIA_PATH = re.compile(
70    rf"(?<!\w)(?:[A-Za-z]:)?(?:[/\\][^/\\\r\n]*?)+\.(?:{_MEDIA_FILE_EXTENSIONS})\b",
71    re.IGNORECASE,
72)
73_RE_MEDIA_FILENAME = re.compile(
74    rf"(?<![\w/\\.])[^\s/\\:*?\"'<>|(][^/\\:*?\"'<>|\r\n]*\.(?:{_MEDIA_FILE_EXTENSIONS})\b",
75    re.IGNORECASE,
76)
77_RE_EMAIL = re.compile(r"\b[\w.+-]+@[\w-]+(?:\.[\w-]+)+\b")
78# MAC addresses (also common as/inside player ids), colon or dash separated
79_RE_MAC_ADDRESS = re.compile(
80    r"(?<![0-9A-Fa-f:-])[0-9A-Fa-f]{2}(?:([:-])[0-9A-Fa-f]{2}){5}(?![0-9A-Fa-f:-])"
81)
82# IPv6 first (so IPv4-mapped addresses are redacted as a whole), candidates are
83# verified with the ipaddress module to avoid eating timestamps and MAC addresses
84_RE_IPV6_CANDIDATE = re.compile(
85    r"(?<![\w.:])(?:[0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}(?![\w:])"
86    r"|(?<![\w.:])[0-9A-Fa-f:]*::[0-9A-Fa-f:.]*(?![\w:.])"
87)
88_RE_IPV4 = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")
89_RE_HOME_DIR = re.compile(r"(?:/(?:Users|home)/|[A-Za-z]:\\Users\\)[^/\\\s:;,'\"]+")
90
91
92def sanitize_text(text: str) -> str:
93    """
94    Redact privacy-sensitive data from a text snippet.
95
96    Applied to every string that ends up in a diagnostics report (log messages,
97    tracebacks, error strings): redacts home directories, media file paths, URL
98    credentials/query strings, secrets/tokens, e-mail addresses, IP addresses and
99    MAC addresses, and shortens absolute code paths to be relative to the
100    application root.
101
102    :param text: The raw text to sanitize.
103    """
104    text = _RE_SITE_PACKAGES_PATH.sub("", text)
105    text = _RE_MASS_ROOT_PATH.sub("music_assistant/", text)
106    text = _RE_URL_QUERY.sub(r"\1?<redacted-query>", text)
107    text = _RE_URL_USERINFO.sub(r"\1<redacted>@", text)
108    text = _RE_PATH_QUERY.sub(r"\1?<redacted-query>", text)
109    text = _RE_JWT.sub("<redacted-token>", text)
110    text = _RE_AUTH_SCHEME.sub(r"\1 <redacted>", text)
111    text = _RE_SECRET_ASSIGNMENT.sub(r"\1\2<redacted>", text)
112    text = _RE_LONG_TOKEN.sub("<redacted-token>", text)
113    text = _RE_MEDIA_PATH.sub(_redact_media_path, text)
114    text = _RE_MEDIA_FILENAME.sub(_redact_media_path, text)
115    text = _RE_EMAIL.sub("<redacted-email>", text)
116    text = _RE_MAC_ADDRESS.sub(_redact_mac, text)
117    text = _RE_IPV6_CANDIDATE.sub(_redact_ip, text)
118    text = _RE_IPV4.sub(_redact_ip, text)
119    return _RE_HOME_DIR.sub("~", text)
120
121
122def sanitize_data[DataT](data: DataT) -> DataT:
123    """
124    Recursively sanitize all strings (including dict keys) inside a data structure.
125
126    :param data: Plain data (dicts/lists/tuples/scalars) to sanitize in depth.
127    """
128    if isinstance(data, str):
129        return sanitize_text(data)  # type: ignore[return-value]
130    if isinstance(data, dict):
131        return {sanitize_data(key): sanitize_data(value) for key, value in data.items()}  # type: ignore[return-value]
132    if isinstance(data, tuple):
133        return tuple(sanitize_data(value) for value in data)  # type: ignore[return-value]
134    if isinstance(data, list):
135        return [sanitize_data(value) for value in data]  # type: ignore[return-value]
136    return data
137
138
139class CapturedLogRecord(NamedTuple):
140    """A single captured (WARNING or higher) log record."""
141
142    created: float
143    level: str
144    logger_name: str
145    message: str
146
147
148@dataclass
149class CapturedException:
150    """Aggregated info for one unique exception fingerprint."""
151
152    fingerprint: str
153    exc_type: str
154    origin: str
155    logger_name: str
156    level: str
157    message: str
158    traceback_exc: traceback.TracebackException
159    first_seen: float
160    last_seen: float
161    count: int = 1
162
163    def render_traceback(self) -> str:
164        """Render the representative traceback as text (may read source files)."""
165        try:
166            return "".join(self.traceback_exc.format())[-MAX_TRACEBACK_CHARS:]
167        except Exception as err:
168            return f"<traceback rendering failed: {err!r}>"
169
170
171class DiagnosticsLogHandler(logging.Handler):
172    """
173    Root-logger handler that captures warnings/errors for the diagnostics report.
174
175    Keeps a bounded ring of recent WARNING+ records plus exception aggregates keyed by
176    fingerprint (exception type + raise site + topmost music_assistant frame). All data
177    stays in memory as-is; sanitization happens only when a report is built.
178    """
179
180    def __init__(self) -> None:
181        """Initialize the capture handler."""
182        super().__init__(level=logging.WARNING)
183        self.installed_at = time.monotonic()
184        self.log_ring: deque[CapturedLogRecord] = deque(maxlen=LOG_RING_MAXLEN)
185        self.exceptions: OrderedDict[str, CapturedException] = OrderedDict()
186
187    def emit(self, record: logging.LogRecord) -> None:
188        """Capture a single log record (may never raise or block)."""
189        try:
190            message = record.getMessage()[:MAX_MESSAGE_CHARS]
191            self.log_ring.append(
192                CapturedLogRecord(record.created, record.levelname, record.name, message)
193            )
194            if record.exc_info and record.exc_info[0] is not None:
195                self._capture_exception(record, message)
196        except Exception:  # noqa: S110 - capturing diagnostics may never break logging itself
197            pass
198
199    def snapshot(self) -> tuple[list[CapturedLogRecord], list[CapturedException]]:
200        """Return a point-in-time copy of the captured log ring and exception aggregates."""
201        self.acquire()
202        try:
203            return list(self.log_ring), [replace(entry) for entry in self.exceptions.values()]
204        finally:
205            self.release()
206
207    def clear(self) -> None:
208        """Clear all captured data (intended for tests)."""
209        self.acquire()
210        try:
211            self.log_ring.clear()
212            self.exceptions.clear()
213        finally:
214            self.release()
215
216    def _capture_exception(self, record: logging.LogRecord, message: str) -> None:
217        """Aggregate an exception attached to a log record by fingerprint."""
218        assert record.exc_info is not None  # guarded by caller
219        exc_type, exc_value, exc_tb = record.exc_info
220        if exc_type is None or exc_value is None:
221            return
222        # walk the raw traceback frames without touching linecache/source files,
223        # keeping this always-on path cheap
224        raise_site = ""
225        raise_site_key = ""
226        origin = ""
227        origin_key = ""
228        frame_tb = exc_tb
229        while frame_tb is not None:
230            code = frame_tb.tb_frame.f_code
231            raise_site = f"{code.co_filename}:{frame_tb.tb_lineno} in {code.co_name}"
232            raise_site_key = f"{code.co_filename}:{code.co_name}"
233            if "music_assistant" in code.co_filename:
234                origin = raise_site
235                origin_key = raise_site_key
236            frame_tb = frame_tb.tb_next
237        exc_type_name = exc_type.__qualname__ if exc_type else "UnknownError"
238        # fingerprint on function granularity (not line numbers) so retries/loops aggregate
239        fingerprint_source = f"{exc_type_name}|{raise_site_key}|{origin_key}"
240        fingerprint = hashlib.sha256(fingerprint_source.encode()).hexdigest()[:12]
241        if entry := self.exceptions.get(fingerprint):
242            entry.count += 1
243            entry.last_seen = record.created
244            self.exceptions.move_to_end(fingerprint)
245            return
246        # capture structured traceback metadata once per unique fingerprint;
247        # lookup_lines=False defers all source file (linecache) reads to report time
248        traceback_exc = traceback.TracebackException(
249            exc_type,
250            exc_value,
251            exc_tb,
252            limit=TRACEBACK_FRAME_LIMIT,
253            lookup_lines=False,
254            compact=True,
255        )
256        self.exceptions[fingerprint] = CapturedException(
257            fingerprint=fingerprint,
258            exc_type=exc_type_name,
259            origin=origin or raise_site,
260            logger_name=record.name,
261            level=record.levelname,
262            message=message,
263            traceback_exc=traceback_exc,
264            first_seen=record.created,
265            last_seen=record.created,
266        )
267        while len(self.exceptions) > MAX_EXCEPTION_FINGERPRINTS:
268            self.exceptions.popitem(last=False)
269
270
271def install_diagnostics_log_handler() -> DiagnosticsLogHandler:
272    """
273    Install (or return the already installed) diagnostics capture handler.
274
275    Idempotent: attaches a single handler instance to the root logger so warnings,
276    errors and exceptions are captured from the earliest moment possible, independent
277    of controller setup order (also covers embedded usage where __main__ is not used).
278    """
279    root_logger = logging.getLogger()
280    for handler in root_logger.handlers:
281        if isinstance(handler, DiagnosticsLogHandler):
282            return handler
283    handler = DiagnosticsLogHandler()
284    root_logger.addHandler(handler)
285    return handler
286
287
288def _redact_media_path(match: re.Match[str]) -> str:
289    """Replace a matched media file path with a stable hash placeholder plus extension."""
290    path = match.group(0)
291    extension = path.rsplit(".", 1)[-1].lower()
292    digest = hashlib.sha256(path.encode()).hexdigest()[:8]
293    return f"<path-{digest}>.{extension}"
294
295
296def _redact_mac(match: re.Match[str]) -> str:
297    """Replace a matched MAC address with a stable hash placeholder."""
298    digest = hashlib.sha256(match.group(0).lower().encode()).hexdigest()[:8]
299    return f"<mac-{digest}>"
300
301
302def _redact_ip(match: re.Match[str]) -> str:
303    """Replace a matched IP address candidate, keeping localhost and invalid candidates."""
304    candidate = match.group(0)
305    try:
306        ip_address = ipaddress.ip_address(candidate)
307    except ValueError:
308        # not an actual IP address (e.g. a timestamp or MAC address lookalike)
309        return candidate
310    if ip_address.is_loopback or ip_address.is_unspecified:
311        return candidate
312    return "<redacted-ip>"
313