Spaces:

cpuai
/

blender-render-studio

Running on Zero

blender-render-studio / tests /test_config_uploads.py

Deploy Blender Render Studio with Gradio API and MCP

94a0de3 verified 30 days ago

9.83 kB

"""配置和资源导入安全边界；不需要 Blender 或 GPU。"""

from __future__ import annotations

import json
import os
import stat
import sys
import zipfile
from contextlib import nullcontext
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from render_config import RenderConfig, json_schema, parse_config
from uploads import UploadError, stage_uploads


def make_zip(path: Path, entries: list[tuple[str | zipfile.ZipInfo, bytes]], compression=zipfile.ZIP_STORED) -> Path:
    with zipfile.ZipFile(path, "w", compression=compression) as archive:
        for name, contents in entries:
            archive.writestr(name, contents)
    return path


@pytest.mark.parametrize(
    "payload",
    [
        {"width": "1920"}, {"width": True}, {"denoise": 1},
        {"schema_version": True}, {"schema_version": 1.0}, {"schema_version": 2},
        {"color_depth": "8"}, {"color_depth": 8.0},
        {"width": 0}, {"height": 8193}, {"samples": 4097},
        {"gpu_seconds": 601}, {"cpu_timeout": 9}, {"fps": 0},
        {"unexpected": "do not silently ignore"},
        {"scene": ""}, {"camera": "Camera\x00script"},
    ],
)
def test_rejects_wrong_types_and_unknown_keys(payload):
    with pytest.raises(ValueError):
        parse_config(json.dumps(payload))


@pytest.mark.parametrize("path", ["../escape.blend", "/tmp/model.blend", "C:/model.blend", "a\\b.blend", "a/../b.blend", "a//b.blend", "a\x00.blend", "a:stream.blend"])
def test_config_rejects_unsafe_blend_path(path):
    with pytest.raises(ValueError):
        parse_config(json.dumps({"blend_file": path}))


@pytest.mark.parametrize("text", ['{"width": 1280,"width":1920}', '{"adaptive_threshold":NaN}', '{"adaptive_threshold":Infinity}', '[]', 'null', '{"width":', '{"scene":"' + "x" * 65536 + '"}'])
def test_json_syntax_and_size(text):
    with pytest.raises(ValueError):
        parse_config(text)


def test_defaults_schema_and_frame_sequence():
    default = parse_config("{}")
    assert default.frame_numbers() == [1]
    assert (default.output_width, default.output_height) == (1280, 720)
    config = parse_config('{"mode":"frames","frame_start":3,"frame_end":10,"frame_step":3}')
    assert config.frame_numbers() == [3, 6, 9]
    assert json_schema()["additionalProperties"] is False
    assert json_schema()["properties"]["device"]["enum"] == ["AUTO", "CUDA", "OPTIX", "CPU"]


@pytest.mark.parametrize("payload", [
    {"mode": "video", "transparent": True},
    {"mode": "video", "width": 1279},
    {"mode": "video", "width": 1282, "resolution_percentage": 50},
    {"mode": "video", "format": "OPEN_EXR", "color_depth": 16},
    {"format": "PNG", "color_depth": 32},
    {"format": "OPEN_EXR", "color_depth": 8},
    {"frame_start": 10, "frame_end": 9},
])
def test_format_and_video_constraints(payload):
    with pytest.raises(ValueError):
        parse_config(json.dumps(payload))


def test_server_limits_count_rendered_frames_and_effective_pixels(monkeypatch):
    monkeypatch.setenv("MAX_FRAMES", "3")
    monkeypatch.setenv("MAX_PIXELS", "1000000")
    assert len(RenderConfig(mode="frames", frame_start=1, frame_end=5, frame_step=2).frame_numbers()) == 3
    assert RenderConfig(width=1920, height=1080, resolution_percentage=50).output_width == 960
    with pytest.raises(ValueError, match="MAX_FRAMES"):
        RenderConfig(mode="frames", frame_end=4)
    with pytest.raises(ValueError, match="MAX_PIXELS"):
        RenderConfig(width=1920, height=1080)


def test_blend_with_resource_zip_and_nested_zip(tmp_path):
    blend = tmp_path / "scene.blend"
    # 不限制 .blend 头：兼容 Blender 自身的压缩格式。
    blend.write_bytes(b"compressed-placeholder")
    resources = make_zip(tmp_path / "resources.zip", [
        ("textures/albedo.png", b"png"), ("textures/", b""),
        ("cache/nested.zip", b"not recursively expanded"),
    ])
    destination = tmp_path / "project"
    selected = stage_uploads([str(blend), str(resources)], destination, None)
    assert selected == destination / "scene.blend"
    assert selected.is_absolute()
    assert (destination / "textures/albedo.png").read_bytes() == b"png"
    assert (destination / "cache/nested.zip").read_bytes() == b"not recursively expanded"


def test_multiple_blends_require_explicit_selection_and_preserve_paths(tmp_path):
    archive = make_zip(tmp_path / "project.zip", [("scenes/a.blend", b"a"), ("scenes/b.blend", b"b")])
    with pytest.raises(UploadError, match="多个"):
        stage_uploads([str(archive)], tmp_path / "failed", None)
    assert not (tmp_path / "failed").exists()
    selected = stage_uploads([str(archive)], tmp_path / "valid", "scenes/b.blend")
    assert selected.read_bytes() == b"b"


@pytest.mark.parametrize("malicious_name", [
    "../escape", "/absolute", "C:/drive", "..\\escape", "folder/../../escape", "folder//escape",
    "folder/file:stream", "folder/evil.py", "EVIL.PY", "folder/file.pyc", "folder/archive.tar.gz",
])
def test_zip_rejects_unsafe_entries_without_partial_output(tmp_path, malicious_name):
    archive = make_zip(tmp_path / "attack.zip", [("good.blend", b"blend"), (malicious_name, b"bad")])
    destination = tmp_path / "project"
    destination.mkdir()
    with pytest.raises(UploadError):
        stage_uploads([str(archive)], destination, None)
    assert list(destination.iterdir()) == []
    assert not list(tmp_path.glob(".project.upload-*"))
    assert not (tmp_path / "escape").exists()


@pytest.mark.parametrize("names", [
    ["scene.blend", "SCENE.blend"],
    ["scene.blend", "scene.blend"],
    ["scene.blend", "assets/x.png", "Assets/y.png"],
    ["scene.blend", "assets", "assets/x.png"],
])
def test_zip_rejects_collisions(tmp_path, names):
    with pytest.warns(UserWarning) if names[0] == names[1] else nullcontext():
        archive = make_zip(tmp_path / "conflict.zip", [(name, b"x") for name in names])
    with pytest.raises(UploadError, match="冲突|重复"):
        stage_uploads([str(archive)], tmp_path / "project", None)


def test_nul_zip_filename_is_not_silently_truncated(tmp_path):
    archive = make_zip(tmp_path / "nul.zip", [("scene.blend", b"blend"), ("evilQname", b"bad")])
    # zipfile.writestr 自身会截断 NUL，因此修改 local header 和 central directory。
    archive.write_bytes(archive.read_bytes().replace(b"evilQname", b"evil\x00name"))
    with pytest.raises(UploadError, match="NUL"):
        stage_uploads([str(archive)], tmp_path / "project", None)


@pytest.mark.parametrize("kind", [stat.S_IFLNK, stat.S_IFIFO, stat.S_IFCHR, stat.S_IFSOCK])
def test_zip_special_files_are_rejected(tmp_path, kind):
    entry = zipfile.ZipInfo("malicious")
    entry.create_system = 3
    entry.external_attr = (kind | 0o777) << 16
    archive = make_zip(tmp_path / "special.zip", [("scene.blend", b"blend"), (entry, b"../outside")])
    with pytest.raises(UploadError, match="符号链接|特殊"):
        stage_uploads([str(archive)], tmp_path / "project", None)


def test_source_symlink_and_fifo_rejected(tmp_path):
    source = tmp_path / "actual.blend"
    source.write_bytes(b"blend")
    link = tmp_path / "link.blend"
    link.symlink_to(source)
    with pytest.raises(UploadError, match="符号链接"):
        stage_uploads([str(link)], tmp_path / "project", None)
    if hasattr(os, "mkfifo"):
        pipe = tmp_path / "pipe.blend"
        os.mkfifo(pipe)
        with pytest.raises(UploadError, match="普通文件"):
            stage_uploads([str(pipe)], tmp_path / "project", None)


def test_zip_bomb_rejected_before_writing(tmp_path):
    archive = make_zip(tmp_path / "bomb.zip", [("scene.blend", b"0" * 2_000_000)], compression=zipfile.ZIP_DEFLATED)
    with pytest.raises(UploadError, match="压缩比"):
        stage_uploads([str(archive)], tmp_path / "project", None)
    assert not (tmp_path / "project").exists()


def test_extraction_size_and_entry_limits(tmp_path, monkeypatch):
    archive = make_zip(tmp_path / "large.zip", [("scene.blend", b"s"), ("large.bin", b"x" * 1_048_576)])
    monkeypatch.setenv("MAX_EXTRACT_MB", "1")
    with pytest.raises(UploadError, match="MAX_EXTRACT_MB"):
        stage_uploads([str(archive)], tmp_path / "project", None)
    assert not (tmp_path / "project").exists()
    monkeypatch.setenv("MAX_ARCHIVE_FILES", "1")
    with pytest.raises(UploadError, match="MAX_ARCHIVE_FILES"):
        stage_uploads([str(archive)], tmp_path / "project", None)


def test_upload_size_limit_and_script_rejection(tmp_path, monkeypatch):
    source = tmp_path / "scene.blend"
    source.write_bytes(b"x" * 1_048_577)
    monkeypatch.setenv("MAX_UPLOAD_MB", "1")
    with pytest.raises(UploadError, match="MAX_UPLOAD_MB"):
        stage_uploads([str(source)], tmp_path / "project", None)
    script = tmp_path / "helper.py"
    script.write_text("raise RuntimeError('must never execute')")
    with pytest.raises(UploadError, match="Python"):
        stage_uploads([str(script)], tmp_path / "project", None)


def test_preexisting_data_is_never_removed_on_failure(tmp_path):
    destination = tmp_path / "project"
    destination.mkdir()
    keep = destination / "keep.txt"
    keep.write_text("important")
    source = tmp_path / "scene.blend"
    source.write_bytes(b"blend")
    with pytest.raises(UploadError, match="覆盖"):
        stage_uploads([str(source)], destination, None)
    assert keep.read_text() == "important"


def test_separate_zip_and_file_cannot_overwrite_each_other(tmp_path):
    source = tmp_path / "scene.blend"
    source.write_bytes(b"original")
    archive = make_zip(tmp_path / "resources.zip", [("scene.blend", b"replace")])
    with pytest.raises(UploadError, match="冲突|重复"):
        stage_uploads([str(source), str(archive)], tmp_path / "project", None)
    assert source.read_bytes() == b"original"

	"""配置和资源导入安全边界；不需要 Blender 或 GPU。"""

	from __future__ import annotations

	import json
	import os
	import stat
	import sys
	import zipfile
	from contextlib import nullcontext
	from pathlib import Path

	import pytest

	sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

	from render_config import RenderConfig, json_schema, parse_config
	from uploads import UploadError, stage_uploads


	def make_zip(path: Path, entries: list[tuple[str \| zipfile.ZipInfo, bytes]], compression=zipfile.ZIP_STORED) -> Path:
	with zipfile.ZipFile(path, "w", compression=compression) as archive:
	for name, contents in entries:
	archive.writestr(name, contents)
	return path


	@pytest.mark.parametrize(
	"payload",
	[
	{"width": "1920"}, {"width": True}, {"denoise": 1},
	{"schema_version": True}, {"schema_version": 1.0}, {"schema_version": 2},
	{"color_depth": "8"}, {"color_depth": 8.0},
	{"width": 0}, {"height": 8193}, {"samples": 4097},
	{"gpu_seconds": 601}, {"cpu_timeout": 9}, {"fps": 0},
	{"unexpected": "do not silently ignore"},
	{"scene": ""}, {"camera": "Camera\x00script"},
	],
	)
	def test_rejects_wrong_types_and_unknown_keys(payload):
	with pytest.raises(ValueError):
	parse_config(json.dumps(payload))


	@pytest.mark.parametrize("path", ["../escape.blend", "/tmp/model.blend", "C:/model.blend", "a\\b.blend", "a/../b.blend", "a//b.blend", "a\x00.blend", "a:stream.blend"])
	def test_config_rejects_unsafe_blend_path(path):
	with pytest.raises(ValueError):
	parse_config(json.dumps({"blend_file": path}))


	@pytest.mark.parametrize("text", ['{"width": 1280,"width":1920}', '{"adaptive_threshold":NaN}', '{"adaptive_threshold":Infinity}', '[]', 'null', '{"width":', '{"scene":"' + "x" * 65536 + '"}'])
	def test_json_syntax_and_size(text):
	with pytest.raises(ValueError):
	parse_config(text)


	def test_defaults_schema_and_frame_sequence():
	default = parse_config("{}")
	assert default.frame_numbers() == [1]
	assert (default.output_width, default.output_height) == (1280, 720)
	config = parse_config('{"mode":"frames","frame_start":3,"frame_end":10,"frame_step":3}')
	assert config.frame_numbers() == [3, 6, 9]
	assert json_schema()["additionalProperties"] is False
	assert json_schema()["properties"]["device"]["enum"] == ["AUTO", "CUDA", "OPTIX", "CPU"]


	@pytest.mark.parametrize("payload", [
	{"mode": "video", "transparent": True},
	{"mode": "video", "width": 1279},
	{"mode": "video", "width": 1282, "resolution_percentage": 50},
	{"mode": "video", "format": "OPEN_EXR", "color_depth": 16},
	{"format": "PNG", "color_depth": 32},
	{"format": "OPEN_EXR", "color_depth": 8},
	{"frame_start": 10, "frame_end": 9},
	])
	def test_format_and_video_constraints(payload):
	with pytest.raises(ValueError):
	parse_config(json.dumps(payload))


	def test_server_limits_count_rendered_frames_and_effective_pixels(monkeypatch):
	monkeypatch.setenv("MAX_FRAMES", "3")
	monkeypatch.setenv("MAX_PIXELS", "1000000")
	assert len(RenderConfig(mode="frames", frame_start=1, frame_end=5, frame_step=2).frame_numbers()) == 3
	assert RenderConfig(width=1920, height=1080, resolution_percentage=50).output_width == 960
	with pytest.raises(ValueError, match="MAX_FRAMES"):
	RenderConfig(mode="frames", frame_end=4)
	with pytest.raises(ValueError, match="MAX_PIXELS"):
	RenderConfig(width=1920, height=1080)


	def test_blend_with_resource_zip_and_nested_zip(tmp_path):
	blend = tmp_path / "scene.blend"
	# 不限制 .blend 头：兼容 Blender 自身的压缩格式。
	blend.write_bytes(b"compressed-placeholder")
	resources = make_zip(tmp_path / "resources.zip", [
	("textures/albedo.png", b"png"), ("textures/", b""),
	("cache/nested.zip", b"not recursively expanded"),
	])
	destination = tmp_path / "project"
	selected = stage_uploads([str(blend), str(resources)], destination, None)
	assert selected == destination / "scene.blend"
	assert selected.is_absolute()
	assert (destination / "textures/albedo.png").read_bytes() == b"png"
	assert (destination / "cache/nested.zip").read_bytes() == b"not recursively expanded"


	def test_multiple_blends_require_explicit_selection_and_preserve_paths(tmp_path):
	archive = make_zip(tmp_path / "project.zip", [("scenes/a.blend", b"a"), ("scenes/b.blend", b"b")])
	with pytest.raises(UploadError, match="多个"):
	stage_uploads([str(archive)], tmp_path / "failed", None)
	assert not (tmp_path / "failed").exists()
	selected = stage_uploads([str(archive)], tmp_path / "valid", "scenes/b.blend")
	assert selected.read_bytes() == b"b"


	@pytest.mark.parametrize("malicious_name", [
	"../escape", "/absolute", "C:/drive", "..\\escape", "folder/../../escape", "folder//escape",
	"folder/file:stream", "folder/evil.py", "EVIL.PY", "folder/file.pyc", "folder/archive.tar.gz",
	])
	def test_zip_rejects_unsafe_entries_without_partial_output(tmp_path, malicious_name):
	archive = make_zip(tmp_path / "attack.zip", [("good.blend", b"blend"), (malicious_name, b"bad")])
	destination = tmp_path / "project"
	destination.mkdir()
	with pytest.raises(UploadError):
	stage_uploads([str(archive)], destination, None)
	assert list(destination.iterdir()) == []
	assert not list(tmp_path.glob(".project.upload-*"))
	assert not (tmp_path / "escape").exists()


	@pytest.mark.parametrize("names", [
	["scene.blend", "SCENE.blend"],
	["scene.blend", "scene.blend"],
	["scene.blend", "assets/x.png", "Assets/y.png"],
	["scene.blend", "assets", "assets/x.png"],
	])
	def test_zip_rejects_collisions(tmp_path, names):
	with pytest.warns(UserWarning) if names[0] == names[1] else nullcontext():
	archive = make_zip(tmp_path / "conflict.zip", [(name, b"x") for name in names])
	with pytest.raises(UploadError, match="冲突\|重复"):
	stage_uploads([str(archive)], tmp_path / "project", None)


	def test_nul_zip_filename_is_not_silently_truncated(tmp_path):
	archive = make_zip(tmp_path / "nul.zip", [("scene.blend", b"blend"), ("evilQname", b"bad")])
	# zipfile.writestr 自身会截断 NUL，因此修改 local header 和 central directory。
	archive.write_bytes(archive.read_bytes().replace(b"evilQname", b"evil\x00name"))
	with pytest.raises(UploadError, match="NUL"):
	stage_uploads([str(archive)], tmp_path / "project", None)


	@pytest.mark.parametrize("kind", [stat.S_IFLNK, stat.S_IFIFO, stat.S_IFCHR, stat.S_IFSOCK])
	def test_zip_special_files_are_rejected(tmp_path, kind):
	entry = zipfile.ZipInfo("malicious")
	entry.create_system = 3
	entry.external_attr = (kind \| 0o777) << 16
	archive = make_zip(tmp_path / "special.zip", [("scene.blend", b"blend"), (entry, b"../outside")])
	with pytest.raises(UploadError, match="符号链接\|特殊"):
	stage_uploads([str(archive)], tmp_path / "project", None)


	def test_source_symlink_and_fifo_rejected(tmp_path):
	source = tmp_path / "actual.blend"
	source.write_bytes(b"blend")
	link = tmp_path / "link.blend"
	link.symlink_to(source)
	with pytest.raises(UploadError, match="符号链接"):
	stage_uploads([str(link)], tmp_path / "project", None)
	if hasattr(os, "mkfifo"):
	pipe = tmp_path / "pipe.blend"
	os.mkfifo(pipe)
	with pytest.raises(UploadError, match="普通文件"):
	stage_uploads([str(pipe)], tmp_path / "project", None)


	def test_zip_bomb_rejected_before_writing(tmp_path):
	archive = make_zip(tmp_path / "bomb.zip", [("scene.blend", b"0" * 2_000_000)], compression=zipfile.ZIP_DEFLATED)
	with pytest.raises(UploadError, match="压缩比"):
	stage_uploads([str(archive)], tmp_path / "project", None)
	assert not (tmp_path / "project").exists()


	def test_extraction_size_and_entry_limits(tmp_path, monkeypatch):
	archive = make_zip(tmp_path / "large.zip", [("scene.blend", b"s"), ("large.bin", b"x" * 1_048_576)])
	monkeypatch.setenv("MAX_EXTRACT_MB", "1")
	with pytest.raises(UploadError, match="MAX_EXTRACT_MB"):
	stage_uploads([str(archive)], tmp_path / "project", None)
	assert not (tmp_path / "project").exists()
	monkeypatch.setenv("MAX_ARCHIVE_FILES", "1")
	with pytest.raises(UploadError, match="MAX_ARCHIVE_FILES"):
	stage_uploads([str(archive)], tmp_path / "project", None)


	def test_upload_size_limit_and_script_rejection(tmp_path, monkeypatch):
	source = tmp_path / "scene.blend"
	source.write_bytes(b"x" * 1_048_577)
	monkeypatch.setenv("MAX_UPLOAD_MB", "1")
	with pytest.raises(UploadError, match="MAX_UPLOAD_MB"):
	stage_uploads([str(source)], tmp_path / "project", None)
	script = tmp_path / "helper.py"
	script.write_text("raise RuntimeError('must never execute')")
	with pytest.raises(UploadError, match="Python"):
	stage_uploads([str(script)], tmp_path / "project", None)


	def test_preexisting_data_is_never_removed_on_failure(tmp_path):
	destination = tmp_path / "project"
	destination.mkdir()
	keep = destination / "keep.txt"
	keep.write_text("important")
	source = tmp_path / "scene.blend"
	source.write_bytes(b"blend")
	with pytest.raises(UploadError, match="覆盖"):
	stage_uploads([str(source)], destination, None)
	assert keep.read_text() == "important"


	def test_separate_zip_and_file_cannot_overwrite_each_other(tmp_path):
	source = tmp_path / "scene.blend"
	source.write_bytes(b"original")
	archive = make_zip(tmp_path / "resources.zip", [("scene.blend", b"replace")])
	with pytest.raises(UploadError, match="冲突\|重复"):
	stage_uploads([str(source), str(archive)], tmp_path / "project", None)
	assert source.read_bytes() == b"original"