bruno/ifcurl
1
0
Fork 0
mirror of https://github.com/brunopostle/ifcurl.git synced 2026-07-12 02:08:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

sandbox.py: subprocess isolation for ifcopenshell parse + render

Browse source 
All ifcopenshell calls (from_string, filter_elements, geom.iterator) now
run inside a short-lived child process via run_sandboxed().  A SIGSEGV or
SIGABRT in the child returns HTTP 422 instead of killing the service worker.
Configurable timeout (IFCURL_SANDBOX_TIMEOUT, default 120s) and optional
resource limits (IFCURL_SANDBOX_MEMORY_MB, IFCURL_SANDBOX_CPU_SECS).

The T3 GUID cache still works: the sandboxed pipeline returns resolved
GUIDs on selector miss so the parent can populate the cache; on hit the
parent passes cached GUIDs and the child converts them to step IDs.

Tests use a sync_sandbox fixture that runs run_sandboxed synchronously so
mock patches remain visible; four dedicated sandbox tests cover normal
return, exception propagation, segfault detection, and timeout.

Closes ifcurl-vm5.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Bruno Postle 2026-04-24 17:38:07 +01:00
parent 02d93e68e1
commit 65a62ffc8b
5 changed files with 322 additions and 82 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.beads/issues.jsonl
View file

@ -1,3 +1,4 @@
{"id":"ifcurl-vm5","title":"ifcopenshell segfault on malformed IFC: service DoS and potential RCE risk","description":"ifcopenshell uses C++ bindings (via web-ifc) that can segfault when given malformed or adversarially crafted IFC data. In the current service architecture this has two serious implications:\n\n1. **DoS**: a segfault kills the uvicorn worker process. Under repeated attack a service with --workers=1 (the common deployment) goes down entirely. Even with multiple workers, each is vulnerable independently.\n\n2. **Potential RCE**: memory-corruption bugs in C++ parsers are a well-known exploit vector. A crafted IFC file that triggers a heap overflow in ifcopenshell could yield arbitrary code execution as the service user. The service runs on the same host as Forgejo, making this a high-value target.\n\nThe two call sites at risk are:\n- `_load_model()` → `ifcopenshell.file.from_string()` in service.py (runs in the uvicorn worker thread)\n- `render_mod.render()` → `ifcopenshell.geom.iterator()` in render.py (already spawns worker subprocesses, so a segfault there kills a worker process but not the service)\n\nThe primary mitigation to investigate is **subprocess isolation for the parse step**: run `from_string()` in a short-lived child process with a timeout and resource limits. If the child exits with a signal (SIGSEGV, SIGABRT) or times out, return HTTP 422 to the caller without killing the service. This is the standard pattern used by document-conversion services (e.g. LibreOffice sandboxing).\n\nSecondary mitigations:\n- Apply `ulimit` / `resource.setrlimit` to cap address-space and CPU time in the child\n- Consider `seccomp` syscall filtering for the parse subprocess (restrict to read-only FS access, no network, no exec)\n- File-size cap on uploaded/fetched IFC bytes (tracked separately in ifcurl-fu5)\n- Fuzz testing with `atheris` or AFL++ against the parse path to surface crashes before deployment\n\nAcceptance criteria: a crafted request that would segfault the current service returns 422 and the service continues to handle subsequent requests.","status":"in_progress","priority":1,"issue_type":"bug","assignee":"Bruno Postle","owner":"bruno@postle.net","created_at":"2026-04-24T16:22:41Z","created_by":"Bruno Postle","updated_at":"2026-04-24T16:25:41Z","started_at":"2026-04-24T16:25:41Z","dependency_count":0,"dependent_count":0,"comment_count":0}
{"id":"ifcurl-p60","title":"Viewer: no element click-to-identify — can't get GlobalId or properties","description":"There is no way to click on a mesh in the viewer and see which IFC element it represents. Reviewers can see geometry but cannot reference specific elements — they can't discover a GlobalId to put in a selector URL, can't see the element name/type/Psets, and can't say 'this specific wall' in a BCF or Forgejo comment. This breaks the collaboration workflow at its foundation: every useful feedback comment needs to identify specific elements. Need: click or hover on a surface to highlight it and show a properties panel (name, type, GlobalId, key Psets). The GlobalId should be copyable into the selector field or URL. The ThatOpen components library provides the tools for this (OBC.IfcRelationsIndexer, element property queries).","status":"closed","priority":1,"issue_type":"feature","assignee":"Bruno Postle","owner":"bruno@postle.net","created_at":"2026-04-24T11:02:33Z","created_by":"Bruno Postle","updated_at":"2026-04-24T11:47:23Z","started_at":"2026-04-24T11:14:57Z","closed_at":"2026-04-24T11:47:23Z","close_reason":"Closed","dependency_count":0,"dependent_count":0,"comment_count":0}
{"id":"ifcurl-4l7","title":"viewer.html: scale= URLs don't load in orthographic mode","description":"applyCameraParam() reads the camera= and fov= parameters from the URL and applies them, but never reads scale= or switches the camera to orthographic (parallel) projection. URLs like ?camera=...\u0026scale=50 will load in perspective mode instead of orthographic. The OBC OrthoPerspectiveCamera can switch modes, but the code path to set parallel projection and apply the scale value is missing from applyCameraParam.","status":"closed","priority":1,"issue_type":"bug","assignee":"Bruno Postle","owner":"bruno@postle.net","created_at":"2026-04-24T09:04:51Z","created_by":"Bruno Postle","updated_at":"2026-04-24T10:56:29Z","started_at":"2026-04-24T10:50:19Z","closed_at":"2026-04-24T10:56:29Z","close_reason":"Closed","dependency_count":0,"dependent_count":0,"comment_count":0}
{"id":"ifcurl-362","title":"Add missing server-config files: gitconfig-ifcmerge and gitattributes","description":"forgejo/README.md documents deploying two files that don't exist in the repo: forgejo/server-config/gitconfig-ifcmerge and forgejo/server-config/gitattributes. Both are required for the ifcmerge git merge driver to work in bare Forgejo repos. The README even includes 'sudo cp forgejo/server-config/gitattributes /etc/gitattributes' and 'sudo git config --system include.path .../gitconfig-ifcmerge', but the files themselves are absent. Anyone following the deployment instructions will hit a 'file not found' error. Need to create both files with the content described in the README.","status":"closed","priority":1,"issue_type":"bug","assignee":"Bruno Postle","owner":"bruno@postle.net","created_at":"2026-04-24T09:04:31Z","created_by":"Bruno Postle","updated_at":"2026-04-24T10:51:47Z","started_at":"2026-04-24T10:50:15Z","closed_at":"2026-04-24T10:51:47Z","close_reason":"Files already exist: forgejo/server-config/gitconfig-ifcmerge and gitattributes are both present and correct. Missed in initial review.","dependency_count":0,"dependent_count":0,"comment_count":0}

137
ifcurl/sandbox.py Normal file
View file

@ -0,0 +1,137 @@
# IFC URL — resolve and render ifc:// URLs
# Copyright (C) 2026 Bruno Postle <bruno@postle.net>
#
# This file is part of IFC URL.
#
# IFC URL is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# IFC URL is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with IFC URL. If not, see <http://www.gnu.org/licenses/>.
"""Subprocess sandbox for ifcopenshell operations.
ifcopenshell uses C++ bindings that can segfault on malformed or adversarially
crafted IFC data. Running untrusted IFC through the main service process
exposes it to DoS (segfault kills the worker) and potential RCE (memory
corruption).
``run_sandboxed`` executes a callable in a short-lived child process. If the
child exits with a fatal signal (SIGSEGV, SIGABRT, …) the parent raises
``SandboxCrashError`` instead of dying. If the child exceeds the configured
timeout, the parent raises ``SandboxTimeoutError`` and kills the child.
Resource limits
---------------
Set ``IFCURL_SANDBOX_MEMORY_MB`` to cap the child's virtual address space (in
MiB). Set ``IFCURL_SANDBOX_CPU_SECS`` to cap CPU time (in seconds). Both
default to 0 (no limit) so that out-of-the-box behaviour is unchanged; set
them in production to bound worst-case resource usage.
Timeout
-------
``IFCURL_SANDBOX_TIMEOUT`` (default 120 s) applies per-call. The timeout
should be longer than the longest expected render time so that slow-but-valid
models aren't rejected.
"""
from __future__ import annotations
import multiprocessing
import os
try:
import resource as _resource
_HAS_RESOURCE = True
except ImportError:
_HAS_RESOURCE = False # Windows
_SANDBOX_TIMEOUT: int = int(os.environ.get("IFCURL_SANDBOX_TIMEOUT", "120"))
_SANDBOX_MEMORY_MB: int = int(os.environ.get("IFCURL_SANDBOX_MEMORY_MB", "0"))
_SANDBOX_CPU_SECS: int = int(os.environ.get("IFCURL_SANDBOX_CPU_SECS", "0"))
class SandboxError(RuntimeError):
pass
class SandboxCrashError(SandboxError):
"""Child process was killed by a signal (segfault, abort, …)."""
class SandboxTimeoutError(SandboxError):
"""Child process exceeded the allowed wall-clock time."""
def _worker(conn: multiprocessing.connection.Connection, fn, args: tuple, kwargs: dict) -> None:
"""Entry point for the sandboxed child process."""
if _HAS_RESOURCE:
if _SANDBOX_MEMORY_MB > 0:
limit = _SANDBOX_MEMORY_MB * 1024 * 1024
_resource.setrlimit(_resource.RLIMIT_AS, (limit, limit))
if _SANDBOX_CPU_SECS > 0:
_resource.setrlimit(_resource.RLIMIT_CPU, (_SANDBOX_CPU_SECS, _SANDBOX_CPU_SECS))
try:
result = fn(*args, **kwargs)
conn.send(("ok", result))
except BaseException as exc:
try:
conn.send(("err", exc))
except Exception:
pass # if the exception is not picklable, the parent sees EOFError
finally:
conn.close()
def run_sandboxed(fn, *args, timeout: int = _SANDBOX_TIMEOUT, **kwargs):
"""Run ``fn(*args, **kwargs)`` in a child process.
:returns: The return value of *fn* on success.
:raises SandboxCrashError: If the child exits with a signal.
:raises SandboxTimeoutError: If the child exceeds *timeout* seconds.
:raises: Any exception raised by *fn* inside the child (re-raised as-is).
"""
parent_conn, child_conn = multiprocessing.Pipe(duplex=False)
proc = multiprocessing.Process(
target=_worker,
args=(child_conn, fn, args, kwargs),
daemon=True,
)
proc.start()
child_conn.close()
got_data = parent_conn.poll(timeout)
if not got_data:
parent_conn.close()
proc.terminate()
proc.join(5)
if proc.is_alive():
proc.kill()
proc.join()
raise SandboxTimeoutError(f"Subprocess timed out after {timeout}s")
try:
status, value = parent_conn.recv()
except EOFError:
parent_conn.close()
proc.join()
sig = -proc.exitcode if (proc.exitcode is not None and proc.exitcode < 0) else proc.exitcode
raise SandboxCrashError(f"Subprocess terminated without sending result (exit {sig})")
parent_conn.close()
proc.join()
if proc.exitcode is not None and proc.exitcode < 0:
raise SandboxCrashError(f"Subprocess killed by signal {-proc.exitcode}")
if status == "err":
raise value
return value

162
ifcurl/service.py
View file

@ -44,19 +44,84 @@ import time
from collections import OrderedDict
from pathlib import Path
import ifcopenshell
import ifcopenshell.util.selector
from fastapi import FastAPI, HTTPException
from fastapi.responses import Response
from platformdirs import user_cache_dir
from pydantic import BaseModel
from ifcurl import render as render_mod
from ifcurl.auth import get_token_for_host
from ifcurl.bcf import build_bcf
from ifcurl.git import fetch_ifc
from ifcurl.sandbox import SandboxCrashError, SandboxTimeoutError, run_sandboxed
from ifcurl.url import IfcUrl
# ---------------------------------------------------------------------------
# Subprocess pipeline functions (run inside run_sandboxed — no service imports)
# ---------------------------------------------------------------------------
def _sandboxed_pipeline(
ifc_bytes: bytes,
selector: str | None,
element_guids: frozenset[str] | None,
camera: tuple | None,
fov: float | None,
scale: float | None,
clips: list,
visibility: str,
) -> tuple[frozenset[str] | None, bytes]:
"""Parse, optionally resolve selection, and render in a child process.
Returns ``(new_guids, png_bytes)`` where *new_guids* is populated only when
*selector* was given (T3 cache miss path) so the caller can update the cache.
"""
import ifcopenshell
import ifcopenshell.util.selector
from ifcurl import render as render_mod
model = ifcopenshell.file.from_string(ifc_bytes.decode())
# T3 cache miss: run selector and return GUIDs for caching
new_guids: frozenset[str] | None = None
element_ids: list[int] | None = None
if selector is not None:
matched = list(ifcopenshell.util.selector.filter_elements(model, selector))
new_guids = frozenset(
e.GlobalId for e in matched if hasattr(e, "GlobalId") and e.GlobalId
)
elif element_guids is not None:
# T3 cache hit: convert cached GUIDs to step IDs for this model instance
ids = []
for guid in element_guids:
try:
ids.append(model.by_guid(guid).id())
except Exception:
pass
element_ids = ids if ids else None
png = render_mod.render(
model,
selector=selector,
element_ids=element_ids,
camera=camera,
fov=fov,
scale=scale,
clips=clips or None,
visibility=visibility,
)
return new_guids, png
def _sandboxed_select(ifc_bytes: bytes, selector: str) -> list[str]:
"""Parse and run selector in a child process. Returns list of GlobalIds."""
import ifcopenshell
import ifcopenshell.util.selector
model = ifcopenshell.file.from_string(ifc_bytes.decode())
matched = list(ifcopenshell.util.selector.filter_elements(model, selector))
return [e.GlobalId for e in matched if hasattr(e, "GlobalId") and e.GlobalId]
app = FastAPI(
title="ifcurl preview service",
description="Renders ifc:// URLs to PNG images for embedding in Gitea and other consumers.",
@ -155,18 +220,6 @@ def _t3_put(hexsha: str, path: str, selector: str, guids: frozenset[str]) -> Non
_t3_cache.popitem(last=False)
def _guids_to_step_ids(model: ifcopenshell.file, guids: frozenset[str]) -> list[int]:
"""Resolve a set of GlobalIds to step IDs in *model*."""
ids = []
for guid in guids:
try:
entity = model.by_guid(guid)
ids.append(entity.id())
except Exception:
pass # entity not present in this model version
return ids
# ---------------------------------------------------------------------------
# Tier 4: sha256(url) → PNG (filesystem, immutable refs only, no expiry)
# ---------------------------------------------------------------------------
@ -219,17 +272,6 @@ def _t4m_put(url: str, png: bytes) -> None:
_t4m_path(url).write_bytes(png)
# ---------------------------------------------------------------------------
# Helper: load model from bytes via a temp file
# ---------------------------------------------------------------------------
def _load_model(ifc_bytes: bytes) -> ifcopenshell.file:
try:
return ifcopenshell.file.from_string(ifc_bytes.decode())
except Exception as exc:
raise HTTPException(status_code=422, detail=f"Could not parse IFC file: {exc}") from exc
# ---------------------------------------------------------------------------
# Request / response models
# ---------------------------------------------------------------------------
@ -327,51 +369,45 @@ def preview(request: PreviewRequest) -> Response:
else:
_t2_put(hexsha, ifc_url.path, ifc_bytes)
# --- Load model ---
model = _load_model(ifc_bytes)
# --- Tier 3: resolve selector via GUID cache ---
element_ids: list[int] | None = None
selector_for_render: str | None = ifc_url.selector
# --- Tier 3: check GUID cache ---
# T3 hit → pass element_guids (child converts to step IDs), skip selector
# T3 miss → pass selector string (child runs filter_elements and returns GUIDs)
cached_guids: frozenset[str] | None = None
selector_for_sandbox: str | None = ifc_url.selector
element_guids: frozenset[str] | None = None
if ifc_url.selector:
cached_guids = _t3_get(hexsha, ifc_url.path, ifc_url.selector)
if cached_guids is not None:
# Convert cached GUIDs to step IDs in this model instance.
# Pass element_ids only (no selector) — filter_elements is skipped.
# Note: for 'isolate' mode this iterates all geometry rather than
# restricting the iterator; correctness is preserved, not optimal.
element_ids = _guids_to_step_ids(model, cached_guids)
selector_for_render = None
else:
# Run selector, cache GUIDs for future requests
try:
matched = list(ifcopenshell.util.selector.filter_elements(model, ifc_url.selector))
except Exception as exc:
raise HTTPException(status_code=422, detail=f"Invalid selector: {exc}") from exc
guids = frozenset(
e.GlobalId for e in matched if hasattr(e, "GlobalId") and e.GlobalId
)
_t3_put(hexsha, ifc_url.path, ifc_url.selector, guids)
# Fall through: render() will re-run filter_elements via selector_for_render
element_guids = cached_guids
selector_for_sandbox = None
# --- Render ---
# --- Sandboxed parse + select + render ---
try:
png_bytes = render_mod.render(
model,
selector=selector_for_render,
element_ids=element_ids,
camera=ifc_url.camera,
fov=ifc_url.fov,
scale=ifc_url.scale,
clips=ifc_url.clips or None,
visibility=ifc_url.visibility,
new_guids, png_bytes = run_sandboxed(
_sandboxed_pipeline,
ifc_bytes,
selector_for_sandbox,
element_guids,
ifc_url.camera,
ifc_url.fov,
ifc_url.scale,
ifc_url.clips or [],
ifc_url.visibility,
)
except SandboxCrashError as exc:
raise HTTPException(status_code=422, detail=f"IFC parse/render crashed: {exc}") from exc
except SandboxTimeoutError as exc:
raise HTTPException(status_code=503, detail=f"Render timed out: {exc}") from exc
except ImportError as exc:
raise HTTPException(status_code=503, detail=str(exc)) from exc
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
# --- Tier 3: populate GUID cache from selector run ---
if new_guids is not None and ifc_url.selector:
_t3_put(hexsha, ifc_url.path, ifc_url.selector, new_guids)
# --- Tier 4 / 4m: store PNG and return with appropriate cache headers ---
if ifc_url.is_mutable_ref():
_t4m_put(request.url, png_bytes)
@ -423,12 +459,14 @@ def bcf_export(request: BcfRequest) -> Response:
if cached is None:
_t2_put(hexsha, ifc_url.path, ifc_bytes)
model = _load_model(ifc_bytes)
try:
matched = list(ifcopenshell.util.selector.filter_elements(model, ifc_url.selector))
guids = run_sandboxed(_sandboxed_select, ifc_bytes, ifc_url.selector)
except SandboxCrashError as exc:
raise HTTPException(status_code=422, detail=f"IFC parse/select crashed: {exc}") from exc
except SandboxTimeoutError as exc:
raise HTTPException(status_code=503, detail=f"Select timed out: {exc}") from exc
except Exception as exc:
raise HTTPException(status_code=422, detail=f"Invalid selector: {exc}") from exc
guids = [e.GlobalId for e in matched if hasattr(e, "GlobalId") and e.GlobalId]
bcf_bytes = build_bcf(
camera=ifc_url.camera,

50
tests/test_sandbox.py Normal file
View file

@ -0,0 +1,50 @@
"""Tests for ifcurl.sandbox — subprocess isolation for ifcopenshell calls."""
from __future__ import annotations
import ctypes
import os
import sys
import pytest
from ifcurl.sandbox import (
SandboxCrashError,
SandboxTimeoutError,
run_sandboxed,
)
def _return_value(x):
return x
def _raise_value_error(msg):
raise ValueError(msg)
def _segfault():
ctypes.string_at(0) # dereference NULL → SIGSEGV
def _sleep_forever():
import time
time.sleep(9999)
class TestRunSandboxed:
def test_returns_value(self):
assert run_sandboxed(_return_value, 42) == 42
def test_propagates_exception(self):
with pytest.raises(ValueError, match="bad input"):
run_sandboxed(_raise_value_error, "bad input")
@pytest.mark.skipif(sys.platform == "win32", reason="SIGSEGV test requires Unix")
def test_crash_raises_sandbox_crash_error(self):
with pytest.raises(SandboxCrashError):
run_sandboxed(_segfault)
def test_timeout_raises_sandbox_timeout_error(self):
with pytest.raises(SandboxTimeoutError):
run_sandboxed(_sleep_forever, timeout=1)

54
tests/test_service.py
View file

@ -36,6 +36,20 @@ def clear_caches(tmp_path, monkeypatch):
configure_allowed_hosts(None)
@pytest.fixture(autouse=True)
def sync_sandbox(monkeypatch):
"""Run run_sandboxed synchronously in tests (no subprocess spawning).
The sandbox pipeline functions are called directly in the test process so
that unittest.mock patches on ifcurl.render.render and
ifcopenshell.util.selector.filter_elements are visible to them.
"""
monkeypatch.setattr(
"ifcurl.service.run_sandboxed",
lambda fn, *a, **kw: fn(*a),
)
@pytest.fixture()
def tmp_t4_cache(tmp_path):
"""Return the temp cache directory (already redirected by clear_caches)."""
@ -79,7 +93,7 @@ class TestPreviewRender:
def test_returns_png_content_type(self, model_with_geometry):
ifc_bytes = model_with_geometry.to_string().encode()
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
r = client.post("/preview", json={"url": MUTABLE_URL})
assert r.status_code == 200
assert r.headers["content-type"] == "image/png"
@ -93,7 +107,7 @@ class TestPreviewRender:
def test_render_error_returns_422(self, model_with_geometry):
ifc_bytes = model_with_geometry.to_string().encode()
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", side_effect=ValueError("No geometry")):
patch("ifcurl.render.render", side_effect=ValueError("No geometry")):
r = client.post("/preview", json={"url": MUTABLE_URL})
assert r.status_code == 422
@ -106,8 +120,8 @@ class TestPreviewRender:
class TestTier2Cache:
def test_bytes_cached_after_first_request(self, model_with_geometry):
ifc_bytes = model_with_geometry.to_string().encode()
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)) as mock_fetch, \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.render.render", return_value=FAKE_PNG):
client.post("/preview", json={"url": MUTABLE_URL})
assert (FAKE_HEXSHA, "model.ifc") in _t2_cache
@ -124,7 +138,7 @@ class TestTier2Cache:
# verify that tier-2 serves ifc_bytes from cache on the second call.
monkeypatch.setattr("ifcurl.service._T4M_TTL", 0)
with patch("ifcurl.service.fetch_ifc", counting_fetch), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
client.post("/preview", json={"url": MUTABLE_URL})
client.post("/preview", json={"url": MUTABLE_URL})
# fetch_ifc still called both times (for the commit hexsha),
@ -142,14 +156,14 @@ class TestTier3Cache:
def test_guids_cached_after_selector_request(self, model_with_geometry):
ifc_bytes = model_with_geometry.to_string().encode()
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
client.post("/preview", json={"url": SELECTOR_URL})
assert (FAKE_HEXSHA, "model.ifc", "IfcWall") in _t3_cache
def test_cached_guids_are_frozenset(self, model_with_geometry):
ifc_bytes = model_with_geometry.to_string().encode()
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
client.post("/preview", json={"url": SELECTOR_URL})
cached = _t3_cache[(FAKE_HEXSHA, "model.ifc", "IfcWall")]
assert isinstance(cached, frozenset)
@ -165,7 +179,7 @@ class TestTier3Cache:
return real_filter(model, selector)
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG), \
patch("ifcurl.render.render", return_value=FAKE_PNG), \
patch("ifcopenshell.util.selector.filter_elements", counting_filter):
client.post("/preview", json={"url": SELECTOR_URL})
client.post("/preview", json={"url": SELECTOR_URL})
@ -189,7 +203,7 @@ class TestTier4Cache:
return FAKE_PNG
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", counting_render):
patch("ifcurl.render.render", counting_render):
r1 = client.post("/preview", json={"url": IMMUTABLE_URL})
r2 = client.post("/preview", json={"url": IMMUTABLE_URL})
@ -207,7 +221,7 @@ class TestTier4Cache:
return FAKE_PNG
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", counting_render):
patch("ifcurl.render.render", counting_render):
client.post("/preview", json={"url": MUTABLE_URL})
client.post("/preview", json={"url": MUTABLE_URL})
@ -224,7 +238,7 @@ class TestTier4Cache:
monkeypatch.setattr("ifcurl.service._T4M_TTL", 0)
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", counting_render):
patch("ifcurl.render.render", counting_render):
client.post("/preview", json={"url": MUTABLE_URL})
client.post("/preview", json={"url": MUTABLE_URL})
@ -233,7 +247,7 @@ class TestTier4Cache:
def test_tier4_cache_file_exists_on_disk(self, tmp_t4_cache, model_with_geometry):
ifc_bytes = model_with_geometry.to_string().encode()
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
client.post("/preview", json={"url": IMMUTABLE_URL})
png_files = list(tmp_t4_cache.rglob("*.png"))
assert len(png_files) == 1
@ -255,7 +269,7 @@ class TestAuthentication:
return FAKE_HEXSHA, ifc_bytes
with patch("ifcurl.service.fetch_ifc", capturing_fetch), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
client.post("/preview", json={"url": MUTABLE_URL, "token": "mytoken123"})
assert received_token == ["mytoken123"]
@ -270,7 +284,7 @@ class TestAuthentication:
monkeypatch.setattr("ifcurl.service.get_token_for_host", lambda host: "config_token")
with patch("ifcurl.service.fetch_ifc", capturing_fetch), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
client.post("/preview", json={"url": MUTABLE_URL})
assert received_token == ["config_token"]
@ -285,7 +299,7 @@ class TestAuthentication:
monkeypatch.setattr("ifcurl.service.get_token_for_host", lambda host: "config_token")
with patch("ifcurl.service.fetch_ifc", capturing_fetch), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
client.post("/preview", json={"url": MUTABLE_URL, "token": "request_token"})
assert received_token == ["request_token"]
@ -300,7 +314,7 @@ class TestGetEndpoint:
def test_get_returns_png(self, model_with_geometry):
ifc_bytes = model_with_geometry.to_string().encode()
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
r = client.get("/preview", params={"url": MUTABLE_URL})
assert r.status_code == 200
assert r.headers["content-type"] == "image/png"
@ -323,7 +337,7 @@ class TestGetEndpoint:
return FAKE_HEXSHA, ifc_bytes
with patch("ifcurl.service.fetch_ifc", mock_fetch), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
r = client.get("/preview", params={"url": MUTABLE_URL, "token": "mytoken"})
assert r.status_code == 200
assert received["token"] == "mytoken"
@ -379,14 +393,14 @@ class TestSSRF:
configure_allowed_hosts({"example.com"})
ifc_bytes = model_with_geometry.to_string().encode()
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
r = client.post("/preview", json={"url": MUTABLE_URL})
assert r.status_code == 200
def test_no_allowlist_permits_public_host(self, model_with_geometry):
ifc_bytes = model_with_geometry.to_string().encode()
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
r = client.post("/preview", json={"url": MUTABLE_URL})
assert r.status_code == 200
@ -406,7 +420,7 @@ class TestSSRF:
configure_allowed_hosts({"localhost:3000"})
ifc_bytes = model_with_geometry.to_string().encode()
with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \
patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG):
patch("ifcurl.render.render", return_value=FAKE_PNG):
r = client.post("/preview", json={
"url": "ifc://localhost:3000/org/repo@heads/main?path=model.ifc"
})