houseplan-card/custom_components/houseplan/decor_assets.py

"""Content-addressed custom images for the decorative layer (#51).

Upload plumbing hands this module untrusted bytes. It either returns one fully
validated canonical asset or rejects the whole file; SVG is never repaired or
partially stripped.
"""
from __future__ import annotations

import hashlib
import json
import math
import re
import stat
import struct
import xml.etree.ElementTree as ET
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from xml.parsers import expat

from .const import CONTENT_URL, MAX_DECOR_ASSET_BYTES

ASSET_ID_RE = re.compile(r"^[0-9a-f]{64}$")
ASSET_EXTENSIONS = frozenset({".png", ".jpg", ".jpeg", ".webp", ".svg"})
MIME_BY_EXT = {
    ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
    ".webp": "image/webp", ".svg": "image/svg+xml",
}
EXT_BY_MIME = {
    "image/png": ".png", "image/jpeg": ".jpg", "image/webp": ".webp",
    "image/svg+xml": ".svg",
}
MAX_RASTER_DIMENSION = 16_384
MAX_RASTER_PIXELS = (128 * 1024 * 1024) // 4
MAX_SVG_ELEMENTS = 5_000
MAX_SVG_DEPTH = 64
# Rendering follows href/url() chains; a chain this long is not art (#498).
MAX_SVG_REF_DEPTH = 64
MAX_SVG_ATTR_CHARS = 512_000
MAX_SVG_ATTR_VALUE_CHARS = 65_536
_SVG_TAGS = frozenset({
    "svg", "g", "defs", "title", "desc", "path", "rect", "circle", "ellipse",
    "line", "polyline", "polygon", "clipPath", "mask", "linearGradient",
    "radialGradient", "stop",
})
_SVG_ATTRS = frozenset({
    "xmlns", "viewBox", "width", "height", "x", "y", "x1", "y1", "x2", "y2",
    "cx", "cy", "r", "rx", "ry", "d", "points", "transform", "fill", "fill-rule",
    "fill-opacity", "stroke", "stroke-width", "stroke-linecap", "stroke-linejoin",
    "stroke-dasharray", "stroke-dashoffset", "stroke-opacity", "opacity", "offset",
    "stop-color", "stop-opacity", "gradientUnits", "gradientTransform", "id",
    "clip-path", "mask", "preserveAspectRatio", "href",
})
_LOCAL_REF = re.compile(r"^url\(#[A-Za-z_][A-Za-z0-9_.:-]*\)$")
_LENGTH = re.compile(r"^\s*([0-9]+(?:\.[0-9]+)?)\s*(?:px)?\s*$", re.I)
_SVG_NAMESPACE = "http://www.w3.org/2000/svg"
_SVG_UNIT_INTERVAL_ATTRS = frozenset({
    "opacity", "fill-opacity", "stroke-opacity", "stop-opacity", "offset",
})


class DecorAssetError(ValueError):
    """Stable validation failure suitable for an API error response."""

    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code


@dataclass(frozen=True)
class ValidatedAsset:
    data: bytes
    mime: str
    ext: str
    width: int
    height: int

    @property
    def asset_id(self) -> str:
        return hashlib.sha256(self.data).hexdigest()


def _check_size(data: bytes) -> None:
    if not data:
        raise DecorAssetError("invalid_image", "The image is empty")
    if len(data) > MAX_DECOR_ASSET_BYTES:
        raise DecorAssetError("too_large", "The image exceeds the 2 MiB limit")


def _raster_dimensions(data: bytes, ext: str) -> tuple[int, int]:
    try:
        if ext == ".png":
            if (len(data) < 45 or data[:8] != b"\x89PNG\r\n\x1a\n"
                    or data[12:16] != b"IHDR" or b"IDAT" not in data
                    or data[-8:-4] != b"IEND"):
                raise ValueError
            return struct.unpack(">II", data[16:24])
        if ext in (".jpg", ".jpeg"):
            if len(data) < 4 or data[:2] != b"\xff\xd8":
                raise ValueError
            pos = 2
            while pos + 4 <= len(data):
                if data[pos] != 0xFF:
                    pos += 1
                    continue
                marker = data[pos + 1]
                pos += 2
                if marker in (0xD8, 0xD9) or 0xD0 <= marker <= 0xD7:
                    continue
                length = int.from_bytes(data[pos:pos + 2], "big")
                if length < 2 or pos + length > len(data):
                    raise ValueError
                if marker in {0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF}:
                    if length < 7:
                        raise ValueError
                    return (
                        int.from_bytes(data[pos + 5:pos + 7], "big"),
                        int.from_bytes(data[pos + 3:pos + 5], "big"),
                    )
                pos += length
            raise ValueError
        if ext == ".webp":
            if len(data) < 30 or data[:4] != b"RIFF" or data[8:12] != b"WEBP":
                raise ValueError
            kind = data[12:16]
            if kind == b"VP8X":
                return (
                    1 + int.from_bytes(data[24:27], "little"),
                    1 + int.from_bytes(data[27:30], "little"),
                )
            if kind == b"VP8 ":
                idx = data.find(b"\x9d\x01\x2a", 20)
                if idx < 0 or idx + 7 > len(data):
                    raise ValueError
                return (
                    int.from_bytes(data[idx + 3:idx + 5], "little") & 0x3FFF,
                    int.from_bytes(data[idx + 5:idx + 7], "little") & 0x3FFF,
                )
            if kind == b"VP8L" and data[20] == 0x2F:
                bits = int.from_bytes(data[21:25], "little")
                return (bits & 0x3FFF) + 1, ((bits >> 14) & 0x3FFF) + 1
            raise ValueError
    except (IndexError, struct.error, ValueError) as err:
        raise DecorAssetError(
            "invalid_image", "The raster image is corrupt or has the wrong type",
        ) from err
    raise DecorAssetError("unsupported_image", "Unsupported image type")


def _validate_dimensions(width: float, height: float) -> tuple[int, int]:
    if not math.isfinite(width) or not math.isfinite(height) or width <= 0 or height <= 0:
        raise DecorAssetError("invalid_image", "The image has no usable dimensions")
    w, h = int(round(width)), int(round(height))
    if w <= 0 or h <= 0:
        raise DecorAssetError("invalid_image", "The image has no usable dimensions")
    if w > MAX_RASTER_DIMENSION or h > MAX_RASTER_DIMENSION or w * h > MAX_RASTER_PIXELS:
        raise DecorAssetError("too_large", "Decoded image dimensions exceed the safety limit")
    return w, h


def _svg_number(value: str | None) -> float | None:
    if value is None:
        return None
    match = _LENGTH.fullmatch(value)
    return float(match.group(1)) if match else None


def _validate_svg_unit_interval(name: str, value: str) -> None:
    """Reject non-finite and out-of-range opacity/gradient values."""
    raw = value.strip()
    percent = raw.endswith("%")
    try:
        number = float(raw[:-1] if percent else raw)
    except ValueError as err:
        raise DecorAssetError("invalid_image", f"The SVG {name} value is invalid") from err
    limit = 100 if percent else 1
    if not math.isfinite(number) or number < 0 or number > limit:
        raise DecorAssetError("invalid_image", f"The SVG {name} value is out of range")


def _reject_svg_declarations(data: bytes) -> None:
    """Parse with encoding-aware expat guards before building an XML tree."""
    parser = expat.ParserCreate()

    def reject(*_args: Any) -> None:
        raise DecorAssetError(
            "invalid_image", "DTD, entities and processing instructions are forbidden",
        )

    parser.StartDoctypeDeclHandler = reject
    parser.EntityDeclHandler = reject
    parser.ExternalEntityRefHandler = reject
    parser.ProcessingInstructionHandler = reject
    try:
        parser.Parse(data, True)
    except DecorAssetError:
        raise
    except (expat.ExpatError, UnicodeError) as err:
        raise DecorAssetError("invalid_image", "The SVG is not valid XML") from err


def _validate_svg(data: bytes) -> ValidatedAsset:
    # Byte-substring checks are bypassable with UTF-16/UTF-32. Expat detects
    # the declared/input encoding first and our handlers reject declarations
    # before entity expansion can allocate an ElementTree.
    _reject_svg_declarations(data)
    try:
        root = ET.fromstring(data)
    except (ET.ParseError, UnicodeError) as err:
        raise DecorAssetError("invalid_image", "The SVG is not valid XML") from err
    if root.tag.rsplit("}", 1)[-1] != "svg":
        raise DecorAssetError("invalid_image", "Only an SVG root is accepted")
    count = attr_chars = 0
    ids: set[str] = set()
    refs: list[str] = []
    ref_graph: dict[str, set[str]] = {}
    stack: list[tuple[ET.Element, int, str | None]] = [(root, 1, None)]
    while stack:
        node, depth, inherited_owner = stack.pop()
        count += 1
        if count > MAX_SVG_ELEMENTS or depth > MAX_SVG_DEPTH:
            raise DecorAssetError("too_large", "The SVG structure exceeds the safety limit")
        if node.tag.startswith("{"):
            namespace, tag = node.tag[1:].split("}", 1)
            if namespace != _SVG_NAMESPACE:
                raise DecorAssetError("unsupported_image", "Unknown SVG namespace")
        else:
            raise DecorAssetError("unsupported_image", "The SVG namespace is required")
        if tag not in _SVG_TAGS:
            raise DecorAssetError("unsupported_image", f"Unsupported SVG element: {tag}")
        node_id = node.attrib.get("id")
        owner = node_id or inherited_owner
        if node.text and node.text.strip():
            if tag not in {"title", "desc"} or len(node.text) > 4096:
                raise DecorAssetError("unsupported_image", "Unsupported or oversized SVG text")
        for raw_name, value in node.attrib.items():
            if len(value) > MAX_SVG_ATTR_VALUE_CHARS:
                raise DecorAssetError("too_large", "An SVG attribute exceeds the safety limit")
            if raw_name.startswith("{"):
                raise DecorAssetError("unsupported_image", "Namespaced SVG attributes are unsupported")
            name = raw_name
            attr_chars += len(raw_name) + len(value)
            if attr_chars > MAX_SVG_ATTR_CHARS:
                raise DecorAssetError("too_large", "The SVG attributes exceed the safety limit")
            if name.lower().startswith("on") or name not in _SVG_ATTRS:
                raise DecorAssetError("unsupported_image", f"Unsupported SVG attribute: {name}")
            if name in _SVG_UNIT_INTERVAL_ATTRS:
                _validate_svg_unit_interval(name, value)
            low = value.strip().lower()
            if any(token in low for token in ("javascript:", "data:", "http:", "https:", "//")):
                raise DecorAssetError("invalid_image", "External SVG resources are forbidden")
            if "url(" in low:
                if not _LOCAL_REF.fullmatch(value.strip()):
                    raise DecorAssetError("invalid_image", "Only local SVG references are allowed")
                ref = value.strip()[5:-1]
                refs.append(ref)
                if owner:
                    ref_graph.setdefault(owner, set()).add(ref)
            if name == "href":
                if not re.fullmatch(r"#[A-Za-z_][A-Za-z0-9_.:-]*", value.strip()):
                    raise DecorAssetError("invalid_image", "Only local SVG references are allowed")
                ref = value.strip()[1:]
                refs.append(ref)
                if owner:
                    ref_graph.setdefault(owner, set()).add(ref)
            if name == "id":
                if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_.:-]*", value) or value in ids:
                    raise DecorAssetError("invalid_image", "SVG ids must be unique and well formed")
                ids.add(value)
        stack.extend((child, depth + 1, owner) for child in node)
    if any(ref not in ids for ref in refs):
        raise DecorAssetError("invalid_image", "The SVG contains an unresolved local reference")

    _walk_reference_graph(ids, ref_graph)
    view_box = root.attrib.get("viewBox")
    width = _svg_number(root.attrib.get("width"))
    height = _svg_number(root.attrib.get("height"))
    if view_box:
        try:
            parts = [float(part) for part in re.split(r"[\s,]+", view_box.strip())]
            if len(parts) != 4:
                raise ValueError
            width, height = parts[2], parts[3]
        except ValueError as err:
            raise DecorAssetError("invalid_image", "The SVG viewBox is invalid") from err
    w, h = _validate_dimensions(width or 0, height or 0)
    ET.register_namespace("", _SVG_NAMESPACE)
    canonical = ET.tostring(
        root, encoding="utf-8", xml_declaration=False, short_empty_elements=True,
    )
    _check_size(canonical)
    try:
        ET.fromstring(canonical)
    except ET.ParseError as err:  # pragma: no cover - serializer invariant
        raise DecorAssetError("invalid_image", "Canonical SVG could not be reparsed") from err
    return ValidatedAsset(canonical, "image/svg+xml", ".svg", w, h)


def _walk_reference_graph(ids: set[str], ref_graph: dict[str, set[str]]) -> None:
    """Reject cycles and chains of local references longer than MAX_SVG_REF_DEPTH.

    Iterative on purpose: a flat chain of a few thousand `href`s passes every
    element/depth/attribute bound yet used to blow the interpreter's recursion
    limit inside a recursive DFS, which the upload view answered with a 500
    instead of a refusal (#498).

    The limit is the longest chain *through* a node, memoised per node, not the
    stack height of whichever traversal happened to reach it first: a chain
    cut into short segments by a hostile `id` order must still be measured
    end to end (spec review #498 r1).
    """
    longest: dict[str, int] = {}
    visiting: set[str] = set()
    for start in sorted(ids):
        if start in longest:
            continue
        stack: list[tuple[str, Iterator[str], int]] = [
            (start, iter(sorted(ref_graph.get(start, ()))), 1),
        ]
        visiting.add(start)
        while stack:
            node_id, children, chain = stack[-1]
            ref = next(children, None)
            if ref is None:
                stack.pop()
                visiting.discard(node_id)
                longest[node_id] = chain
                if chain > MAX_SVG_REF_DEPTH:
                    raise DecorAssetError("too_large", "The SVG reference chain exceeds the safety limit")
                if stack:
                    parent, parent_children, parent_chain = stack[-1]
                    stack[-1] = (parent, parent_children, max(parent_chain, chain + 1))
                continue
            if ref in visiting:
                raise DecorAssetError("invalid_image", "The SVG contains a cyclic local reference")
            if ref in longest:
                stack[-1] = (node_id, children, max(chain, longest[ref] + 1))
                continue
            if len(stack) > MAX_SVG_REF_DEPTH:
                raise DecorAssetError("too_large", "The SVG reference chain exceeds the safety limit")
            visiting.add(ref)
            stack.append((ref, iter(sorted(ref_graph.get(ref, ()))), 1))


def validate_asset(
    data: bytes, filename: str, declared_mime: str | None = None,
) -> ValidatedAsset:
    """Validate and canonicalise one complete upload."""
    _check_size(data)
    ext = Path(filename).suffix.lower()
    if ext not in ASSET_EXTENSIONS:
        raise DecorAssetError("unsupported_image", "Use PNG, JPEG, WebP or SVG")
    expected_mime = MIME_BY_EXT[ext]
    claimed = str(declared_mime or "").split(";", 1)[0].strip().lower()
    if claimed and claimed != "application/octet-stream" and claimed != expected_mime:
        raise DecorAssetError("invalid_format", "The filename, MIME type and image bytes disagree")
    if ext == ".svg":
        return _validate_svg(data)
    width, height = _validate_dimensions(*_raster_dimensions(data, ext))
    # Home Assistant ships Pillow. Header parsing above keeps this helper
    # independently testable, while a full decode here catches valid-looking
    # but truncated/corrupt payloads before they enter the authenticated store.
    try:
        from io import BytesIO

        from PIL import Image

        with Image.open(BytesIO(data)) as image:
            image.load()
            if image.size != (width, height):
                raise DecorAssetError("invalid_image", "Image dimensions are inconsistent")
            if getattr(image, "is_animated", False):
                raise DecorAssetError("unsupported_image", "Animated images are unsupported")
    except ImportError:
        # The Home Assistant runtime includes Pillow. Keeping the pure parser
        # usable without the full HA dependency set lets repository-local
        # security tests still exercise signatures, dimensions and SVG.
        pass
    except DecorAssetError:
        raise
    except Exception as err:  # noqa: BLE001 - decoder failures are one public error
        raise DecorAssetError("invalid_image", "The raster image cannot be decoded") from err
    mime = expected_mime
    return ValidatedAsset(data, mime, EXT_BY_MIME[mime], width, height)


def asset_refs(config: dict[str, Any] | None) -> dict[str, list[dict[str, str]]]:
    """Return server-authoritative references grouped by asset id."""
    out: dict[str, list[dict[str, str]]] = {}
    for space in (config or {}).get("spaces") or []:
        if not isinstance(space, dict):
            continue
        sid = str(space.get("id") or "")
        for shape in space.get("decor") or []:
            if not isinstance(shape, dict) or shape.get("kind") != "image":
                continue
            aid = shape.get("asset_id")
            if isinstance(aid, str) and ASSET_ID_RE.fullmatch(aid):
                out.setdefault(aid, []).append({
                    "space_id": sid, "decor_id": str(shape.get("id") or ""),
                })
    return out


def asset_meta_path(root: Path, asset_id: str) -> Path:
    return root / f"{asset_id}.json"


def _physical_asset_inventory(root: Path) -> list[tuple[Path, int]]:
    """Observe regular promoted blobs once, tolerating concurrent removals."""
    try:
        if not root.is_dir():
            return []
        entries = list(root.iterdir())
    except OSError:
        return []

    inventory: list[tuple[Path, int]] = []
    for path in entries:
        if path.suffix not in ASSET_EXTENSIONS or not ASSET_ID_RE.fullmatch(path.stem):
            continue
        try:
            current = path.stat(follow_symlinks=False)
        except OSError:
            continue
        if stat.S_ISREG(current.st_mode):
            inventory.append((path, current.st_size))
    return sorted(inventory, key=lambda item: item[0].name)


def physical_asset_blobs(root: Path) -> list[Path]:
    """Return exact promoted blob files, independently from their sidecars."""
    return [path for path, _size in _physical_asset_inventory(root)]


def physical_asset_usage(root: Path) -> tuple[int, int]:
    """Return promoted blob count and actual bytes used for quota checks."""
    inventory = _physical_asset_inventory(root)
    return len(inventory), sum(size for _path, size in inventory)


def _read_catalog_row(root: Path, path: Path) -> dict[str, Any] | None:
    """Read one sidecar through the validation shared by list and resolve."""
    try:
        row = json.loads(path.read_text(encoding="utf-8"))
        if not isinstance(row, dict):
            return None
        aid = str(row.get("asset_id") or "")
        ext = row.get("ext")
        blob = root / f"{aid}{ext}"
        if (
            path.stem != aid
            or not ASSET_ID_RE.fullmatch(aid)
            or ext not in ASSET_EXTENSIONS
            or not blob.is_file()
        ):
            return None
        return row
    except (OSError, ValueError, TypeError):
        return None


def read_asset(root: Path, asset_id: str) -> dict[str, Any] | None:
    """Read one exact catalog row without scanning unrelated sidecars."""
    if not ASSET_ID_RE.fullmatch(asset_id):
        return None
    return _read_catalog_row(root, asset_meta_path(root, asset_id))


def read_catalog(root: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    if not root.is_dir():
        return rows
    for path in root.glob("*.json"):
        row = _read_catalog_row(root, path)
        if row is not None:
            rows.append(row)
    return sorted(
        rows,
        key=lambda row: (str(row.get("created_at", "")), str(row["asset_id"])),
        reverse=True,
    )


def public_asset(
    row: dict[str, Any], used_by: list[dict[str, str]] | None = None,
) -> dict[str, Any]:
    aid = str(row["asset_id"])
    ext = str(row["ext"])
    return {
        "asset_id": aid,
        "name": row.get("name", "image"),
        "mime": row.get("mime"),
        "width": row.get("width"),
        "height": row.get("height"),
        "bytes": row.get("bytes"),
        "url": f"{CONTENT_URL}/assets/_/{aid}{ext}",
        "used_by": used_by or [],
    }