Blame for app.py

"""
🦊 Sisi PDF API - Universal HTML to PDF Converter
Perfect support for Chinese (Simplified & Traditional), Emoji, SVG, and modern CSS
Powered by Playwright + Chromium
🔒 Protected with API Key Authentication

Hardening (v2.0.0):
- API_KEY is required; the service refuses to start without it (no default key)
- Shared Chromium browser launched once at startup; concurrency-limited (no per-request launch)
- HTML payload size limit (default 2MB)
- Content-Disposition filename sanitized (no header injection)
- Renderer network requests to internal/private hosts blocked (SSRF guard)
- CORS origins configurable via env (defaults to permissive to avoid breakage)
"""

import asyncio
import base64
import ipaddress
import logging
import os
import re
import secrets
from contextlib import asynccontextmanager
from typing import Optional
from urllib.parse import quote, urlparse

from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import Response
from playwright.async_api import async_playwright
from pydantic import BaseModel

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ── Configuration (all from environment) ──────────────────────────────────────
# 🔐 API key is mandatory. Refuse to start without it (no insecure default).
API_KEY = os.environ.get("API_KEY")
if not API_KEY:
    raise RuntimeError(
        "API_KEY environment variable is required. "
        "Set it as a Space secret before starting the service."
    )

# HTML payload cap (~2MB, aligned with the branded-report proxy guideline)
MAX_HTML_BYTES = int(os.environ.get("MAX_HTML_BYTES", 2_000_000))
# Concurrent PDF renders sharing the single Chromium process
PDF_MAX_CONCURRENCY = int(os.environ.get("PDF_MAX_CONCURRENCY", 4))
# When "strict", block ALL external http(s) sub-resources (only data:/blob:/about:)
BLOCK_EXTERNAL_STRICT = os.environ.get("PDF_BLOCK_EXTERNAL", "").lower() == "strict"

CHROMIUM_ARGS = [
    "--no-sandbox",
    "--disable-setuid-sandbox",
    "--disable-dev-shm-usage",
    "--disable-gpu",
]

ALWAYS_ALLOWED_SCHEMES = {"data", "blob", "about"}


def verify_api_key(x_api_key: Optional[str] = Header(None)):
    """Verify API key from request header (constant-time comparison)."""
    if not x_api_key or not secrets.compare_digest(x_api_key, API_KEY):
        raise HTTPException(
            status_code=403,
            detail="🔒 Invalid or missing API key. Please contact SisiTheFox for access.",
        )
    return x_api_key


# ── Lifespan: launch the shared browser once ──────────────────────────────────
@asynccontextmanager
async def lifespan(app: FastAPI):
    pw = await async_playwright().start()
    browser = await pw.chromium.launch(headless=True, args=CHROMIUM_ARGS)
    app.state.playwright = pw
    app.state.browser = browser
    app.state.semaphore = asyncio.Semaphore(PDF_MAX_CONCURRENCY)
    logger.info("🦊 Chromium launched (shared, max concurrency=%s)", PDF_MAX_CONCURRENCY)
    try:
        yield
    finally:
        await browser.close()
        await pw.stop()
        logger.info("🦊 Chromium closed")


app = FastAPI(
    title="Sisi PDF API",
    description="Universal HTML to PDF converter with perfect Chinese, Emoji, and SVG support",
    version="2.0.0",
    lifespan=lifespan,
)

# ── CORS (env-configurable; default permissive to avoid breaking callers) ──────
_origins_env = os.environ.get("ALLOWED_ORIGINS", "*").strip()
if _origins_env == "*":
    _allow_origins = ["*"]
    _allow_credentials = False  # browsers forbid credentials with wildcard origin
else:
    _allow_origins = [o.strip() for o in _origins_env.split(",") if o.strip()]
    _allow_credentials = True

app.add_middleware(
    CORSMiddleware,
    allow_origins=_allow_origins,
    allow_credentials=_allow_credentials,
    allow_methods=["*"],
    allow_headers=["*"],
)


class PDFRequest(BaseModel):
    """PDF generation request"""
    html_content: str  # Full HTML content to convert
    filename: Optional[str] = "document.pdf"
    page_size: Optional[str] = "A4"  # A4, Letter, A3, etc.
    margin_top: Optional[str] = "20mm"
    margin_bottom: Optional[str] = "20mm"
    margin_left: Optional[str] = "20mm"
    margin_right: Optional[str] = "20mm"
    print_background: Optional[bool] = True
    landscape: Optional[bool] = False
    prefer_css_page_size: Optional[bool] = False
    render_timeout_ms: Optional[int] = 30000


# ── Helpers ───────────────────────────────────────────────────────────────────
def sanitize_filename(name: Optional[str]) -> str:
    """Make a safe filename for Content-Disposition (prevents header injection)."""
    name = (name or "").strip()
    name = name.replace("\\", "/").split("/")[-1]          # strip any path
    name = re.sub(r'[\r\n"]', "", name)                    # CR/LF/quotes
    name = "".join(ch for ch in name if ch >= " ")         # drop control chars
    name = name.strip()
    if not name:
        name = "document.pdf"
    if not name.lower().endswith(".pdf"):
        name += ".pdf"
    return name[:200]


def content_disposition(name: str) -> str:
    """Return an ASCII-safe header while preserving UTF-8 names for browsers."""
    ascii_name = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip(".-")
    if not ascii_name:
        ascii_name = "document.pdf"
    elif not ascii_name.lower().endswith(".pdf"):
        ascii_name += ".pdf"
    encoded_name = quote(name, safe="")
    return f"attachment; filename=\"{ascii_name}\"; filename*=UTF-8''{encoded_name}"


async def _host_is_internal(host: str) -> bool:
    """True if host resolves to a private/loopback/link-local/reserved address."""
    try:
        loop = asyncio.get_running_loop()
        infos = await loop.getaddrinfo(host, None)
    except Exception:
        return True  # cannot resolve -> treat as unsafe
    for info in infos:
        addr = info[4][0]
        try:
            ip = ipaddress.ip_address(addr)
        except ValueError:
            return True
        if (
            ip.is_private or ip.is_loopback or ip.is_link_local
            or ip.is_reserved or ip.is_multicast or ip.is_unspecified
        ):
            return True
    return False


async def _guard_route(route):
    """SSRF guard for sub-resource requests made while rendering the HTML."""
    try:
        parsed = urlparse(route.request.url)
        scheme = parsed.scheme.lower()
        if scheme in ALWAYS_ALLOWED_SCHEMES:
            await route.continue_()
            return
        if scheme not in ("http", "https"):
            await route.abort()
            return
        if BLOCK_EXTERNAL_STRICT:
            await route.abort()
            return
        host = parsed.hostname
        if not host or await _host_is_internal(host):
            await route.abort()
            return
        await route.continue_()
    except Exception:
        # Never let the guard hang the render; fail closed.
        try:
            await route.abort()
        except Exception:
            pass


async def prepare_page(page, request: PDFRequest):
    """Load HTML and wait for print assets before Chromium snapshots the page."""
    timeout = max(5000, min(request.render_timeout_ms or 30000, 60000))
    await page.emulate_media(media="print")
    await page.set_content(request.html_content, wait_until="networkidle", timeout=timeout)
    await page.evaluate("document.fonts.ready")
    await page.evaluate(
        """
        async () => {
          await Promise.all([...document.images].map((image) => {
            if (image.complete) return Promise.resolve();
            return new Promise((resolve) => {
              image.addEventListener('load', resolve, { once: true });
              image.addEventListener('error', resolve, { once: true });
            });
          }));
        }
        """
    )


async def render_pdf(request: PDFRequest) -> bytes:
    """Render a PDF using the shared browser, isolated per-request context."""
    if len(request.html_content.encode("utf-8")) > MAX_HTML_BYTES:
        raise HTTPException(
            status_code=413,
            detail=f"HTML content too large (max {MAX_HTML_BYTES} bytes).",
        )

    async with app.state.semaphore:
        context = await app.state.browser.new_context()
        try:
            page = await context.new_page()
            await page.route("**/*", _guard_route)
            await prepare_page(page, request)
            return await page.pdf(
                format=request.page_size,
                prefer_css_page_size=request.prefer_css_page_size,
                print_background=request.print_background,
                landscape=request.landscape,
                margin={
                    "top": request.margin_top,
                    "bottom": request.margin_bottom,
                    "left": request.margin_left,
                    "right": request.margin_right,
                },
            )
        finally:
            await context.close()


# ── Routes ────────────────────────────────────────────────────────────────────
@app.get("/")
def root():
    """Health check"""
    return {
        "status": "online",
        "service": "Sisi PDF API",
        "version": "2.0.0",
        "features": [
            "Perfect Chinese support (Simplified & Traditional)",
            "Colorful emoji rendering",
            "SVG graphics (100% accurate)",
            "Modern CSS (Flexbox, Grid, Gradients)",
            "Powered by Chromium",
        ],
    }


@app.get("/health")
@app.head("/health")
def health_check():
    """Health check endpoint for monitoring"""
    return {"status": "healthy", "service": "Sisi PDF API", "version": "2.0.0"}


@app.post("/generate-pdf")
async def generate_pdf(request: PDFRequest, api_key: str = Depends(verify_api_key)):
    """
    Generate PDF from HTML content. 🔒 Requires API Key authentication.
    Perfect support for Chinese, Emoji, SVG and modern CSS3.
    """
    safe_name = sanitize_filename(request.filename)
    try:
        logger.info("📄 Generating PDF: %s", safe_name)
        pdf_bytes = await render_pdf(request)
        logger.info("✅ PDF generated: %d bytes", len(pdf_bytes))
        return Response(
            content=pdf_bytes,
            media_type="application/pdf",
            headers={"Content-Disposition": content_disposition(safe_name)},
        )
    except HTTPException:
        raise
    except Exception as e:
        logger.error("❌ PDF generation error: %s", str(e))
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/generate-pdf-base64")
async def generate_pdf_base64(request: PDFRequest, api_key: str = Depends(verify_api_key)):
    """
    Generate PDF and return as base64 string. 🔒 Requires API Key authentication.
    Useful for embedding in JSON responses.
    """
    safe_name = sanitize_filename(request.filename)
    try:
        logger.info("📄 Generating PDF (base64): %s", safe_name)
        pdf_bytes = await render_pdf(request)
        pdf_base64 = base64.b64encode(pdf_bytes).decode("utf-8")
        logger.info("✅ PDF generated (base64): %d bytes", len(pdf_bytes))
        return {
            "success": True,
            "filename": safe_name,
            "pdf_base64": pdf_base64,
            "size_bytes": len(pdf_bytes),
        }
    except HTTPException:
        raise
    except Exception as e:
        logger.error("❌ PDF generation error: %s", str(e))
        raise HTTPException(status_code=500, detail=str(e))


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=7860)