Skip to content
Open
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
30 changes: 6 additions & 24 deletions ReportEngine/flask_interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
from .agent import ReportAgent, create_agent
from .nodes import ChapterJsonParseError
from .utils.config import settings
from .utils.filenames import report_export_filename


# 创建Blueprint
Expand Down Expand Up @@ -230,19 +231,6 @@ def _format_sse(event: Dict[str, Any]) -> str:
return f"id: {event_id}\nevent: {event_type}\ndata: {payload}\n\n"


def _safe_filename_segment(value: str, fallback: str = "report") -> str:
"""
生成可用于文件名的安全片段,保留字母数字与常见分隔符。

参数:
value: 原始字符串。
fallback: 兜底文本,当value为空或清洗后为空时使用。
"""
sanitized = "".join(c for c in str(value) if c.isalnum() or c in (" ", "-", "_")).strip()
sanitized = sanitized.replace(" ", "_")
return sanitized or fallback


def initialize_report_engine():
"""
初始化Report Engine。
Expand Down Expand Up @@ -1258,11 +1246,7 @@ def export_markdown(task_id: str):
# 传入 ir_file_path,修复后的图表会自动保存到 IR 文件
markdown_text = renderer.render(document_ir, ir_file_path=task.ir_file_path)

metadata = document_ir.get('metadata') if isinstance(document_ir, dict) else {}
topic = (metadata or {}).get('topic') or (metadata or {}).get('title') or (metadata or {}).get('query') or task.query
safe_topic = _safe_filename_segment(topic or 'report')
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"report_{safe_topic}_{timestamp}.md"
filename = report_export_filename(document_ir, 'md', fallback=task.query or 'report')

output_dir = Path(settings.OUTPUT_DIR)
output_dir.mkdir(parents=True, exist_ok=True)
Expand Down Expand Up @@ -1357,9 +1341,8 @@ def export_pdf(task_id: str):
# 生成PDF字节流
pdf_bytes = renderer.render_to_bytes(document_ir, optimize_layout=optimize)

# 确定下载文件名
topic = document_ir.get('metadata', {}).get('topic', 'report')
pdf_filename = f"report_{topic}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
# 确定下载文件名,清洗 topic 避免路径穿越和 Content-Disposition 注入
pdf_filename = report_export_filename(document_ir, 'pdf', fallback=task.query or 'report')

# 返回PDF文件
return Response(
Expand Down Expand Up @@ -1434,9 +1417,8 @@ def export_pdf_from_ir():
# 生成PDF字节流
pdf_bytes = renderer.render_to_bytes(document_ir, optimize_layout=optimize)

# 确定下载文件名
topic = document_ir.get('metadata', {}).get('topic', 'report')
pdf_filename = f"report_{topic}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
# 确定下载文件名,清洗 topic 避免路径穿越和 Content-Disposition 注入
pdf_filename = report_export_filename(document_ir, 'pdf')

# 返回PDF文件
return Response(
Expand Down
48 changes: 48 additions & 0 deletions ReportEngine/utils/filenames.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
"""文件名清洗工具,避免导出路径穿越和响应头注入。"""

from __future__ import annotations

from datetime import datetime
from typing import Any, Mapping, Optional


def safe_filename_segment(value: Any, fallback: str = "report") -> str:
"""
生成可用于文件名的安全片段,仅保留字母数字与常见分隔符。

参数:
value: 原始字符串。
fallback: 清洗后为空时使用的回退值。
"""
sanitized = "".join(c for c in str(value) if c.isalnum() or c in (" ", "-", "_")).strip()
sanitized = sanitized.replace(" ", "_")
return sanitized or fallback


def topic_from_document_ir(document_ir: Any, fallback: str = "report") -> str:
"""
从 Document IR 中取出用于命名的主题,缺失时回退到 fallback。
"""
metadata = document_ir.get("metadata") if isinstance(document_ir, Mapping) else None
if not isinstance(metadata, Mapping):
metadata = {}
topic = metadata.get("topic") or metadata.get("title") or metadata.get("query")
if topic is None or not str(topic).strip():
return fallback
return str(topic)


def report_export_filename(
document_ir: Any,
extension: str,
fallback: str = "report",
timestamp: Optional[str] = None,
) -> str:
"""
生成 report_<topic>_<timestamp>.<ext> 形式的安全导出文件名。
"""
topic = topic_from_document_ir(document_ir, fallback=fallback)
safe_topic = safe_filename_segment(topic, fallback=fallback)
stamp = timestamp or datetime.now().strftime("%Y%m%d_%H%M%S")
ext = extension.lstrip(".")
return f"report_{safe_topic}_{stamp}.{ext}"
14 changes: 8 additions & 6 deletions export_pdf.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
import json
import os
import sys
from datetime import datetime
from pathlib import Path

# 动态获取项目根目录,避免硬编码绝对路径
Expand Down Expand Up @@ -33,13 +32,16 @@ def export_pdf(ir_file_path):
print("正在生成PDF...")
pdf_bytes = renderer.render_to_bytes(document_ir, optimize_layout=True)

# 确定输出文件名
topic = document_ir.get('metadata', {}).get('topic', 'report')
output_dir = PROJECT_ROOT / 'final_reports' / 'pdf'
# 确定输出文件名:清洗 topic,并确保最终路径仍落在导出目录内
from ReportEngine.utils.filenames import report_export_filename

output_dir = (PROJECT_ROOT / 'final_reports' / 'pdf').resolve()
output_dir.mkdir(parents=True, exist_ok=True)

pdf_filename = f"report_{topic}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
output_path = output_dir / pdf_filename
pdf_filename = report_export_filename(document_ir, 'pdf')
output_path = (output_dir / pdf_filename).resolve()
if not output_path.is_relative_to(output_dir):
raise ValueError('Refusing to write PDF outside the export directory')

# 保存PDF文件
print(f"正在保存PDF到: {output_path}")
Expand Down
148 changes: 148 additions & 0 deletions tests/test_export_pdf_headers.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,148 @@
import importlib
import importlib.util
import json
import sys
from pathlib import Path
from types import ModuleType, SimpleNamespace

import pytest
from flask import Flask


def _remove_modules(prefix):
saved = {
name: module
for name, module in list(sys.modules.items())
if name == prefix or name.startswith(f"{prefix}.")
}
for name in saved:
sys.modules.pop(name, None)
return saved


def _load_filenames_module():
path = Path(__file__).resolve().parents[1] / "ReportEngine" / "utils" / "filenames.py"
spec = importlib.util.spec_from_file_location("ReportEngine.utils.filenames", path)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
return module


@pytest.fixture
def report_interface(monkeypatch, tmp_path):
saved_modules = _remove_modules("ReportEngine")

agent_module = ModuleType("ReportEngine.agent")
agent_module.ReportAgent = type("ReportAgent", (), {})
agent_module.create_agent = lambda *args, **kwargs: None

nodes_module = ModuleType("ReportEngine.nodes")
nodes_module.ChapterJsonParseError = type("ChapterJsonParseError", (Exception,), {})

utils_package = ModuleType("ReportEngine.utils")
utils_package.__path__ = []
config_module = ModuleType("ReportEngine.utils.config")
config_module.settings = SimpleNamespace(OUTPUT_DIR=str(tmp_path / "output"))

filenames_module = _load_filenames_module()
dep_module = ModuleType("ReportEngine.utils.dependency_check")
dep_module.check_pango_available = lambda: (True, "ok")
dep_module.log_dependency_status = lambda: None

class FakePDFRenderer:
def render_to_bytes(self, document_ir, optimize_layout=True):
return b"%PDF-fake"

renderers_module = ModuleType("ReportEngine.renderers")
renderers_module.PDFRenderer = FakePDFRenderer
renderers_module.MarkdownRenderer = type("MarkdownRenderer", (), {})

monkeypatch.setitem(sys.modules, "ReportEngine.agent", agent_module)
monkeypatch.setitem(sys.modules, "ReportEngine.nodes", nodes_module)
monkeypatch.setitem(sys.modules, "ReportEngine.utils", utils_package)
monkeypatch.setitem(sys.modules, "ReportEngine.utils.config", config_module)
monkeypatch.setitem(sys.modules, "ReportEngine.utils.filenames", filenames_module)
monkeypatch.setitem(sys.modules, "ReportEngine.utils.dependency_check", dep_module)
monkeypatch.setitem(sys.modules, "ReportEngine.renderers", renderers_module)

module = importlib.import_module("ReportEngine.flask_interface")
flask_app = Flask(__name__)
flask_app.register_blueprint(module.report_bp, url_prefix="/api/report")

try:
yield module, flask_app.test_client(), tmp_path
finally:
_remove_modules("ReportEngine")
sys.modules.update(saved_modules)


def _completed_task(module, tmp_path, document_ir, task_id="task-pdf"):
ir_path = tmp_path / f"{task_id}.json"
ir_path.write_text(json.dumps(document_ir), encoding="utf-8")
task = module.ReportTask(query="fallback-query", task_id=task_id)
task.status = "completed"
task.ir_file_path = str(ir_path)
module.current_task = task
module.tasks_registry.clear()
module.tasks_registry[task.task_id] = task
return task


def _assert_safe_pdf_disposition(response, expected_topic_segment):
assert response.status_code == 200
assert response.data == b"%PDF-fake"
disposition = response.headers.get("Content-Disposition", "")
assert "attachment" in disposition
assert expected_topic_segment in disposition
assert ".." not in disposition
assert "\r" not in disposition
assert "\n" not in disposition
assert 'filename="' in disposition
filename = disposition.split("filename=", 1)[1].strip().strip('"')
assert "/" not in filename
assert "\\" not in filename
assert '"' not in filename
assert filename.startswith(f"report_{expected_topic_segment}_")
assert filename.endswith(".pdf")


def test_export_pdf_sanitizes_content_disposition_topic(report_interface):
module, client, tmp_path = report_interface
_completed_task(
module,
tmp_path,
{"metadata": {"topic": '../../tmp/pwned\r\nSet-Cookie: a=1"; filename="evil'}},
)

response = client.get("/api/report/export/pdf/task-pdf")
_assert_safe_pdf_disposition(response, "tmppwnedSet-Cookie_a1_filenameevil")


def test_export_pdf_from_ir_sanitizes_content_disposition_topic(report_interface):
module, client, tmp_path = report_interface
response = client.post(
"/api/report/export/pdf-from-ir",
json={
"document_ir": {
"metadata": {"topic": '..\\..\\tmp\\pwned"; filename="evil'}
},
"optimize": False,
},
)
_assert_safe_pdf_disposition(response, "tmppwned_filenameevil")

def test_export_pdf_from_ir_uses_report_fallback_when_metadata_missing(report_interface):
_module, client, _tmp_path = report_interface
response = client.post(
"/api/report/export/pdf-from-ir",
json={"document_ir": {"metadata": None}},
)
_assert_safe_pdf_disposition(response, "report")


def test_export_pdf_uses_task_query_when_topic_missing(report_interface):
module, client, tmp_path = report_interface
_completed_task(module, tmp_path, {"metadata": {}}, task_id="task-pdf")
response = client.get("/api/report/export/pdf/task-pdf")
_assert_safe_pdf_disposition(response, "fallback-query")
73 changes: 65 additions & 8 deletions tests/test_export_pdf_paths.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import importlib.util
import json
import sys
from pathlib import Path
Expand All @@ -6,19 +7,27 @@
import export_pdf


def test_project_root_is_derived_from_script_location():
assert export_pdf.PROJECT_ROOT == Path(export_pdf.__file__).resolve().parent
def _install_filenames(monkeypatch, module=None):
if module is None:
path = Path(__file__).resolve().parents[1] / "ReportEngine" / "utils" / "filenames.py"
spec = importlib.util.spec_from_file_location("ReportEngine.utils.filenames", path)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)

utils_pkg = ModuleType("ReportEngine.utils")
utils_pkg.__path__ = []
utils_pkg.filenames = module
monkeypatch.setitem(sys.modules, "ReportEngine.utils", utils_pkg)
monkeypatch.setitem(sys.modules, "ReportEngine.utils.filenames", module)
return module

def test_export_pdf_writes_to_project_reports_directory(monkeypatch, tmp_path):
ir_path = tmp_path / "report.json"
ir_path.write_text(json.dumps({"metadata": {"topic": "compatibility"}}), encoding="utf-8")

def _stub_pdf_renderer(monkeypatch, payload=b"%PDF-test", filenames_module=None):
class FakeRenderer:
def render_to_bytes(self, document_ir, optimize_layout):
assert document_ir["metadata"]["topic"] == "compatibility"
assert optimize_layout is True
return b"%PDF-test"
return payload

report_engine = ModuleType("ReportEngine")
report_engine.__path__ = []
Expand All @@ -29,10 +38,58 @@ def render_to_bytes(self, document_ir, optimize_layout):
monkeypatch.setitem(sys.modules, "ReportEngine", report_engine)
monkeypatch.setitem(sys.modules, "ReportEngine.renderers", renderers)
monkeypatch.setitem(sys.modules, "ReportEngine.renderers.pdf_renderer", pdf_renderer)
_install_filenames(monkeypatch, filenames_module)


def test_project_root_is_derived_from_script_location():
assert export_pdf.PROJECT_ROOT == Path(export_pdf.__file__).resolve().parent


def test_export_pdf_writes_to_project_reports_directory(monkeypatch, tmp_path):
ir_path = tmp_path / "report.json"
ir_path.write_text(json.dumps({"metadata": {"topic": "compatibility"}}), encoding="utf-8")
_stub_pdf_renderer(monkeypatch)
monkeypatch.setattr(export_pdf, "PROJECT_ROOT", tmp_path)

result = export_pdf.export_pdf(ir_path)

output_path = Path(result)
assert output_path.parent == tmp_path / "final_reports" / "pdf"
assert output_path.parent == (tmp_path / "final_reports" / "pdf").resolve()
assert output_path.read_bytes() == b"%PDF-test"
assert output_path.name.startswith("report_compatibility_")
assert output_path.suffix == ".pdf"


def test_export_pdf_sanitizes_traversal_topic(monkeypatch, tmp_path):
ir_path = tmp_path / "report.json"
ir_path.write_text(
json.dumps({"metadata": {"topic": "../../tmp/pwned"}}),
encoding="utf-8",
)
_stub_pdf_renderer(monkeypatch)
monkeypatch.setattr(export_pdf, "PROJECT_ROOT", tmp_path)

result = export_pdf.export_pdf(ir_path)

output_path = Path(result).resolve()
export_dir = (tmp_path / "final_reports" / "pdf").resolve()
assert output_path.is_relative_to(export_dir)
assert ".." not in output_path.name
assert output_path.name.startswith("report_tmppwned_")
assert not (tmp_path.parent / "pwned").exists()


def test_export_pdf_rejects_filename_that_escapes_output_dir(monkeypatch, tmp_path):
ir_path = tmp_path / "report.json"
ir_path.write_text(json.dumps({"metadata": {"topic": "ok"}}), encoding="utf-8")

fake_filenames = ModuleType("ReportEngine.utils.filenames")
fake_filenames.report_export_filename = lambda document_ir, extension: "../outside.pdf"
_stub_pdf_renderer(monkeypatch, filenames_module=fake_filenames)
monkeypatch.setattr(export_pdf, "PROJECT_ROOT", tmp_path)

result = export_pdf.export_pdf(ir_path)

assert result is None
assert not (tmp_path / "outside.pdf").exists()
assert not list((tmp_path / "final_reports" / "pdf").glob("*.pdf"))
Loading