tenuo-ai/tenuo
GitHub: tenuo-ai/tenuo
Tenuo 是一款面向 AI Agent 的加密能力令牌引擎，通过任务级衰减委托和离线验证防止提示注入导致的越权操作。
Stars: 35 | Forks: 3
AI Agent 的能力令牌。
## 快速开始
```
from tenuo import configure, SigningKey, mint_sync, guard, Capability, Pattern
from tenuo.exceptions import AuthorizationDenied
# 1. 一次性设置：生成密钥并配置 Tenuo
configure(issuer_key=SigningKey.generate(), dev_mode=True, audit_log=False)
# 2. 保护函数 —— 除非 warrant 允许，否则调用将被阻止
@guard(tool="send_email")
def send_email(to: str) -> str:
return f"Sent to {to}"
# 3. 铸造一个仅允许发送至 @company.com 的 warrant
with mint_sync(Capability("send_email", to=Pattern("*@company.com"))):
print(send_email(to="alice@company.com")) # -> "Sent to alice@company.com"
try:
send_email(to="attacker@evil.com")
except AuthorizationDenied:
print("Blocked: attacker@evil.com") # -> "Blocked: attacker@evil.com"
```
Agent 可能遭受提示注入。授权层并不在意。Warrant 规定只能是 `*@company.com`。请求却说是 `attacker@evil.com`。**拒绝。**
当 `mint_sync` 代码块退出时，warrant 就消失了。无需清理，无需撤销——它直接过期。
## 为什么选择 Tenuo？
IAM 回答“你是谁？”Tenuo 回答“你现在能做什么？”
| 问题 | Tenuo 的回答 |
|---------|----------------|
| 静态 IAM 角色寿命长于任务 | Warrant 随任务过期 (TTL) |
| 权限宽泛，爆炸半径大 | 约束在每次委托时收窄 |
| 令牌可能被窃取和重放 | 占有证明将 warrant 绑定到密钥 |
| 中心化策略服务器增加延迟 | 离线验证，无网络调用 |
## 工作原理
Tenuo 实现了**减法委托**：链中的每一步只能减少权限，永远不能增加。
```
┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐
│ Control Plane │ │ Orchestrator │ │ Worker │
│ │ │ │ │ │
│ Issues root │────▶│ Attenuates │────▶│ Executes with │
│ warrant │ │ for task │ │ proof │
└──────────────────┘ └──────────────────┘ └──────────────────┘
Full scope --> Narrower --> Narrowest
```
1. **控制平面** 发布具有广泛能力的根 warrant
2. **编排器** 针对特定任务对其进行削减（作用域只能缩小）
3. **Worker** 证明其拥有绑定的密钥并执行
4. **Warrant 过期** —— 无需清理
## Tenuo 不是什么
- **不是沙箱** —— Tenuo 授权操作，但不隔离执行。请结合容器/沙箱/VM 进行纵深防御。
- **不是提示工程** —— 没有“请不要做坏事”的指令。这是加密强制执行，而非行为层面的。
- **不是 LLM 过滤器** —— 我们不解析模型输出。我们在执行时把关工具调用。
- **不是 IAM 的替代品** —— Tenuo *补充* IAM，在身份之上添加任务范围的、递减的能力。
## 关键特性
| 特性 | 描述 |
|---------|-------------|
| **离线验证** | 无网络调用，约 27μs |
| **持有者绑定** | 没有密钥，被盗令牌毫无用处 |
| **语义约束** | [11 种约束类型](https://tenuo.ai/constraints)，包括 `Subpath`、`UrlSafe`、`Shlex`、`CEL` —— 它们以目标系统的方式解析输入（[为什么这很重要](https://niyikiza.com/posts/cve-2025-66032/)） |
| **单调衰减** | 能力只能缩小，永不扩大 |
| **框架集成** | OpenAI, Google ADK, CrewAI, Temporal, LangChain, LangGraph, FastAPI, MCP, A2A, AutoGen |
## 集成
**OpenAI** —— 带 streaming TOCTOU 防御的直接 API 保护
```
from tenuo.openai import GuardBuilder, Subpath, UrlSafe, Shlex, Pattern
client = (GuardBuilder(openai.OpenAI())
.allow("read_file", path=Subpath("/data")) # Path traversal protection
.allow("fetch_url", url=UrlSafe()) # SSRF protection
.allow("run_command", cmd=Shlex(allow=["ls"])) # Shell injection protection
.allow("send_email", to=Pattern("*@company.com"))
.build())
# Prompt injection -> send_email(to="attacker@evil.com") -> DENIED
```
**LangChain / LangGraph**
```
from tenuo.langchain import guard_tools
from tenuo.langgraph import TenuoToolNode
protected = guard_tools([search_tool, email_tool]) # LangChain
graph.add_node("tools", TenuoToolNode([search, email])) # LangGraph
```
**MCP** —— Model Context Protocol 客户端和服务器验证
```
from tenuo.mcp import SecureMCPClient, MCPVerifier
# Client：自动将 warrant 证明注入 tool 参数
async with SecureMCPClient("python", ["server.py"]) as client:
async with mint(Capability("read_file", path=Subpath("/data"))):
result = await client.tools["read_file"](path="/data/file.txt")
# Server：在执行前离线验证 tool 约束
verifier = MCPVerifier(...)
@mcp.tool()
async def read_file(path: str, **kwargs) -> str:
clean = verifier.verify_or_raise("read_file", {"path": path, **kwargs})
return open(clean["path"]).read()
```
更多集成：Google ADK, CrewAI, A2A, Temporal, FastAPI
**Google ADK** ``` from tenuo.google_adk import GuardBuilder from tenuo.constraints import Subpath, UrlSafe guard = (GuardBuilder() .allow("read_file", path=Subpath("/data")) .allow("web_search", url=UrlSafe(allow_domains=["*.google.com"])) .build()) agent = Agent(name="assistant", before_tool_callback=guard.before_tool) ``` **CrewAI** —— 具有基于能力授权的多 Agent Crew ``` from tenuo.crewai import GuardBuilder guard = (GuardBuilder() .allow("search", query=Pattern("*")) .allow("write_file", path=Subpath("/workspace")) .build()) crew = guard.protect(my_crew) # All agents get enforced constraints ``` **A2A (Agent-to-Agent)** —— 基于 Warrant 的 Agent 间委托 ``` from tenuo.a2a import A2AServerBuilder server = A2AServerBuilder().name("Search Agent").url("https://...").key(my_key).trust(orchestrator_key).build() @server.skill("search", constraints={"url": UrlSafe}) async def search(query: str, url: str) -> dict: return await do_search(query, url) client = A2AClient("https://...") warrant = await client.request_warrant(signing_key=worker_key, capabilities={"search": {}}) result = await client.send_task(skill="search", warrant=warrant, signing_key=worker_key) ``` **Temporal** —— 具有基于 warrant 活动授权的持久工作流 ``` from tenuo.temporal import AuthorizedWorkflow, TenuoInterceptor, TenuoInterceptorConfig @workflow.defn class MyWorkflow(AuthorizedWorkflow): @workflow.run async def run(self, path: str) -> str: return await self.execute_authorized_activity( read_file, args=[path], start_to_close_timeout=timedelta(seconds=30), ) ``` 查看完整的 Temporal 示例：[`authorized_workflow_demo.py`](tenuo-python/examples/temporal/authorized_workflow_demo.py) | [`multi_warrant.py`](tenuo-python/examples/temporal/multi_warrant.py) | [`delegation.py`](tenuo-python/examples/temporal/delegation.py) **FastAPI** —— 从 Header 提取 warrant，离线验证 PoP ``` @app.get("/search") async def search(query: str, ctx: SecurityContext = Depends(TenuoGuard("search"))): return {"results": do_search(query)} ``` **Kubernetes** —— 参见 [Kubernetes 指南](https://tenuo.ai/kubernetes)标签：Capabilty-based Security, JSONLines, PE 加载器, Python, Rust, Streamlit, 云端原生, 人工智能代理, 令牌衰减, 任务范围授权, 可视化界面, 委托授权, 子域名突变, 密码学, 工具调用控制, 微服务安全, 手动系统调用, 授权框架, 提示词注入防御, 无后门, 最小权限原则, 权限管理, 模型越狱, 离线验证, 网络流量审计, 能力令牌, 访问控制, 请求拦截, 软件供应链安全, 远程方法调用, 逆向工具, 通知系统, 零信任