Skip to content
43 changes: 41 additions & 2 deletions backend/geolibre_server/geolibre_server/app/conversion.py
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,10 @@ def shapefile_field_warnings(column_names):
_JOBS_LOCK = threading.Lock()
_RUNTIME_SETUP_LOCK = threading.Lock()
MAX_RETAINED_JOBS = 100
# Concurrent pending/running conversion (and raster) jobs. Finished jobs are
# retained up to MAX_RETAINED_JOBS; in-flight work is refused with HTTP 429 once
# this cap is hit so a burst of /run calls cannot spawn unbounded subprocesses.
MAX_IN_FLIGHT_JOBS = 8


class VectorToVectorRequest(BaseModel):
Expand Down Expand Up @@ -1033,6 +1037,11 @@ def _append_job_message(job_id: str, message: str) -> None:
)


def _count_in_flight_jobs_locked() -> int:
"""Return how many jobs are pending or running. Caller must hold ``_JOBS_LOCK``."""
return sum(1 for job in _JOBS.values() if job.status in {"pending", "running"})


def _evict_finished_jobs_locked() -> None:
"""Drop the oldest finished jobs once the retention cap is exceeded.

Expand Down Expand Up @@ -1122,7 +1131,25 @@ def _run_conversion_job(
outputs={output_name: {"path": output_path}} if output_path else {},
)
except Exception as exc:
_job_update(job_id, status="failed", error=str(exc))
# Mirror Whitebox: log the raw failure server-side and surface only a
# generic message. DuckDB/GDAL stderr often embeds absolute paths that
# must not reach the browser-proxied /jobs/{id} response — including
# via ``messages``, which accumulates every subprocess stdout line.
with _JOBS_LOCK:
leaked = list(_JOBS.get(job_id).messages) if job_id in _JOBS else []
logger.warning(
"Conversion job %s failed: %s; subprocess output=%r",
job_id,
exc,
leaked,
exc_info=True,
)
_job_update(
job_id,
status="failed",
error="Conversion failed. See the sidecar logs for details.",
messages=[],
)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Comment thread
giswqs marked this conversation as resolved.
Comment thread
giswqs marked this conversation as resolved.
# Remove a partial output so a retry starts clean and stale bytes do not
# confuse downstream tools.
output_path = params.get("output_path")
Expand All @@ -1148,6 +1175,11 @@ def _start_job(
job_id = str(uuid.uuid4())
now = _utc_now()
with _JOBS_LOCK:
if _count_in_flight_jobs_locked() >= MAX_IN_FLIGHT_JOBS:
raise HTTPException(
status_code=429,
detail="Too many conversion jobs in progress; try again shortly.",
)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
_JOBS[job_id] = JobState(
id=job_id,
status="pending",
Expand All @@ -1161,7 +1193,14 @@ def _start_job(
args=(job_id, script, params, output_name),
daemon=True,
)
thread.start()
try:
thread.start()
except RuntimeError:
# Drop the reserved pending slot so a failed Thread.start does not
# permanently consume an in-flight capacity slot (and force 429s).
with _JOBS_LOCK:
_JOBS.pop(job_id, None)
raise
# The worker may have already flipped the job to "running" by the time this
# lock is re-acquired, so callers must not assume the response is "pending".
with _JOBS_LOCK:
Expand Down
84 changes: 84 additions & 0 deletions backend/geolibre_server/geolibre_server/app/raster.py
Original file line number Diff line number Diff line change
Expand Up @@ -929,6 +929,7 @@ def model(hh):


_RASTER_CALC_SCRIPT = """
import ast
import json, re, sys

import numpy as np
Expand Down Expand Up @@ -1024,6 +1025,89 @@ def load_bands(path, letter, namespace, base):
"e": np.e,
}
namespace.update(safe_funcs)
# Band arrays are full ndarrays, so attribute access (``A.tofile(...)``,
# ``A.dump(...)``) would still write files even with an empty ``__builtins__``
# and no bare ``np``. Allowlist the expression AST before ``eval`` so only
# curated calls/operators run — and so list/bytes/tuple multiplication cannot
# allocate unbounded memory before the shape check.
try:
tree = ast.parse(expression, mode="eval")
except SyntaxError as exc:
raise SystemExit(f"Failed to evaluate expression: {exc}") from exc
allowed_calls = set(safe_funcs)
_allowed_nodes = (
ast.Expression,
ast.Name,
ast.Load,
ast.Constant,
ast.Call,
ast.BinOp,
ast.UnaryOp,
ast.BoolOp,
ast.Compare,
ast.IfExp,
ast.Add,
ast.Sub,
ast.Mult,
ast.Div,
ast.FloorDiv,
ast.Mod,
ast.Pow,
ast.UAdd,
ast.USub,
ast.Not,
ast.And,
ast.Or,
ast.Eq,
ast.NotEq,
ast.Lt,
ast.LtE,
ast.Gt,
ast.GtE,
ast.keyword,
)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
# ``**`` on two plain integers stays exact, so ``9**9**6`` builds a
# half-million-digit result and pins a core for tens of seconds before the shape
# check can reject it. Require the exponent to be a small numeric literal unless
# it derives from a band or a curated call, whose element-wise float power is
# bounded by the raster's shape. 64 is far past anything band math needs
# (squares, gamma, roots).
MAX_POW_EXPONENT = 64
for node in ast.walk(tree):
if not isinstance(node, _allowed_nodes):
raise SystemExit(
f"Expression may not use {type(node).__name__} "
"(band math only allows curated functions and operators)"
)
if isinstance(node, ast.Constant) and not isinstance(node.value, (int, float, bool)):
raise SystemExit(
"Expression may only use numeric constants "
"(band math only allows curated functions and operators)"
)
if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Pow):
exponent = node.right
while isinstance(exponent, ast.UnaryOp):
exponent = exponent.operand
band_backed = any(
isinstance(inner, (ast.Name, ast.Call)) for inner in ast.walk(exponent)
)
if not band_backed:
if not isinstance(exponent, ast.Constant):
raise SystemExit(
"Exponent must be a plain number or derive from a band "
f"(band math caps '**' at {MAX_POW_EXPONENT})"
)
if abs(exponent.value) > MAX_POW_EXPONENT:
raise SystemExit(
f"Exponent may not exceed {MAX_POW_EXPONENT} "
"(band math caps '**' to keep evaluation bounded)"
)
if isinstance(node, ast.Call):
# ``np.where(...)`` / ``A.tofile(...)`` parse as Attribute-backed calls;
# Attribute is already rejected above, but keep an explicit Name check.
if not isinstance(node.func, ast.Name) or node.func.id not in allowed_calls:
name = node.func.id if isinstance(node.func, ast.Name) else type(node.func).__name__
raise SystemExit(f"Call to '{name}' is not allowed in band math")
Comment thread
coderabbitai[bot] marked this conversation as resolved.
try:
with np.errstate(all="ignore"):
result = eval(expression, {"__builtins__": {}}, namespace)
Expand Down
47 changes: 46 additions & 1 deletion backend/geolibre_server/geolibre_server/app/whitebox.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel

from geolibre_server.vector_ops import MAX_FEATURES as MAX_LAYER_FEATURES

from . import conversion
from .runtime import (
RUNTIME_CATALOG_TIMEOUT_SECS,
Expand Down Expand Up @@ -118,6 +120,10 @@ class WhiteboxRunRequest(BaseModel):
_JOBS_LOCK = threading.Lock()
_RUNTIME_SETUP_LOCK = threading.Lock()
MAX_RETAINED_JOBS = 100
# Concurrent pending/running Whitebox jobs. Finished jobs are retained up to
# MAX_RETAINED_JOBS; in-flight work is refused with HTTP 429 once this cap is
# hit so a burst of /run calls cannot spawn unbounded tool sessions.
MAX_IN_FLIGHT_JOBS = 8


def _check_python_import(python_executable: str) -> None:
Expand Down Expand Up @@ -731,10 +737,19 @@ def _write_layer_input(param_name: str, layer: dict[str, Any], temp_paths: list[

Returns:
Path to the materialized input file.

Raises:
ValueError: When the layer payload is not GeoJSON, or exceeds the
shared feature cap used by vector/PostGIS/Sedona paths.
"""
geojson = layer.get("geojson")
if not isinstance(geojson, dict):
raise ValueError(f"Layer input for {param_name} does not contain GeoJSON.")
features = geojson.get("features") or []
if isinstance(features, list) and len(features) > MAX_LAYER_FEATURES:
raise ValueError(
f"Layer input for {param_name} exceeds the {MAX_LAYER_FEATURES}-feature limit"
)
folder = Path(tempfile.mkdtemp(prefix="geolibre-whitebox-input-"))
temp_paths.append(folder)
path = folder / f"{_safe_output_stem('input', param_name)}.geojson"
Expand Down Expand Up @@ -1138,15 +1153,38 @@ def _evict_finished_jobs_locked() -> None:
_JOBS.pop(job_id, None)


def _count_in_flight_jobs_locked() -> int:
"""Return how many jobs are pending or running. Caller must hold ``_JOBS_LOCK``."""
return sum(1 for job in _JOBS.values() if job.status in {"pending", "running"})


@router.post("/run")
def whitebox_run(request: WhiteboxRunRequest):
"""Start a background Whitebox tool run."""
tool_id = request.tool_id.strip()
if not tool_id:
raise HTTPException(status_code=400, detail="tool_id is required")
# Reject oversized embedded layers before enqueueing work, matching the
# 413 vector/PostGIS/Sedona feature cap (defense-in-depth also lives in
# ``_write_layer_input``).
for name, layer in request.layer_inputs.items():
geojson = layer.get("geojson") if isinstance(layer, dict) else None
if not isinstance(geojson, dict):
continue
features = geojson.get("features") or []
if isinstance(features, list) and len(features) > MAX_LAYER_FEATURES:
raise HTTPException(
status_code=413,
detail=(f"Layer input for {name} exceeds the {MAX_LAYER_FEATURES}-feature limit"),
)
job_id = str(uuid.uuid4())
now = _utc_now()
with _JOBS_LOCK:
if _count_in_flight_jobs_locked() >= MAX_IN_FLIGHT_JOBS:
raise HTTPException(
status_code=429,
detail="Too many Whitebox jobs in progress; try again shortly.",
)
_JOBS[job_id] = JobState(
id=job_id,
status="pending",
Expand All @@ -1156,7 +1194,14 @@ def whitebox_run(request: WhiteboxRunRequest):
)
_evict_finished_jobs_locked()
thread = threading.Thread(target=_run_job, args=(job_id, request), daemon=True)
thread.start()
try:
thread.start()
except RuntimeError:
# Drop the reserved pending slot so a failed Thread.start does not
# permanently consume an in-flight capacity slot (and force 429s).
with _JOBS_LOCK:
_JOBS.pop(job_id, None)
raise
with _JOBS_LOCK:
return _JOBS[job_id]

Expand Down
11 changes: 10 additions & 1 deletion backend/geolibre_server/geolibre_server/sedona_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,7 @@ def run_sql(sql: str, layers: Optional[list[dict]] = None) -> dict:
``rows`` and as GeoJSON in ``geojson``.

Raises:
SqlInputTooLarge: A layer exceeds :data:`MAX_FEATURES`.
SqlInputTooLarge: A layer or the query result exceeds :data:`MAX_FEATURES`.
ValueError: Invalid input.
Exception: Whatever SedonaDB raises for an invalid SQL statement.
"""
Expand All @@ -148,9 +148,18 @@ def run_sql(sql: str, layers: Optional[list[dict]] = None) -> dict:
connection.create_data_frame(gdf).to_view(name)

result = connection.sql(sql)
# Cap before materializing: to_pandas() would otherwise hold an
# unbounded cross-join / generate_series expansion in memory. Fetch one
# past the limit so overflow still raises SqlInputTooLarge.
result = result.limit(MAX_FEATURES + 1)
# to_pandas() returns a GeoDataFrame when the result has a geometry
# column, otherwise a plain DataFrame.
frame = result.to_pandas()
if len(frame) > MAX_FEATURES:
# Input registration already caps each layer, but a query can still
# expand rows (cross joins, generate_series, etc.). Bound the
# response the same way vector/PostGIS paths bound payloads.
raise SqlInputTooLarge(f"Query result exceeds the {MAX_FEATURES}-feature limit")
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Comment thread
giswqs marked this conversation as resolved.
columns = [str(column) for column in frame.columns]

geometry_column: Optional[str] = None
Expand Down
74 changes: 74 additions & 0 deletions backend/geolibre_server/tests/test_conversion.py
Original file line number Diff line number Diff line change
Expand Up @@ -459,3 +459,77 @@ def test_evict_finished_jobs_never_drops_running(monkeypatch) -> None:
_evict_finished_jobs_locked()
# Excess is 2, but only the one finished job is eligible for eviction.
assert set(jobs) == {"old_running", "pending"}


def test_start_job_rejects_when_in_flight_cap_reached(monkeypatch) -> None:
"""A new conversion job is refused with 429 once in-flight work is at the cap."""
monkeypatch.setattr(conversion, "MAX_IN_FLIGHT_JOBS", 1)
monkeypatch.setattr(
conversion,
"_JOBS",
{"busy": _job("busy", "running", "2026-01-01T00:00:00+00:00")},
)

with pytest.raises(HTTPException) as exc:
conversion._start_job("vector-to-geoparquet", "pass", {}, "output")
assert exc.value.status_code == 429
assert "Too many conversion jobs" in str(exc.value.detail)


def test_conversion_job_does_not_leak_error(monkeypatch, tmp_path: Path) -> None:
"""Failed conversion jobs store a generic error and scrub subprocess messages."""
job_id = "test-conversion-leak"
now = conversion._utc_now()
out = tmp_path / "out.parquet"
with conversion._JOBS_LOCK:
conversion._JOBS[job_id] = conversion.JobState(
id=job_id,
status="pending",
tool_id="vector-to-geoparquet",
created_at=now,
updated_at=now,
)

secret = "/secret/path/to/duckdb: boom traceback leak"

def _boom(*_args, **_kwargs):
# Simulate lines already streamed into messages before the failure —
# the realistic GDAL/DuckDB stderr path that GET /jobs/{id} would leak.
conversion._append_job_message(job_id, secret)
raise RuntimeError(secret)

monkeypatch.setattr(conversion, "_runtime_python", _boom)
try:
conversion._run_conversion_job(
job_id,
"pass",
{"output_path": str(out)},
"output",
)
job = conversion._JOBS[job_id]
assert job.status == "failed"
assert job.error == "Conversion failed. See the sidecar logs for details."
assert secret not in (job.error or "")
assert job.messages == []
assert all(secret not in message for message in job.messages)
finally:
with conversion._JOBS_LOCK:
conversion._JOBS.pop(job_id, None)


def test_start_job_rolls_back_when_thread_start_fails(monkeypatch) -> None:
"""A failed Thread.start must not leave a permanent pending job slot."""
monkeypatch.setattr(conversion, "_JOBS", {})
monkeypatch.setattr(conversion, "MAX_IN_FLIGHT_JOBS", 8)

class _BoomThread:
def __init__(self, *args, **kwargs): # noqa: ANN002, ANN003
pass

def start(self) -> None:
raise RuntimeError("thread start failed")

monkeypatch.setattr(conversion.threading, "Thread", _BoomThread)
with pytest.raises(RuntimeError, match="thread start failed"):
conversion._start_job("vector-to-geoparquet", "pass", {}, "output")
assert conversion._JOBS == {}
Loading
Loading