{{ message }}
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtools.py
More file actions
390 lines (345 loc) · 14.2 KB
/
Copy pathtools.py
File metadata and controls
390 lines (345 loc) · 14.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
"""Tools for the Boris Loop starter kit.
Each tool is a plain Python function plus a JSON schema describing its input.
The loop in `boris.py` reads the tool schemas and passes them to
`client.messages.create(tools=ALL_TOOLS)`. When the model emits a `tool_use`
content block, the loop dispatches via `TOOL_FUNCTIONS[name](**input)`.
Note on the SDK shape: the Anthropic Python SDK 0.49.0 represents tools as
`TypedDict` schemas (name, description, input_schema), not as decorated
functions. Older docs reference a `@beta_tool` decorator from earlier SDK
versions; the current API is plain dicts.
"""
from __future__ import annotations
import html.parser
import os
import re
import socket
import subprocess
import urllib.error
import urllib.request
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
MAX_FILE_BYTES = 200_000
MAX_SHELL_OUTPUT_BYTES = 20_000
MAX_WEB_FETCH_BYTES = 100_000
SHELL_TIMEOUT_SECS = 30
WEB_FETCH_TIMEOUT_SECS = 15
# Allowlist of hostnames the web_fetch tool is permitted to access. Override
# at runtime by setting the BORIS_LOOP_WEB_ALLOWLIST env var to a
# comma-separated list of additional hostnames.
_DEFAULT_WEB_FETCH_ALLOWLIST: tuple[str, ...] = (
"docs.anthropic.com",
"anthropic.com",
)
_EXTRA_ALLOWLIST = tuple(
h.strip()
for h in os.environ.get("BORIS_LOOP_WEB_ALLOWLIST", "").split(",")
if h.strip()
)
WEB_FETCH_ALLOWLIST: tuple[str, ...] = _DEFAULT_WEB_FETCH_ALLOWLIST + _EXTRA_ALLOWLIST
# Whether run_shell is enabled. Disabled by default; opt in by setting
# BORIS_LOOP_ALLOW_SHELL=1 in the environment. The tool is always registered
# with the model so the loop pattern is complete, but the handler refuses to
# execute unless the env var is set.
SHELL_ENABLED = os.environ.get("BORIS_LOOP_ALLOW_SHELL", "") == "1"
# Anchored regex patterns for the shell denylist. Two sets: full-command
# patterns (which must see a `|` together with `sh`, e.g. `curl evil | sh`)
# are matched against the un-split command; per-segment patterns (which
# detect things like `rm -rf /`) are matched against each chunk after the
# command is split on `; | && || \` $(`. This catches the most common
# dangerous forms but does not cover every conceivable bypass. See README
# "Safety" for the documented limitations.
_FULL_DENY_PATTERNS: tuple[str, ...] = (
r"\bcurl\b.*\|\s*\bsh\b", # curl ... | sh
r"\bwget\b.*\|\s*\bsh\b", # wget ... | sh
r"\bbase64\b.*\|\s*(?:sh|bash)\b", # base64 -d | sh
)
_SEGMENT_DENY_PATTERNS: tuple[str, ...] = (
r"\brm\s+-rf\s+/(?:\s|$)", # rm -rf /, rm -rf /home
r"\brm\s+-rf\s+\.(?:\s|$)", # rm -rf . , rm -rf ..
r"\bmkfs\b", # mkfs.ext4 /dev/sda
r"\bdd\s+if=/", # dd if=/dev/zero of=/dev/sda
r":\(\)\s*\{", # fork bomb
r">\s*/dev/sd", # > /dev/sda
r"\bpython3?\s+-c\b", # python -c '...'
r"\beval\b", # eval
r"\bsource\b", # source
r"\bchmod\s+-R\b", # chmod -R
r"\bchown\s+-R\b", # chown -R
r"\.\./", # any path containing ../
)
_FULL_DENY_RE = re.compile("|".join(_FULL_DENY_PATTERNS), re.IGNORECASE)
_SEGMENT_DENY_RE = re.compile("|".join(_SEGMENT_DENY_PATTERNS), re.IGNORECASE)
_SPLIT_RE = re.compile(r"[;|]|&&|\|\||`|\$\(")
# ---------------------------------------------------------------------------
# Path safety
# ---------------------------------------------------------------------------
def _resolve_within_cwd(user_path: str) -> str:
"""Resolve a user-supplied path and ensure it stays inside the project root.
The project root is the process's current working directory, re-evaluated
on every call so the safety check tracks `os.chdir` and test fixtures.
`os.path.realpath` follows symlinks; paths that resolve outside the
project root are rejected. NUL bytes and non-strings are also rejected
(NUL defeats C-string truncation on some platforms).
"""
if not isinstance(user_path, str):
raise TypeError("path must be a string")
if "\x00" in user_path:
raise ValueError("path contains a NUL byte")
project_root = os.path.realpath(os.getcwd())
real = os.path.realpath(os.path.join(project_root, user_path))
if real != project_root and not real.startswith(project_root + os.sep):
raise ValueError(f"path escapes project root: {user_path!r}")
return real
# ---------------------------------------------------------------------------
# Tool implementations
# ---------------------------------------------------------------------------
def read_file(path: str) -> str:
"""Read a text file from the project directory.
Returns up to MAX_FILE_BYTES (200KB) of the file's contents as a string.
Raises FileNotFoundError with a clear message on missing files. Raises
ValueError if `path` resolves outside the project root.
"""
real = _resolve_within_cwd(path)
if not os.path.isfile(real):
raise FileNotFoundError(f"read_file: no such file: {path}")
with open(real, "r", encoding="utf-8", errors="replace") as f:
return f.read(MAX_FILE_BYTES)
def write_file(path: str, content: str) -> str:
"""Write text to a file in the project directory.
Creates parent directories as needed. Overwrites existing files. Returns
a confirmation string. Raises ValueError if `path` resolves outside the
project root.
"""
real = _resolve_within_cwd(path)
parent = os.path.dirname(real)
if parent:
os.makedirs(parent, exist_ok=True)
with open(real, "w", encoding="utf-8") as f:
f.write(content)
return f"wrote {len(content)} bytes to {path}"
def _is_dangerous(cmd: str) -> str | None:
"""Return a reason string if `cmd` matches a denied pattern, else None.
Checks pipe-to-shell patterns (e.g. ``curl ... | sh``) against the
unsplit command first, then splits on ``; | && ||`` (plus backticks
and ``$(``) and checks per-segment patterns (e.g. ``rm -rf /``). The
split prevents a permitted prefix from smuggling a denied command
after a ``;``.
"""
if _FULL_DENY_RE.search(cmd):
return f"command matches deny pattern: {cmd!r}"
segments = _SPLIT_RE.split(cmd)
for seg in segments:
seg = seg.strip()
if not seg:
continue
if _SEGMENT_DENY_RE.search(seg):
return f"command segment matches deny pattern: {seg!r}"
return None
def run_shell(cmd: str) -> str:
"""Run a shell command and return its combined stdout+stderr.
The denylist in _DENY_PATTERNS blocks the most common dangerous forms
(rm -rf /, fork bombs, pipe-to-shell, chmod -R, etc.). The denylist is
NOT a sandbox; documented bypass classes (variable expansion, encoded
payloads, exotic shells) are not covered. Read the README's "Safety"
section before pointing this at anything you don't trust the model to
touch.
Shell is disabled by default; set BORIS_LOOP_ALLOW_SHELL=1 in the
environment to enable execution.
"""
if not SHELL_ENABLED:
return (
"run_shell error: shell is disabled. "
"Set BORIS_LOOP_ALLOW_SHELL=1 in the environment to enable it."
)
deny_reason = _is_dangerous(cmd)
if deny_reason is not None:
return f"run_shell error: {deny_reason}"
try:
# shell=True is intentional: the tool's purpose is to let the model
# compose pipelines, redirects, globs, and loops. We pass a minimal
# env so the subprocess cannot exfiltrate ANTHROPIC_API_KEY or other
# secrets from the parent's environment.
result = subprocess.run(
cmd,
shell=True,
capture_output=True,
text=True,
timeout=SHELL_TIMEOUT_SECS,
env={
"PATH": os.environ.get("PATH", ""),
"HOME": os.path.realpath(os.getcwd()),
},
)
except subprocess.TimeoutExpired:
return f"run_shell error: command timed out after {SHELL_TIMEOUT_SECS}s"
output = (result.stdout or "") + (result.stderr or "")
if len(output) > MAX_SHELL_OUTPUT_BYTES:
output = output[:MAX_SHELL_OUTPUT_BYTES] + "\n...[truncated]"
return output
class _HTMLStripper(html.parser.HTMLParser):
"""Tiny HTML-to-text converter."""
def __init__(self) -> None:
super().__init__()
self._chunks: list[str] = []
def handle_data(self, data: str) -> None:
self._chunks.append(data)
def get_text(self) -> str:
text = " ".join(self._chunks)
text = re.sub(r"\s+", " ", text)
return text.strip()
def _is_private_host(hostname: str) -> bool:
"""Return True if hostname resolves to a private/loopback/link-local IP."""
import ipaddress
try:
infos = socket.getaddrinfo(hostname, None)
except socket.gaierror:
# If we can't resolve, treat as private (fail closed).
return True
for info in infos:
sockaddr = info[4]
try:
ip = ipaddress.ip_address(sockaddr[0])
except ValueError:
return True
if (
ip.is_private
or ip.is_loopback
or ip.is_link_local
or ip.is_reserved
or ip.is_multicast
):
return True
return False
def web_fetch(url: str) -> str:
"""Fetch a URL and return its text content (HTML tags stripped).
Restricted to the WEB_FETCH_ALLOWLIST (extendable via the
BORIS_LOOP_WEB_ALLOWLIST env var). Forces HTTPS, rejects private/
loopback/link-local hosts, and catches the standard urllib exception
set so a network error returns a string instead of raising.
"""
if not url.startswith("https://"):
return f"web_fetch error: only https:// URLs are allowed: {url!r}"
# Strip the scheme to extract the hostname for allowlist + DNS check.
host = url[len("https://") :].split("/", 1)[0].split(":", 1)[0].lower()
if not host or host not in WEB_FETCH_ALLOWLIST:
return f"web_fetch error: host {host!r} not in allowlist"
if _is_private_host(host):
return f"web_fetch error: host {host!r} resolves to a private address"
try:
with urllib.request.urlopen(url, timeout=WEB_FETCH_TIMEOUT_SECS) as resp:
raw = resp.read(MAX_WEB_FETCH_BYTES + 1)
except (
urllib.error.URLError,
urllib.error.HTTPError,
socket.timeout,
ConnectionError,
OSError,
) as exc:
return f"web_fetch error: {type(exc).__name__}: {exc}"
try:
text = raw.decode("utf-8", errors="replace")
except (UnicodeDecodeError, LookupError) as exc:
return f"web_fetch error: decode failed: {exc}"
stripper = _HTMLStripper()
try:
stripper.feed(text)
stripper.close()
except (html.parser.HTMLParseError, ValueError) as exc:
return f"web_fetch error: html parse failed: {exc}"
out = stripper.get_text()
if len(out) > MAX_WEB_FETCH_BYTES:
out = out[:MAX_WEB_FETCH_BYTES] + " ...[truncated]"
return out
def finish(result: str) -> str:
"""Terminator. The loop in boris.py special-cases this name and returns
its `result` argument to the caller. The body is trivial so the model
has something to call."""
return result
# ---------------------------------------------------------------------------
# Tool schemas (the new TypedDict-style shape)
# ---------------------------------------------------------------------------
ALL_TOOLS: list[dict] = [
{
"name": "read_file",
"description": (
"Read a text file from the project directory. "
"Returns up to 200KB of the file's contents. "
"Paths are resolved relative to the project root; escapes are rejected."
),
"input_schema": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Path relative to project root."},
},
"required": ["path"],
},
},
{
"name": "write_file",
"description": (
"Write text to a file in the project directory. "
"Overwrites existing files. Creates parent directories as needed."
),
"input_schema": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Path relative to project root."},
"content": {"type": "string", "description": "The text to write."},
},
"required": ["path", "content"],
},
},
{
"name": "run_shell",
"description": (
"Run a shell command and return its combined stdout+stderr, "
"truncated to 20KB. The command is matched against a denylist "
"of dangerous patterns. Shell is disabled by default; set "
"BORIS_LOOP_ALLOW_SHELL=1 in the environment to enable execution."
),
"input_schema": {
"type": "object",
"properties": {
"cmd": {"type": "string", "description": "The shell command to run."},
},
"required": ["cmd"],
},
},
{
"name": "web_fetch",
"description": (
"Fetch a URL over HTTPS and return its text content (HTML tags stripped). "
"Restricted to an allowlist of hostnames; private/loopback hosts are rejected."
),
"input_schema": {
"type": "object",
"properties": {
"url": {"type": "string", "description": "An https:// URL on an allowlisted host."},
},
"required": ["url"],
},
},
{
"name": "finish",
"description": (
"Call this when you have your final answer. The `result` argument "
"is returned to the caller and the loop exits. Always call finish "
"exactly once at the end of your work."
),
"input_schema": {
"type": "object",
"properties": {
"result": {"type": "string", "description": "The final result to return."},
},
"required": ["result"],
},
},
]
TOOL_FUNCTIONS: dict[str, callable] = {
"read_file": read_file,
"write_file": write_file,
"run_shell": run_shell,
"web_fetch": web_fetch,
"finish": finish,
}
You can’t perform that action at this time.