diff --git a/extractive/orchestrator.py b/extractive/orchestrator.py index 9badc78..946979f 100644 --- a/extractive/orchestrator.py +++ b/extractive/orchestrator.py @@ -24,6 +24,7 @@ map_sentences_to_pages, sumy_language, ) +from shared.path_safety import resolve_source_path from shared.schemas.atomic_note_extractive import AtomicNoteExtractive EXTRACTIVE_VERSION = "extractive-v0.2.0" @@ -71,9 +72,10 @@ def main(): ap.add_argument("--dry-run", action="store_true", help="Keine Dateien schreiben, nur Eval") args = ap.parse_args() - source = Path(args.source) - if not source.exists(): - sys.exit(f"Fehler: PDF nicht gefunden: {source}") + try: + source = resolve_source_path(args.source) + except FileNotFoundError as exc: + sys.exit(f"Fehler: PDF nicht gefunden: {exc}") out_dir = Path(args.out_dir) out_dir.mkdir(parents=True, exist_ok=True) diff --git a/generative/eval_chunk_recall.py b/generative/eval_chunk_recall.py index 3c64465..35b3a50 100644 --- a/generative/eval_chunk_recall.py +++ b/generative/eval_chunk_recall.py @@ -188,6 +188,7 @@ def main(argv: list[str] | None = None) -> int: from generative.config import CHUNK_WORDS from generative.pipeline.pdf_chunker import pdf_to_text + from shared.path_safety import resolve_source_path p = argparse.ArgumentParser(description="Phase-A Chunk-/Planner-Recall-Messung (LLM-frei)") p.add_argument("pdf", type=Path, help="Pfad zum Quell-PDF") @@ -209,8 +210,10 @@ def main(argv: list[str] | None = None) -> int: p.add_argument("--overlaps", type=int, nargs="+", default=[0, 50, 100, 200], help="Zu testende Overlap-Wortzahlen.") args = p.parse_args(argv) - if not args.pdf.exists(): - print(f"PDF nicht gefunden: {args.pdf}", file=sys.stderr) + try: + args.pdf = resolve_source_path(args.pdf) + except FileNotFoundError as exc: + print(f"PDF nicht gefunden: {exc}", file=sys.stderr) return 1 text = pdf_to_text(args.pdf) diff --git a/generative/gui/app.py b/generative/gui/app.py index 563a46f..78acedf 100644 --- a/generative/gui/app.py +++ b/generative/gui/app.py @@ -28,6 +28,7 @@ from generative.gui import env_file, gui_settings, run_history, runner from generative.pipeline.export_runner import EXPORT_FILE_SUFFIXES +from shared.path_safety import resolve_source_path logger = logging.getLogger(__name__) @@ -780,8 +781,20 @@ async def start_run(request: Request) -> JSONResponse: if options.get("export_formats"): export_formats_dir = (exports_dir / run_history.make_run_id(clock())).resolve() options = {**options, "export_formats_dir": str(export_formats_dir)} - if not pdf or not Path(pdf).exists(): + if not pdf: return JSONResponse({"error": f"PDF nicht gefunden: {pdf}"}, status_code=400) + # #186-Nachbesserung (Cross-Model-Review, Punkt 3): der Root-Check muss + # VOR dem Apostroph-Glob-Fallback laufen -- sonst wuerde resolve_source_path() + # Verzeichnisse ausserhalb der erlaubten Roots durchsuchen, bevor der + # eigentliche #2-Check (unten) das ueberhaupt ablehnen kann (Defense-in- + # Depth-Aufweichung; der Fallback sucht ohnehin nur im selben Verzeichnis + # wie der Roh-Pfad, dessen Erlaubtheit hier schon geprueft wird). + if not any(_is_within(str(Path(pdf).parent), root) for root in _allowed_roots): + return JSONResponse({"error": "PDF liegt ausserhalb der erlaubten Verzeichnisse."}, status_code=400) + try: + pdf = str(resolve_source_path(pdf)) + except FileNotFoundError as exc: + return JSONResponse({"error": f"PDF nicht gefunden: {exc}"}, status_code=400) # #2: Quelle muss unter einem erlaubten Root liegen (gelistet/hochgeladen). if not any(_is_within(pdf, root) for root in _allowed_roots): return JSONResponse({"error": "PDF liegt ausserhalb der erlaubten Verzeichnisse."}, status_code=400) diff --git a/generative/gui/tests/test_app.py b/generative/gui/tests/test_app.py index e7db06f..2dddd7c 100644 --- a/generative/gui/tests/test_app.py +++ b/generative/gui/tests/test_app.py @@ -577,6 +577,40 @@ def test_run_rejects_pdf_outside_allowed_dirs(tmp_path): assert r.status_code == 400 +def test_run_glob_fallback_not_attempted_outside_allowed_dirs(tmp_path, monkeypatch): + # #186-Nachbesserung (Cross-Model-Review, Punkt 3): resolve_source_path() + # lief bisher VOR dem _allowed_roots-Check -- der Apostroph-Glob-Fallback + # haette so Verzeichnisse ausserhalb der erlaubten Roots durchsuchen koennen + # (Defense-in-Depth-Aufweichung). Der Root-Check muss zuerst laufen, sodass + # resolve_source_path fuer nicht-erlaubte Verzeichnisse gar nicht erst + # aufgerufen wird. + import generative.gui.app as app_module + + allowed = tmp_path / "pdfs" + allowed.mkdir() + outside = tmp_path / "aussen" + outside.mkdir() + (outside / "Porst’s-Buch.pdf").write_bytes(b"%PDF-1.4") + queried = outside / "Porst's-Buch.pdf" # existiert nicht exakt, nur die Apostroph-Variante + + def boom(*_args, **_kwargs): + raise AssertionError("resolve_source_path lief fuer ein nicht-erlaubtes Verzeichnis -- Reihenfolge falsch") + + monkeypatch.setattr(app_module, "resolve_source_path", boom) + + app = app_module.create_app( + run_factory=fake_run, + pdf_dirs=[allowed], + vault_path=tmp_path, + backend="subscription", + uploads_dir=tmp_path / "uploads", + doctor_fn=fake_doctor, + ) + r = TestClient(app, base_url="http://localhost").post("/api/run", json={"pdf": str(queried), "dry_run": True}) + assert r.status_code == 400 + assert "ausserhalb der erlaubten Verzeichnisse" in r.json()["error"] + + def test_run_accepts_pdf_from_uploads_dir(tmp_path): # Hochgeladene PDFs (in uploads_dir) bleiben gültige Lauf-Quellen. uploads = tmp_path / "uploads" diff --git a/generative/orchestrator.py b/generative/orchestrator.py index c48367e..f6c0623 100644 --- a/generative/orchestrator.py +++ b/generative/orchestrator.py @@ -91,6 +91,7 @@ def _reconfigure_streams_utf8() -> None: from generative.pipeline.page_index import build_page_index from generative.schemas.atomic_note import AtomicNoteDraft, ConceptPlan from generative.schemas.citation import CitationMeta, build_citation_meta, crossref_override_blocked +from shared.path_safety import resolve_source_path from generative.config import ( AGENT_VERSION, CRITIC_AUTO_THRESHOLD, @@ -2129,9 +2130,14 @@ def main(argv: list[str] | None = None): print(f"\n=== Atomic Agent (load-drafts): {source_path.name} ===\n") print(f" [load-drafts] {len(drafts)} Drafts geladen · Stage 1–5 übersprungen") else: - source_path = Path(args.source) - if not source_path.exists(): - sys.exit(f"Datei nicht gefunden: {source_path}") + # #186-Nachbesserung: derselbe Apostroph-/Anfuehrungszeichen-Glob-Fallback + # wie extractive/orchestrator.py und eval_chunk_recall.py -- vorher brach + # dieser Haupt-CLI-Pfad mit einem nackten sys.exit bei reinen Apostroph- + # Varianten ab. + try: + source_path = resolve_source_path(args.source) + except FileNotFoundError as exc: + sys.exit(f"Datei nicht gefunden: {exc}") print(f"\n=== Atomic Agent: {source_path.name} ===\n") ( drafts, diff --git a/generative/tests/test_orchestrator_source_resolution.py b/generative/tests/test_orchestrator_source_resolution.py new file mode 100644 index 0000000..c90738d --- /dev/null +++ b/generative/tests/test_orchestrator_source_resolution.py @@ -0,0 +1,62 @@ +"""Tests fuer die Quell-Pfad-Aufloesung im CLI-Hauptpfad von `orchestrator.main` +(#186-Nachbesserung, Cross-Model-Review). + +Der `--load-drafts`-Zweig und die extractive-/eval_chunk_recall-Einstiegspunkte +nutzen bereits `shared.path_safety.resolve_source_path` (Apostroph-/ +Anfuehrungszeichen-Glob-Fallback). Der normale `--source`-Pfad in +`orchestrator.main` wurde im Review uebersehen und brach mit einem nackten +`sys.exit` bei reinen Apostroph-Varianten ab. Kein voller Pipeline-Lauf hier -- +`_run_extraction_stages` wird als Seam gestubbt, analog zu +test_maintainer_optin.py/test_orchestrator_export.py. +""" + +from __future__ import annotations + +import pytest + +from generative import orchestrator + + +@pytest.fixture(autouse=True) +def _kein_globaler_llm_state(monkeypatch): + """orchestrator.main ruft set_llm_runtime_config VOR der Quell-Aufloesung -- + das setzt sonst das globale _LLM_RUNTIME_SETTINGS und leakt Backend-Kwargs + (call_timeout_sec) in spaeter laufende Tests (test_phoenix_span). Hier als + No-op patchen; Phoenix-Tracing analog zu test_orchestrator_export.py stummschalten.""" + from generative.agents import base as agents_base + + monkeypatch.setattr(agents_base, "set_llm_runtime_config", lambda _cfg: None) + monkeypatch.setattr(orchestrator, "_setup_phoenix_tracing", lambda: None) + + +def test_missing_source_exits_with_datei_nicht_gefunden(monkeypatch, tmp_path): + monkeypatch.setenv("ATOMIC_AGENT_GUI", "1") + missing = tmp_path / "nicht-vorhanden.pdf" + with pytest.raises(SystemExit) as exc: + orchestrator.main(["--source", str(missing)]) + msg = str(exc.value) + assert "Datei nicht gefunden" in msg + assert "nicht-vorhanden.pdf" in msg + + +def test_source_apostrophe_variant_resolved_via_glob_fallback(monkeypatch, tmp_path): + # Datei liegt mit typografischem Apostroph (U+2019) vor, --source wird mit + # dem geraden ' aufgerufen -- muss trotzdem gefunden werden (wie extractive/ + # eval_chunk_recall bereits per resolve_source_path). + monkeypatch.setenv("ATOMIC_AGENT_GUI", "1") + real = tmp_path / "Porst’s-Buch.pdf" + real.write_bytes(b"%PDF-1.4") + queried = tmp_path / "Porst's-Buch.pdf" + + captured = {} + + def stop_here(_args, source_path, _runtime_config): + captured["source_path"] = source_path + raise RuntimeError("stop-here") + + monkeypatch.setattr(orchestrator, "_run_extraction_stages", stop_here) + + with pytest.raises(RuntimeError, match="stop-here"): + orchestrator.main(["--source", str(queried)]) + + assert captured["source_path"] == real diff --git a/shared/path_safety.py b/shared/path_safety.py index 71fa62b..1124a59 100644 --- a/shared/path_safety.py +++ b/shared/path_safety.py @@ -1,7 +1,9 @@ from __future__ import annotations +import glob import os import re +import sys from pathlib import Path _UNSAFE_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]+') @@ -15,6 +17,11 @@ *(f"LPT{i}" for i in range(1, 10)), } +# Apostroph-/Anfuehrungszeichen-Varianten, die beim Kopieren aus PDFs/Web +# unbemerkt gegeneinander vertauscht werden (#186): gerades Apostroph, rechtes/ +# linkes typografisches Apostroph, Gravis, Akut. +_QUOTE_LIKE_CHARS = "'’‘`´" + def safe_filename_stem(value: str, *, max_len: int = 60, fallback: str = "note") -> str: stem = str(value or "").lower().replace(" ", "-") @@ -28,6 +35,65 @@ def safe_filename_stem(value: str, *, max_len: int = 60, fallback: str = "note") return stem +def resolve_source_path(source: str | Path) -> Path: + """Loest einen Quell-Dateipfad (z. B. PDF) auf, inkl. Glob-Fallback fuer + Apostroph-/Anfuehrungszeichen-Varianten (#186). + + Existiert `source` exakt, wird er unveraendert zurueckgegeben. Sonst wird + im selben Verzeichnis nach Dateien gesucht, deren Name bis auf gerade vs. + typografische Apostrophe/Anfuehrungszeichen (' ’ ‘ ` ´) + identisch ist. Genau ein Treffer wird verwendet (Hinweis auf stderr); + bei 0 oder >1 Treffern wirft die Funktion FileNotFoundError mit + `repr(str(source))` und -- falls mehrdeutig -- den Kandidaten, damit alle + Aufrufer dieselbe, gut lesbare Fehlermeldung erhalten (statt sie je Stelle + zu duplizieren). + """ + path = Path(source) + if path.exists(): + return path + + pattern = glob.escape(path.name) + for ch in _QUOTE_LIKE_CHARS: + pattern = pattern.replace(ch, "?") + + # Nachbesserung (#186 Review): `?` im Glob matcht JEDES Zeichen, nicht nur + # Quote-Varianten -- ohne den Re-Filter unten wuerden z.B. "PorstXs-Buch.pdf", + # "Porst_s-Buch.pdf" oder "Porst5s-Buch.pdf" faelschlich als Apostroph- + # Variante von "Porst's-Buch.pdf" durchgehen. Deshalb: nach dem Glob streng + # nachfiltern -- Name muss nach Normalisierung aller Quote-Zeichen exakt dem + # normalisierten Query-Namen entsprechen, plus is_file() (keine Verzeichnisse). + normalized_query = _normalize_quote_chars(path.name) + candidates = ( + sorted( + c for c in path.parent.glob(pattern) if c.is_file() and _normalize_quote_chars(c.name) == normalized_query + ) + if path.parent.is_dir() + else [] + ) + + if len(candidates) == 1: + print( + f"Hinweis: {repr(str(path))} nicht gefunden, verwende Fallback-Treffer " + f"(Apostroph-/Anfuehrungszeichen-Variante): {candidates[0]}", + file=sys.stderr, + ) + return candidates[0] + + if candidates: + listing = ", ".join(str(c) for c in candidates) + raise FileNotFoundError(f"{repr(str(path))} nicht eindeutig -- mehrere Kandidaten gefunden: {listing}") + + raise FileNotFoundError(repr(str(path))) + + +def _normalize_quote_chars(name: str) -> str: + """Bildet alle Apostroph-/Anfuehrungszeichen-Varianten auf ein einheitliches + Zeichen ab, damit zwei Namen die sich nur darin unterscheiden vergleichbar sind.""" + for ch in _QUOTE_LIKE_CHARS: + name = name.replace(ch, "'") + return name + + def contained_child_path(parent: Path, filename: str) -> Path: candidate = parent / filename base = os.path.abspath(os.path.normpath(os.fspath(parent))) diff --git a/shared/tests/test_path_safety.py b/shared/tests/test_path_safety.py index e4456be..b72a0db 100644 --- a/shared/tests/test_path_safety.py +++ b/shared/tests/test_path_safety.py @@ -15,7 +15,7 @@ import pytest -from shared.path_safety import contained_child_path, safe_filename_stem +from shared.path_safety import contained_child_path, resolve_source_path, safe_filename_stem # Unsichere Titel aus dem alten write_note-Test: Traversal, absolute Pfade, # Windows-Sondernamen, NUL-Byte. Der Slug muss jeden davon im Zielordner halten. @@ -94,3 +94,85 @@ def test_contained_child_path_rejects_absolute_sibling(): target = os.path.join(outside, "evil.md") with pytest.raises(ValueError): contained_child_path(Path(inside), target) + + +# -- resolve_source_path: Apostroph-/Anfuehrungszeichen-Varianten (#186) -------- + + +def test_resolve_source_path_returns_exact_match_unchanged(tmp_path: Path): + f = tmp_path / "plain.pdf" + f.write_text("x") + assert resolve_source_path(f) == f + + +def test_resolve_source_path_finds_curly_apostrophe_file_via_straight_query(tmp_path: Path): + """Datei liegt mit typografischem Apostroph (U+2019) vor, Nutzer tippt den geraden ' -- muss trotzdem gefunden werden.""" + f = tmp_path / "Porst’s-Buch.pdf" + f.write_text("x") + queried = tmp_path / "Porst's-Buch.pdf" + assert resolve_source_path(queried) == f + + +def test_resolve_source_path_finds_straight_apostrophe_file_via_curly_query(tmp_path: Path): + """Umgekehrter Fall: Datei mit geradem ', Anfrage mit typografischem U+2019.""" + f = tmp_path / "Porst's-Buch.pdf" + f.write_text("x") + queried = tmp_path / "Porst’s-Buch.pdf" + assert resolve_source_path(queried) == f + + +def test_resolve_source_path_raises_with_repr_when_no_candidate(tmp_path: Path): + missing = tmp_path / "nichts-hier.pdf" + with pytest.raises(FileNotFoundError) as exc_info: + resolve_source_path(missing) + assert repr(str(missing)) in str(exc_info.value) + + +def test_resolve_source_path_raises_and_lists_candidates_when_ambiguous(tmp_path: Path): + a = tmp_path / "Porst’s-Buch.pdf" + b = tmp_path / "Porst‘s-Buch.pdf" + a.write_text("x") + b.write_text("x") + queried = tmp_path / "Porst's-Buch.pdf" + with pytest.raises(FileNotFoundError) as exc_info: + resolve_source_path(queried) + msg = str(exc_info.value) + assert repr(str(queried)) in msg + assert str(a) in msg + assert str(b) in msg + + +def test_resolve_source_path_no_fallback_without_quote_chars(tmp_path: Path): + """Fehlt ein Apostroph-Zeichen ganz (Tippfehler ohne Zweifelszeichen), soll kein Glob-Fallback greifen.""" + missing = tmp_path / "voellig-anderer-name.pdf" + (tmp_path / "anderes-dokument.pdf").write_text("x") + with pytest.raises(FileNotFoundError): + resolve_source_path(missing) + + +# -- Nachbesserung (#186 Review): das `?`-Glob matcht JEDES Zeichen, nicht nur +# Quote-Varianten -- ohne strikten Re-Filter wuerden beliebige Ein-Zeichen- +# Abweichungen (Buchstabe/Ziffer/Unterstrich) faelschlich als "Apostroph- +# Variante" durchgehen. ----------------------------------------------------- + + +@pytest.mark.parametrize( + "wrong_name", + ["PorstXs-Buch.pdf", "Porst_s-Buch.pdf", "Porst5s-Buch.pdf"], +) +def test_resolve_source_path_rejects_non_quote_char_glob_matches(tmp_path: Path, wrong_name: str): + """`?` im Glob-Pattern matcht auch Nicht-Quote-Zeichen -- das darf NICHT als + Apostroph-Variante durchgehen (Fehlmatch aus Cross-Model-Review, 3x konvergent).""" + (tmp_path / wrong_name).write_text("x") + queried = tmp_path / "Porst's-Buch.pdf" + with pytest.raises(FileNotFoundError): + resolve_source_path(queried) + + +def test_resolve_source_path_ignores_directory_candidates(tmp_path: Path): + """Ein gleichnamiges VERZEICHNIS (Apostroph-Variante im Namen) darf nie als + Datei-Fallback zurueckgegeben werden.""" + (tmp_path / "Porst’s-Buch.pdf").mkdir() + queried = tmp_path / "Porst's-Buch.pdf" + with pytest.raises(FileNotFoundError): + resolve_source_path(queried)