把 Agent 接入企业 IM：从权限校验到指令误触发的防护

关键词

企业智能代理、IM 集成、权限管理、指令安全、误触发防护、访问控制、身份认证、多因素认证、自然语言处理安全

摘要

本文深入探讨了将智能代理（Agent）接入企业即时通讯（IM）系统的完整技术路径，重点聚焦于权限校验机制与指令误触发防护两大核心安全挑战。我们从第一性原理出发，构建了企业级 Agent-IM 集成的理论框架，提供了系统化的架构设计方案，并通过实际代码实现展示了关键技术点。本文不仅涵盖了基础的身份认证与授权模型，还深入探讨了上下文感知权限系统、指令意图验证、多模态安全检查等高级主题，为企业构建安全、可靠的 Agent-IM 集成系统提供了全面指导。

1. 概念基础

1.1 领域背景化

核心概念

企业 IM（即时通讯）系统是企业内部协作的核心基础设施，而智能 Agent 则是能够自主执行任务的软件实体。将两者结合，意味着赋予企业员工通过自然语言与企业系统交互的能力，极大提升工作效率。

问题背景

随着人工智能技术的快速发展，越来越多的企业开始探索将智能 Agent 集成到其现有的 IM 平台（如 Slack、Microsoft Teams、企业微信、钉钉等）中。这种集成可以让员工通过自然语言指令查询数据、触发工作流、执行操作，极大提升生产力。然而，这种便利也带来了前所未有的安全挑战：如何确保只有授权用户可以执行特定操作？如何防止误操作或恶意指令导致的安全事故？

问题描述

在没有适当安全措施的情况下，Agent-IM 集成可能面临以下关键问题：

1. 越权访问：低权限用户通过 Agent 执行高权限操作
2. 指令注入：恶意用户构造特殊指令绕过安全检查
3. 误触发：用户无意的表述被 Agent 误解析为执行指令
4. 会话劫持：攻击者窃取用户会话后通过 Agent 执行操作
5. 数据泄露：Agent 返回结果中包含未授权访问的敏感信息

问题解决

解决这些问题需要一个多层安全架构，涵盖身份认证、细粒度授权、指令验证、上下文感知安全决策等多个维度。本文将系统性地介绍这些解决方案的理论基础、设计原则和实现方法。

1.2 历史轨迹

为了更好地理解当前的解决方案，我们简要回顾一下企业 IM 安全与智能代理发展的历史轨迹：

这一演进过程表明，随着 IM 系统与智能代理技术的发展，安全挑战也在不断演变，从最初的基本访问控制发展到如今的复杂上下文感知安全机制。

1.3 问题空间定义

在深入探讨解决方案之前，我们需要精确界定 Agent-IM 集成系统中的安全问题空间。我们可以从以下几个维度来构建这个问题空间：

1. 身份维度：谁在发起请求？

   • 用户身份认证
   • 设备认证
   • 位置认证
   • 行为生物特征认证

2. 权限维度：发起者可以做什么？

   • 角色权限
   • 资源权限
   • 操作权限
   • 时间权限
   • 上下文权限

3. 意图维度：用户真正想要做什么？

   • 指令意图理解
   • 歧义消解
   • 上下文依赖解析
   • 隐含意图识别

4. 指令维度：指令本身是否安全？

   • 指令格式验证
   • 恶意指令检测
   • 指令范围限制
   • 敏感操作识别

5. 环境维度：执行环境是否安全？

   • 网络环境评估
   • 设备安全状态
   • 时间因素
   • 地理位置

通过这五个维度的分析，我们可以构建一个全面的问题空间模型，为后续设计解决方案提供清晰的指导。

1.4 术语精确性

在继续深入讨论之前，我们需要明确定义本文中使用的关键术语，以确保交流的精确性：

1. 智能代理（Agent）：一种能够感知环境、做出决策并执行行动的软件实体，在本文中特指可以通过自然语言交互的企业级智能系统。

2. 企业 IM：企业内部使用的即时通讯系统，如 Slack、Microsoft Teams、企业微信、钉钉等，提供消息传递、群组协作等功能。

3. 身份认证（Authentication）：验证用户或系统身份的过程，确认"你是谁"。

4. 授权（Authorization）：确定已认证用户可以执行哪些操作或访问哪些资源的过程，回答"你可以做什么"。

5. 权限校验（Permission Validation）：在执行操作前验证用户是否有足够权限的过程。

6. 指令误触发（False Trigger）：Agent 将用户无意的表述错误识别为需要执行的指令。

7. 上下文感知安全（Context-Aware Security）：根据用户、设备、位置、时间等上下文信息动态调整安全策略的机制。

8. 最小权限原则（Principle of Least Privilege）：安全设计原则，指用户只应拥有完成其工作所需的最小权限集合。

9. 意图识别（Intent Recognition）：理解用户输入背后真实意图的过程，通常通过自然语言处理技术实现。

10. 多因素认证（Multi-Factor Authentication, MFA）：要求用户提供两种或多种不同类型的认证因素才能访问系统的安全机制。

2. 理论框架

2.1 第一性原理推导

在构建 Agent-IM 集成安全框架时，我们从第一性原理出发，首先确定几个不可简化的基本公理：

公理 1：身份真实性公理
系统必须能够以足够高的可信度确认发起交互的实体身份。
$$P(I_{valid})\ge {\tau }_{identity}$$
其中，$P(I_{valid})$ 是身份验证的可信概率，${\tau }_{identity}$ 是组织设定的身份验证可信度阈值。

公理 2：权限最小化公理
任何实体在任何时间点只应拥有完成当前任务所必需的最小权限集合。
$$Permissions(t)=\bigcap Permission{s}_{tas{k}_i}(t)$$
其中，$Permissions(t)$ 是实体在时间 $t$ 的权限集合，$Permission{s}_{tas{k}_i}(t)$ 是完成任务 $tas{k}_i$ 所需的权限集合。

公理 3：指令确定性公理
在执行任何具有副作用的指令前，系统必须以足够高的确定性确认指令意图与用户真实意图一致。
$$P(I_{command}=I_{user})\ge {\tau }_{intent}$$
其中，$P(I_{command}=I_{user})$ 是系统理解的指令意图与用户真实意图一致的概率，${\tau }_{intent}$ 是组织设定的意图确认阈值。

公理 4：操作可追溯公理
所有通过 Agent 执行的操作必须完整记录，且记录不可篡改，以便事后审计。
$$\forall op\in Operations,(op,t,u,ctx)\in ImmutableLog$$
其中，$op$ 是操作，$t$ 是时间，$u$ 是用户，$ctx$ 是上下文信息，$ImmutableLog$ 是不可篡改日志。

从这四条公理出发，我们可以推导出 Agent-IM 集成安全框架的必要组件：

1. 从公理 1 推导出：强身份认证系统、会话管理机制
2. 从公理 2 推导出：细粒度授权系统、动态权限调整机制
3. 从公理 3 推导出：意图验证系统、误触发防护机制
4. 从公理 4 推导出：完整审计日志系统、操作追溯机制

2.2 数学形式化

为了更精确地描述 Agent-IM 集成系统的安全机制，我们引入以下数学模型：

2.2.1 身份认证模型

身份认证可以形式化为一个假设检验问题：

$$H_0:\text{声称的身份是真实的}$$
$$H_1:\text{声称的身份是伪造的}$$

认证系统根据收集到的证据 E={e1,e2,...,en}E = \{e_1, e_2, ..., e_n\}E={e1​,e2​,...,en​}（如密码、令牌、生物特征等）计算似然比：

$$LR(E)=\frac{P(E|H_0)}{P(E|H_1)}$$

然后将似然比与组织设定的阈值 $\theta$ 比较，做出认证决策：

$$\text{Decision}(E)=\{\begin{array}{ll}\text{接受}& \text{如果 }LR(E)\ge \theta \\ \text{拒绝}& \text{如果 }LR(E)<\theta \end{array}$$

在多因素认证场景下，我们可以使用贝叶斯网络来组合多个认证因素的证据：

$$P(H_0|E_1,E_2,...,E_n)=\frac{P(E_1,E_2,...,E_n|H_0)P(H_0)}{P(E_1,E_2,...,E_n)}$$

假设各认证因素条件独立，则：

$$P(H_0|E_1,E_2,...,E_n)=\frac{P(H_0){\prod }_{i=1}^{n}P(E_i|H_0)}{P(H_0){\prod }_{i=1}^{n}P(E_i|H_0)+P(H_1){\prod }_{i=1}^{n}P(E_i|H_1)}$$

2.2.2 权限控制模型

我们使用基于属性的访问控制（ABAC）模型，这是一种灵活的权限控制模型，非常适合动态的 Agent-IM 集成场景。

在 ABAC 模型中，访问控制决策基于主体属性（$S$）、客体属性（$O$）、操作属性（$A$）和环境属性（$E$）：

$$P=\{(s,o,a,e)|\text{Policy}(s,o,a,e)=\text{True}\}$$

其中，$P$ 是允许的访问集合，Policy 是访问控制策略函数。

为了支持细粒度和上下文感知的权限控制，我们定义策略规则为带有约束的逻辑表达式：

$$Rul{e}_i:{\text{Condition}}_i\to ({\text{Decision}}_i,{\text{Obligation}}_i)$$

其中，${\text{Condition}}_i$ 是基于属性的条件表达式，${\text{Decision}}_i$ 是允许或拒绝，${\text{Obligation}}_i$ 是决策后需要执行的义务（如记录日志）。

策略评估过程可以形式化为：

1. 收集相关属性：$Attr=S\cup O\cup A\cup E$
2. 评估所有规则，找到匹配规则集：Match={Rulei∣Conditioni(Attr)=True}Match = \{ Rule_i | Condition_i(Attr) = True \}Match={Rulei​∣Conditioni​(Attr)=True}
3. 使用冲突解决策略（如拒绝优先、优先级等）从 $Match$ 中得出最终决策

2.2.3 指令误触发防护模型

指令误触发防护可以建模为一个二分类问题，我们需要区分真实指令和非指令（或误触发）。

设 $X$ 为输入特征空间（包含用户输入、对话历史、用户属性等），Y={0,1}Y = \{0, 1\}Y={0,1} 为标签空间（0 表示非指令，1 表示真实指令）。

我们的目标是学习一个分类器 $f:X\to Y$，使得误分类风险最小：

$$R(f)=\int L(y,f(x))dP(x,y)$$

其中，$L(y,f(x))$ 是损失函数。考虑到误触发的代价通常高于漏检（即把真实指令误认为非指令），我们使用加权损失函数：

$$L(y,f(x))=\{\begin{array}{ll}{w}_{FP}& \text{如果 }y=0,f(x)=1\text{ (误触发)}\\ w_{FN}& \text{如果 }y=1,f(x)=0\text{ (漏检)}\\ 0& \text{否则}\end{array}$$

通常设置 $w_{FP}>w_{FN}$，以更严格地控制误触发。

对于高风险操作，我们还可以引入确认机制，形式化为顺序决策过程：

1. 初始分类：$y_1=f_1(x)$
2. 如果 $y_1=1$ 且置信度 $con{f}_1<{\tau }_{high}$，请求用户确认：$x^{\prime }=\text{ConfirmRequest}(x)$
3. 二次分类：$y_2=f_2(x,x^{\prime })$
4. 最终决策：$y_{final}=y_2$

2.3 理论局限性

虽然我们构建了相对完善的理论框架，但必须承认其存在一些局限性：

1. 身份认证的局限性：无论多么复杂的认证系统，都无法完全防止身份盗用，特别是在面对定向的社会工程攻击时。

2. 权限模型的局限性：ABAC 等灵活权限模型的复杂性可能导致配置错误，从而引入安全漏洞。同时，"最小权限"在实践中很难精确界定，往往需要在安全性和可用性之间做出权衡。

3. 意图识别的局限性：自然语言的歧义性和人类表达的多样性使得意图识别不可避免地存在错误率。特别是在专业领域，术语的多义性可能导致严重的误触发。

4. 上下文建模的局限性：我们无法捕获和建模所有可能的上下文信息，一些关键的情境因素可能被忽略，导致安全决策不完整。

5. 适应攻击者演化的局限性：我们的模型主要基于已知的攻击模式，面对零日攻击或创新性攻击方法时可能失效。

2.4 竞争范式分析

在 Agent-IM 集成安全领域，存在几种不同的竞争范式，各有优缺点：

我们认为，对于大多数企业场景，混合范式是最佳选择，结合多种技术手段，并在必要时引入人工审批环节，形成多层防御体系。

3. 架构设计

3.1 系统分解

基于前面的理论框架，我们将 Agent-IM 集成安全系统分解为以下核心组件：

这个分层架构遵循安全设计的纵深防御原则，每层都有相应的安全机制，确保即使某一层被突破，整体系统仍然安全。

3.2 组件交互模型

让我们更详细地描述这些核心组件的交互模型，特别是在处理一条用户指令时的完整流程：

这个交互模型展示了处理一条用户指令的完整安全流程，从身份认证到最终执行，每个环节都有相应的安全检查。

3.3 权限模型设计

权限模型是系统安全的核心，我们设计了一个混合权限模型，结合了 RBAC（基于角色的访问控制）和 ABAC（基于属性的访问控制）的优点：

这个权限模型支持：

1. 通过角色管理用户组权限（RBAC部分）
2. 通过属性和条件实现细粒度、上下文感知的权限控制（ABAC部分）
3. 风险分级，不同风险级别的操作有不同的安全要求
4. 时间限制的角色分配
5. 权限条件的灵活组合（AND/OR逻辑）

3.4 指令验证流程设计

指令验证是防止误触发的关键环节，我们设计了一个多层次的指令验证流程：

这个多层次验证流程结合了规则检查、机器学习模型、上下文分析和用户确认等多种手段，有效降低了误触发的风险，同时保持了良好的用户体验。

4. 实现机制

4.1 身份认证实现

企业 IM 集成中的身份认证需要支持多种认证方式，我们将实现一个灵活的认证框架，支持多种 IM 平台和认证方法。

4.1.1 IM 平台签名验证

大多数企业 IM 平台都提供请求签名机制，用于验证 Webhook 请求确实来自该平台。以下是一个通用的签名验证实现：

import hmac
import hashlib
import json
from typing import Dict, Any, Optional, Tuple
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum


class IMPlatform(Enum):
    SLACK = "slack"
    TEAMS = "teams"
    WECHAT_WORK = "wechat_work"
    DINGTALK = "dingtalk"
    FEISHU = "feishu"


@dataclass
class AuthResult:
    success: bool
    user_id: Optional[str] = None
    error_message: Optional[str] = None
    metadata: Dict[str, Any] = None


class IMPlatformAuthenticator(ABC):
    """IM平台认证器抽象基类"""
    
    @abstractmethod
    def authenticate_request(self, headers: Dict[str, str], body: bytes) -> AuthResult:
        """验证请求是否来自合法的IM平台"""
        pass
    
    @abstractmethod
    def extract_user_identity(self, payload: Dict[str, Any]) -> Optional[str]:
        """从有效负载中提取用户身份信息"""
        pass


class SlackAuthenticator(IMPlatformAuthenticator):
    """Slack平台认证器"""
    
    def __init__(self, signing_secret: str):
        self.signing_secret = signing_secret
    
    def authenticate_request(self, headers: Dict[str, str], body: bytes) -> AuthResult:
        # 获取请求头中的时间戳和签名
        timestamp = headers.get("X-Slack-Request-Timestamp")
        signature = headers.get("X-Slack-Signature")
        
        if not timestamp or not signature:
            return AuthResult(False, error_message="Missing required headers")
        
        # 检查时间戳是否在5分钟内，防止重放攻击
        import time
        if abs(time.time() - int(timestamp)) > 60 * 5:
            return AuthResult(False, error_message="Request timestamp too old")
        
        # 构造签名基准字符串
        sig_basestring = f"v0:{timestamp}:{body.decode('utf-8')}"
        
        # 计算HMAC-SHA256哈希
        my_signature = f"v0={hmac.new(
            self.signing_secret.encode('utf-8'),
            sig_basestring.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()}"
        
        # 比较签名
        if not hmac.compare_digest(my_signature, signature):
            return AuthResult(False, error_message="Invalid signature")
        
        # 解析请求体获取用户ID
        try:
            payload = json.loads(body)
            user_id = self.extract_user_identity(payload)
            return AuthResult(True, user_id=user_id, metadata={"platform": IMPlatform.SLACK})
        except Exception as e:
            return AuthResult(False, error_message=f"Failed to parse payload: {str(e)}")
    
    def extract_user_identity(self, payload: Dict[str, Any]) -> Optional[str]:
        # 从Slack事件中提取用户ID
        if "event" in payload and "user" in payload["event"]:
            return payload["event"]["user"]
        elif "user_id" in payload:
            return payload["user_id"]
        elif "user" in payload:
            return payload["user"]
        return None


class TeamsAuthenticator(IMPlatformAuthenticator):
    """Microsoft Teams平台认证器"""
    
    def __init__(self, app_id: str, app_password: str, open_id_config_url: str = None):
        self.app_id = app_id
        self.app_password = app_password
        self.open_id_config_url = open_id_config_url or \
            "https://login.botframework.com/v1/.well-known/openidconfiguration"
        self._signing_keys = None
    
    def authenticate_request(self, headers: Dict[str, str], body: bytes) -> AuthResult:
        # 获取Authorization头
        auth_header = headers.get("Authorization")
        if not auth_header:
            return AuthResult(False, error_message="Missing Authorization header")
        
        try:
            # 验证JWT令牌（简化实现，实际应该使用完整的JWT验证库）
            import jwt
            
            # 提取Bearer令牌
            token = auth_header.replace("Bearer ", "")
            
            # 获取并缓存签名密钥
            if not self._signing_keys:
                self._signing_keys = self._fetch_signing_keys()
            
            # 验证令牌
            payload = jwt.decode(
                token,
                self._signing_keys,
                audience=self.app_id,
                issuer="https://api.botframework.com",
                algorithms=["RS256"]
            )
            
            # 提取用户ID
            user_id = self.extract_user_identity(json.loads(body))
            return AuthResult(True, user_id=user_id, metadata={"platform": IMPlatform.TEAMS})
        except Exception as e:
            return AuthResult(False, error_message=f"Token validation failed: {str(e)}")
    
    def _fetch_signing_keys(self):
        # 从OpenID配置获取签名密钥（简化实现）
        import requests
        jwks_uri = requests.get(self.open_id_config_url).json()["jwks_uri"]
        return requests.get(jwks_uri).json()
    
    def extract_user_identity(self, payload: Dict[str, Any]) -> Optional[str]:
        # 从Teams活动中提取用户ID
        if "from" in payload and "id" in payload["from"]:
            return payload["from"]["id"]
        return None


class AuthenticationManager:
    """认证管理器，支持多个IM平台"""
    
    def __init__(self):
        self._authenticators = {}
    
    def register_authenticator(self, platform: IMPlatform, authenticator: IMPlatformAuthenticator):
        """注册IM平台认证器"""
        self._authenticators[platform] = authenticator
    
    def authenticate(self, platform: IMPlatform, headers: Dict[str, str], body: bytes) -> AuthResult:
        """执行认证"""
        if platform not in self._authenticators:
            return AuthResult(False, error_message=f"No authenticator registered for platform {platform}")
        
        return self._authenticators[platform].authenticate_request(headers, body)

这个实现提供了一个灵活的认证框架，可以轻松扩展支持更多的 IM 平台。我们实现了 Slack 和 Microsoft Teams 的认证器，其他平台可以通过继承 IMPlatformAuthenticator 类来添加。

4.1.2 多因素认证实现

对于高风险操作，我们需要实现多因素认证（MFA）。以下是一个基于时间的一次性密码（TOTP）实现：

import hmac
import hashlib
import base64
import struct
import time
import qrcode
import io
from typing import Optional, Tuple


class TOTPManager:
    """时间-based一次性密码管理器"""
    
    def __init__(self, secret_length: int = 16, interval: int = 30, digits: int = 6):
        self.secret_length = secret_length
        self.interval = interval
        self.digits = digits
    
    def generate_secret(self) -> str:
        """生成随机密钥"""
        import secrets
        random_bytes = secrets.token_bytes(self.secret_length)
        return base64.b32encode(random_bytes).decode('utf-8').replace('=', '')
    
    def generate_totp(self, secret: str, for_time: Optional[int] = None) -> str:
        """生成TOTP码"""
        if for_time is None:
            for_time = int(time.time())
        
        # 解码密钥
        secret_bytes = base64.b32decode(secret.upper() + '=' * ((8 - len(secret) % 8) % 8))
        
        # 计算时间步长
        time_step = struct.pack(">Q", int(for_time // self.interval))
        
        # 计算HMAC-SHA1哈希
        hmac_hash = hmac.new(secret_bytes, time_step, hashlib.sha1).digest()
        
        # 提取动态密码
        offset = hmac_hash[-1] & 0x0F
        truncated_hash = struct.unpack(">I", hmac_hash[offset:offset+4])[0] & 0x7FFFFFFF
        code = truncated_hash % (10 ** self.digits)
        
        # 补齐前导零
        return str(code).zfill(self.digits)
    
    def verify_totp(self, secret: str, code: str, valid_window: int = 1) -> bool:
        """验证TOTP码"""
        code = code.replace(' ', '')
        if len(code) != self.digits or not code.isdigit():
            return False
        
        current_time = int(time.time())
        
        # 检查当前时间窗口及前后valid_window个窗口
        for time_offset in range(-valid_window, valid_window + 1):
            check_time = current_time + time_offset * self.interval
            expected_code = self.generate_totp(secret, for_time=check_time)
            if hmac.compare_digest(code, expected_code):
                return True
        
        return False
    
    def get_provisioning_uri(self, secret: str, account_name: str, issuer_name: str = "EnterpriseAgent") -> str:
        """生成TOTP配置URI"""
        from urllib.parse import quote, urlencode
        
        params = {
            'secret': secret,
            'issuer': issuer_name,
            'algorithm': 'SHA1',
            'digits': self.digits,
            'period': self.interval
        }
        
        return f"otpauth://totp/{quote(issuer_name)}:{quote(account_name)}?{urlencode(params)}"
    
    def generate_qr_code(self, provisioning_uri: str) -> bytes:
        """生成QR码图像"""
        qr = qrcode.QRCode(
            version=1,
            error_correction=qrcode.constants.ERROR_CORRECT_L,
            box_size=10,
            border=4,
        )
        qr.add_data(provisioning_uri)
        qr.make(fit=True)
        
        img = qr.make_image(fill_color="black", back_color="white")
        
        img_buffer = io.BytesIO()
        img.save(img_buffer, format='PNG')
        img_buffer.seek(0)
        
        return img_buffer.getvalue()


class MFAManager:
    """多因素认证管理器"""
    
    def __init__(self, totp_manager: TOTPManager = None):
        self.totp_manager = totp_manager or TOTPManager()
        self._user_secrets = {}  # 实际应用中应存储在安全的数据库中
    
    def enroll_user(self, user_id: str) -> Tuple[str, bytes]:
        """为用户注册MFA，返回配置URI和QR码"""
        secret = self.totp_manager.generate_secret()
        self._user_secrets[user_id] = secret
        
        provisioning_uri = self.totp_manager.get_provisioning_uri(secret, user_id)
        qr_code = self.totp_manager.generate_qr_code(provisioning_uri)
        
        return provisioning_uri, qr_code
    
    def verify_user_code(self, user_id: str, code: str) -> bool:
        """验证用户的MFA码"""
        if user_id not in self._user_secrets:
            return False
        
        secret = self._user_secrets[user_id]
        return self.totp_manager.verify_totp(secret, code)
    
    def is_user_enrolled(self, user_id: str) -> bool:
        """检查用户是否已注册MFA"""
        return user_id in self._user_secrets

这个多因素认证实现可以用于高风险操作的二次确认，增加了一层额外的安全保障。

4.2 权限校验实现

接下来，我们实现基于属性的访问控制（ABAC）权限系统：

import re
import operator
from enum import Enum
from typing import Any, Dict, List, Optional, Callable, Set
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
import json


class ConditionOperator(Enum):
    """条件运算符"""
    EQUALS = "equals"
    NOT_EQUALS = "not_equals"
    CONTAINS = "contains"
    NOT_CONTAINS = "not_contains"
    IN = "in"
    NOT_IN = "not_in"
    GREATER_THAN = "greater_than"
    GREATER_THAN_OR_EQUAL = "greater_than_or_equal"
    LESS_THAN = "less_than"
    LESS_THAN_OR_EQUAL = "less_than_or_equal"
    MATCHES = "matches"
    STARTS_WITH = "starts_with"
    ENDS_WITH = "ends_with"
    IS_TRUE = "is_true"
    IS_FALSE = "is_false"
    IS_NULL = "is_null"
    IS_NOT_NULL = "is_not_null"


class ConditionCombination(Enum):
    """条件组合方式"""
    AND = "and"
    OR = "or"


class PermissionEffect(Enum):
    """权限效果"""
    ALLOW = "allow"
    DENY = "deny"


@dataclass
class AttributeValue:
    """属性值"""
    attribute_name: str
    value: Any
    source: str  # user, resource, environment, request


@dataclass
class Condition:
    """条件表达式"""
    id: str
    attribute_source: str  # user, resource, environment, request
    attribute_name: str
    operator: ConditionOperator
    value: Any
    description: str = ""
    
    def evaluate(self, attributes: Dict[str, Dict[str, Any]]) -> bool:
        """评估条件是否满足"""
        # 获取属性值
        if self.attribute_source not in attributes:
            return False
        
        source_attributes = attributes[self.attribute_source]
        if self.attribute_name not in source_attributes:
            attribute_value = None
        else:
            attribute_value = source_attributes[self.attribute_name]
        
        # 根据运算符评估
        operators = {
            ConditionOperator.EQUALS: operator.eq,
            ConditionOperator.NOT_EQUALS: operator.ne,
            ConditionOperator.GREATER_THAN: operator.gt,
            ConditionOperator.GREATER_THAN_OR_EQUAL: operator.ge,
            ConditionOperator.LESS_THAN: operator.lt,
            ConditionOperator.LESS_THAN_OR_EQUAL: operator.le,
        }
        
        if self.operator in operators:
            try:
                return operators[self.operator](attribute_value, self.value)
            except TypeError:
                return False
        
        # 处理需要特殊逻辑的运算符
        if self.operator == ConditionOperator.CONTAINS:
            return self.value in attribute_value if attribute_value is not None else False
        elif self.operator == ConditionOperator.NOT_CONTAINS:
            return self.value not in attribute_value if attribute_value is not None else True
        elif self.operator == ConditionOperator.IN:
            return attribute_value in self.value if attribute_value is not None else False
        elif self.operator == ConditionOperator.NOT_IN:
            return attribute_value not in self.value if attribute_value is not None else True
        elif self.operator == ConditionOperator.MATCHES:
            if attribute_value is None:
                return False
            try:
                return bool(re.match(self.value, str(attribute_value)))
            except re.error:
                return False
        elif self.operator == ConditionOperator.STARTS_WITH:
            return str(attribute_value).startswith(self.value) if attribute_value is not None else False
        elif self.operator == ConditionOperator.ENDS_WITH:
            return str(attribute_value).endswith(self.value) if attribute_value is not None else False
        elif self.operator == ConditionOperator.IS_TRUE:
            return attribute_value is True
        elif self.operator == ConditionOperator.IS_FALSE:
            return attribute_value is False
        elif self.operator == ConditionOperator.IS_NULL:
            return attribute_value is None
        elif self.operator == ConditionOperator.IS_NOT_NULL:
            return attribute_value is not None
        
        return False


@dataclass
class Permission:
    """权限定义"""
    id: str
    name: str
    description: str
    effect: PermissionEffect
    resource_type: str
    actions: List[str]
    conditions: List[Condition] = field(default_factory=list)
    condition_combination: ConditionCombination = ConditionCombination.AND
    priority: int = 0  # 数字越大优先级越高
    
    def evaluate(self, attributes: Dict[str, Dict[str, Any]], action: str) -> Optional[bool]:
        """
        评估权限是否适用于当前请求
        返回: True=允许, False=拒绝, None=不适用
        """
        # 检查资源类型和操作是否匹配
        if action not in self.actions:
            return None
        
        # 评估条件
        if not self.conditions:
            return self.effect == PermissionEffect.ALLOW
        
        # 根据组合方式评估条件
        if self.condition_combination == ConditionCombination.AND:
            conditions_met = all(cond.evaluate(attributes) for cond in self.conditions)
        else:  # OR
            conditions_met = any(cond.evaluate(attributes) for cond in self.conditions)
        
        if not conditions_met:
            return None
        
        return self.effect == PermissionEffect.ALLOW


@dataclass
class AccessRequest:
    """访问请求"""
    user_id: str
    resource_type: str
    resource_id: str
    action: str
    environment: Dict[str, Any] = field(default_factory=dict)
    request_context: Dict[str, Any] = field(default_factory=dict)


@dataclass
class AccessDecision:
    """访问决策"""
    allowed: bool
    reason: str
    matched_permissions: List[str] = field(default_factory=list)
    decision_details: Dict[str, Any] = field(default_factory=dict)


class AttributeProvider(ABC):
    """属性提供者抽象基类"""
    
    @abstractmethod
    def get_user_attributes(self, user_id: str) -> Dict[str, Any]:
        """获取用户属性"""
        pass
    
    @abstractmethod
    def get_resource_attributes(self, resource_type: str, resource_id: str) -> Dict[str, Any]:
        """获取资源属性"""
        pass


class SimpleAttributeProvider(AttributeProvider):
    """简单的内存属性提供者"""
    
    def __init__(self):
        self._user_attributes = {}
        self._resource_attributes = {}
    
    def set_user