From 4eae454c8b5cefe023ce476eac6dea08c221c8ba Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Wed, 14 Oct 2026 12:01:40 -0400 Subject: [PATCH 1/6] Remove code-carrying transform card from feature catalog seed; keep v1 snapshot Signed-off-by: Dhaval Patel --- src/couchdb/.allowed_datafiles | 3 +- .../shared/tsfm/feature_catalog.json | 36 ----- .../shared/tsfm/feature_catalog_orig.json | 46 +++++++ .../tsfm/tests/test_feature_catalog_store.py | 126 ------------------ 4 files changed, 48 insertions(+), 163 deletions(-) create mode 100644 src/couchdb/scenarios_data/shared/tsfm/feature_catalog_orig.json diff --git a/src/couchdb/.allowed_datafiles b/src/couchdb/.allowed_datafiles index f2dcdb9c3..ec38afc0d 100644 --- a/src/couchdb/.allowed_datafiles +++ b/src/couchdb/.allowed_datafiles @@ -16,4 +16,5 @@ src/couchdb/scenarios_data/shared/iot/motor_01.json src/couchdb/scenarios_data/shared/work_order/workorders.csv src/couchdb/scenarios_data/shared/tsfm/feature_catalog.json src/couchdb/scenarios_data/shared/tsfm/model_catalog.json -src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json \ No newline at end of file +src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json +src/couchdb/scenarios_data/shared/tsfm/feature_catalog_orig.json \ No newline at end of file diff --git a/src/couchdb/scenarios_data/shared/tsfm/feature_catalog.json b/src/couchdb/scenarios_data/shared/tsfm/feature_catalog.json index 853c616ac..c88033ea7 100644 --- a/src/couchdb/scenarios_data/shared/tsfm/feature_catalog.json +++ b/src/couchdb/scenarios_data/shared/tsfm/feature_catalog.json @@ -1,40 +1,4 @@ [ - { - "feature_id": "efe_time_robust_norm_v1", - "name": "Invertible robust normalization", - "modality": "timeseries", - "interface": "fit_transform_inverse", - "class_name": "Transformation", - "invertible": true, - "code": "import numpy as np\nclass Transformation:\n \"\"\"Invertible robust per-channel normalization (median/IQR) for forecasting inputs.\"\"\"\n def fit(self, X, metadata):\n X = np.asarray(X, dtype=float)\n center = np.median(X, axis=0)\n q1, q3 = np.percentile(X, 25, axis=0), np.percentile(X, 75, axis=0)\n scale = np.where((q3 - q1) > 1e-8, (q3 - q1), 1.0)\n return {\"center\": center, \"scale\": scale}\n def transform(self, X, state):\n X = np.asarray(X, dtype=float)\n return (X - state[\"center\"]) / state[\"scale\"]\n def inverse_transform(self, Y, state):\n Y = np.asarray(Y, dtype=float)\n return Y * state[\"scale\"] + state[\"center\"]\n", - "provenance": "evolved", - "method": "EFE-Time", - "parent_feature_id": null, - "generation": 0, - "target_task": "tsfm_forecasting", - "target_model": null, - "dataset": null, - "output_type": "normalized_series", - "columns_added": [], - "columns_dropped": [], - "validity": { - "entry_points": true, - "no_inplace": true, - "invertible_ok": true, - "leakage_checked": true - }, - "tags": [ - "timeseries", - "EFE-Time", - "Future State Prediction" - ], - "status": "active", - "version": "1", - "created_by": "seed", - "created_at": "2026-06-20T00:00:00+00:00", - "kind": "transform", - "description": "Invertible per-channel robust normalization (median/IQR) for forecasting inputs." - }, { "feature_id": "abs_2nd_diff_mean", "name": "abs_2nd_diff_mean", diff --git a/src/couchdb/scenarios_data/shared/tsfm/feature_catalog_orig.json b/src/couchdb/scenarios_data/shared/tsfm/feature_catalog_orig.json new file mode 100644 index 000000000..853c616ac --- /dev/null +++ b/src/couchdb/scenarios_data/shared/tsfm/feature_catalog_orig.json @@ -0,0 +1,46 @@ +[ + { + "feature_id": "efe_time_robust_norm_v1", + "name": "Invertible robust normalization", + "modality": "timeseries", + "interface": "fit_transform_inverse", + "class_name": "Transformation", + "invertible": true, + "code": "import numpy as np\nclass Transformation:\n \"\"\"Invertible robust per-channel normalization (median/IQR) for forecasting inputs.\"\"\"\n def fit(self, X, metadata):\n X = np.asarray(X, dtype=float)\n center = np.median(X, axis=0)\n q1, q3 = np.percentile(X, 25, axis=0), np.percentile(X, 75, axis=0)\n scale = np.where((q3 - q1) > 1e-8, (q3 - q1), 1.0)\n return {\"center\": center, \"scale\": scale}\n def transform(self, X, state):\n X = np.asarray(X, dtype=float)\n return (X - state[\"center\"]) / state[\"scale\"]\n def inverse_transform(self, Y, state):\n Y = np.asarray(Y, dtype=float)\n return Y * state[\"scale\"] + state[\"center\"]\n", + "provenance": "evolved", + "method": "EFE-Time", + "parent_feature_id": null, + "generation": 0, + "target_task": "tsfm_forecasting", + "target_model": null, + "dataset": null, + "output_type": "normalized_series", + "columns_added": [], + "columns_dropped": [], + "validity": { + "entry_points": true, + "no_inplace": true, + "invertible_ok": true, + "leakage_checked": true + }, + "tags": [ + "timeseries", + "EFE-Time", + "Future State Prediction" + ], + "status": "active", + "version": "1", + "created_by": "seed", + "created_at": "2026-06-20T00:00:00+00:00", + "kind": "transform", + "description": "Invertible per-channel robust normalization (median/IQR) for forecasting inputs." + }, + { + "feature_id": "abs_2nd_diff_mean", + "name": "abs_2nd_diff_mean", + "description": "Mean absolute second difference.", + "kind": "extractor", + "extractor_name": "abs_2nd_diff_mean", + "status": "active" + } +] \ No newline at end of file diff --git a/src/servers/tsfm/tests/test_feature_catalog_store.py b/src/servers/tsfm/tests/test_feature_catalog_store.py index bd96bb405..4d1358734 100644 --- a/src/servers/tsfm/tests/test_feature_catalog_store.py +++ b/src/servers/tsfm/tests/test_feature_catalog_store.py @@ -42,34 +42,6 @@ def seeded_feature_store() -> MemoryStore: store.put(feature_store.collection_name(), _feature_doc(doc)) return store - -def test_feature_store_lists_searches_and_versions_catalog_cards(): - store = seeded_feature_store() - - transforms = feature_store.find_features(store, kind="transform") - assert [doc["feature_id"] for doc in transforms] == ["efe_time_robust_norm_v1"] - - extractors = feature_store.list_extractors(store) - assert [doc["feature_id"] for doc in extractors] == ["abs_2nd_diff_mean"] - - matches = feature_store.search(store, "normalization") - assert [doc["feature_id"] for doc in matches] == ["efe_time_robust_norm_v1"] - - successor = feature_store.new_version( - store, - "efe_time_robust_norm_v1", - {"description": "updated normalization"}, - ) - assert successor["parent_feature_id"] == "efe_time_robust_norm_v1" - assert successor["version"] == "2" - assert feature_store.get_feature(store, "efe_time_robust_norm_v1")[ - "status" - ] == "superseded" - assert feature_store.get_lineage(store, successor["feature_id"])[ - "ancestors" - ] == ["efe_time_robust_norm_v1"] - - def test_feature_store_reads_collection_name_from_env(monkeypatch): monkeypatch.setenv("FEATURE_CATALOG_DBNAME", "custom_feature_catalog") store = MemoryStore() @@ -101,101 +73,3 @@ def test_register_feature_rejects_in_place_transform(): with pytest.raises(Exception, match="mutated"): feature_store.register_feature(store, bad) - - -@pytest.mark.anyio -async def test_feature_catalog_mcp_tools(monkeypatch): - monkeypatch.setattr(tsfm_main, "_FEATURE_STORE", seeded_feature_store()) - - listed = await call_tool( - tsfm_main.mcp, "list_features", {"kind": "transform"} - ) - assert listed["message"] == "listed 1 transform feature with status active." - assert listed["features"][0]["feature_id"] == "efe_time_robust_norm_v1" - - found = await call_tool( - tsfm_main.mcp, "search_features", {"text": "second difference"} - ) - assert ( - found["message"] - == "found 1 feature matching text 'second difference' with status active." - ) - assert found["features"][0]["feature_id"] == "abs_2nd_diff_mean" - - card = await call_tool( - tsfm_main.mcp, - "get_feature", - {"feature_id": "efe_time_robust_norm_v1"}, - ) - assert ( - card["message"] - == "found transform feature efe_time_robust_norm_v1 with status active." - ) - - lineage = await call_tool( - tsfm_main.mcp, - "get_feature_lineage", - {"feature_id": "efe_time_robust_norm_v1"}, - ) - assert ( - lineage["message"] - == "lineage for feature efe_time_robust_norm_v1 has 0 ancestors and " - "0 descendants; root is efe_time_robust_norm_v1." - ) - assert lineage["feature_id"] == "efe_time_robust_norm_v1" - - feature = { - "feature_id": "smoke_identity", - "interface": "fit_transform", - "code": ( - "class Transformation:\n" - " def fit(self, X, metadata):\n" - " return {}\n" - " def transform(self, X, state):\n" - " return X.copy()\n" - ), - } - registered = await call_tool( - tsfm_main.mcp, - "register_feature", - {"feature": feature}, - ) - assert registered["message"] == "registered feature smoke_identity." - assert registered["card"]["feature_id"] == "smoke_identity" - - updated = await call_tool( - tsfm_main.mcp, - "update_feature", - { - "feature_id": "smoke_identity", - "fields": {"description": "updated identity transform"}, - }, - ) - assert updated["message"] == "updated feature smoke_identity with 1 field." - assert updated["description"] == "updated identity transform" - - versioned = await call_tool( - tsfm_main.mcp, - "new_feature_version", - { - "feature_id": "smoke_identity", - "fields": {"description": "successor identity transform"}, - "new_feature_id": "smoke_identity_v2", - }, - ) - assert ( - versioned["message"] - == "created feature version smoke_identity_v2 from smoke_identity." - ) - assert versioned["feature_id"] == "smoke_identity_v2" - - deprecated = await call_tool( - tsfm_main.mcp, - "deprecate_feature", - {"feature_id": "smoke_identity_v2", "reason": "covered by successor"}, - ) - assert ( - deprecated["message"] - == "deprecated feature smoke_identity_v2. Reason: covered by successor." - ) - assert deprecated["status"] == "deprecated" From c1f406d505313c9504b13579ca61170615245e72 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Wed, 14 Oct 2026 12:26:57 -0400 Subject: [PATCH 2/6] tsfm: no execution of agent-supplied code - resolver: allowlist the estimator classes importable from a dotted path - param_space: route introspect through the same allowlist - feature_runner: refuse instead of exec(); unregister register_feature / new_feature_version - main: default HF_HUB_OFFLINE=1 so weights load only from the preloaded cache - tests: test_no_code_exec.py; update feature catalog tests Signed-off-by: Dhaval Patel --- src/servers/tsfm/engine/feature_runner.py | 22 +--- src/servers/tsfm/main.py | 7 +- src/servers/tsfm/reasoning/param_space.py | 6 +- src/servers/tsfm/substrate/resolver.py | 48 +++++++- src/servers/tsfm/tests/conftest.py | 3 + .../tsfm/tests/test_feature_catalog_store.py | 6 +- src/servers/tsfm/tests/test_no_code_exec.py | 109 ++++++++++++++++++ 7 files changed, 173 insertions(+), 28 deletions(-) create mode 100644 src/servers/tsfm/tests/test_no_code_exec.py diff --git a/src/servers/tsfm/engine/feature_runner.py b/src/servers/tsfm/engine/feature_runner.py index 7e46f5e0b..4005cef95 100644 --- a/src/servers/tsfm/engine/feature_runner.py +++ b/src/servers/tsfm/engine/feature_runner.py @@ -8,7 +8,7 @@ from __future__ import annotations import copy -from typing import Any, Dict, Optional +from typing import Optional class FeatureValidationError(Exception): @@ -16,24 +16,8 @@ class FeatureValidationError(Exception): def load_program(code: str, class_name: str = "Transformation"): - """Exec a stored program string and return its Transformation class. - - NOTE: executes code. In the benchmark/agent setting run this in a sandbox - (restricted globals, subprocess, or container). Here we exec in a fresh namespace - with a minimal import surface, matching how EFE executes candidate programs. - """ - ns: Dict[str, Any] = {} - try: - exec(code, ns) # noqa: S102 - intentional; sandbox in production - except Exception as e: - raise FeatureValidationError(f"program failed to import: {e}") - cls = ns.get(class_name) - if cls is None: - raise FeatureValidationError(f"no class '{class_name}' defined") - for m in ("fit", "transform"): - if not callable(getattr(cls, m, None)): - raise FeatureValidationError(f"missing required method '{m}'") - return cls + """Disabled: the server never executes feature-card source (it used to exec() it).""" + raise FeatureValidationError("executing feature-card code is disabled in the TSFM server") def _no_inplace(before, after) -> bool: diff --git a/src/servers/tsfm/main.py b/src/servers/tsfm/main.py index 7107240e2..69092f91d 100644 --- a/src/servers/tsfm/main.py +++ b/src/servers/tsfm/main.py @@ -15,6 +15,11 @@ import logging import sys import os + +# Load weights only from the preloaded HF cache, never a repo the agent names: MOIRAIForecaster +# unpickles model.ckpt (weights_only=False) from any non-Salesforce repo. Set before any +# huggingface_hub import; override with HF_HUB_OFFLINE=0 for local development. +os.environ.setdefault("HF_HUB_OFFLINE", "1") from typing import List, Optional, Union import pandas as pd @@ -1329,7 +1334,6 @@ def get_feature(feature_id: str) -> Union[CardResult, ErrorResult]: return ErrorResult(error=str(exc)) -@mcp.tool(title="Register Feature") def register_feature( feature: dict, overwrite: bool = False, @@ -1426,7 +1430,6 @@ def deprecate_feature( return ErrorResult(error=str(exc)) -@mcp.tool(title="New Feature Version") def new_feature_version( feature_id: str, fields: Optional[dict] = None, diff --git a/src/servers/tsfm/reasoning/param_space.py b/src/servers/tsfm/reasoning/param_space.py index 8a46f2024..4324a2e0f 100644 --- a/src/servers/tsfm/reasoning/param_space.py +++ b/src/servers/tsfm/reasoning/param_space.py @@ -14,15 +14,15 @@ from __future__ import annotations -import importlib import inspect from typing import Dict, Optional def introspect(sktime_class: str) -> dict: """Constructor params (name → default/required/type) + sktime example param sets.""" - module, cls = sktime_class.rsplit(".", 1) - Est = getattr(importlib.import_module(module), cls) + from ..substrate.resolver import _import_target + + Est = _import_target(sktime_class) sig = inspect.signature(Est.__init__) _empty = inspect.Parameter.empty params: Dict[str, dict] = {} diff --git a/src/servers/tsfm/substrate/resolver.py b/src/servers/tsfm/substrate/resolver.py index afd72b129..9797913ee 100644 --- a/src/servers/tsfm/substrate/resolver.py +++ b/src/servers/tsfm/substrate/resolver.py @@ -26,8 +26,54 @@ _TARGET_KEY = "_target_" +# Only these classes may be imported from a dotted path. Paths reach this function from agent +# input (inline sktime_class, nested _target_, catalog cards the agent can rewrite), so an +# open import ran any callable, e.g. subprocess.Popen. Operators extend the list with +# TSFM_EXTRA_ESTIMATOR_CLASSES (comma-separated exact paths). +ALLOWED_CLASSES = frozenset({ + "sktime.forecasting.ttm.TinyTimeMixerForecaster", + "sktime.forecasting.chronos.ChronosForecaster", + "sktime.forecasting.timesfm.TimesFMForecaster", + "sktime.forecasting.timesfm2.TimesFM2Forecaster", + "sktime.forecasting.moirai.MOIRAIForecaster", + "sktime.forecasting.moirai2.Moirai2Forecaster", + "sktime.forecasting.toto2.Toto2Forecaster", + "sktime.forecasting.kronos.KronosForecaster", + "sktime.forecasting.momentfm.MomentFMForecaster", + "sktime.forecasting.timemoe.TimeMoEForecaster", + "sktime.forecasting.patch_tst.PatchTSTForecaster", + "sktime.forecasting.patch_tsmixer.PatchTSMixerForecaster", + "sktime.forecasting.naive.NaiveForecaster", + "sktime.forecasting.theta.ThetaForecaster", + "sktime.forecasting.arima.ARIMA", + "sktime.forecasting.arima.AutoARIMA", + "sktime.forecasting.conformal.ConformalIntervals", + "sktime.detection.tspulse.TSPulseAnomalyDetector", + "sktime.detection.hampel.HampelDetector", + "sktime.detection.lof.SubLOF", + "sktime.detection.adapters._pyod.PyODDetector", + "sktime.detection.dummy._zero_an.ZeroAnomalies", + "pyod.models.iforest.IForest", + "sktime.classification.foundation_models.tspulse.TSPulseClassifier", + "sklearn.ensemble.RandomForestClassifier", + "sktime.clustering.k_means.TimeSeriesKMeans", + "sktime.clustering.k_medoids.TimeSeriesKMedoids", + "sktime.clustering.dbscan.TimeSeriesDBSCAN", + "sktime.clustering.kernel_k_means.TimeSeriesKernelKMeans", + "sktime.clustering.agglomerative.TimeSeriesAgglomerativeClustering", +}) + + def _import_target(path: str): - """Import a dotted path 'pkg.module.Name' -> the class/callable Name.""" + """Import an allowlisted dotted path 'pkg.module.Name' -> the class Name. + + The check runs before the import, so a refused path never loads its module. + """ + import os + + extra = {p.strip() for p in os.environ.get("TSFM_EXTRA_ESTIMATOR_CLASSES", "").split(",")} + if path not in ALLOWED_CLASSES and path not in extra - {""}: + raise ValueError(f"'{path}' is not an allowed estimator class; use a catalog model_id") module, name = path.rsplit(".", 1) return getattr(importlib.import_module(module), name) diff --git a/src/servers/tsfm/tests/conftest.py b/src/servers/tsfm/tests/conftest.py index 7f1a051b6..776a9e6bb 100644 --- a/src/servers/tsfm/tests/conftest.py +++ b/src/servers/tsfm/tests/conftest.py @@ -11,6 +11,9 @@ import pytest os.environ["TSFM_STORE"] = "memory" +os.environ.setdefault( + "TSFM_EXTRA_ESTIMATOR_CLASSES", "servers.tsfm.tests.fake_hf_forecaster.FakeTTMForecaster" +) _SEEDS = os.environ.get("TSFM_SEEDS_DIR") or os.path.normpath( os.path.join( diff --git a/src/servers/tsfm/tests/test_feature_catalog_store.py b/src/servers/tsfm/tests/test_feature_catalog_store.py index 4d1358734..5e66fe503 100644 --- a/src/servers/tsfm/tests/test_feature_catalog_store.py +++ b/src/servers/tsfm/tests/test_feature_catalog_store.py @@ -53,10 +53,10 @@ def test_feature_store_reads_collection_name_from_env(monkeypatch): assert feature_store.get_feature(store, doc["feature_id"])["feature_id"] == doc[ "feature_id" ] - assert feature_store.find_features(store, kind="transform") + assert feature_store.find_features(store, kind="extractor") -def test_register_feature_rejects_in_place_transform(): +def test_register_feature_never_executes_code(): store = MemoryStore() bad = { "feature_id": "bad_inplace", @@ -71,5 +71,5 @@ def test_register_feature_rejects_in_place_transform(): ), } - with pytest.raises(Exception, match="mutated"): + with pytest.raises(Exception, match="disabled"): feature_store.register_feature(store, bad) diff --git a/src/servers/tsfm/tests/test_no_code_exec.py b/src/servers/tsfm/tests/test_no_code_exec.py new file mode 100644 index 000000000..6a0dcd031 --- /dev/null +++ b/src/servers/tsfm/tests/test_no_code_exec.py @@ -0,0 +1,109 @@ +"""The TSFM server must never run agent-supplied code. + +Each test sends the payload from the reported run (subprocess.Popen through a JSON spec) down +one route and checks both that the call errors and that the canary file was never written. +""" + +import asyncio +import json +import os +import sys +import tempfile +import time +import uuid + +import numpy as np + +from ..io import refs +from ..main import mcp + + +def _call(name, args): + content, _ = asyncio.run(mcp.call_tool(name, args)) + out = json.loads(content[0].text) + return out.get("result", out) if isinstance(out, dict) else out + + +def _popen(): + canary = os.path.join(tempfile.gettempdir(), f"tsfm_canary_{uuid.uuid4().hex}") + spec = {"_target_": "subprocess.Popen", + "params": {"args": [sys.executable, "-c", f"open({canary!r}, 'w')"]}} + return canary, spec + + +def _assert_blocked(result, canary): + time.sleep(0.5) # Popen is asynchronous + assert "error" in json.dumps(result), result + assert not os.path.exists(canary), "agent-supplied code executed" + + +def _series(): + return refs.materialize_iot(np.sin(np.arange(120) / 6.0), asset_id="noexec") + + +def _forecast(recipe): + return _call("run_recipe", {"dataset_path": _series(), "timestamp_column": "timestamp", + "target_columns": ["value"], "recipe": recipe}) + + +def test_inline_class_is_refused(): + canary, spec = _popen() + _assert_blocked(_forecast({"estimator": {"sktime_class": spec["_target_"], + "params": spec["params"]}}), canary) + + +def test_nested_target_is_refused(): + canary, spec = _popen() + _assert_blocked(_forecast({"estimator": { + "sktime_class": "sktime.forecasting.naive.NaiveForecaster", + "params": {"strategy": spec}}}), canary) + + +def test_finetune_block_is_refused(): + canary, spec = _popen() + _assert_blocked(_forecast({"estimator": {"model_id": "ttm_512_96"}, + "finetune": {"x": spec}}), canary) + + +def test_run_plan_step_is_refused(): + canary, spec = _popen() + _assert_blocked(_call("run_plan", {"plan_spec": {"steps": [{ + "id": "a", "task": "anomaly", "args": {"data_ref": _series()}, + "recipe": {"estimator": {"sktime_class": spec["_target_"], + "params": spec["params"]}}}]}}), canary) + + +def test_rewritten_catalog_card_is_refused(): + canary, spec = _popen() + _call("update_model", {"model_id": "ttm_512_96", + "fields": {"sktime_class": spec["_target_"], + "params": spec["params"]}}) + try: + _assert_blocked(_forecast({"estimator": {"model_id": "ttm_512_96"}}), canary) + finally: + from .conftest import _seed_into + from .. import main + _seed_into(main._STORE) # restore the seeded card for later tests + + +def test_feature_code_is_never_executed(): + names = {t.name for t in asyncio.run(mcp.list_tools())} + assert "register_feature" not in names and "new_feature_version" not in names + from ..engine.feature_runner import FeatureValidationError, load_program + + canary = os.path.join(tempfile.gettempdir(), f"tsfm_canary_{uuid.uuid4().hex}") + try: + load_program(f"open({canary!r}, 'w')") + except FeatureValidationError: + pass + assert not os.path.exists(canary) + + +def test_hub_downloads_default_to_offline(): + assert os.environ.get("HF_HUB_OFFLINE") == "1" + + +def test_allowlisted_estimator_still_runs(): + r = _forecast({"estimator": {"sktime_class": "sktime.forecasting.naive.NaiveForecaster"}, + "fh": [1, 2, 3]}) + assert r.get("status") == "success", r From 1f30953ec281d5060946b55542c766fd6552001d Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Wed, 14 Oct 2026 13:20:47 -0400 Subject: [PATCH 3/6] revised code Data confinment Signed-off-by: Dhaval Patel --- src/servers/tsfm/engine/composition.py | 4 +- src/servers/tsfm/io/refs.py | 38 +++++++++++++-- src/servers/tsfm/tests/test_no_code_exec.py | 53 +++++++++++++++++++++ 3 files changed, 90 insertions(+), 5 deletions(-) diff --git a/src/servers/tsfm/engine/composition.py b/src/servers/tsfm/engine/composition.py index 9fc079993..af3830adf 100644 --- a/src/servers/tsfm/engine/composition.py +++ b/src/servers/tsfm/engine/composition.py @@ -204,7 +204,9 @@ def _save_checkpoint(fc, path: str) -> str: f"{type(fc).__name__} has none. A card can only point at weights via " f"params.model_path, so there is no card-loadable way to store this estimator." ) - path = os.path.abspath(os.path.expanduser(path)) + from ..io import refs + + path = refs._path(str(path)) # confined to the data roots os.makedirs(path, exist_ok=True) model.save_pretrained(path) return path diff --git a/src/servers/tsfm/io/refs.py b/src/servers/tsfm/io/refs.py index 71d5102db..fc1e4075d 100644 --- a/src/servers/tsfm/io/refs.py +++ b/src/servers/tsfm/io/refs.py @@ -11,6 +11,8 @@ import json import os +import re +import tempfile import uuid from typing import List, Optional @@ -20,8 +22,36 @@ WORKDIR = os.environ.get("TSFM_WORKDIR", "/tmp/tsfm_work") +def _data_roots() -> List[str]: + """Directories a file pointer may resolve into: this server's WORKDIR, the system temp dir + and /tmp (code_exec workspaces, other servers' outputs), the Harbor shared workspace, and + artifacts/output (where agents save fine-tuned checkpoints via recipe save_to). + TSFM_DATA_ROOTS (os.pathsep-separated) replaces the default.""" + env = os.environ.get("TSFM_DATA_ROOTS") + roots = env.split(os.pathsep) if env else [ + WORKDIR, tempfile.gettempdir(), "/tmp", "/workspace-share", + os.path.join(os.getcwd(), "artifacts", "output")] # where agents save checkpoints + return [os.path.realpath(r) for r in roots if r] + + def _path(ref: str) -> str: - return ref[7:] if ref.startswith("file://") else ref + """Resolve a file pointer, confined to the data roots. + + Unconfined, every dataset_path read any file the server could open, e.g. + /proc/self/environ (the CouchDB password). realpath() follows symlinks, so a link + planted in a workspace cannot escape; URLs (http://, s3://) are refused. + """ + if "://" in ref and not ref.startswith("file://"): + raise ValueError("only local file pointers are supported") + p = os.path.realpath(ref[7:] if ref.startswith("file://") else ref) + if not any(p == r or p.startswith(r + os.sep) for r in _data_roots()): + raise ValueError(f"file pointer is outside the allowed data directories: {ref}") + return p + + +def _safe_name(name) -> str: + """Output file stem; plan step ids are agent-supplied, so strip path characters.""" + return re.sub(r"[^A-Za-z0-9_.-]", "_", str(name)).lstrip(".") or "result" def _ensure_workdir(): @@ -91,7 +121,7 @@ def load_series( def write_series(obj, *, name: str = "result") -> str: """Write a forecast/series/frame to a file pointer and return the ref.""" _ensure_workdir() - p = os.path.join(WORKDIR, f"{name}_{uuid.uuid4().hex[:8]}.csv") + p = os.path.join(WORKDIR, f"{_safe_name(name)}_{uuid.uuid4().hex[:8]}.csv") (obj.to_frame() if isinstance(obj, pd.Series) else pd.DataFrame(obj)).to_csv( p, index=False ) @@ -100,7 +130,7 @@ def write_series(obj, *, name: str = "result") -> str: def write_json(obj: dict, *, name: str = "result") -> str: _ensure_workdir() - p = os.path.join(WORKDIR, f"{name}_{uuid.uuid4().hex[:8]}.json") + p = os.path.join(WORKDIR, f"{_safe_name(name)}_{uuid.uuid4().hex[:8]}.json") json.dump(obj, open(p, "w"), indent=2, default=str) return f"file://{p}" @@ -124,6 +154,6 @@ def materialize_iot( df.insert( 0, "timestamp", pd.date_range("2020-01-01", periods=len(arr), freq=freq) ) - p = os.path.join(WORKDIR, f"iot_{asset_id}.csv") + p = os.path.join(WORKDIR, f"iot_{_safe_name(asset_id)}.csv") df.to_csv(p, index=False) return f"file://{p}" diff --git a/src/servers/tsfm/tests/test_no_code_exec.py b/src/servers/tsfm/tests/test_no_code_exec.py index 6a0dcd031..e6490cc61 100644 --- a/src/servers/tsfm/tests/test_no_code_exec.py +++ b/src/servers/tsfm/tests/test_no_code_exec.py @@ -107,3 +107,56 @@ def test_allowlisted_estimator_still_runs(): r = _forecast({"estimator": {"sktime_class": "sktime.forecasting.naive.NaiveForecaster"}, "fh": [1, 2, 3]}) assert r.get("status") == "success", r + + +# File pointers: the read-back channel ------------------------------------------------------ +def test_file_pointer_outside_data_roots_is_refused(): + for p in ("/proc/self/environ", "file:///etc/passwd", "/etc/hostname", + os.path.join(refs.WORKDIR, "..", "..", "etc", "passwd")): + r = _call("data_quality", {"dataset_path": p}) + assert "error" in r, (p, r) + assert "COUCHDB" not in json.dumps(r) + + +def test_url_pointer_is_refused(): + for p in ("http://couchdb:5984/_all_dbs", "s3://bucket/x.csv"): + r = _call("data_quality", {"dataset_path": p}) + assert "only local file pointers" in r.get("error", ""), r + + +def test_valid_csv_outside_roots_and_symlink_to_it_are_refused(monkeypatch): + # a readable CSV outside the roots: the old code returned its contents + outside = tempfile.mkdtemp() + target = os.path.join(outside, "secret.csv") + with open(target, "w") as fh: + fh.write("timestamp,value\n2020-01-01,1\n2020-01-02,2\n") + monkeypatch.setenv("TSFM_DATA_ROOTS", refs._ensure_workdir()) + link = os.path.join(refs.WORKDIR, f"link_{uuid.uuid4().hex}.csv") + os.symlink(target, link) + try: + for p in (target, link): + r = _call("data_quality", {"dataset_path": p}) + assert "outside the allowed data directories" in r.get("error", ""), (p, r) + finally: + os.unlink(link) + + +def test_file_pointer_inside_data_roots_still_works(): + assert "error" not in _call("data_quality", {"dataset_path": _series()}) + + +def test_plan_step_id_cannot_write_outside_workdir(): + assert refs._safe_name("../../etc/evil") == "_.._etc_evil" + out = refs.write_json({}, name="../../evil") + assert os.path.realpath(refs._path(out)).startswith(os.path.realpath(refs.WORKDIR)) + + +def test_agent_checkpoint_folder_is_allowed_and_repo_code_is_not(): + # save_to "artifacts/output/..." is how agents persist a fine-tune; the code next to it is not + assert refs._path("artifacts/output/tuned_models/x").endswith("artifacts/output/tuned_models/x") + for p in ("artifacts/tsfm_models", "src/servers/tsfm/main.py", "artifacts/output/../../src"): + try: + refs._path(p) + raise AssertionError(f"{p} accepted") + except ValueError: + pass From 08e817bee3accce7c37241d86a7a358f680a0450 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Wed, 14 Oct 2026 22:26:37 -0400 Subject: [PATCH 4/6] tsfm: address PR 633 review (allowlist transformers/regressors, open-then-verify reads, confine training_args) Signed-off-by: Dhaval Patel --- src/servers/tsfm/io/refs.py | 45 +++++++- src/servers/tsfm/main.py | 4 +- src/servers/tsfm/substrate/resolver.py | 47 +++++++++ src/servers/tsfm/tests/test_no_code_exec.py | 107 ++++++++++++++++++++ 4 files changed, 199 insertions(+), 4 deletions(-) diff --git a/src/servers/tsfm/io/refs.py b/src/servers/tsfm/io/refs.py index fc1e4075d..a3fe69ba1 100644 --- a/src/servers/tsfm/io/refs.py +++ b/src/servers/tsfm/io/refs.py @@ -49,6 +49,48 @@ def _path(ref: str) -> str: return p +def open_data(ref: str): + """Open a file pointer for reading, verified AFTER opening. + + _path() checks the path, but the file could be swapped between that check and the open + (e.g. a parent folder replaced by a symlink), and a FIFO passes the check and then blocks + the reader forever. So: open without blocking, require a regular file, and check where the + OPENED file really lives. Callers read from the returned handle, never re-open the path. + """ + import stat + + p = _path(ref) + fd = os.open(p, os.O_RDONLY | os.O_NONBLOCK | getattr(os, "O_NOFOLLOW", 0)) + try: + if not stat.S_ISREG(os.fstat(fd).st_mode): + raise ValueError(f"file pointer is not a regular file: {ref}") + real = os.path.realpath(_fd_path(fd)) + if not any(real == r or real.startswith(r + os.sep) for r in _data_roots()): + raise ValueError(f"file pointer is outside the allowed data directories: {ref}") + return os.fdopen(fd, "rb") + except BaseException: + os.close(fd) + raise + + +def read_frame(ref: str) -> pd.DataFrame: + """Read a CSV / Parquet file pointer through open_data().""" + with open_data(ref) as fh: + if ref.endswith((".parquet", ".pq")): + return pd.read_parquet(fh) + return pd.read_csv(fh) + + +def _fd_path(fd: int) -> str: + """Path of an open file descriptor: /proc on Linux, F_GETPATH on macOS.""" + proc = f"/proc/self/fd/{fd}" + if os.path.exists(proc): + return os.readlink(proc) + import fcntl + + return fcntl.fcntl(fd, fcntl.F_GETPATH, b"\0" * 1024).rstrip(b"\0").decode() + + def _safe_name(name) -> str: """Output file stem; plan step ids are agent-supplied, so strip path characters.""" return re.sub(r"[^A-Za-z0-9_.-]", "_", str(name)).lstrip(".") or "result" @@ -80,8 +122,7 @@ def load_series( Column selection: `channels` picks a subset (multivariate); `value_col` picks the single value column (univariate). If neither is given, every numeric column is kept. """ - p = _path(data_ref) - df = pd.read_parquet(p) if p.endswith((".parquet", ".pq")) else pd.read_csv(p) + df = read_frame(data_ref) # locate the time column (explicit name wins; else auto-detect by common names) if time_col is None: diff --git a/src/servers/tsfm/main.py b/src/servers/tsfm/main.py index 69092f91d..120d856aa 100644 --- a/src/servers/tsfm/main.py +++ b/src/servers/tsfm/main.py @@ -272,7 +272,7 @@ def data_quality( if not dataset_path.strip(): return ErrorResult(error="dataset_path is required") try: - df = pd.read_csv(refs._path(dataset_path)) + df = refs.read_frame(dataset_path) nan = _dq._df_nan_stats(df) out = _dq._efficient_nan_removal(df) cleaned = out["df_filter"] @@ -1769,7 +1769,7 @@ def run_tabular_recipe( if bad: return ErrorResult(error=bad) try: - df = pd.read_csv(refs._path(dataset_path)) + df = refs.read_frame(dataset_path) y = None if label_column: if label_column not in df.columns: diff --git a/src/servers/tsfm/substrate/resolver.py b/src/servers/tsfm/substrate/resolver.py index 9797913ee..f803dcd90 100644 --- a/src/servers/tsfm/substrate/resolver.py +++ b/src/servers/tsfm/substrate/resolver.py @@ -61,6 +61,27 @@ "sktime.clustering.dbscan.TimeSeriesDBSCAN", "sktime.clustering.kernel_k_means.TimeSeriesKernelKMeans", "sktime.clustering.agglomerative.TimeSeriesAgglomerativeClustering", + # forecasting `transforms` (TransformedTargetForecaster steps) + "sktime.transformations.series.detrend.Detrender", + "sktime.transformations.series.detrend.Deseasonalizer", + "sktime.transformations.series.difference.Differencer", + "sktime.transformations.series.boxcox.LogTransformer", + "sktime.transformations.series.boxcox.BoxCoxTransformer", + "sktime.transformations.series.exponent.ExponentTransformer", + "sktime.transformations.series.impute.Imputer", + "sktime.transformations.series.adapt.TabularToSeriesAdaptor", + "sktime.forecasting.trend.PolynomialTrendForecaster", # Detrender's default forecaster + "sklearn.preprocessing.StandardScaler", + "sklearn.preprocessing.MinMaxScaler", + # run_tabular_recipe transforms (panel -> features) + "sktime.transformations.series.summarize.SummaryTransformer", + "sktime.transformations.panel.catch22.Catch22", + "sktime.transformations.panel.rocket.MiniRocket", + # run_tabular_recipe estimators: tsfm_regression / tsfm_clustering + "sklearn.ensemble.RandomForestRegressor", + "sklearn.ensemble.GradientBoostingRegressor", + "sklearn.linear_model.Ridge", + "sklearn.cluster.KMeans", }) @@ -108,9 +129,35 @@ def resolve(card: dict): raise ValueError(f"card '{card.get('model_id')}' has no sktime_class") Est = _import_target(path) params = {k: _build(v) for k, v in (card.get("params") or {}).items()} + if isinstance(params.get("training_args"), dict): + params["training_args"] = _safe_training_args(params["training_args"]) return Est(**params) +# HF TrainingArguments keys that name a path or an outside service. output_dir is forced to a +# server-chosen folder; the rest are refused, so a fine-tune writes (and pickles) only there. +_TRAINING_ARGS_REFUSED = frozenset({ + "logging_dir", "resume_from_checkpoint", "deepspeed", "fsdp_config", "accelerator_config", + "push_to_hub", "hub_model_id", "hub_token", "hub_strategy", "hub_private_repo", + "hub_always_push", "run_name", +}) + + +def _safe_training_args(ta: dict) -> dict: + import os + import uuid + + from ..io import refs + + bad = sorted(k for k in ta if k in _TRAINING_ARGS_REFUSED or k.startswith("hub_")) + if bad: + raise ValueError(f"training_args may not set {bad}") + out = dict(ta) + out["output_dir"] = os.path.join(refs._ensure_workdir(), "trainer", uuid.uuid4().hex[:8]) + out["report_to"] = "none" + return out + + def discover(scitype: str, filter_tags: Optional[dict] = None) -> List[str]: """sktime registry as live model discovery (installed estimators).""" from sktime.registry import all_estimators diff --git a/src/servers/tsfm/tests/test_no_code_exec.py b/src/servers/tsfm/tests/test_no_code_exec.py index e6490cc61..d6a0e57bb 100644 --- a/src/servers/tsfm/tests/test_no_code_exec.py +++ b/src/servers/tsfm/tests/test_no_code_exec.py @@ -160,3 +160,110 @@ def test_agent_checkpoint_folder_is_allowed_and_repo_code_is_not(): raise AssertionError(f"{p} accepted") except ValueError: pass + + +# PR #633 review ---------------------------------------------------------------------------- +import pandas as pd + +from ..substrate import resolver as R + + +def _tab_panel(numeric_label=False): + rng = np.random.RandomState(0) + X = rng.randn(30, 16) + df = pd.DataFrame(X, columns=[f"f{i}" for i in range(16)]) + df["label"] = X.mean(axis=1) if numeric_label else [0, 1] * 15 + p = os.path.join(refs._ensure_workdir(), f"tab_{uuid.uuid4().hex[:6]}.csv") + df.to_csv(p, index=False) + return p + + +def test_forecasting_transforms_still_run(): + r = _forecast({"estimator": {"sktime_class": "sktime.forecasting.naive.NaiveForecaster"}, + "transforms": [ + {"sktime_class": "sktime.transformations.series.detrend.Detrender"}, + {"sktime_class": "sktime.transformations.series.adapt.TabularToSeriesAdaptor", + "params": {"transformer": { + "_target_": "sklearn.preprocessing.StandardScaler"}}}], + "fh": [1, 2, 3]}) + assert r.get("status") == "success", r + + +def test_tabular_regression_still_runs(): + r = _call("run_tabular_recipe", {"dataset_path": _tab_panel(numeric_label=True), + "label_column": "label", + "recipe": {"task": "tsfm_regression", + "estimator": {"sktime_class": "sklearn.ensemble.RandomForestRegressor", + "params": {"n_estimators": 10}}}}) + assert r.get("status") == "success", r + + +def test_tabular_sktime_transform_still_runs(): + r = _call("run_tabular_recipe", {"dataset_path": _tab_panel(), "label_column": "label", + "recipe": {"task": "tsfm_classification", + "estimator": {"sktime_class": "sklearn.ensemble.RandomForestClassifier", + "params": {"n_estimators": 10}}, + "transforms": [{"sktime_class": + "sktime.transformations.series.summarize.SummaryTransformer"}]}}) + assert r.get("status") == "success", r + + +def test_fifo_is_refused_without_blocking(): + import threading + + fifo = os.path.join(refs._ensure_workdir(), f"pipe_{uuid.uuid4().hex[:6]}.csv") + os.mkfifo(fifo) + out = {} + t = threading.Thread(target=lambda: out.update( + r=_call("data_quality", {"dataset_path": fifo})), daemon=True) + try: + t.start() + t.join(10) + assert not t.is_alive(), "reading a FIFO blocked the server" + assert "not a regular file" in out["r"].get("error", ""), out + finally: + os.unlink(fifo) + + +def test_open_file_is_verified_after_open(monkeypatch): + # Simulate the race: the path check passes, but what gets opened lives outside the roots. + outside = os.path.join(tempfile.mkdtemp(), "secret.csv") + with open(outside, "w") as fh: + fh.write("timestamp,value\n2020-01-01,1\n") + monkeypatch.setenv("TSFM_DATA_ROOTS", refs._ensure_workdir()) + monkeypatch.setattr(refs, "_path", lambda ref: outside) + try: + refs.open_data("whatever.csv") + raise AssertionError("opened a file outside the roots") + except ValueError as exc: + assert "outside the allowed data directories" in str(exc) + + +def test_training_args_output_dir_is_forced_under_workdir(): + ta = R._safe_training_args({"output_dir": "/etc/evil", "num_train_epochs": 1, + "report_to": ["wandb"]}) + assert os.path.realpath(ta["output_dir"]).startswith(os.path.realpath(refs.WORKDIR)) + assert ta["report_to"] == "none" and ta["num_train_epochs"] == 1 + for key in ("logging_dir", "resume_from_checkpoint", "deepspeed", "push_to_hub", + "hub_model_id"): + try: + R._safe_training_args({key: "/etc/x"}) + raise AssertionError(f"{key} accepted") + except ValueError: + pass + + +def test_training_args_are_sanitized_on_every_resolve(): + built = {} + + class _Spy: + def __init__(self, training_args=None): + built["ta"] = training_args + + real = R._import_target + R._import_target = lambda path: _Spy + try: + R.resolve({"sktime_class": "x", "params": {"training_args": {"output_dir": "/etc/x"}}}) + finally: + R._import_target = real + assert os.path.realpath(built["ta"]["output_dir"]).startswith(os.path.realpath(refs.WORKDIR)) From ebd792c917684cd249024b4aa7e391b042d6ec16 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Wed, 14 Oct 2026 22:35:35 -0400 Subject: [PATCH 5/6] Revert "tsfm: address PR 633 review (allowlist transformers/regressors, open-then-verify reads, confine training_args)" This reverts commit 08e817bee3accce7c37241d86a7a358f680a0450. Signed-off-by: Dhaval Patel --- src/servers/tsfm/io/refs.py | 45 +------- src/servers/tsfm/main.py | 4 +- src/servers/tsfm/substrate/resolver.py | 47 --------- src/servers/tsfm/tests/test_no_code_exec.py | 107 -------------------- 4 files changed, 4 insertions(+), 199 deletions(-) diff --git a/src/servers/tsfm/io/refs.py b/src/servers/tsfm/io/refs.py index a3fe69ba1..fc1e4075d 100644 --- a/src/servers/tsfm/io/refs.py +++ b/src/servers/tsfm/io/refs.py @@ -49,48 +49,6 @@ def _path(ref: str) -> str: return p -def open_data(ref: str): - """Open a file pointer for reading, verified AFTER opening. - - _path() checks the path, but the file could be swapped between that check and the open - (e.g. a parent folder replaced by a symlink), and a FIFO passes the check and then blocks - the reader forever. So: open without blocking, require a regular file, and check where the - OPENED file really lives. Callers read from the returned handle, never re-open the path. - """ - import stat - - p = _path(ref) - fd = os.open(p, os.O_RDONLY | os.O_NONBLOCK | getattr(os, "O_NOFOLLOW", 0)) - try: - if not stat.S_ISREG(os.fstat(fd).st_mode): - raise ValueError(f"file pointer is not a regular file: {ref}") - real = os.path.realpath(_fd_path(fd)) - if not any(real == r or real.startswith(r + os.sep) for r in _data_roots()): - raise ValueError(f"file pointer is outside the allowed data directories: {ref}") - return os.fdopen(fd, "rb") - except BaseException: - os.close(fd) - raise - - -def read_frame(ref: str) -> pd.DataFrame: - """Read a CSV / Parquet file pointer through open_data().""" - with open_data(ref) as fh: - if ref.endswith((".parquet", ".pq")): - return pd.read_parquet(fh) - return pd.read_csv(fh) - - -def _fd_path(fd: int) -> str: - """Path of an open file descriptor: /proc on Linux, F_GETPATH on macOS.""" - proc = f"/proc/self/fd/{fd}" - if os.path.exists(proc): - return os.readlink(proc) - import fcntl - - return fcntl.fcntl(fd, fcntl.F_GETPATH, b"\0" * 1024).rstrip(b"\0").decode() - - def _safe_name(name) -> str: """Output file stem; plan step ids are agent-supplied, so strip path characters.""" return re.sub(r"[^A-Za-z0-9_.-]", "_", str(name)).lstrip(".") or "result" @@ -122,7 +80,8 @@ def load_series( Column selection: `channels` picks a subset (multivariate); `value_col` picks the single value column (univariate). If neither is given, every numeric column is kept. """ - df = read_frame(data_ref) + p = _path(data_ref) + df = pd.read_parquet(p) if p.endswith((".parquet", ".pq")) else pd.read_csv(p) # locate the time column (explicit name wins; else auto-detect by common names) if time_col is None: diff --git a/src/servers/tsfm/main.py b/src/servers/tsfm/main.py index 120d856aa..69092f91d 100644 --- a/src/servers/tsfm/main.py +++ b/src/servers/tsfm/main.py @@ -272,7 +272,7 @@ def data_quality( if not dataset_path.strip(): return ErrorResult(error="dataset_path is required") try: - df = refs.read_frame(dataset_path) + df = pd.read_csv(refs._path(dataset_path)) nan = _dq._df_nan_stats(df) out = _dq._efficient_nan_removal(df) cleaned = out["df_filter"] @@ -1769,7 +1769,7 @@ def run_tabular_recipe( if bad: return ErrorResult(error=bad) try: - df = refs.read_frame(dataset_path) + df = pd.read_csv(refs._path(dataset_path)) y = None if label_column: if label_column not in df.columns: diff --git a/src/servers/tsfm/substrate/resolver.py b/src/servers/tsfm/substrate/resolver.py index f803dcd90..9797913ee 100644 --- a/src/servers/tsfm/substrate/resolver.py +++ b/src/servers/tsfm/substrate/resolver.py @@ -61,27 +61,6 @@ "sktime.clustering.dbscan.TimeSeriesDBSCAN", "sktime.clustering.kernel_k_means.TimeSeriesKernelKMeans", "sktime.clustering.agglomerative.TimeSeriesAgglomerativeClustering", - # forecasting `transforms` (TransformedTargetForecaster steps) - "sktime.transformations.series.detrend.Detrender", - "sktime.transformations.series.detrend.Deseasonalizer", - "sktime.transformations.series.difference.Differencer", - "sktime.transformations.series.boxcox.LogTransformer", - "sktime.transformations.series.boxcox.BoxCoxTransformer", - "sktime.transformations.series.exponent.ExponentTransformer", - "sktime.transformations.series.impute.Imputer", - "sktime.transformations.series.adapt.TabularToSeriesAdaptor", - "sktime.forecasting.trend.PolynomialTrendForecaster", # Detrender's default forecaster - "sklearn.preprocessing.StandardScaler", - "sklearn.preprocessing.MinMaxScaler", - # run_tabular_recipe transforms (panel -> features) - "sktime.transformations.series.summarize.SummaryTransformer", - "sktime.transformations.panel.catch22.Catch22", - "sktime.transformations.panel.rocket.MiniRocket", - # run_tabular_recipe estimators: tsfm_regression / tsfm_clustering - "sklearn.ensemble.RandomForestRegressor", - "sklearn.ensemble.GradientBoostingRegressor", - "sklearn.linear_model.Ridge", - "sklearn.cluster.KMeans", }) @@ -129,35 +108,9 @@ def resolve(card: dict): raise ValueError(f"card '{card.get('model_id')}' has no sktime_class") Est = _import_target(path) params = {k: _build(v) for k, v in (card.get("params") or {}).items()} - if isinstance(params.get("training_args"), dict): - params["training_args"] = _safe_training_args(params["training_args"]) return Est(**params) -# HF TrainingArguments keys that name a path or an outside service. output_dir is forced to a -# server-chosen folder; the rest are refused, so a fine-tune writes (and pickles) only there. -_TRAINING_ARGS_REFUSED = frozenset({ - "logging_dir", "resume_from_checkpoint", "deepspeed", "fsdp_config", "accelerator_config", - "push_to_hub", "hub_model_id", "hub_token", "hub_strategy", "hub_private_repo", - "hub_always_push", "run_name", -}) - - -def _safe_training_args(ta: dict) -> dict: - import os - import uuid - - from ..io import refs - - bad = sorted(k for k in ta if k in _TRAINING_ARGS_REFUSED or k.startswith("hub_")) - if bad: - raise ValueError(f"training_args may not set {bad}") - out = dict(ta) - out["output_dir"] = os.path.join(refs._ensure_workdir(), "trainer", uuid.uuid4().hex[:8]) - out["report_to"] = "none" - return out - - def discover(scitype: str, filter_tags: Optional[dict] = None) -> List[str]: """sktime registry as live model discovery (installed estimators).""" from sktime.registry import all_estimators diff --git a/src/servers/tsfm/tests/test_no_code_exec.py b/src/servers/tsfm/tests/test_no_code_exec.py index d6a0e57bb..e6490cc61 100644 --- a/src/servers/tsfm/tests/test_no_code_exec.py +++ b/src/servers/tsfm/tests/test_no_code_exec.py @@ -160,110 +160,3 @@ def test_agent_checkpoint_folder_is_allowed_and_repo_code_is_not(): raise AssertionError(f"{p} accepted") except ValueError: pass - - -# PR #633 review ---------------------------------------------------------------------------- -import pandas as pd - -from ..substrate import resolver as R - - -def _tab_panel(numeric_label=False): - rng = np.random.RandomState(0) - X = rng.randn(30, 16) - df = pd.DataFrame(X, columns=[f"f{i}" for i in range(16)]) - df["label"] = X.mean(axis=1) if numeric_label else [0, 1] * 15 - p = os.path.join(refs._ensure_workdir(), f"tab_{uuid.uuid4().hex[:6]}.csv") - df.to_csv(p, index=False) - return p - - -def test_forecasting_transforms_still_run(): - r = _forecast({"estimator": {"sktime_class": "sktime.forecasting.naive.NaiveForecaster"}, - "transforms": [ - {"sktime_class": "sktime.transformations.series.detrend.Detrender"}, - {"sktime_class": "sktime.transformations.series.adapt.TabularToSeriesAdaptor", - "params": {"transformer": { - "_target_": "sklearn.preprocessing.StandardScaler"}}}], - "fh": [1, 2, 3]}) - assert r.get("status") == "success", r - - -def test_tabular_regression_still_runs(): - r = _call("run_tabular_recipe", {"dataset_path": _tab_panel(numeric_label=True), - "label_column": "label", - "recipe": {"task": "tsfm_regression", - "estimator": {"sktime_class": "sklearn.ensemble.RandomForestRegressor", - "params": {"n_estimators": 10}}}}) - assert r.get("status") == "success", r - - -def test_tabular_sktime_transform_still_runs(): - r = _call("run_tabular_recipe", {"dataset_path": _tab_panel(), "label_column": "label", - "recipe": {"task": "tsfm_classification", - "estimator": {"sktime_class": "sklearn.ensemble.RandomForestClassifier", - "params": {"n_estimators": 10}}, - "transforms": [{"sktime_class": - "sktime.transformations.series.summarize.SummaryTransformer"}]}}) - assert r.get("status") == "success", r - - -def test_fifo_is_refused_without_blocking(): - import threading - - fifo = os.path.join(refs._ensure_workdir(), f"pipe_{uuid.uuid4().hex[:6]}.csv") - os.mkfifo(fifo) - out = {} - t = threading.Thread(target=lambda: out.update( - r=_call("data_quality", {"dataset_path": fifo})), daemon=True) - try: - t.start() - t.join(10) - assert not t.is_alive(), "reading a FIFO blocked the server" - assert "not a regular file" in out["r"].get("error", ""), out - finally: - os.unlink(fifo) - - -def test_open_file_is_verified_after_open(monkeypatch): - # Simulate the race: the path check passes, but what gets opened lives outside the roots. - outside = os.path.join(tempfile.mkdtemp(), "secret.csv") - with open(outside, "w") as fh: - fh.write("timestamp,value\n2020-01-01,1\n") - monkeypatch.setenv("TSFM_DATA_ROOTS", refs._ensure_workdir()) - monkeypatch.setattr(refs, "_path", lambda ref: outside) - try: - refs.open_data("whatever.csv") - raise AssertionError("opened a file outside the roots") - except ValueError as exc: - assert "outside the allowed data directories" in str(exc) - - -def test_training_args_output_dir_is_forced_under_workdir(): - ta = R._safe_training_args({"output_dir": "/etc/evil", "num_train_epochs": 1, - "report_to": ["wandb"]}) - assert os.path.realpath(ta["output_dir"]).startswith(os.path.realpath(refs.WORKDIR)) - assert ta["report_to"] == "none" and ta["num_train_epochs"] == 1 - for key in ("logging_dir", "resume_from_checkpoint", "deepspeed", "push_to_hub", - "hub_model_id"): - try: - R._safe_training_args({key: "/etc/x"}) - raise AssertionError(f"{key} accepted") - except ValueError: - pass - - -def test_training_args_are_sanitized_on_every_resolve(): - built = {} - - class _Spy: - def __init__(self, training_args=None): - built["ta"] = training_args - - real = R._import_target - R._import_target = lambda path: _Spy - try: - R.resolve({"sktime_class": "x", "params": {"training_args": {"output_dir": "/etc/x"}}}) - finally: - R._import_target = real - assert os.path.realpath(built["ta"]["output_dir"]).startswith(os.path.realpath(refs.WORKDIR)) From 6b036094107aded296cd2cb29e7bc12d5cfe3de1 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Wed, 14 Oct 2026 22:42:59 -0400 Subject: [PATCH 6/6] tsfm: allowlist the transformers and tabular estimators the recipe tools advertise Signed-off-by: Dhaval Patel --- src/servers/tsfm/substrate/resolver.py | 36 ++++++++++++++ src/servers/tsfm/tests/test_no_code_exec.py | 54 +++++++++++++++++++++ 2 files changed, 90 insertions(+) diff --git a/src/servers/tsfm/substrate/resolver.py b/src/servers/tsfm/substrate/resolver.py index 9797913ee..3f3953078 100644 --- a/src/servers/tsfm/substrate/resolver.py +++ b/src/servers/tsfm/substrate/resolver.py @@ -61,6 +61,42 @@ "sktime.clustering.dbscan.TimeSeriesDBSCAN", "sktime.clustering.kernel_k_means.TimeSeriesKernelKMeans", "sktime.clustering.agglomerative.TimeSeriesAgglomerativeClustering", + # forecasting `transforms` (TransformedTargetForecaster steps) and their nested parts + "sktime.transformations.series.detrend.Detrender", + "sktime.transformations.series.detrend.Deseasonalizer", + "sktime.transformations.series.detrend.ConditionalDeseasonalizer", + "sktime.transformations.series.difference.Differencer", + "sktime.transformations.series.boxcox.LogTransformer", + "sktime.transformations.series.boxcox.BoxCoxTransformer", + "sktime.transformations.series.exponent.ExponentTransformer", + "sktime.transformations.series.exponent.SqrtTransformer", + "sktime.transformations.series.impute.Imputer", + "sktime.transformations.series.outlier_detection.HampelFilter", + "sktime.transformations.series.adapt.TabularToSeriesAdaptor", + "sktime.forecasting.trend.PolynomialTrendForecaster", + "sktime.forecasting.trend.TrendForecaster", + "sklearn.preprocessing.StandardScaler", + "sklearn.preprocessing.MinMaxScaler", + "sklearn.preprocessing.RobustScaler", + # run_tabular_recipe transforms + "sktime.transformations.series.summarize.SummaryTransformer", + "sktime.transformations.panel.catch22.Catch22", + "sktime.transformations.panel.rocket.Rocket", + "sktime.transformations.panel.rocket.MiniRocket", + "sktime.transformations.panel.rocket.MiniRocketMultivariate", + # run_tabular_recipe estimators: regression / classification / clustering + "sklearn.ensemble.RandomForestRegressor", + "sklearn.ensemble.ExtraTreesRegressor", + "sklearn.ensemble.GradientBoostingRegressor", + "sklearn.linear_model.LinearRegression", + "sklearn.linear_model.Ridge", + "sklearn.linear_model.Lasso", + "sklearn.neighbors.KNeighborsRegressor", + "sklearn.ensemble.ExtraTreesClassifier", + "sklearn.ensemble.GradientBoostingClassifier", + "sklearn.linear_model.LogisticRegression", + "sklearn.neighbors.KNeighborsClassifier", + "sklearn.cluster.KMeans", }) diff --git a/src/servers/tsfm/tests/test_no_code_exec.py b/src/servers/tsfm/tests/test_no_code_exec.py index e6490cc61..a37a480ba 100644 --- a/src/servers/tsfm/tests/test_no_code_exec.py +++ b/src/servers/tsfm/tests/test_no_code_exec.py @@ -160,3 +160,57 @@ def test_agent_checkpoint_folder_is_allowed_and_repo_code_is_not(): raise AssertionError(f"{p} accepted") except ValueError: pass + + +# Allowlist covers what the recipe tools advertise (PR #633 review) --------------------------- +import importlib + +import pandas as pd + +from ..substrate.resolver import ALLOWED_CLASSES + + +def test_every_allowlisted_class_imports(): + for path in sorted(ALLOWED_CLASSES): + module, name = path.rsplit(".", 1) + assert isinstance(getattr(importlib.import_module(module), name), type), path + + +def _tab_panel(numeric_label=False): + rng = np.random.RandomState(0) + X = rng.randn(30, 16) + df = pd.DataFrame(X, columns=[f"f{i}" for i in range(16)]) + df["label"] = X.mean(axis=1) if numeric_label else [0, 1] * 15 + p = os.path.join(refs._ensure_workdir(), f"tab_{uuid.uuid4().hex[:6]}.csv") + df.to_csv(p, index=False) + return p + + +def test_forecasting_transforms_run(): + r = _forecast({"estimator": {"sktime_class": "sktime.forecasting.naive.NaiveForecaster"}, + "transforms": [ + {"sktime_class": "sktime.transformations.series.detrend.Detrender"}, + {"sktime_class": "sktime.transformations.series.adapt.TabularToSeriesAdaptor", + "params": {"transformer": { + "_target_": "sklearn.preprocessing.StandardScaler"}}}], + "fh": [1, 2, 3]}) + assert r.get("status") == "success", r + + +def test_tabular_regression_runs(): + r = _call("run_tabular_recipe", {"dataset_path": _tab_panel(numeric_label=True), + "label_column": "label", + "recipe": {"task": "tsfm_regression", + "estimator": {"sktime_class": "sklearn.ensemble.RandomForestRegressor", + "params": {"n_estimators": 10}}}}) + assert r.get("status") == "success", r + + +def test_tabular_sktime_transform_runs(): + r = _call("run_tabular_recipe", {"dataset_path": _tab_panel(), "label_column": "label", + "recipe": {"task": "tsfm_classification", + "estimator": {"sktime_class": "sklearn.ensemble.RandomForestClassifier", + "params": {"n_estimators": 10}}, + "transforms": [{"sktime_class": + "sktime.transformations.series.summarize.SummaryTransformer"}]}}) + assert r.get("status") == "success", r