"""Unified tabular XLSX export primitive. Every admin / self-service list surface that offers a download goes through :func:`xlsx_export_response` so the platform has exactly one implementation of: * the two-row header convention (human label row + machine field row — the same shape the asset bulk-update template established, so any export can round-trip through a future import flow); * Excel formula-injection neutralisation (OWASP "CSV injection") for every text cell, not just the columns someone remembered to guard; * tz-aware datetime / date / Decimal / UUID cell coercion (TiDB rows come back tz-aware; openpyxl only accepts naive datetimes); * the ``X-Export-Total-Matched / X-Export-Returned / X-Export-Truncated`` response headers the shared frontend ``ExportButton`` reads to tell the admin "you got 5 000 of 12 340 rows — narrow the filter"; * the export audit trail (`*.export` actions) — downloading a list is a data-egress event and must leave the same forensic residue as any state-changing operation. Endpoints stay thin: mirror the list endpoint's filter signature, call the same service/list builder with ``page=1, page_size=``, flatten rows to dicts, then hand off to this module. Reusing the list query path means every filter/RBAC fix on the list applies to the export automatically — the two surfaces cannot drift. """ from __future__ import annotations import io import re from collections.abc import Iterable, Mapping, Sequence from datetime import UTC, date, datetime from decimal import Decimal from typing import Any from urllib.parse import quote from uuid import UUID from fastapi import Request from fastapi.responses import StreamingResponse from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.dependencies import get_real_ip # Hard ceiling on a single export. Generous enough to cover every # current list surface end-to-end (the largest, provisioning_tasks, is # ~20k rows lifetime), but bounds the memory a runaway filter can # materialise. Past this point the admin should narrow the filter — # there is no value in a 100k-row spreadsheet. EXPORT_ROW_LIMIT = 20000 XLSX_MEDIA_TYPE = ( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ) # Excel treats a leading ``= + - @`` as a formula. Tab / CR prefixes are # the historical Excel DDE vector. Neutralise with a leading apostrophe # (renders invisibly in Excel, keeps the cell text-typed). _FORMULA_PREFIXES = ("=", "+", "-", "@", "\t", "\r") def to_cell(value: Any) -> Any: """Coerce a Python value into an openpyxl-safe, injection-safe cell.""" if value is None: return "" if isinstance(value, datetime): # tz-aware → naive UTC string. openpyxl rejects tz-aware # datetimes; iso-stringifying keeps Excel happy without a stray # ``+00:00`` suffix. if value.tzinfo is not None: value = value.astimezone(UTC).replace(tzinfo=None) return value.isoformat(sep=" ", timespec="seconds") if isinstance(value, date): return value.isoformat() if isinstance(value, Decimal): return float(value) if isinstance(value, UUID): return str(value) if isinstance(value, bool): # openpyxl renders bools as TRUE/FALSE which Excel localises # unpredictably; a lowercase literal string is stable across # locales and survives re-import comparisons. return "true" if value else "false" if isinstance(value, (list, tuple, set)): return ", ".join(str(to_cell(v)) for v in value) if isinstance(value, dict): import json return json.dumps(value, ensure_ascii=False, default=str) if isinstance(value, str) and value.startswith(_FORMULA_PREFIXES): return "'" + value return value def build_xlsx( *, columns: Sequence[tuple[str, str]], rows: Iterable[Mapping[str, Any]], sheet_title: str = "Export", ) -> io.BytesIO: """Build a styled workbook: label header, machine-key row, data rows.""" from openpyxl import Workbook from openpyxl.styles import Alignment, Font, PatternFill wb = Workbook() ws = wb.active # Excel sheet titles are capped at 31 chars and reject ``[]:*?/\``. safe_title = "".join( c for c in sheet_title if c not in "[]:*?/\\" )[:31] or "Export" ws.title = safe_title header_fill = PatternFill( start_color="4472C4", end_color="4472C4", fill_type="solid", ) header_font = Font(color="FFFFFF", bold=True, size=11) machine_fill = PatternFill( start_color="D9E2F3", end_color="D9E2F3", fill_type="solid", ) machine_font = Font(color="555555", italic=True, size=9) for col_idx, (_key, label) in enumerate(columns, 1): cell = ws.cell(row=1, column=col_idx, value=label) cell.fill = header_fill cell.font = header_font cell.alignment = Alignment(horizontal="center") for col_idx, (key, _label) in enumerate(columns, 1): cell = ws.cell(row=2, column=col_idx, value=key) cell.fill = machine_fill cell.font = machine_font cell.alignment = Alignment(horizontal="center") keys = [k for k, _ in columns] row_idx = 2 for item in rows: row_idx += 1 for col_idx, key in enumerate(keys, start=1): ws.cell(row=row_idx, column=col_idx, value=to_cell(item.get(key))) for col_idx in range(1, len(keys) + 1): letter = ws.cell(row=1, column=col_idx).column_letter ws.column_dimensions[letter].width = 18 buf = io.BytesIO() wb.save(buf) buf.seek(0) return buf async def audit_export( db: AsyncSession, request: Request, current_user: AuthUser, *, action: str, resource_type: str, row_count: int, total_matched: int, filters: Mapping[str, Any] | None = None, ) -> None: """Write the standard export audit row. ``action`` follows the ``{domain}.{entity}.export`` convention (the asset domain's pre-existing ``asset.export`` is grandfathered). Filters are stringified shallowly so UUID / list values serialize. Commits immediately: ``get_db`` only commits after the download response has fully streamed, so a client that disconnects mid- download would otherwise roll the audit row back — the egress would leave no trail. Export endpoints only SELECT before this point, so the early commit has no other side effects. """ safe_filters: dict[str, Any] = {} for key, value in (filters or {}).items(): if value is None: safe_filters[key] = None elif isinstance(value, (list, tuple, set)): safe_filters[key] = [str(v) for v in value] elif isinstance(value, (str, int, float, bool)): safe_filters[key] = value else: safe_filters[key] = str(value) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action=action, resource_type=resource_type, resource_id=None, status="success", detail={ "row_count": row_count, "total_matched": total_matched, "truncated": total_matched > row_count, "filters": safe_filters, }, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() _FILENAME_UNSAFE_RE = re.compile(r'[\\/:*?"<>|\x00-\x1f]') def content_disposition(filename: str) -> str: """Public alias — see :func:`_content_disposition`. Exported because the download endpoints that do NOT go through ``xlsx_response`` need the same header construction. One of them (the self-attestation docx/pdf export) hand-rolled an ``f"attachment; filename={stem}"`` and 500'd on every real campaign: the label is Chinese by construction and starlette encodes header values as latin-1 (#2127). """ return _content_disposition(filename) def _content_disposition(filename: str) -> str: """Build a header-safe Content-Disposition for an arbitrary filename. Filename stems can carry free text (e.g. vendor names) — quotes/CR/LF must never reach the header, and non-ASCII needs the RFC 5987 ``filename*`` form with an ASCII fallback for old clients. """ cleaned = _FILENAME_UNSAFE_RE.sub("_", filename).strip() or "export.xlsx" ascii_fallback = cleaned.encode("ascii", "replace").decode("ascii").replace('"', "_") disposition = f'attachment; filename="{ascii_fallback}"' if cleaned != ascii_fallback: disposition += f"; filename*=UTF-8''{quote(cleaned)}" return disposition def xlsx_response( buf: io.BytesIO, *, filename_stem: str, returned: int, total_matched: int, ) -> StreamingResponse: """Wrap a built workbook in the standard download response.""" stamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S") filename = f"{filename_stem}_{stamp}.xlsx" return StreamingResponse( buf, media_type=XLSX_MEDIA_TYPE, headers={ "Content-Disposition": _content_disposition(filename), "X-Export-Total-Matched": str(total_matched), "X-Export-Returned": str(returned), "X-Export-Truncated": ( "true" if total_matched > returned else "false" ), "X-Content-Type-Options": "nosniff", }, ) async def xlsx_export_response( db: AsyncSession, request: Request, current_user: AuthUser, *, columns: Sequence[tuple[str, str]], rows: Sequence[Mapping[str, Any]], total_matched: int, filename_stem: str, sheet_title: str, audit_action: str, audit_resource_type: str, filters: Mapping[str, Any] | None = None, ) -> StreamingResponse: """One-call composition: audit + build + response headers. ``rows`` must already be flattened dicts (the endpoint owns the row shape — typically ``model_dump()`` of the list row schema plus any join-enriched display fields). """ await audit_export( db, request, current_user, action=audit_action, resource_type=audit_resource_type, row_count=len(rows), total_matched=total_matched, filters=filters, ) buf = build_xlsx(columns=columns, rows=rows, sheet_title=sheet_title) return xlsx_response( buf, filename_stem=filename_stem, returned=len(rows), total_matched=total_matched, )