Studio: close remaining sandbox review gaps
- Freeze the runtime import guard's block decision in the non-removable audit hook so sandbox code cannot neutralise it by rebinding sitecustomize globals, restoring builtins.__import__, or detaching the meta-path finder. - Keep child interpreters guarded when sandbox code lowers UNSLOTH_STUDIO_SANDBOXED before spawning them (present-but-not-"1" flag + sandbox site dir on PYTHONPATH); bypass (flag removed) is unaffected. - Recurse into shells launched via find/fd -exec when scanning for Python startup-guard bypasses. - Treat optional-chained computed document.write and document.open().write() receivers as HTML sinks. - Honour a declared charset (or fail closed on an unknown one) when decoding active data: documents. - Add regression tests for each.
This commit is contained in:
Michael Han
parent
67d01e9cfb
commit
a37f225cd6
5 changed files with 176 additions and 15 deletions
|
|
@ -279,14 +279,6 @@ def _guarded_import_module(name, package = None):
|
|||
return module
|
||||
|
||||
|
||||
def _network_import_audit(event, args):
|
||||
if event != "import" or not args:
|
||||
return
|
||||
root = _blocked_network_module(args[0])
|
||||
if root is not None:
|
||||
_raise_blocked_network_module(root)
|
||||
|
||||
|
||||
class _BlockedNetworkModuleFinder:
|
||||
_unsloth_blocked_network_guard = True
|
||||
|
||||
|
|
@ -302,12 +294,73 @@ class _BlockedNetworkModuleFinder:
|
|||
return None
|
||||
|
||||
|
||||
def _loaded_from_sandbox_site():
|
||||
"""True when this shim is imported from the sandbox site dir on PYTHONPATH.
|
||||
|
||||
The parent adds this directory to a sandbox child's PYTHONPATH, so its
|
||||
presence confirms the child is still running under the sandbox launcher even
|
||||
if ``UNSLOTH_STUDIO_SANDBOXED`` has been altered in ``os.environ``.
|
||||
"""
|
||||
try:
|
||||
module_dir = os.path.realpath(os.path.dirname(__file__))
|
||||
except (OSError, NameError, TypeError):
|
||||
return False
|
||||
for entry in os.environ.get("PYTHONPATH", "").split(os.pathsep):
|
||||
if not entry:
|
||||
continue
|
||||
try:
|
||||
if os.path.realpath(entry) == module_dir:
|
||||
return True
|
||||
except OSError:
|
||||
continue
|
||||
return False
|
||||
|
||||
|
||||
def _sandbox_guard_should_activate():
|
||||
"""Decide whether to install the runtime network guard.
|
||||
|
||||
Normal sandbox children set ``UNSLOTH_STUDIO_SANDBOXED=1``. Bypass (full
|
||||
access) removes the variable entirely, so an absent flag means "do not
|
||||
guard". A flag that is PRESENT but not ``"1"`` (e.g. sandbox code running
|
||||
``os.environ['UNSLOTH_STUDIO_SANDBOXED']='0'`` before spawning a child to
|
||||
escape the guard) is tampering: keep the guard on as long as this shim was
|
||||
still loaded from the sandbox site dir the launcher put on PYTHONPATH.
|
||||
"""
|
||||
flag = os.environ.get("UNSLOTH_STUDIO_SANDBOXED")
|
||||
if flag == "1":
|
||||
return True
|
||||
if flag is None:
|
||||
return False
|
||||
return _loaded_from_sandbox_site()
|
||||
|
||||
|
||||
def _install_import_guard():
|
||||
global _import_guard_installed
|
||||
if os.environ.get("UNSLOTH_STUDIO_SANDBOXED") != "1":
|
||||
if not _sandbox_guard_should_activate():
|
||||
return
|
||||
if not _import_guard_installed:
|
||||
sys.addaudithook(_network_import_audit)
|
||||
# Capture the block sets and trust probe as closure locals. Audit hooks
|
||||
# cannot be removed once registered, so this hook is the backstop for the
|
||||
# import wrapper and meta-path finder (both of which sandbox code can
|
||||
# restore/detach). Reading globals here would let sandbox code neutralise
|
||||
# it by rebinding this module's attributes, so the decision is frozen.
|
||||
blocked_roots = frozenset(_BLOCKED_NETWORK_MODULES)
|
||||
direct_roots = frozenset(_DIRECT_BLOCKED_NETWORK_MODULES)
|
||||
sandbox_requested = _sandbox_code_requested_import
|
||||
|
||||
def _immutable_network_import_audit(event, args):
|
||||
if event != "import" or not args:
|
||||
return
|
||||
name = args[0]
|
||||
if not isinstance(name, str):
|
||||
return
|
||||
root = name.split(".", 1)[0]
|
||||
if root in blocked_roots or (root in direct_roots and sandbox_requested()):
|
||||
raise ModuleNotFoundError(
|
||||
f"Blocked: low-level network module {root!r} is unavailable in sandboxed code"
|
||||
)
|
||||
|
||||
sys.addaudithook(_immutable_network_import_audit)
|
||||
builtins.__import__ = _guarded_import
|
||||
importlib.import_module = _guarded_import_module
|
||||
_import_guard_installed = True
|
||||
|
|
|
|||
|
|
@ -325,9 +325,18 @@ def _sandbox_python_startup_bypasses_guard(command: str, depth: int = 0) -> bool
|
|||
for segment in _shell_command_segments(command):
|
||||
first = os.path.basename(segment[0].replace("\\", "/")).lower()
|
||||
wrapper_context = first in _PYTHON_LAUNCH_WRAPPERS
|
||||
find_exec = first in {"find", "fd"}
|
||||
for shell_index, shell_token in enumerate(segment):
|
||||
shell = os.path.basename(shell_token.replace("\\", "/")).lower()
|
||||
if shell not in shell_names or (shell_index and not wrapper_context):
|
||||
# A shell after the first token is only a real launch when it follows
|
||||
# a launch wrapper (env/xargs/...) or a find/fd -exec flag; otherwise
|
||||
# it is an argument (e.g. a path) and is ignored.
|
||||
find_exec_context = find_exec and any(
|
||||
token in _FIND_EXEC_FLAGS for token in segment[:shell_index]
|
||||
)
|
||||
if shell not in shell_names or (
|
||||
shell_index and not wrapper_context and not find_exec_context
|
||||
):
|
||||
continue
|
||||
for index in range(shell_index + 1, len(segment) - 1):
|
||||
token = segment[index]
|
||||
|
|
@ -2596,7 +2605,11 @@ _RENDER_HTML_MARKUP_ASSIGNMENT_START_RE = re.compile(
|
|||
)
|
||||
_RENDER_HTML_MARKUP_CALL_START_RE = re.compile(
|
||||
r"(?:\.\s*(?P<insert>insertAdjacentHTML)|"
|
||||
r"\bdocument\s*(?:\?\.\s*|\.\s*)(?P<write>write|writeln))"
|
||||
# document.write / writeln, optionally reached through a document-valued
|
||||
# receiver such as document.open(): document.open().write('<img src=...>')
|
||||
# returns the same document and inserts the remote-loading markup.
|
||||
r"\bdocument\s*(?:(?:\?\.\s*|\.\s*)open\s*\([^()]*\)\s*)?"
|
||||
r"(?:\?\.\s*|\.\s*)(?P<write>write|writeln))"
|
||||
r"\s*(?:\?\.\s*)?\(",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
|
@ -2605,7 +2618,7 @@ _RENDER_HTML_COMPUTED_ASSIGNMENT_START_RE = re.compile(
|
|||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
_RENDER_HTML_COMPUTED_CALL_START_RE = re.compile(
|
||||
r"(?:(?P<document>\bdocument)\s*)?\[\s*(?P<member>[^\]]+)\s*\]\s*(?:\?\.\s*)?\(",
|
||||
r"(?:(?P<document>\bdocument)\s*(?:\?\.\s*)?)?\[\s*(?P<member>[^\]]+)\s*\]\s*(?:\?\.\s*)?\(",
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
_RENDER_HTML_REFLECT_SET_START_RE = re.compile(
|
||||
|
|
@ -2776,12 +2789,19 @@ def _render_html_data_document_reaches_network(value: str, depth: int) -> bool:
|
|||
media_type = (parts[0] or "text/plain").lower()
|
||||
if media_type not in _RENDER_HTML_ACTIVE_DATA_MIME_TYPES:
|
||||
return False
|
||||
# Honour a declared charset so a UTF-16/Latin-1 document decodes the same way
|
||||
# the browser would; an unknown/undecodable charset fails closed rather than
|
||||
# letting a mangled UTF-8 read hide a remote load.
|
||||
charset = "utf-8"
|
||||
for part in parts[1:]:
|
||||
if part.lower().startswith("charset="):
|
||||
charset = part.split("=", 1)[1].strip() or "utf-8"
|
||||
try:
|
||||
payload_bytes = urllib.parse.unquote_to_bytes(payload)
|
||||
if any(part.lower() == "base64" for part in parts[1:]):
|
||||
payload_bytes = base64.b64decode(b"".join(payload_bytes.split()), validate = True)
|
||||
markup = payload_bytes.decode("utf-8", errors = "replace")
|
||||
except (ValueError, binascii.Error):
|
||||
markup = payload_bytes.decode(charset, errors = "replace")
|
||||
except (ValueError, binascii.Error, LookupError):
|
||||
return True
|
||||
return _render_html_code_reaches_network(markup, depth + 1)
|
||||
|
||||
|
|
|
|||
|
|
@ -181,6 +181,10 @@ def test_bash_blocklist_enforced_when_sandboxed(captured_popen):
|
|||
'env -u PYTHONPATH sh -c "python -c import\\ boto3"',
|
||||
'command sh -c "python -I -c import\\ boto3"',
|
||||
'find . -exec python -S -c "import boto3" ;',
|
||||
# A find/fd -exec that hides the interpreter behind a nested shell must
|
||||
# still be recursed into, not left as an opaque exec target.
|
||||
'find . -exec sh -c "python -S -c import\\ boto3" ;',
|
||||
'find . -type f -execdir bash -c "python -I -c import\\ boto3" ;',
|
||||
],
|
||||
)
|
||||
def test_bash_blocks_python_startup_guard_bypasses(captured_popen, command):
|
||||
|
|
|
|||
|
|
@ -1113,6 +1113,16 @@ def test_render_html_gated_only_when_networked():
|
|||
assert rh("<script>document.write('<img sr','c=https://evil/x>')</script>") is True
|
||||
assert rh("<script>document.writeln('<p>Local</p>')</script>") is False
|
||||
assert rh("<script>writer.write('<img src=https://evil/x>')</script>") is False
|
||||
# Optional-chained computed document.write still recurses into the markup.
|
||||
assert rh("<script>document?.['write']('<img src=https://evil/x>')</script>") is True
|
||||
assert rh("<script>document?.['write']('<p>Local</p>')</script>") is False
|
||||
# document.open() returns the document, so a write through it is an HTML sink.
|
||||
assert rh("<script>document.open().write('<img src=https://evil/x>')</script>") is True
|
||||
assert (
|
||||
rh("<script>document.open('text/html').writeln('<img src=https://evil/x>')</script>")
|
||||
is True
|
||||
)
|
||||
assert rh("<script>document.open().write('<p>Local</p>')</script>") is False
|
||||
# A computed bracket key spliced from string fragments on a global host object.
|
||||
assert rh("<script>window['fet'+'ch']('https://attacker.example')</script>") is True
|
||||
assert rh("<script>self['open' + '']('https://x')</script>") is True
|
||||
|
|
@ -1142,6 +1152,25 @@ def test_render_html_gated_only_when_networked():
|
|||
assert rh('<img src="data:image/png;base64,iVBORw0KGgo=">') is False
|
||||
assert rh('<object data="data:image/svg+xml,<image href=https://evil/x>"></object>') is True
|
||||
assert rh('<iframe src="data:text/html;base64,not-valid-***"></iframe>') is True
|
||||
# A declared charset is honoured so a UTF-16 document is decoded like the
|
||||
# browser would; an unknown charset fails closed instead of hiding the load.
|
||||
assert (
|
||||
rh(
|
||||
'<iframe src="data:text/html;charset=utf-16le;base64,'
|
||||
'PABpAG0AZwAgAHMAcgBjAD0AaAB0AHQAcABzADoALwAvAGUAdgBpAGwALwB4AD4A"></iframe>'
|
||||
)
|
||||
is True
|
||||
)
|
||||
assert (
|
||||
rh(
|
||||
'<iframe src="data:text/html;charset=utf-16le;base64,'
|
||||
'PABoADEAPgBMAG8AYwBhAGwAPAAvAGgAMQA+AA=="></iframe>'
|
||||
)
|
||||
is False
|
||||
)
|
||||
assert (
|
||||
rh('<iframe src="data:text/html;charset=nonesuch,%3Cimg%3E"></iframe>') is True
|
||||
) # unknown charset fails closed
|
||||
assert rh("<script>frame.src='data:text/html,<img src=https://evil/x>'</script>") is True
|
||||
assert (
|
||||
rh(
|
||||
|
|
|
|||
|
|
@ -394,6 +394,61 @@ class TestSandboxEnvIsolation:
|
|||
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
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code",
|
||||
[
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue