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/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/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/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/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..3f3953078 100644 --- a/src/servers/tsfm/substrate/resolver.py +++ b/src/servers/tsfm/substrate/resolver.py @@ -26,8 +26,90 @@ _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", + # 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", +}) + + 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 bd96bb405..5e66fe503 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() @@ -81,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", @@ -99,103 +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) - - -@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" 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..a37a480ba --- /dev/null +++ b/src/servers/tsfm/tests/test_no_code_exec.py @@ -0,0 +1,216 @@ +"""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 + + +# 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 + + +# 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