preserve(security): duplicate log-forging repair pending canonical #1055

.jules/sentinel.md

@@ -28,3 +28,9 @@
 **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-20 - Prevent Log Forging / Log Injection (CWE-117) via Unsanitized Input in Python Logging
+**Vulnerability:** Found unsanitized untrusted user input (`path_str`) logged directly via f-strings (`logger.info(f"Loading and decoding audio: {path_str}")`), allowing attackers to inject newline characters (`
+`) to forge fake log entries or exploit log viewers.
+**Learning:** Python logging standard practices require using deferred string interpolation (e.g. `logger.info("msg %s", var)`) rather than f-strings to prevent interpolation performance overhead and align with log parsers. However, using `%s` alone does not escape control characters. To prevent Log Forging / CWE-117, untrusted input must be wrapped in `repr()` before passing to the logger.
+**Prevention:** When logging untrusted user input in Python, always use deferred interpolation (`%s`) AND wrap the untrusted input in `repr()` (e.g., `logger.info("msg %s", repr(untrusted_input))`) to escape control characters.

services/analysis-engine/src/bandscope_analysis/cli.py

@@ -90,7 +90,7 @@ def main() -> int:
             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.",

services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py

@@ -73,7 +73,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", repr(path_str))
 
         try:
             with path.open("rb") as fileobj:
@@ -128,7 +128,11 @@ 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 +144,5 @@ def analyze(self, audio_path: str | Path) -> TemporalFeatures:
             }
 
         except Exception as e:
-            logger.error(f"Failed to analyze audio {path_str}: {e}")
+            logger.error("Failed to analyze audio %s: %s", repr(path_str), e)
             raise ValueError(f"Temporal analysis failed: {e}") from e

services/analysis-engine/tests/test_supply_chain_policy.py

@@ -1275,9 +1275,7 @@ def test_workflow_concurrency_cancels_only_superseded_pr_heads() -> None:
         workflow = (workflows_dir / workflow_name).read_text(encoding="utf-8")
         assert "concurrency:" in workflow, workflow_name
         assert "cancel-in-progress: false" in workflow, workflow_name
-        assert "contents: read" in workflow or "permissions: read-all" in workflow, (
-            workflow_name
-        )
+        assert "contents: read" in workflow or "permissions: read-all" in workflow, workflow_name
 
     assert "pull_request:" not in (workflows_dir / "release.yml").read_text(encoding="utf-8")