Blender add-on: importGltf, setCamera, setLight, setWorld, renderStill #2
7 changed files with 1026 additions and 0 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
226
surfaces/blender/brokkr_bridge/scene_actions.py
Normal file
226
surfaces/blender/brokkr_bridge/scene_actions.py
Normal 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.")
|
||||
|
|
@ -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.
|
||||
|
|
|
|||
99
surfaces/blender/tests/live/headless_smoke.py
Normal file
99
surfaces/blender/tests/live/headless_smoke.py
Normal 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)
|
||||
455
surfaces/blender/tests/test_scene_actions.py
Normal file
455
surfaces/blender/tests/test_scene_actions.py
Normal 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()
|
||||
Loading…
Reference in a new issue