epub2pdf/converters/common.py

"""Shared building blocks for the PHASE 1 multi-format converters.

The EPUB pipeline (app.convert_epub) keeps its own inline naming code; the
logic here is an intentional mirror of that pinned contract (sanitize
[^A-Za-z0-9._ -]+ -> '_', 60-char cap, no doubled .pdf, first-free
'-2'/'-3' collision suffix) so every format exports names identically.

No module-level import of app: app.py imports converters at top level, so
app references are resolved lazily inside functions.
"""
from __future__ import annotations

import html as _html
import io
import os
import re
from pathlib import Path
from urllib.parse import urlsplit
from weasyprint import default_url_fetcher as _weasy_default_fetcher

__all__ = [
    "PRE_CSS",
    "convert_single",
    "decode_declared",
    "decode_text",
    "derive_export_name",
    "escape",
    "finalize_pdf",
    "html_to_pdf_bytes",
    "report",
]

# CSS for <pre>-based outputs (txt, log, json, xml): monospace, wrapped.
PRE_CSS = """
@page { size: A4; margin: 18mm 16mm; }
body {
    font-family: "DejaVu Sans Mono", "Courier New", monospace;
    font-size: 9pt; line-height: 1.5; color: #1c1b1a;
    font-kerning: none; /* pypdf misreads WeasyPrint kerning as spaces */
}
pre { white-space: pre-wrap; word-wrap: break-word; margin: 0; }
"""


def escape(text: str) -> str:
    """HTML-escape for safe embedding in generated <pre>/<table> bodies."""
    return _html.escape(text, quote=False)


def default_output_dir(output_dir=None) -> Path:
    """CLI-supplied output dir (created), else the app-managed OUTPUT_DIR."""
    if output_dir:
        d = Path(output_dir)
        d.mkdir(parents=True, exist_ok=True)
        return d
    import app  # lazy: avoids a module-level import cycle
    return app.OUTPUT_DIR


def decode_declared(raw: bytes) -> str:
    """Decode honoring an XML-declaration / <meta charset> encoding, else UTF-8.

    Reuses the pinned charset-detection regexes from the EPUB path so an HTML
    file with a windows-1252 declaration decodes like an EPUB chapter.
    Undecodable input never raises.
    """
    import app
    import codecs

    m = app._XML_DECL_ENC.search(raw[:4096]) or app._META_ENC.search(raw[:8192])
    if m:
        enc = m.group(1).decode("ascii", errors="replace").strip()
        try:
            codecs.lookup(enc)
            return raw.decode(enc)
        except (LookupError, UnicodeDecodeError):
            pass
    return raw.decode("utf-8", errors="replace")


def decode_text(raw: bytes) -> str:
    """Best-effort text decode: strict UTF-8, then charset-normalizer, then replace.

    Strict UTF-8 first so valid UTF-8 is never second-guessed; the
    charset-normalizer pass catches legacy encodings (windows-1252, latin-1)
    without mojibake.
    """
    try:
        return raw.decode("utf-8")
    except UnicodeDecodeError:
        pass
    from charset_normalizer import from_bytes

    best = from_bytes(raw).best()
    if best is not None:
        return str(best)
    return raw.decode("utf-8", errors="replace")


def derive_export_name(filename: str, out_dir: Path, title: str | None = None) -> str:
    """Collision-safe PDF name, mirroring the pinned EPUB export-name contract.

    base = clean(title) if title else clean(stem); empty -> 'book'.
    clean: [^A-Za-z0-9._ -]+ -> '_', [:60] cap, strip trailing '.pdf'
    (case-insensitive). Collision: first free of base.pdf, base-2.pdf, ... .
    """
    stem = Path(filename).stem
    if re.fullmatch(r"\.\w+", stem):
        # bare dotfile (".txt"): Path keeps the dot in .stem; treat as empty
        stem = ""
    if title and re.sub(r"[^A-Za-z0-9._ -]+", "_", title).strip():
        base = re.sub(r"[^A-Za-z0-9._ -]+", "_", title).strip()[:60].strip()
    else:
        base = re.sub(r"[^A-Za-z0-9._ -]+", "_", stem).strip()[:60].strip() or "book"
    if base.lower().endswith(".pdf"):
        base = base[:-4].rstrip() or "book"
    filename_out = f"{base}.pdf"
    counter = 2
    while (Path(out_dir) / filename_out).exists():
        filename_out = f"{base}-{counter}.pdf"
        counter += 1
    return filename_out


# Schemes that must never be fetched from a user-supplied document: they let a
# hostile upload make the server contact remote hosts (SSRF / data egress),
# directly contradicting the 'nothing leaves your machine' guarantee.
_REMOTE_SCHEMES = ("http", "https", "ftp", "ftps", "ws", "wss", "file")


def _blocking_url_fetcher(url: str) -> bytes:
    """Raise if `url` points at any remote/external scheme (C2: SSRF guard).

    Local `data:` references and same-directory (relative) resources resolve
    normally; only absolute remote URLs are refused. Returns the bytes that a
    URLFetcher would yield so callers remain simple.
    Satisfies: C2
    """
    import urllib.parse as _urllib_parse

    scheme = _urllib_parse.urlsplit(url).scheme.lower()
    if scheme in _REMOTE_SCHEMES:
        raise ValueError(
            f"blocked remote resource reference ({scheme}://) — the server "
            "does not fetch external URLs"
        )
    # data: and relative/local refs are handled by WeasyPrint's default loader,
    # which asserts the fetcher returns a dict or URLFetcherResponse instance
    # (WeasyPrint 69.0). Delegate here instead of returning None.
    # Satisfies: C2
    return _weasy_default_fetcher(url)


def _strip_scripts(html: str) -> str:
    """Neutralize stored-XSS vectors in user-supplied HTML (C3 shared helper).

    Removes <script>/<iframe> tags and their content, strips inline event-
    handler attributes (onclick=, onmouseover=, ...), and blanks href/src
    attributes whose value is a javascript:/vbscript: URL. Applied to every
    non-pandoc converter (html, svg, epub, md, csv, xml).
    Satisfies: C3
    """
    # <script>...</script> blocks, then any orphan open/close tags.
    html = re.sub(
        r"<\s*script\b[^>]*>.*?<\s*/\s*script\s*>", "", html,
        flags=re.IGNORECASE | re.DOTALL,
    )
    html = re.sub(r"<\s*/?\s*script\s*/?>", "", html, flags=re.IGNORECASE)
    # <iframe>...</iframe> blocks, then orphan tags.
    html = re.sub(
        r"<\s*iframe\b[^>]*>.*?<\s*/\s*iframe\s*>", "", html,
        flags=re.IGNORECASE | re.DOTALL,
    )
    html = re.sub(r"<\s*/?\s*iframe\s*/?>", "", html, flags=re.IGNORECASE)
    # Inline event handlers: onclick=, onmouseover=, ...
    html = re.sub(
        r"\s+on\w+\s*=\s*(\"[^\"]*\"|'[^']*'|[^\s>]+)", "", html,
        flags=re.IGNORECASE,
    )

    # javascript:/vbscript: URLs in href/src -> blank the attribute.
    # group(4) captures the value INCLUDING its surrounding quotes, so the
    # scheme check strips the quotes first; otherwise the leading quote makes
    # "javascript:" never match and the handler silently survives (the C3 bug).
    def _blank(match):
        quote, scheme = (
            match.group(4)[0],
            match.group(4).strip().strip('"').strip("'").strip(),
        )
        if re.match(r"(?:javascript|vbscript):", scheme, re.IGNORECASE):
            return f"{match.group(1)}{match.group(2)}{match.group(3)}{quote}{quote}"
        return match.group(0)

    html = re.sub(
        r"(\s)(href|src)(\s*=\s*)(\"[^\"]*\"|'[^']*'|[^\s>]+)",
        _blank,
        html, flags=re.IGNORECASE,
    )
    return html


def zip_header_guard(infos, cap) -> None:
    """Header-based pre-check shared by both zip-extraction paths (C1).

    Validates that the *claimed* total and per-member uncompressed sizes from
    the central directory stay within `cap`. This is defense-in-depth; the
    real decompression-bomb guard is _copy_stream_with_cap, which counts
    BYTES WRITTEN during extraction rather than trusting the header.
    """
    total = sum(i.file_size for i in infos)
    if total > cap:
        raise ValueError(
            f"zip-bomb guard: declared uncompressed size {total} exceeds "
            f"cap {cap}"
        )
    for info in infos:
        if info.file_size > cap:
            raise ValueError(
                f"zip-bomb guard: member {info.filename} declared "
                f"{info.file_size} bytes (cap {cap})"
            )


def _copy_stream_with_cap(src, dst, cap) -> int:
    """Stream-copy src\u2192dst counting WRITTEN bytes; abort mid-stream if the
    decompressed total exceeds `cap` (C1 real enforcement).

    Unlike a header-only check, this counts actual bytes delivered to disk, so a
    zip that inflates a 5 MB header to 50 GB is stopped mid-write. `dst` may be
    None: when None the bytes are counted but not written (a pure pre-check that
    must not touch disk), e.g. streaming the decompressed size of an office zip
    before pandoc reads it. Always returns the total WRITTEN bytes; callers that
    write to a file ignore the return.

    Satisfies: C1
    """
    chunk_size = 64 * 1024
    written = 0
    while True:
        chunk = src.read(chunk_size)
        if not chunk:
            return written
        written += len(chunk)
        if written > cap:
            raise ValueError(
                f"zip-bomb guard: member exceeded decompressed cap "
                f"{written} bytes (cap {cap}); aborting extraction to "
                "prevent disk-fill"
            )
        if dst is not None:
            dst.write(chunk)
    return written


# Resource-attribute names whose string value WeasyPrint may fetch from the
# network. Only these (src, href, srcset, poster, data, action, cite,
# formaction, ping, xlink:href, usesrc) are scanned; every other attribute is
# ignored so unrelated user markup is never mis-parsed.
_RESOURCE_ATTRS = frozenset(
    {"src", "href", "srcset", "poster", "data", "action", "cite",
     "formaction", "ping", "xlink:href", "usesrc"}
)

# Matches `name = <value>` where value is single- or double-quoted.
_ATTR_RE = re.compile(r"""(\w[\w-]*)\s*=\s*["']([^"']*)["']""", re.IGNORECASE)


def scan_for_remote(html: str) -> None:
    """Pre-render SSRF guard (C2): reject HTML referencing any remote resource.

    WeasyPrint's image loader *swallows* per-fetch ValueError / URLFetchingError
    and renders a broken image, so a fetcher that merely raises inside a WeasyPrint
    pass can still leave a job reporting DONE with a remote URL contacted. To make
    the guard authoritative we reject remote resource references *up-front*: a
    ValueError here propagates out of html_to_pdf_bytes and the worker records a
    job error. Local `data:` URIs and same-directory (relative) references pass.
    Satisfies: C2
    """
    for match in _ATTR_RE.finditer(html):
        if match.group(1).lower() not in _RESOURCE_ATTRS:
            continue
        value = match.group(2).strip()
        if not value:
            continue
        for part in value.split(","):
            tokens = part.split()
            url = (tokens[0] if tokens else "").strip()
            if not url:
                continue
            scheme = urlsplit(url).scheme.lower()
            if scheme in _REMOTE_SCHEMES:
                raise ValueError(
                    f"blocked remote resource reference ({scheme}://) - the "
                    "server does not fetch external URLs"
                )
    return


def html_to_pdf_bytes(html: str,
                      css: str | None = None,
                      url_fetcher=None) -> bytes:
    """Render an HTML string to PDF bytes via WeasyPrint.

    css: extra stylesheet (e.g. PRE_CSS for text dumps). Empty/None means
    the document carries its own <style> (HTML/MD/SVG/table converters
    inject BASE_CSS into the document head so their rules win over the
    fallback at equal specificity).

    url_fetcher: retained for API compatibility; the authoritative SSRF guard is
    scan_for_remote() which rejects remote resource references before WeasyPrint
    ever runs, so every converter rendering user HTML/SVG is SSRF-safe by default.
    Satisfies: C2
    """
    # C3: strip stored-XSS vectors BEFORE WeasyPrint renders the document.
    # This is the single choke point every HTML/SVG/MD/CSV/XML/text converter
    # flows through (svg renders here directly; the rest reach it via
    # common.convert_single -> html_to_pdf_bytes), so one strip here covers the
    # whole non-pandoc render surface. Harmless no-op for json/text, which escape
    # their content into &lt;...&gt; so the script regex never matches.
    html = _strip_scripts(html)
    scan_for_remote(html)
    from weasyprint import CSS, HTML

    buf = io.BytesIO()
    kwargs = {"stylesheets": [CSS(string=css)]} if css else {}
    HTML(string=html, url_fetcher=url_fetcher).write_pdf(buf, **kwargs)
    return buf.getvalue()


def finalize_pdf(pdf_bytes: bytes, out_dir: Path, filename: str,
                 job_id: str, progress_cb=None, title: str | None = None) -> str:
    """Add metadata, write atomically (.tmp + os.replace); return the name.

    title: explicit PDF /Title (e.g. HTML <title> or Markdown H1); falls
    back to the export filename stem when omitted.

    Enforces the documented output-size cap (finding: H2): any produced PDF
    over MAX_OUTPUT_MB is rejected before it is written.
    Satisfies: H2
    """
    import app  # lazy: MAX_OUTPUT_MB lives with the EPUB pipeline config
    from pypdf import PdfReader, PdfWriter

    max_output_bytes = getattr(app, "MAX_OUTPUT_MB", 200) * 1024 * 1024
    if len(pdf_bytes) > max_output_bytes:
        raise ValueError(
            f"output too large: PDF is {len(pdf_bytes)} bytes, exceeds the "
            f"{getattr(app, 'MAX_OUTPUT_MB', 200)} MB limit"
        )

    writer = PdfWriter()
    for page in PdfReader(io.BytesIO(pdf_bytes)).pages:
        writer.add_page(page)
    if not writer.pages:
        raise ValueError("no renderable content: conversion produced 0 PDF pages")
    writer.add_metadata({"/Title": (title or "").strip() or Path(filename).stem})
    if progress_cb:
        progress_cb(99.0, "Finalizing PDF\u2026")
    tmp = Path(out_dir) / (filename + ".tmp")  # atomic: write .tmp then rename
    with open(tmp, "wb") as out:
        writer.write(out)
    os.replace(tmp, Path(out_dir) / filename)
    return filename


def report(progress_cb, pct, message):
    """Progress callback with a message matching the UI regex \\((\\d+)/(\\d+)\\)\n    when a count label is supplied, else a plain status line."""
    if progress_cb:
        progress_cb(pct, message)


def convert_single(path: Path, job_id: str, *, build_html, title=None,
                   css=PRE_CSS, progress_cb=None, output_dir=None,
                   name_src: str | None = None) -> str:
    """Common single-document pipeline: build_html() -> render -> finalize.

    build_html() must return (html_body, page_label) where page_label feeds
    the UI progress line. Raising ValueError here becomes a job error with the
    same message the user sees.

    name_src: original upload/CLI filename to derive the export name from
    (web uploads land on disk as {job_id}<ext>, so path.name is unusable
    there); defaults to path.name (CLI mode).
    """
    out_dir = default_output_dir(output_dir)
    body, label = build_html()
    report(progress_cb, 30.0, f"Rendering {label} (1/1)")
    pdf_bytes = html_to_pdf_bytes(body, css)
    filename = derive_export_name(name_src or path.name, out_dir, title=title)
    return finalize_pdf(pdf_bytes, out_dir, filename, job_id,
                        progress_cb=progress_cb, title=title)