saintparish4/openscript
GitHub: saintparish4/openscript
OpenScript 是一个 Python LLM Agent 安全网关 SDK，通过可插拔的策略管道在操作前后拦截注入、脱敏敏感数据、管控工具调用并审批高风险行为。
Stars: 0 | Forks: 0
# OpenScript
**用于 LLM/agent 工作流的安全网关 SDK。** 阻断 prompt 注入，脱敏 PII 和机密信息，为 tool 调用设置防火墙，通过人工审批机制对高风险操作进行管控，并评估每一次操作的风险——适用于任何 LLM 提供商。
OpenScript 将任何 agent 包装在 **policy pipeline** 中：policy 在每次操作前后运行，且每一个都可以执行允许、修改（脱敏）、拒绝或要求人工审批。pipeline 本身不包含任何检测逻辑——所有的一切都是一个 `Policy`，你可以随意替换它们、通过 YAML 进行配置，或者自己编写。
## 5 分钟入门
### 1. 安装
```
pip install openscript
```
或从源码安装：
```
git clone https://github.com/OrdinalScale/openscript.git
cd openscript
python -m venv .venv
.venv/Scripts/activate # Windows
# source .venv/bin/activate # Unix
pip install -r requirements.txt
pip install -e .
```
可选的额外依赖：`openscript[redis]`（共享审批存储）、`openscript[metrics]`（Prometheus）、`openscript[otel]`（链路追踪）、`openscript[ml]`（基于 embedding 的检查）。
### 2. 包装 Agent
```
import asyncio
from sdk import PIIPolicy, PromptInjectionPolicy, SecretsPolicy, SecureAgent
class MyAgent:
async def ainvoke(self, input_data, **kwargs):
return {"output": f"Hello, {input_data.get('input', 'world')}!"}
async def main():
secure = SecureAgent(
agent=MyAgent(),
policies=[
PromptInjectionPolicy(threshold=0.5), # blocks injection attempts
PIIPolicy(mode="redact"), # redacts PII from output
SecretsPolicy(mode="redact"), # redacts credentials, flags internal URLs
],
)
result = await secure.invoke({"input": "OpenScript"})
print(result) # {"output": "Hello, OpenScript!"}
asyncio.run(main())
```
被阻断的操作会引发 `ActionBlockedError`，并附带原因、执行阻断的 policy，以及该操作汇总后的 `risk_score`。
### 3. 或者通过 YAML 配置 Policy
```
# policies.yaml
policies:
prompt_injection:
threshold: 0.6
toxicity:
threshold: 0.5
pii:
mode: redact
secrets:
mode: deny
internal_url_mode: annotate
compliance:
rules: [phi_detection, credential_output_guard]
tool_firewall:
rules_path: tools.yaml
```
```
from sdk import SecureAgent, load_policies
secure = SecureAgent(agent, policies=load_policies("policies.yaml"))
```
## 内置 Policy
| Policy | 阶段 | 功能描述 |
|--------|-------|--------------|
| `PromptInjectionPolicy` | input | 评估角色注入、prompt 提取、目标劫持、分隔符/间接注入；在达到阈值时拒绝 |
| `ToxicityPolicy` | input | 检测威胁、仇恨言论、骚扰、自残内容；在达到阈值时拒绝 |
| `PIIPolicy` | output | 脱敏或拒绝邮箱、电话、社会安全号码（SSN）、信用卡（经 Luhn 校验）、API key、IP |
| `SecretsPolicy` | input + output | 脱敏或拒绝 AWS/GitHub/Slack token、JWT、私钥块；单独标记内部 URL/私有 IP（`internal_url_mode`：默认为 annotate，并提供 allowlist） |
| `CompliancePolicy` | input + output | 严格界定范围的预设：`phi_detection`、`credential_output_guard`、`data_access_audit` — 参见 [合规定位](#compliance-positioning) |
| `ToolFirewallPolicy` | input | 针对 tool 调用的 allowlist/deny/RBAC/参数约束；可要求人工审批。也可通过 `validate_tool_call()` 或 `POST /v1/tools/validate` 独立使用 |
| `OutputSchemaPolicy` | output | Pydantic schema 校验，危险内容扫描，针对来源的可选幻觉/接地检查 |
| `AuditPolicy` | both | 将每次操作写入事件存储；请将其放在**最后**，以确保其事件包含最终的风险评分 |
每个 policy 都会写入标准化的元数据 —— `{"risk": float, "category": str, ...}` —— 内置的 `RiskScorer` 会将其汇总为每次操作的单一 `risk_score`：
```
result, ctx = await secure.invoke_with_context({"input": "..."})
print(ctx.risk_score) # 0.0 – 1.0
print(ctx.risk_categories) # {"pii": 0.4, "prompt_injection": 0.0, ...}
```
## 人工审批（审批后重试）
当 policy 返回 `REQUIRE_APPROVAL`（例如被防火墙拦截的 tool 调用）时，该操作会被阻断，并创建一条待处理的审批记录：
```
from sdk import ActionBlockedError, RedisApprovalStore, SecureAgent
secure = SecureAgent(
agent,
policies=[...],
# Redis is REQUIRED when approvals are decided via the server API —
# the default in-memory store only works within a single process.
approval_store=RedisApprovalStore("redis://localhost:6379/0"),
)
try:
await secure.invoke({"input": "transfer $5,000"})
except ActionBlockedError as e:
approval_id = e.approval_id # a human decides via POST /v1/approvals/{id}/decide
# 批准后，使用 approval id 重试相同的操作：
result = await secure.invoke({"input": "transfer $5,000"}, approval_id=approval_id)
```
审批是**一次性的**，在 1 小时后过期，且绑定到具体的操作 + 输入哈希值——为某次转账授予的审批无法被重用于另一次转账。
## 流式传输
`stream()` 支持三种保护模式（通过构造函数或单次调用设置 `stream_output=`）：
| 模式 | 保护机制 | 延迟 | 用途 |
|------|-----------|---------|---------|
| `buffer`（默认） | 全面—— output policy 可以在输出任何内容之前，查看、脱敏并阻断完整的响应 | 完整响应时间 | 供机器消费的输出 |
| `guarded` | 通过带有保留窗口的增量扫描，对有边界的模式（机密信息、PII）进行全面保护；若触发拒绝（deny），则会中止且不输出任何匹配内容。Schema/接地检查在流结束时运行 | 约一个窗口（几十个 token） | 面向人类的聊天 UI（文本流） |
| `passthrough` | **无**—— 分块直接交付，不进行扫描；policy 仅在事后运行以提供元数据/审计，事后的拒绝操作只会在事后引发异常 | 无 | 仅用于可观测性设置，请谨慎使用 |
```
async for chunk in secure.stream({"input": "..."}, stream_output="guarded"):
print(chunk, end="")
```
自定义 policy 可以通过实现 `stream_guard()` 来参与受保护的流式传输（参见 `contracts.interceptor.StreamGuard`）。
## 可观测性
```
from sdk import MetricsRecorder, SecureAgent
secure = SecureAgent(agent, policies=[...], metrics=MetricsRecorder())
```
Prometheus 指标（`pip install openscript[metrics]`）：按决策分类的操作、`risk_score` 直方图、按类别的违规计数器、注入/ tool 拒绝/ PII 脱敏计数器，以及按 policy 划分的延迟直方图。服务器通过 `GET /metrics` 暴露这些指标（受 API key 控制）。
OpenTelemetry（`pip install openscript[otel]`）：设置 `OPENSCRIPT_OTEL=1`，即可为每个操作生成一个 span，其中包含最终决策和风险评分；通过标准的 `OTEL_*` 环境变量配置你的 OTLP exporter。
## 编写自定义 Policy
```
from contracts.types import ActionContext, InterceptorDecision
from sdk import BasePolicy
class BusinessHoursPolicy(BasePolicy):
async def before_action(self, context: ActionContext) -> ActionContext:
if not is_business_hours():
context.decision = InterceptorDecision.DENY
context.decision_reason = "agent actions are restricted to business hours"
return context
```
任何包含 `before_action`、`after_action` 和 `failure_mode` 的对象都满足 `Policy` 协议——继承 `BasePolicy` 只是为你提供了直通的默认值。声明 `failure_mode` 可以控制错误处理方式：
| 模式 | 行为 |
|------|----------|
| `FAIL_OPEN` | 记录警告日志，允许操作继续执行 |
| `FAIL_CLOSED` | 记录错误日志，阻断操作 |
| `FAIL_EXCEPTION` | 重新抛出原始异常 |
安全 policy 默认使用 `FAIL_CLOSED`；可观测性 policy 默认使用 `FAIL_OPEN`。
## 架构
```
User Request
│
▼
┌────────────────────────────────────────────┐
│ SecureAgent │
│ │
│ ┌─ before_action ─────────────────────┐ │
│ │ PromptInjectionPolicy ──► DENY? │ │
│ │ ToxicityPolicy ──► DENY? │ │
│ │ ToolFirewallPolicy ──► APPROVAL? │ │
│ └─────────────────────────────────────┘ │
│ │ │
│ Agent.invoke() / stream() │
│ │ │
│ ┌─ after_action ──────────────────────┐ │
│ │ OutputSchemaPolicy ──► DENY? │ │
│ │ PIIPolicy / SecretsPolicy (redact) │ │
│ │ CompliancePolicy │ │
│ │ AuditPolicy (events + risk) │ │
│ └─────────────────────────────────────┘ │
│ │ │
│ RiskScorer ──► risk_score, metrics │
└────────────────────────────────────────────┘
│
▼
Response (or ActionBlockedError with
reason, risk_score, approval_id)
```
pipeline 是刻意设计得简单的——所有的检测逻辑都位于 policy 中。在 *before* 阶段拒绝（deny）会在 agent 运行前阻断；在 *after* 阶段拒绝会在所有 policy（包括审计）完成后阻断响应。
## 框架集成
目前已提供 LangChain 和 LangGraph 的包装器：
```
from sdk import wrap_agent, wrap_graph_agent, load_policies
secure = wrap_agent(langchain_agent, policies=load_policies("policies.yaml"))
result = await secure.invoke({"input": "What is prompt injection?"})
```
CrewAI、PydanticAI、AutoGen 和 OpenAI Agents SDK 适配器已列入路线图（每一个都将作为可选的额外依赖，基于各框架官方的防护机制/回调钩子构建）。
## 合规定位
`CompliancePolicy` **协助**合规计划——它的检查项（PHI 标识符检测、凭据输出防护、数据访问审计）对应了常见的 GDPR/HIPAA/SOC 2 控制措施。它**不**代表授予或认证符合任何法规，并且其预设名称特意根据它们检查的内容来命名，而不是以法规名称命名。
## 服务器
可选的 FastAPI 服务器（`uvicorn server.app:app`）提供事件存储、SSE 推送、会话仪表板（`/dashboard/`）、无状态评分端点（`/v1/threat/score`、`/v1/tools/validate`）、审批队列（`/v1/approvals`）以及 Prometheus 指标（`/metrics`）。除 `/health` 外，所有端点都需要在请求头中提供 `X-API-KEY`（`OPENSCRIPT_API_KEY`）。
尝试端到端运行完整的 pipeline：
```
python demo/injection_demo.py
```
## 从 Interceptor API 迁移
1.0 版本之前的名称仍然有效，但会发出 `DeprecationWarning`：`OpenScriptMiddleware` → `SecureAgent`，`interceptors=` → `policies=`，`ThreatInterceptor` → `PromptInjectionPolicy`，`PIIInterceptor` → `PIIPolicy`，`EventWriterInterceptor` → `AuditPolicy`，`Interceptor` 协议 → `Policy`。
## 开发
```
# 设置
python -m venv .venv && .venv/Scripts/activate
pip install -r requirements.txt && pip install -e .
# 测试
pytest
# Lint + format + type-check
ruff check .
black .
mypy sdk/ contracts/
```
详情请参阅 [CONTRIBUTING.md](CONTRIBUTING.md)。
## 许可证
Apache 2.0 — 参见 [LICENSE](LICENSE)。
标签：Kerberos, 搜索引擎查询, 用户代理, 自定义请求头, 逆向工具