Blender add-on: importGltf, setCamera, setLight, setWorld, renderStill #2

Merged
libby merged 6 commits from libby/blender-scene-actions into main 2026-11-07 15:13:27 +00:00
7 changed files with 1026 additions and 0 deletions

View file

@ -215,6 +215,11 @@ All Blender writes use `brokkr.blender.command_intent.v0` and receive
- `setObjectCustomProperty`
- `selectObject`
- `assignMaterial`
- `importGltf`
- `setCamera`
- `setLight`
- `setWorld`
- `renderStill`
Blender owns the mutation. Brokkr advertises the command surface; Verse clients
write typed command intents; Blender executes recognized intents and publishes

View file

@ -61,6 +61,20 @@ class BrokkrPreferences(bpy.types.AddonPreferences):
description="Optional JSON debug export/import root; CultMesh remains the mirror owner",
)
import_dir: bpy.props.StringProperty(
name="Import Directory",
subtype="DIR_PATH",
default="",
description="importGltf may only read files inside this directory; empty disables importGltf",
)
output_dir: bpy.props.StringProperty(
name="Render Output Directory",
subtype="DIR_PATH",
default="",
description="renderStill may only write files inside this directory; empty disables renderStill",
)
auto_capture: bpy.props.BoolProperty(
name="Auto Capture",
default=False,
@ -270,6 +284,8 @@ class BrokkrPreferences(bpy.types.AddonPreferences):
layout.prop(self, "cultmesh_cache_path")
layout.prop(self, "cultlib_py_src")
layout.prop(self, "debug_mirror_root")
layout.prop(self, "import_dir")
layout.prop(self, "output_dir")
layout.prop(self, "auto_capture")
layout.prop(self, "auto_drain_commands")
layout.prop(self, "serve_host")
@ -434,6 +450,7 @@ class BROKKR_OT_drain_commands(bpy.types.Operator):
def execute(self, context):
prefs = context.preferences.addons[__name__].preferences
target().configure_action_dirs(prefs.import_dir, prefs.output_dir)
receipts = target().drain_commands(
context,
prefs.cultmesh_cache_path,
@ -616,6 +633,7 @@ def _auto_drain_commands():
return 1.0
try:
adapter.configure_action_dirs(prefs.import_dir, prefs.output_dir)
adapter.drain_commands(
context,
prefs.cultmesh_cache_path,

View file

@ -9,6 +9,7 @@ from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
from . import scene_actions as sa
from .server_guard import BindRefused, SessionGuard, check_bind, guarded_server_class, is_loopback_host, purge_command_intents
PROVIDER_ID = "brokkr.blender_editor"
@ -44,6 +45,11 @@ CAPABILITIES = (
"material.assign",
"selection.read",
"selection.write",
"asset.import.gltf",
"camera.write",
"light.write",
"world.write",
"render.still",
"command.receipt.publish",
)
@ -65,6 +71,13 @@ class BrokkrBlenderTarget:
self._guard: SessionGuard | None = None
self._cultmesh_module: str = ""
self._debug_mirror_root: str = ""
self.import_dir: str = ""
self.output_dir: str = ""
def configure_action_dirs(self, import_dir: str, output_dir: str) -> None:
"""The directories importGltf and renderStill may touch; empty leaves that action disabled."""
self.import_dir = self.bpy.path.abspath(import_dir) if import_dir else ""
self.output_dir = self.bpy.path.abspath(output_dir) if output_dir else ""
def resolve_cache_path(self, cache_path: str) -> str:
return self.bpy.path.abspath(cache_path or DEFAULT_CULTMESH_CACHE_PATH)
@ -467,7 +480,19 @@ class BrokkrBlenderTarget:
return self._select_object(context, command)
if action == "assignMaterial":
return self._assign_material(context, command)
if action == "importGltf":
return self._import_gltf(context, command)
if action == "setCamera":
return self._set_camera(context, command)
if action == "setLight":
return self._set_light(context, command)
if action == "setWorld":
return self._set_world(context, command)
if action == "renderStill":
return self._render_still(context, command)
return self._receipt(command, "failed", f"Unsupported Blender command action: {action}", "")
except sa.ActionRefused as error:
return self._receipt(command, "failed", str(error), "")
except Exception as error:
return self._receipt(command, "failed", str(error), "")
@ -568,6 +593,152 @@ class BrokkrBlenderTarget:
return self._receipt(command, "accepted", "Blender material assigned.", obj.name)
return self._receipt(command, "failed", "Target object cannot receive materials.", obj.name)
def _import_gltf(self, context: Any, command: dict[str, Any]) -> dict[str, Any]:
name = str(command.get("filepath") or "")
path = sa.resolve_inside(self.import_dir, name, "import")
sa.require_gltf(path)
stamped = sa.fingerprint([path]) # before the parse, so a swap during it shows
references = sa.gltf_references(path)
existing = references # gltf_references refuses a missing one
checked = (stamped, sa.fingerprint(existing))
# Re-check immediately before use: the same path, the same files, the same bytes' stats.
# Blender then opens the files itself, so a swap in that last gap is not prevented
# (residual TOCTOU): the import directory must only be writable by trusted users.
if sa.resolve_inside(self.import_dir, name, "import") != path or \
(sa.fingerprint([path]), sa.fingerprint(existing)) != checked:
raise sa.ActionRefused("Import file changed while it was being checked.")
before = set(self.bpy.data.objects.keys())
self.bpy.ops.import_scene.gltf(filepath=str(path))
added = sorted(set(self.bpy.data.objects.keys()) - before)
return self._receipt(command, "accepted", f"Imported {len(added)} object(s) from {path.name}.", added[0] if added else "")
def _resolve_typed(self, command: dict[str, Any], object_type: str) -> Any:
obj = self._resolve_object(command)
if obj is None:
raise sa.ActionRefused("Target Blender object was not found.")
if obj.type != object_type:
raise sa.ActionRefused(f"Target Blender object is not a {object_type}.")
return obj
def _set_camera(self, context: Any, command: dict[str, Any]) -> dict[str, Any]:
obj = self._resolve_typed(command, "CAMERA")
camera = obj.data
# Validate everything before changing anything, so a refused intent leaves the camera alone.
lens = sa.number(command, "lens", 1.0, sa.MAX_LENS_MM)
camera_type = sa.choice(command, "cameraType", sa.CAMERA_TYPES)
ortho_scale = sa.number(command, "orthoScale", 0.001, 100000.0)
clip_start = sa.number(command, "clipStart", 0.0001, 1.0e6)
clip_end = sa.number(command, "clipEnd", 0.001, 1.0e7)
panorama_type = sa.choice(command, "panoramaType", self._panorama_types(camera))
if lens is not None:
camera.lens = lens
if camera_type is not None:
camera.type = camera_type
if ortho_scale is not None:
camera.ortho_scale = ortho_scale
if clip_start is not None:
camera.clip_start = clip_start
if clip_end is not None:
camera.clip_end = clip_end
if panorama_type is not None:
# Blender 4.2 moved panorama_type from camera.cycles to the camera itself.
holder = camera if hasattr(camera, "panorama_type") else camera.cycles
holder.panorama_type = panorama_type
if command.get("active"):
context.scene.camera = obj
return self._receipt(command, "accepted", "Blender camera updated.", obj.name)
def _panorama_types(self, camera: Any) -> tuple[str, ...]:
holder = camera if hasattr(camera, "panorama_type") else getattr(camera, "cycles", camera)
try:
return tuple(item.identifier for item in holder.bl_rna.properties["panorama_type"].enum_items)
except (KeyError, AttributeError):
return ("EQUIRECTANGULAR", "EQUIDISTANT", "FISHEYE_EQUISOLID", "MIRRORBALL")
def _set_light(self, context: Any, command: dict[str, Any]) -> dict[str, Any]:
obj = self._resolve_typed(command, "LIGHT")
energy = sa.number(command, "energy", 0.0, 1.0e7)
light_color = sa.color(command)
light_type = sa.choice(command, "lightType", sa.LIGHT_TYPES)
if energy is not None:
obj.data.energy = energy
if light_color is not None:
obj.data.color = light_color
if light_type is not None:
obj.data.type = light_type
return self._receipt(command, "accepted", "Blender light updated.", obj.name)
def _set_world(self, context: Any, command: dict[str, Any]) -> dict[str, Any]:
world_color = sa.color(command)
strength = sa.number(command, "strength", 0.0, 1.0e6)
name = str(command.get("worldName") or "")
scene = context.scene
world = self.bpy.data.worlds.get(name) if name else scene.world
if world is None:
world = self.bpy.data.worlds.new(name or "Brokkr World")
scene.world = world
world.use_nodes = True
background = world.node_tree.nodes.get("Background")
if background is None:
raise sa.ActionRefused("World has no Background node to configure.")
if world_color is not None:
background.inputs["Color"].default_value = (*world_color, 1.0)
if strength is not None:
background.inputs["Strength"].default_value = strength
return self._receipt(command, "accepted", "Blender world updated.", world.name)
def _render_still(self, context: Any, command: dict[str, Any]) -> dict[str, Any]:
scene = context.scene
render = scene.render
name = str(command.get("outputPath") or "")
path = sa.resolve_inside(self.output_dir, name, "output", new_file=True)
file_format = sa.still_format(path)
# The engine enum is dynamic (Cycles and add-on engines register at runtime), and RNA's
# enum_items on a dynamic enum lists only its static defaults, so the running Blender is
# asked by assignment below and a TypeError there is the refusal.
engine = command.get("engine")
if engine is not None and (not isinstance(engine, str) or not engine.strip()):
raise sa.ActionRefused("engine must be a render engine identifier.")
engine = engine.strip().upper() if engine else None
width = sa.integer(command, "resolutionX", 1, sa.MAX_RESOLUTION)
height = sa.integer(command, "resolutionY", 1, sa.MAX_RESOLUTION)
camera_name = str(command.get("cameraName") or "")
camera = None
if camera_name:
camera = self.bpy.data.objects.get(camera_name)
if camera is None or camera.type != "CAMERA":
raise sa.ActionRefused(f"Camera object was not found: {camera_name}")
# Render settings are the intent's for this render only; put them back afterwards.
saved = (render.engine, render.resolution_x, render.resolution_y, render.filepath,
render.image_settings.file_format, scene.camera)
try:
if engine is not None:
try:
render.engine = engine
except TypeError:
raise sa.ActionRefused(f"engine {engine} is not available in this Blender.") from None
if width is not None:
render.resolution_x = width
if height is not None:
render.resolution_y = height
if camera is not None:
scene.camera = camera
render.filepath = str(path)
render.image_settings.file_format = file_format
# Folders are made only after every refusal that can still come from Blender (an unknown
# engine), so a refused render leaves none behind. Then re-check after the folders exist
# and immediately before use: no link was swapped in, and the name is still free. Blender
# opens the file itself afterwards, so a swap in that last gap is not prevented (residual
# TOCTOU): the output directory must only be writable by trusted users.
path.parent.mkdir(parents=True, exist_ok=True)
if sa.resolve_inside(self.output_dir, name, "output", new_file=True) != path:
raise sa.ActionRefused("Output path changed while it was being checked.")
self.bpy.ops.render.render(write_still=True)
finally:
(render.engine, render.resolution_x, render.resolution_y, render.filepath,
render.image_settings.file_format, scene.camera) = saved
return self._receipt(command, "accepted", f"Rendered still to {path.name}.", scene.name)
def _resolve_object(self, command: dict[str, Any]) -> Any:
name = command.get("targetObjectName") or command.get("objectName")
return self.bpy.data.objects.get(name) if name else None

View file

@ -0,0 +1,226 @@
"""Validation for the Blender scene command actions. Stdlib only, no bpy.
Everything an intent can say is checked here before bpy is touched, so the rules
are testable without Blender. Failures raise ActionRefused; the target turns the
message into a failed receipt.
"""
from __future__ import annotations
import json
import math
import os
import struct
from pathlib import Path
from typing import Any, Iterable
from urllib.parse import unquote
GLTF_EXTENSIONS = (".gltf", ".glb")
STILL_FORMATS = {".png": "PNG", ".jpg": "JPEG", ".jpeg": "JPEG", ".exr": "OPEN_EXR"}
CAMERA_TYPES = ("PERSP", "ORTHO", "PANO")
LIGHT_TYPES = ("POINT", "SUN", "SPOT", "AREA")
MAX_RESOLUTION = 16384
MAX_LENS_MM = 5000.0
class ActionRefused(ValueError):
"""The intent is malformed or reaches outside what the action may touch."""
WINDOWS_DEVICE_NAMES = frozenset(
["CON", "PRN", "AUX", "NUL", "CONIN$", "CONOUT$"]
+ [f"COM{i}" for i in "123456789\u00b9\u00b2\u00b3"] + [f"LPT{i}" for i in "123456789\u00b9\u00b2\u00b3"]
)
MAX_GLTF_JSON_BYTES = 16 * 1024 * 1024
def check_relative(candidate: str, label: str) -> list[str]:
"""The components of a path that is relative, plain and safe on every platform, or ActionRefused.
Checked as text, the same on Linux and Windows, because the machine that validates is
not the one whose rules bite: both / and \\ separate; a drive (C:), drive-relative
(C:foo), UNC (\\\\host), extended (\\\\?\\) or rooted path is refused; so is any ':' at all
(it also names NTFS alternate data streams, a.glb:stream), a NUL, a .. component,
a component ending in a dot or space (Windows drops them, so a. is a), and a
Windows device name (CON, NUL.txt, COM1)."""
if not candidate:
raise ActionRefused(f"{label} path is required.")
if "\0" in candidate:
raise ActionRefused(f"{label} path contains a NUL.")
if ":" in candidate:
raise ActionRefused(f"{label} path may not contain ':' (drive, scheme or stream).")
if candidate[0] in "/\\":
raise ActionRefused(f"{label} path must be relative (no leading separator, UNC or extended path).")
parts = [part for part in candidate.replace("\\", "/").split("/") if part not in ("", ".")]
if not parts:
raise ActionRefused(f"{label} path is required.")
for part in parts:
if part == "..":
raise ActionRefused(f"{label} path may not contain '..'.")
if part[-1] in ". ":
raise ActionRefused(f"{label} path component {part!r} ends in a dot or space.")
# Windows reads "CON .txt" as CON: trailing spaces before the first dot do not count.
if part.split(".")[0].rstrip(" ").upper() in WINDOWS_DEVICE_NAMES:
raise ActionRefused(f"{label} path component {part!r} is a Windows device name.")
return parts
def _same(a: Path, b: Path) -> bool:
return os.path.normcase(str(a)) == os.path.normcase(str(b))
def resolve_inside(root: str, candidate: str, label: str, *, new_file: bool = False) -> Path:
"""The path of candidate under root, or ActionRefused.
root is the directory configured in the add-on preferences; an empty root means the
action is not enabled. The candidate is relative (check_relative). No existing
component between root and the target may be a symlink or junction: each is compared
with its resolved form, which also catches junctions on Pythons without
os.path.isjunction. The resolved result must lie inside the resolved root. With
new_file the target must not exist at all, as a file, a symlink (even a dangling one)
or a hardlink, so an output can never write through an existing name."""
if not root:
raise ActionRefused(f"{label} directory is not configured in the add-on preferences.")
parts = check_relative(candidate, label)
base = Path(root).resolve()
current = base
for part in parts:
step = current / part
try:
linked = os.path.lexists(step) and not _same(step.resolve(), current / part)
except OSError:
# Windows refuses to traverse some junctions outright (WinError 448, untrusted mount point).
linked = True
if linked:
raise ActionRefused(f"{label} path goes through a link at {part!r}.")
current = step
try:
current.resolve().relative_to(base)
except ValueError:
raise ActionRefused(f"{label} path is outside the configured {label} directory.") from None
if new_file and os.path.lexists(current):
raise ActionRefused(f"{label} path already exists; renderStill never overwrites or writes through an existing name.")
return current
def fingerprint(paths: Iterable[Path]) -> tuple:
"""What a re-check compares: each path's resolved name, size and modification time."""
out = []
for path in paths:
stat = path.stat()
out.append((str(path.resolve()), stat.st_size, stat.st_mtime_ns))
return tuple(out)
def gltf_references(path: Path) -> list[Path]:
"""Every external file a .gltf or .glb names, checked to exist inside path's folder; [] for self-contained.
Parses the glTF JSON (the .gltf file, or a .glb's first chunk) and takes every
buffers[].uri and images[].uri. A data: URI is inline and fine. Anything else is
percent-decoded and must pass check_relative (so file:, http:, drive, UNC, rooted and
.. forms are refused), then resolve under the .gltf's own folder (itself already checked under the import root)
with no link on the way, and must exist."""
raw = path.read_bytes() if path.suffix.lower() == ".gltf" else _glb_json_chunk(path)
if len(raw) > MAX_GLTF_JSON_BYTES:
raise ActionRefused("glTF JSON is too large to check.")
try:
document = json.loads(raw.decode("utf-8"))
except (UnicodeDecodeError, ValueError):
raise ActionRefused("Import file is not valid glTF JSON.") from None
if not isinstance(document, dict):
raise ActionRefused("Import file is not a glTF object.")
references: list[Path] = []
here = path.parent.resolve()
for kind in ("buffers", "images"):
entries = document.get(kind, [])
if not isinstance(entries, list):
raise ActionRefused(f"glTF {kind} is not a list.")
for entry in entries:
uri = entry.get("uri") if isinstance(entry, dict) else None
if uri is None:
continue
if not isinstance(uri, str):
raise ActionRefused(f"glTF {kind} uri is not a string.")
if uri.lower().startswith("data:"):
continue
decoded = unquote(uri)
relative = "/".join(check_relative(decoded, f"glTF {kind} uri"))
# path is already under root, so here is too, and resolve_inside keeps the reference under here.
target = resolve_inside(str(here), relative, f"glTF {kind} uri")
if not target.exists():
raise ActionRefused(f"glTF {kind} uri names a file that does not exist: {relative}")
references.append(target)
return references
def _glb_json_chunk(path: Path) -> bytes:
with path.open("rb") as handle:
header = handle.read(20)
if len(header) < 20 or header[:4] != b"glTF":
raise ActionRefused("Import file is not a binary glTF.")
length, kind = struct.unpack("<I4s", header[12:20])
if kind != b"JSON" or length > MAX_GLTF_JSON_BYTES:
raise ActionRefused("Binary glTF does not start with a JSON chunk of a checkable size.")
data = handle.read(length)
if len(data) != length:
raise ActionRefused("Binary glTF is truncated.")
return data
def number(command: dict[str, Any], key: str, low: float, high: float) -> float | None:
"""A finite number in [low, high], or None when the key is absent."""
if key not in command or command[key] is None:
return None
value = command[key]
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
raise ActionRefused(f"{key} must be a finite number.")
if not low <= value <= high:
raise ActionRefused(f"{key} must be between {low} and {high}.")
return float(value)
def integer(command: dict[str, Any], key: str, low: int, high: int) -> int | None:
value = number(command, key, low, high)
if value is None:
return None
if value != int(value):
raise ActionRefused(f"{key} must be a whole number.")
return int(value)
def color(command: dict[str, Any], key: str = "color") -> tuple[float, float, float] | None:
"""Linear RGB, three finite numbers in [0, 1], or None when absent."""
if key not in command or command[key] is None:
return None
value = command[key]
if not isinstance(value, (list, tuple)) or len(value) != 3:
raise ActionRefused(f"{key} must be three numbers [r, g, b].")
parts = []
for item in value:
if isinstance(item, bool) or not isinstance(item, (int, float)) or not math.isfinite(item) or not 0.0 <= item <= 1.0:
raise ActionRefused(f"{key} components must be numbers between 0 and 1.")
parts.append(float(item))
return (parts[0], parts[1], parts[2])
def choice(command: dict[str, Any], key: str, allowed: Iterable[str]) -> str | None:
if key not in command or command[key] is None:
return None
allowed = tuple(allowed)
value = command[key]
if not isinstance(value, str) or value.upper() not in allowed:
raise ActionRefused(f"{key} must be one of {', '.join(allowed)}.")
return value.upper()
def still_format(path: Path) -> str:
try:
return STILL_FORMATS[path.suffix.lower()]
except KeyError:
raise ActionRefused(f"Output path must end in one of {', '.join(STILL_FORMATS)}.") from None
def require_gltf(path: Path) -> None:
if path.suffix.lower() not in GLTF_EXTENSIONS:
raise ActionRefused(f"Import path must end in {' or '.join(GLTF_EXTENSIONS)}.")
if not path.is_file():
raise ActionRefused("Import path is not an existing file.")

View file

@ -101,3 +101,55 @@ Admitted command actions:
- `setObjectCustomProperty`
- `selectObject`
- `assignMaterial`
- `importGltf`
- `setCamera`
- `setLight`
- `setWorld`
- `renderStill`
Scene action fields (all other keys are ignored; a refused intent changes nothing and
gets a `failed` receipt whose message names the rule):
- `importGltf`: `filepath`, `.gltf` or `.glb`, relative to the add-on preference
`Import Directory` (absolute paths, `..` and symlinks that resolve outside it are
refused). Empty `Import Directory` disables the action. Needs Blender's glTF add-on.
- `setCamera`: `targetObjectName` (a camera object); optional `lens` (mm),
`cameraType` (`PERSP`, `ORTHO`, `PANO`), `panoramaType` (for example
`EQUIRECTANGULAR`; read from the camera, or from `camera.cycles` before Blender 4.2),
`orthoScale`, `clipStart`, `clipEnd`, `active` (true makes it the scene camera).
Panoramic cameras render in Cycles.
- `setLight`: `targetObjectName` (a light object); optional `energy` (>= 0),
`color` (linear `[r, g, b]`, each 0 to 1), `lightType` (`POINT`, `SUN`, `SPOT`, `AREA`).
- `setWorld`: optional `worldName` (default: the scene's world, created if there is
none), `color` (`[r, g, b]`), `strength`. Edits the world's Background node.
- `renderStill`: `outputPath` relative to the preference `Render Output Directory`,
ending `.png`, `.jpg` or `.exr` (same containment rules as import); optional
`engine` (any engine of the running Blender, case-insensitive), `resolutionX`,
`resolutionY` (whole numbers, 1 to 16384), `cameraName`. Engine, resolution,
camera, file path and format are restored after the render. The render blocks
Blender's UI until it finishes.
The scene actions were run against Blender 5.2.3 LTS headless with
`surfaces/blender/tests/live/headless_smoke.py` (usage in its docstring). One real
failure it found: Blender's `engine` enum is dynamic, so `bl_rna` lists only
`BLENDER_EEVEE`; `renderStill` now asks Blender by assignment instead.
### Path rules and the residual race
Both directories take plain relative names only, judged as text the same on every OS:
no drive (`C:`, `C:foo`), UNC or `\\?\` form, leading separator, `..`, NUL, `:` anywhere
(also NTFS streams), trailing dot or space, or Windows device name. No existing component
may be a symlink or junction (Windows' "untrusted mount point" error on a junction counts
as one). `renderStill` refuses any output name that already exists, including symlinks
(even dangling) and hardlinks. `importGltf` parses the `.gltf` JSON (or a `.glb`'s first
chunk) and refuses every `buffers[].uri` / `images[].uri` that is not `data:` or a
relative path inside the import directory; `.gltf` files from the same folder tree work.
**Residual TOCTOU.** The checks run in the add-on, and the path is re-checked (resolved
name, plus size and modification time of the file and its references) immediately before
use, but Blender's own importer and renderer then open the files themselves. Someone who
can write to the import or output directory in that last gap can still swap a file or
link in. The add-on cannot close that window from Python: make both directories writable
only by trusted users, and treat the checks as protection against a malicious *intent*,
not against a hostile local writer. A same-size, same-mtime content swap is also not
detected by the re-check.

View file

@ -0,0 +1,99 @@
"""Live Blender smoke for the scene actions. Not part of the unittest run: it needs Blender.
BROKKR_RUN=<scratch dir> BROKKR_ADDON=<path to surfaces/blender/brokkr_bridge> \
blender --background --factory-startup --python surfaces/blender/tests/live/headless_smoke.py
Writes a tiny .glb into <scratch>/in, runs every scene action against real bpy, and prints
RESULT lines (accepted/failed) plus the state each action should have produced.
Windows only: also makes real mklink /J junctions (one pointing outside the root, one inside it) and
expects importGltf and renderStill to refuse paths through them, plus an existing-output refusal.
Last run: see the PR.
"""
import json, struct, sys, types, os, subprocess, shutil, traceback
from pathlib import Path
import bpy
ROOT = Path(os.environ["BROKKR_RUN"]).resolve()
ADDON = Path(os.environ["BROKKR_ADDON"]).resolve()
(ROOT / "in").mkdir(parents=True, exist_ok=True)
(ROOT / "out").mkdir(parents=True, exist_ok=True)
# tiny triangle glb
pos = struct.pack("<9f", 0,0,0, 1,0,0, 0,1,0)
gltf = {"asset": {"version": "2.0"}, "scene": 0, "scenes": [{"nodes": [0]}], "nodes": [{"mesh": 0, "name": "Tri"}],
"meshes": [{"primitives": [{"attributes": {"POSITION": 0}}]}],
"buffers": [{"byteLength": len(pos)}],
"bufferViews": [{"buffer": 0, "byteLength": len(pos), "target": 34962}],
"accessors": [{"bufferView": 0, "componentType": 5126, "count": 3, "type": "VEC3", "min": [0,0,0], "max": [1,1,0]}]}
j = json.dumps(gltf).encode(); j += b" " * (-len(j) % 4)
b = pos + b"\0" * (-len(pos) % 4)
glb = struct.pack("<4sII", b"glTF", 2, 12 + 8 + len(j) + 8 + len(b)) + struct.pack("<I4s", len(j), b"JSON") + j + struct.pack("<I4s", len(b), b"BIN\0") + b
(ROOT / "in" / "tiny.glb").write_bytes(glb)
pkg = types.ModuleType("brokkr_bridge"); pkg.__path__ = [str(ADDON)]
sys.modules["brokkr_bridge"] = pkg
from brokkr_bridge import blender_target as bt
bpy.ops.wm.read_factory_settings(use_empty=True)
target = bt.BrokkrBlenderTarget(bpy)
target.configure_action_dirs(str(ROOT / "in"), str(ROOT / "out"))
ctx = bpy.context
results = []
def run(label, **cmd):
cmd.setdefault("commandId", label)
try:
r = target.execute_command(bpy.context, cmd)
except Exception:
r = {"status": "EXCEPTION", "message": traceback.format_exc()}
print("RESULT", label, r["status"], r.get("message", "").replace("\n", " | ")[:400], flush=True)
results.append((label, r["status"]))
return r
print("BLENDER", bpy.app.version_string)
run("createCam", action="createObject", objectType="CAMERA", name="Cam", location=[0, -3, 1])
run("createLight", action="createObject", objectType="LIGHT", name="Lamp", location=[2, -2, 3])
run("importGltf", action="importGltf", filepath="tiny.glb")
print("OBJECTS", sorted(bpy.data.objects.keys()))
run("importGltf_escape", action="importGltf", filepath="../x.glb")
run("setCamera", action="setCamera", targetObjectName="Cam", lens=24, cameraType="PANO", panoramaType="EQUIRECTANGULAR", active=True, clipEnd=200)
cam = bpy.data.objects["Cam"].data
print("CAM", cam.lens, cam.type, getattr(cam, "panorama_type", None), bpy.context.scene.camera and bpy.context.scene.camera.name)
run("setLight", action="setLight", targetObjectName="Lamp", energy=500, color=[1, 0.5, 0.2], lightType="AREA")
l = bpy.data.objects["Lamp"].data
print("LIGHT", l.energy, tuple(l.color), l.type)
run("setWorld", action="setWorld", color=[0.1, 0.2, 0.4], strength=1.5)
w = bpy.context.scene.world
print("WORLD", w and w.name, w and tuple(w.node_tree.nodes["Background"].inputs["Color"].default_value), w and w.node_tree.nodes["Background"].inputs["Strength"].default_value)
sc = bpy.context.scene
print("ENGINES", [i.identifier for i in sc.render.bl_rna.properties["engine"].enum_items])
sc.cycles.samples = 1
run("renderStill_cycles_pano", action="renderStill", outputPath="pano.png", engine="CYCLES", resolutionX=64, resolutionY=32, cameraName="Cam")
print("FILE pano", (ROOT / "out" / "pano.png").exists(), (ROOT / "out" / "pano.png").stat().st_size if (ROOT / "out" / "pano.png").exists() else 0)
run("renderStill_workbench", action="renderStill", outputPath="wb/shot.jpg", engine="BLENDER_WORKBENCH", resolutionX=64, resolutionY=64)
print("FILE wb", (ROOT / "out" / "wb" / "shot.jpg").exists())
run("renderStill_escape", action="renderStill", outputPath="../evil.png")
def junction(link, target):
link = Path(link)
if link.exists() or os.path.lexists(link):
os.rmdir(link)
subprocess.run(["cmd", "/c", "mklink", "/J", str(link), str(target)], check=True, capture_output=True)
if os.name == "nt":
elsewhere = ROOT.parent / (ROOT.name + "-elsewhere"); elsewhere.mkdir(exist_ok=True)
shutil.copy(ROOT / "in" / "tiny.glb", elsewhere / "tiny.glb")
(ROOT / "in" / "realdir").mkdir(exist_ok=True); shutil.copy(ROOT / "in" / "tiny.glb", ROOT / "in" / "realdir" / "tiny.glb")
junction(ROOT / "in" / "jout", elsewhere) # junction leaving the root
junction(ROOT / "in" / "jin", ROOT / "in" / "realdir") # junction staying inside the root
junction(ROOT / "out" / "jout", elsewhere)
for label, cmd in (("junction_import_outside", dict(action="importGltf", filepath="jout/tiny.glb")),
("junction_import_inside", dict(action="importGltf", filepath="jin/tiny.glb")),
("junction_render_outside", dict(action="renderStill", outputPath="jout/shot.png", engine="BLENDER_WORKBENCH", resolutionX=8, resolutionY=8))):
r = run(label, **cmd)
print("JUNCTION_RULE", label, r["status"] == "failed" and "link" in r.get("message", ""))
print("JUNCTION_LEAK", (elsewhere / "shot.png").exists())
run("render_existing_output", action="renderStill", outputPath="wb/shot.jpg", engine="BLENDER_WORKBENCH", resolutionX=8, resolutionY=8)
os.rmdir(ROOT / "in" / "jout"); os.rmdir(ROOT / "in" / "jin"); os.rmdir(ROOT / "out" / "jout")
print("RESTORED", sc.render.engine, sc.render.resolution_x, sc.render.resolution_y, sc.camera and sc.camera.name)
print("SUMMARY", results)

View file

@ -0,0 +1,455 @@
"""Scene command action tests. No Blender: a fake bpy records what the target does to it.
Run: python3 -m unittest discover -s surfaces/blender/tests -v
"""
from __future__ import annotations
import json
import os
import struct
import sys
import tempfile
import types
import unittest
from pathlib import Path
from unittest import mock
ADDON = Path(__file__).resolve().parents[1] / "brokkr_bridge"
package = types.ModuleType("brokkr_bridge")
package.__path__ = [str(ADDON)]
sys.modules.setdefault("brokkr_bridge", package)
from brokkr_bridge import blender_target as bt # noqa: E402
from brokkr_bridge import scene_actions as sa # noqa: E402
NS = types.SimpleNamespace
def glb_bytes(document=None):
document = document or {"asset": {"version": "2.0"}}
raw = json.dumps(document).encode()
raw += b" " * (-len(raw) % 4)
return struct.pack("<4sII", b"glTF", 2, 12 + 8 + len(raw)) + struct.pack("<I4s", len(raw), b"JSON") + raw
class Enum:
def __init__(self, *names):
self.enum_items = [NS(identifier=n) for n in names]
class FakeRender:
"""Assigning an engine Blender does not have raises TypeError, as RNA enums do. The enum's
bl_rna lists only the static default, as a real dynamic enum does (Blender 5.2.3 reports
only BLENDER_EEVEE there while Cycles is installed)."""
ENGINES = ("BLENDER_EEVEE", "CYCLES", "BLENDER_WORKBENCH")
def __init__(self):
self._engine = "BLENDER_EEVEE"
self.resolution_x, self.resolution_y, self.filepath = 1920, 1080, "//old"
self.image_settings = NS(file_format="PNG")
self.bl_rna = NS(properties={"engine": Enum("BLENDER_EEVEE")})
@property
def engine(self):
return self._engine
@engine.setter
def engine(self, value):
if value not in self.ENGINES:
raise TypeError(f"enum {value!r} not found")
self._engine = value
def fake_render():
return FakeRender()
class Objects(dict):
"""bpy.data.objects: a mapping with keys() and get()."""
class FakeBpy:
def __init__(self):
self.imported = []
self.rendered = []
self.scene = NS(name="Scene", camera=None, world=None, render=fake_render())
self.camera_data = NS(lens=50.0, type="PERSP", ortho_scale=6.0, clip_start=0.1, clip_end=100.0,
panorama_type="FISHEYE_EQUISOLID",
bl_rna=NS(properties={"panorama_type": Enum("EQUIRECTANGULAR", "FISHEYE_EQUISOLID")}))
self.light_data = NS(energy=1000.0, color=(1.0, 1.0, 1.0), type="POINT")
self.cam = NS(name="Cam", type="CAMERA", data=self.camera_data)
self.lamp = NS(name="Lamp", type="LIGHT", data=self.light_data)
self.mesh = NS(name="Cube", type="MESH", data=None)
self.data = NS(
objects=Objects({o.name: o for o in (self.cam, self.lamp, self.mesh)}),
worlds=NS(get=lambda name: None, new=self.new_world),
)
self.path = NS(abspath=lambda p: p)
self.ops = NS(import_scene=NS(gltf=self.import_gltf), render=NS(render=self.render_still))
def new_world(self, name):
background = NS(inputs={"Color": NS(default_value=None), "Strength": NS(default_value=None)})
world = NS(name=name, use_nodes=False,
node_tree=NS(nodes=NS(get=lambda n: background if n == "Background" else None)))
world.background = background
return world
def import_gltf(self, filepath):
self.imported.append(filepath)
self.data.objects["Imported"] = NS(name="Imported", type="MESH", data=None)
def render_still(self, write_still):
r = self.scene.render
self.rendered.append((r.engine, r.resolution_x, r.resolution_y, r.filepath,
r.image_settings.file_format, self.scene.camera))
class ActionTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve()
(self.root / "in").mkdir()
(self.root / "out").mkdir()
(self.root / "in" / "model.glb").write_bytes(glb_bytes())
(self.root / "secret.glb").write_bytes(glb_bytes())
self.bpy = FakeBpy()
self.target = bt.BrokkrBlenderTarget(self.bpy)
self.target.configure_action_dirs(str(self.root / "in"), str(self.root / "out"))
self.context = NS(scene=self.bpy.scene)
def run_action(self, **command):
command.setdefault("commandId", "c1")
return self.target.execute_command(self.context, command)
# importGltf
def test_import_inside_the_import_dir_runs(self):
receipt = self.run_action(action="importGltf", filepath="model.glb")
self.assertEqual(receipt["status"], "accepted")
self.assertEqual(self.bpy.imported, [str(self.root / "in" / "model.glb")])
self.assertEqual(receipt["objectName"], "Imported")
def test_import_outside_the_import_dir_is_refused_before_bpy(self):
for path in ("../secret.glb", str(self.root / "secret.glb"), "/etc/passwd.glb", "sub/../../secret.glb"):
receipt = self.run_action(action="importGltf", filepath=path)
self.assertEqual(receipt["status"], "failed", path)
self.assertEqual(self.bpy.imported, [])
def test_import_through_a_symlink_that_escapes_is_refused(self):
link = self.root / "in" / "link.glb"
try:
os.symlink(self.root / "secret.glb", link)
except OSError:
self.skipTest("symlinks unavailable")
self.assertEqual(self.run_action(action="importGltf", filepath="link.glb")["status"], "failed")
self.assertEqual(self.bpy.imported, [])
def test_import_needs_a_configured_dir_a_gltf_extension_and_an_existing_file(self):
self.target.configure_action_dirs("", "")
self.assertEqual(self.run_action(action="importGltf", filepath="model.glb")["status"], "failed")
self.target.configure_action_dirs(str(self.root / "in"), "")
(self.root / "in" / "x.obj").write_text("x")
for path in ("x.obj", "missing.glb", ""):
self.assertEqual(self.run_action(action="importGltf", filepath=path)["status"], "failed", path)
self.assertEqual(self.bpy.imported, [])
# setCamera
def test_set_camera_sets_lens_type_panorama_and_active(self):
receipt = self.run_action(action="setCamera", targetObjectName="Cam", lens=35, cameraType="pano",
panoramaType="EQUIRECTANGULAR", active=True, clipEnd=500)
self.assertEqual(receipt["status"], "accepted")
d = self.bpy.camera_data
self.assertEqual((d.lens, d.type, d.panorama_type, d.clip_end), (35.0, "PANO", "EQUIRECTANGULAR", 500.0))
self.assertIs(self.bpy.scene.camera, self.bpy.cam)
def test_set_camera_falls_back_to_the_cycles_panorama_property(self):
cycles = NS(panorama_type="FISHEYE_EQUISOLID",
bl_rna=NS(properties={"panorama_type": Enum("EQUIRECTANGULAR")}))
data = NS(lens=50.0, type="PERSP", cycles=cycles)
self.bpy.data.objects["Cam"] = NS(name="Cam", type="CAMERA", data=data)
receipt = self.run_action(action="setCamera", targetObjectName="Cam", panoramaType="EQUIRECTANGULAR")
self.assertEqual(receipt["status"], "accepted")
self.assertEqual(cycles.panorama_type, "EQUIRECTANGULAR")
def test_set_camera_refused_input_changes_nothing(self):
for bad in ({"lens": -1}, {"lens": "wide"}, {"lens": float("nan")}, {"cameraType": "FISHEYE"},
{"panoramaType": "NOPE"}, {"lens": True}):
receipt = self.run_action(action="setCamera", targetObjectName="Cam", clipEnd=9, **bad)
self.assertEqual(receipt["status"], "failed", bad)
self.assertEqual(self.bpy.camera_data.clip_end, 100.0)
self.assertEqual(self.run_action(action="setCamera", targetObjectName="Lamp", lens=35)["status"], "failed")
self.assertEqual(self.run_action(action="setCamera", targetObjectName="Nope", lens=35)["status"], "failed")
# setLight
def test_set_light_energy_and_colour(self):
receipt = self.run_action(action="setLight", targetObjectName="Lamp", energy=250, color=[1, 0.5, 0])
self.assertEqual(receipt["status"], "accepted")
self.assertEqual((self.bpy.light_data.energy, self.bpy.light_data.color), (250.0, (1.0, 0.5, 0.0)))
def test_set_light_rejects_bad_values_whole(self):
for bad in ({"energy": -5}, {"color": [2, 0, 0]}, {"color": [1, 0]}, {"color": "red"}, {"lightType": "LASER"}):
self.assertEqual(self.run_action(action="setLight", targetObjectName="Lamp", **bad)["status"], "failed", bad)
self.assertEqual(self.bpy.light_data.energy, 1000.0)
self.assertEqual(self.run_action(action="setLight", targetObjectName="Cam", energy=1)["status"], "failed")
# setWorld
def test_set_world_creates_and_assigns_a_world_with_colour_and_strength(self):
receipt = self.run_action(action="setWorld", color=[0.1, 0.2, 0.3], strength=2)
self.assertEqual(receipt["status"], "accepted")
world = self.bpy.scene.world
self.assertTrue(world.use_nodes)
self.assertEqual(world.background.inputs["Color"].default_value, (0.1, 0.2, 0.3, 1.0))
self.assertEqual(world.background.inputs["Strength"].default_value, 2.0)
def test_set_world_refuses_bad_colour_before_creating_a_world(self):
self.assertEqual(self.run_action(action="setWorld", color=[9, 0, 0])["status"], "failed")
self.assertIsNone(self.bpy.scene.world)
# renderStill
def test_render_still_writes_inside_output_dir_and_restores_settings(self):
receipt = self.run_action(action="renderStill", outputPath="shots/a.png", engine="cycles",
resolutionX=640, resolutionY=360, cameraName="Cam")
self.assertEqual(receipt["status"], "accepted")
engine, x, y, filepath, fmt, camera = self.bpy.rendered[0]
self.assertEqual((engine, x, y, fmt), ("CYCLES", 640, 360, "PNG"))
self.assertEqual(filepath, str(self.root / "out" / "shots" / "a.png"))
self.assertIs(camera, self.bpy.cam)
render = self.bpy.scene.render
self.assertEqual((render.engine, render.resolution_x, render.resolution_y, render.filepath),
("BLENDER_EEVEE", 1920, 1080, "//old"))
self.assertIsNone(self.bpy.scene.camera)
def test_render_still_restores_settings_when_the_render_fails(self):
def boom(write_still):
raise RuntimeError("render failed")
self.bpy.ops.render.render = boom
self.assertEqual(self.run_action(action="renderStill", outputPath="a.png", engine="CYCLES")["status"], "failed")
self.assertEqual(self.bpy.scene.render.engine, "BLENDER_EEVEE")
def test_render_still_outside_the_output_dir_or_with_bad_input_never_renders(self):
cases = [{"outputPath": "../a.png"}, {"outputPath": str(self.root / "a.png")}, {"outputPath": "a.txt"},
{"outputPath": ""}, {"outputPath": "a.png", "engine": "NOPE"},
{"outputPath": "a.png", "resolutionX": 0}, {"outputPath": "a.png", "resolutionX": 99999},
{"outputPath": "a.png", "resolutionX": 10.5}, {"outputPath": "a.png", "cameraName": "Lamp"}]
for case in cases:
self.assertEqual(self.run_action(action="renderStill", **case)["status"], "failed", case)
self.assertEqual(self.bpy.rendered, [])
self.target.configure_action_dirs(str(self.root / "in"), "")
self.assertEqual(self.run_action(action="renderStill", outputPath="a.png")["status"], "failed")
# importGltf: names
def test_import_refuses_drive_unc_extended_stream_and_device_names(self):
for name in ("C:foo.glb", "C:\\x\\model.glb", "\\\\host\\share\\model.glb", "\\\\?\\C:\\model.glb",
"model.glb:stream", "/model.glb", "\\model.glb", "CON.glb", "sub/nul.glb", "model.glb.",
"model.glb ", "a\x00.glb", "..\\secret.glb"):
self.assertEqual(self.run_action(action="importGltf", filepath=name)["status"], "failed", name)
self.assertEqual(self.bpy.imported, [])
def test_import_refuses_a_symlinked_folder_even_when_it_points_elsewhere_inside_the_root(self):
(self.root / "in" / "real").mkdir()
(self.root / "in" / "real" / "m.glb").write_bytes(glb_bytes())
try:
os.symlink(self.root / "in" / "real", self.root / "in" / "alias", target_is_directory=True)
os.symlink(self.root, self.root / "in" / "up", target_is_directory=True)
except OSError:
self.skipTest("symlinks unavailable")
for name in ("alias/m.glb", "up/secret.glb"):
self.assertEqual(self.run_action(action="importGltf", filepath=name)["status"], "failed", name)
self.assertEqual(self.run_action(action="importGltf", filepath="real/m.glb")["status"], "accepted")
# importGltf: references inside the glTF
def write_gltf(self, name, document, glb=False):
path = self.root / "in" / name
if glb:
path.write_bytes(glb_bytes(document))
else:
path.write_text(json.dumps(document))
return path
def test_gltf_with_references_inside_the_root_and_data_uris_is_imported(self):
(self.root / "in" / "tex").mkdir()
(self.root / "in" / "tex" / "a.png").write_bytes(b"png")
(self.root / "in" / "m.bin").write_bytes(b"bin")
for glb, name in ((False, "ok.gltf"), (True, "ok.glb")):
self.write_gltf(name, {"asset": {"version": "2.0"},
"buffers": [{"uri": "m.bin", "byteLength": 3},
{"uri": "data:application/octet-stream;base64,AAAA", "byteLength": 3}],
"images": [{"uri": "tex/a%2Epng"}]}, glb=glb)
self.assertEqual(self.run_action(action="importGltf", filepath=name)["status"], "accepted", name)
self.assertEqual(len(self.bpy.imported), 2)
def test_gltf_references_that_leave_the_root_or_name_a_scheme_are_refused(self):
bad = ["../secret.glb", "..\\secret.glb", "%2e%2e/secret.glb", "/etc/passwd", "\\\\host\\share\\x.bin",
"C:\\Windows\\x.bin", "C:x.bin", "file:///etc/passwd", "http://example.com/x.bin",
"x.bin:stream", "sub/../../secret.glb", "CON", "x.bin."]
for kind in ("buffers", "images"):
for uri in bad:
for glb in (False, True):
name = f"bad{glb}.{'glb' if glb else 'gltf'}"
self.write_gltf(name, {"asset": {"version": "2.0"}, kind: [{"uri": uri}]}, glb=glb)
receipt = self.run_action(action="importGltf", filepath=name)
self.assertEqual(receipt["status"], "failed", (kind, uri, glb))
self.assertEqual(self.bpy.imported, [])
def test_gltf_reference_through_a_symlink_that_escapes_is_refused(self):
try:
os.symlink(self.root / "secret.glb", self.root / "in" / "buf.bin")
except OSError:
self.skipTest("symlinks unavailable")
self.write_gltf("link.gltf", {"asset": {"version": "2.0"}, "buffers": [{"uri": "buf.bin"}]})
receipt = self.run_action(action="importGltf", filepath="link.gltf")
self.assertEqual(receipt["status"], "failed")
self.assertIn("link", receipt["message"]) # the rule that fired, not a generic "outside"
self.assertEqual(self.bpy.imported, [])
def test_gltf_reference_to_a_missing_file_is_refused(self):
try:
os.symlink(self.root / "nowhere.bin", self.root / "in" / "dangling.bin")
dangling = True
except OSError:
dangling = False
for kind in ("buffers", "images"):
for uri in ["absent.bin", "sub/absent.png"] + (["dangling.bin"] if dangling else []):
self.write_gltf("miss.gltf", {"asset": {"version": "2.0"}, kind: [{"uri": uri}]})
receipt = self.run_action(action="importGltf", filepath="miss.gltf")
self.assertEqual(receipt["status"], "failed", (kind, uri))
self.assertRegex(receipt["message"], "does not exist|link", (kind, uri))
self.assertEqual(self.bpy.imported, [])
def test_malformed_gltf_is_refused(self):
(self.root / "in" / "junk.gltf").write_text("{not json")
(self.root / "in" / "list.gltf").write_text("[]")
(self.root / "in" / "trunc.glb").write_bytes(glb_bytes()[:21])
(self.root / "in" / "notglb.glb").write_bytes(b"GLTF" + bytes(30))
(self.root / "in" / "uri.gltf").write_text(json.dumps({"buffers": [{"uri": 5}]}))
for name in ("junk.gltf", "list.gltf", "trunc.glb", "notglb.glb", "uri.gltf"):
self.assertEqual(self.run_action(action="importGltf", filepath=name)["status"], "failed", name)
self.assertEqual(self.bpy.imported, [])
def test_import_is_refused_when_the_file_changes_between_check_and_use(self):
target_file = self.root / "in" / "model.glb"
original = sa.gltf_references
def swap_after_check(path):
result = original(path)
target_file.write_bytes(glb_bytes({"asset": {"version": "2.0"}, "extras": "x" * 50}))
return result
with mock.patch.object(sa, "gltf_references", swap_after_check):
receipt = self.run_action(action="importGltf", filepath="model.glb")
self.assertEqual(receipt["status"], "failed")
self.assertIn("changed", receipt["message"])
self.assertEqual(self.bpy.imported, [])
# renderStill: existing names and links
def test_render_refuses_an_existing_file_symlink_dangling_symlink_or_hardlink(self):
out = self.root / "out"
(out / "taken.png").write_bytes(b"old")
cases = ["taken.png"]
try:
os.symlink(self.root / "secret.glb", out / "sym.png")
os.symlink(self.root / "nowhere.png", out / "dangling.png")
os.link(self.root / "secret.glb", out / "hard.png")
cases += ["sym.png", "dangling.png", "hard.png"]
except OSError:
pass
for name in cases:
self.assertEqual(self.run_action(action="renderStill", outputPath=name)["status"], "failed", name)
self.assertEqual(self.bpy.rendered, [])
self.assertEqual((out / "taken.png").read_bytes(), b"old")
def test_render_refuses_a_symlinked_output_folder(self):
outside = self.root / "outside"
outside.mkdir()
try:
os.symlink(outside, self.root / "out" / "dir", target_is_directory=True)
except OSError:
self.skipTest("symlinks unavailable")
self.assertEqual(self.run_action(action="renderStill", outputPath="dir/a.png")["status"], "failed")
self.assertEqual(self.bpy.rendered, [])
self.assertEqual(list(outside.iterdir()), [])
def test_a_refused_render_leaves_no_folders_behind(self):
(self.root / "out" / "taken").mkdir()
(self.root / "out" / "taken" / "a.png").write_bytes(b"x")
for case in ({"outputPath": "new/deep/a.png", "resolutionX": 0}, {"outputPath": "new/deep/a.png", "engine": ""}, {"outputPath": "new/deep/a.png", "engine": "NOPE"},
{"outputPath": "new/deep/a.png", "cameraName": "Nope"}, {"outputPath": "new/deep/a.txt"},
{"outputPath": "taken/a.png"}):
self.assertEqual(self.run_action(action="renderStill", **case)["status"], "failed", case)
self.assertIn("already exists", self.run_action(action="renderStill", outputPath="taken/a.png")["message"])
self.assertEqual(sorted(p.name for p in (self.root / "out").iterdir()), ["taken"])
self.assertEqual(self.bpy.rendered, [])
def test_render_refuses_drive_unc_extended_stream_and_device_names(self):
for name in ("C:a.png", "C:\\a.png", "\\\\host\\share\\a.png", "\\\\?\\C:\\a.png", "a.png:stream",
"/a.png", "NUL.png", "a.png.", "x/../a.png"):
self.assertEqual(self.run_action(action="renderStill", outputPath=name)["status"], "failed", name)
self.assertEqual(self.bpy.rendered, [])
def test_render_is_refused_when_a_link_appears_after_the_check(self):
outside = self.root / "outside"
outside.mkdir()
real_mkdir = Path.mkdir
def swap(self_path, *args, **kwargs):
real_mkdir(self_path, *args, **kwargs)
if self_path.name == "late":
self_path.rmdir()
os.symlink(outside, self_path, target_is_directory=True)
try:
with mock.patch.object(Path, "mkdir", swap):
receipt = self.run_action(action="renderStill", outputPath="late/a.png")
except OSError:
self.skipTest("symlinks unavailable")
self.assertEqual(receipt["status"], "failed")
self.assertEqual(self.bpy.rendered, [])
def test_capabilities_advertise_the_new_actions(self):
for capability in ("asset.import.gltf", "camera.write", "light.write", "world.write", "render.still"):
self.assertIn(capability, bt.CAPABILITIES)
class PathTests(unittest.TestCase):
def test_resolve_inside_takes_plain_relative_paths_only(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp).resolve()
self.assertEqual(sa.resolve_inside(tmp, "a/./b.png", "output"), root / "a" / "b.png")
for name in ("a/../b.png", "a/../../b.png", "..", "", "."):
with self.assertRaises(sa.ActionRefused, msg=name):
sa.resolve_inside(tmp, name, "output")
def test_check_relative_refuses_device_names_however_windows_spells_them(self):
# Windows drops spaces before the first dot (CON .txt is CON) and reads superscript digits as COM/LPT digits.
for name in ("CON", "con", "CON .txt", "NUL .png", "sub/NUL .png", "COM1", "COM\u00b9", "LPT\u00b2.txt", "LPT\u00b3",
"CONIN$", "a/PRN.tar.gz"):
with self.assertRaisesRegex(sa.ActionRefused, "device", msg=name):
sa.check_relative(name, "output")
for name in ("CONSOLE.png", "COM10.png", "com.png", "a.CON.png", "NULL"):
sa.check_relative(name, "output")
def test_check_relative_itself_refuses_dotdot_and_nul(self):
# Each rule on its own: resolve_inside's link walk also trips on '..', so only a
# direct call proves the text rule.
for name in ("a/../b", "..", "a\\..\\b", "a\x00b"):
with self.assertRaises(sa.ActionRefused, msg=repr(name)):
sa.check_relative(name, "output")
self.assertEqual(sa.check_relative("a\\b/./c", "output"), ["a", "b", "c"])
def test_a_component_the_os_refuses_to_resolve_counts_as_a_link(self):
# Windows raises WinError 448 resolving some junctions; that is a link refusal, not a crash.
with tempfile.TemporaryDirectory() as tmp:
(Path(tmp) / "j").mkdir()
real = Path.resolve
def resolve(self, *args, **kwargs):
if self.name == "j":
raise OSError(448, "untrusted mount point")
return real(self, *args, **kwargs)
with mock.patch.object(Path, "resolve", resolve):
with self.assertRaisesRegex(sa.ActionRefused, "link"):
sa.resolve_inside(tmp, "j/x.png", "output")
if __name__ == "__main__":
unittest.main()