# SPDX-License-Identifier: AGPL-3.0-only # Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. """Tests for the sandboxed-Python AST policy in core/inference/tools.py.""" import os import subprocess import sys from pathlib import Path import pytest _BACKEND_ROOT = Path(__file__).resolve().parents[1] if str(_BACKEND_ROOT) not in sys.path: sys.path.insert(0, str(_BACKEND_ROOT)) from core.inference.tools import _check_code_safety def _ok(code: str): assert _check_code_safety(code) is None, code def _blocked(code: str, *, expect_phrase: str): msg = _check_code_safety(code) assert msg is not None, code assert expect_phrase in msg, (expect_phrase, msg) class TestMetadataHostDenylist: def test_aws_imds_literal_blocked(self): _blocked( 'import requests; requests.get("http://169.254.169.254/latest/meta-data/")', expect_phrase = "Blocked: cloud-metadata host", ) def test_gcp_metadata_dns_blocked(self): _blocked( 'import requests; requests.get("http://metadata.google.internal/")', expect_phrase = "Blocked: cloud-metadata host", ) def test_alibaba_ecs_literal_blocked(self): _blocked( 'import socket; s=socket.socket(); s.connect(("100.100.100.200", 80))', expect_phrase = "Blocked: cloud-metadata host", ) def test_ipv6_imds_literal_blocked(self): _blocked( 'import urllib.request; urllib.request.urlopen("http://[fd00:ec2::254]/")', expect_phrase = "Blocked: cloud-metadata host", ) def test_metadata_link_local_prefix_blocked(self): _blocked( 'import requests; requests.get("http://169.254.170.2/v3/")', expect_phrase = "Blocked: cloud-metadata host", ) class TestTrustedHostAllowlist: @pytest.mark.parametrize( "url", [ "https://en.wikipedia.org/wiki/Python_(programming_language)", "https://fr.wikipedia.org/wiki/Python_(langage)", "https://www.google.com/search?q=foo", "https://duckduckgo.com/?q=foo", "https://huggingface.co/unsloth", "https://cdn-lfs.huggingface.co/repos/abc/def/file.bin", "https://raw.githubusercontent.com/foo/bar/main/README.md", "https://api.github.com/repos/foo/bar", "https://arxiv.org/abs/2401.12345", "https://export.arxiv.org/abs/2401.12345", "https://stackoverflow.com/questions/12345", "https://math.stackexchange.com/questions/12345", "https://developer.mozilla.org/en-US/docs/Web/JavaScript", "https://docs.python.org/3/library/asyncio.html", "https://pypi.org/project/requests/", "https://files.pythonhosted.org/packages/foo/bar.whl", "https://www.bbc.com/news", "https://api.weather.gov/points/40,-90", "https://numpy.org/doc/stable/", "https://pytorch.org/docs/stable/index.html", ], ) def test_trusted_host_passes(self, url): _ok(f"import requests; requests.get({url!r})") def test_wikipedia_subdomain_passes(self): _ok('import urllib.request; urllib.request.urlopen("https://m.en.wikipedia.org/wiki/Foo")') def test_hf_co_short_form_passes(self): _ok('import requests; requests.get("https://hf.co/unsloth/Qwen3.5-4B-GGUF")') def test_github_io_pages_pass(self): _ok('import requests; requests.get("https://unslothai.github.io/")') class TestUntrustedHostBlock: def test_example_com_blocked(self): _blocked( 'import requests; requests.get("https://example.com/")', expect_phrase = "Blocked: host not in sandbox allowlist", ) def test_random_blog_blocked(self): _blocked( 'import urllib.request; urllib.request.urlopen("https://random-blog-host.example/")', expect_phrase = "Blocked: host not in sandbox allowlist", ) def test_socket_connect_random_host_blocked(self): _blocked( 'import socket; s=socket.socket(); s.connect(("evil.example", 80))', expect_phrase = "Blocked: host not in sandbox allowlist", ) def test_dynamic_url_not_statically_blocked(self): # Static AST can't resolve runtime URLs; bash blocklist is the fallback. _ok('import requests; url = "https://example.com/"; requests.get(url)') class TestLowLevelNetworkModules: @pytest.mark.parametrize( "code", [ 'import httpcore; httpcore.request("GET", "https://example.com")', 'import boto3; boto3.client("s3").list_buckets()', "from botocore.session import get_session; get_session()", "m = __import__('boto3'); print(m.__name__)", ("import importlib as il; m = il.import_module('http' + 'core'); print(m.__name__)"), ( "from importlib import import_module as load; " "name = 'botocore.session'; print(load(name).__name__)" ), ( "from builtins import __import__ as load; " "loader = load; print(loader('boto3').__name__)" ), ( "import importlib; " "load = getattr(importlib, 'import_' + 'module'); " "print(load('boto3').__name__)" ), ("import importlib; print(getattr(importlib, 'import_module')(name='boto3').__name__)"), ("import importlib; print(importlib.import_module(name='botocore.session').__name__)"), ("import importlib; print(vars(importlib)['import_module']('httpcore').__name__)"), ("import importlib; print(importlib.__dict__['import_module']('boto3').__name__)"), ("import builtins; print(getattr(builtins, '__import__')('botocore').__name__)"), ( "import importlib.machinery, importlib.util\n" "spec = importlib.util.find_spec('httpcore')\n" "loader = importlib.machinery.SourceFileLoader('httpcore', spec.origin)\n" "loader.load_module()" ), ], ) def test_low_level_client_blocked(self, code): _blocked(code, expect_phrase = "Blocked: low-level network module") @pytest.mark.parametrize( "code", [ "m = __import__('statistics'); print(m.mean([1, 2]))", ("from importlib import import_module as load; print(load('statistics').mean([1, 2]))"), ( "import importlib; " "print(getattr(importlib, 'import_module')(name='statistics').mean([1, 2]))" ), ( "import importlib; " "print(vars(importlib)['import_module']('statistics').mean([1, 2]))" ), ], ) def test_other_dynamic_imports_stay_available(self, code): _ok(code) class TestHostNormalization: def test_trailing_dot_treated_same(self): _ok('import requests; requests.get("https://wikipedia.org./")') def test_explicit_port_does_not_unblock_or_misblock(self): _ok('import requests; requests.get("https://en.wikipedia.org:443/wiki/Foo")') _blocked( 'import requests; requests.get("https://example.com:8080/")', expect_phrase = "Blocked: host not in sandbox allowlist", ) def test_userinfo_at_does_not_smuggle_metadata_host(self): _blocked( 'import requests; requests.get("https://wikipedia.org@169.254.169.254/latest/")', expect_phrase = "Blocked: cloud-metadata host", ) def test_uppercase_host_normalised(self): _ok('import requests; requests.get("https://EN.WIKIPEDIA.ORG/wiki/Foo")') class TestUploadDenylist: def test_requests_post_files_blocked(self): _blocked( ( "import requests\n" 'requests.post("https://huggingface.co/api/repos/upload", ' 'files={"f": open("x.bin", "rb")})' ), expect_phrase = "Blocked: file upload disallowed in sandbox", ) def test_requests_put_data_bytes_blocked(self): _blocked( ( "import requests\n" 'requests.put("https://huggingface.co/api/repos/upload", ' 'data=b"\\x00\\x01\\x02")' ), expect_phrase = "Blocked: file upload disallowed in sandbox", ) def test_requests_post_data_open_handle_blocked(self): _blocked( ( "import requests\n" 'requests.post("https://huggingface.co/api/repos/upload", ' 'data=open("x.bin", "rb"))' ), expect_phrase = "Blocked: file upload disallowed in sandbox", ) def test_httpx_post_files_blocked(self): _blocked( ( "import httpx\n" 'httpx.post("https://huggingface.co/api/repos/upload", ' 'files={"f": open("x.bin", "rb")})' ), expect_phrase = "Blocked: file upload disallowed in sandbox", ) def test_hf_api_upload_sandbox_local_allowed(self): # Sandbox-local relative path is the canonical safe shape. _ok( "from huggingface_hub import HfApi\n" 'HfApi().upload_file(path_or_fileobj="x.bin", ' 'path_in_repo="x.bin", repo_id="foo/bar")' ) def test_hf_module_upload_folder_sandbox_local_allowed(self): _ok( "import huggingface_hub\n" 'huggingface_hub.upload_folder(folder_path="outputs", repo_id="foo/bar")' ) def test_hf_create_commit_empty_operations_allowed(self): _ok( "import huggingface_hub\n" "api = huggingface_hub.HfApi()\n" 'api.create_commit(repo_id="foo/bar", operations=[])' ) def test_hf_upload_absolute_path_blocked(self): _blocked( "from huggingface_hub import HfApi\n" 'HfApi().upload_file(path_or_fileobj="/etc/passwd", path_in_repo="x", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_hf_upload_parent_dir_escape_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj="../escape.bin", path_in_repo="x", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_plain_post_json_not_blocked(self): _ok('import requests\nrequests.post("https://api.weather.gov/lookup", json={"k": "v"})') class TestSandboxEnvIsolation: """Sandbox env is built from a whitelist, so credential-shaped parent vars stay absent regardless of operator config (Linux/macOS/WSL/Windows).""" _SECRET_KEYS = ( # HF + ML tooling "HF_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACEHUB_API_TOKEN", "WANDB_API_KEY", "WANDB_USERNAME", "MLFLOW_TRACKING_TOKEN", "COMET_API_KEY", "NEPTUNE_API_TOKEN", # Generic cloud "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN", "GCP_SERVICE_ACCOUNT_KEY", "GOOGLE_APPLICATION_CREDENTIALS", "AZURE_STORAGE_KEY", "AZURE_CLIENT_SECRET", # Forge / git / package "GH_TOKEN", "GITHUB_TOKEN", "GITLAB_TOKEN", "BITBUCKET_TOKEN", "NPM_TOKEN", "PYPI_TOKEN", "CARGO_REGISTRY_TOKEN", # LLM provider "OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GOOGLE_API_KEY", "MISTRAL_API_KEY", "COHERE_API_KEY", "TOGETHER_API_KEY", # Loader injection / sudo state "LD_PRELOAD", "LD_LIBRARY_PATH", "DYLD_INSERT_LIBRARIES", "DYLD_LIBRARY_PATH", # Windows "USERPROFILE", "APPDATA", "LOCALAPPDATA", "ProgramData", ) def test_no_secret_keys_leak_into_sandbox(self, monkeypatch, tmp_path): from core.inference.tools import _build_safe_env for key in self._SECRET_KEYS: monkeypatch.setenv(key, f"sentinel-{key}") env = _build_safe_env(str(tmp_path)) for key in self._SECRET_KEYS: assert key not in env, f"parent env var {key!r} leaked into sandbox env" def test_sandbox_env_is_minimal_whitelist(self, monkeypatch, tmp_path): from core.inference.tools import _build_safe_env # Pollute parent env with arbitrary keys for key in ("EVIL", "RANDOM", "ATTACK_VEC", "MY_TOKEN", "X_API_KEY"): monkeypatch.setenv(key, "leak-me") env = _build_safe_env(str(tmp_path)) allowed = { "PATH", "HOME", "TMPDIR", "LANG", "TERM", "PYTHONIOENCODING", "PYTHONPATH", "UNSLOTH_STUDIO_SANDBOXED", "VIRTUAL_ENV", "SystemRoot", } extras = set(env.keys()) - allowed assert not extras, f"sandbox env added unexpected keys: {extras}" # PYTHONPATH is whitelist-built, never inherited: only the sandbox # sitecustomize shim dir (code-interpreter path remap). assert env["PYTHONPATH"].endswith("sandbox_site") assert "leak-me" not in env["PYTHONPATH"] assert env["UNSLOTH_STUDIO_SANDBOXED"] == "1" def test_runtime_import_guard_does_not_apply_to_bypass(self, monkeypatch, tmp_path): from core.inference.tools import _build_bypass_env, _build_safe_env monkeypatch.setenv("UNSLOTH_STUDIO_SANDBOXED", "1") (tmp_path / "boto3.py").write_text("VALUE = 7\n", encoding = "utf-8") code = ( "import sys\n" "sys.meta_path[:] = [f for f in sys.meta_path " "if not getattr(f, '_unsloth_blocked_network_guard', False)]\n" "name = ''.join(['bo', 'to3'])\n" "print(__import__(name).VALUE)" ) sandboxed = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert sandboxed.returncode != 0 assert "Blocked: low-level network module 'boto3'" in sandboxed.stderr bypass = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_bypass_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert bypass.returncode == 0, bypass.stderr assert bypass.stdout.strip() == "7" def test_runtime_import_guard_survives_global_tampering(self, monkeypatch, tmp_path): # Sandbox code can restore builtins.__import__, detach the meta-path # finder and rebind this module's globals, but the audit hook (which # cannot be removed) freezes its decision in a closure and still blocks. from core.inference.tools import _build_safe_env monkeypatch.setenv("UNSLOTH_STUDIO_SANDBOXED", "1") code = ( "import sys, builtins, sitecustomize\n" "sitecustomize._blocked_network_module = lambda _: None\n" "sitecustomize._BLOCKED_NETWORK_MODULES = frozenset()\n" "builtins.__import__ = sitecustomize._original_import\n" "sys.meta_path[:] = [f for f in sys.meta_path " "if not getattr(f, '_unsloth_blocked_network_guard', False)]\n" "name = ''.join(['bo', 'to3'])\n" "print(__import__(name).__name__)\n" ) result = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert result.returncode != 0 assert "Blocked: low-level network module 'boto3'" in result.stderr def test_runtime_import_guard_survives_env_flag_reset_for_children(self, tmp_path): # Clearing UNSLOTH_STUDIO_SANDBOXED before spawning a child must not # unguard the child: the child re-imports this shim from the sandbox site # dir still on PYTHONPATH, which is itself the sandbox signal. from core.inference.tools import _build_safe_env code = ( "import os, subprocess, sys\n" "os.environ['UNSLOTH_STUDIO_SANDBOXED'] = '0'\n" "r = subprocess.run([sys.executable, '-c', 'import boto3'], " "capture_output=True, text=True)\n" "sys.stdout.write('RC=%d\\n' % r.returncode)\n" "sys.stdout.write('BLOCKED=%d\\n' % " "(\"low-level network module 'boto3'\" in r.stderr))\n" ) result = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert result.returncode == 0, result.stderr assert "RC=1" in result.stdout assert "BLOCKED=1" in result.stdout def test_runtime_import_guard_survives_env_flag_deletion_for_children(self, tmp_path): # Deleting UNSLOTH_STUDIO_SANDBOXED (not just setting it to "0") before # spawning a child must not unguard it: the child still re-imports this # shim from the sandbox site dir on PYTHONPATH, which is the real signal. from core.inference.tools import _build_safe_env code = ( "import os, subprocess, sys\n" "os.environ.pop('UNSLOTH_STUDIO_SANDBOXED', None)\n" "r = subprocess.run([sys.executable, '-c', 'import boto3'], " "capture_output=True, text=True)\n" "sys.stdout.write('RC=%d\\n' % r.returncode)\n" "sys.stdout.write('BLOCKED=%d\\n' % " "(\"low-level network module 'boto3'\" in r.stderr))\n" ) result = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert result.returncode == 0, result.stderr assert "RC=1" in result.stdout assert "BLOCKED=1" in result.stdout @pytest.mark.parametrize( "code", [ "name = ''.join(['http', 'core']); print(__import__(name).__name__)", ( "import importlib; name = ''.join(['http', 'core']); " "print(importlib.import_module(name).__name__)" ), ("import httpx; name = ''.join(['http', 'core']); print(__import__(name).__name__)"), ( "import httpx, importlib; suffix = ''.join(['_', 'api']); " "print(importlib.import_module('.' + suffix, package='httpcore')" ".__name__.split('.')[0])" ), ], ) def test_runtime_import_guard_blocks_direct_dynamic_httpcore(self, tmp_path, code): from core.inference.tools import _build_bypass_env, _build_safe_env sandboxed = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert sandboxed.returncode != 0 assert "Blocked: low-level network module 'httpcore'" in sandboxed.stderr bypass = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_bypass_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert bypass.returncode == 0, bypass.stderr assert bypass.stdout.strip() == "httpcore" def test_runtime_import_guard_rejects_spoofed_httpx_globals(self, tmp_path): from core.inference.tools import _build_safe_env code = ( "import httpx\n" "__name__ = 'httpx._client'\n" "__file__ = httpx.__file__\n" "name = ''.join(['http', 'core'])\n" "module = __import__(name)\n" "print(module.__name__)\n" ) result = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert result.returncode != 0 assert "Blocked: low-level network module 'httpcore'" in result.stderr def test_runtime_import_guard_blocks_legacy_loader_httpcore(self, tmp_path): from core.inference.tools import _build_safe_env code = ( "import importlib.machinery, importlib.util\n" "name = ''.join(['http', 'core'])\n" "spec = importlib.util.find_spec(name)\n" "loader = importlib.machinery.SourceFileLoader(name, spec.origin)\n" "loader.load_module()\n" ) result = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert result.returncode != 0 assert "Blocked: low-level network module 'httpcore'" in result.stderr def test_runtime_import_guard_blocks_aliased_httpcore_origin(self, tmp_path): from core.inference.tools import _build_safe_env code = ( "import importlib.machinery, importlib.util, sys\n" "spec = importlib.machinery.PathFinder.find_spec('httpcore')\n" "alias = importlib.util.spec_from_file_location(\n" " 'hc', spec.origin,\n" " submodule_search_locations=list(spec.submodule_search_locations or []),\n" ")\n" "module = importlib.util.module_from_spec(alias)\n" "sys.modules['hc'] = module\n" "alias.loader.exec_module(module)\n" "module.request('GET', 'http://127.0.0.1:9')\n" ) result = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert result.returncode != 0 assert "Blocked: low-level network module 'httpcore'" in result.stderr def test_runtime_import_guard_blocks_httpcore_backend_reflection(self, tmp_path): from core.inference.tools import _build_safe_env code = ( "import sys, types\n" "import httpx\n" "client = httpx.Client()\n" "client.close()\n" "module = sys.modules['httpcore._backends.sync']\n" "module_dict = types.ModuleType.__getattribute__(module, '__dict__')\n" "backend_type = module_dict['SyncBackend']\n" "guarded = type.__getattribute__(backend_type, '__dict__')['connect_tcp']\n" "dispatch = key = None\n" "for cell in guarded.__closure__ or ():\n" " value = cell.cell_contents\n" " if callable(value) and getattr(value, '__name__', '') == 'dispatch':\n" " dispatch = value\n" " elif type(value) is object:\n" " key = value\n" "originals = None\n" "for cell in dispatch.__closure__ or ():\n" " value = cell.cell_contents\n" " if type(value) is dict:\n" " originals = value\n" "original = originals[key]\n" "original(\n" " backend_type(), '127.0.0.1', 9,\n" " timeout=0.01, local_address=None, socket_options=None,\n" ")\n" ) result = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert result.returncode != 0 assert "Blocked: low-level network module 'httpcore'" in result.stderr def test_runtime_import_guard_allows_httpx_backend_connect(self, tmp_path): from core.inference.tools import _build_safe_env code = ( "import httpx\n" "try:\n" " httpx.get('http://127.0.0.1:9/probe', timeout=0.01)\n" "except Exception as exc:\n" " print(type(exc).__name__)\n" ) result = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert result.returncode == 0, result.stderr assert "Blocked: low-level network module" not in result.stderr def test_runtime_import_guard_blocks_local_module_httpcore_import(self, tmp_path): from core.inference.tools import _build_safe_env (tmp_path / "loader.py").write_text( "name = ''.join(['http', 'core'])\nprint(__import__(name).__name__)\n", encoding = "utf-8", ) result = subprocess.run( [sys.executable, "-c", "import loader"], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert result.returncode != 0 assert "Blocked: low-level network module 'httpcore'" in result.stderr @pytest.mark.parametrize("module_name", ["loader", "httpx"]) def test_runtime_import_guard_blocks_external_module_httpcore_import( self, tmp_path, module_name ): from core.inference.tools import _build_safe_env workdir = tmp_path / "sandbox" external = tmp_path / "external" workdir.mkdir() external.mkdir() (external / f"{module_name}.py").write_text( "name = ''.join(['http', 'core'])\nprint(__import__(name).__name__)\n", encoding = "utf-8", ) code = f"import sys; sys.path.insert(0, {str(external)!r}); import {module_name}" result = subprocess.run( [sys.executable, "-c", code], cwd = workdir, env = _build_safe_env(str(workdir)), capture_output = True, text = True, check = False, ) assert result.returncode != 0 assert "Blocked: low-level network module 'httpcore'" in result.stderr @pytest.mark.parametrize( "code", [ ( "import httpx, sys; client = httpx.Client(); client.close(); " "print(sys.modules['httpcore'].request)" ), ( "import httpx, sys; client = httpx.Client(); client.close(); " "print(sys.modules['httpcore._sync.connection_pool'].ConnectionPool)" ), ( "import httpx, sys, types; client = httpx.Client(); client.close(); " "module = sys.modules['httpcore']; " "request = types.ModuleType.__getattribute__(module, 'request'); " "request('GET', 'http://127.0.0.1:9/probe')" ), ( "import asyncio, httpx, sys, types\n" "async def main():\n" " async with httpx.AsyncClient():\n" " pass\n" " module = sys.modules['httpcore']\n" " pool_type = types.ModuleType.__getattribute__(module, 'AsyncConnectionPool')\n" " async with pool_type() as pool:\n" " await pool.request('GET', 'http://127.0.0.1:9/probe')\n" "asyncio.run(main())" ), ], ) def test_runtime_import_guard_blocks_cached_httpcore_access(self, tmp_path, code): from core.inference.tools import _build_safe_env result = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert result.returncode != 0 assert "Blocked: low-level network module 'httpcore'" in result.stderr @pytest.mark.parametrize("module", ["httpx", "requests", "huggingface_hub"]) def test_runtime_import_guard_keeps_supported_clients_available(self, tmp_path, module): from core.inference.tools import _build_safe_env result = subprocess.run( [sys.executable, "-c", f"import {module}; print({module}.__name__)"], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert result.returncode == 0, result.stderr assert result.stdout.strip() == module def test_runtime_import_guard_keeps_httpx_transport_available(self, tmp_path): from core.inference.tools import _build_safe_env code = "import httpx; client = httpx.Client(); print(type(client).__name__); client.close()" result = subprocess.run( [sys.executable, "-c", code], cwd = tmp_path, env = _build_safe_env(str(tmp_path)), capture_output = True, text = True, check = False, ) assert result.returncode == 0, result.stderr assert result.stdout.strip() == "Client" def test_home_points_at_sandbox_workdir(self, tmp_path): from core.inference.tools import _build_safe_env env = _build_safe_env(str(tmp_path)) assert env["HOME"] == str(tmp_path) assert env["TMPDIR"] == str(tmp_path) def test_term_is_dumb(self, tmp_path): from core.inference.tools import _build_safe_env # Avoid re-using the operator's TERM (e.g. xterm-256color) that # could trigger color-escape parsing in downstream tools. env = _build_safe_env(str(tmp_path)) assert env["TERM"] == "dumb" def test_bypass_env_installs_sitecustomize_path_shim(self, tmp_path): # Bypass mode keeps path remapping without installing network guards. from core.inference.tools import _BYPASS_SITE_DIR, _build_bypass_env env = _build_bypass_env(str(tmp_path)) assert _BYPASS_SITE_DIR in env["PYTHONPATH"].split(os.pathsep) def test_bypass_env_prepends_shim_and_keeps_inherited_pythonpath(self, monkeypatch, tmp_path): from core.inference.tools import _BYPASS_SITE_DIR, _build_bypass_env monkeypatch.setenv("PYTHONPATH", "/operator/libs") env = _build_bypass_env(str(tmp_path)) parts = env["PYTHONPATH"].split(os.pathsep) # Shim first so its open()/makedirs remap wins, operator entries kept. assert parts[0] == _BYPASS_SITE_DIR assert "/operator/libs" in parts class TestSandboxCpuRlimitDefault: """Pin the default so a regression below 600s without opt-in is caught.""" def test_default_cpu_s_is_600(self): src = (_BACKEND_ROOT / "core" / "inference" / "tools.py").read_text() assert 'UNSLOTH_STUDIO_SANDBOX_CPU_S", "600"' in src def test_clone_newnet_removed(self): src = (_BACKEND_ROOT / "core" / "inference" / "tools.py").read_text() assert "_libc.unshare(0x40000000)" not in src # Explanatory comment retained. assert "CLONE_NEWNET" in src def test_nofile_env_tunable(self): src = (_BACKEND_ROOT / "core" / "inference" / "tools.py").read_text() # Parity with the other rlimits: must come from the env, not be hardcoded. assert "UNSLOTH_STUDIO_SANDBOX_NOFILE" in src class TestMaxBodyDefault: def test_default_is_500_mb(self): src = (_BACKEND_ROOT / "utils" / "upload_limits.py").read_text() assert "DEFAULT_UPLOAD_LIMIT_MB = 500" in src assert "UNSLOTH_STUDIO_MAX_BODY_MB" in src class TestBashBlocklistPosition: """The blocklist must fire at command position only, so args like `grep -r curl .` and `echo source` are not falsely rejected.""" @staticmethod def _find(): from core.inference.tools import _find_blocked_commands return _find_blocked_commands # ---- argument-position: must NOT be blocked ---- def test_grep_for_curl_string_allowed(self): assert self._find()("grep -r curl .") == set() def test_echo_source_allowed(self): assert self._find()("echo source the data") == set() def test_cat_with_word_source_allowed(self): # 'source' is an argument to echo, and echo isn't blocked either. assert self._find()("cat README.md && echo source") == set() assert "source" not in self._find()("cat README.md && echo source") assert "echo" not in self._find()("cat README.md && echo source") def test_ls_path_containing_curl_allowed(self): assert self._find()("ls /usr/bin/curl") == set() def test_find_for_wget_string_allowed(self): assert self._find()("find . -name wget") == set() def test_quoted_curl_arg_allowed(self): assert self._find()('echo "curl is a tool"') == set() # ---- command-position: must be blocked ---- def test_bare_rm_blocked(self): assert "rm" in self._find()("rm -rf /") def test_curl_at_command_position_blocked(self): assert "curl" in self._find()("curl https://example.com") def test_after_semicolon_blocked(self): # `rm` after `;` even without surrounding whitespace. assert "rm" in self._find()("echo done; rm -rf /tmp/x") assert "rm" in self._find()("echo done;rm -rf /tmp/x") def test_after_double_ampersand_blocked(self): assert "wget" in self._find()("cd /tmp && wget https://bad") def test_split_quotes_obfuscation_blocked(self): # shlex collapses 'r''m' -> 'rm' at command position. assert "rm" in self._find()("r''m -rf /") def test_path_prefixed_command_blocked(self): assert "sudo" in self._find()("/usr/bin/sudo whoami") def test_nested_bash_c_blocked(self): # Recursion into the nested command string catches command-position curl. assert "curl" in self._find()("bash -c 'curl https://x'") def test_subshell_command_blocked(self): assert "rm" in self._find()("echo $(rm -rf /tmp)") def test_backtick_command_blocked(self): assert "rm" in self._find()("echo `rm -rf /tmp`") # ---- shell prefixes / wrappers: must still be blocked ---- @pytest.mark.parametrize( "command, blocked_cmd", [ ("FOO=bar curl https://example.com", "curl"), ("HTTPS_PROXY=http://x wget https://bad", "wget"), ("env curl https://example.com", "curl"), ("env FOO=1 /usr/bin/curl https://x", "curl"), ("/usr/bin/env rm -rf /tmp/x", "rm"), ("command rm -rf /tmp/x", "rm"), ("time curl https://example.com", "curl"), ("nice rm -rf /tmp/x", "rm"), ("nohup wget https://bad", "wget"), ("timeout 1 rm -rf /tmp/x", "rm"), ("setsid rm -rf /tmp/x", "rm"), ("stdbuf -oL rm -rf /tmp/x", "rm"), ("sudo rm -rf /tmp/x", "rm"), ("cd /tmp; FOO=bar rm -rf x", "rm"), ], ) def test_command_prefix_wrappers_blocked(self, command, blocked_cmd): assert blocked_cmd in self._find()(command) # ---- split-quoted command name after attached separators ---- def test_split_quotes_after_semicolon_blocked(self): assert "rm" in self._find()("echo done; r''m -rf /tmp/x") assert "rm" in self._find()("echo done;r''m -rf /tmp/x") assert "curl" in self._find()("echo done; c''url --version") assert "curl" in self._find()("echo done; /usr/bin/c''url --version") # ---- find -exec / xargs invoke a command directly ---- def test_find_exec_blocked(self): assert "rm" in self._find()("find . -type f -exec rm -f {} +") assert "rm" in self._find()("find . -type f -exec rm -f {} ';'") assert "rm" in self._find()("find . -execdir rm -f {} ';'") def test_xargs_command_blocked(self): assert "rm" in self._find()("printf /tmp/x | xargs rm") assert "rm" in self._find()("printf /tmp/x | xargs -- rm") # ---- brace groups and bash compound statements ---- def test_brace_group_blocked(self): assert "rm" in self._find()("{ rm -rf /tmp/x; }") def test_if_then_blocked(self): assert "curl" in self._find()("if true; then curl --version; fi") def test_while_do_blocked(self): assert "curl" in self._find()("while true; do curl --version; break; done") class TestHfUploadImportGate: """Upload-method blocking requires an HF import in scope, so paramiko / boto3 / internal SDKs with the same method names don't false-positive.""" def test_paramiko_upload_file_allowed_without_hf_import(self): _ok("import paramiko; sftp=None; sftp.upload_file('a','b')") def test_boto3_create_commit_allowed_without_hf_import(self): _ok("client=None; client.create_commit(Repo='x')") def test_hf_api_upload_safe_path_allowed(self): # Sandbox-local relative path -- the permitted call shape. _ok("from huggingface_hub import HfApi; HfApi().upload_file('a','b','c')") def test_hf_upload_file_fq_safe_path_allowed(self): _ok("import huggingface_hub; huggingface_hub.upload_file('a','b','c')") def test_dynamic_builtin_import_safe_path_allowed(self): # `__import__('huggingface_hub')` puts HF in scope; relative literal is safe. _ok("hf=__import__('huggingface_hub'); hf.HfApi().upload_file('a','b','c')") def test_dynamic_importlib_safe_path_allowed(self): _ok( "import importlib; hf=importlib.import_module('huggingface_hub');" " hf.HfApi().upload_file('a','b','c')" ) def test_from_importlib_import_module_safe_create_commit_allowed(self): _ok( "from importlib import import_module;" " api=import_module('huggingface_hub').HfApi(); api.create_commit()" ) def test_hf_bare_name_upload_safe_path_allowed(self): # Bare `upload_file(...)` (imported from huggingface_hub) with a # sandbox-local relative-path literal is allowed. _ok( "from huggingface_hub import upload_file;" " upload_file(path_or_fileobj='x', path_in_repo='x', repo_id='r')" ) def test_hf_bare_name_upload_folder_safe_allowed(self): _ok( "from huggingface_hub import upload_folder; upload_folder(folder_path='x', repo_id='r')" ) def test_hf_bare_name_create_commit_safe_allowed(self): _ok("from huggingface_hub import create_commit; create_commit(operations=[], repo_id='r')") def test_bare_name_upload_file_without_hf_import_allowed(self): # No HF import -- local helper named upload_file passes. _ok("def upload_file(*a, **k):\n pass\nupload_file('x', 'y', 'z')") class TestHfUploadSandboxLocalPaths: """HF upload gate allows only files in the sandbox workdir. Absolute paths, `..` traversal, home expansion, and Windows drives are rejected (they could lift secrets from outside the sandbox).""" def test_relative_literal_allowed(self): _ok( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj="model.bin",' ' path_in_repo="model.bin", repo_id="me/r")' ) def test_dotted_relative_allowed(self): _ok( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj="./outputs/m.bin",' ' path_in_repo="m.bin", repo_id="me/r")' ) def test_nested_relative_allowed(self): _ok( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj="outputs/run42/model.bin",' ' path_in_repo="m.bin", repo_id="me/r")' ) def test_open_of_relative_literal_allowed(self): _ok( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj=open("model.bin", "rb"),' ' path_in_repo="m.bin", repo_id="me/r")' ) def test_inline_bytes_literal_allowed(self): _ok( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj=b"\\x00\\x01\\x02",' ' path_in_repo="m.bin", repo_id="me/r")' ) def test_absolute_unix_path_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj="/etc/passwd",' ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_absolute_windows_drive_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj="C:\\\\Windows\\\\creds",' ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_home_expansion_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj="~/.aws/credentials",' ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_parent_traversal_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj="../../etc/shadow",' ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_parent_traversal_mid_path_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj="outputs/../../../etc",' ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_open_of_absolute_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj=open("/etc/passwd","rb"),' ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_open_of_parent_traversal_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj=open("../escape","rb"),' ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_dynamic_variable_path_blocked(self): # A non-literal expr could resolve to any path at runtime; the # static checker can't prove safety, so block. _blocked( "import huggingface_hub, os\n" "p = os.path.join('outputs', 'x.bin')\n" 'huggingface_hub.upload_file(path_or_fileobj=p, path_in_repo="x", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_upload_folder_absolute_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_folder(folder_path="/var/log", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_upload_folder_parent_traversal_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_folder(folder_path="../..", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_upload_large_folder_absolute_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_large_folder(folder_path="/etc", repo_id="r")', expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) def test_create_commit_operation_safe_allowed(self): _ok( "import huggingface_hub\n" "from huggingface_hub import CommitOperationAdd\n" "huggingface_hub.HfApi().create_commit(\n" " repo_id='r',\n" " operations=[CommitOperationAdd(path_or_fileobj='m.bin', path_in_repo='m.bin')],\n" ")" ) def test_create_commit_operation_absolute_blocked(self): _blocked( "import huggingface_hub\n" "from huggingface_hub import CommitOperationAdd\n" "huggingface_hub.HfApi().create_commit(\n" " repo_id='r',\n" " operations=[CommitOperationAdd(path_or_fileobj='/etc/passwd', path_in_repo='x')],\n" ")", expect_phrase = "HF upload path must be a sandbox-local relative-path literal", ) class TestHfUploadEnvAndSecretLeakBlock: """HF upload gate rejects any arg sourced from os.environ / os.getenv / subprocess env reads, since a script can reach the parent env directly despite the safe-env shell wrapper.""" def test_path_from_os_environ_subscript_blocked(self): _blocked( "import huggingface_hub, os\n" 'huggingface_hub.upload_file(path_or_fileobj=os.environ["HF_TOKEN"],' ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload cannot include os.environ", ) def test_path_from_os_environ_get_blocked(self): _blocked( "import huggingface_hub, os\n" 'huggingface_hub.upload_file(path_or_fileobj=os.environ.get("HF_TOKEN"),' ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload cannot include os.environ", ) def test_path_from_os_getenv_blocked(self): _blocked( "import huggingface_hub, os\n" 'huggingface_hub.upload_file(path_or_fileobj=os.getenv("HF_TOKEN"),' ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload cannot include os.environ", ) def test_path_from_bare_getenv_blocked(self): _blocked( "import huggingface_hub\n" "from os import getenv\n" 'huggingface_hub.upload_file(path_or_fileobj=getenv("HF_TOKEN"),' ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload cannot include os.environ", ) def test_path_from_subprocess_printenv_blocked(self): _blocked( "import huggingface_hub, subprocess\n" "huggingface_hub.upload_file(" 'path_or_fileobj=subprocess.check_output(["printenv","HF_TOKEN"]),' ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload cannot include os.environ", ) def test_token_kwarg_with_literal_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj="x.bin",' ' path_in_repo="x", repo_id="r", token="hf_xyzabc123")', expect_phrase = "HF upload token= cannot be set", ) def test_hf_token_kwarg_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_file(path_or_fileobj="x.bin",' ' path_in_repo="x", repo_id="r", hf_token="hf_secret")', expect_phrase = "HF upload hf_token= cannot be set", ) def test_api_key_kwarg_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.upload_folder(folder_path="outputs",' ' repo_id="r", api_key="abc")', expect_phrase = "HF upload api_key= cannot be set", ) def test_token_kwarg_from_env_blocked(self): # Both rules fire; the sensitive-kwarg check trips first. _blocked( "import huggingface_hub, os\n" 'huggingface_hub.upload_file(path_or_fileobj="x.bin",' ' path_in_repo="x", repo_id="r", token=os.environ["HF_TOKEN"])', expect_phrase = "HF upload token= cannot be set", ) def test_env_dict_unpacked_via_environ_attr_blocked(self): # Bare `os.environ` reference (passed somewhere it gets serialized). _blocked( "import huggingface_hub, os\n" "huggingface_hub.upload_file(path_or_fileobj=str(os.environ)," ' path_in_repo="x", repo_id="r")', expect_phrase = "HF upload cannot include os.environ", ) def test_repo_id_from_env_also_blocked(self): # Non-path args must not source env vars either -- an attacker # could encode secrets in repo_id or path_in_repo. _blocked( "import huggingface_hub, os\n" 'huggingface_hub.upload_file(path_or_fileobj="x.bin",' ' path_in_repo=os.environ["HF_TOKEN"], repo_id="r")', expect_phrase = "HF upload cannot include os.environ", ) def test_create_commit_with_env_in_operation_blocked(self): _blocked( "import huggingface_hub, os\n" "from huggingface_hub import CommitOperationAdd\n" "huggingface_hub.HfApi().create_commit(\n" " repo_id='r',\n" " operations=[CommitOperationAdd(" 'path_or_fileobj=os.environ["HF_TOKEN"], path_in_repo="x")],\n' ")", expect_phrase = "HF upload cannot include os.environ", ) def test_create_commit_token_kwarg_blocked(self): _blocked( "import huggingface_hub\n" 'huggingface_hub.HfApi().create_commit(repo_id="r",' ' operations=[], token="hf_xxx")', expect_phrase = "HF upload token= cannot be set", )