Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions extractive/orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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)

Expand Down
7 changes: 5 additions & 2 deletions generative/eval_chunk_recall.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand All @@ -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)
Expand Down
15 changes: 14 additions & 1 deletion generative/gui/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)

Expand Down Expand Up @@ -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)
Expand Down
34 changes: 34 additions & 0 deletions generative/gui/tests/test_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
12 changes: 9 additions & 3 deletions generative/orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
62 changes: 62 additions & 0 deletions generative/tests/test_orchestrator_source_resolution.py
Original file line number Diff line number Diff line change
@@ -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
66 changes: 66 additions & 0 deletions shared/path_safety.py
Original file line number Diff line number Diff line change
@@ -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]+')
Expand All @@ -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(" ", "-")
Expand All @@ -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)))
Expand Down
84 changes: 83 additions & 1 deletion shared/tests/test_path_safety.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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)
Loading