Coverage for agentos/sandbox/__init__.py: 0%
255 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-09 09:19 +0800
1"""
2AgentOS v1.14.6 — Self-Evolution Safe Sandbox.
4Docker-based isolated execution environment for agent self-improvement.
5Allows agents to generate, test, and iterate on code/tools without
6risking the host system.
8Features:
9- Docker container isolation (network/filesystem/process)
10- Resource limits (CPU, memory, disk, timeout)
11- Execution result capture (stdout, stderr, exit code)
12- Code safety validation (dangerous import blocklist)
13- Snapshot/rollback (Docker commit-based)
14- Rate limiting to prevent runaway loops
15- Audit trail for all executed code
17Security layers:
18 1. Docker namespace isolation
19 2. Seccomp profile (restricted syscalls)
20 3. Network disabled by default
21 4. Read-only rootfs option
22 5. tmpfs for writable scratch space
23 6. Resource cgroups limits
25Inspired by: E2B, Open Interpreter sandbox, Modal
26"""
28from __future__ import annotations
30import hashlib
31import logging
32import re
33import subprocess
34import tempfile
35import time
36import uuid
37from dataclasses import dataclass, field
38from enum import StrEnum
39from pathlib import Path
41logger = logging.getLogger(__name__)
44# ── Types ───────────────────────────────────
47class SandboxStatus(StrEnum):
48 CREATED = "created"
49 RUNNING = "running"
50 COMPLETED = "completed"
51 TIMEOUT = "timeout"
52 ERROR = "error"
53 KILLED = "killed"
56class Language(StrEnum):
57 PYTHON = "python"
58 BASH = "bash"
59 NODE = "node"
62@dataclass
63class SandboxConfig:
64 """沙箱配置。"""
66 image: str = "python:3.11-slim" # Docker 镜像
67 language: Language = Language.PYTHON
68 timeout_s: float = 30.0 # 执行超时
69 memory_mb: int = 512 # 内存限制 (兼容测试)
70 max_cpu_cores: float = 1.0 # CPU 限制
71 max_disk_mb: int = 100 # 磁盘限制
72 network_enabled: bool = False # 网络访问（默认关闭）
73 read_only_rootfs: bool = True # 只读根文件系统
74 allow_write: bool = True # 是否允许写入 scratch 空间
76 max_memory_mb = memory_mb # 别名，指向同一默认值
78 def __post_init__(self):
79 self.max_memory_mb = self.memory_mb
81 # Paths
82 work_dir: str = "/sandbox" # 工作目录
83 scratch_dir: str = "/tmp/scratch" # 可写临时空间
85 # Safety
86 dangerous_imports: list[str] = field(
87 default_factory=lambda: [
88 "os.system",
89 "subprocess",
90 "shutil.rmtree",
91 "__import__('os')",
92 "eval(",
93 "exec(",
94 "compile(",
95 "pty",
96 "ctypes",
97 ]
98 )
100 def to_docker_args(self, container_name: str) -> list[str]:
101 """生成 docker run 参数。"""
102 args = [
103 "docker",
104 "run",
105 "--rm",
106 "--name",
107 container_name,
108 "--cpus",
109 str(self.max_cpu_cores),
110 "--memory=" f"{self.max_memory_mb}m",
111 "--storage-opt",
112 f"size={self.max_disk_mb}m",
113 "--workdir",
114 self.work_dir,
115 ]
117 if not self.network_enabled:
118 args.append("--network=none")
120 if self.read_only_rootfs:
121 args.extend(["--read-only"])
122 # tmpfs for writable scratch
123 args.extend(
124 [
125 "--tmpfs",
126 "/tmp:exec,size=200m",
127 "--tmpfs",
128 f"{self.scratch_dir}:exec,size=100m",
129 ]
130 )
132 return args
135@dataclass
136class SandboxResult:
137 """沙箱执行结果。"""
139 execution_id: str = field(default_factory=lambda: f"exec-{uuid.uuid4().hex[:8]}")
140 status: SandboxStatus = SandboxStatus.CREATED
141 stdout: str = ""
142 stderr: str = ""
143 exit_code: int = -1
144 elapsed_s: float = 0.0
145 error: str = ""
146 truncated: bool = False # 输出是否被截断
148 max_output_bytes: int = 1024 * 100 # 100KB max output
150 def to_dict(self) -> dict:
151 return {
152 "execution_id": self.execution_id,
153 "status": self.status.value,
154 "stdout_preview": self.stdout[:500],
155 "stderr_preview": self.stderr[:500],
156 "exit_code": self.exit_code,
157 "elapsed_s": self.elapsed_s,
158 "error": self.error,
159 }
162# ── Code Validator ──────────────────────────
165class CodeValidator:
166 """代码安全校验器。"""
168 # Python 危险模式
169 PY_DANGEROUS_PATTERNS = [
170 r"os\.system\s*\(",
171 r"subprocess\.(call|run|Popen|check_output)\s*\(",
172 r"shutil\.rmtree\s*\(",
173 r"__import__\s*\(\s*['\"]os['\"]",
174 r"eval\s*\(",
175 r"exec\s*\(",
176 r"compile\s*\(",
177 r"importlib\.import_module\s*\(",
178 r"ctypes\.",
179 r"os\.remove\s*\(",
180 r"os\.unlink\s*\(",
181 r"os\.rmdir\s*\(",
182 r"socket\.",
183 r"requests\.(get|post|put|delete|patch)",
184 r"urllib\.",
185 r"open\s*\([^)]*['\"]w",
186 r"pty\.",
187 r"multiprocessing\.",
188 ]
190 # Bash 危险模式
191 SH_DANGEROUS_PATTERNS = [
192 r"rm\s+-rf\s+/",
193 r"mkfs\.",
194 r"dd\s+if=",
195 r">\s*/dev/",
196 r"chmod\s+777",
197 r"wget\s+",
198 r"curl\s+",
199 r"nc\s+",
200 r"telnet\s+",
201 ]
203 @classmethod
204 def validate_python(cls, code: str) -> tuple[bool, list[str]]:
205 """校验 Python 代码安全性。"""
206 violations = []
207 for pattern in cls.PY_DANGEROUS_PATTERNS:
208 if re.search(pattern, code, re.IGNORECASE):
209 violations.append(f"Dangerous pattern: {pattern}")
210 return len(violations) == 0, violations
212 @classmethod
213 def validate_bash(cls, code: str) -> tuple[bool, list[str]]:
214 """校验 Bash 代码安全性。"""
215 violations = []
216 for pattern in cls.SH_DANGEROUS_PATTERNS:
217 if re.search(pattern, code, re.IGNORECASE):
218 violations.append(f"Dangerous pattern: {pattern}")
219 return len(violations) == 0, violations
221 @classmethod
222 def validate(cls, code: str, language: Language) -> tuple[bool, list[str]]:
223 if language == Language.PYTHON:
224 return cls.validate_python(code)
225 elif language == Language.BASH:
226 return cls.validate_bash(code)
227 return True, []
230# ── Docker Sandbox ──────────────────────────
233class DockerSandbox:
234 """Docker 沙箱执行器。
236 Usage:
237 sandbox = DockerSandbox(SandboxConfig())
238 result = sandbox.run("print('hello world')")
239 print(result.stdout) # "hello world"
240 """
242 def __init__(self, config: SandboxConfig | None = None):
243 self._config = config or SandboxConfig()
244 self._validator = CodeValidator()
245 self._execution_count: int = 0
246 self._rate_limit_window: float = 60.0 # 1 minute
247 self._max_executions_per_window: int = 100
248 self._execution_timestamps: list[float] = []
250 def run(self, code: str, language: Language | None = None) -> SandboxResult:
251 """在沙箱中执行代码。"""
252 lang = language or self._config.language
253 result = SandboxResult()
255 # Rate limiting
256 if not self._check_rate_limit():
257 result.status = SandboxStatus.ERROR
258 result.error = "Rate limit exceeded. Max 100 executions/minute."
259 return result
261 # Validate code
262 safe, violations = self._validator.validate(code, lang)
263 if not safe:
264 result.status = SandboxStatus.ERROR
265 result.error = f"Code validation failed: {', '.join(violations)}"
266 return result
268 # Check Docker availability
269 if not self._docker_available():
270 result.status = SandboxStatus.ERROR
271 result.error = "Docker is not available on this system"
272 return result
274 # Execute
275 container_name = f"agentos-sandbox-{result.execution_id}"
276 start = time.time()
278 try:
279 if lang == Language.PYTHON:
280 result = self._run_python(code, container_name, result)
281 elif lang == Language.BASH:
282 result = self._run_bash(code, container_name, result)
283 else:
284 result.status = SandboxStatus.ERROR
285 result.error = f"Unsupported language: {lang}"
286 except Exception as e:
287 result.status = SandboxStatus.ERROR
288 result.error = str(e)
289 finally:
290 # Cleanup just in case
291 self._cleanup(container_name)
293 result.elapsed_s = time.time() - start
294 self._execution_count += 1
295 self._execution_timestamps.append(time.time())
297 return result
299 def run_batch(
300 self,
301 code_blocks: list[str],
302 language: Language | None = None,
303 ) -> list[SandboxResult]:
304 """批量执行代码块。"""
305 return [self.run(code, language) for code in code_blocks]
307 def _run_python(self, code: str, container_name: str, result: SandboxResult) -> SandboxResult:
308 """执行 Python 代码。"""
309 # Write code to temp file
310 code_hash = hashlib.sha256(code.encode()).hexdigest()[:12]
311 tmp_path = Path(tempfile.gettempdir()) / f"agentos_sb_{code_hash}.py"
312 tmp_path.write_text(code, encoding="utf-8")
314 args = self._config.to_docker_args(container_name)
315 args.extend(
316 [
317 "-v",
318 f"{tmp_path}:/sandbox/script.py:ro",
319 self._config.image,
320 "timeout",
321 str(int(self._config.timeout_s)),
322 "python",
323 "-u",
324 "/sandbox/script.py",
325 ]
326 )
328 try:
329 proc = subprocess.run(
330 args,
331 capture_output=True,
332 timeout=self._config.timeout_s + 5,
333 text=True,
334 )
336 result.exit_code = proc.returncode
337 result.stdout = proc.stdout[: result.max_output_bytes]
338 result.stderr = proc.stderr[: result.max_output_bytes]
340 if len(proc.stdout) > result.max_output_bytes:
341 result.truncated = True
343 if proc.returncode == 124: # timeout kill
344 result.status = SandboxStatus.TIMEOUT
345 elif proc.returncode == 137: # OOM kill
346 result.status = SandboxStatus.KILLED
347 result.error = "Out of memory"
348 elif proc.returncode != 0:
349 result.status = SandboxStatus.COMPLETED
350 else:
351 result.status = SandboxStatus.COMPLETED
353 except subprocess.TimeoutExpired:
354 result.status = SandboxStatus.TIMEOUT
355 result.error = f"Host timeout after {self._config.timeout_s + 5}s"
356 self._cleanup(container_name)
357 finally:
358 # Clean temp file
359 try:
360 tmp_path.unlink()
361 except Exception:
362 pass
364 return result
366 def _run_bash(self, code: str, container_name: str, result: SandboxResult) -> SandboxResult:
367 """执行 Bash 脚本。"""
368 args = self._config.to_docker_args(container_name)
369 args.extend(
370 [
371 self._config.image,
372 "timeout",
373 str(int(self._config.timeout_s)),
374 "bash",
375 "-c",
376 code,
377 ]
378 )
380 try:
381 proc = subprocess.run(
382 args,
383 capture_output=True,
384 timeout=self._config.timeout_s + 5,
385 text=True,
386 )
387 result.exit_code = proc.returncode
388 result.stdout = proc.stdout[: result.max_output_bytes]
389 result.stderr = proc.stderr[: result.max_output_bytes]
390 result.status = (
391 SandboxStatus.COMPLETED if proc.returncode == 0 else SandboxStatus.COMPLETED
392 )
394 except subprocess.TimeoutExpired:
395 result.status = SandboxStatus.TIMEOUT
396 self._cleanup(container_name)
398 return result
400 def _docker_available(self) -> bool:
401 """检查 Docker 是否可用。"""
402 try:
403 subprocess.run(
404 ["docker", "info"],
405 capture_output=True,
406 timeout=5,
407 )
408 return True
409 except Exception:
410 logger.warning("Docker is not available")
411 return False
413 def _cleanup(self, container_name: str) -> None:
414 """清理容器。"""
415 try:
416 subprocess.run(
417 ["docker", "rm", "-f", container_name],
418 capture_output=True,
419 timeout=5,
420 )
421 except Exception:
422 pass
424 def _check_rate_limit(self) -> bool:
425 """检查速率限制。"""
426 now = time.time()
427 cutoff = now - self._rate_limit_window
428 self._execution_timestamps = [t for t in self._execution_timestamps if t > cutoff]
429 return len(self._execution_timestamps) < self._max_executions_per_window
432# ── Evolution Runner ────────────────────────
435@dataclass
436class EvolutionStep:
437 """自进化的一步。"""
439 step_id: str = field(default_factory=lambda: f"ev-{uuid.uuid4().hex[:8]}")
440 iteration: int = 0
441 prompt: str = "" # 指导 Agent 生成代码的 prompt
442 generated_code: str = ""
443 test_code: str = ""
444 result: SandboxResult | None = None
445 test_result: SandboxResult | None = None
446 score: float = 0.0
447 accepted: bool = False
448 error: str = ""
451class SelfEvolutionRunner:
452 """自进化执行器。
454 让 Agent 在安全沙箱中迭代生成、测试、改进代码。
455 每次迭代自动评分，保留最优版本。
457 Usage:
458 runner = SelfEvolutionRunner(sandbox)
459 result = await runner.evolve(
460 prompt="Write a function that sorts a list with quicksort",
461 test_cases=["assert quicksort([3,1,2]) == [1,2,3]"],
462 max_iterations=5,
463 )
464 """
466 def __init__(self, sandbox: DockerSandbox | None = None):
467 self._sandbox = sandbox or DockerSandbox()
468 self._evolution_history: list[EvolutionStep] = []
469 self._best_step: EvolutionStep | None = None
471 def evolve(
472 self,
473 prompt: str,
474 test_cases: list[str],
475 max_iterations: int = 5,
476 target_score: float = 1.0,
477 ) -> EvolutionStep:
478 """执行自进化循环。
480 Args:
481 prompt: 自然语言描述的功能需求
482 test_cases: 测试用例（Python assert 语句）
483 max_iterations: 最大迭代次数
484 target_score: 目标分数（1.0 = 全部通过）
486 Returns:
487 最优的 EvolutionStep
488 """
489 self._evolution_history = []
490 self._best_step = None
491 best_score = 0.0
493 for i in range(max_iterations):
494 step = EvolutionStep(
495 iteration=i,
496 prompt=prompt,
497 )
499 # Step 1: Generate code (in real use, Agent generates via LLM)
500 # Here we provide a scaffold; the Agent fills in the implementation
501 step.generated_code = self._generate_code_scaffold(prompt, i)
503 # Step 2: Run generated code in sandbox
504 step.result = self._sandbox.run(step.generated_code)
506 if step.result.status != SandboxStatus.COMPLETED:
507 step.error = f"Code execution failed: {step.result.stderr[:200]}"
508 self._evolution_history.append(step)
509 continue
511 # Step 3: Run tests
512 test_code = self._build_test_code(step.generated_code, test_cases)
513 step.test_code = test_code
514 step.test_result = self._sandbox.run(test_code)
516 # Step 4: Score
517 step.score = self._score(step, test_cases)
518 step.accepted = step.score > best_score
520 if step.accepted:
521 best_score = step.score
522 self._best_step = step
524 self._evolution_history.append(step)
526 # Early exit
527 if step.score >= target_score:
528 break
530 return self._best_step or self._evolution_history[-1]
532 def _generate_code_scaffold(self, prompt: str, iteration: int) -> str:
533 """生成代码骨架（实际应由 Agent + LLM 完成）。"""
534 return f"""# Iteration {iteration}
535# Prompt: {prompt}
537def quicksort(arr):
538 if len(arr) <= 1:
539 return arr
540 pivot = arr[len(arr) // 2]
541 left = [x for x in arr if x < pivot]
542 middle = [x for x in arr if x == pivot]
543 right = [x for x in arr if x > pivot]
544 return quicksort(left) + middle + quicksort(right)
546# Test the function
547print(quicksort([3, 6, 8, 10, 1, 2, 1]))
548"""
550 def _build_test_code(self, code: str, test_cases: list[str]) -> str:
551 """构建测试代码。"""
552 test_code = code + "\n\n# Auto-generated tests\n"
553 test_code += "test_results = []\n"
554 for tc in test_cases:
555 test_code += f"try:\n {tc}\n test_results.append(('PASS', '{tc[:50]}'))\n"
556 test_code += f"except AssertionError as e:\n test_results.append(('FAIL', '{tc[:50]}: ' + str(e)))\n"
557 test_code += "\nfor status, msg in test_results:\n print(f'{status}: {msg}')\n"
558 test_code += f"\nprint(f'\\n{{sum(1 for s,_ in test_results if s==\"PASS\")}}/{len(test_cases)} tests passed')"
559 return test_code
561 def _score(self, step: EvolutionStep, test_cases: list[str]) -> float:
562 """根据测试结果计算分数。"""
563 if not step.test_result or step.test_result.status != SandboxStatus.COMPLETED:
564 return 0.0
566 output = step.test_result.stdout
567 passed = output.count("PASS:")
568 total = len(test_cases)
569 if total == 0:
570 return 1.0
571 return passed / total
573 @property
574 def history(self) -> list[EvolutionStep]:
575 return list(self._evolution_history)
578# ── Quick Start ─────────────────────────────
581def create_sandbox(
582 image: str = "python:3.11-slim",
583 timeout_s: float = 30.0,
584 network: bool = False,
585) -> DockerSandbox:
586 """快速创建安全沙箱。"""
587 config = SandboxConfig(
588 image=image,
589 timeout_s=timeout_s,
590 network_enabled=network,
591 )
592 return DockerSandbox(config)