摘要

当你的 OpenClaw 从个人工具变成团队平台，权限控制就成了刚需——谁能调用哪些工具？谁能管理哪些渠道？谁能查看哪些会话？本文从 RBAC（基于角色的访问控制）模型出发，系统讲解 OpenClaw 的权限架构设计——包括角色定义、权限粒度、资源级隔离、策略引擎实现。通过两个完整实战案例（团队协作权限体系和多租户 SaaS 权限隔离），你将掌握从零搭建生产级权限系统的全部技能。读完你会发现：权限不是"加个 if 判断"，而是一套完整的策略引擎。

---

1. 引言：当"都能做"变成"谁该做什么"

1.1 权限缺失的代价

先看一个真实场景。

你的团队有5个人在用 OpenClaw：

• 小王（开发）：用 Agent 写代码、查文档
• 小李（运维）：用 Agent 管理服务器、查看日志
• 老张（产品）：用 Agent 查数据、生成报告
• 你（管理员）：管理整个平台

一开始，大家都用同一个 API Key，所有工具都能调用。直到有一天——

小李不小心调用了"发送全员通知"的工具，凌晨3点给全公司发了条测试消息。老张好奇点了"服务器重启"，把生产环境搞挂了。

问题根源：没有权限控制，每个人都能做任何事。

1.2 RBAC 核心概念

概念	说明	示例
用户（User）	使用系统的个体	小王、小李
角色（Role）	权限的集合	开发者、运维
权限（Permission）	对资源的操作许可	tool:exec、channel:feishu
资源（Resource）	被保护的对象	工具、渠道、会话
策略（Policy）	权限判断规则	“开发者可以调用 exec 但不能管理用户”

---

2. 权限模型设计

2.1 权限粒度定义

# permissions.yaml - 权限定义
permissions:
  # ===== 工具权限 =====
  tools:
    exec:
      description: "执行系统命令"
      risk_level: high
      requires_approval: true
    read:
      description: "读取文件"
      risk_level: low
    write:
      description: "写入文件"
      risk_level: medium
    browser:
      description: "浏览器操作"
      risk_level: medium
    message:
      description: "发送消息"
      risk_level: medium
      scoped: true  # 需要渠道范围限制
  
  # ===== 渠道权限 =====
  channels:
    feishu:
      description: "飞书渠道管理"
      operations: [read, send, configure]
    wecom:
      description: "企业微信渠道管理"
      operations: [read, send, configure]
    telegram:
      description: "Telegram 渠道管理"
      operations: [read, send, configure]
  
  # ===== 管理权限 =====
  admin:
    users:
      description: "用户管理"
      operations: [list, create, update, delete]
    roles:
      description: "角色管理"
      operations: [list, create, update, delete]
    audit:
      description: "审计日志查看"
      operations: [read, export]
    config:
      description: "系统配置管理"
      operations: [read, update]

2.2 角色定义

# roles.yaml - 角色与权限绑定
roles:
  # ===== 预置角色 =====
  admin:
    description: "平台管理员——拥有所有权限"
    permissions:
      - "tools:*"          # 所有工具
      - "channels:*"       # 所有渠道
      - "admin:*"          # 所有管理功能
    is_system: true
  
  developer:
    description: "开发者——代码和文件操作"
    permissions:
      - "tools:exec"
      - "tools:read"
      - "tools:write"
      - "tools:browser"
      - "tools:message{channel:dev-*}"  # 仅限 dev- 开头的渠道
    restrictions:
      exec:
        blocked_commands: ["rm -rf", "shutdown", "reboot"]
        require_confirmation: true
  
  operator:
    description: "运维人员——服务器管理和监控"
    permissions:
      - "tools:exec"
      - "tools:read"
      - "tools:message{channel:ops-*}"
      - "admin:audit:read"
    restrictions:
      exec:
        allowed_paths: ["/var/log/", "/etc/openclaw/", "/data/"]
        require_confirmation: false
  
  analyst:
    description: "数据分析师——只读查询和导出"
    permissions:
      - "tools:read"
      - "tools:search"
      - "tools:export"
    restrictions:
      export:
        max_rows: 10000
        require_approval: true
  
  viewer:
    description: "只读观察者——仅查看"
    permissions:
      - "tools:read"
    restrictions:
      read:
        allowed_paths: ["/data/reports/", "/data/dashboards/"]

---

3. 策略引擎实现

3.1 核心引擎代码

"""
policy_engine.py
RBAC 策略引擎——权限判断的核心

设计原则：
1. 策略与代码分离（YAML 定义，引擎执行）
2. 支持通配符匹配（tools:* 匹配所有工具）
3. 支持条件权限（仅限特定渠道/路径）
4. 支持审批流程（高风险操作需要二次确认）
5. 拒绝优先（deny > allow）
"""

import re
import yaml
from typing import Dict, List, Optional, Set, Any
from dataclasses import dataclass, field
from enum import Enum


class Decision(Enum):
    ALLOW = "allow"
    DENY = "deny"
    REQUIRE_APPROVAL = "require_approval"


@dataclass
class PermissionCheck:
    """一次权限检查的上下文"""
    user_id: str
    action: str           # 如 "tool:exec"
    resource: str         # 如 "/etc/openclaw/config.yaml"
    channel: Optional[str] = None
    metadata: Dict = field(default_factory=dict)


class PolicyEngine:
    """
    RBAC 策略引擎
    
    核心流程：
    1. 查找用户的所有角色
    2. 合并角色权限（取并集）
    3. 检查是否有拒绝规则（deny 优先）
    4. 检查是否需要审批
    5. 返回决策
    """
    
    def __init__(self, roles_path: str, permissions_path: str):
        with open(roles_path) as f:
            self.roles = yaml.safe_load(f)["roles"]
        with open(permissions_path) as f:
            self.permissions = yaml.safe_load(f)["permissions"]
        
        # 用户→角色映射（生产环境从数据库读取）
        self.user_roles: Dict[str, List[str]] = {}
    
    def assign_role(self, user_id: str, role_name: str):
        """为用户分配角色"""
        if role_name not in self.roles:
            raise ValueError(f"角色不存在: {role_name}")
        if user_id not in self.user_roles:
            self.user_roles[user_id] = []
        if role_name not in self.user_roles[user_id]:
            self.user_roles[user_id].append(role_name)
    
    def revoke_role(self, user_id: str, role_name: str):
        """撤销用户角色"""
        if user_id in self.user_roles:
            self.user_roles[user_id] = [
                r for r in self.user_roles[user_id] if r != role_name
            ]
    
    def get_user_permissions(self, user_id: str) -> Set[str]:
        """获取用户的所有权限（合并所有角色）"""
        permissions = set()
        role_names = self.user_roles.get(user_id, [])
        
        for role_name in role_names:
            role = self.roles.get(role_name)
            if role:
                permissions.update(role.get("permissions", []))
        
        return permissions
    
    def check(self, check: PermissionCheck) -> Dict[str, Any]:
        """
        权限检查主入口
        
        返回决策和详细信息
        """
        # 1. 获取用户权限
        user_perms = self.get_user_permissions(check.user_id)
        
        # 2. 检查是否有匹配的权限
        matched = self._match_permission(check.action, user_perms)
        
        if not matched:
            return {
                "decision": Decision.DENY.value,
                "reason": f"用户 {check.user_id} 没有 {check.action} 权限",
                "user_roles": self.user_roles.get(check.user_id, [])
            }
        
        # 3. 检查条件限制
        restrictions = self._get_restrictions(check.user_id, check.action)
        if restrictions:
            condition_result = self._check_conditions(check, restrictions)
            if not condition_result["allowed"]:
                return {
                    "decision": Decision.DENY.value,
                    "reason": condition_result["reason"],
                    "restriction": restrictions
                }
        
        # 4. 检查是否需要审批
        if self._requires_approval(check.action, check.user_id):
            return {
                "decision": Decision.REQUIRE_APPROVAL.value,
                "reason": f"{check.action} 需要管理员审批",
                "approval_required": True
            }
        
        return {
            "decision": Decision.ALLOW.value,
            "matched_permission": matched,
            "user_roles": self.user_roles.get(check.user_id, [])
        }
    
    def _match_permission(self, action: str, permissions: Set[str]) -> Optional[str]:
        """
        权限匹配（支持通配符）
        
        匹配规则：
        - "tools:*" 匹配 "tools:exec", "tools:read" 等
        - "tools:exec" 精确匹配
        - "tools:message{channel:dev-*}" 带条件匹配
        """
        for perm in permissions:
            # 提取基础权限（去掉条件部分）
            base_perm = perm.split("{")[0] if "{" in perm else perm
            
            # 通配符匹配
            if base_perm.endswith(":*"):
                prefix = base_perm[:-2]
                if action.startswith(prefix):
                    return perm
            
            # 精确匹配
            if base_perm == action:
                return perm
        
        return None
    
    def _get_restrictions(self, user_id: str, action: str) -> Optional[Dict]:
        """获取用户对该操作的额外限制"""
        for role_name in self.user_roles.get(user_id, []):
            role = self.roles.get(role_name, {})
            restrictions = role.get("restrictions", {})
            
            # 查找匹配的限制
            for restrict_key, restrict_rules in restrictions.items():
                if action.startswith(restrict_key) or action == restrict_key:
                    return restrict_rules
        
        return None
    
    def _check_conditions(
        self, check: PermissionCheck, restrictions: Dict
    ) -> Dict:
        """检查条件限制是否满足"""
        # 检查命令黑名单
        if "blocked_commands" in restrictions:
            for cmd in restrictions["blocked_commands"]:
                if cmd in check.metadata.get("command", ""):
                    return {
                        "allowed": False,
                        "reason": f"命令 '{cmd}' 在禁止列表中"
                    }
        
        # 检查路径白名单
        if "allowed_paths" in restrictions:
            resource = check.resource
            allowed = any(
                resource.startswith(path)
                for path in restrictions["allowed_paths"]
            )
            if not allowed:
                return {
                    "allowed": False,
                    "reason": f"路径 '{resource}' 不在允许列表中"
                }
        
        return {"allowed": True}
    
    def _requires_approval(self, action: str, user_id: str) -> bool:
        """检查是否需要审批"""
        for role_name in self.user_roles.get(user_id, []):
            role = self.roles.get(role_name, {})
            restrictions = role.get("restrictions", {})
            
            for restrict_key, rules in restrictions.items():
                if action.startswith(restrict_key):
                    if rules.get("require_confirmation", False):
                        return True
                    if rules.get("require_approval", False):
                        return True
        
        return False


# ============================================
# 使用示例
# ============================================

if __name__ == "__main__":
    engine = PolicyEngine("roles.yaml", "permissions.yaml")
    
    # 分配角色
    engine.assign_role("wang", "developer")
    engine.assign_role("li", "operator")
    engine.assign_role("zhang", "analyst")
    engine.assign_role("admin", "admin")
    
    # 测试用例
    test_cases = [
        PermissionCheck("wang", "tool:exec", "/tmp/test.py",
                       metadata={"command": "python /tmp/test.py"}),
        PermissionCheck("wang", "tool:exec", "/tmp/test.py",
                       metadata={"command": "rm -rf /"}),
        PermissionCheck("li", "tool:exec", "/var/log/app.log",
                       metadata={"command": "tail -f /var/log/app.log"}),
        PermissionCheck("li", "tool:exec", "/home/user/secret.txt",
                       metadata={"command": "cat /home/user/secret.txt"}),
        PermissionCheck("zhang", "tool:exec", "/tmp/query.py",
                       metadata={"command": "python /tmp/query.py"}),
        PermissionCheck("admin", "admin:users:delete", "user_123"),
    ]
    
    for tc in test_cases:
        result = engine.check(tc)
        icon = {"allow": "✅", "deny": "❌", "require_approval": "⚠️"}
        print(f"{icon[result['decision']]} {tc.user_id} → {tc.action}")
        print(f"   决策: {result['decision']}")
        print(f"   原因: {result.get('reason', 'N/A')}")
        if "user_roles" in result:
            print(f"   角色: {result['user_roles']}")
        print()

预期输出：

✅ wang → tool:exec
   决策: allow
   角色: ['developer']

❌ wang → tool:exec
   决策: deny
   原因: 命令 'rm -rf' 在禁止列表中

✅ li → tool:exec
   决策: allow
   角色: ['operator']

❌ li → tool:exec
   决策: deny
   原因: 路径 '/home/user/secret.txt' 不在允许列表中

❌ zhang → tool:exec
   决策: deny
   原因: 用户 zhang 没有 tool:exec 权限

✅ admin → admin:users:delete
   决策: allow
   角色: ['admin']

📸 截图位置：策略引擎测试的终端输出，展示6种不同权限检查场景的结果。

---

4. 多租户权限隔离

4.1 租户模型设计

"""
tenant_isolation.py
多租户 SaaS 权限隔离

每个租户拥有独立的：
- 用户和角色
- 渠道配置
- 会话数据
- 工具白名单
"""

class TenantIsolation:
    """
    租户隔离管理器
    
    核心设计：
    1. 每个租户有独立的命名空间
    2. 租户间数据完全隔离
    3. 平台管理员可跨租户管理
    4. 租户管理员只能管理自己的租户
    """
    
    def __init__(self, policy_engine: PolicyEngine):
        self.engine = policy_engine
        self.tenants: Dict[str, Dict] = {}
    
    def create_tenant(self, tenant_id: str, admin_user_id: str):
        """创建租户并指定管理员"""
        self.tenants[tenant_id] = {
            "id": tenant_id,
            "admin": admin_user_id,
            "users": {admin_user_id},
            "channels": set(),
            "created_at": datetime.now().isoformat()
        }
        # 分配租户管理员角色
        self.engine.assign_role(admin_user_id, "tenant_admin")
    
    def add_user_to_tenant(self, tenant_id: str, user_id: str, role: str):
        """向租户添加用户"""
        if tenant_id not in self.tenants:
            raise ValueError(f"租户不存在: {tenant_id}")
        
        self.tenants[tenant_id]["users"].add(user_id)
        self.engine.assign_role(user_id, role)
    
    def check_tenant_access(
        self, user_id: str, tenant_id: str, action: str
    ) -> Dict:
        """
        租户级权限检查
        
        先检查用户是否属于该租户，再检查具体操作权限
        """
        # 平台管理员可以访问所有租户
        if "admin" in self.engine.user_roles.get(user_id, []):
            return {"decision": "allow", "reason": "平台管理员"}
        
        # 检查是否属于该租户
        tenant = self.tenants.get(tenant_id)
        if not tenant or user_id not in tenant["users"]:
            return {
                "decision": "deny",
                "reason": f"用户 {user_id} 不属于租户 {tenant_id}"
            }
        
        # 检查具体权限
        check = PermissionCheck(
            user_id=user_id,
            action=action,
            resource=f"tenant:{tenant_id}"
        )
        return self.engine.check(check)

4.2 租户隔离配置

# tenants.yaml
tenants:
  t-company-a:
    name: "A公司"
    admin: "user_admin_a"
    plan: "enterprise"
    limits:
      max_users: 100
      max_channels: 10
      max_daily_requests: 50000
    allowed_tools:
      - exec
      - read
      - write
      - browser
      - message
  
  t-company-b:
    name: "B公司"
    admin: "user_admin_b"
    plan: "professional"
    limits:
      max_users: 20
      max_channels: 3
      max_daily_requests: 10000
    allowed_tools:
      - read
      - write
      - message

---

5. 审批流程集成

5.1 高风险操作审批

"""
approval_workflow.py
高风险操作审批流程

触发条件：
- 操作风险等级为 high
- 用户角色配置了 require_approval
- 操作涉及生产环境
"""

from enum import Enum
from datetime import datetime
from typing import Dict, List, Optional


class ApprovalStatus(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"
    EXPIRED = "expired"


class ApprovalWorkflow:
    """审批工作流管理器"""
    
    def __init__(self, policy_engine):
        self.engine = policy_engine
        self.pending_approvals: Dict[str, Dict] = {}
    
    def request_approval(self, check: PermissionCheck) -> Dict:
        """发起审批请求"""
        approval_id = f"apr-{datetime.now().strftime('%Y%m%d%H%M%S')}-{check.user_id}"
        
        approval = {
            "id": approval_id,
            "user_id": check.user_id,
            "action": check.action,
            "resource": check.resource,
            "reason": f"用户 {check.user_id} 请求执行 {check.action}",
            "status": ApprovalStatus.PENDING.value,
            "created_at": datetime.now().isoformat(),
            "expires_at": (datetime.now() + timedelta(hours=1)).isoformat()
        }
        
        self.pending_approvals[approval_id] = approval
        
        # 通知管理员（实际生产环境会通过消息渠道发送）
        print(f"📋 新审批请求: {approval_id}")
        print(f"   用户: {check.user_id}")
        print(f"   操作: {check.action}")
        print(f"   资源: {check.resource}")
        
        return approval
    
    def approve(self, approval_id: str, approver_id: str) -> Dict:
        """批准审批请求"""
        approval = self.pending_approvals.get(approval_id)
        if not approval:
            return {"error": "审批请求不存在"}
        
        approval["status"] = ApprovalStatus.APPROVED.value
        approval["approver_id"] = approver_id
        approval["approved_at"] = datetime.now().isoformat()
        
        print(f"✅ 审批通过: {approval_id} (审批人: {approver_id})")
        return approval
    
    def reject(self, approval_id: str, approver_id: str, reason: str) -> Dict:
        """拒绝审批请求"""
        approval = self.pending_approvals.get(approval_id)
        if not approval:
            return {"error": "审批请求不存在"}
        
        approval["status"] = ApprovalStatus.REJECTED.value
        approval["approver_id"] = approver_id
        approval["reject_reason"] = reason
        
        print(f"❌ 审批拒绝: {approval_id} (原因: {reason})")
        return approval

---

6. 总结

核心要点：

1. RBAC 三角模型：用户 → 角色 → 权限，角色是权限的集合，用户通过角色获得权限

2. 策略引擎三要素：通配符匹配（tools:*）、条件限制（路径白名单/命令黑名单）、审批流程（高风险操作二次确认）

3. 拒绝优先原则：deny > allow，即使有 allow 规则，deny 规则也会覆盖

4. 多租户隔离：租户级命名空间 + 用户归属检查 + 平台管理员跨租户能力

5. 审批工作流：高风险操作自动触发审批，1小时过期，支持批准/拒绝

思考题：

1. 如果你的权限系统需要支持"临时提权"——比如开发者需要临时获得运维权限处理紧急故障，你会如何设计这个机制？如何确保临时权限到期自动回收？

2. RBAC 的局限性之一是"角色爆炸"——随着业务增长，角色数量可能膨胀到难以管理。ABAC（基于属性的访问控制）如何解决这个问题？你会在什么场景下从 RBAC 升级到 ABAC？

3. 权限策略引擎的性能至关重要——每次工具调用都要检查权限。如果用户有20个角色，每个角色有50条权限，你会如何优化匹配算法避免成为性能瓶颈？

---

参考资料

• NIST RBAC 标准 (ANSI/INCITS 359-2004)
• AWS IAM 策略评估逻辑
• OWASP 访问控制最佳实践
• OpenClaw 安全配置文档
• Kubernetes RBAC 文档
• Casbin - Go 语言权限管理库