"""Tenant branding (logo + cemetery details) for report PDF headers. The logo a tenant uploads under Settings → Cemetery Profile (``PUT /api/v1/settings/profile/logo``) is stored two ways, and this module reads it back for embedding into report PDFs. SECURITY — why ``branding_configs.logo_url`` is never dereferenced ----------------------------------------------------------------- ``logo_url`` is a tenant-controlled string. Fetching it over HTTP would hand an authenticated tenant admin a server-side request primitive (OWASP A10 / API7), which is exactly what src/apps/reports/services/export.py's "no user-supplied URL is ever fetched" guarantee and scheduling_pdf.py's text-only header both exist to avoid. Both storage layouts are fully derivable from ``tenant_id`` plus a filename, so this module uses ``logo_url`` only as (a) a presence flag and (b) a cache key, and takes nothing from it but the final path segment — which is then validated as a bare filename before it is used to build a local path or an S3 key: local : /uploads/tenants/{tenant_id}/logo/{filename} S3 : s3://{settings.S3_BUCKET}/tenants/{tenant_id}/logo/{filename} The host in the stored URL is deliberately ignored (CLOUDFRONT_URL can be empty while previously-stored URLs still carry a CloudFront host). Every failure is non-fatal: a missing, oversized, corrupt or unreadable logo logs a warning and yields ``None`` so the export still succeeds unbranded. """ from __future__ import annotations import asyncio import hashlib import io import logging import re from dataclasses import dataclass from pathlib import Path from typing import Optional from urllib.parse import unquote, urlparse from sqlalchemy import text as sa_text from sqlalchemy.ext.asyncio import AsyncSession from src.core.config import settings from src.database.session import get_redis logger = logging.getLogger(__name__) # Mirrors _LOGO_ALLOWED_MIME / _LOGO_MAX_BYTES in src/apps/settings/router.py — # the upload side's allowlist and 2 MB cap, re-asserted on the way back out # because storage may predate (or outlive) any given upload rule. _ALLOWED_EXTENSIONS = {".png", ".jpg", ".jpeg", ".svg"} _MAX_LOGO_BYTES = 2 * 1024 * 1024 # A logo filename must be a single bare path segment. Anything with a separator, # a parent-directory hop, or a NUL is rejected outright rather than sanitised. _SAFE_FILENAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._\- ]{0,120}$") # Uploads root, matching _LOCAL_UPLOADS_ROOT in src/apps/settings/router.py: # /uploads. This file is src/apps/reports/services/branding.py, so # the repo root is five parents up. _LOCAL_UPLOADS_ROOT = Path(__file__).resolve().parents[4] / "uploads" # Header logos render ~14 mm tall in a landscape A4 header; 300 px on the long # edge is plenty at print resolution and keeps an 850 KB upload from inflating # every single export. _MAX_EDGE_PX = 300 # Refuse absurd pixel dimensions before Pillow allocates a decode buffer # (decompression-bomb guard — Image.open is lazy, so this is checked pre-load). _MAX_PIXELS = 40_000_000 _CACHE_PREFIX = "reports:logo:v1" _CACHE_TTL_SECONDS = 3600 @dataclass(frozen=True) class LogoAsset: """A report-ready logo: raster bytes as PNG, or the raw bytes of an SVG.""" data: bytes is_svg: bool @dataclass(frozen=True) class ReportHeader: """Everything the PDF header prints: the logo plus the cemetery's identity. Any field may be None/empty — the renderer simply omits the missing lines, so a tenant with no logo or a blank address still gets a sensible header. """ logo: Optional[LogoAsset] = None cemetery_name: Optional[str] = None address: Optional[str] = None email: Optional[str] = None def __bool__(self) -> bool: """True when there is anything at all worth rendering.""" return any((self.logo, self.cemetery_name, self.address, self.email)) def _safe_basename(logo_url: str) -> Optional[str]: """Extract and validate the bare filename from a stored logo URL. Returns None for anything that is not a plain, allow-extension filename. The URL is parsed only to find its last path segment; it is never fetched. """ path = urlparse(logo_url.strip()).path candidate = unquote(path.rsplit("/", 1)[-1]).strip() if not candidate or not _SAFE_FILENAME.match(candidate): return None # Belt-and-braces: reject separators/traversal that survived the regex. if "/" in candidate or "\\" in candidate or ".." in candidate or "\x00" in candidate: return None if Path(candidate).suffix.lower() not in _ALLOWED_EXTENSIONS: return None return candidate def _read_local(tenant_id: str, filename: str) -> Optional[bytes]: """Read the logo from the local uploads dir, if it is there. The resolved path is asserted to sit inside *this* tenant's own logo directory, so a filename that somehow escapes validation still cannot reach another tenant's files or the wider filesystem. """ logo_dir = (_LOCAL_UPLOADS_ROOT / "tenants" / tenant_id / "logo").resolve() target = (logo_dir / filename).resolve() if logo_dir not in target.parents: logger.warning("[reports/logo] path escapes tenant logo dir — refusing read") return None if not target.is_file(): return None if target.stat().st_size > _MAX_LOGO_BYTES: logger.warning("[reports/logo] local logo exceeds %d bytes — skipping", _MAX_LOGO_BYTES) return None return target.read_bytes() def _make_s3_client(): """Per-call S3 client, matching the pattern in documents/document_service.py.""" import boto3 return boto3.client( "s3", region_name=settings.AWS_REGION, aws_access_key_id=settings.AWS_ACCESS_KEY_ID, aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY, ) def _fetch_s3_sync(key: str) -> Optional[bytes]: """Blocking S3 read of an object key we constructed ourselves.""" from botocore.exceptions import BotoCoreError, ClientError try: s3 = _make_s3_client() head = s3.head_object(Bucket=settings.S3_BUCKET, Key=key) if head.get("ContentLength", 0) > _MAX_LOGO_BYTES: logger.warning("[reports/logo] s3 logo exceeds %d bytes — skipping", _MAX_LOGO_BYTES) return None obj = s3.get_object(Bucket=settings.S3_BUCKET, Key=key) # amt caps the read even if ContentLength lied. return obj["Body"].read(_MAX_LOGO_BYTES + 1)[:_MAX_LOGO_BYTES] except (BotoCoreError, ClientError) as exc: logger.warning("[reports/logo] s3 fetch failed for %s: %s", key, exc) return None def _normalise_raster(raw: bytes) -> Optional[bytes]: """Downscale and re-encode a raster logo as PNG, preserving transparency.""" from PIL import Image try: with Image.open(io.BytesIO(raw)) as img: width, height = img.size if width * height > _MAX_PIXELS: logger.warning("[reports/logo] logo dimensions too large (%dx%d)", width, height) return None img = img.convert("RGBA" if img.mode in ("RGBA", "LA", "P") else "RGB") if max(img.size) > _MAX_EDGE_PX: img.thumbnail((_MAX_EDGE_PX, _MAX_EDGE_PX), Image.LANCZOS) out = io.BytesIO() img.save(out, format="PNG", optimize=True) return out.getvalue() except Exception as exc: # Pillow raises a wide variety on malformed input logger.warning("[reports/logo] could not decode logo image: %s", exc) return None async def _cache_get(cache_key: str) -> Optional[bytes]: redis = await get_redis() if redis is None: # get_redis() returns None on connection failure return None try: return await redis.get(cache_key) except Exception: return None async def _cache_set(cache_key: str, payload: bytes) -> None: redis = await get_redis() if redis is None: return try: await redis.setex(cache_key, _CACHE_TTL_SECONDS, payload) except Exception: pass async def load_tenant_logo(db: AsyncSession, tenant_id: str) -> Optional[LogoAsset]: """Return the tenant's report-ready logo, or None if there isn't a usable one. Never raises: any problem resolving, reading or decoding the logo is logged and reported as None so a broken logo cannot break a report export. """ try: logo_url = ( await db.execute( # BrandingConfig's PK *is* account_id (== tenant id), so this # needs no join. Queried as raw SQL rather than via the ORM # entity because BrandingConfig.relationship("Account") only # resolves once the whole model registry is loaded. sa_text( "SELECT logo_url FROM branding_configs " "WHERE account_id = CAST(:tid AS uuid)" ), {"tid": tenant_id}, ) ).scalar_one_or_none() except Exception as exc: logger.warning("[reports/logo] branding lookup failed: %s", exc) return None if not logo_url or not logo_url.strip(): return None filename = _safe_basename(logo_url) if filename is None: logger.warning("[reports/logo] rejected unsafe logo filename for tenant %s", tenant_id) return None is_svg = Path(filename).suffix.lower() == ".svg" # Cache on the URL itself, so re-uploading a new logo under the same # filename still busts the entry. url_digest = hashlib.sha256(logo_url.encode()).hexdigest()[:16] cache_key = f"{_CACHE_PREFIX}:{tenant_id}:{url_digest}" cached = await _cache_get(cache_key) if cached: return LogoAsset(data=cached, is_svg=is_svg) raw = _read_local(tenant_id, filename) if raw is None and settings.AWS_ACCESS_KEY_ID and settings.AWS_SECRET_ACCESS_KEY: key = f"tenants/{tenant_id}/logo/{filename}" raw = await asyncio.to_thread(_fetch_s3_sync, key) if not raw: return None payload = raw if is_svg else _normalise_raster(raw) if not payload: return None await _cache_set(cache_key, payload) return LogoAsset(data=payload, is_svg=is_svg) def render_logo_flowable(logo: LogoAsset, *, max_height: float): """Build a reportlab flowable for a tenant logo, or None if it won't render. Raster logos become an ``Image``; SVGs are rasterised through svglib. Both are scaled to ``max_height`` with aspect ratio preserved, so a very wide logo shrinks rather than shoving surrounding header text out of place. Shared by every PDF generator that renders a tenant letterhead (reports export, sales proposal, sales contract/invoice) so logo-rendering stays in one place next to logo-loading. Returns None on any rendering problem — a logo is decoration, and must never be the reason a PDF export fails. """ import io try: if logo.is_svg: from svglib.svglib import svg2rlg drawing = svg2rlg(io.BytesIO(logo.data)) if drawing is None or not drawing.height: return None scale = min(max_height / drawing.height, 1.0) drawing.scale(scale, scale) drawing.width *= scale drawing.height *= scale return drawing from reportlab.lib.utils import ImageReader from reportlab.platypus import Image as RLImage reader = ImageReader(io.BytesIO(logo.data)) px_w, px_h = reader.getSize() if not px_h or not px_w: return None width = max_height * (px_w / px_h) return RLImage(io.BytesIO(logo.data), width=width, height=max_height) except Exception as exc: logger.warning("[branding/logo] could not render tenant logo: %s", exc) return None async def load_report_header(db: AsyncSession, tenant_id: str) -> ReportHeader: """Assemble the report PDF header: logo plus the cemetery's name/address/email. Reads the identity fields straight off ``accounts`` (the same values shown in Settings → Cemetery Profile), preferring the branding override ``public_site_name`` for the display name when one is set. Never raises — on any failure the caller gets an empty ReportHeader and the PDF falls back to a plain title-only header. """ name = address = email = None try: row = ( await db.execute( # Raw column select against both tables, deliberately avoiding # the ORM entities: BrandingConfig/Account carry relationships # that require the full mapper registry to be initialised. sa_text( """ SELECT a.organization_name, a.address, a.contact_email, b.public_site_name FROM accounts a LEFT JOIN branding_configs b ON b.account_id = a.id WHERE a.id = CAST(:tid AS uuid) """ ), {"tid": tenant_id}, ) ).first() if row: org_name, addr, contact_email, public_site_name = row name = (public_site_name or org_name or "").strip() or None address = (addr or "").strip() or None email = (contact_email or "").strip() or None except Exception as exc: logger.warning("[reports/header] cemetery profile lookup failed: %s", exc) return ReportHeader( logo=await load_tenant_logo(db, tenant_id), cemetery_name=name, address=address, email=email, )