e8945cab46* Add whole-document context mode to RAG chat attachments
Thread-attached files are injected in full when they fit a token budget,
instead of only top-K retrieved chunks, so the model reads the entire file
for summarize/reason-over-document requests. Oversized files fall back to
top-K retrieval so the context window is never blown. KB and project
corpora are unchanged (still retrieval).
- core/rag/store.py: all_chunks_for_scope returns every completed-document
chunk for a scope, ordered document-then-index, joined with filename.
- core/rag/tool.py: whole_document_context renders the chunks as the same
<chunk> blocks + citation source-map retrieval produces, returns None
when empty or over budget.
- core/inference/tools.py: build_rag_autoinject tries whole-document first
for thread scopes, falls through to search_for_autoinject otherwise.
- core/rag/config.py: THREAD_WHOLE_DOC + WHOLE_DOC_MAX_TOKENS (env-tunable).
- tests/test_rag_whole_document.py: store ordering, whole-doc render +
budget cutoff, auto-inject whole-doc vs top-K fallback, KB never whole-doc.
* Add scanned-PDF OCR fallback to RAG ingestion
A PDF page with no extractable text layer (a scanned or image-only page)
previously ingested as empty, so image PDFs were invisible to retrieval and
whole-document context. Such pages are now rendered and transcribed by the
loaded vision model during ingestion, so they become searchable and readable
like any other page. This restores OCR for the RAG document flow without a
separate extraction pipeline.
- core/rag/parsers.py: render_pdf_pages renders whole pages (1-based) to PNG.
- core/rag/captioner.py: factor the shared vision call into _vision_complete;
add _ocr_one + ocr_pages (transcribe rendered pages, OCR_MAX_PAGES bound).
- core/rag/ingestion.py: _ocr_scanned_pages runs right after parse, replacing
text on near-empty PDF pages. No-op when OCR is off, no page is scanned, or
no vision model is loaded (degrades like figure captioning).
- core/rag/config.py: OCR_SCANNED, OCR_MIN_CHARS, OCR_MAX_PAGES, OCR_DPI,
OCR_TIMEOUT_S, OCR_MAX_TOKENS (env-tunable).
- tests/test_rag_ocr_fallback.py: page render, ocr_pages gating + cap, scanned
PDF end-to-end OCR into chunks + whole-doc, born-digital skips OCR, disabled
leaves the page empty.
* Broaden OCR prompt to figures/tables and guard against repetition runaway
The OCR prompt now asks the vision model to also transcribe text inside figures,
diagrams, charts and tables, so labels and table cells on scanned pages are
indexed rather than skipped. Verified on real documents that this does not
regress plain-text transcription.
Some vision models loop on sparse images (e.g. a title-only cover) and emit the
same line hundreds of times. _collapse_runaway caps any run of identical
consecutive lines so a pathological page cannot flood the index; legitimate
short repeats (a label appearing a few times) survive. Applied in ocr_pages.
* Restrict whole-document injection to thread attachments only
whole_document_context resolved the combined project+thread scope, so a project
chat (the frontend sends both thread_id and project_id) injected the entire
project corpus in full, contradicting the design that project and KB corpora stay
retrieval-only. A large project corpus could also push the total over budget and
drop a small thread attachment back to top-K.
Resolve the thread scope alone in whole_document_context, and in
build_rag_autoinject only enter whole-doc mode when a thread attachment is present
and no KB is selected (a KB pick is exclusive: search that corpus). Project
sources and KBs keep top-K retrieval. Adds regression tests for the mixed
project+thread payload, the budget isolation, and KB precedence.
* Address review: keep project retrieval, harden budget + OCR guards
Follow-up to the 8-reviewer pass on the whole-document + OCR work.
- Preserve project grounding in project chats. The thread-scope-only fix made
whole-doc exclusive of retrieval, so a thread attachment silently dropped the
project corpus for that turn. build_rag_autoinject now whole-docs the thread
attachment AND retrieves the project sources top-K, merged under one citation
numbering via tool.render_sources. KB selection stays exclusive.
- Budget: a NULL/zero token_count no longer bypasses the cap (length-based
fallback in _row_token_count), so a malformed huge doc can't inject in full.
- OCR runaway guard: _collapse_runaway now also caps each distinct line at a
generous total across the page (not just consecutive), bounding the
interleaved/alternating loops weak models emit; blank-line floods collapse too.
- OCR: warn when a scanned PDF exceeds OCR_MAX_PAGES (pages past the cap stay
untranscribed) instead of silently dropping them.
- Document the known limits: OCR'd pages have no PDF highlight regions; vision
models need a micro-batch >= image tokens (Gemma-family) or the server aborts.
- Tests for project-retrieval composition, NULL-token budget, and interleaved
runaway; drop the now-superseded exclude-project test.
* Add OCR toggle to RAG retrieval settings
Make scanned-PDF OCR user-controllable per upload instead of only via the
RAG_OCR_SCANNED config default. The retrieval settings panel gains an OCR
scanned pages switch (persisted in localStorage, on by default); the chosen
value is read fresh at upload time and sent with each document upload.
Backend: the three upload routes accept an optional ocr form field and pass it
through start_ingestion to _ocr_scanned_pages, which now treats None as use the
config default and an explicit bool as an override. The on/off policy lives only
in _ocr_scanned_pages now, so ocr_pages no longer re-checks the config (that
double gate would have blocked a per-upload ocr=True while the default was off).
Tests cover both override directions (force on while config off, force off while
config on).
* Add "Describe figures & charts" toggle with chart-aware captions
Surface RAG figure captioning as a user control and make it actually useful for
graphs and plots. The figure detection already clustered vector drawings and
raster images into regions and rendered them, but captioning was off by default,
had no UI, and used a thin generic prompt.
Accuracy: the caption prompt now asks for chart type, axis titles and units,
legend or series, salient trends and readable values, and table columns, while
forbidding invented numbers. The token budget is configurable (CAPTION_MAX_TOKENS)
and captions pass through the same runaway guard as OCR so a looping vision model
cannot flood the index.
Control: a per-upload caption override threads from the three upload routes through
start_ingestion and _run, with the on/off policy single-sourced in _run (caption
self-gating removed from caption_images, mirroring the OCR change) so a force-on
override works when the config default is off. The frontend adds a "Describe
figures & charts" switch in the retrieval settings, persisted in localStorage and
sent with each upload. Default on; it is a no-op without a vision model and bounded
to CAPTION_MAX_IMAGES figures per document.
Tests cover the new caption_images contract, the runaway guard on captions, the
chart-aware prompt and token budget (and that OCR keeps its own prompt and budget),
and both override directions end to end through ingestion.
* Generalize figure understanding: transcribe-first prompt + high-DPI tiling
Make figure/chart description work across any visual and any model strength, not
just a strong VLM on simple figures. Two changes, validated by a recall benchmark
on authoritative documents (ResNet/Attention papers, USDA, UN UDHR).
1. Transcribe-first caption prompt. The caption now asks the model to transcribe
every visible label verbatim (titles, axis labels and units, legends, every
box/node/arrow label, table cells, equations) and then add a one-line summary,
instead of only describing the figure. Transcription is the most model-robust
visual task, so weak models that cannot reason about a chart still recover its
labels.
2. High-DPI tiling of figure pages. Figure-bearing pages are rendered as an
overlapping grid of high-DPI tiles (plus a full-page pass for context); each
tile is transcribed, then merged and de-duplicated. This keeps small diagram
labels legible and covers every sub-figure without relying on exact region
detection, which previously missed sub-figures and small labels.
Supporting changes: figure render DPI 130 -> 200 with a clip margin so edge labels
are not lost; vision calls are deterministic (temperature 0) so transcription does
not randomly drop labels; the repetition guard now applies to captions too. New
config knobs: FIGURE_DPI, FIGURE_MARGIN_FRAC, FIGURE_TILE_ROWS/COLS, FIGURE_TILE_
OVERLAP, FIGURE_FULLPAGE, CAPTION_MAX_PAGES, larger CAPTION_MAX_TOKENS, and
CAPTION_MAX_IMAGES as a per-document tile budget.
Measured figure context recall (per-label, dense academic figures):
Qwen2.5-VL: 0.50 -> 0.83 (overall 0.81 -> 0.94)
Gemma-4-E2B (weak): ~0 with loops -> 0.83 (overall 0.91)
Born-digital text and scanned-page recall are unchanged (no regression).
parsers gains _figure_boxes (shared detection), pages_with_figures, and
render_pdf_figure_tiles; captioner gains merge_page_captions and a temperature
parameter; ingestion routes figure captioning through the tiled path.
* Fix RAG review issues: whole-doc budget pre-check, figure gating, empty re-ingest, vision auth
Whole-document context now runs a cheap token-sum pre-check (store.scope_token_estimate)
before hydrating every chunk's text, so an attachment that cannot fit the budget is
rejected without loading the whole corpus into memory. The estimate mirrors
all_chunks_for_scope's filter and the per-row token-count fallback exactly.
Ingestion skips all figure work (PDF rasterization and detection, not just the caption
call) unless a vision model is loaded, so a text-only deployment pays nothing. When OCR
is enabled, scanned/image-only pages are excluded from figure tiling since OCR already
transcribes them whole, avoiding double vision work and overlapping index entries; a
scanned figure page is still tiled when OCR is off.
start_ingestion no longer dedupes forever to a prior ingest that produced zero chunks
(e.g. a scanned PDF uploaded before a vision model was loaded): the empty record is
dropped and the content is re-ingested.
Vision OCR and caption requests now send the backend Authorization header, so they
match the chat endpoint and do not 401 under direct-stream (--api-key) mode.
Adds tests for the budget estimate, scanned-page exclusion, the vision-model gate, the
empty re-ingest path, and the auth-header passthrough.
* Trim RAG vision-ingestion comments and docstrings
Tighten the verbose multi-line docstrings and comments added across the RAG vision
ingestion work (captioner, config, parsers, ingestion, store, tool, build_rag_autoinject,
the RAG tests, and the chat-store/upload-hook frontend toggles) to one or two lines while
keeping their intent. No code changed: verified comment/docstring-only against the prior
commit, and the RAG test suite still passes.
* Fix figure-tiling exclusion and client dedupe for re-ingestable docs
Figure tiling now excludes only the pages OCR actually transcribed, not every
text-less page. _ocr_scanned_pages returns the set of pages it OCR'd, and _run passes
that to pages_with_figures as exclude_pages (replacing the ocr_on-keyed min_text_chars
heuristic). A scanned page that OCR skipped (past OCR_MAX_PAGES, or whose OCR returned
empty) is no longer dropped from captioning, so a chart on such a page still gets a
caption.
The document panel's upload dedupe no longer skips re-selecting a file whose only
matching doc completed with zero chunks. Such a doc is re-ingestable (e.g. a scan
attached before a vision model loaded), and the backend re-ingests on the same content
hash, so the client must let it reach the backend; healthy or still-indexing docs are
still skipped. The SSE complete frame's chunk count is recorded on the doc so the
check is exact.
Adds a regression test for the un-OCR'd scanned figure page and updates the
pages_with_figures test to the exclude_pages interface.
* Address review findings: whole-doc budget guard, job numChunks, dead code, upload cap
whole_document_context now treats a non-positive max_tokens as "never inject" instead
of injecting the whole corpus unbounded, so RAG_WHOLE_DOC_MAX_TOKENS=0 tightens rather
than disables the budget (the real off switch stays RAG_THREAD_WHOLE_DOC=0).
The job-status endpoint and get_job_status now expose num_chunks (joined from the
document), and the upload hook threads it through the SSE-fallback completion paths
(reconcile + poll). Previously a document that finished via the connection-cap fallback
had no chunk count client-side, so the re-ingest dedupe wrongly treated it as empty and
re-uploaded it. IndexJob/JobEvent gain the field and the untyped cast is dropped.
Removes the dead render_pdf_figures function (superseded by the tiling path), its test,
and the unused FIGURE_MARGIN_FRAC config knob.
Adds an upload size cap (RAG_MAX_UPLOAD_BYTES, default 200 MB; 413 on exceed with the
partial file cleaned up) so a pathological file can't drive unbounded parse + vision
work. render_pdf_figure_tiles clamps rows/cols to >= 1 (no ZeroDivisionError on a
misconfigured grid). Captioning progress is reported after OCR so the bar is monotonic.
sqlite connections set busy_timeout=5000 so a long figure/scan ingest holding its
connection doesn't make a concurrent ingest/read fail with "database is locked".
Adds tests for the non-positive budget, the zero-grid clamp, job-status num_chunks, and
the oversize-upload rejection.
* Extract PDF text as layout-aware Markdown via pymupdf4llm
parsers._pdf now extracts each PDF page as Markdown with pymupdf4llm.to_markdown
(page_chunks=True) instead of flat page.get_text("text"), so tables, headings and lists
keep their structure in the indexed chunks and retrieve far better (a table's cells stay
associated with their row instead of flattening into a token stream). Gated by
RAG_PDF_MARKDOWN (default on); falls back to plain PyMuPDF text when the toggle is off,
pymupdf4llm is missing, extraction fails, or a page yields no Markdown. The scanned-page
OCR and figure-tiling passes operate on rendered pixels and are unaffected; docx/html/txt
keep their existing extractors.
The preview-highlight locator already strips Markdown punctuation when building anchors;
it now also splits anchor tokens on pipes so a Markdown table row still anchors to the
raw PDF word stream.
Declares pymupdf4llm as a studio/RAG dependency (was only transitively present via the
data-designer plugin). Adds parser tests (Markdown table reaches the page text, the
plain-text fallback, the missing-lib fallback) and a locator test for table-pipe anchoring.
* Pin pymupdf4llm to 0.3.4 so the package scan does not pull onnxruntime
The lockstep pymupdf4llm 1.27.x line makes pymupdf-layout a hard dependency,
which in turn pulls onnxruntime (plus numpy/networkx/protobuf). The security-audit
pip scan-packages job resolves requirements --with-deps, so adding pymupdf4llm to
no-torch-runtime.txt and studio.txt surfaced onnxruntime's un-baselined CRITICAL
finding and flipped the hf-stack shard from pass to fail.
pymupdf4llm 0.3.x keeps pymupdf-layout behind an optional [layout] extra, so a plain
install resolves to pymupdf + tabulate only and never touches onnxruntime. 0.3.4
requires pymupdf>=1.27.1, satisfied by our pinned pymupdf==1.27.2.3, and to_markdown
(page_chunks=True) produces equivalent layout-aware Markdown on real PDFs (verified on
the Attention, ResNet and USDA documents). Production already installs these files
--no-deps, so onnxruntime was never shipped at runtime; this only fixes the scanner.
The parser test now asserts Markdown markup (heading or table pipes) rather than table
pipes specifically, since 0.3.4 emits a heading but not a pipe table on the tiny
borderless synthetic fixture; both markers are absent from the plain-text fallback.
* Fix RAG whole-doc review findings
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Address RAG whole-doc review follow-ups
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Address RAG review follow-up edge cases
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Reserve image budget for whole-document RAG
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
---------
Co-authored-by: danielhanchen <michaelhan2050@gmail.com>
Co-authored-by: Lee Jackson <130007945+Imagineer99@users.noreply.github.com>
Co-authored-by: wasimysaid <wasimysdev@gmail.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
2807 lines
104 KiB
Python
2807 lines
104 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
|
|
|
|
"""Tool definitions and executors for LLM tool calling: web search
|
|
(DuckDuckGo), Python code execution, and terminal commands."""
|
|
|
|
import ast
|
|
import http.client
|
|
import os
|
|
import signal
|
|
|
|
os.environ["UNSLOTH_IS_PRESENT"] = "1"
|
|
|
|
import asyncio
|
|
import random
|
|
import re
|
|
import shlex
|
|
import ssl
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import urllib.request
|
|
|
|
from core.inference.mcp_client import (
|
|
MCP_TOOL_PREFIX,
|
|
TOOL_CACHE_INVALIDATING_FIELDS,
|
|
cache_tools,
|
|
call_tool_sync,
|
|
get_cached_tools,
|
|
in_failure_cooloff,
|
|
is_stdio,
|
|
list_tools_async,
|
|
parse_server_headers,
|
|
probe_timeout,
|
|
record_probe_failure,
|
|
stdio_mcp_enabled,
|
|
)
|
|
from storage import mcp_servers_db
|
|
|
|
from loggers import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
_EXEC_TIMEOUT = 300 # 5 minutes
|
|
|
|
# Splits the UI source-map from the result; loops strip it (like __IMAGES__).
|
|
RAG_SOURCES_SENTINEL = "\n__RAG_SOURCES__:"
|
|
|
|
# Import these at module level so the preexec_fn closure triggers no imports in
|
|
# the forked child (which can deadlock multi-threaded servers).
|
|
_libc = None
|
|
if sys.platform == "linux":
|
|
try:
|
|
import ctypes
|
|
import ctypes.util
|
|
|
|
_libc_name = ctypes.util.find_library("c")
|
|
if _libc_name:
|
|
_libc = ctypes.CDLL(_libc_name, use_errno = True)
|
|
except (OSError, AttributeError):
|
|
pass
|
|
|
|
_resource = None
|
|
if sys.platform != "win32":
|
|
try:
|
|
import resource as _resource
|
|
except ImportError:
|
|
pass
|
|
|
|
# Raster-image allowlist for sandbox file serving.
|
|
# No .svg (XSS via embedded scripts), no .html, no .pdf.
|
|
_IMAGE_EXTS = frozenset({".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp"})
|
|
_MAX_OUTPUT_CHARS = 8000 # truncate long output
|
|
_BLOCKED_COMMANDS_COMMON = frozenset(
|
|
{
|
|
"rm",
|
|
"dd",
|
|
"chmod",
|
|
"chown",
|
|
"mkfs",
|
|
"mount",
|
|
"umount",
|
|
"fdisk",
|
|
"sudo",
|
|
"su",
|
|
"doas",
|
|
"pkexec",
|
|
"shutdown",
|
|
"reboot",
|
|
"halt",
|
|
"poweroff",
|
|
"kill",
|
|
"killall",
|
|
"pkill",
|
|
"passwd",
|
|
"curl",
|
|
"wget",
|
|
"nc",
|
|
"ncat",
|
|
"netcat",
|
|
"socat",
|
|
"ssh",
|
|
"scp",
|
|
"sftp",
|
|
"rsync",
|
|
"eval",
|
|
"source",
|
|
}
|
|
)
|
|
_BLOCKED_COMMANDS_WIN = frozenset(
|
|
{
|
|
"rmdir",
|
|
"takeown",
|
|
"icacls",
|
|
"runas",
|
|
"powershell",
|
|
"pwsh",
|
|
}
|
|
)
|
|
_BLOCKED_COMMANDS = (
|
|
_BLOCKED_COMMANDS_COMMON | _BLOCKED_COMMANDS_WIN
|
|
if sys.platform == "win32"
|
|
else _BLOCKED_COMMANDS_COMMON
|
|
)
|
|
|
|
|
|
_SHELL_SEPARATORS = frozenset({";", "&&", "||", "|", "&", "\n", "(", ")", "`", "{", "}"})
|
|
# Bash keywords starting a new command position (then $cmd, do $cmd, etc.).
|
|
_SHELL_KEYWORDS_AS_SEP = frozenset({"then", "do", "else", "elif"})
|
|
# Wrappers whose next non-flag argument is the command Bash will exec.
|
|
_COMMAND_PREFIXES = frozenset(
|
|
{
|
|
"env",
|
|
"command",
|
|
"builtin",
|
|
"exec",
|
|
"time",
|
|
"nohup",
|
|
"nice",
|
|
"setsid",
|
|
"stdbuf",
|
|
"timeout",
|
|
"ionice",
|
|
"chroot",
|
|
"sudo",
|
|
"doas",
|
|
"su",
|
|
"xargs",
|
|
}
|
|
)
|
|
_ASSIGNMENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=")
|
|
_FIND_EXEC_FLAGS = frozenset({"-exec", "-execdir", "-ok", "-okdir"})
|
|
|
|
|
|
def _find_blocked_commands(command: str) -> set[str]:
|
|
"""Detect blocked commands at shell command position only.
|
|
|
|
A token is at command position if it is the first token, or follows a
|
|
shell separator / brace-group opener / new-command keyword (`then`, `do`,
|
|
etc.), or a command-prefix wrapper like `env` / `time` / `xargs` (next
|
|
token is the real command). Tokens in argument position (`grep -r curl .`,
|
|
`echo source the data`, `ls /usr/bin/curl`) pass through. Also scans
|
|
`find ... -exec CMD` and recurses into bash -c / cmd /c.
|
|
"""
|
|
blocked: set[str] = set()
|
|
|
|
# punctuation_chars splits separators into their own tokens, so command
|
|
# position is detected even in `echo done; rm -rf x` (no whitespace) or
|
|
# quote-split names (`r''m` collapses to `rm` after `;`).
|
|
try:
|
|
if sys.platform == "win32":
|
|
tokens = shlex.split(command, posix = False)
|
|
else:
|
|
lexer = shlex.shlex(command, posix = True, punctuation_chars = ";&|()`")
|
|
lexer.whitespace_split = True
|
|
tokens = list(lexer)
|
|
except ValueError:
|
|
tokens = command.split()
|
|
|
|
def _token_basename(tok: str) -> str:
|
|
# Strip glued-on meta-chars (`rm;`) so the basename still matches `rm`.
|
|
tok = tok.strip(";&|()`{}")
|
|
base = os.path.basename(tok).lower()
|
|
stem, ext = os.path.splitext(base)
|
|
if ext in {".exe", ".com", ".bat", ".cmd"}:
|
|
base = stem
|
|
return base
|
|
|
|
expect_command = True # start of string is a command position
|
|
prefix_pending = False # last cmd-position token was a wrapper (env/time/xargs/...)
|
|
for token in tokens:
|
|
if token in _SHELL_SEPARATORS or token in _SHELL_KEYWORDS_AS_SEP:
|
|
expect_command = True
|
|
prefix_pending = False
|
|
continue
|
|
if token.startswith("-"):
|
|
# Flags belong to the active command, but keep expect_command while a
|
|
# wrapper prefix awaits its command (`stdbuf -oL cmd`, `xargs -- cmd`).
|
|
if not prefix_pending:
|
|
expect_command = False
|
|
continue
|
|
if not expect_command:
|
|
continue
|
|
# FOO=bar assignment prefix; next non-assignment token is the command.
|
|
if _ASSIGNMENT_RE.match(token):
|
|
continue
|
|
# Numeric wrapper arg: `timeout 1 cmd` / `nice -n 5 cmd`.
|
|
if prefix_pending and token.lstrip("-").isdigit():
|
|
continue
|
|
base = _token_basename(token)
|
|
if base in _BLOCKED_COMMANDS:
|
|
blocked.add(base)
|
|
# Wrappers (env/time/xargs/sudo) consume one command; the next non-flag,
|
|
# non-numeric token is the real command. sudo is also in _BLOCKED_COMMANDS.
|
|
if base in _COMMAND_PREFIXES:
|
|
prefix_pending = True
|
|
continue
|
|
expect_command = False
|
|
prefix_pending = False
|
|
|
|
# `find ... -exec CMD ... ;` and `-execdir CMD ... ;` invoke CMD directly.
|
|
for i, tok in enumerate(tokens):
|
|
if tok in _FIND_EXEC_FLAGS and i + 1 < len(tokens):
|
|
base = _token_basename(tokens[i + 1])
|
|
if base in _BLOCKED_COMMANDS:
|
|
blocked.add(base)
|
|
|
|
# Regex catches blocked words at command boundaries shlex misses: inside
|
|
# $(rm -rf), <(rm), backtick chains, or "foo;rm". Anchored to command-position
|
|
# delimiters, so it doesn't match in argument position.
|
|
lowered = command.lower()
|
|
if _BLOCKED_COMMANDS:
|
|
words_alt = "|".join(re.escape(w) for w in sorted(_BLOCKED_COMMANDS))
|
|
pattern = (
|
|
rf"(?:^|[;&|`\n(]\s*|[$]\(\s*|<\(\s*)"
|
|
rf"(?:[\w./\\-]*/|[a-zA-Z]:[/\\][\w./\\-]*)?"
|
|
rf"({words_alt})(?:\.(?:exe|com|bat|cmd))?\b"
|
|
)
|
|
blocked.update(re.findall(pattern, lowered))
|
|
|
|
# Nested shell invocations (bash -c '...', bash -lc '...', cmd /c '...'):
|
|
# on a -c/-/c flag, look back for a shell name (skipping flags) and
|
|
# recursively scan the nested command string.
|
|
_SHELLS = {"bash", "sh", "zsh", "dash", "ksh", "csh", "tcsh", "fish"}
|
|
_SHELLS_WIN = {"cmd", "cmd.exe"}
|
|
for i, token in enumerate(tokens):
|
|
tok_lower = token.lower()
|
|
# Match -c exactly, or combined flags ending in c (e.g. -lc, -xc)
|
|
is_unix_c = tok_lower == "-c" or (
|
|
tok_lower.startswith("-") and tok_lower.endswith("c") and not tok_lower.startswith("--")
|
|
)
|
|
is_win_c = tok_lower == "/c"
|
|
if not (is_unix_c or is_win_c) or i < 1 or i + 1 >= len(tokens):
|
|
continue
|
|
# Look back past flags for the shell binary. Windows flags and absolute
|
|
# paths both start with /, so only skip short /X flags (not /bin/bash).
|
|
for j in range(i - 1, -1, -1):
|
|
prev = tokens[j]
|
|
if prev.startswith("-"):
|
|
continue # skip Unix flags like --login, -l
|
|
if is_win_c and prev.startswith("/") and len(prev) <= 3:
|
|
continue # skip Windows flags like /s, /q (not /bin/bash)
|
|
prev_base = os.path.basename(prev).lower()
|
|
if is_unix_c and prev_base in _SHELLS:
|
|
blocked |= _find_blocked_commands(tokens[i + 1])
|
|
elif is_win_c and prev_base in _SHELLS_WIN:
|
|
blocked |= _find_blocked_commands(tokens[i + 1])
|
|
break # stop at first non-flag token
|
|
|
|
return blocked
|
|
|
|
|
|
def _build_safe_env(workdir: str) -> dict[str, str]:
|
|
"""Build a minimal, credential-free environment for sandboxed subprocesses.
|
|
|
|
Whitelist-built from scratch (parent env NOT inherited): only PATH/HOME/
|
|
TMPDIR/LANG/TERM/PYTHONIOENCODING (+VIRTUAL_ENV or Windows SystemRoot) reach
|
|
the child; all credential vars (HF_TOKEN, AWS_*, etc.) are absent. HOME
|
|
points at the sandbox workdir so SDKs can't read the operator's cached creds.
|
|
"""
|
|
# Start from the running interpreter's dir so 'python'/'pip' resolve to the
|
|
# same environment the Studio server runs in.
|
|
exe_dir = os.path.dirname(sys.executable)
|
|
path_entries = [exe_dir] if exe_dir else []
|
|
|
|
# If a virtualenv is active, include its bin/Scripts directory.
|
|
venv = os.environ.get("VIRTUAL_ENV")
|
|
if venv:
|
|
venv_bin = os.path.join(venv, "Scripts" if sys.platform == "win32" else "bin")
|
|
if venv_bin not in path_entries:
|
|
path_entries.append(venv_bin)
|
|
|
|
if sys.platform == "win32":
|
|
sysroot = os.environ.get("SystemRoot", r"C:\Windows")
|
|
path_entries.extend([os.path.join(sysroot, "System32"), sysroot])
|
|
else:
|
|
path_entries.extend(["/usr/local/bin", "/usr/bin", "/bin"])
|
|
|
|
# Deduplicate, preserving order.
|
|
deduped = list(dict.fromkeys(p for p in path_entries if p))
|
|
|
|
env = {
|
|
"PATH": os.pathsep.join(deduped),
|
|
"HOME": workdir,
|
|
"TMPDIR": workdir,
|
|
"LANG": os.environ.get("LANG", "C.UTF-8"),
|
|
"TERM": "dumb",
|
|
"PYTHONIOENCODING": "utf-8",
|
|
}
|
|
if venv:
|
|
env["VIRTUAL_ENV"] = venv
|
|
# Windows needs SystemRoot for Python/subprocess to work.
|
|
if sys.platform == "win32":
|
|
env["SystemRoot"] = os.environ.get("SystemRoot", r"C:\Windows")
|
|
return env
|
|
|
|
|
|
# Credential env vars dropped even in bypass mode so tool code cannot read the
|
|
# operator's keys. Over-strips on purpose (a benign var is harmless to lose).
|
|
_BYPASS_ENV_SECRET_NAMES = frozenset(
|
|
{
|
|
"HF_TOKEN",
|
|
"HF_HUB_TOKEN",
|
|
"HUGGING_FACE_HUB_TOKEN",
|
|
"HUGGINGFACE_TOKEN",
|
|
"HUGGINGFACEHUB_API_TOKEN",
|
|
"WANDB_API_KEY",
|
|
"GH_TOKEN",
|
|
"GITHUB_TOKEN",
|
|
"OPENAI_API_KEY",
|
|
"ANTHROPIC_API_KEY",
|
|
"GEMINI_API_KEY",
|
|
"GOOGLE_API_KEY",
|
|
"GROQ_API_KEY",
|
|
"OPENROUTER_API_KEY",
|
|
"REPLICATE_API_TOKEN",
|
|
"COHERE_API_KEY",
|
|
"MISTRAL_API_KEY",
|
|
"NGC_API_KEY",
|
|
"KAGGLE_KEY",
|
|
"MYSQL_PWD", # exact name: markers use PASSWD, not PWD (PWD is the cwd var)
|
|
"LD_PRELOAD",
|
|
# Auth brokers / capability handles: not secrets by value, but they
|
|
# hand the child the operator's live agent (ssh/gpg), kube config, or
|
|
# docker daemon. Names are listed because there is no value signal to
|
|
# key off. URL config vars (HTTP_PROXY, PIP_INDEX_URL, DATABASE_URL,
|
|
# ...) are intentionally NOT name-listed: a benign proxy/index without
|
|
# credentials must keep working in bypass mode, while a credentialed
|
|
# value is dropped by _is_secret_env_value() regardless of its name.
|
|
"SSH_AUTH_SOCK",
|
|
"SSH_AGENT_PID",
|
|
"GPG_AGENT_INFO",
|
|
"GNUPGHOME",
|
|
"KUBECONFIG",
|
|
"DOCKER_HOST",
|
|
}
|
|
)
|
|
_BYPASS_ENV_SECRET_PREFIXES = ("AWS_", "AZURE_", "GOOGLE_", "GCP_", "GCLOUD_", "DYLD_")
|
|
_BYPASS_ENV_SECRET_MARKERS = (
|
|
"TOKEN",
|
|
"API_KEY",
|
|
"APIKEY",
|
|
"SECRET",
|
|
"PASSWORD",
|
|
"PASSWD",
|
|
"CREDENTIAL",
|
|
"PRIVATE_KEY",
|
|
"AUTH", # e.g. NPM_CONFIG__AUTH (npm _auth), REDISCLI_AUTH
|
|
# Azure App Service connection strings: SQLCONNSTR_/CUSTOMCONNSTR_/... and
|
|
# WEBSITE_CONTENTAZUREFILECONNECTIONSTRING carry DB/storage credentials.
|
|
"CONNSTR",
|
|
"CONNECTIONSTRING",
|
|
)
|
|
# Non-secret hardening flags that match a secret prefix/marker but must be KEPT
|
|
# so bypass mode does not silently undo an operator's opt-out. AWS_EC2_METADATA_
|
|
# DISABLED tells the AWS SDK/CLI not to pull instance-role creds from IMDS;
|
|
# dropping it would re-open that path for a bypassed tool.
|
|
_BYPASS_ENV_KEEP_NAMES = frozenset(
|
|
{
|
|
"AWS_EC2_METADATA_DISABLED",
|
|
"AWS_EC2_METADATA_V1_DISABLED",
|
|
}
|
|
)
|
|
# Matches a URL that embeds userinfo before the host, covering both
|
|
# "scheme://user:pass@host" and token-only "scheme://token@host" (and
|
|
# percent-encoded variants). The userinfo must precede the first '/', so an '@'
|
|
# in a path or query does not false-positive. Used to scrub credential-bearing
|
|
# URL values regardless of the variable's name.
|
|
_URL_USERINFO_RE = re.compile(r"://[^/\s@]+@")
|
|
# Connection-string credential fields (ADO.NET / Azure storage / Service Bus):
|
|
# "...;Password=...", "...;AccountKey=...", "...;SharedAccessKey=...". Catches
|
|
# credential-bearing values whose names dodge the name classifier. "accesskey"
|
|
# also covers Shared/Secret AccessKey via substring; the Name fields (e.g.
|
|
# SharedAccessKeyName=) do not match since "=" must follow the keyword.
|
|
_SECRET_VALUE_RE = re.compile(r"(?i)(?:password|pwd|accountkey|accesskey)\s*=\s*[^\s;]")
|
|
|
|
# Names that hold no secret value but point SDKs at the operator's real
|
|
# home/cache/config (cached tokens, cred files), defeating the HOME repoint.
|
|
# Startup always sets HF_HOME (-> $HF_HOME/token), so this is the live leak.
|
|
# Dropped in bypass mode so tools fall back to the empty repointed HOME.
|
|
_BYPASS_ENV_CRED_LOCATION_NAMES = frozenset(
|
|
{
|
|
# HF cache roots (token lives under $HF_HOME/token)
|
|
"HF_HOME",
|
|
"HF_HUB_CACHE",
|
|
"HUGGINGFACE_HUB_CACHE",
|
|
"HF_XET_CACHE",
|
|
"TRANSFORMERS_CACHE",
|
|
"HF_DATASETS_CACHE",
|
|
"HF_ASSETS_CACHE",
|
|
# XDG base dirs (resolved before $HOME)
|
|
"XDG_CONFIG_HOME",
|
|
"XDG_CACHE_HOME",
|
|
"XDG_DATA_HOME",
|
|
# explicit cred/config file pointers honoured before $HOME
|
|
"NETRC",
|
|
"PGPASSFILE",
|
|
"BOTO_CONFIG",
|
|
"PIP_CONFIG_FILE",
|
|
"CLOUDSDK_CONFIG",
|
|
"KAGGLE_CONFIG_DIR",
|
|
"DOCKER_CONFIG",
|
|
"WANDB_DIR",
|
|
"WANDB_CONFIG_DIR",
|
|
"WANDB_CACHE_DIR",
|
|
# package-manager / git / cloud config pointers to real cred files
|
|
"NPM_CONFIG_USERCONFIG",
|
|
"NPM_CONFIG_GLOBALCONFIG",
|
|
"YARN_RC_FILENAME",
|
|
"GIT_CONFIG_GLOBAL",
|
|
"GIT_CONFIG_SYSTEM",
|
|
"CARGO_HOME",
|
|
"RCLONE_CONFIG",
|
|
# auth-helper scripts that hand creds to git/ssh
|
|
"GIT_ASKPASS",
|
|
"SSH_ASKPASS",
|
|
# shell startup hook: bash -c sources $BASH_ENV (can re-export secrets)
|
|
"BASH_ENV",
|
|
# Windows: HOMEDRIVE+HOMEPATH compose a home that bypasses HOME
|
|
"HOMEDRIVE",
|
|
"HOMEPATH",
|
|
}
|
|
)
|
|
# Windows profile dirs SDKs read creds under; repointed (not dropped) since
|
|
# callers expect them present.
|
|
_BYPASS_ENV_WINDOWS_PROFILE_VARS = ("USERPROFILE", "APPDATA", "LOCALAPPDATA")
|
|
|
|
|
|
def _is_secret_env_name(name: str) -> bool:
|
|
"""True if an env var name looks like it carries a credential."""
|
|
upper = name.upper()
|
|
if upper in _BYPASS_ENV_KEEP_NAMES:
|
|
return False # non-secret hardening flag; keep it
|
|
if upper in _BYPASS_ENV_SECRET_NAMES:
|
|
return True
|
|
if any(upper.startswith(p) for p in _BYPASS_ENV_SECRET_PREFIXES):
|
|
return True
|
|
return any(marker in upper for marker in _BYPASS_ENV_SECRET_MARKERS)
|
|
|
|
|
|
def _is_cred_location_env_name(name: str) -> bool:
|
|
"""True for vars that point SDKs at the real home/cache/config (cached creds)."""
|
|
return name.upper() in _BYPASS_ENV_CRED_LOCATION_NAMES
|
|
|
|
|
|
def _is_secret_env_value(value: str) -> bool:
|
|
"""True if a value embeds credentials regardless of its name.
|
|
|
|
Catches URL userinfo (``scheme://user:token@host`` in DATABASE_URL /
|
|
PIP_INDEX_URL / HTTP_PROXY) and connection-string credential fields
|
|
(``...;Password=...`` / ``...;AccountKey=...``) whose names dodge the name
|
|
classifier.
|
|
"""
|
|
if not value:
|
|
return False
|
|
return _URL_USERINFO_RE.search(value) is not None or _SECRET_VALUE_RE.search(value) is not None
|
|
|
|
|
|
def _build_bypass_env(workdir: str) -> dict[str, str]:
|
|
"""Env for bypass exec: full host env (unrestricted) minus credential vars,
|
|
with HOME/TMPDIR repointed at the workdir so SDKs cannot read cached creds.
|
|
|
|
Note: stripping the child env is necessary but not sufficient on its own -
|
|
a same-UID child can still read the parent's environment via procfs, so
|
|
callers also harden the parent (see _harden_parent_against_proc_env_leak).
|
|
"""
|
|
env = {
|
|
k: v
|
|
for k, v in os.environ.items()
|
|
if not _is_secret_env_name(k)
|
|
and not _is_secret_env_value(v)
|
|
and not _is_cred_location_env_name(k)
|
|
}
|
|
env["HOME"] = workdir
|
|
env["TMPDIR"] = workdir
|
|
# Windows tempfile / SDKs honour TEMP/TMP, not TMPDIR; repoint all three so
|
|
# the bypassed tool writes under the per-session sandbox dir on every OS.
|
|
env["TEMP"] = workdir
|
|
env["TMP"] = workdir
|
|
# Windows SDKs read creds under the profile dirs, not $HOME; repoint set
|
|
# ones to the workdir (HOMEDRIVE/HOMEPATH are dropped above).
|
|
for var in _BYPASS_ENV_WINDOWS_PROFILE_VARS:
|
|
if var in os.environ:
|
|
env[var] = workdir
|
|
return env
|
|
|
|
|
|
def _sandbox_preexec():
|
|
"""Best-effort sandbox setup for sandboxed subprocesses (modules are
|
|
resolved at import time so the forked child runs no imports)."""
|
|
try:
|
|
os.setsid()
|
|
except OSError:
|
|
pass
|
|
|
|
try:
|
|
os.umask(0o077)
|
|
except OSError:
|
|
pass
|
|
|
|
if _libc is not None:
|
|
try:
|
|
_libc.prctl(38, 1, 0, 0, 0) # PR_SET_NO_NEW_PRIVS
|
|
except (OSError, AttributeError):
|
|
pass
|
|
|
|
try:
|
|
_libc.prctl(1, 9, 0, 0, 0) # PR_SET_PDEATHSIG = SIGKILL
|
|
except (OSError, AttributeError):
|
|
pass
|
|
|
|
# CLONE_NEWNET not applied: with userns enabled it blocks all egress,
|
|
# including allowlisted hosts. Network policy is enforced by the AST
|
|
# host check and the bash blocklist.
|
|
|
|
if _resource is not None:
|
|
# RLIMIT_NPROC is per-real-UID, so the cap is well above normal usage.
|
|
try:
|
|
nproc = int(os.environ.get("UNSLOTH_STUDIO_SANDBOX_NPROC", "10000"))
|
|
_resource.setrlimit(_resource.RLIMIT_NPROC, (nproc, nproc))
|
|
except (ValueError, OSError, AttributeError):
|
|
pass
|
|
try:
|
|
_resource.setrlimit(_resource.RLIMIT_FSIZE, (100 * 1024 * 1024, 100 * 1024 * 1024))
|
|
except (ValueError, OSError):
|
|
pass
|
|
try:
|
|
as_bytes = int(os.environ.get("UNSLOTH_STUDIO_SANDBOX_AS_GB", "8")) * 1024 * 1024 * 1024
|
|
_resource.setrlimit(_resource.RLIMIT_AS, (as_bytes, as_bytes))
|
|
except (ValueError, OSError, AttributeError):
|
|
pass
|
|
try:
|
|
cpu_s = int(os.environ.get("UNSLOTH_STUDIO_SANDBOX_CPU_S", "600"))
|
|
_resource.setrlimit(_resource.RLIMIT_CPU, (cpu_s, cpu_s))
|
|
except (ValueError, OSError, AttributeError):
|
|
pass
|
|
try:
|
|
# High enough for multi-shard safetensors mmaps; tunable via env.
|
|
# Clamp to the inherited hard limit so setrlimit doesn't ValueError
|
|
# when the parent's hard cap is below the request.
|
|
nofile = int(os.environ.get("UNSLOTH_STUDIO_SANDBOX_NOFILE", "16384"))
|
|
_soft_cur, hard_cur = _resource.getrlimit(_resource.RLIMIT_NOFILE)
|
|
target = nofile if hard_cur == _resource.RLIM_INFINITY else min(nofile, hard_cur)
|
|
_resource.setrlimit(_resource.RLIMIT_NOFILE, (target, target))
|
|
except (ValueError, OSError, AttributeError):
|
|
pass
|
|
|
|
|
|
def _bypass_preexec():
|
|
"""Minimal pre-exec for bypass exec: os.setsid() only.
|
|
|
|
Required, not a restriction: _kill_process_tree does killpg(getpgid(child)),
|
|
so without a new session a timeout/cancel would kill the Studio server too.
|
|
"""
|
|
try:
|
|
os.setsid()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
# Hardening the Studio parent is done once (PR_SET_DUMPABLE is process-global
|
|
# and sticky); guarded so repeated bypass calls do not re-issue the prctl.
|
|
_parent_proc_hardened = False
|
|
|
|
|
|
def _harden_parent_against_proc_env_leak() -> bool:
|
|
"""Make the Studio process's /proc/<pid>/environ unreadable to its children.
|
|
|
|
Stripping the child env is not enough on Linux: a bypassed same-UID child
|
|
runs unsandboxed and can read /proc/<getppid()>/environ to recover the
|
|
tool-executing process's *unfiltered* secrets (HF_TOKEN, cloud keys, ...).
|
|
Clearing the dumpable flag (PR_SET_DUMPABLE=0) reparents this process's
|
|
/proc entries to root, so a same-UID child can no longer read its environ.
|
|
|
|
Returns True when the process is hardened or hardening is unnecessary (no
|
|
/proc leak off Linux), and False when it is needed but could not be applied
|
|
(e.g. prctl denied by a seccomp policy). Callers must fail closed - refuse
|
|
the unsandboxed exec - when this returns False, rather than running with the
|
|
parent environ still readable.
|
|
|
|
Scope: this closes the direct parent read (the demonstrated leak). It is a
|
|
mitigation, not a full boundary - a bypassed tool is unsandboxed by design,
|
|
so it can still walk /proc to a same-UID *ancestor* (e.g. the launching
|
|
shell) or read on-disk credentials by absolute path. Complete isolation
|
|
needs a separate uid / PID+mount namespace, which is out of scope here; the
|
|
UI already warns the mode is dangerous. Applied lazily on first bypass exec
|
|
so non-bypass operation is unchanged.
|
|
"""
|
|
global _parent_proc_hardened
|
|
if _parent_proc_hardened:
|
|
return True
|
|
if sys.platform != "linux":
|
|
return True # no /proc/<pid>/environ same-UID leak to close
|
|
if _libc is None:
|
|
return False # on Linux but cannot issue prctl -> cannot harden
|
|
try:
|
|
# prctl(PR_SET_DUMPABLE=4, SUID_DUMP_DISABLE=0). ctypes returns the
|
|
# syscall result (-1 on failure) and does NOT raise, so check it.
|
|
ret = _libc.prctl(4, 0, 0, 0, 0)
|
|
except (OSError, AttributeError):
|
|
return False
|
|
if ret != 0:
|
|
return False
|
|
_parent_proc_hardened = True
|
|
return True
|
|
|
|
|
|
def _get_shell_cmd(command: str) -> list[str]:
|
|
"""Return the platform-appropriate shell invocation for a command string."""
|
|
if sys.platform == "win32":
|
|
return ["cmd", "/c", command]
|
|
return ["bash", "-c", command]
|
|
|
|
|
|
# Per-session working directories so each chat thread gets its own sandbox.
|
|
# Falls back to ~/studio_sandbox/_default for callers without a session_id.
|
|
_workdirs: dict[str, str] = {}
|
|
|
|
|
|
# Non-matching session_ids collapse to ``_invalid`` to block cross-session escapes.
|
|
_SESSION_ID_RE = re.compile(r"\A[A-Za-z0-9_\-]{1,64}\Z")
|
|
_PROJECT_SESSION_PREFIX = "project-"
|
|
|
|
|
|
def _get_project_workdir(session_id: str) -> str | None:
|
|
if not session_id.startswith(_PROJECT_SESSION_PREFIX):
|
|
return None
|
|
project_id = session_id[len(_PROJECT_SESSION_PREFIX) :]
|
|
if not project_id or not _SESSION_ID_RE.match(project_id):
|
|
return None
|
|
try:
|
|
from storage.studio_db import ensure_chat_project_workspace
|
|
project = ensure_chat_project_workspace(project_id)
|
|
except Exception:
|
|
logger.warning("Failed to resolve project sandbox for %s", session_id, exc_info = True)
|
|
return None
|
|
if not project:
|
|
return None
|
|
root_path = project.get("rootPath")
|
|
sandbox_path = project.get("sandboxPath")
|
|
if not root_path or not sandbox_path:
|
|
return None
|
|
root_real = os.path.realpath(root_path)
|
|
sandbox_real = os.path.realpath(sandbox_path)
|
|
if sandbox_real != root_real and not sandbox_real.startswith(root_real + os.sep):
|
|
return None
|
|
return sandbox_real
|
|
|
|
|
|
def _get_workdir(session_id: str | None = None) -> str:
|
|
"""Return a per-session sandbox dir at mode 0o700."""
|
|
global _workdirs
|
|
key = session_id or "_default"
|
|
if key not in _workdirs or not os.path.isdir(_workdirs[key]):
|
|
home = os.path.expanduser("~")
|
|
sandbox_root = os.path.join(home, "studio_sandbox")
|
|
project_workdir = (
|
|
_get_project_workdir(session_id)
|
|
if session_id and _SESSION_ID_RE.match(session_id)
|
|
else None
|
|
)
|
|
if project_workdir:
|
|
workdir = project_workdir
|
|
elif session_id and _SESSION_ID_RE.match(session_id):
|
|
workdir = os.path.join(sandbox_root, session_id)
|
|
if not os.path.realpath(workdir).startswith(os.path.realpath(sandbox_root) + os.sep):
|
|
workdir = os.path.join(sandbox_root, "_invalid")
|
|
elif session_id:
|
|
workdir = os.path.join(sandbox_root, "_invalid")
|
|
else:
|
|
workdir = os.path.join(sandbox_root, "_default")
|
|
os.makedirs(workdir, exist_ok = True)
|
|
try:
|
|
os.chmod(sandbox_root, 0o700)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
os.chmod(workdir, 0o700)
|
|
except OSError:
|
|
pass
|
|
_workdirs[key] = workdir
|
|
return _workdirs[key]
|
|
|
|
|
|
def get_sandbox_workdir(session_id: str | None = None) -> str:
|
|
return _get_workdir(session_id)
|
|
|
|
|
|
WEB_SEARCH_TOOL = {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_search",
|
|
"description": (
|
|
"Search the web and fetch page content. Returns snippets for all results. "
|
|
"Use the url parameter to fetch full page text from a specific URL."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {
|
|
"type": "string",
|
|
"description": "The search query",
|
|
},
|
|
"url": {
|
|
"type": "string",
|
|
"description": "A URL to fetch full page content from (instead of searching). Use this to read a page found in search results.",
|
|
},
|
|
},
|
|
"required": [],
|
|
},
|
|
},
|
|
}
|
|
|
|
PYTHON_TOOL = {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "python",
|
|
"description": "Execute Python code in a sandbox and return stdout/stderr.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"code": {
|
|
"type": "string",
|
|
"description": "The Python code to run",
|
|
}
|
|
},
|
|
"required": ["code"],
|
|
},
|
|
},
|
|
}
|
|
|
|
TERMINAL_TOOL = {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "terminal",
|
|
"description": "Execute a terminal command and return stdout/stderr.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"command": {
|
|
"type": "string",
|
|
"description": "The command to run",
|
|
}
|
|
},
|
|
"required": ["command"],
|
|
},
|
|
},
|
|
}
|
|
|
|
RENDER_HTML_TOOL = {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"description": (
|
|
"Render a self-contained HTML/CSS/JavaScript canvas for the user. "
|
|
"Call this at most once per assistant response unless the user "
|
|
"explicitly asks for changes in that response. Future user requests "
|
|
"for new canvases may call render_html once. Put the entire document "
|
|
"in code, including any CSS in <style> tags and JavaScript in <script> tags."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"code": {
|
|
"type": "string",
|
|
"description": "A complete self-contained HTML document.",
|
|
},
|
|
"title": {
|
|
"type": "string",
|
|
"description": "Short display title for the canvas.",
|
|
},
|
|
},
|
|
"required": ["code"],
|
|
},
|
|
},
|
|
}
|
|
|
|
# Duplicated (not imported from core.rag.tool) so the registry never pulls in
|
|
# the RAG stack; dispatch imports it lazily.
|
|
SEARCH_KNOWLEDGE_BASE_TOOL = {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "search_knowledge_base",
|
|
"description": (
|
|
"Search the user's uploaded documents and knowledge bases for "
|
|
"relevant passages. Use this whenever the question may be answered "
|
|
"by the attached documents, then cite the returned chunks."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {
|
|
"type": "string",
|
|
"description": "Natural-language search query.",
|
|
},
|
|
"top_k": {
|
|
"type": "integer",
|
|
"description": "Max chunks to return.",
|
|
},
|
|
},
|
|
"required": ["query"],
|
|
},
|
|
},
|
|
}
|
|
|
|
ALL_TOOLS = [
|
|
WEB_SEARCH_TOOL,
|
|
PYTHON_TOOL,
|
|
TERMINAL_TOOL,
|
|
RENDER_HTML_TOOL,
|
|
SEARCH_KNOWLEDGE_BASE_TOOL,
|
|
]
|
|
|
|
|
|
# OpenAI's function.name regex ^[a-zA-Z0-9_-]{1,64}$, enforced before streaming.
|
|
# MCP tool names with '.', '/', spaces, etc. would 400 the whole request, so we
|
|
# validate up front and skip with a warning.
|
|
_OPENAI_FN_NAME_RE = re.compile(r"^[a-zA-Z0-9_-]{1,64}$")
|
|
|
|
|
|
def _mcp_specs_for_server(server: dict, mcp_tools: list[dict]) -> list[dict]:
|
|
"""Convert an MCP server's tool list into OpenAI function specs."""
|
|
display = server.get("display_name") or server["id"]
|
|
specs: list[dict] = []
|
|
seen_names: set[str] = set()
|
|
for tool in mcp_tools:
|
|
raw_name = tool.get("name") or ""
|
|
if not raw_name:
|
|
logger.warning("Skipping MCP tool on '%s': empty name.", display)
|
|
continue
|
|
name = f"{MCP_TOOL_PREFIX}{server['id']}__{raw_name}"
|
|
# Bad chars or oversized names would 400 the whole request; skip + warn
|
|
# so the rest of the tools still ship.
|
|
if not _OPENAI_FN_NAME_RE.fullmatch(name):
|
|
logger.warning(
|
|
"Skipping MCP tool '%s' on '%s': composed name '%s' is not "
|
|
"valid OpenAI function.name (regex ^[a-zA-Z0-9_-]{1,64}$).",
|
|
raw_name,
|
|
display,
|
|
name,
|
|
)
|
|
continue
|
|
# Duplicate tool names would also 400 OpenAI; drop dupes.
|
|
if name in seen_names:
|
|
logger.warning("Skipping duplicate MCP tool '%s' on '%s'.", raw_name, display)
|
|
continue
|
|
seen_names.add(name)
|
|
specs.append(
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": name,
|
|
"description": f"[{display}] {tool.get('description') or ''}".strip(),
|
|
"parameters": tool.get("inputSchema") or {"type": "object", "properties": {}},
|
|
},
|
|
}
|
|
)
|
|
return specs
|
|
|
|
|
|
async def get_enabled_mcp_tools() -> list[dict]:
|
|
servers = [s for s in mcp_servers_db.list_servers() if s.get("is_enabled")]
|
|
# Never spawn stdio servers when stdio is disabled on this host (e.g. a DB
|
|
# carried from a desktop install onto a Colab/network deployment).
|
|
if not stdio_mcp_enabled():
|
|
servers = [s for s in servers if not is_stdio(s["url"])]
|
|
if not servers:
|
|
return []
|
|
|
|
# Skip servers still in their post-failure cool-off, otherwise a down
|
|
# server gets re-probed -- and blocks the send for the full timeout -- on
|
|
# every message.
|
|
uncached = [
|
|
s for s in servers if get_cached_tools(s["id"]) is None and not in_failure_cooloff(s["id"])
|
|
]
|
|
if uncached:
|
|
results = await asyncio.gather(
|
|
*(
|
|
list_tools_async(
|
|
url = s["url"],
|
|
headers = parse_server_headers(s),
|
|
timeout = probe_timeout(s["url"], bool(s.get("use_oauth"))),
|
|
use_oauth = bool(s.get("use_oauth")),
|
|
)
|
|
for s in uncached
|
|
),
|
|
return_exceptions = True,
|
|
)
|
|
# An edit/delete can land while we await a probe (up to 305 s for
|
|
# OAuth); its cache eviction is a no-op against an entry we haven't
|
|
# written yet. Re-read and drop a result whose server changed or
|
|
# was removed mid-probe, else a stale tool list caches indefinitely.
|
|
current = {s["id"]: s for s in mcp_servers_db.list_servers()}
|
|
for server, payload in zip(uncached, results):
|
|
# Guard the failure branch too: a stale failure must not park a
|
|
# cool-off on the fresh config, or the server the user just fixed
|
|
# is skipped for the whole window.
|
|
fresh = current.get(server["id"])
|
|
if fresh is None or any(
|
|
fresh.get(k) != server.get(k) for k in TOOL_CACHE_INVALIDATING_FIELDS
|
|
):
|
|
continue
|
|
if isinstance(payload, BaseException):
|
|
logger.warning(
|
|
"MCP server '%s' (%s) discovery failed: %s",
|
|
server.get("display_name") or server["id"],
|
|
server.get("url"),
|
|
payload,
|
|
)
|
|
# Failures aren't cached, but record one so a down server
|
|
# isn't re-probed every send during the cool-off.
|
|
record_probe_failure(server["id"], bool(fresh.get("use_oauth")))
|
|
continue
|
|
cache_tools(server["id"], payload)
|
|
|
|
specs: list[dict] = []
|
|
for server in servers:
|
|
payload = get_cached_tools(server["id"])
|
|
if payload is None:
|
|
continue
|
|
specs.extend(_mcp_specs_for_server(server, payload))
|
|
return specs
|
|
|
|
|
|
_TIMEOUT_UNSET = object()
|
|
|
|
|
|
def _render_html_result(arguments: dict) -> str:
|
|
code = arguments.get("code")
|
|
if not isinstance(code, str) or not code.strip():
|
|
return "Error: render_html requires a non-empty code string."
|
|
title = arguments.get("title")
|
|
if isinstance(title, str) and title.strip():
|
|
safe_title = title.strip()[:120]
|
|
return (
|
|
f"Rendered HTML canvas: {safe_title}. Do not call render_html "
|
|
"again in this response unless the user asks for changes. For a later "
|
|
"user request for a new canvas, call render_html once."
|
|
)
|
|
return (
|
|
"Rendered HTML canvas. Do not call render_html again in this response "
|
|
"unless the user asks for changes. For a later user request for a new "
|
|
"canvas, call render_html once."
|
|
)
|
|
|
|
|
|
def execute_tool(
|
|
name: str,
|
|
arguments: dict,
|
|
cancel_event = None,
|
|
timeout: int | None = _TIMEOUT_UNSET,
|
|
session_id: str | None = None,
|
|
rag_scope: dict | None = None,
|
|
disable_sandbox: bool = False,
|
|
) -> str:
|
|
"""Execute a tool by name with the given arguments; returns a string.
|
|
|
|
``timeout``: int seconds, ``None`` = no limit, unset = ``_EXEC_TIMEOUT``.
|
|
``session_id``: optional ID for per-conversation sandbox isolation.
|
|
``rag_scope``: hidden per-request RAG context the model never sees; consumed
|
|
by ``search_knowledge_base``.
|
|
``disable_sandbox``: Bypass Permissions; run python/terminal without the
|
|
safety checks, blocklist, or resource caps (secrets still stripped). Only
|
|
affects local code tools; web_search / MCP are unchanged.
|
|
"""
|
|
logger.info(f"execute_tool: name={name}, session_id={session_id}, timeout={timeout}")
|
|
effective_timeout = _EXEC_TIMEOUT if timeout is _TIMEOUT_UNSET else timeout
|
|
if name == "search_knowledge_base":
|
|
return _search_knowledge_base(arguments, rag_scope)
|
|
if name == "render_html":
|
|
return _render_html_result(arguments)
|
|
if name.startswith(MCP_TOOL_PREFIX):
|
|
try:
|
|
_, server_id, tool_name = name.split("__", 2)
|
|
except ValueError:
|
|
return f"Error: malformed MCP tool name '{name}'"
|
|
server = mcp_servers_db.get_server(server_id)
|
|
if not server:
|
|
return f"Error: MCP server '{server_id}' not found"
|
|
if not server.get("is_enabled"):
|
|
return f"Error: MCP server '{server_id}' is disabled"
|
|
if is_stdio(server["url"]) and not stdio_mcp_enabled():
|
|
return f"Error: stdio MCP server '{server_id}' is disabled on this host"
|
|
return call_tool_sync(
|
|
url = server["url"],
|
|
headers = parse_server_headers(server),
|
|
name = tool_name,
|
|
args = arguments,
|
|
timeout = effective_timeout,
|
|
use_oauth = bool(server.get("use_oauth")),
|
|
cancel_event = cancel_event,
|
|
)
|
|
if name == "web_search":
|
|
return _web_search(
|
|
arguments.get("query", ""),
|
|
url = arguments.get("url"),
|
|
timeout = effective_timeout,
|
|
)
|
|
if name == "python":
|
|
return _python_exec(
|
|
arguments.get("code", ""),
|
|
cancel_event,
|
|
effective_timeout,
|
|
session_id,
|
|
disable_sandbox = disable_sandbox,
|
|
)
|
|
if name == "terminal":
|
|
return _bash_exec(
|
|
arguments.get("command", ""),
|
|
cancel_event,
|
|
effective_timeout,
|
|
session_id,
|
|
disable_sandbox = disable_sandbox,
|
|
)
|
|
return f"Unknown tool: {name}"
|
|
|
|
|
|
def _opt_int(v) -> int | None:
|
|
try:
|
|
return int(v) if v is not None else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _scope_retrieval_kwargs(scope: dict) -> dict:
|
|
"""Retrieval mode from rag_scope; candidate pools and RRF come from config."""
|
|
mode = scope.get("mode")
|
|
return {"mode": mode if mode in ("hybrid", "dense", "lexical") else "hybrid"}
|
|
|
|
|
|
def _search_knowledge_base(arguments: dict, rag_scope: dict | None) -> str:
|
|
"""Run the RAG search bound to the hidden per-request ``rag_scope`` (the model
|
|
supplies only ``query``/``top_k``). Lazy import; missing sqlite-vec degrades
|
|
to a friendly message."""
|
|
scope = rag_scope or {}
|
|
query = (arguments or {}).get("query", "")
|
|
if not query or not str(query).strip():
|
|
return "Error: query is empty."
|
|
try:
|
|
from storage import rag_db
|
|
if not rag_db.RAG_AVAILABLE:
|
|
return "Knowledge base search is unavailable on this server."
|
|
from core.rag.tool import search_knowledge_base_with_sources
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("RAG tool unavailable: %s", exc)
|
|
return "Knowledge base search is unavailable on this server."
|
|
|
|
top_k = _opt_int((arguments or {}).get("top_k") or scope.get("default_top_k"))
|
|
text, sources = search_knowledge_base_with_sources(
|
|
query = str(query),
|
|
scope_kb_id = scope.get("kb_id"),
|
|
scope_thread_id = scope.get("thread_id"),
|
|
scope_project_id = scope.get("project_id"),
|
|
top_k = top_k,
|
|
**_scope_retrieval_kwargs(scope),
|
|
)
|
|
# Append the UI source-map after the sentinel; loops strip it before the model.
|
|
if sources:
|
|
import json as _json
|
|
return text + RAG_SOURCES_SENTINEL + _json.dumps(sources, ensure_ascii = False)
|
|
return text
|
|
|
|
|
|
# Forced first-pass RAG retrieval: a high cosine floor keeps it precise (fires on
|
|
# on-topic queries, skips weak ones) and helps small models that under-call the tool.
|
|
# Tunable via RAG_AUTOINJECT_MIN_SCORE.
|
|
_AUTOINJECT_DEFAULT_FLOOR = 0.70
|
|
|
|
|
|
def _autoinject_enabled() -> bool:
|
|
return os.environ.get("RAG_AUTOINJECT", "1").strip().lower() not in (
|
|
"0",
|
|
"false",
|
|
"no",
|
|
"off",
|
|
)
|
|
|
|
|
|
def _autoinject_floor() -> float:
|
|
raw = os.environ.get("RAG_AUTOINJECT_MIN_SCORE")
|
|
if raw is not None:
|
|
try:
|
|
return float(raw)
|
|
except ValueError:
|
|
pass
|
|
return _AUTOINJECT_DEFAULT_FLOOR
|
|
|
|
|
|
# Lean: injecting the full top_k every turn prefills thousands of tokens.
|
|
_AUTOINJECT_DEFAULT_TOP_K = 4
|
|
|
|
|
|
def _autoinject_top_k() -> int:
|
|
raw = os.environ.get("RAG_AUTOINJECT_TOP_K")
|
|
if raw is not None:
|
|
try:
|
|
return max(1, int(raw))
|
|
except ValueError:
|
|
pass
|
|
return _AUTOINJECT_DEFAULT_TOP_K
|
|
|
|
|
|
def _thread_whole_doc_enabled(scope: dict) -> bool:
|
|
"""Whether a thread-attached file should be injected in full rather than
|
|
retrieved top-K. ``rag_scope.whole_doc=False`` disables it for this request."""
|
|
override = scope.get("whole_doc")
|
|
if override is False:
|
|
return False
|
|
try:
|
|
from core.rag import config as _rag_config
|
|
except Exception: # noqa: BLE001
|
|
return True
|
|
return _rag_config.THREAD_WHOLE_DOC
|
|
|
|
|
|
_IMAGE_PART_TOKEN_ESTIMATE = 1024
|
|
|
|
|
|
def _message_token_estimate(conversation: list[dict]) -> int:
|
|
"""Cheap prompt-size estimate for budget guards; exact tokenization happens later."""
|
|
total = 0
|
|
for msg in conversation:
|
|
content = msg.get("content")
|
|
if isinstance(content, str):
|
|
total += max(1, len(content) // 4)
|
|
elif isinstance(content, list):
|
|
for part in content:
|
|
if isinstance(part, dict):
|
|
if part.get("type") in ("image_url", "input_image"):
|
|
total += _IMAGE_PART_TOKEN_ESTIMATE
|
|
else:
|
|
total += max(1, len(str(part.get("text") or "")) // 4)
|
|
total += 4 # chat-template role / separator overhead estimate
|
|
return total
|
|
|
|
|
|
def _whole_doc_budget(scope: dict | None = None, conversation: list[dict] | None = None) -> int:
|
|
try:
|
|
from core.rag import config as _rag_config
|
|
except Exception: # noqa: BLE001
|
|
budget = 6000
|
|
else:
|
|
budget = _rag_config.WHOLE_DOC_MAX_TOKENS
|
|
if not scope:
|
|
return budget
|
|
context = _opt_int(scope.get("context_length") or scope.get("max_context_tokens"))
|
|
if context is None or context <= 0:
|
|
return budget
|
|
headroom = _opt_int(scope.get("response_headroom"))
|
|
if headroom is None:
|
|
headroom = max(1024, context // 4)
|
|
used = _message_token_estimate(conversation or [])
|
|
# Leave room for tool XML wrappers, citation metadata, and chat-template overhead.
|
|
available = context - headroom - used - 512
|
|
return min(budget, max(0, available))
|
|
|
|
|
|
def _last_user_text(conversation: list[dict]) -> str:
|
|
"""Plain text of the most recent user turn (text parts only)."""
|
|
for msg in reversed(conversation):
|
|
if msg.get("role") != "user":
|
|
continue
|
|
content = msg.get("content")
|
|
if isinstance(content, str):
|
|
return content.strip()
|
|
if isinstance(content, list):
|
|
parts = [
|
|
p.get("text", "")
|
|
for p in content
|
|
if isinstance(p, dict) and p.get("type") in ("text", "input_text")
|
|
]
|
|
return " ".join(t for t in parts if t).strip()
|
|
return ""
|
|
return ""
|
|
|
|
|
|
def build_rag_autoinject(conversation: list[dict], rag_scope: dict | None) -> dict | None:
|
|
"""Pre-retrieve the latest user turn; if a hit clears the cosine floor return
|
|
``{"events": [...], "messages": [...]}`` to splice into the loop, else ``None``.
|
|
Toggle via ``rag_scope.autoinject`` (else env ``RAG_AUTOINJECT``); floor via
|
|
``rag_scope.autoinject_min_score`` (else env ``RAG_AUTOINJECT_MIN_SCORE``).
|
|
|
|
Also the small-model fallback: models below ~4B often answer from memory
|
|
instead of calling ``search_knowledge_base``, so forcing retrieval here keeps
|
|
attachments consulted regardless of model size."""
|
|
if not rag_scope:
|
|
return None
|
|
enabled = rag_scope.get("autoinject")
|
|
if enabled is None:
|
|
enabled = _autoinject_enabled()
|
|
thread_id = rag_scope.get("thread_id")
|
|
whole_doc_requested = (
|
|
bool(thread_id) and not rag_scope.get("kb_id") and _thread_whole_doc_enabled(rag_scope)
|
|
)
|
|
if not enabled and not whole_doc_requested:
|
|
return None
|
|
query = _last_user_text(conversation)
|
|
if not query:
|
|
return None
|
|
try:
|
|
from storage import rag_db
|
|
if not rag_db.RAG_AVAILABLE:
|
|
return None
|
|
from core.rag.tool import render_sources, search_for_autoinject, whole_document_context
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("RAG auto-inject unavailable: %s", exc)
|
|
return None
|
|
|
|
text: str | None = None
|
|
sources: list[dict] = []
|
|
|
|
floor_override = rag_scope.get("autoinject_min_score")
|
|
floor = float(floor_override) if floor_override is not None else _autoinject_floor()
|
|
# Cap at the lean top_k, but honor a lower user setting.
|
|
lean_k = _autoinject_top_k()
|
|
sidebar_k = _opt_int(rag_scope.get("default_top_k"))
|
|
top_k = min(sidebar_k, lean_k) if sidebar_k is not None else lean_k
|
|
|
|
# Whole-document mode: a thread-attached file under budget is injected in full so
|
|
# the model reads everything. A KB selection is exclusive, so whole-doc never
|
|
# preempts it; in a project chat the project sources are still retrieved top-K and
|
|
# appended under one citation numbering. Oversized files (or no thread doc) fall
|
|
# through to the combined top-K retrieval below.
|
|
if whole_doc_requested:
|
|
try:
|
|
budget = _whole_doc_budget(rag_scope, conversation)
|
|
|
|
whole = whole_document_context(
|
|
scope_thread_id = thread_id,
|
|
max_tokens = budget,
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("RAG whole-document context failed: %s", exc)
|
|
whole = None
|
|
if whole is not None:
|
|
text, sources = whole
|
|
project_id = rag_scope.get("project_id")
|
|
if project_id:
|
|
try:
|
|
proj = search_for_autoinject(
|
|
query = query,
|
|
scope_project_id = project_id,
|
|
top_k = top_k,
|
|
min_dense_score = floor,
|
|
**_scope_retrieval_kwargs(rag_scope),
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("RAG project retrieval (whole-doc companion) failed: %s", exc)
|
|
proj = None
|
|
if proj is not None:
|
|
merged = sources + proj[1]
|
|
merged_text = render_sources(merged)
|
|
if max(1, len(merged_text) // 4) <= budget:
|
|
sources = merged
|
|
text = merged_text
|
|
logger.info("RAG auto-inject: whole-document context (%d chunk(s))", len(sources))
|
|
|
|
if text is None and enabled:
|
|
try:
|
|
found = search_for_autoinject(
|
|
query = query,
|
|
scope_kb_id = rag_scope.get("kb_id"),
|
|
scope_thread_id = rag_scope.get("thread_id"),
|
|
scope_project_id = rag_scope.get("project_id"),
|
|
top_k = top_k,
|
|
min_dense_score = floor,
|
|
**_scope_retrieval_kwargs(rag_scope),
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("RAG auto-inject retrieval failed: %s", exc)
|
|
return None
|
|
if not found:
|
|
logger.info("RAG auto-inject: no passage >= %.2f; skipping", floor)
|
|
return None
|
|
text, sources = found
|
|
if text is None:
|
|
return None
|
|
|
|
import json as _json
|
|
import uuid as _uuid
|
|
|
|
call_id = "rag_auto_" + _uuid.uuid4().hex[:12]
|
|
args = {"query": query}
|
|
full_result = text + RAG_SOURCES_SENTINEL + _json.dumps(sources, ensure_ascii = False)
|
|
events = [
|
|
{"type": "status", "text": f"Searching documents: {query[:60]}"},
|
|
{
|
|
"type": "tool_start",
|
|
"tool_name": "search_knowledge_base",
|
|
"tool_call_id": call_id,
|
|
"arguments": args,
|
|
},
|
|
{
|
|
"type": "tool_end",
|
|
"tool_name": "search_knowledge_base",
|
|
"tool_call_id": call_id,
|
|
"result": full_result,
|
|
},
|
|
{"type": "status", "text": ""},
|
|
]
|
|
messages = [
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [
|
|
{
|
|
"id": call_id,
|
|
"type": "function",
|
|
"function": {
|
|
"name": "search_knowledge_base",
|
|
"arguments": _json.dumps(args, ensure_ascii = False),
|
|
},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"role": "tool",
|
|
"name": "search_knowledge_base",
|
|
"tool_call_id": call_id,
|
|
"content": text,
|
|
},
|
|
]
|
|
logger.info("RAG auto-inject: %d passage(s) for %r", len(sources), query[:80])
|
|
return {"events": events, "messages": messages}
|
|
|
|
|
|
_MAX_PAGE_CHARS = 16000 # cap fetched page text (after HTML-to-MD conversion)
|
|
# Raw download cap > _MAX_PAGE_CHARS because SSR pages embed large <head>
|
|
# sections stripped during conversion; 512 KB reaches article content even
|
|
# where <head> alone is ~200 KB.
|
|
_MAX_FETCH_BYTES = 512 * 1024
|
|
|
|
_USER_AGENTS = (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
|
|
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:133.0) Gecko/20100101 Firefox/133.0",
|
|
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:133.0) Gecko/20100101 Firefox/133.0",
|
|
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.2 Safari/605.1.15",
|
|
)
|
|
|
|
_tls_ctx = ssl.create_default_context()
|
|
|
|
|
|
class _NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
return None
|
|
|
|
|
|
class _PinnedHTTPSConnection(http.client.HTTPSConnection):
|
|
"""HTTPS connection to a pinned IP, using a different hostname for SNI and
|
|
cert verification.
|
|
|
|
SSRF IP-pinning rewrites URLs to raw IPs; a normal HTTPSConnection would then
|
|
send no SNI and verify the cert against the IP (both fail). This splits the
|
|
concerns: TCP connects to the pinned IP (``host``), TLS uses ``sni_hostname``.
|
|
"""
|
|
|
|
def __init__(self, host: str, *, sni_hostname: str, **kwargs):
|
|
super().__init__(host, **kwargs)
|
|
self._sni_hostname = sni_hostname
|
|
|
|
def connect(self):
|
|
# TCP connect to the pinned IP in self.host.
|
|
http.client.HTTPConnection.connect(self)
|
|
# TLS handshake with the real hostname for SNI + cert verification.
|
|
self.sock = self._context.wrap_socket(
|
|
self.sock,
|
|
server_hostname = self._sni_hostname,
|
|
)
|
|
|
|
|
|
class _SNIHTTPSHandler(urllib.request.HTTPSHandler):
|
|
"""HTTPS handler sending the correct SNI hostname during TLS handshake.
|
|
|
|
SSRF IP-pinning breaks SNI and cert verification; this returns a
|
|
``_PinnedHTTPSConnection`` that connects to the pinned IP but verifies TLS
|
|
against the original hostname.
|
|
"""
|
|
|
|
def __init__(self, hostname: str):
|
|
super().__init__(context = _tls_ctx)
|
|
self._sni_hostname = hostname
|
|
|
|
def https_open(self, req):
|
|
return self.do_open(self._sni_connection, req)
|
|
|
|
def _sni_connection(self, host, **kwargs):
|
|
kwargs["context"] = _tls_ctx
|
|
return _PinnedHTTPSConnection(host, sni_hostname = self._sni_hostname, **kwargs)
|
|
|
|
|
|
def _validate_and_resolve_host(hostname: str, port: int) -> tuple[bool, str, str]:
|
|
"""Resolve *hostname*, reject non-public IPs, return a pinned IP string.
|
|
|
|
Returns ``(ok, reason_or_empty, resolved_ip)``. The caller should connect
|
|
to *resolved_ip* (with a ``Host`` header) to prevent DNS rebinding between
|
|
validation and the actual fetch.
|
|
"""
|
|
import ipaddress
|
|
import socket
|
|
|
|
try:
|
|
infos = socket.getaddrinfo(hostname, port, type = socket.SOCK_STREAM)
|
|
except OSError as e:
|
|
return False, f"Failed to resolve host: {e}", ""
|
|
|
|
if not infos:
|
|
return False, f"Failed to resolve host: no addresses for {hostname!r}", ""
|
|
|
|
for *_, sockaddr in infos:
|
|
ip = ipaddress.ip_address(sockaddr[0])
|
|
# `not ip.is_global` is the source of truth: it rejects every category
|
|
# below PLUS shared/CGNAT (100.64.0.0/10) and benchmarking/doc ranges
|
|
# Python marks is_private=False and is_global=False. The explicit
|
|
# predicates only give human-readable categories in the error message.
|
|
if (
|
|
not ip.is_global
|
|
or ip.is_private
|
|
or ip.is_loopback
|
|
or ip.is_link_local
|
|
or ip.is_multicast
|
|
or ip.is_reserved
|
|
or ip.is_unspecified
|
|
):
|
|
return False, f"Blocked: refusing to fetch non-public address {ip}.", ""
|
|
|
|
# Return the first resolved address for pinning.
|
|
first_ip = infos[0][4][0]
|
|
return True, "", first_ip
|
|
|
|
|
|
def _fetch_page_text(
|
|
url: str,
|
|
max_chars: int = _MAX_PAGE_CHARS,
|
|
timeout: int = 30,
|
|
) -> str:
|
|
"""Fetch a URL and return plain text content (HTML tags stripped).
|
|
|
|
Blocks private/loopback/link-local targets (SSRF protection) and caps
|
|
the download size to avoid unbounded memory usage.
|
|
"""
|
|
from urllib.parse import urlparse
|
|
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in ("http", "https"):
|
|
return f"Blocked: only http/https URLs are allowed (got {parsed.scheme!r})."
|
|
if not parsed.hostname:
|
|
return "Blocked: URL is missing a hostname."
|
|
|
|
port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
|
ok, reason, pinned_ip = _validate_and_resolve_host(parsed.hostname, port)
|
|
if not ok:
|
|
return reason
|
|
|
|
try:
|
|
from urllib.error import HTTPError as _HTTPError
|
|
from urllib.parse import urljoin, urlunparse
|
|
|
|
max_bytes = _MAX_FETCH_BYTES
|
|
current_url = url
|
|
current_host = parsed.hostname
|
|
ua = random.choice(_USER_AGENTS)
|
|
|
|
for _hop in range(5):
|
|
# Pin to the validated IP (prevents DNS rebinding): rewrite URL to
|
|
# the IP, set the Host header.
|
|
cp = urlparse(current_url)
|
|
# Bracket IPv6 addresses so the netloc is valid in a URL.
|
|
ip_str = f"[{pinned_ip}]" if ":" in pinned_ip else pinned_ip
|
|
ip_netloc = f"{ip_str}:{cp.port}" if cp.port else ip_str
|
|
pinned_url = urlunparse(cp._replace(netloc = ip_netloc))
|
|
|
|
opener = urllib.request.build_opener(
|
|
_NoRedirect,
|
|
_SNIHTTPSHandler(current_host),
|
|
)
|
|
|
|
req = urllib.request.Request(
|
|
pinned_url,
|
|
headers = {
|
|
"User-Agent": ua,
|
|
"Host": current_host,
|
|
},
|
|
)
|
|
try:
|
|
resp = opener.open(req, timeout = timeout)
|
|
except _HTTPError as e:
|
|
if e.code not in (301, 302, 303, 307, 308):
|
|
return f"Failed to fetch URL: HTTP {e.code} {getattr(e, 'reason', '')}"
|
|
location = e.headers.get("Location")
|
|
if not location:
|
|
return "Failed to fetch URL: redirect missing Location header."
|
|
current_url = urljoin(current_url, location)
|
|
rp = urlparse(current_url)
|
|
if rp.scheme not in ("http", "https") or not rp.hostname:
|
|
return "Blocked: redirect target is not a valid http/https URL."
|
|
rp_port = rp.port or (443 if rp.scheme == "https" else 80)
|
|
ok2, reason2, pinned_ip = _validate_and_resolve_host(
|
|
rp.hostname,
|
|
rp_port,
|
|
)
|
|
if not ok2:
|
|
return reason2
|
|
current_host = rp.hostname
|
|
continue
|
|
# Success: read capped body.
|
|
raw_bytes = resp.read(max_bytes)
|
|
break
|
|
else:
|
|
return "Failed to fetch URL: too many redirects."
|
|
|
|
charset = resp.headers.get_content_charset() or "utf-8"
|
|
raw_html = raw_bytes.decode(charset, errors = "replace")
|
|
except _HTTPError as e:
|
|
return f"Failed to fetch URL: HTTP {e.code} {getattr(e, 'reason', '')}"
|
|
except Exception as e:
|
|
return f"Failed to fetch URL: {e}"
|
|
|
|
# Convert HTML to Markdown with the builtin converter (no external deps).
|
|
from ._html_to_md import html_to_markdown
|
|
|
|
text = html_to_markdown(raw_html)
|
|
|
|
if not text:
|
|
return "(page returned no readable text)"
|
|
if len(text) > max_chars:
|
|
text = text[:max_chars] + f"\n\n... (truncated, {len(text)} chars total)"
|
|
return text
|
|
|
|
|
|
def _web_search(
|
|
query: str,
|
|
max_results: int = 5,
|
|
timeout: int = _EXEC_TIMEOUT,
|
|
url: str | None = None,
|
|
) -> str:
|
|
"""Search the web using DuckDuckGo and return formatted results.
|
|
|
|
If ``url`` is provided, fetches that page directly instead of searching.
|
|
"""
|
|
# Direct URL fetch mode.
|
|
if url and url.strip():
|
|
fetch_timeout = 60 if timeout is None else min(timeout, 60)
|
|
return _fetch_page_text(url.strip(), timeout = fetch_timeout)
|
|
|
|
if not query or not query.strip():
|
|
return "No query provided."
|
|
try:
|
|
from ddgs import DDGS
|
|
|
|
results = DDGS(timeout = timeout).text(query, max_results = max_results)
|
|
if not results:
|
|
return "No results found."
|
|
parts = []
|
|
for r in results:
|
|
parts.append(
|
|
f"Title: {r.get('title', '')}\n"
|
|
f"URL: {r.get('href', '')}\n"
|
|
f"Snippet: {r.get('body', '')}"
|
|
)
|
|
text = "\n\n---\n\n".join(parts)
|
|
text += (
|
|
"\n\n---\n\nIMPORTANT: These are only short snippets. "
|
|
"To get the full page content, call web_search with "
|
|
'the url parameter (e.g. {"url": "<URL>"}).'
|
|
)
|
|
return text
|
|
except Exception as e:
|
|
return f"Search failed: {e}"
|
|
|
|
|
|
def _check_signal_escape_patterns(code: str):
|
|
"""Check for patterns that could escape signal-based timeouts. Returns
|
|
(safe: bool, details: dict). Vendored from unsloth_zoo.rl_environments to
|
|
avoid importing unsloth_zoo (needs GPU drivers; fails on Apple Silicon)."""
|
|
try:
|
|
tree = ast.parse(code)
|
|
except SyntaxError as e:
|
|
return False, {
|
|
"error": f"SyntaxError: {e}",
|
|
"signal_tampering": [],
|
|
"exception_catching": [],
|
|
"warnings": [],
|
|
}
|
|
|
|
signal_tampering = []
|
|
exception_catching = []
|
|
shell_escapes = []
|
|
warnings = []
|
|
|
|
def _ast_name_matches(node, names):
|
|
if isinstance(node, ast.Name):
|
|
return node.id in names
|
|
elif isinstance(node, ast.Attribute):
|
|
full_name = []
|
|
current = node
|
|
while isinstance(current, ast.Attribute):
|
|
full_name.append(current.attr)
|
|
current = current.value
|
|
if isinstance(current, ast.Name):
|
|
full_name.append(current.id)
|
|
full_name = ".".join(reversed(full_name))
|
|
return full_name in names
|
|
return False
|
|
|
|
# Dangerous os/subprocess functions that can execute shell commands.
|
|
_SHELL_EXEC_FUNCS = frozenset(
|
|
{
|
|
"os.system",
|
|
"os.popen",
|
|
"os.popen2",
|
|
"os.popen3",
|
|
"os.popen4",
|
|
"os.execl",
|
|
"os.execle",
|
|
"os.execlp",
|
|
"os.execlpe",
|
|
"os.execv",
|
|
"os.execve",
|
|
"os.execvp",
|
|
"os.execvpe",
|
|
"os.spawnl",
|
|
"os.spawnle",
|
|
"os.spawnlp",
|
|
"os.spawnlpe",
|
|
"os.spawnv",
|
|
"os.spawnve",
|
|
"os.spawnvp",
|
|
"os.spawnvpe",
|
|
"os.posix_spawn",
|
|
"os.posix_spawnp",
|
|
"subprocess.run",
|
|
"subprocess.call",
|
|
"subprocess.check_call",
|
|
"subprocess.check_output",
|
|
"subprocess.Popen",
|
|
"subprocess.getoutput",
|
|
"subprocess.getstatusoutput",
|
|
}
|
|
)
|
|
|
|
def _extract_string_from_node(node):
|
|
"""Extract a plain string value from an AST node, if it is a constant."""
|
|
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
|
return node.value
|
|
return None
|
|
|
|
def _extract_strings_from_list(node):
|
|
"""Extract string elements from an AST List or Tuple node."""
|
|
if isinstance(node, (ast.List, ast.Tuple)):
|
|
parts = []
|
|
for elt in node.elts:
|
|
s = _extract_string_from_node(elt)
|
|
if s is not None:
|
|
parts.append(s)
|
|
return parts
|
|
return []
|
|
|
|
# Kwarg names that carry command content (not control flags like
|
|
# check=True, text=True, capture_output=True).
|
|
_CMD_KWARGS = frozenset({"args", "command", "executable", "path", "file"})
|
|
|
|
def _check_args_for_blocked(args_nodes):
|
|
"""Check if any call arguments contain blocked commands."""
|
|
found = set()
|
|
for arg in args_nodes:
|
|
s = _extract_string_from_node(arg)
|
|
if s is not None:
|
|
found |= _find_blocked_commands(s)
|
|
strs = _extract_strings_from_list(arg)
|
|
for s in strs:
|
|
found |= _find_blocked_commands(s)
|
|
return found
|
|
|
|
class SignalEscapeVisitor(ast.NodeVisitor):
|
|
def __init__(self):
|
|
self.imports_signal = False
|
|
self.signal_aliases = {"signal"}
|
|
self.os_aliases = {"os"}
|
|
self.subprocess_aliases = {"subprocess"}
|
|
# Bare name -> fully-qualified form for from-import tracking
|
|
# (e.g. "system" -> "os.system").
|
|
self.shell_exec_aliases: dict[str, str] = {}
|
|
self.loop_depth = 0
|
|
|
|
def visit_Import(self, node):
|
|
for alias in node.names:
|
|
if alias.name == "signal":
|
|
self.imports_signal = True
|
|
if alias.asname:
|
|
self.signal_aliases.add(alias.asname)
|
|
elif alias.name == "os":
|
|
self.os_aliases.add(alias.asname or "os")
|
|
elif alias.name == "subprocess":
|
|
self.subprocess_aliases.add(alias.asname or "subprocess")
|
|
self.generic_visit(node)
|
|
|
|
def visit_ImportFrom(self, node):
|
|
if node.module == "signal":
|
|
self.imports_signal = True
|
|
for alias in node.names:
|
|
if alias.name in (
|
|
"signal",
|
|
"SIGALRM",
|
|
"SIG_IGN",
|
|
"setitimer",
|
|
"ITIMER_REAL",
|
|
"pthread_sigmask",
|
|
"SIG_BLOCK",
|
|
"alarm",
|
|
):
|
|
self.signal_aliases.add(alias.asname or alias.name)
|
|
elif node.module in ("os", "subprocess"):
|
|
if node.module == "os":
|
|
self.os_aliases.add("os")
|
|
else:
|
|
self.subprocess_aliases.add("subprocess")
|
|
# Track from-imports of dangerous functions.
|
|
for alias in node.names:
|
|
fq = f"{node.module}.{alias.name}"
|
|
if fq in _SHELL_EXEC_FUNCS:
|
|
self.shell_exec_aliases[alias.asname or alias.name] = fq
|
|
self.generic_visit(node)
|
|
|
|
def visit_While(self, node):
|
|
self.loop_depth += 1
|
|
self.generic_visit(node)
|
|
self.loop_depth -= 1
|
|
|
|
def visit_For(self, node):
|
|
self.loop_depth += 1
|
|
self.generic_visit(node)
|
|
self.loop_depth -= 1
|
|
|
|
def visit_Call(self, node):
|
|
func = node.func
|
|
func_name = None
|
|
if isinstance(func, ast.Attribute):
|
|
if isinstance(func.value, ast.Name):
|
|
if func.value.id in self.signal_aliases:
|
|
func_name = f"signal.{func.attr}"
|
|
elif isinstance(func, ast.Name):
|
|
if func.id in ("signal", "setitimer", "alarm", "pthread_sigmask"):
|
|
func_name = func.id
|
|
|
|
if func_name:
|
|
if func_name in ("signal.signal", "signal"):
|
|
if len(node.args) >= 1:
|
|
if _ast_name_matches(node.args[0], ("SIGALRM", "signal.SIGALRM")):
|
|
signal_tampering.append(
|
|
{
|
|
"type": "signal_handler_override",
|
|
"line": node.lineno,
|
|
"description": "Overrides SIGALRM handler",
|
|
}
|
|
)
|
|
elif func_name in ("signal.setitimer", "setitimer"):
|
|
if len(node.args) >= 1:
|
|
if _ast_name_matches(node.args[0], ("ITIMER_REAL", "signal.ITIMER_REAL")):
|
|
signal_tampering.append(
|
|
{
|
|
"type": "timer_manipulation",
|
|
"line": node.lineno,
|
|
"description": "Manipulates ITIMER_REAL timer",
|
|
}
|
|
)
|
|
elif func_name in ("signal.alarm", "alarm"):
|
|
signal_tampering.append(
|
|
{
|
|
"type": "alarm_manipulation",
|
|
"line": node.lineno,
|
|
"description": "Manipulates alarm timer",
|
|
}
|
|
)
|
|
elif func_name in ("signal.pthread_sigmask", "pthread_sigmask"):
|
|
signal_tampering.append(
|
|
{
|
|
"type": "signal_mask",
|
|
"line": node.lineno,
|
|
"description": "Modifies signal mask (may block SIGALRM)",
|
|
}
|
|
)
|
|
|
|
# --- Shell escape detection ---
|
|
# Resolve the FQ function name for os.*/subprocess.*
|
|
shell_func = None
|
|
if isinstance(func, ast.Attribute):
|
|
if isinstance(func.value, ast.Name):
|
|
if func.value.id in self.os_aliases:
|
|
shell_func = f"os.{func.attr}"
|
|
elif func.value.id in self.subprocess_aliases:
|
|
shell_func = f"subprocess.{func.attr}"
|
|
elif isinstance(func, ast.Name):
|
|
# from-import aliases: from os import system; system(...)
|
|
shell_func = self.shell_exec_aliases.get(func.id)
|
|
|
|
if shell_func and shell_func in _SHELL_EXEC_FUNCS:
|
|
# Expand **kwargs dicts to inspect their keys.
|
|
expanded_kwargs: dict[str, ast.AST] = {}
|
|
has_opaque_kwargs = False
|
|
for kw in node.keywords:
|
|
if kw.arg is not None:
|
|
expanded_kwargs[kw.arg] = kw.value
|
|
elif isinstance(kw.value, ast.Dict):
|
|
for k, v in zip(kw.value.keys, kw.value.values):
|
|
key = _extract_string_from_node(k) if k else None
|
|
if key is not None:
|
|
expanded_kwargs[key] = v
|
|
else:
|
|
has_opaque_kwargs = True
|
|
|
|
cmd_kw_values = [v for k, v in expanded_kwargs.items() if k in _CMD_KWARGS]
|
|
all_call_args = list(node.args) + cmd_kw_values
|
|
blocked_in_args = _check_args_for_blocked(all_call_args)
|
|
|
|
if has_opaque_kwargs:
|
|
# Can't inspect dynamic **kwargs; flag as unsafe.
|
|
shell_escapes.append(
|
|
{
|
|
"type": "shell_escape_dynamic",
|
|
"line": node.lineno,
|
|
"description": (f"{shell_func}() called with dynamic **kwargs"),
|
|
}
|
|
)
|
|
elif blocked_in_args:
|
|
shell_escapes.append(
|
|
{
|
|
"type": "shell_escape",
|
|
"line": node.lineno,
|
|
"description": (
|
|
f"{shell_func}() invokes blocked command(s): "
|
|
f"{', '.join(sorted(blocked_in_args))}"
|
|
),
|
|
}
|
|
)
|
|
else:
|
|
# Only flag dynamic args for funcs that interpret strings as
|
|
# shell commands, or when shell= might be on. Any non-literal-
|
|
# False shell= is treated as potentially True (conservative).
|
|
_STRING_SHELL_FUNCS = frozenset(
|
|
{
|
|
"os.system",
|
|
"os.popen",
|
|
"os.popen2",
|
|
"os.popen3",
|
|
"os.popen4",
|
|
"subprocess.getoutput",
|
|
"subprocess.getstatusoutput",
|
|
}
|
|
)
|
|
shell_node = expanded_kwargs.get("shell")
|
|
shell_safe = shell_node is None or (
|
|
isinstance(shell_node, ast.Constant) and shell_node.value is False
|
|
)
|
|
# Dynamic shell-exec args (chr/format/concat bypasses).
|
|
if (
|
|
shell_func in _STRING_SHELL_FUNCS
|
|
or shell_func in _SHELL_EXEC_FUNCS
|
|
or not shell_safe
|
|
):
|
|
|
|
def _is_safe_literal(n):
|
|
if _extract_string_from_node(n) is not None:
|
|
return True
|
|
if isinstance(n, (ast.List, ast.Tuple)):
|
|
return all(_extract_string_from_node(e) is not None for e in n.elts)
|
|
return False
|
|
|
|
has_non_literal = any(not _is_safe_literal(a) for a in all_call_args)
|
|
if has_non_literal:
|
|
shell_escapes.append(
|
|
{
|
|
"type": "shell_escape_dynamic",
|
|
"line": node.lineno,
|
|
"description": (
|
|
f"{shell_func}() called with non-literal "
|
|
f"shell command (potential shell escape)"
|
|
),
|
|
}
|
|
)
|
|
|
|
self.generic_visit(node)
|
|
|
|
def visit_ExceptHandler(self, node):
|
|
if self.loop_depth == 0:
|
|
self.generic_visit(node)
|
|
return
|
|
if node.type is None:
|
|
exception_catching.append(
|
|
{
|
|
"type": "bare_except_in_loop",
|
|
"line": node.lineno,
|
|
"description": "Bare except in loop catches TimeoutError and continues looping",
|
|
}
|
|
)
|
|
elif isinstance(node.type, ast.Name):
|
|
# Flag BaseException/TimeoutError but NOT Exception: `except
|
|
# Exception` can't catch SystemExit/KeyboardInterrupt, so it
|
|
# can't suppress timeout enforcement.
|
|
if node.type.id in ("TimeoutError", "BaseException"):
|
|
exception_catching.append(
|
|
{
|
|
"type": f"catches_{node.type.id}_in_loop",
|
|
"line": node.lineno,
|
|
"description": f"Catches {node.type.id} in loop - may suppress timeout and continue",
|
|
}
|
|
)
|
|
elif isinstance(node.type, ast.Tuple):
|
|
for elt in node.type.elts:
|
|
if isinstance(elt, ast.Name):
|
|
if elt.id in ("TimeoutError", "BaseException"):
|
|
exception_catching.append(
|
|
{
|
|
"type": f"catches_{elt.id}_in_loop",
|
|
"line": node.lineno,
|
|
"description": f"Catches {elt.id} in loop - may suppress timeout and continue",
|
|
}
|
|
)
|
|
self.generic_visit(node)
|
|
|
|
visitor = SignalEscapeVisitor()
|
|
visitor.visit(tree)
|
|
|
|
if visitor.imports_signal and not signal_tampering:
|
|
warnings.append("Code imports 'signal' module - review manually for safety")
|
|
|
|
# Static host policy: block metadata hosts and any literal host outside the
|
|
# trusted allowlist; uploads blocked regardless of host. Dynamic hosts are
|
|
# caught by the bash blocklist.
|
|
network_calls: list[dict] = []
|
|
sensitive_file_reads: list[dict] = []
|
|
_NETWORK_FQ_PREFIXES = (
|
|
"socket.socket",
|
|
"socket.create_connection",
|
|
"socket.getaddrinfo",
|
|
"urllib.request.urlopen",
|
|
"urllib.request.urlretrieve",
|
|
"urllib3.",
|
|
"requests.get",
|
|
"requests.post",
|
|
"requests.put",
|
|
"requests.delete",
|
|
"requests.patch",
|
|
"requests.head",
|
|
"requests.request",
|
|
"requests.Session",
|
|
"http.client.HTTPConnection",
|
|
"http.client.HTTPSConnection",
|
|
"httpx.get",
|
|
"httpx.post",
|
|
"httpx.put",
|
|
"httpx.patch",
|
|
"httpx.delete",
|
|
"httpx.request",
|
|
"httpx.Client",
|
|
"httpx.AsyncClient",
|
|
"aiohttp.ClientSession",
|
|
)
|
|
_UPLOAD_HTTP_METHODS = (
|
|
"requests.post",
|
|
"requests.put",
|
|
"requests.patch",
|
|
"requests.delete",
|
|
"requests.request",
|
|
"httpx.post",
|
|
"httpx.put",
|
|
"httpx.patch",
|
|
"httpx.delete",
|
|
"httpx.request",
|
|
"urllib.request.urlopen",
|
|
"urllib.request.Request",
|
|
)
|
|
_UPLOAD_HF_FQ = (
|
|
"huggingface_hub.upload_file",
|
|
"huggingface_hub.upload_folder",
|
|
"huggingface_hub.upload_large_folder",
|
|
"huggingface_hub.create_commit",
|
|
)
|
|
_UPLOAD_HF_METHODS = frozenset(
|
|
{
|
|
"upload_file",
|
|
"upload_folder",
|
|
"upload_large_folder",
|
|
"create_commit",
|
|
}
|
|
)
|
|
# Cloud-metadata / link-local hosts.
|
|
_METADATA_HOST_LITERALS = {
|
|
"169.254.169.254",
|
|
"fd00:ec2::254",
|
|
"metadata.google.internal",
|
|
"metadata",
|
|
"metadata.tencentyun.com",
|
|
"100.100.100.200",
|
|
"100.100.100.110",
|
|
"169.254.170.2",
|
|
"169.254.170.23",
|
|
}
|
|
_METADATA_HOST_PREFIXES = (
|
|
"169.254.",
|
|
"100.64.",
|
|
)
|
|
# Allowlist kept explicit so each entry is auditable.
|
|
_TRUSTED_PUBLIC_HOST_LITERALS = frozenset(
|
|
{
|
|
# search
|
|
"www.google.com",
|
|
"google.com",
|
|
"www.bing.com",
|
|
"bing.com",
|
|
"duckduckgo.com",
|
|
"html.duckduckgo.com",
|
|
# encyclopedic / reference
|
|
"wikipedia.org",
|
|
"www.wikipedia.org",
|
|
"wikimedia.org",
|
|
"www.wikimedia.org",
|
|
"wikidata.org",
|
|
"www.wikidata.org",
|
|
"commons.wikimedia.org",
|
|
"www.britannica.com",
|
|
"openlibrary.org",
|
|
"www.openstreetmap.org",
|
|
# ML / dev / data
|
|
"huggingface.co",
|
|
"hf.co",
|
|
"github.com",
|
|
"api.github.com",
|
|
"raw.githubusercontent.com",
|
|
"gist.github.com",
|
|
"docs.github.com",
|
|
"pypi.org",
|
|
"files.pythonhosted.org",
|
|
"www.npmjs.com",
|
|
"registry.npmjs.org",
|
|
"crates.io",
|
|
"static.crates.io",
|
|
# docs
|
|
"docs.python.org",
|
|
"python.org",
|
|
"www.python.org",
|
|
"developer.mozilla.org",
|
|
"developer.apple.com",
|
|
"learn.microsoft.com",
|
|
"docs.docker.com",
|
|
"pytorch.org",
|
|
"docs.pytorch.org",
|
|
"tensorflow.org",
|
|
"www.tensorflow.org",
|
|
"numpy.org",
|
|
"pandas.pydata.org",
|
|
"scipy.org",
|
|
"scikit-learn.org",
|
|
"matplotlib.org",
|
|
"fastapi.tiangolo.com",
|
|
"starlette.io",
|
|
# academic
|
|
"arxiv.org",
|
|
"export.arxiv.org",
|
|
"scholar.google.com",
|
|
"openreview.net",
|
|
"semanticscholar.org",
|
|
"www.semanticscholar.org",
|
|
"biorxiv.org",
|
|
"www.biorxiv.org",
|
|
"medrxiv.org",
|
|
"www.medrxiv.org",
|
|
"pubmed.ncbi.nlm.nih.gov",
|
|
"www.ncbi.nlm.nih.gov",
|
|
# Q&A / community
|
|
"stackoverflow.com",
|
|
"stackexchange.com",
|
|
"askubuntu.com",
|
|
"superuser.com",
|
|
"serverfault.com",
|
|
# standards
|
|
"www.w3.org",
|
|
"tools.ietf.org",
|
|
"datatracker.ietf.org",
|
|
"www.rfc-editor.org",
|
|
# reputable news
|
|
"www.bbc.com",
|
|
"www.bbc.co.uk",
|
|
"www.reuters.com",
|
|
"apnews.com",
|
|
"www.nature.com",
|
|
"www.science.org",
|
|
# government / open data
|
|
"data.gov",
|
|
"catalog.data.gov",
|
|
"www.census.gov",
|
|
"www.nasa.gov",
|
|
"data.nasa.gov",
|
|
"www.cdc.gov",
|
|
"www.nih.gov",
|
|
"www.who.int",
|
|
# weather / time
|
|
"api.weather.gov",
|
|
"worldtimeapi.org",
|
|
}
|
|
)
|
|
_TRUSTED_PUBLIC_HOST_SUFFIXES = (
|
|
".wikipedia.org",
|
|
".wikimedia.org",
|
|
".wiktionary.org",
|
|
".wikibooks.org",
|
|
".wikiquote.org",
|
|
".wikisource.org",
|
|
".wikiversity.org",
|
|
".wikivoyage.org",
|
|
".stackexchange.com",
|
|
".hf.co",
|
|
".huggingface.co",
|
|
".githubusercontent.com",
|
|
".github.io",
|
|
".arxiv.org",
|
|
".readthedocs.io",
|
|
".readthedocs.org",
|
|
)
|
|
_SENSITIVE_FILE_PREFIXES = (
|
|
"/etc/passwd",
|
|
"/etc/shadow",
|
|
"/etc/sudoers",
|
|
"/etc/ssh/",
|
|
)
|
|
_SENSITIVE_FILE_RE = re.compile(r"^/proc/(?:self|\d+)/(?:environ|cmdline|task/\d+/environ)$")
|
|
|
|
def _normalize_host(host: str) -> str:
|
|
if not host:
|
|
return ""
|
|
h = host.strip().lower().rstrip(".")
|
|
if "@" in h:
|
|
h = h.split("@", 1)[1]
|
|
if h.startswith("[") and "]" in h:
|
|
h = h[1 : h.index("]")]
|
|
elif h.count(":") == 1:
|
|
h = h.split(":", 1)[0]
|
|
return h
|
|
|
|
def _is_metadata_host(host: str) -> bool:
|
|
h = _normalize_host(host)
|
|
if not h:
|
|
return False
|
|
if h in _METADATA_HOST_LITERALS:
|
|
return True
|
|
if any(h.startswith(p) for p in _METADATA_HOST_PREFIXES):
|
|
return True
|
|
return False
|
|
|
|
def _is_trusted_host(host: str) -> bool:
|
|
h = _normalize_host(host)
|
|
if not h:
|
|
return False
|
|
if h in _TRUSTED_PUBLIC_HOST_LITERALS:
|
|
return True
|
|
return any(h.endswith(s) for s in _TRUSTED_PUBLIC_HOST_SUFFIXES)
|
|
|
|
def _call_is_upload_shape(node: ast.Call, fq: str) -> bool:
|
|
"""True for statically obvious upload shapes (files=, data=open(), bytes literal)."""
|
|
if fq in _UPLOAD_HF_FQ:
|
|
return True
|
|
if fq not in _UPLOAD_HTTP_METHODS:
|
|
return False
|
|
for kw in node.keywords or []:
|
|
if kw.arg == "files":
|
|
return True
|
|
if kw.arg == "data":
|
|
v = kw.value
|
|
if isinstance(v, ast.Call) and isinstance(v.func, ast.Name) and v.func.id == "open":
|
|
return True
|
|
if isinstance(v, ast.Constant) and isinstance(v.value, (bytes, bytearray)):
|
|
return True
|
|
return False
|
|
|
|
# Bare method-name fallback (`x.upload_file(...)`) is fuzzy, so it fires only
|
|
# when huggingface_hub/hf_api is imported; else paramiko.upload_file,
|
|
# boto3.create_commit, etc. would false-positive. Pre-scan for the imports.
|
|
_HF_IMPORT_MODULES = (
|
|
"huggingface_hub",
|
|
"hf_api",
|
|
"huggingface_hub.hf_api",
|
|
)
|
|
|
|
def _module_has_hf_import(tree: ast.AST) -> bool:
|
|
for n in ast.walk(tree):
|
|
if isinstance(n, ast.Import):
|
|
for alias in n.names:
|
|
if alias.name.split(".", 1)[0] in _HF_IMPORT_MODULES:
|
|
return True
|
|
elif isinstance(n, ast.ImportFrom):
|
|
root = (n.module or "").split(".", 1)[0]
|
|
if root in _HF_IMPORT_MODULES:
|
|
return True
|
|
elif isinstance(n, ast.Call) and n.args:
|
|
# __import__('huggingface_hub'), importlib.import_module(...),
|
|
# and bare import_module(...) (via `from importlib import ...`).
|
|
arg0 = n.args[0]
|
|
if not (isinstance(arg0, ast.Constant) and isinstance(arg0.value, str)):
|
|
continue
|
|
if arg0.value.split(".", 1)[0] not in _HF_IMPORT_MODULES:
|
|
continue
|
|
func = n.func
|
|
if isinstance(func, ast.Name) and func.id in {
|
|
"__import__",
|
|
"import_module",
|
|
}:
|
|
return True
|
|
if isinstance(func, ast.Attribute) and func.attr == "import_module":
|
|
return True
|
|
return False
|
|
|
|
_hf_in_scope = _module_has_hf_import(tree)
|
|
|
|
def _method_call_hf_upload_name(node: ast.Call) -> str | None:
|
|
"""Return the HF upload method name (`upload_file`, ...) or None. Covers
|
|
the Attribute and bare-Name forms; the bare-name branch fires only when
|
|
an HF import is in scope so paramiko/boto3 don't false-positive."""
|
|
if not _hf_in_scope:
|
|
return None
|
|
f = node.func
|
|
if isinstance(f, ast.Attribute) and f.attr in _UPLOAD_HF_METHODS:
|
|
return f.attr
|
|
if isinstance(f, ast.Name) and f.id in _UPLOAD_HF_METHODS:
|
|
return f.id
|
|
return None
|
|
|
|
# Kwargs that ship a credential over the wire. The sandbox env strips
|
|
# credentials up front, so any value here is hard-coded or lifted from parent.
|
|
_HF_SENSITIVE_KWARGS = frozenset(
|
|
{
|
|
"token",
|
|
"hf_token",
|
|
"api_token",
|
|
"api_key",
|
|
"auth_token",
|
|
"access_token",
|
|
"password",
|
|
"secret",
|
|
}
|
|
)
|
|
|
|
def _is_os_environ(node: ast.AST) -> bool:
|
|
return (
|
|
isinstance(node, ast.Attribute)
|
|
and node.attr == "environ"
|
|
and isinstance(node.value, ast.Name)
|
|
and node.value.id == "os"
|
|
)
|
|
|
|
def _reads_env_or_secret(node: ast.AST | None) -> bool:
|
|
"""True if any node in the subtree resolves to an env/process read.
|
|
|
|
Walks the whole subtree (not just the root) to catch wrappers like
|
|
`str(os.environ)`. Covers os.environ[/.get]/os.getenv, bare getenv, and
|
|
subprocess.{run,check_output,...} that could lift parent env via printenv.
|
|
"""
|
|
if node is None:
|
|
return False
|
|
for sub in ast.walk(node):
|
|
if _is_os_environ(sub):
|
|
return True
|
|
if isinstance(sub, ast.Call):
|
|
f = sub.func
|
|
if isinstance(f, ast.Attribute):
|
|
if (
|
|
f.attr in {"getenv", "getenvb"}
|
|
and isinstance(f.value, ast.Name)
|
|
and f.value.id == "os"
|
|
):
|
|
return True
|
|
if (
|
|
f.attr
|
|
in {
|
|
"check_output",
|
|
"run",
|
|
"Popen",
|
|
"getoutput",
|
|
"getstatusoutput",
|
|
}
|
|
and isinstance(f.value, ast.Name)
|
|
and f.value.id in {"subprocess", "commands"}
|
|
):
|
|
return True
|
|
if isinstance(f, ast.Name) and f.id in {"getenv", "getenvb"}:
|
|
return True
|
|
return False
|
|
|
|
def _is_safe_relative_path(path: str) -> bool:
|
|
"""Relative path with no leading `/`, `~`, drive letter, or `..` segments."""
|
|
if not isinstance(path, str) or not path:
|
|
return False
|
|
if path[0] in ("/", "\\", "~"):
|
|
return False
|
|
if len(path) >= 2 and path[1] == ":":
|
|
return False
|
|
return ".." not in path.replace("\\", "/").split("/")
|
|
|
|
def _path_arg_is_sandbox_local(node: ast.AST | None) -> bool:
|
|
"""Whether the path argument resolves to a sandbox-local literal."""
|
|
if node is None:
|
|
return False
|
|
if isinstance(node, ast.Constant) and isinstance(node.value, (bytes, bytearray)):
|
|
return True # inline bytes, no file access
|
|
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
|
return _is_safe_relative_path(node.value)
|
|
if isinstance(node, ast.Call):
|
|
f = node.func
|
|
is_open = (isinstance(f, ast.Name) and f.id == "open") or (
|
|
isinstance(f, ast.Attribute) and f.attr == "open"
|
|
)
|
|
if is_open and node.args:
|
|
a0 = node.args[0]
|
|
return (
|
|
isinstance(a0, ast.Constant)
|
|
and isinstance(a0.value, str)
|
|
and _is_safe_relative_path(a0.value)
|
|
)
|
|
return False
|
|
|
|
def _hf_upload_violation(node: ast.Call, method_name: str) -> str | None:
|
|
"""Inspect an HF upload call; return a violation reason or None.
|
|
|
|
Policy: HF uploads are allowed only when (a) no sensitive kwarg is set,
|
|
(b) no positional / keyword value reads `os.environ` or related env
|
|
readers, and (c) the path arg is a sandbox-local literal: a relative
|
|
string with no `..`, an `open(<literal>)`, or inline bytes. Dynamic /
|
|
variable paths are rejected since safety can't be proven statically and
|
|
a wrong-allow means credential exfiltration.
|
|
"""
|
|
for kw in node.keywords or []:
|
|
if kw.arg in _HF_SENSITIVE_KWARGS:
|
|
return (
|
|
f"HF upload {kw.arg}= cannot be set from sandboxed code; "
|
|
"uploads run with the sandbox identity only"
|
|
)
|
|
all_values = list(node.args or []) + [kw.value for kw in (node.keywords or [])]
|
|
for v in all_values:
|
|
if _reads_env_or_secret(v):
|
|
return (
|
|
"HF upload cannot include os.environ / os.getenv / subprocess "
|
|
"env reads; secrets and tokens must not be exfiltrated"
|
|
)
|
|
if method_name == "create_commit":
|
|
for kw in node.keywords or []:
|
|
if kw.arg == "operations" and isinstance(kw.value, ast.List):
|
|
for elt in kw.value.elts:
|
|
if isinstance(elt, ast.Call):
|
|
inner = _hf_upload_violation(elt, "upload_file")
|
|
if inner:
|
|
return inner
|
|
return None
|
|
path_node: ast.AST | None = node.args[0] if node.args else None
|
|
for kw in node.keywords or []:
|
|
if kw.arg in ("path_or_fileobj", "folder_path"):
|
|
path_node = kw.value
|
|
break
|
|
if not _path_arg_is_sandbox_local(path_node):
|
|
return (
|
|
"HF upload path must be a sandbox-local relative-path literal "
|
|
"(no absolute paths, no '..' segments, no dynamic expressions)"
|
|
)
|
|
return None
|
|
|
|
class NetworkAndIoVisitor(ast.NodeVisitor):
|
|
def visit_Call(self, node):
|
|
parts: list[str] = []
|
|
cur = node.func
|
|
while isinstance(cur, ast.Attribute):
|
|
parts.insert(0, cur.attr)
|
|
cur = cur.value
|
|
if isinstance(cur, ast.Name):
|
|
parts.insert(0, cur.id)
|
|
fq = ".".join(parts) if parts else ""
|
|
|
|
hf_upload_name = _method_call_hf_upload_name(node)
|
|
if hf_upload_name is not None:
|
|
violation = _hf_upload_violation(node, hf_upload_name)
|
|
if violation is not None:
|
|
network_calls.append(
|
|
{
|
|
"type": "upload_blocked",
|
|
"line": getattr(node, "lineno", -1),
|
|
"description": f"Blocked: {violation}",
|
|
}
|
|
)
|
|
|
|
# Direct sock.connect((host, port)) bypasses the FQ-prefix branch.
|
|
if isinstance(node.func, ast.Attribute) and node.func.attr == "connect" and node.args:
|
|
a0 = node.args[0]
|
|
host_lit = None
|
|
if isinstance(a0, ast.Tuple) and a0.elts:
|
|
e0 = a0.elts[0]
|
|
if isinstance(e0, ast.Constant) and isinstance(e0.value, str):
|
|
host_lit = e0.value
|
|
elif isinstance(a0, ast.Constant) and isinstance(a0.value, str):
|
|
host_lit = a0.value
|
|
if host_lit:
|
|
if _is_metadata_host(host_lit):
|
|
network_calls.append(
|
|
{
|
|
"type": "metadata_host_blocked",
|
|
"line": getattr(node, "lineno", -1),
|
|
"description": "Blocked: cloud-metadata host",
|
|
}
|
|
)
|
|
elif not _is_trusted_host(host_lit):
|
|
network_calls.append(
|
|
{
|
|
"type": "untrusted_host_blocked",
|
|
"line": getattr(node, "lineno", -1),
|
|
"description": (
|
|
"Blocked: host not in sandbox allowlist; "
|
|
"use an allowed informational source"
|
|
),
|
|
}
|
|
)
|
|
|
|
if fq and any(fq.startswith(p) for p in _NETWORK_FQ_PREFIXES):
|
|
# 1) Upload-shape check (host-independent).
|
|
if _call_is_upload_shape(node, fq):
|
|
network_calls.append(
|
|
{
|
|
"type": "upload_blocked",
|
|
"line": getattr(node, "lineno", -1),
|
|
"description": ("Blocked: file upload disallowed in sandbox"),
|
|
}
|
|
)
|
|
|
|
# 2) Extract literal host (URL string or (host, port) tuple).
|
|
host_arg = None
|
|
url_arg = None
|
|
if node.args:
|
|
a0 = node.args[0]
|
|
if isinstance(a0, ast.Constant) and isinstance(a0.value, str):
|
|
url_arg = a0.value
|
|
elif isinstance(a0, ast.Tuple) and a0.elts:
|
|
e0 = a0.elts[0]
|
|
if isinstance(e0, ast.Constant) and isinstance(e0.value, str):
|
|
host_arg = e0.value
|
|
if url_arg and host_arg is None:
|
|
m = re.match(r"^\w+://([^/?#]+)", url_arg)
|
|
if m:
|
|
host_arg = m.group(1)
|
|
|
|
if host_arg:
|
|
if _is_metadata_host(host_arg):
|
|
network_calls.append(
|
|
{
|
|
"type": "metadata_host_blocked",
|
|
"line": getattr(node, "lineno", -1),
|
|
"description": "Blocked: cloud-metadata host",
|
|
}
|
|
)
|
|
elif not _is_trusted_host(host_arg):
|
|
network_calls.append(
|
|
{
|
|
"type": "untrusted_host_blocked",
|
|
"line": getattr(node, "lineno", -1),
|
|
"description": (
|
|
"Blocked: host not in sandbox allowlist; "
|
|
"use an allowed informational source"
|
|
),
|
|
}
|
|
)
|
|
|
|
is_open_call = (
|
|
(isinstance(node.func, ast.Name) and node.func.id == "open")
|
|
or fq in ("io.open", "pathlib.Path.open")
|
|
or fq.endswith(".open")
|
|
)
|
|
if is_open_call and node.args:
|
|
a0 = node.args[0]
|
|
path_lit = None
|
|
if isinstance(a0, ast.Constant) and isinstance(a0.value, str):
|
|
path_lit = a0.value
|
|
if path_lit:
|
|
flagged = False
|
|
if any(path_lit.startswith(p) for p in _SENSITIVE_FILE_PREFIXES):
|
|
flagged = True
|
|
elif _SENSITIVE_FILE_RE.match(path_lit):
|
|
flagged = True
|
|
if flagged:
|
|
sensitive_file_reads.append(
|
|
{
|
|
"type": "sensitive_file_read",
|
|
"line": getattr(node, "lineno", -1),
|
|
"description": (
|
|
f"open({path_lit!r}) targets a host identity / "
|
|
"credential file; sandboxed code may not read it"
|
|
),
|
|
}
|
|
)
|
|
self.generic_visit(node)
|
|
|
|
NetworkAndIoVisitor().visit(tree)
|
|
|
|
is_safe = (
|
|
len(signal_tampering) == 0
|
|
and len(exception_catching) == 0
|
|
and len(shell_escapes) == 0
|
|
and len(network_calls) == 0
|
|
and len(sensitive_file_reads) == 0
|
|
)
|
|
return is_safe, {
|
|
"signal_tampering": signal_tampering,
|
|
"exception_catching": exception_catching,
|
|
"shell_escapes": shell_escapes,
|
|
"network_calls": network_calls,
|
|
"sensitive_file_reads": sensitive_file_reads,
|
|
"warnings": warnings,
|
|
}
|
|
|
|
|
|
def _check_code_safety(code: str) -> str | None:
|
|
"""Validate code safety via static analysis.
|
|
|
|
Returns an error message string if the code is unsafe, or None if OK.
|
|
"""
|
|
safe, info = _check_signal_escape_patterns(code)
|
|
if not safe:
|
|
# Let SyntaxError from ast.parse through so the subprocess produces a
|
|
# normal Python traceback instead of a misleading "unsafe code" message.
|
|
if info.get("error"):
|
|
return None
|
|
|
|
reasons = [item.get("description", "") for item in info.get("signal_tampering", [])]
|
|
shell_reasons = [item.get("description", "") for item in info.get("shell_escapes", [])]
|
|
exception_reasons = [
|
|
item.get("description", "") for item in info.get("exception_catching", [])
|
|
]
|
|
network_reasons = [item.get("description", "") for item in info.get("network_calls", [])]
|
|
file_reasons = [
|
|
item.get("description", "") for item in info.get("sensitive_file_reads", [])
|
|
]
|
|
all_reasons = [
|
|
r
|
|
for r in reasons + shell_reasons + exception_reasons + network_reasons + file_reasons
|
|
if r
|
|
]
|
|
if all_reasons:
|
|
return (
|
|
f"Error: unsafe code detected ({'; '.join(all_reasons)}). "
|
|
f"Please remove unsafe patterns from your code."
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
def _kill_process_tree(proc) -> None:
|
|
"""SIGKILL the setsid process group; fall back to single-pid kill."""
|
|
if proc.poll() is not None:
|
|
return
|
|
try:
|
|
pgid = os.getpgid(proc.pid)
|
|
except (ProcessLookupError, PermissionError):
|
|
pgid = None
|
|
if pgid is not None:
|
|
try:
|
|
os.killpg(pgid, signal.SIGKILL)
|
|
return
|
|
except (ProcessLookupError, PermissionError):
|
|
pass
|
|
try:
|
|
proc.kill()
|
|
except (ProcessLookupError, PermissionError):
|
|
pass
|
|
|
|
|
|
def _cancel_watcher(
|
|
proc,
|
|
cancel_event,
|
|
poll_interval = 0.2,
|
|
):
|
|
"""Daemon thread that kills a process when cancel_event is set."""
|
|
while proc.poll() is None:
|
|
if cancel_event is not None and cancel_event.is_set():
|
|
_kill_process_tree(proc)
|
|
return
|
|
cancel_event.wait(poll_interval) if cancel_event else None
|
|
|
|
|
|
def _truncate(text: str, limit: int = _MAX_OUTPUT_CHARS) -> str:
|
|
if len(text) > limit:
|
|
return text[:limit] + f"\n\n... (truncated, {len(text)} chars total)"
|
|
return text
|
|
|
|
|
|
def _python_exec(
|
|
code: str,
|
|
cancel_event = None,
|
|
timeout: int = _EXEC_TIMEOUT,
|
|
session_id: str | None = None,
|
|
disable_sandbox: bool = False,
|
|
) -> str:
|
|
"""Execute Python code in a subprocess sandbox.
|
|
|
|
disable_sandbox (Bypass Permissions): skip the safety analysis and rlimit
|
|
pre-exec, and use the host env minus secrets.
|
|
"""
|
|
if not code or not code.strip():
|
|
return "No code provided."
|
|
|
|
# Validate imports and code safety (skipped when the sandbox is disabled)
|
|
if not disable_sandbox:
|
|
error = _check_code_safety(code)
|
|
if error:
|
|
return error
|
|
elif not _harden_parent_against_proc_env_leak():
|
|
# Close the /proc/<parent>/environ secret-recovery path first; if it
|
|
# cannot be applied, fail closed rather than leak the parent environ.
|
|
return (
|
|
"Execution error: could not harden the Studio process against "
|
|
"/proc environment reads; refusing bypass execution."
|
|
)
|
|
|
|
tmp_path = None
|
|
workdir = _get_workdir(session_id)
|
|
# Snapshot image mtimes to detect new and overwritten files.
|
|
_before: dict[str, int] = {}
|
|
if os.path.isdir(workdir):
|
|
for _name in os.listdir(workdir):
|
|
if os.path.splitext(_name)[1].lower() in _IMAGE_EXTS:
|
|
_p = os.path.join(workdir, _name)
|
|
if os.path.isfile(_p):
|
|
try:
|
|
_before[_name] = os.stat(_p).st_mtime_ns
|
|
except OSError:
|
|
pass
|
|
try:
|
|
fd, tmp_path = tempfile.mkstemp(suffix = ".py", prefix = "studio_exec_", dir = workdir)
|
|
# utf-8 so non-ASCII in model-written code survives the OS default codec
|
|
# (Windows cp1252 would otherwise raise UnicodeEncodeError).
|
|
with os.fdopen(fd, "w", encoding = "utf-8") as f:
|
|
f.write(code)
|
|
|
|
safe_env = _build_bypass_env(workdir) if disable_sandbox else _build_safe_env(workdir)
|
|
if disable_sandbox:
|
|
# Match the sandboxed Python path without changing bypass shell I/O.
|
|
safe_env = dict(safe_env)
|
|
safe_env["PYTHONIOENCODING"] = "utf-8"
|
|
popen_kwargs = dict(
|
|
stdout = subprocess.PIPE,
|
|
stderr = subprocess.STDOUT,
|
|
text = True,
|
|
# Decode child output as utf-8 (it emits utf-8 via PYTHONIOENCODING);
|
|
# replace so non-ASCII output never crashes the read on Windows.
|
|
encoding = "utf-8",
|
|
errors = "replace",
|
|
cwd = workdir,
|
|
env = safe_env,
|
|
)
|
|
if sys.platform != "win32":
|
|
popen_kwargs["preexec_fn"] = _bypass_preexec if disable_sandbox else _sandbox_preexec
|
|
else:
|
|
popen_kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
|
|
|
|
proc = subprocess.Popen([sys.executable, tmp_path], **popen_kwargs)
|
|
|
|
# Spawn cancel watcher if we have a cancel event
|
|
if cancel_event is not None:
|
|
watcher = threading.Thread(
|
|
target = _cancel_watcher, args = (proc, cancel_event), daemon = True
|
|
)
|
|
watcher.start()
|
|
|
|
try:
|
|
output, _ = proc.communicate(timeout = timeout)
|
|
except subprocess.TimeoutExpired:
|
|
_kill_process_tree(proc)
|
|
try:
|
|
proc.communicate(timeout = 5)
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
return _truncate(f"Execution timed out after {timeout} seconds.")
|
|
|
|
if cancel_event is not None and cancel_event.is_set():
|
|
return "Execution cancelled."
|
|
|
|
result = output or ""
|
|
if proc.returncode != 0:
|
|
result = f"Exit code {proc.returncode}:\n{result}"
|
|
result = _truncate(result) if result.strip() else "(no output)"
|
|
|
|
# Detect new/overwritten images and append sentinel for the frontend
|
|
if session_id and os.path.isdir(workdir):
|
|
new_images = []
|
|
for _name in os.listdir(workdir):
|
|
if os.path.splitext(_name)[1].lower() not in _IMAGE_EXTS:
|
|
continue
|
|
_p = os.path.join(workdir, _name)
|
|
if not os.path.isfile(_p):
|
|
continue
|
|
try:
|
|
_mtime = os.stat(_p).st_mtime_ns
|
|
except OSError:
|
|
continue
|
|
if _name not in _before or _mtime != _before[_name]:
|
|
new_images.append(_name)
|
|
if new_images:
|
|
import json as _json
|
|
result += f"\n__IMAGES__:{_json.dumps(sorted(new_images))}"
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
return f"Execution error: {e}"
|
|
finally:
|
|
if tmp_path and os.path.exists(tmp_path):
|
|
try:
|
|
os.unlink(tmp_path)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _bash_exec(
|
|
command: str,
|
|
cancel_event = None,
|
|
timeout: int = _EXEC_TIMEOUT,
|
|
session_id: str | None = None,
|
|
disable_sandbox: bool = False,
|
|
) -> str:
|
|
"""Execute a bash command in a subprocess sandbox.
|
|
|
|
disable_sandbox (Bypass Permissions): skip the command blocklist and rlimit
|
|
pre-exec, and use the host env minus secrets.
|
|
"""
|
|
if not command or not command.strip():
|
|
return "No command provided."
|
|
|
|
# Block dangerous commands (skipped when the sandbox is disabled)
|
|
if not disable_sandbox:
|
|
blocked = _find_blocked_commands(command)
|
|
if blocked:
|
|
return f"Blocked command(s) for safety: {', '.join(sorted(blocked))}"
|
|
elif not _harden_parent_against_proc_env_leak():
|
|
# Close the /proc/<parent>/environ secret-recovery path first; if it
|
|
# cannot be applied, fail closed rather than leak the parent environ.
|
|
return (
|
|
"Execution error: could not harden the Studio process against "
|
|
"/proc environment reads; refusing bypass execution."
|
|
)
|
|
|
|
try:
|
|
workdir = _get_workdir(session_id)
|
|
safe_env = _build_bypass_env(workdir) if disable_sandbox else _build_safe_env(workdir)
|
|
popen_kwargs = dict(
|
|
stdout = subprocess.PIPE,
|
|
stderr = subprocess.STDOUT,
|
|
text = True,
|
|
cwd = workdir,
|
|
env = safe_env,
|
|
)
|
|
if sys.platform != "win32":
|
|
popen_kwargs["preexec_fn"] = _bypass_preexec if disable_sandbox else _sandbox_preexec
|
|
else:
|
|
popen_kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
|
|
|
|
proc = subprocess.Popen(_get_shell_cmd(command), **popen_kwargs)
|
|
|
|
if cancel_event is not None:
|
|
watcher = threading.Thread(
|
|
target = _cancel_watcher, args = (proc, cancel_event), daemon = True
|
|
)
|
|
watcher.start()
|
|
|
|
try:
|
|
output, _ = proc.communicate(timeout = timeout)
|
|
except subprocess.TimeoutExpired:
|
|
_kill_process_tree(proc)
|
|
try:
|
|
proc.communicate(timeout = 5)
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
return _truncate(f"Execution timed out after {timeout} seconds.")
|
|
|
|
if cancel_event is not None and cancel_event.is_set():
|
|
return "Execution cancelled."
|
|
|
|
result = output or ""
|
|
if proc.returncode != 0:
|
|
result = f"Exit code {proc.returncode}:\n{result}"
|
|
return _truncate(result) if result.strip() else "(no output)"
|
|
|
|
except Exception as e:
|
|
return f"Execution error: {e}"
|