Harden local audit boundaries

ROADMAP.md

@@ -1,6 +1,6 @@
 # WhisperForge Roadmap
 
-Last reviewed: 2026-07-01
+Last reviewed: 2026-07-03
 
 The May 2026 reset wave is now complete. The immediate next-round plan is
 [`docs/NEXT-ROUND-PLAN-2026-05-19.md`](docs/NEXT-ROUND-PLAN-2026-05-19.md); current
@@ -20,7 +20,9 @@ WhisperForge is now a single-user voice-to-knowledge workbench:
 - local run workspace with manifests, stage artifacts, reopen, and downstream
   export retry;
 - local report-only resurfacing digest;
-- text-first SongForge creative pack mode.
+- text-first SongForge creative pack mode;
+- audit hardening for knowledge-base writes, audio upload spooling, and trusted
+  local cache reads.
 
 ## Delivery State
 
@@ -32,7 +34,10 @@ WhisperForge is now a single-user voice-to-knowledge workbench:
 - Latest ops baseline: lightweight company OS manifest merged in PR `#57`.
 - Open GitHub issues: none as of 2026-05-30 (`#49` through `#52` are shipped).
 - Open GitHub PRs: none.
-- Current unit baseline: `302 passed`.
+- Current unit baseline: `324 passed`.
+- Current audit-hardening branch: `codex/audit-hardening-cleanup` covers KB
+  write path constraints, temp-file upload transcription, services upload copy
+  hygiene, and pickle cache trust guards.
 - Next round plan:
   [`docs/NEXT-ROUND-PLAN-2026-05-19.md`](docs/NEXT-ROUND-PLAN-2026-05-19.md).
 - Release target decision (owner): local-first personal workbench, recorded in

STATUS.md

@@ -1,6 +1,6 @@
 # WhisperForge Status
 
-Last updated: 2026-07-01
+Last updated: 2026-07-03
 
 ## Current State
 
@@ -12,6 +12,9 @@ Last updated: 2026-07-01
 - Latest feature baseline: the merged local-first reliability swarm for KB
   governance, transcription-router media planning, approved digest routing, and
   SongForge export polish on current `main`.
+- Current audit-hardening branch: `codex/audit-hardening-cleanup` adds
+  knowledge-base write path constraints, temp-file audio upload transcription,
+  services upload copy hygiene, and pickle cache trust guards.
 - GitHub repo: `WalksWithASwagger/spektorAI`.
 - GitHub metadata: description, topics, and homepage point to this canonical
   WhisperForge repo.
@@ -37,6 +40,22 @@ Last updated: 2026-07-01
 
 ## Verified Baseline
 
+- Verified on 2026-07-03 during audit-hardening closeout.
+- `git rebase origin/main` completed successfully on
+  `codex/audit-hardening-cleanup`.
+- Ignored Python/test cache cruft was removed earlier in this lane:
+  `__pycache__/`, `.pytest_cache/`, `.ruff_cache/`, `.mypy_cache/`, and
+  `.DS_Store` files. `.env`, `.cache/`, `venv/`, and `whisperforge-env/` were
+  intentionally preserved.
+- `git diff --check` passes.
+- `make docs-check` passes documentation link, command reference, and freshness
+  checks.
+- `make lint` passes Python syntax and high-signal Ruff checks.
+- `make test` -> `324 passed, 2 warnings`.
+- `make pip-check` -> `No broken requirements found`.
+- `make eval-fixture` -> editorial and SongForge fixtures pass.
+- `make smoke` passes Streamlit health smoke on the default smoke port.
+
 - Verified on 2026-05-30 during shutdown closeout.
 - `git fetch origin --prune` completed successfully.
 - `git status --short --branch` -> clean `main...origin/main` before docs

services/transcription/service.py

@@ -8,6 +8,7 @@
 """
 
 import os
+import shutil
 import tempfile
 
 from fastapi import Depends, FastAPI, File, HTTPException, UploadFile
@@ -39,10 +40,7 @@ async def transcribe(
             detail=f"Invalid file format. Allowed: {sorted(ALLOWED_EXTENSIONS)}",
         )
 
-    # Spill to a temp file so detailed transcription can size-check + chunk.
-    with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp:
-        tmp.write(await file.read())
-        tmp_path = tmp.name
+    tmp_path = _copy_upload_to_temp(file, ext)
 
     try:
         details = audio.transcribe_audio_detailed(tmp_path)
@@ -60,3 +58,18 @@ async def transcribe(
             os.unlink(tmp_path)
         except OSError:
             pass
+
+
+def _copy_upload_to_temp(file: UploadFile, suffix: str) -> str:
+    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
+        tmp_path = tmp.name
+        try:
+            file.file.seek(0)
+            shutil.copyfileobj(file.file, tmp, length=1 << 20)
+        except Exception:
+            try:
+                os.unlink(tmp_path)
+            except OSError:
+                pass
+            raise
+    return tmp_path

tests/test_cache.py

@@ -5,6 +5,9 @@
 return the stored value without re-running compute.
 """
 
+import os
+import pickle
+import stat
 from unittest.mock import MagicMock
 
 import pytest
@@ -107,3 +110,39 @@ def test_clear_removes_all_entries(self, cache_on):
             cache.put(cache.make_key([str(i)]), f"value-{i}")
         assert cache.clear() == 3
         assert cache.clear() == 0
+
+
+class TestTrustedPickleLoading:
+    def test_loads_pickle_under_private_cache_root(self, tmp_path):
+        path = tmp_path / "value.pkl"
+        path.write_bytes(pickle.dumps({"ok": True}))
+
+        assert cache.load_pickle(path, root=tmp_path, label="test") == {"ok": True}
+
+    def test_rejects_pickle_outside_cache_root(self, tmp_path):
+        root = tmp_path / "cache"
+        root.mkdir()
+        outside = tmp_path / "outside.pkl"
+        outside.write_bytes(pickle.dumps("nope"))
+
+        assert cache.load_pickle(outside, root=root, label="outside") is None
+
+    def test_rejects_pickle_symlink(self, tmp_path):
+        if not hasattr(os, "symlink"):
+            pytest.skip("symlink unavailable")
+        target = tmp_path / "target.pkl"
+        target.write_bytes(pickle.dumps("secret"))
+        link = tmp_path / "link.pkl"
+        os.symlink(target, link)
+
+        assert cache.load_pickle(link, root=tmp_path, label="symlink") is None
+
+    def test_rejects_group_or_world_writable_cache_path(self, tmp_path):
+        path = tmp_path / "value.pkl"
+        path.write_bytes(pickle.dumps("secret"))
+        original = tmp_path.stat().st_mode
+        try:
+            tmp_path.chmod(original | stat.S_IWGRP)
+            assert cache.load_pickle(path, root=tmp_path, label="shared") is None
+        finally:
+            tmp_path.chmod(original)

tests/test_http_adapters.py

@@ -37,6 +37,30 @@ def fake_post(*args, **kwargs):
     assert calls[0]["kwargs"]["files"]["file"][0] == "upload.wav"
 
 
+def test_http_transcribe_path_posts_file_object_without_prereading(monkeypatch, tmp_path):
+    path = tmp_path / "clip.wav"
+    path.write_bytes(b"fake-audio")
+    inspected = {}
+
+    def fake_post(*args, **kwargs):
+        file_name, file_obj = kwargs["files"]["file"]
+        inspected["file_name"] = file_name
+        inspected["is_bytes"] = isinstance(file_obj, bytes)
+        inspected["body"] = file_obj.read()
+        return FakeResponse({"text": "Transcript body", "segments": [], "language": None})
+
+    monkeypatch.setattr(http_adapters.requests, "post", fake_post)
+
+    details = http_adapters.HttpTranscriber().transcribe_detailed(path, suffix=".wav")
+
+    assert details.text == "Transcript body"
+    assert inspected == {
+        "file_name": "clip.wav",
+        "is_bytes": False,
+        "body": b"fake-audio",
+    }
+
+
 def test_http_generate_forwards_modern_options(monkeypatch):
     calls = []
 

tests/test_primary_loop_flow.py

@@ -1,9 +1,12 @@
 """Integration coverage for the primary local run loop."""
 
+import io
+from pathlib import Path
 from types import SimpleNamespace
 
 from ui import dialogs, input as input_mod, output as output_mod, pipeline as pipeline_mod
 from whisperforge_core import adapters as adapters_mod
+from whisperforge_core import audio
 from whisperforge_core import captures
 from whisperforge_core import pipeline as core_pipeline
 from whisperforge_core import run_artifacts
@@ -155,6 +158,79 @@ def fake_save_to_notion():
     assert state["last_notion_url"] == "https://notion.so/primary-loop"
 
 
+def test_audio_upload_spools_to_temp_path_without_getvalue(tmp_path, monkeypatch):
+    monkeypatch.setattr(run_artifacts, "RUNS_DIR", tmp_path / "runs")
+
+    class FakeUpload:
+        name = "clip.wav"
+
+        def __init__(self):
+            self._body = io.BytesIO(b"fake-audio")
+
+        def seek(self, *args):
+            return self._body.seek(*args)
+
+        def read(self, size=-1):
+            return self._body.read(size)
+
+        def getvalue(self):
+            raise AssertionError("getvalue should not be used for audio spooling")
+
+    state = FakeSessionState({
+        "pending_input": input_mod.PendingInput(
+            source="upload",
+            payload=FakeUpload(),
+            filename="clip.wav",
+        ),
+        "_submit_mode": "transcribe_only",
+        "pipeline_running": True,
+        "pipeline_stage_idx": 0,
+        "selected_user": None,
+        "ai_provider": "Anthropic",
+        "ai_model": "claude-haiku-4-5",
+        "cleanup_enabled": True,
+        "chapters_enabled": True,
+        "agentic_drafting": False,
+        "fact_check_enabled": False,
+        "images_enabled": False,
+        "rag_mode": "auto",
+        "compare_provider": None,
+        "compare_model": None,
+        "selected_personas": [],
+        "capture_id": None,
+    })
+
+    fake_st = SimpleNamespace(
+        session_state=state,
+        status=lambda *_args, **_kwargs: FakeStatus(),
+        toast=lambda *_args, **_kwargs: None,
+    )
+    monkeypatch.setattr(pipeline_mod, "st", fake_st)
+    monkeypatch.setattr(pipeline_mod, "_ensure_capture", lambda _pending, _run_id: None)
+    seen = {}
+
+    def fake_transcribe_detailed(source, suffix=".mp3"):
+        path = Path(source)
+        seen["path"] = path
+        seen["suffix"] = suffix
+        assert path.exists()
+        assert path.read_bytes() == b"fake-audio"
+        return audio.TranscriptionDetails(text="Transcript body")
+
+    fake_adapters = adapters_mod.Adapters(
+        transcriber=SimpleNamespace(transcribe_detailed=fake_transcribe_detailed),
+        processor=SimpleNamespace(),
+        storage=SimpleNamespace(),
+    )
+    monkeypatch.setattr(pipeline_mod.adapters_mod, "get_adapters", lambda: fake_adapters)
+
+    pipeline_mod._execute_run()
+
+    assert state["transcription"] == "Transcript body"
+    assert seen["suffix"] == ".wav"
+    assert not seen["path"].exists()
+
+
 def test_mark_capture_status_refreshes_run_manifest_capture_metadata(tmp_path, monkeypatch):
     monkeypatch.setattr(run_artifacts, "RUNS_DIR", tmp_path / "runs")
     monkeypatch.setattr(captures, "CAPTURES_DIR", tmp_path / "captures")

tests/test_prompts.py

@@ -78,6 +78,40 @@ def test_generation_warning_summarizes_unresolved_governance(self, tmp_prompts_d
         assert "private-token.md" in warning
 
 
+class TestKnowledgeBaseWriteTarget:
+    @pytest.mark.parametrize(
+        ("name", "expected"),
+        [
+            ("voice", "voice.md"),
+            ("voice guide", "voice guide.md"),
+            ("voice.md", "voice.md"),
+            ("notes.txt", "notes.md"),
+            ("release.v1", "release.v1.md"),
+        ],
+    )
+    def test_normalizes_safe_names(self, tmp_prompts_dir, name, expected):
+        target = prompts.knowledge_base_write_target("alice", name)
+
+        assert target == tmp_prompts_dir / "alice" / "knowledge_base" / expected
+
+    @pytest.mark.parametrize(
+        "name",
+        ["", "   ", "../evil", "folder/evil", "folder\\evil", ".hidden", "..", "bad:name"],
+    )
+    def test_rejects_unsafe_names(self, tmp_prompts_dir, name):
+        with pytest.raises(ValueError):
+            prompts.knowledge_base_write_target("alice", name)
+
+    def test_write_target_preserves_replace_behavior(self, tmp_prompts_dir):
+        target = prompts.knowledge_base_write_target("alice", "voice")
+        target.parent.mkdir(parents=True)
+
+        target.write_text("first")
+        prompts.knowledge_base_write_target("alice", "voice.md").write_text("second")
+
+        assert target.read_text() == "second"
+
+
 class TestPromptPrecedence:
     def test_custom_prompt_overrides_md(self, tmp_prompts_dir):
         user = tmp_prompts_dir / "alice"

tests/test_rag_store.py

@@ -6,6 +6,7 @@
 """
 
 import hashlib
+import os
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -99,6 +100,26 @@ def test_second_load_uses_disk_index(self, tmp_path):
             assert build_spy.call_count == 0
         assert s2.chunk_count() == s1.chunk_count()
 
+    def test_untrusted_chunks_pickle_triggers_rebuild(self, tmp_path):
+        if not hasattr(os, "symlink"):
+            pytest.skip("symlink unavailable")
+        _seed_kb(tmp_path / "prompts", "alice", {
+            "voice.md": "# Voice\n" + " ".join(f"word{i}" for i in range(200)),
+        })
+        s1 = KBStore("alice")
+        s1.ensure_built()
+        chunks_path = s1.dir / "chunks.pkl"
+        target = tmp_path / "outside.pkl"
+        target.write_bytes(b"not trusted")
+        chunks_path.unlink()
+        os.symlink(target, chunks_path)
+
+        s2 = KBStore("alice")
+        with patch.object(s2, "_build") as build_spy:
+            s2.ensure_built()
+
+        assert build_spy.called
+
 
 class TestInvalidation:
     def test_kb_edit_triggers_rebuild(self, tmp_path):

tests/test_services_contract.py

@@ -1,3 +1,7 @@
+import io
+import os
+from types import SimpleNamespace
+
 from fastapi.testclient import TestClient
 
 from services.processing import service as processing_service
@@ -39,6 +43,19 @@ def fake_transcribe_audio_detailed(_path):
     }
 
 
+def test_transcription_service_copy_upload_uses_file_stream_not_async_read():
+    async def fail_read():
+        raise AssertionError("UploadFile.read should not be used for spooling")
+
+    upload = SimpleNamespace(file=io.BytesIO(b"fake-audio"), read=fail_read)
+    path = transcription_service._copy_upload_to_temp(upload, ".wav")
+    try:
+        with open(path, "rb") as handle:
+            assert handle.read() == b"fake-audio"
+    finally:
+        os.unlink(path)
+
+
 def test_processing_pipeline_accepts_modern_options_and_returns_full_result(monkeypatch):
     captured = {}