fix(sidecar): harden band-math eval, job caps, and result bounds

backend/geolibre_server/geolibre_server/app/conversion.py

@@ -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):
@@ -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.
 
@@ -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=[],
+        )
         # Remove a partial output so a retry starts clean and stale bytes do not
         # confuse downstream tools.
         output_path = params.get("output_path")
@@ -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.",
+            )
         _JOBS[job_id] = JobState(
             id=job_id,
             status="pending",
@@ -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:

backend/geolibre_server/geolibre_server/app/raster.py

@@ -929,6 +929,7 @@ def model(hh):
 
 
 _RASTER_CALC_SCRIPT = """
+import ast
 import json, re, sys
 
 import numpy as np
@@ -1024,6 +1025,64 @@ 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,
+)
+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.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")
 try:
     with np.errstate(all="ignore"):
         result = eval(expression, {"__builtins__": {}}, namespace)

backend/geolibre_server/geolibre_server/app/whitebox.py

@@ -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,
@@ -88,6 +90,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:
@@ -701,10 +707,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"
@@ -1041,15 +1056,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",
@@ -1059,7 +1097,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]
 

backend/geolibre_server/geolibre_server/sedona_ops.py

@@ -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.
     """
@@ -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")
         columns = [str(column) for column in frame.columns]
 
         geometry_column: Optional[str] = None

backend/geolibre_server/tests/test_conversion.py

@@ -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 == {}