Coverage for agentos/system/permissions.py: 49%
118 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 12:29 +0800
1"""
2分层权限系统 — 让 Agent 操作系统的每一步都在受控范围内。
4权限层级设计理念:
5- 参考 Android 权限模型的分级思想
6- 默认最小权限原则 (Principle of Least Privilege)
7- 高风险操作需二次确认
8- 支持会话级/全局级权限配置
9"""
11from __future__ import annotations
13from collections.abc import Callable
14from dataclasses import dataclass, field
15from enum import IntEnum
17# ── 权限层级定义 ──────────────────────────────────────────────
20class PermissionTier(IntEnum):
21 """权限层级，数值越大权限越高。"""
23 READ = 0 # 只读访问
24 WRITE_SANDBOX = 1 # 沙箱写入
25 WRITE_ALL = 2 # 全盘写入
26 SHELL_READONLY = 3 # 只读 Shell
27 SHELL_STANDARD = 4 # 标准 Shell（超时/目录限制）
28 SHELL_FULL = 5 # 全权限 Shell
29 BROWSER = 6 # 浏览器自动化
30 ADMIN = 7 # 系统管理
32 @property
33 def label(self) -> str:
34 """中文标签，用于 UI 展示。"""
35 return {
36 0: "只读访问",
37 1: "沙箱写入",
38 2: "全盘写入",
39 3: "只读Shell",
40 4: "标准Shell",
41 5: "全权限Shell",
42 6: "浏览器自动化",
43 7: "系统管理",
44 }.get(self.value, "未知")
47@dataclass
48class SystemPermission:
49 """单个系统权限定义。"""
51 tier: PermissionTier
52 resource: str # 资源标识，如 "/home/user/*", "apt:install"
53 description: str = ""
54 requires_confirmation: bool = False # 是否需要用户二次确认
55 rate_limit_per_minute: int = 0 # 0 表示不限
58# ── 预设权限策略 ──────────────────────────────────────────────
60# 安全模式（默认）: 允许读写但 Shell 受限
61SAFE_PERMISSIONS: list[SystemPermission] = [
62 SystemPermission(PermissionTier.READ, "*", "读取任意文件"),
63 SystemPermission(PermissionTier.WRITE_SANDBOX, "/tmp/agentos/**", "沙箱写入"),
64 SystemPermission(
65 PermissionTier.SHELL_READONLY,
66 "ls,cat,head,tail,find,ps,df,du,whoami,pwd,env,echo,date,wc,stat,file,which,uname",
67 "只读 Shell 命令",
68 ),
69 SystemPermission(PermissionTier.BROWSER, "*", "浏览器自动化", requires_confirmation=True),
70]
72# 开发模式: 允许写 + 标准 Shell
73DEV_PERMISSIONS: list[SystemPermission] = SAFE_PERMISSIONS + [
74 SystemPermission(PermissionTier.WRITE_ALL, "/home/**", "用户目录读写"),
75 SystemPermission(PermissionTier.SHELL_STANDARD, "*", "标准 Shell（超时60s，沙箱目录）"),
76 SystemPermission(PermissionTier.BROWSER, "*", "浏览器自动化"),
77]
79# 全权限模式: 无限制（高风险，仅限受信环境）
80FULL_PERMISSIONS: list[SystemPermission] = [
81 SystemPermission(PermissionTier.ADMIN, "*", "完全权限"),
82]
85# ── 权限上下文 ─────────────────────────────────────────────────
88@dataclass
89class PermissionContext:
90 """权限检查的上下文信息。"""
92 session_id: str
93 agent_id: str = ""
94 user_id: str = ""
95 tier: PermissionTier = PermissionTier.READ
96 granted_permissions: list[SystemPermission] = field(
97 default_factory=lambda: SAFE_PERMISSIONS.copy()
98 )
99 # 回调：当需要用户确认时触发
100 on_confirm_needed: Callable[[SystemPermission, str], bool] | None = None
103class PermissionDenied(Exception):
104 """权限拒绝异常。"""
106 def __init__(self, required: PermissionTier, resource: str, detail: str = ""):
107 self.required = required
108 self.resource = resource
109 self.detail = detail
110 super().__init__(
111 f"权限不足: 需要 {required.name} 访问 {resource}{' — ' + detail if detail else ''}"
112 )
115# ── 权限管理器 ─────────────────────────────────────────────────
118class SystemPermissionManager:
119 """系统权限管理器 — 检查、授权、升级、审计。"""
121 def __init__(self, default_tier: PermissionTier = PermissionTier.READ):
122 self._contexts: dict[str, PermissionContext] = {}
123 self._default_tier = default_tier
125 # ── 会话管理 ──
127 def create_session(
128 self,
129 session_id: str,
130 tier: PermissionTier | None = None,
131 permissions: list[SystemPermission] | None = None,
132 ) -> PermissionContext:
133 """创建权限会话。"""
134 ctx = PermissionContext(
135 session_id=session_id,
136 tier=tier or self._default_tier,
137 granted_permissions=permissions or SAFE_PERMISSIONS.copy(),
138 )
139 self._contexts[session_id] = ctx
140 return ctx
142 def get_session(self, session_id: str) -> PermissionContext:
143 """获取会话，不存在则创建默认会话。"""
144 if session_id not in self._contexts:
145 return self.create_session(session_id)
146 return self._contexts[session_id]
148 def close_session(self, session_id: str) -> None:
149 self._contexts.pop(session_id, None)
151 # ── 权限检查 ──
153 def check(self, session_id: str, required_tier: PermissionTier, resource: str) -> bool:
154 """检查是否拥有指定资源的权限。"""
155 ctx = self.get_session(session_id)
157 # 检查是否有匹配的权限
158 for perm in ctx.granted_permissions:
159 if self._tier_covers(perm.tier, required_tier) and self._resource_matches(
160 perm.resource, resource
161 ):
162 # 需要确认则触发回调
163 if perm.requires_confirmation:
164 if ctx.on_confirm_needed and not ctx.on_confirm_needed(perm, resource):
165 return False
166 return True
167 return False
169 def require(self, session_id: str, required_tier: PermissionTier, resource: str) -> None:
170 """要求权限，不满足则抛出 PermissionDenied。"""
171 if not self.check(session_id, required_tier, resource):
172 ctx = self.get_session(session_id)
173 current_max = max(
174 (p.tier for p in ctx.granted_permissions), default=PermissionTier.READ
175 )
176 raise PermissionDenied(
177 required_tier,
178 resource,
179 f"当前最高权限: {current_max.name}，需要: {required_tier.name}",
180 )
182 # ── 权限升级 ──
184 def elevate(
185 self,
186 session_id: str,
187 tier: PermissionTier,
188 permissions: list[SystemPermission] | None = None,
189 require_user_approval: bool = True,
190 ) -> bool:
191 """升级会话权限级别。"""
192 ctx = self.get_session(session_id)
194 if require_user_approval:
195 # 触发用户确认流程
196 if ctx.on_confirm_needed:
197 dummy_perm = SystemPermission(tier, "*", f"升级到 {tier.name}")
198 if not ctx.on_confirm_needed(dummy_perm, "elevate"):
199 return False
201 ctx.tier = tier
202 if permissions:
203 ctx.granted_permissions = permissions
204 return True
206 # ── 临时提权（供 ApprovalEngine 审批通过后调用）──
208 def escalate(self, session_id: str, tier: PermissionTier, resource: str) -> None:
209 """单次临时提权 — 审批通过后临时授予某资源访问权，本会话有效。
211 与 elevate 不同：escalate 是细粒度的、单资源、可撤销的临时授权；
212 elevate 是整体层级提升。
213 """
214 ctx = self.get_session(session_id)
215 # 添加临时权限（不持久化，仅本会话）
216 temp_perm = SystemPermission(
217 tier=tier,
218 resource=resource,
219 description=f"临时授权: {tier.label} → {resource}",
220 requires_confirmation=False, # 已经审批过，无需二次确认
221 )
222 ctx.granted_permissions.append(temp_perm)
224 def revoke_escalation(self, session_id: str, resource: str) -> None:
225 """撤销某资源的临时提权。"""
226 ctx = self.get_session(session_id)
227 ctx.granted_permissions = [
228 p
229 for p in ctx.granted_permissions
230 if not (p.description.startswith("临时授权:") and p.resource == resource)
231 ]
233 # ── 预设模式快捷切换 ──
235 def set_safe_mode(self, session_id: str) -> None:
236 """切换到安全模式。"""
237 ctx = self.get_session(session_id)
238 ctx.granted_permissions = SAFE_PERMISSIONS.copy()
239 ctx.tier = PermissionTier.WRITE_SANDBOX
241 def set_dev_mode(self, session_id: str) -> None:
242 """切换到开发模式。"""
243 ctx = self.get_session(session_id)
244 ctx.granted_permissions = DEV_PERMISSIONS.copy()
245 ctx.tier = PermissionTier.SHELL_STANDARD
247 def set_full_mode(self, session_id: str) -> None:
248 """切换到全权限模式（需确认）。"""
249 ctx = self.get_session(session_id)
250 ctx.granted_permissions = FULL_PERMISSIONS.copy()
251 ctx.tier = PermissionTier.ADMIN
253 # ── 辅助方法 ──
255 @staticmethod
256 def _tier_covers(granted: PermissionTier, required: PermissionTier) -> bool:
257 """检查授权层级是否覆盖需求层级。"""
258 return granted.value >= required.value
260 @staticmethod
261 def _resource_matches(pattern: str, resource: str) -> bool:
262 """简单的资源匹配（支持 * 通配符）。"""
263 if pattern == "*":
264 return True
265 # 支持 ** 递归匹配
266 if "**" in pattern:
267 prefix = pattern.replace("**", "")
268 return resource.startswith(prefix)
269 # 支持 * 单层匹配
270 if "*" in pattern:
271 import fnmatch
273 return fnmatch.fnmatch(resource, pattern)
274 return resource == pattern or resource.startswith(pattern)
277# ── Auto-generated compat stubs ──
280class SAFE_PERMISSIONS:
281 pass
284class DEV_PERMISSIONS:
285 pass
288class FULL_PERMISSIONS:
289 pass