mirror of
https://github.com/Matysh/houseplan-card
synced 2026-10-07 06:59:46 +00:00
fix: exact upload quota, support palette allowlist, bounded SVG reference chains (#498)
Attachment uploads stage the body as `.upload-*` under files_root and then asked the quota to count that file as stored usage *and* as the incoming size, so the last file that still fit was refused at the boundary — by bytes and by count. check_quota/dir_usage now take `exclude` for the caller's own staged file; other staged files keep counting, so two concurrent uploads can never both land past the limit. The support package copied every string key of settings.fill_colors. The schema stays open for compatibility, but the projection now keeps only the eleven slots the card defines (SUPPORT_FILL_COLOR_KEYS, pinned to src/logic.ts DEFAULT_FILL_COLORS by a test); an empty palette is omitted. The SVG local-reference walk was a recursive DFS: a flat chain of a few thousand hrefs passed every #436 bound and died with RecursionError, which the upload view turned into a 500. The walk is iterative and measures the longest chain through each node (memoised, order-independent); chains deeper than MAX_SVG_REF_DEPTH = 64 are refused as too_large, cycles stay invalid_image. Tests: quota boundaries on the validator and the HA endpoint, a barrier test for concurrent uploads, palette allowlist and TS parity, reference chains (plain, hostile id order, cycle) on the validator and the endpoint; six mutants caught by the standard runner. Issue: #498 User-Visible: yes
This commit is contained in:
@@ -13,6 +13,7 @@ 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
|
||||
@@ -34,6 +35,8 @@ 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({
|
||||
@@ -269,22 +272,7 @@ def _validate_svg(data: bytes) -> ValidatedAsset:
|
||||
if any(ref not in ids for ref in refs):
|
||||
raise DecorAssetError("invalid_image", "The SVG contains an unresolved local reference")
|
||||
|
||||
visiting: set[str] = set()
|
||||
visited: set[str] = set()
|
||||
|
||||
def _visit(node_id: str) -> None:
|
||||
if node_id in visiting:
|
||||
raise DecorAssetError("invalid_image", "The SVG contains a cyclic local reference")
|
||||
if node_id in visited:
|
||||
return
|
||||
visiting.add(node_id)
|
||||
for ref in ref_graph.get(node_id, ()):
|
||||
_visit(ref)
|
||||
visiting.remove(node_id)
|
||||
visited.add(node_id)
|
||||
|
||||
for node_id in ids:
|
||||
_visit(node_id)
|
||||
_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"))
|
||||
@@ -309,6 +297,52 @@ def _validate_svg(data: bytes) -> ValidatedAsset:
|
||||
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:
|
||||
|
||||
@@ -445,9 +445,13 @@ class HouseplanUploadView(HomeAssistantView):
|
||||
|
||||
tmp_path = temps[0]
|
||||
try:
|
||||
# The staged file already sits under files_root: hand it to
|
||||
# the quota as `incoming` only, not as stored usage too (#498).
|
||||
await hass.async_add_executor_job(
|
||||
check_quota, files_root, tmp_path.stat().st_size,
|
||||
MAX_FILES_BYTES, MAX_FILES_COUNT,
|
||||
partial(
|
||||
check_quota, files_root, tmp_path.stat().st_size,
|
||||
MAX_FILES_BYTES, MAX_FILES_COUNT, exclude=tmp_path,
|
||||
),
|
||||
)
|
||||
except QuotaError as err:
|
||||
_LOGGER.warning("House Plan upload refused: %s", err.detail)
|
||||
|
||||
@@ -198,12 +198,20 @@ class QuotaError(Exception):
|
||||
self.detail = detail
|
||||
|
||||
|
||||
def dir_usage(path: Path) -> tuple[int, int]:
|
||||
"""(bytes, files) below `path`, ignoring what we cannot read."""
|
||||
def dir_usage(path: Path, *, exclude: Path | None = None) -> tuple[int, int]:
|
||||
"""(bytes, files) below `path`, ignoring what we cannot read.
|
||||
|
||||
`exclude` is the caller's own staged upload: it already lives under `path`
|
||||
and its size arrives separately as `incoming`, so counting it here would
|
||||
charge the same bytes and the same file twice (#498). Any *other* staged
|
||||
file stays in the count — it is about to become an attachment.
|
||||
"""
|
||||
total = count = 0
|
||||
if not path.is_dir():
|
||||
return 0, 0
|
||||
for item in path.rglob("*"):
|
||||
if exclude is not None and item == exclude:
|
||||
continue
|
||||
try:
|
||||
if item.is_file():
|
||||
total += item.stat().st_size
|
||||
@@ -213,8 +221,10 @@ def dir_usage(path: Path) -> tuple[int, int]:
|
||||
return total, count
|
||||
|
||||
|
||||
def check_quota(path: Path, incoming: int, max_bytes: int, max_files: int) -> None:
|
||||
"""Raise QuotaError unless `incoming` more bytes fit.
|
||||
def check_quota(
|
||||
path: Path, incoming: int, max_bytes: int, max_files: int, *, exclude: Path | None = None,
|
||||
) -> None:
|
||||
"""Raise QuotaError unless `incoming` more bytes fit (`exclude`: see dir_usage).
|
||||
|
||||
Deliberately not an age rule. Files are never removed for getting old — that
|
||||
cost real plans twice — so the limit sits where a decision is being made
|
||||
@@ -222,7 +232,7 @@ def check_quota(path: Path, incoming: int, max_bytes: int, max_files: int) -> No
|
||||
"""
|
||||
import shutil
|
||||
|
||||
used, count = dir_usage(path)
|
||||
used, count = dir_usage(path, exclude=exclude)
|
||||
if count + 1 > max_files:
|
||||
raise QuotaError("too_many_files", f"{count} files already stored, the limit is {max_files}")
|
||||
if used + incoming > max_bytes:
|
||||
|
||||
@@ -136,6 +136,15 @@ def _custom_fill(value: object) -> dict[str, Any] | None:
|
||||
return _copy_keys(value, ("c", "a"))
|
||||
|
||||
|
||||
# The eleven palette slots the card reads (src/logic.ts DEFAULT_FILL_COLORS).
|
||||
# The config schema stays open on purpose so older or extended configs keep
|
||||
# loading; a package must not carry a key the product never defined (#498).
|
||||
SUPPORT_FILL_COLOR_KEYS = (
|
||||
"light_on", "light_off", "light_none", "temp_cold", "temp_ok", "temp_hot",
|
||||
"lqi_low", "lqi_high", "glow_base", "glow_light", "wall_fill",
|
||||
)
|
||||
|
||||
|
||||
def _global_settings(value: object) -> dict[str, Any]:
|
||||
out = _copy_keys(value, ("glow_radius_cm", "bg_color", "north_deg", "bg_mode", "sun_rays"))
|
||||
if not isinstance(value, dict):
|
||||
@@ -145,11 +154,13 @@ def _global_settings(value: object) -> dict[str, Any]:
|
||||
out["show_room_tooltip"] = show_room_tooltip
|
||||
fill_colors = value.get("fill_colors")
|
||||
if isinstance(fill_colors, dict):
|
||||
out["fill_colors"] = {
|
||||
str(key): _copy_keys(item, ("c", "a"))
|
||||
for key, item in fill_colors.items()
|
||||
if isinstance(key, str) and isinstance(item, dict)
|
||||
palette = {
|
||||
key: _copy_keys(fill_colors[key], ("c", "a"))
|
||||
for key in SUPPORT_FILL_COLOR_KEYS
|
||||
if isinstance(fill_colors.get(key), dict)
|
||||
}
|
||||
if palette:
|
||||
out["fill_colors"] = palette
|
||||
style = value.get("decor_default_style")
|
||||
if isinstance(style, dict):
|
||||
out["decor_default_style"] = _copy_keys(
|
||||
|
||||
Reference in New Issue
Block a user