epub2pdf/tests/test_review_findings_20260903.py

"""Regression tests for the 2026-09-03 adversarial-review findings.

Source of truth:
  .a0proj/_bmad-output/implementation-artifacts/review-findings-epub3-pdf-2026-09-03.md

C2 - Remote resource fetching by WeasyPrint (blind SSRF / server-side data egress).
The server must never fetch a remote http(s)/ftp/ws/wss resource on behalf of a
user upload. The fix lives in converters.common._blocking_url_fetcher
(common.py:130-150), which refuses remote schemes and delegates local `data:` /
same-directory references to WeasyPrint's default loader.

C1 - Decompression-bomb guard on zip extraction (app.py zip-bomb guard + the
SHARED streaming guard common._copy_stream_with_cap).

C3 - Stored-XSS stripping in rendered HTML/SVG (common._strip_scripts in
common.html_to_pdf_bytes, called by the pandoc/html/svg converters).

H1 - CSV row/column render cap in converters.csv_converter._table_html (rows
capped to MAX_ROWS, columns to MAX_COLS, visible truncation note).

M2 - Stale output-file sweep (app._sweep_stale_output_files).

Satisfies: C1, C2, C3, H1, M2
"""
import base64
import io
import os
import tempfile
import time
import zipfile

import pytest

import app
import converters.common as C
from tests.helpers import wait_for_job, download_bytes, pdf_text


def _upload(client, content, filename="doc.html"):
    """POST an upload, wait for the terminal job state, return the snapshot.

    Accepts either a str (encoded to utf-8) or bytes so the HTML/SVG string
    tests and the binary-epub C1 zip-bomb test can both share it.
    """
    if isinstance(content, str):
        content = content.encode("utf-8")
    files = {"file": (io.BytesIO(content), filename)}
    resp = client.post("/api/convert", data=files, content_type="multipart/form-data")
    assert resp.status_code == 202, resp.get_json()
    return wait_for_job(client, resp.get_json()["job_id"])


# ---------------------------------------------------------------------------
# Finding C2: WeasyPrint SSRF guard - remote resource references are refused
# ---------------------------------------------------------------------------

def test_remote_http_image_reference_rejected(client):
    """A user-supplied .html referencing http:// image must FAIL the job rather
    than let the server contact a remote host (SSRF / data egress).
    Satisfies: C2
    """
    html = (
        "<!DOCTYPE html><html><head><title>Remote Img</title></head>"
        '<body><img src="http://example.com/pic.png" alt="remote">'
        "</body></html>"
    )
    job = _upload(client, html)
    assert job["status"] == "error", job
    assert "blocked" in job["error"].lower() or "remote" in job["error"].lower()


def test_remote_https_image_reference_rejected(client):
    """Same guard for https:// - the common metadata / SSRF vector.
    Satisfies: C2
    """
    html = (
        "<!DOCTYPE html><html><body>"
        '<img src="https://evil.example.com/x.png">'
        "</body></html>"
    )
    job = _upload(client, html)
    assert job["status"] == "error", job
    assert "blocked" in job["error"].lower() or "remote" in job["error"].lower()


def test_local_data_uri_image_succeeds(client):
    """A local `data:` URI resolves through WeasyPrint's default loader and the
    conversion completes. Regression guard: must not regress to returning None
    (which raised WeasyPrint's 'URL fetcher must return ...' assertion).
    Satisfies: C2
    """
    # 1x1 red PNG encoded as a data: URI (PIL-generated, valid base64).
    png = base64.b64decode(
        "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAADElEQVR4nGP4z8AAAAMBAQDJ/pLvAAAAAElFTkSuQmCC"
    )
    data_uri = "data:image/png;base64," + base64.b64encode(png).decode("ascii")
    html = (
        "<!DOCTYPE html><html><head><title>Local Img</title></head>"
        '<body><img src="' + data_uri + '" alt="local">'
        "</body></html>"
    )
    job = _upload(client, html)
    assert job["status"] == "done", job


# ---------------------------------------------------------------------------
# Finding C1 (regression): decompression-bomb guard on zip extraction
# ---------------------------------------------------------------------------
# Two independent mechanisms are exercised:
#   * the SHARED streaming guard common._copy_stream_with_cap (proven by the unit
#     tests: it counts WRITTEN decompressed bytes and aborts mid-stream past the
#     cap), and
#   * app.py's zip-bomb guard end-to-end (proven by the e2e: a genuinely large
#     member trips the per-member size cap and the job fails with 'zip-bomb').


def _append_member(epub_bytes, name, data=b"x", external_attr=None):
    """Append a member to an existing epub byte string (mirrors test_review_fixes.py)."""
    src = io.BytesIO(epub_bytes)
    out = io.BytesIO()
    with zipfile.ZipFile(src, "r") as zin, zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as zout:
        for item in zin.namelist():
            zout.writestr(item, zin.read(item))
        zi = zipfile.ZipInfo(name)
        if external_attr is not None:
            zi.external_attr = external_attr
        zout.writestr(zi, data)
    return out.getvalue()


def test_copy_stream_with_cap_counts_written_bytes():
    """_copy_stream_with_cap returns the exact byte count it streamed to dst.
    Satisfies: C1
    """
    data = b"x" * (10 * 1024)  # 10 KiB
    buf = io.BytesIO()
    written = C._copy_stream_with_cap(io.BytesIO(data), buf, cap=10 * 1024 * 1024)
    assert written == len(data)
    assert buf.getvalue() == data


def test_copy_stream_with_cap_count_only_when_dst_none():
    """When dst is None the bytes are counted but NOT written (used by the pandoc
    pre-check, which must not touch disk). Satisfies: C1
    """
    data = b"x" * (10 * 1024)
    written = C._copy_stream_with_cap(io.BytesIO(data), None, cap=10 * 1024 * 1024)
    assert written == len(data)


def test_copy_stream_with_cap_aborts_on_overflow():
    """_copy_stream_with_cap raises a zip-bomb guard error mid-stream when the
    decompressed total exceeds the cap - the real decompression-bomb protection.
    Satisfies: C1
    """
    data = b"x" * (10 * 1024)  # 10 KiB
    with pytest.raises(ValueError) as excinfo:
        C._copy_stream_with_cap(io.BytesIO(data), None, cap=1024)  # 1 KiB cap
    assert "zip-bomb" in str(excinfo.value).lower()


def test_c1_zip_bomb_guard_rejects_large_member(client, fixture_epub_bytes, monkeypatch):
    """A genuinely large member (ZIP_BOMB_MIN_CAP) fails the job with a 'zip-bomb'
    error through the public HTTP API. Production floor is 2 GiB (impractical in a
    unit test), so the floor is shrunk so the guard logic is exercised with a 4 MiB
    member in a ~8 KiB upload. Satisfies: C1
    """
    monkeypatch.setattr(app, "ZIP_BOMB_MIN_CAP", 1024 * 1024)  # 1 MiB floor
    large = b"\x00" * (4 * 1024 * 1024)  # 4 MiB of NUL bytes, exceeds the 1 MiB cap
    content = _append_member(fixture_epub_bytes, "OEBPS/bomb.bin", large)
    job = _upload(client, content, "bomb.epub")
    assert job["status"] == "error", job
    assert "zip-bomb" in job["error"].lower()


# ---------------------------------------------------------------------------
# Finding C3 (regression): stored-XSS stripping in rendered HTML/SVG
# ---------------------------------------------------------------------------
# WeasyPrint already drops <script> content and event-handler attribute VALUES
# from rendered PDF *text*, so 'script absent from the PDF' is NOT by itself a
# proof the guard works. The AUTHORITATIVE proof is the C3 unit test on
# common._strip_scripts; the e2e below proves the guard is wired into the render
# path without blanking the document.


def test_strip_scripts_removes_all_vectors():
    """_strip_scripts removes <script>/<iframe> content, on* handlers and
    javascript:/vbscript: URLs. This is the authoritative C3 guard test.
    Satisfies: C3
    """
    html = (
        '<div onmouseover="EVIL_ATTR()">BODY</div>'
        "<script>SCRIPT_BODY</script>"
        "<iframe>IFRAME_BODY</iframe>"
        '<a href="javascript:EVIL_JS()">JS_LINK</a>'
        '<a href="vbscript:EVIL_VB()">VB_LINK</a>'
    )
    out = C._strip_scripts(html)
    assert "SCRIPT_BODY" not in out
    assert "IFRAME_BODY" not in out
    assert "EVIL_ATTR()" not in out
    assert "EVIL_JS()" not in out
    assert "EVIL_VB()" not in out


def test_strip_scripts_preserves_visible_and_safe_urls():
    """_strip_scripts leaves visible text and safe data:/https: references intact.
    Satisfies: C3
    """
    html = (
        "<p>VISIBLE_TEXT</p>"
        '<img src="data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1Pe'
        'AAAADElEQVR4nGP4z8AAAAMBAQDJ/pLvAAAAAElFTkSuQmCC" alt="local>"'
        '<a href="https://example.com/ok">SAFE_LINK</a>'
    )
    out = C._strip_scripts(html)
    assert "VISIBLE_TEXT" in out
    assert "SAFE_LINK" in out
    assert "https://example.com/ok" in out
    assert "data:image/png;base64," in out


def test_c3_html_guard_wired_job_done_and_content_preserved(client):
    """An .html upload with <script>, an on* handler and a javascript: link converts
    successfully (guard is wired into the render path) and the visible content is
    preserved. Satisfies: C3
    """
    html = (
        "<!DOCTYPE html><html><head><title>C3</title></head>"
        "<body>"
        "<script>SCRIPT_MARKER_TEXT</script>"
        "<div onmouseover='EVIL_MARKER'>VISIBLE_MAIN_TEXT</div>"
        '<a href="javascript:EVIL_LINK()">LINK</a>'
        "<p>SECOND_VISIBLE</p>"
        "</body></html>"
    )
    job = _upload(client, html, "xss.html")
    assert job["status"] == "done", job
    full = pdf_text(download_bytes(client, job["filename"]))
    # Guard must not blank the document: visible content still renders.
    assert "VISIBLE_MAIN_TEXT" in full
    assert "SECOND_VISIBLE" in full
    # Guarded vectors must not be reachable in the rendered document.
    assert "SCRIPT_MARKER_TEXT" not in full
    assert "EVIL_MARKER" not in full
    assert "EVIL_LINK" not in full


def test_c3_svg_guard_wired_job_done(client):
    """An .svg upload with a <script> converts successfully (guard is wired into the
    SVG render path, which also flows through common._strip_scripts).
    Satisfies: C3
    """
    svg = (
        '<svg xmlns="http://www.w3.org/2000/svg" width="100" height="100">'
        "<script>SVG_SCRIPT_MARKER</script>"
        '<text x="10" y="50">SVG_VISIBLE_TEXT</text>'
        "</svg>"
    )
    job = _upload(client, svg, "xss.svg")
    assert job["status"] == "done", job


# ---------------------------------------------------------------------------
# Finding H1 (regression): CSV row/column render cap in _table_html
# ---------------------------------------------------------------------------
# A single oversized CSV must not render unbounded rows/columns on the single
# worker thread (CPU/memory DoS). Rows/columns are TRUNCATED (not rejected) to
# MAX_ROWS x MAX_COLS, and a visible note is appended so the truncation is honest.
# Satisfies: H1


@pytest.fixture
def csv_converter():
    from converters import csv_converter as C
    return C


def test_table_html_caps_rows_to_max_rows(csv_converter):
    """More than 5000 data rows are truncated to the first 5000 total rows
    (header included); a visible note is added.
    Satisfies: H1
    """
    # header + 5001 body rows = 5002 non-empty rows total.
    body_rows = [["r" + str(i)] for i in range(5001)]
    rows = [["h"]] + body_rows
    out = csv_converter._table_html(rows)
    # rows[:5000] keeps the first 5000 total rows: 1 header + 4999 body = 5000 <tr>.
    assert out.count("<tr>") == 5000, out.count("<tr>")
    assert csv_converter._TRUNCATION_NOTE in out


def test_table_html_caps_cols_to_max_cols(csv_converter):
    """More than 50 columns are truncated to exactly 50 in header and body.
    Satisfies: H1
    """
    # 60-column header + one 60-column body row.
    wide = ["c" + str(i) for i in range(60)]
    out = csv_converter._table_html([wide, wide])
    header_cells = out.split("</tr>")[0].count("<td>") + out.split("</tr>")[0].count("<th>")
    assert header_cells == 50, header_cells
    body_cells = out.split("</tr>")[-2].count("<td>")
    assert body_cells == 50, body_cells
    assert csv_converter._TRUNCATION_NOTE not in out


def test_table_html_undersizes_are_not_truncated(csv_converter):
    """A small table renders in full with no truncation note.
    Satisfies: H1
    """
    rows = [["h1", "h2", "h3"], ["a", "b", "c"]]
    out = csv_converter._table_html(rows)
    assert out.count("<tr>") == 2
    assert "truncated to 5000 rows x 50 cols" not in out


# ---------------------------------------------------------------------------
# Finding M2 (regression): stale output-file sweep
# ---------------------------------------------------------------------------
# A symmetric counterpart to _sweep_stale_work_dirs: output files older than the
# same window are removed at startup, fresh files are kept.


def test_m2_output_sweep_removes_only_stale_output_files(tmp_path, monkeypatch):
    """_sweep_stale_output_files removes OUTPUT_DIR files older than the window,
    keeping fresh ones. Symmetric to the work-dir sweep test.
    Satisfies: M2
    """
    monkeypatch.setattr(app, "OUTPUT_DIR", tmp_path)
    stale = tmp_path / "stale.pdf"
    stale.write_bytes(b"%PDF-1.7 old\n")
    fresh = tmp_path / "fresh.pdf"
    fresh.write_bytes(b"%PDF-1.7 new\n")
    old = time.time() - 7 * 3600  # 7h > 6h window
    os.utime(stale, (old, old))
    app._sweep_stale_output_files()
    assert not stale.exists(), "stale output file must be swept"
    assert fresh.exists(), "fresh output file must be kept"