diff --git a/.jules/sentinel.md b/.jules/sentinel.md index 34122c2b4..2dc3a51f0 100644 --- a/.jules/sentinel.md +++ b/.jules/sentinel.md @@ -6,7 +6,7 @@ ## 2024-07-07 - Unsanitized Directory Input Paths API Validation **Vulnerability:** The API logic allowed user-controlled local data directory paths (`cacheRoot` and `tempRoot`) to be directly used without mitigating cross-platform path traversal vulnerabilities. **Learning:** Checking for '..' sequences in untrusted paths fails to parse cross-platform separators reliably for untrusted inputs (e.g., Windows backslashes on POSIX). Relying solely on `os.sep` or `os.altsep` is inadequate because absolute paths can bypass restrictions if not resolved correctly, or if `os.altsep` is None. -**Prevention:** Manually replace backslashes with forward slashes and split by forward slash (e.g., `if '..' in path.replace('\\', '/').split('/')`) to enforce path traversal protections explicitly for restricted directory inputs provided via the API. Do not block `~` for user-selected input files. +**Prevention:** Manually replace backslashes with forward slashes and split by forward slash (e.g. `if '..' in path.replace('\\', '/').split('/')`) to enforce path traversal protections explicitly for restricted directory inputs provided via the API. Do not block `~` for user-selected input files. ## 2024-05-20 - Python Path Traversal Mitigation bypass **Vulnerability:** Path traversal detection in Python backend APIs relied solely on checking the input path string or basic parsed parts which might not adequately catch sequences like `..` when intermixed with different path separators. @@ -28,3 +28,8 @@ **Vulnerability:** The Rust backend (`apps/desktop/src-tauri/src/main.rs`) did not enforce a maximum URL length limit when processing YouTube URLs via `import_youtube_url`. While the frontend enforced `MAX_YOUTUBE_URL_LENGTH = 2000` via the input element, this could be bypassed by an attacker sending requests directly to the Tauri backend API, potentially causing a Denial of Service (DoS) due to unbounded URL parsing and regex matching. **Learning:** Input validation must occur at the entry point of untrusted data on the backend, even if it is also validated on the frontend. Relying solely on frontend validation for constraints like string length can expose the backend to resource exhaustion vulnerabilities. **Prevention:** Always enforce constraints like maximum length, format validation, and sanitization at the earliest possible point on the backend, typically at the API boundary, regardless of frontend safeguards. + +## 2026-09-21 - Prevent log forging and excessive diagnostic output +**Vulnerability:** Untrusted file names, paths, or decoder exception text could carry carriage returns, line separators, terminal controls, or oversized payloads into line-oriented logs or the analyzer's wrapped failure text. Invoking an arbitrary dependency exception's `__repr__` or `__str__` while handling a failure also executes dependency-controlled code on an error path and can replace the original decoder failure or allocate an unbounded representation before sanitization. +**Learning:** Deferred logging interpolation avoids eager template construction but does not neutralize controls or bound diagnostic size. Sanitizing the result of `repr()` or `str()` is too late to make arbitrary dependency representation code a trusted boundary. Diagnostics should consume bounded plain-text fields and code-owned exception metadata without invoking dependency `str`/`repr`; control escaping and size limits belong before both log formatting and error-envelope construction. +**Prevention:** Keep log templates parameterized. Route buyer/dependency text through the shared bounded single-line renderer, which stops consuming input when the rendered budget is exhausted and marks truncation. For arbitrary exceptions, read only the runtime type name and an exact-string base exception argument; do not call dependency `__repr__` or `__str__` for the log record or the analyzer's wrapped `ValueError`. Regression tests cover LF/CRLF, tab, ANSI ESC, NUL/C0, Unicode line separators, printable Unicode, hostile/broken `__repr__`, broken `__str__`, oversized exception messages, and oversized buyer-controlled file names. One logical event must remain one bounded physical log record, and diagnostic handling must not mask the original decoder failure. diff --git a/services/analysis-engine/src/bandscope_analysis/cli.py b/services/analysis-engine/src/bandscope_analysis/cli.py index 6838ee711..422b6d6f7 100644 --- a/services/analysis-engine/src/bandscope_analysis/cli.py +++ b/services/analysis-engine/src/bandscope_analysis/cli.py @@ -8,6 +8,7 @@ from datetime import UTC, datetime from bandscope_analysis.api import get_analysis_status, run_analysis_job, run_analysis_job_updates +from bandscope_analysis.logging_safety import safe_log_value from bandscope_analysis.temporal import TemporalAnalyzer logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") @@ -86,15 +87,15 @@ def main() -> int: audio_path = local_source.get("sourcePath") file_name = local_source.get("fileName", "selected audio") if audio_path: - logging.info("Extracting temporal features from %s...", file_name) + logging.info("Extracting temporal features from %s...", safe_log_value(file_name)) try: temporal_analyzer = TemporalAnalyzer() features = temporal_analyzer.analyze(audio_path) - logging.info(f"Extracted BPM: {features['bpm']}") + logging.info("Extracted BPM: %s", features["bpm"]) except Exception: logging.warning( "Temporal analysis failed for %s; continuing with safe fallback.", - file_name, + safe_log_value(file_name), ) requested_at = datetime.now(UTC).isoformat().replace("+00:00", "Z") diff --git a/services/analysis-engine/src/bandscope_analysis/logging_safety.py b/services/analysis-engine/src/bandscope_analysis/logging_safety.py new file mode 100644 index 000000000..e3ad51e7d --- /dev/null +++ b/services/analysis-engine/src/bandscope_analysis/logging_safety.py @@ -0,0 +1,89 @@ +"""Bounded rendering helpers for untrusted analysis diagnostics.""" + +from __future__ import annotations + +MAX_LOG_DIAGNOSTIC_CHARS = 1024 +MAX_LOG_TYPE_CHARS = 128 +_LOG_TRUNCATION_SUFFIX = "..." + + +def _safe_type_name(value: object) -> str: + """Read a bounded runtime type name without invoking instance or metaclass hooks.""" + value_type = object.__getattribute__(value, "__class__") + type_name = type.__getattribute__(value_type, "__name__") + if type(type_name) is not str: + return "unknown" + return single_line_log_text(type_name, max_chars=MAX_LOG_TYPE_CHARS) + + +def single_line_log_text( + value: str, + *, + max_chars: int = MAX_LOG_DIAGNOSTIC_CHARS, +) -> str: + """Escape controls and cap one untrusted textual diagnostic field. + + The function stops reading input once the rendered budget is exhausted, so + oversized buyer/dependency text cannot force a full-size escaped copy solely + for diagnostics. + """ + if max_chars < len(_LOG_TRUNCATION_SUFFIX): + raise ValueError("max_chars is too small for the truncation marker") + + chunks: list[str] = [] + rendered_chars = 0 + truncated = False + for character in value: + safe_character = ( + character + if character.isprintable() + else character.encode("unicode_escape").decode("ascii") + ) + if rendered_chars + len(safe_character) > max_chars: + truncated = True + break + chunks.append(safe_character) + rendered_chars += len(safe_character) + + if truncated: + while chunks and rendered_chars + len(_LOG_TRUNCATION_SUFFIX) > max_chars: + rendered_chars -= len(chunks.pop()) + chunks.append(_LOG_TRUNCATION_SUFFIX) + + return "".join(chunks) + + +def safe_log_value(value: object) -> str: + """Render an untrusted log value without executing arbitrary representation code.""" + if type(value) is str: + return single_line_log_text(value) + return f"<{_safe_type_name(value)}>" + + +def safe_exception_message(error: BaseException) -> str: + """Return bounded exception text without calling dependency ``str``/``repr``.""" + try: + args = BaseException.args.__get__(error, type(error)) + except Exception: + return _safe_type_name(error) + + if not args or type(args[0]) is not str or not args[0]: + return _safe_type_name(error) + return single_line_log_text(args[0]) + + +def safe_exception_summary(error: BaseException) -> str: + """Render a bounded typed exception summary without dependency ``str``/``repr``.""" + error_type = _safe_type_name(error) + try: + args = BaseException.args.__get__(error, type(error)) + except Exception: + return error_type + + if not args or type(args[0]) is not str or not args[0]: + return error_type + + message_budget = MAX_LOG_DIAGNOSTIC_CHARS - len(error_type) - 2 + if message_budget < len(_LOG_TRUNCATION_SUFFIX): + return error_type + return f"{error_type}: {single_line_log_text(args[0], max_chars=message_budget)}" diff --git a/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py b/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py index 7fe5ae6f7..7a5064ce2 100644 --- a/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py +++ b/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py @@ -12,6 +12,7 @@ import numpy as np from numpy.typing import NDArray +from ..logging_safety import safe_exception_message, safe_exception_summary, safe_log_value from .model import TemporalFeatures logger = logging.getLogger(__name__) @@ -73,7 +74,7 @@ def analyze(self, audio_path: str | Path) -> TemporalFeatures: if not path.exists() or not path.is_file(): raise FileNotFoundError(f"Audio file not found: {path_str}") - logger.info(f"Loading and decoding audio: {path_str}") + logger.info("Loading and decoding audio: %s", safe_log_value(path_str)) try: with path.open("rb") as fileobj: @@ -128,7 +129,7 @@ def analyze(self, audio_path: str | Path) -> TemporalFeatures: bpm_val = float(tempo[0]) if isinstance(tempo, np.ndarray) else float(tempo) - logger.info(f"Analysis complete: {bpm_val:.1f} BPM, {len(beat_times)} beats detected.") + logger.info("Analysis complete: %.1f BPM, %d beats detected.", bpm_val, len(beat_times)) return { "bpm": bpm_val, @@ -140,5 +141,9 @@ def analyze(self, audio_path: str | Path) -> TemporalFeatures: } except Exception as e: - logger.error(f"Failed to analyze audio {path_str}: {e}") - raise ValueError(f"Temporal analysis failed: {e}") from e + logger.error( + "Failed to analyze audio %s: %s", + safe_log_value(path_str), + safe_exception_summary(e), + ) + raise ValueError(f"Temporal analysis failed: {safe_exception_message(e)}") from e diff --git a/services/analysis-engine/tests/test_logging_safety.py b/services/analysis-engine/tests/test_logging_safety.py new file mode 100644 index 000000000..5ba6ad44c --- /dev/null +++ b/services/analysis-engine/tests/test_logging_safety.py @@ -0,0 +1,346 @@ +"""Logging safety regressions for untrusted analysis inputs.""" + +from __future__ import annotations + +import io +import json +import logging +from pathlib import Path + +import pytest + +from bandscope_analysis import cli +from bandscope_analysis.temporal import TemporalAnalyzer +from bandscope_analysis.temporal import analyzer as analyzer_module + +_LOG_CONTROL_ESCAPES = ( + ("\n", "\\n"), + ("\r", "\\r"), + ("\t", "\\t"), + ("\x1b", "\\x1b"), + ("\x00", "\\x00"), + ("\u2028", "\\u2028"), + ("\u2029", "\\u2029"), +) +_HOSTILE_LOG_VALUES = ( + "FORGED\nSECURITY EVENT", + "FORGED\r\nSECURITY EVENT", + "FORGED\tSECURITY EVENT", + "FORGED\x1b[31mSECURITY EVENT", + "FORGED\x00SECURITY EVENT", + "FORGED\u2028SECURITY EVENT", + "FORGED\u2029SECURITY EVENT", + "정상-유니코드-é", +) + + +def _assert_log_value_is_single_record(rendered: str, untrusted_value: str) -> None: + for control_character, escaped_form in _LOG_CONTROL_ESCAPES: + assert control_character not in rendered + if control_character in untrusted_value: + assert escaped_form in rendered + if untrusted_value == "정상-유니코드-é": + assert untrusted_value in rendered + + +@pytest.mark.parametrize("untrusted_value", _HOSTILE_LOG_VALUES) +def test_temporal_error_log_escapes_exception_control_characters( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, + untrusted_value: str, +) -> None: + """Decoder failures must not inject a second physical log line.""" + audio_path = tmp_path / "buyer-audio.wav" + audio_path.write_bytes(b"not-a-real-wave") + + def fail_decode(*_args: object, **_kwargs: object) -> object: + """Raise a decoder error carrying the parameterized untrusted value.""" + raise RuntimeError(untrusted_value) + + monkeypatch.setattr(analyzer_module.librosa, "load", fail_decode) + caplog.set_level(logging.ERROR, logger=analyzer_module.__name__) + + with pytest.raises(ValueError, match="Temporal analysis failed"): + TemporalAnalyzer().analyze(audio_path) + + messages = [ + record.getMessage() + for record in caplog.records + if record.name == analyzer_module.__name__ and record.levelno >= logging.ERROR + ] + assert len(messages) == 1 + _assert_log_value_is_single_record(messages[0], untrusted_value) + + +def test_temporal_error_log_does_not_execute_hostile_exception_repr( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + """Log rendering must not execute dependency-controlled exception repr code.""" + audio_path = tmp_path / "buyer-audio.wav" + audio_path.write_bytes(b"not-a-real-wave") + repr_calls = 0 + + class HostileDecoderError(RuntimeError): + """Model a dependency exception whose repr has attacker-controlled behavior.""" + + def __repr__(self) -> str: + """Record execution and return an intentionally unsafe representation.""" + nonlocal repr_calls + repr_calls += 1 + return "HostileDecoderError('FORGED\nSECURITY EVENT\x1b[31m\u2028NEXT')" + + def fail_decode(*_args: object, **_kwargs: object) -> object: + """Raise the dependency-shaped exception through the real analyzer path.""" + raise HostileDecoderError("decoder failed") + + monkeypatch.setattr(analyzer_module.librosa, "load", fail_decode) + caplog.set_level(logging.ERROR, logger=analyzer_module.__name__) + + with pytest.raises(ValueError, match="Temporal analysis failed: decoder failed"): + TemporalAnalyzer().analyze(audio_path) + + messages = [ + record.getMessage() + for record in caplog.records + if record.name == analyzer_module.__name__ and record.levelno >= logging.ERROR + ] + assert len(messages) == 1 + assert repr_calls == 0 + assert "HostileDecoderError" in messages[0] + assert "decoder failed" in messages[0] + assert "FORGED" not in messages[0] + assert "\n" not in messages[0] + assert "\x1b" not in messages[0] + assert "\u2028" not in messages[0] + + +def test_temporal_error_log_survives_exception_repr_failure( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + """Logging must not call a broken repr while handling the decoder failure.""" + audio_path = tmp_path / "buyer-audio.wav" + audio_path.write_bytes(b"not-a-real-wave") + repr_calls = 0 + + class BrokenReprDecoderError(RuntimeError): + """Model a dependency exception whose repr is itself faulty.""" + + def __repr__(self) -> str: + """Raise if logging improperly executes dependency representation code.""" + nonlocal repr_calls + repr_calls += 1 + raise RuntimeError("repr failed") + + def fail_decode(*_args: object, **_kwargs: object) -> object: + """Raise the dependency-shaped exception through the real analyzer path.""" + raise BrokenReprDecoderError("decoder failed") + + monkeypatch.setattr(analyzer_module.librosa, "load", fail_decode) + caplog.set_level(logging.ERROR, logger=analyzer_module.__name__) + + with pytest.raises(ValueError, match="Temporal analysis failed: decoder failed"): + TemporalAnalyzer().analyze(audio_path) + + messages = [ + record.getMessage() + for record in caplog.records + if record.name == analyzer_module.__name__ and record.levelno >= logging.ERROR + ] + assert len(messages) == 1 + assert repr_calls == 0 + assert "BrokenReprDecoderError" in messages[0] + assert "decoder failed" in messages[0] + assert "repr unavailable" not in messages[0] + + +def test_temporal_failure_does_not_execute_dependency_str( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Wrapping a decoder failure must not execute dependency-controlled str code.""" + audio_path = tmp_path / "buyer-audio.wav" + audio_path.write_bytes(b"not-a-real-wave") + str_calls = 0 + + class BrokenStrDecoderError(RuntimeError): + """Model a decoder exception whose string conversion is unsafe.""" + + def __str__(self) -> str: + """Raise if the analyzer executes dependency string conversion.""" + nonlocal str_calls + str_calls += 1 + raise RuntimeError("str failed") + + def fail_decode(*_args: object, **_kwargs: object) -> object: + """Raise the dependency-shaped exception through the real analyzer path.""" + raise BrokenStrDecoderError("decoder failed") + + monkeypatch.setattr(analyzer_module.librosa, "load", fail_decode) + + with pytest.raises(ValueError, match="Temporal analysis failed: decoder failed"): + TemporalAnalyzer().analyze(audio_path) + + assert str_calls == 0 + + +def test_temporal_error_log_bounds_oversized_exception_message( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + """One decoder failure cannot emit an unbounded diagnostic log field.""" + audio_path = tmp_path / "buyer-audio.wav" + audio_path.write_bytes(b"not-a-real-wave") + oversized_message = "X" * 5000 + + def fail_decode(*_args: object, **_kwargs: object) -> object: + """Raise a dependency error with a deliberately oversized message.""" + raise RuntimeError(oversized_message) + + monkeypatch.setattr(analyzer_module.librosa, "load", fail_decode) + caplog.set_level(logging.ERROR, logger=analyzer_module.__name__) + + with pytest.raises(ValueError, match="Temporal analysis failed") as raised: + TemporalAnalyzer().analyze(audio_path) + + messages = [ + record.getMessage() + for record in caplog.records + if record.name == analyzer_module.__name__ and record.levelno >= logging.ERROR + ] + assert len(messages) == 1 + assert "" in messages[0] + assert len(messages[0]) <= 1200 + assert "" in str(raised.value) + assert len(str(raised.value)) <= 1200 + + +@pytest.mark.parametrize("untrusted_value", _HOSTILE_LOG_VALUES) +def test_cli_logs_untrusted_filename_as_single_line( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + untrusted_value: str, +) -> None: + """Local-audio labels must stay on one physical log line on fallback.""" + malicious_name = f"buyer-{untrusted_value}.wav" + stdin = io.StringIO( + json.dumps( + { + "jobId": "log-safety", + "request": { + "sourceKind": "local_audio", + "projectId": "project-log-safety", + "sourceLabel": "buyer.wav", + "roleFocus": [], + "localSource": { + "sourcePath": "/synthetic/buyer.wav", + "fileName": malicious_name, + "extension": "wav", + "fileSizeBytes": 1, + }, + }, + } + ) + ) + stdout = io.StringIO() + + class FailingAnalyzer: + """Exercise the CLI fallback without touching a real decoder.""" + + def analyze(self, _path: object) -> object: + """Fail after the buyer-controlled file name has been logged.""" + raise RuntimeError("expected test failure") + + monkeypatch.setattr(cli, "TemporalAnalyzer", FailingAnalyzer) + monkeypatch.setattr( + cli, + "run_analysis_job", + lambda _job_id, _request, requested_at: { + "jobId": "log-safety", + "state": "failed", + "requestedAt": requested_at, + "updatedAt": requested_at, + }, + ) + monkeypatch.setattr(cli.sys, "stdin", stdin) + monkeypatch.setattr(cli.sys, "stdout", stdout) + monkeypatch.setattr(cli.sys, "argv", ["cli.py"]) + caplog.set_level(logging.INFO) + + assert cli.main() == 0 + + messages = [ + record.getMessage() + for record in caplog.records + if "buyer-" in record.getMessage() + ] + assert len(messages) == 2 + for message in messages: + _assert_log_value_is_single_record(message, untrusted_value) + + +def test_cli_bounds_oversized_filename_logs( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """A buyer-controlled file name cannot create oversized routine log records.""" + oversized_name = f"buyer-{'A' * 5000}.wav" + stdin = io.StringIO( + json.dumps( + { + "jobId": "log-size-safety", + "request": { + "sourceKind": "local_audio", + "projectId": "project-log-size-safety", + "sourceLabel": "buyer.wav", + "roleFocus": [], + "localSource": { + "sourcePath": "/synthetic/buyer.wav", + "fileName": oversized_name, + "extension": "wav", + "fileSizeBytes": 1, + }, + }, + } + ) + ) + + class FailingAnalyzer: + """Exercise both CLI file-name log events without invoking a decoder.""" + + def analyze(self, _path: object) -> object: + """Fail after the first buyer-controlled file-name log event.""" + raise RuntimeError("expected test failure") + + monkeypatch.setattr(cli, "TemporalAnalyzer", FailingAnalyzer) + monkeypatch.setattr( + cli, + "run_analysis_job", + lambda _job_id, _request, requested_at: { + "jobId": "log-size-safety", + "state": "failed", + "requestedAt": requested_at, + "updatedAt": requested_at, + }, + ) + monkeypatch.setattr(cli.sys, "stdin", stdin) + monkeypatch.setattr(cli.sys, "stdout", io.StringIO()) + monkeypatch.setattr(cli.sys, "argv", ["cli.py"]) + caplog.set_level(logging.INFO) + + assert cli.main() == 0 + + messages = [ + record.getMessage() + for record in caplog.records + if "buyer-" in record.getMessage() + ] + assert len(messages) == 2 + for message in messages: + assert "" in message + assert len(message) <= 1200