pigway-frame-flow/pigway_frame_flow/renderers/images.py

import base64
import ipaddress
import socket
from io import BytesIO
from urllib.parse import urlsplit
from uuid import uuid4

from PIL import Image, ImageFilter, ImageOps, ImageStat

from pigway_frame_flow.errors import ErrorCode as E
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.storage.database import now_iso

Image.MAX_IMAGE_PIXELS = 25_000_000


def convert(data, fit="contain", profile="color"):
    try:
        with Image.open(BytesIO(data)) as source:
            if source.format not in {"JPEG", "PNG", "WEBP"}:
                raise ValueError()
            if source.width * source.height > 25_000_000:
                raise ValueError()
            img = ImageOps.exif_transpose(source).convert("RGBA")
            canvas = Image.new("RGBA", img.size, "white")
            canvas.alpha_composite(img)
            img = canvas.convert("RGB")
            if fit == "contain":
                result = ImageOps.pad(
                    img, (1200, 1600), method=Image.Resampling.LANCZOS, color="white"
                )
            else:
                center = (0.5, 0.5)
                if fit == "smart":
                    # Lightweight edge saliency: slide an aspect-correct window
                    # over a thumbnail; retain the region with most detail.
                    thumb = img.copy()
                    thumb.thumbnail((240, 240))
                    edges = thumb.convert("L").filter(ImageFilter.FIND_EDGES)
                    w, h = edges.size
                    cw, ch = (int(h * 0.75), h) if w / h > 0.75 else (w, int(w / 0.75))
                    choices = []
                    for step in range(11):
                        x, y = int((w - cw) * step / 10), int((h - ch) * step / 10)
                        score = ImageStat.Stat(edges.crop((x, y, x + cw, y + ch))).sum[
                            0
                        ]
                        # Favor centered crops in ties, suppress border artifacts.
                        choices.append((score - abs(step - 5) * 0.01, x, y))
                    _, x, y = max(choices)
                    center = (
                        (x / (w - cw)) if w != cw else 0.5,
                        (y / (h - ch)) if h != ch else 0.5,
                    )
                result = ImageOps.fit(
                    img, (1200, 1600), method=Image.Resampling.LANCZOS, centering=center
                )
            if profile == "grayscale":
                result = ImageOps.grayscale(result).convert("RGB")
            out = BytesIO()
            result.save(out, "JPEG", quality=93, optimize=True)
            return out.getvalue()
    except (OSError, ValueError, Image.DecompressionBombError):
        raise FlowError(E.INVALID_REQUEST, "Unsupported or invalid image") from None


class Assets:
    def __init__(self, db, settings):
        self.db, self.settings = db, settings

    def put(self, data, name="image.jpg", kind="images"):
        from pathlib import Path

        name = Path(name).stem[:110] + ".jpg"
        asset_id = uuid4().hex
        (self.settings.data_dir / kind / (asset_id + ".jpg")).write_bytes(data)
        self.db.execute(
            "INSERT INTO assets VALUES(?,?,?,?)", (asset_id, name, kind, now_iso())
        )
        return self.info(asset_id)

    def promote(self, asset_id):
        row = self.db.one("SELECT * FROM assets WHERE id=?", (asset_id,))
        if row and row["kind"] == "cache":
            target = self.settings.data_dir / "images" / (asset_id + ".jpg")
            self.path(asset_id).replace(target)
            self.db.execute("UPDATE assets SET kind='images' WHERE id=?", (asset_id,))
        return self.info(asset_id)

    def info(self, asset_id):
        row = self.db.one("SELECT * FROM assets WHERE id=?", (asset_id,))
        if not row:
            raise FlowError(E.NOT_FOUND, "Asset not found", 404)
        return {
            "asset_id": row["id"],
            "source": "generated" if row["kind"] == "generated" else "uploaded",
            "name": row["name"],
            "url": f"/api/v1/assets/{row['id']}",
            "created_at": row["created_at"],
        }

    def path(self, asset_id):
        row = self.db.one("SELECT * FROM assets WHERE id=?", (asset_id,))
        if not row:
            raise FlowError(E.NOT_FOUND, "Asset not found", 404)
        return self.settings.data_dir / row["kind"] / (row["id"] + ".jpg")

    def decode(self, value):
        try:
            data = base64.b64decode(value, validate=True)
        except ValueError:
            raise FlowError(E.INVALID_REQUEST, "Invalid base64 image") from None
        if len(data) > self.settings.max_bytes:
            raise FlowError(E.INVALID_REQUEST, "Image exceeds 20 MiB")
        return data

    def fetch_url(
        self, url, *, method="GET", headers=None, limit=None, accepted_status=(200,)
    ):
        limit = limit or self.settings.max_bytes
        parsed = urlsplit(url)
        if (
            parsed.scheme != "https"
            or not parsed.hostname
            or parsed.username
            or parsed.password
            or parsed.port not in (None, 443)
        ):
            raise FlowError(E.INVALID_REQUEST, "Image URL requires HTTPS on port 443")
        if parsed.hostname.lower() not in self.settings.image_url_hosts:
            raise FlowError(E.FORBIDDEN, "Image host is not allowed", 403)
        try:
            addresses = socket.getaddrinfo(
                parsed.hostname, 443, type=socket.SOCK_STREAM
            )
            if not addresses or any(
                not ipaddress.ip_address(x[4][0]).is_global for x in addresses
            ):
                raise FlowError(
                    E.FORBIDDEN, "Private image addresses are forbidden", 403
                )
            # Pin the validated resolution for this request, including TLS SNI,
            # instead of performing a second DNS lookup (rebinding).
            import http.client
            import ssl

            ip = str(addresses[0][4][0])

            class PinnedHTTPS(http.client.HTTPSConnection):
                def connect(self):
                    sock = socket.create_connection((ip, 443), timeout=10)
                    self.sock = ssl.create_default_context().wrap_socket(
                        sock, server_hostname=parsed.hostname
                    )

            connection = PinnedHTTPS(parsed.hostname, timeout=10)
            try:
                connection.request(
                    method,
                    (parsed.path or "/") + ("?" + parsed.query if parsed.query else ""),
                    headers=headers or {},
                )
                response = connection.getresponse()
                if response.status not in accepted_status:
                    raise FlowError(
                        E.INVALID_REQUEST, "Image URL returned an invalid response"
                    )
                data = response.read(limit + 1)
                if len(data) > limit:
                    raise FlowError(E.INVALID_REQUEST, "Image exceeds 20 MiB")
                return data
            finally:
                connection.close()
        except (OSError, ValueError):
            raise FlowError(E.INVALID_REQUEST, "Unable to fetch image URL") from None