Files
orisys-app/shared/tests/test_scene3d_protect.py
T
2026-09-25 17:46:37 +08:00

675 lines
26 KiB
Python

"""Stage 5 acceptance: sealing, signatures, the platform lock and the record block.
The design document's §5.2 lists five protections and is explicit about what each one
buys; these tests are about holding the code to *those* claims and to no larger ones:
* **origin.** A signed package verifies against the shipped public key, and every way
of editing it — the manifest, an entry's bytes, the envelope's metadata, the
ciphertext — is refused. The tests also assert the negative: a signature over a
*plain* package is not accepted as encryption, and a package with no signature is
reported as unsigned rather than quietly accepted as if it were signed.
* **optional encryption.** AES-256-GCM with a fresh nonce per seal, the envelope bound
as associated data, and the plaintext digest inside the signed part. Two seals of
the same content must not share a nonce or a ciphertext.
* **key isolation.** Reading a sealed package needs a :class:`KeyProvider` and a
verification key; a *wrong* key is a refusal, not a crash, and no message anywhere
contains key material.
* **the platform lock.** A package compiled for one operating system, architecture,
GL profile or GPU is refused elsewhere — with the reason — and an unlocked package
is not the same as one locked to nothing.
* **the record block.** ``session.json`` says what the pane was; a recording without
the block (an older file) reads as "not supported" instead of raising.
"""
from __future__ import annotations
import json
import zipfile
from dataclasses import replace
from pathlib import Path
import numpy as np
import pytest
from orisys_app_shared.record.visualization3d import (
BLOCK_VERSION,
build_block,
read_block,
)
from orisys_app_shared.scene3d import protect, synthetic
from orisys_app_shared.scene3d.contracts import (
PlatformInfo,
PlatformLock,
Sealing,
ThreeDError,
check_platform_lock,
)
from orisys_app_shared.scene3d.package import (
ENVELOPE_VERSION,
PAYLOAD_ENTRY,
SUPPORTED_ENVELOPE_VERSIONS,
describe_envelope,
read_package,
scene_from_contents,
write_package,
)
pytest.importorskip("cryptography", reason="stage 5 sealing needs the crypto backend")
@pytest.fixture(scope="module")
def signing_key() -> bytes:
return protect.generate_signing_key()
@pytest.fixture(scope="module")
def verify_key(signing_key) -> bytes:
return protect.public_key_pem(signing_key)
@pytest.fixture
def provider() -> protect.MappingKeyProvider:
return protect.MappingKeyProvider({"customer-1": "11" * 32})
@pytest.fixture
def scene():
return synthetic.poc_scene()
def _seal(tmp_path, scene, *, signing_key, verify_key, provider, name="sealed.oryscene"):
path = tmp_path / name
write_package(
scene,
path,
sealing=Sealing(
key_id="customer-1",
provider=provider,
signing_key_pem=signing_key,
signing_key_reference="build-2024",
verify_key_pem=verify_key,
),
)
return path
def _sign_only(tmp_path, scene, *, signing_key, verify_key, name="signed.oryscene"):
path = tmp_path / name
write_package(
scene,
path,
sealing=Sealing(
signing_key_pem=signing_key,
signing_key_reference="build-2024",
verify_key_pem=verify_key,
),
)
return path
# ------------------------------------------------------------------ the envelope
def test_a_sealed_package_round_trips_and_carries_the_same_content(
tmp_path, scene, signing_key, verify_key, provider
):
sealed = _seal(tmp_path, scene, signing_key=signing_key, verify_key=verify_key, provider=provider)
plain = tmp_path / "plain.oryscene"
write_package(scene, plain)
contents = read_package(sealed, keys=provider, verify_key_pem=verify_key)
reference = read_package(plain)
assert set(contents.arrays) == set(reference.arrays)
for name in contents.arrays:
assert np.array_equal(contents.arrays[name], reference.arrays[name]), name
assert contents.manifest["asset_hash"] == reference.manifest["asset_hash"], (
"sealing must not change the asset identity"
)
assets = scene_from_contents(contents)
assert assets.geometry_id == scene.geometry_id
def test_the_envelope_says_it_is_sealed_without_any_key(
tmp_path, scene, signing_key, verify_key, provider
):
sealed = _seal(tmp_path, scene, signing_key=signing_key, verify_key=verify_key, provider=provider)
summary = describe_envelope(sealed)
assert summary["envelope_version"] == ENVELOPE_VERSION == 2
assert summary["payload"] == "aes-gcm"
assert summary["sealed"] is True and summary["signed"] is True
assert summary["key_id"] == "customer-1"
assert summary["signing_key"] == "build-2024"
assert summary["ciphertext_bytes"] > 0
assert PAYLOAD_ENTRY in summary["members"]
# The summary is what a support ticket quotes: no key material, no nonce, no salt.
text = json.dumps(summary)
assert "11" * 32 not in text and "nonce" not in text and "salt" not in text
assert SUPPORTED_ENVELOPE_VERSIONS == (1, 2)
def test_a_plain_package_still_reads_and_says_it_is_unsigned(tmp_path, scene):
plain = tmp_path / "plain.oryscene"
write_package(scene, plain)
summary = describe_envelope(plain)
assert summary["envelope_version"] == 1
assert summary["payload"] == "plain"
assert summary["signed"] is False and summary["sealed"] is False
assert scene_from_contents(read_package(plain)).asset_id == scene.asset_id
def test_a_signed_plain_package_needs_the_public_key_and_verifies_with_it(
tmp_path, scene, signing_key, verify_key
):
signed = _sign_only(tmp_path, scene, signing_key=signing_key, verify_key=verify_key)
summary = describe_envelope(signed)
assert summary["payload"] == "plain" and summary["signed"] is True
assert summary["sealed"] is False
with pytest.raises(ThreeDError) as excinfo:
read_package(signed)
assert "verification key" in excinfo.value.diagnostic.message
contents = read_package(signed, verify_key_pem=verify_key)
assert contents.manifest["asset_id"] == scene.asset_id
# ------------------------------------------------------------------- tampering
def _rewrite(path: Path, target: Path, mutate) -> Path:
"""Copy a package while changing one member (or the envelope)."""
with zipfile.ZipFile(path) as source:
members = {name: source.read(name) for name in source.namelist()}
members = mutate(members)
with zipfile.ZipFile(target, "w") as archive:
for name, payload in members.items():
archive.writestr(name, payload)
return target
@pytest.mark.parametrize("what", ["ciphertext", "entry", "manifest", "metadata"])
def test_every_edit_of_a_sealed_package_is_refused(
tmp_path, scene, signing_key, verify_key, provider, what
):
sealed = _seal(tmp_path, scene, signing_key=signing_key, verify_key=verify_key, provider=provider)
def mutate(members: dict[str, bytes]) -> dict[str, bytes]:
if what == "ciphertext":
payload = bytearray(members[PAYLOAD_ENTRY])
payload[len(payload) // 2] ^= 0x01
members[PAYLOAD_ENTRY] = bytes(payload)
elif what == "entry":
# A signed *plain* package is where an entry can be edited directly.
raise AssertionError("covered by the signed-plain case below")
elif what == "manifest":
payload = members[PAYLOAD_ENTRY]
members[PAYLOAD_ENTRY] = payload + b"\x00"
else:
envelope = json.loads(members["envelope.json"])
envelope["key_id"] = "customer-2"
members["envelope.json"] = json.dumps(envelope).encode()
return members
if what == "entry":
pytest.skip("an entry cannot be edited inside a ciphertext; see the signed case")
broken = _rewrite(sealed, tmp_path / f"broken-{what}.oryscene", mutate)
with pytest.raises(ThreeDError) as excinfo:
read_package(broken, keys=provider, verify_key_pem=verify_key)
assert excinfo.value.diagnostic.code in ("asset_auth_failed",)
assert excinfo.value.diagnostic.message
def test_editing_an_entry_of_a_signed_package_is_refused(
tmp_path, scene, signing_key, verify_key
):
"""The signature covers the entries: this is what "origin" has to mean."""
signed = _sign_only(tmp_path, scene, signing_key=signing_key, verify_key=verify_key)
def mutate(members):
payload = bytearray(members["sensing/positions.npy"])
payload[-1] ^= 0x01
members["sensing/positions.npy"] = bytes(payload)
return members
broken = _rewrite(signed, tmp_path / "broken-entry.oryscene", mutate)
with pytest.raises(ThreeDError) as excinfo:
read_package(broken, verify_key_pem=verify_key)
assert excinfo.value.diagnostic.code == "asset_auth_failed"
def test_a_wrong_verification_key_is_refused(tmp_path, scene, signing_key, verify_key, provider):
sealed = _seal(tmp_path, scene, signing_key=signing_key, verify_key=verify_key, provider=provider)
other = protect.public_key_pem(protect.generate_signing_key())
with pytest.raises(ThreeDError) as excinfo:
read_package(sealed, keys=provider, verify_key_pem=other)
assert "signature" in excinfo.value.diagnostic.message
def test_a_wrong_content_key_is_refused_and_reveals_nothing(
tmp_path, scene, signing_key, verify_key, provider
):
sealed = _seal(tmp_path, scene, signing_key=signing_key, verify_key=verify_key, provider=provider)
wrong = protect.MappingKeyProvider({"customer-1": "22" * 32})
with pytest.raises(ThreeDError) as excinfo:
read_package(sealed, keys=wrong, verify_key_pem=verify_key)
item = excinfo.value.diagnostic
assert item.code == "asset_auth_failed"
assert "22" * 32 not in item.message
assert "11" * 32 not in item.message
def test_a_missing_key_for_the_key_id_is_named(tmp_path, scene, signing_key, verify_key, provider):
sealed = _seal(tmp_path, scene, signing_key=signing_key, verify_key=verify_key, provider=provider)
empty = protect.MappingKeyProvider({})
with pytest.raises(ThreeDError) as excinfo:
read_package(sealed, keys=empty, verify_key_pem=verify_key)
assert "customer-1" in excinfo.value.diagnostic.message
# ------------------------------------------------------------- the primitives
def test_a_nonce_is_never_reused_and_the_salt_is_fresh(provider):
first, first_bytes = protect.seal_payload(
b"the same payload", key_id="customer-1", provider=provider
)
second, second_bytes = protect.seal_payload(
b"the same payload", key_id="customer-1", provider=provider
)
assert first["nonce"] != second["nonce"]
assert first["salt"] != second["salt"]
assert first_bytes != second_bytes, "the same plaintext must not produce the same bytes"
assert len(protect._unb64(first["nonce"], field="nonce", length=protect.NONCE_BYTES)) == 12
def test_the_envelope_metadata_is_bound_into_the_ciphertext(provider):
envelope, ciphertext = protect.seal_payload(
b"payload", key_id="customer-1", provider=provider, extra={"asset_id": "x"}
)
assert protect.open_payload(envelope, ciphertext, provider=provider) == b"payload"
changed = dict(envelope)
changed["asset_id"] = "y"
with pytest.raises(ThreeDError):
protect.open_payload(changed, ciphertext, provider=provider)
def test_a_signed_envelope_covers_the_plaintext_digest(signing_key, verify_key, provider):
envelope, _ = protect.seal_payload(
b"payload", key_id="customer-1", provider=provider,
signing_key_pem=signing_key, signing_key_reference="k",
)
protect.verify_envelope(envelope, public_key_pem=verify_key)
changed = dict(envelope)
changed["payload_sha256"] = "0" * 64
with pytest.raises(ThreeDError) as excinfo:
protect.verify_envelope(changed, public_key_pem=verify_key)
assert "signature" in excinfo.value.diagnostic.message
def test_a_key_provider_must_return_a_usable_key():
with pytest.raises(ThreeDError):
protect.MappingKeyProvider({"short": "abcd"})
with pytest.raises(ThreeDError):
protect.MappingKeyProvider({"not-text": 17})
class Wrong:
def get_content_key(self, key_id: str) -> bytes:
return b"too short"
with pytest.raises(ThreeDError) as excinfo:
protect.seal_payload(b"x", key_id="k", provider=Wrong())
assert "32" in excinfo.value.diagnostic.message
def test_a_key_file_is_read_as_hex_base64_or_raw(tmp_path):
raw = bytes(range(32))
for name, text in (
("hex.key", raw.hex()),
("b64.key", __import__("base64").b64encode(raw).decode()),
):
path = tmp_path / name
path.write_text(text, encoding="utf-8")
assert protect.load_key_file(path, key_id="k").get_content_key("k") == raw
binary = tmp_path / "raw.key"
binary.write_bytes(raw)
assert protect.load_key_file(binary, key_id="k").get_content_key("k") == raw
def test_the_writer_refuses_to_seal_without_a_backend_or_without_a_key(
tmp_path, scene, monkeypatch
):
with pytest.raises(ThreeDError) as missing:
write_package(
scene,
tmp_path / "no-key.oryscene",
sealing=Sealing(key_id="customer-1", provider=None),
)
assert missing.value.diagnostic.code == "asset_auth_failed"
import orisys_app_shared.scene3d.package as package_module
monkeypatch.setattr(package_module, "backend_available", lambda: False)
from orisys_app_shared.scene3d.assets import SceneCompiler
options = _compile_options(
encryption_key_id="customer-1",
sealing=Sealing(
key_id="customer-1", provider=protect.MappingKeyProvider({"customer-1": "11" * 32})
),
)
with pytest.raises(ThreeDError) as no_backend:
SceneCompiler().compile(scene, None, tmp_path / "no-backend.oryscene", options)
assert "cryptography" in no_backend.value.diagnostic.message
def _compile_options(*, encryption_key_id=None, sealing=None, lock=None):
from orisys_app_shared.scene3d.contracts import CompileOptions
return CompileOptions(
asset_id="x",
version="0",
expected=synthetic.poc_expected(),
transform_sensor_from_glb=np.eye(4),
max_surface_vertices=0,
max_housing_triangles=1000,
signing_key_reference="build-key",
encryption_key_id=encryption_key_id,
platform_lock=lock,
sealing=sealing,
)
# ------------------------------------------------------------------ the lock
def test_an_unlocked_package_is_not_locked_to_nothing():
assert PlatformLock().is_unrestricted()
assert PlatformLock().to_dict()["os_family"] == []
assert check_platform_lock(None, PlatformInfo("linux", "aarch64")) == ()
assert check_platform_lock(PlatformLock(), PlatformInfo("linux", "aarch64")) == ()
def test_the_lock_covers_os_arch_profile_gpu_and_software_renderers():
lock = PlatformLock(
os_family=("windows",),
machine=("amd64",),
gl_profile="desktop33",
allowed_vendors=("NVIDIA",),
allowed_renderers=("RTX 4080",),
min_texture_size=16384,
note="dev box",
)
# The machine this was qualified on passes.
assert (
check_platform_lock(
lock,
PlatformInfo("windows", "amd64", "desktop33", "NVIDIA Corporation",
"NVIDIA GeForce RTX 4080 SUPER", False, 32768),
)
== ()
)
# Operating system and architecture are checkable with no graphics information.
static = check_platform_lock(lock, PlatformInfo("linux", "aarch64"))
assert len(static) == 2
assert all(item.code == "platform_mismatch" for item in static)
# Every graphics violation at once: profile, software renderer, vendor,
# renderer and texture size.
full = check_platform_lock(
lock,
PlatformInfo("windows", "amd64", "gles30", "Mesa", "llvmpipe", True, 4096),
)
joined = " ".join(item.message for item in full)
assert "profile" in joined
assert "software rasteriser" in joined
assert "vendor" in joined or "renderer" in joined
assert "texture" in joined
def test_the_lock_can_be_used_to_endorse_a_gpu_by_name():
"""Vendor is 'NVIDIA Corporation', the renderer is the specific card string."""
lock = PlatformLock(allowed_renderers=("RTX 4080",))
assert check_platform_lock(
lock, PlatformInfo("windows", "amd64", "desktop33", "NVIDIA",
"NVIDIA GeForce RTX 4080 SUPER/PCIe/SSE2", False, 32768)
) == ()
assert check_platform_lock(
lock, PlatformInfo("windows", "amd64", "desktop33", "Intel", "Intel UHD", False, 16384)
)
def test_a_malformed_lock_is_rejected_rather_than_ignored():
with pytest.raises(ThreeDError):
PlatformLock.from_dict({"gl_profile": "vulkan"})
with pytest.raises(ThreeDError):
PlatformLock.from_dict({"os_family": "windows"})
with pytest.raises(ThreeDError):
PlatformLock.from_dict({"min_texture_size": -4})
with pytest.raises(ThreeDError):
PlatformLock.from_dict({"os_family": ["windows"], "min_texture_size": "big"})
def test_the_lock_travels_into_the_manifest_and_the_asset(tmp_path, scene):
locked = replace(
scene,
platform_lock=PlatformLock(os_family=("windows",), machine=("amd64",), note="dev"),
)
path = tmp_path / "locked.oryscene"
write_package(locked, path)
contents = read_package(path)
assert contents.manifest["platform_lock"]["os_family"] == ["windows"]
assets = scene_from_contents(contents)
assert assets.platform_lock is not None
assert assets.platform_lock.machine == ("amd64",)
assert assets.platform_lock.note == "dev"
# And a locked asset is a different package from an unlocked one.
other = tmp_path / "unlocked.oryscene"
write_package(scene, other)
assert read_package(path).manifest["asset_hash"] != read_package(other).manifest["asset_hash"]
# --------------------------------------------------------------- the record block
def test_the_block_records_what_the_pane_was():
block = build_block(
enabled=True,
available=True,
scene={"asset_id": "finger-display", "asset_hash": "a" * 64},
gl={"profile": "desktop33", "renderer": "RTX 4080", "software_renderer": False},
display={"deform_enabled": True, "display_gain": 50.0, "max_fps": 30},
stream={"quantity": "flow_magnitude", "unit": "px", "quality": "approximate"},
)
assert block["block_version"] == BLOCK_VERSION
record = read_block({"visualization3d": block})
assert record.supported and record.enabled and record.available
assert record.asset_hash() == "a" * 64
assert record.was_showing_measurements() is True
assert record.display["display_gain"] == 50.0
def test_a_disabled_pane_records_its_reason_and_no_measurement():
block = build_block(enabled=False, available=True, reason="the operator turned it off")
record = read_block({"visualization3d": block})
assert record.supported and not record.enabled
assert record.reason == "the operator turned it off"
assert record.was_showing_measurements() is False
def test_a_reference_model_is_not_an_observation():
block = build_block(
enabled=True,
available=True,
stream={"quantity": "unavailable", "unit": "1", "quality": "display_only"},
)
assert read_block({"visualization3d": block}).was_showing_measurements() is False
@pytest.mark.parametrize(
"session,reason_part",
[
(None, "no session metadata"),
({}, "no visualization3d block"),
({"visualization3d": 7}, "not an object"),
({"visualization3d": {"enabled": True}}, "version"),
({"visualization3d": {"block_version": 99}}, "not"),
],
)
def test_an_old_or_broken_recording_reads_as_unsupported(session, reason_part):
record = read_block(session)
assert record.supported is False
assert reason_part in record.reason
assert record.enabled is False
def test_an_empty_reason_is_not_written_for_a_working_pane():
block = build_block(enabled=True, available=True, reason="ignored")
assert block["reason"] == "", "a working pane must not carry a reason"
# ------------------------------------------------- the release workflow, end to end
def _tool(name: str):
"""Import a tool by path: the tools are scripts, not an installed package."""
import importlib.util
path = Path(__file__).resolve().parents[2] / "tools" / "scene_assets" / f"{name}.py"
spec = importlib.util.spec_from_file_location(f"scene_tool_{name}", path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def _shipped_asset_dirs() -> tuple[Path, Path] | None:
"""The Blender export and its unroll map, when the repository carries them.
The map's name follows ``camera.source_scale`` (the SDK's ``scaled_map_cache_path``), so it is
derived from the bundle's own ``sensor.json`` rather than pinned: a scale change renames the
file, and a hard-coded name turns this acceptance test into a silent skip.
"""
bundle = (
Path(__file__).resolve().parents[2]
/ "apps"
/ "studio"
/ "config"
/ "sensors"
/ "finger"
)
base = bundle / "assets" / "30fps-0703"
config_path = bundle / "sensor.json"
if not config_path.is_file():
return None
from orisys_app_shared.acquisition.geometry import map_path_for
scale = (json.loads(config_path.read_text(encoding="utf-8")).get("camera") or {}).get(
"source_scale"
)
mapping = map_path_for(base, scale)
if (base / "mesh.npz").is_file() and mapping.is_file():
return base, mapping
return None
def test_the_release_workflow_produces_a_signed_sealed_locked_package(tmp_path):
"""``compile_scene.py`` → ``validate_scene.py``, the way a release script runs it.
This is the stage-5 acceptance in one test: a real export compiled with a signing
key, a content key and a platform lock taken from a probe report, then validated
with only the public key and the content key — and then the same validation run
against a package with one byte changed, which must fail.
"""
shipped = _shipped_asset_dirs()
if shipped is None:
pytest.skip("the shipped finger export is not in this checkout")
legacy_dir, mapping = shipped
compile_tool = _tool("compile_scene")
validate_tool = _tool("validate_scene")
keys = tmp_path / "keys"
keys.mkdir()
signing = keys / "build-ed25519.pem"
signing.write_bytes(protect.generate_signing_key())
public = keys / "build-ed25519.pub.pem"
public.write_bytes(protect.public_key_pem(signing.read_bytes()))
content = keys / "finger-display.key"
content.write_text(bytes(range(32)).hex(), encoding="utf-8")
probe = tmp_path / "gl-report.json"
probe.write_text(
json.dumps(
{
"profile_requested": "desktop33",
"host": {"os_family": "windows", "machine": "amd64"},
"context": {
"profile_used": "desktop33",
"vendor": "NVIDIA Corporation",
"renderer": "NVIDIA GeForce RTX 4080 SUPER/PCIe/SSE2",
"software_renderer": False,
"max_texture_size": 32768,
},
}
),
encoding="utf-8",
)
package = tmp_path / "finger-display-1.0.0.oryscene"
code = compile_tool.main(
[
"--legacy-dir", str(legacy_dir),
"--map", str(mapping),
"--meters-per-source-unit", "0.001",
"--sign-key", str(signing),
"--sign-key-id", "build-2024",
"--write-verify-key", str(public),
"--encrypt-key-id", "finger-display",
"--key-file", str(content),
"--lock-from-probe", str(probe),
"--output", str(package),
]
)
assert code == 0
summary = describe_envelope(package)
assert summary["sealed"] and summary["signed"]
assert summary["key_id"] == "finger-display"
contents = read_package(
package,
keys=protect.load_key_file(content, key_id="finger-display"),
verify_key_pem=public.read_bytes(),
)
assets = scene_from_contents(contents)
assert assets.sensing_part().positions_sensor_m.shape[0] == 1032
assert assets.platform_lock is not None
assert assets.platform_lock.gl_profile == "desktop33"
assert assets.platform_lock.allowed_renderers == (
"NVIDIA GeForce RTX 4080 SUPER/PCIe/SSE2",
)
# The lock is inside the signed manifest, so the asset hash covers it.
assert contents.manifest["platform_lock"]["min_texture_size"] == 32768
release_args = [
str(package),
"--key-file", str(content),
"--key-id", "finger-display",
"--verify-key", str(public),
"--require-signature",
]
assert validate_tool.main(release_args) == 0
# One flipped byte in the ciphertext: the release gate must fail, not warn.
damaged = tmp_path / "damaged.oryscene"
with zipfile.ZipFile(package) as source:
members = {name: source.read(name) for name in source.namelist()}
payload = bytearray(members[PAYLOAD_ENTRY])
payload[0] ^= 0x01
members[PAYLOAD_ENTRY] = bytes(payload)
with zipfile.ZipFile(damaged, "w") as target:
for name, value in members.items():
target.writestr(name, value)
assert (
validate_tool.main(
[
str(damaged),
"--key-file", str(content),
"--key-id", "finger-display",
"--verify-key", str(public),
"--require-signature",
]
)
== 2
)
def test_the_release_gate_fails_an_unsigned_package(tmp_path, scene):
"""``--require-signature`` is the difference between a build and a release."""
validate_tool = _tool("validate_scene")
plain = tmp_path / "plain.oryscene"
write_package(scene, plain)
assert validate_tool.main([str(plain)]) == 0, "an engineering package is acceptable"
assert validate_tool.main([str(plain), "--require-signature"]) == 2