Agent 开发实战：数据泄露的 5 个解决方案

0. 痛点：用户问我天气，我返回了全公司的数据

你有没有遇到过这种情况：

用户：帮我查一下今天的天气
Agent：好的，让我调用 get_weather(city)...

[bug] 查询语句写错：
SELECT * FROM users WHERE city = '北京'
[返回] 全公司 5000 名员工信息

Agent：今天天气晴朗，顺便告诉你：
用户 ID: user_001, 姓名: 张三, 电话: 13800138000, 邮箱: zhangsan@example.com
用户 ID: user_002, 姓名: 李四, 电话: 13900139000, 邮箱: lisi@example.com
...

或者更糟：

[生产环境]
用户：我的 API Key 是多少？
Agent：让我查询数据库...

[返回] SELECT api_key FROM users WHERE id = 'user_001'
Agent：你的 API Key 是：sk-abc123def456...

[用户截图发到微博]
"XX 公司的 Agent 泄露我的 API Key！"
#数据泄露 #隐私侵权

这就是数据泄露（Data Leakage）。

---

1. 问题根源：为什么 Agent 会泄露数据？

原因 1：过度数据收集（Over-collection）

# 收集不必要的数据
def process_user_query(user_id: str, query: str) -> str:
    """处理用户问题（过度收集数据）"""
    # 问题：查询了所有用户数据，而不仅仅是需要的
    user_data = db.query(f"SELECT * FROM users WHERE id = '{user_id}'")
    # user_data 包含：密码、API Key、身份证号等敏感信息
    
    # 调用 LLM（可能把敏感数据发给 LLM 服务商）
    response = call_llm(f"用户数据：{user_data}\n问题：{query}")
    return response

原因 2：缺少数据过滤（No Data Filtering）

# 没有过滤敏感数据
def call_llm(prompt: str) -> str:
    """调用 LLM（没有过滤敏感数据）"""
    # 问题：prompt 可能包含敏感数据（密码、API Key、身份证号）
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# 敏感数据泄露
user_data = {
    "user_id": "user_001",
    "name": "张三",
    "password": "mypassword123",  # 敏感
    "api_key": "sk-abc123def456",  # 敏感
    "id_card": "110101199001011234"  # 敏感
}

prompt = f"用户数据：{user_data}\n问题：今天天气怎么样？"
response = call_llm(prompt)
# 问题：敏感数据被发送给 LLM 服务商

原因 3：不安全的存储（Insecure Storage）

# 明文存储敏感数据
def save_user_data(user_id: str, api_key: str):
    """保存用户数据（明文存储）"""
    # 问题：API Key 明文存储
    db.execute(f"INSERT INTO users (id, api_key) VALUES ('{user_id}', '{api_key}')")
    # 如果数据库被攻破，所有 API Key 泄露

# 明文传输
def send_to_third_party(data: dict):
    """发送数据到第三方（明文传输）"""
    import requests
    # 问题：HTTP 明文传输
    requests.post("http://third-party.com/api", json=data)
    # 如果网络被窃听，数据泄露

---

解决方案 1：数据最小化（Data Minimization）

原理

只收集和使用必要的数据。

from typing import Dict, Any, List
import json


class DataMinimizer:
    """
    数据最小化器
    
    策略：
    1. 只查询需要的字段
    2. 只使用必要的数据
    3. 定期清理不必要的数据
    """
    def __init__(self):
        # 必需字段（按任务类型）
        self.required_fields = {
            "weather_query": ["user_id", "city"],
            "order_query": ["user_id", "order_id"],
            "profile_query": ["user_id", "name", "email"]
        }
    
    def minimize_query(self, task_type: str, user_data: Dict) -> Dict:
        """
        最小化查询（只查询需要的字段）
        
        Args:
            task_type: 任务类型
            user_data: 原始用户数据
        
        Returns:
            最小化后的数据
        """
        if task_type not in self.required_fields:
            raise ValueError(f"未知任务类型：{task_type}")
        
        # 只保留必需字段
        required = set(self.required_fields[task_type])
        minimized = {k: v for k, v in user_data.items() if k in required}
        
        print(f"【数据最小化】原始字段：{len(user_data)} → 最小化后：{len(minimized)}")
        return minimized
    
    def minimize_db_query(self, task_type: str) -> str:
        """
        最小化数据库查询（只查询需要的字段）
        
        Args:
            task_type: 任务类型
        
        Returns:
            SQL 查询语句（只查询必需字段）
        """
        if task_type not in self.required_fields:
            raise ValueError(f"未知任务类型：{task_type}")
        
        # 构造 SELECT 子句（只查询必需字段）
        fields = ", ".join(self.required_fields[task_type])
        query = f"SELECT {fields} FROM users WHERE ..."
        
        print(f"【查询最小化】{query}")
        return query


# 使用示例
minimizer = DataMinimizer()

# 原始用户数据（包含敏感信息）
user_data = {
    "user_id": "user_001",
    "name": "张三",
    "password": "mypassword123",  # 敏感
    "api_key": "sk-abc123def456",  # 敏感
    "id_card": "110101199001011234",  # 敏感
    "city": "北京"
}

# 最小化（只保留必需字段）
minimized = minimizer.minimize_query("weather_query", user_data)
print(f"最小化后：{minimized}")
# {'user_id': 'user_001', 'city': '北京'}

# 最小化数据库查询
query = minimizer.minimize_db_query("weather_query")
print(f"查询语句：{query}")
# SELECT user_id, city FROM users WHERE ...

更先进的最小化：用 LLM 判断必需字段

class LLMDataMinimizer:
    """用 LLM 判断必需字段"""
    def __init__(self, llm):
        self.llm = llm
    
    def minimize_with_llm(self, task_description: str, user_data: Dict) -> Dict:
        """用 LLM 判断必需字段"""
        prompt = f"""
任务：{task_description}

用户数据：
{json.dumps(user_data, ensure_ascii=False, indent=2)}

请判断完成任务必需的字段（只需要字段名，不需要值）。
输出格式（JSON）：
```json
["field1", "field2", ...]

“”"

    response = self.llm(prompt)
    
    try:
        required_fields = json.loads(response)
        minimized = {k: v for k, v in user_data.items() if k in required_fields}
        return minimized
    except json.JSONDecodeError:
        # 解析失败，返回空字典
        return {}

**优点**：减少数据泄露风险  
**缺点**：可能漏掉必需字段

---

## 解决方案 2：数据加密（Data Encryption）

### 原理

加密存储和传输敏感数据。

```python
from cryptography.fernet import Fernet
from typing import Dict, Any
import base64


class DataEncryptor:
    """
    数据加密器
    
    策略：
    1. 静态加密（Encryption at Rest）：加密存储
    2. 传输加密（Encryption in Transit）：加密传输
    3. 端到端加密（End-to-End Encryption）：端到端加密
    """
    def __init__(self, key: bytes = None):
        """
        Args:
            key: 加密密钥（None 则自动生成）
        """
        if key:
            self.key = key
            self.cipher = Fernet(self.key)
        else:
            self.key = Fernet.generate_key()
            self.cipher = Fernet(self.key)
    
    def encrypt(self, data: str) -> bytes:
        """加密数据"""
        return self.cipher.encrypt(data.encode("utf-8"))
    
    def decrypt(self, encrypted_data: bytes) -> str:
        """解密数据"""
        return self.cipher.decrypt(encrypted_data).decode("utf-8")
    
    def encrypt_dict(self, data: Dict) -> Dict:
        """加密字典中的敏感字段"""
        encrypted = data.copy()
        
        # 敏感字段
        sensitive_fields = ["password", "api_key", "id_card", "phone", "email"]
        
        for field in sensitive_fields:
            if field in encrypted:
                encrypted[field] = base64.b64encode(
                    self.encrypt(str(encrypted[field]))
                ).decode("utf-8")
        
        return encrypted
    
    def decrypt_dict(self, data: Dict) -> Dict:
        """解密字典中的敏感字段"""
        decrypted = data.copy()
        
        # 敏感字段
        sensitive_fields = ["password", "api_key", "id_card", "phone", "email"]
        
        for field in sensitive_fields:
            if field in decrypted:
                try:
                    decrypted[field] = self.decrypt(
                        base64.b64decode(decrypted[field])
                    )
                except Exception as e:
                    print(f"【解密失败】{field}：{e}")
        
        return decrypted


# 使用示例
encryptor = DataEncryptor()

# 加密敏感数据
sensitive_data = {
    "user_id": "user_001",
    "name": "张三",
    "password": "mypassword123",
    "api_key": "sk-abc123def456",
    "id_card": "110101199001011234"
}

encrypted = encryptor.encrypt_dict(sensitive_data)
print(f"加密后：{encrypted}")
# password、api_key、id_card 被加密

# 解密
decrypted = encryptor.decrypt_dict(encrypted)
print(f"解密后：{decrypted}")

更先进的加密：用 Hashicorp Vault 管理密钥

# 注意：需要安装 hvac 库
# pip install hvac

import hvac


class VaultKeyManager:
    """用 Hashicorp Vault 管理密钥"""
    def __init__(self, vault_addr: str, token: str):
        """
        Args:
            vault_addr: Vault 地址
            token: Vault Token
        """
        self.client = hvac.Client(url=vault_addr, token=token)
    
    def get_encryption_key(self, key_name: str) -> bytes:
        """从 Vault 获取加密密钥"""
        response = self.client.secrets.kv.v2.read_secret_version(
            path=f"keys/{key_name}"
        )
        
        key = response["data"]["data"]["key"]
        return base64.b64decode(key)
    
    def rotate_key(self, key_name: str):
        """轮换密钥"""
        # 生成新密钥
        new_key = Fernet.generate_key()
        
        # 保存到 Vault
        self.client.secrets.kv.v2.create_or_update_secret(
            path=f"keys/{key_name}",
            secret={"key": base64.b64encode(new_key).decode("utf-8")}
        )
        
        print(f"【密钥轮换】{key_name} 已轮换")

优点：即使数据泄露，也无法解密
缺点：增加处理时间

---

解决方案 3：访问控制（Access Control）

原理

限制数据访问权限。

from typing import Dict, Any, List
import functools


class DataAccessControl:
    """
    数据访问控制
    
    策略：
    1. 基于角色的访问控制（RBAC）
    2. 基于属性的访问控制（ABAC）
    3. 最小权限原则
    """
    def __init__(self):
        # 角色权限表
        self.role_permissions = {
            "admin": ["read", "write", "delete", "manage_users"],
            "user": ["read", "write"],
            "guest": ["read"]
        }
        
        # 数据权限表（哪些角色可以访问哪些数据）
        self.data_permissions = {
            "user_data": ["admin", "user"],  # 用户数据：admin 和 user 可以访问
            "order_data": ["admin", "user"],  # 订单数据：admin 和 user 可以访问
            "financial_data": ["admin"]  # 财务数据：只有 admin 可以访问
        }
    
    def check_permission(self, user_id: str, data_type: str, permission: str) -> bool:
        """
        检查用户是否有权限访问数据
        
        Args:
            user_id: 用户 ID
            data_type: 数据类型（user_data/order_data/financial_data）
            permission: 权限（read/write/delete）
        
        Returns:
            True: 有权限
            False: 无权限
        """
        # 1. 获取用户角色
        role = self._get_user_role(user_id)
        if not role:
            print(f"【权限拒绝】用户 {user_id} 未分配角色")
            return False
        
        # 2. 检查角色权限
        if permission not in self.role_permissions.get(role, []):
            print(f"【权限拒绝】用户 {user_id}（角色 {role}）缺少权限：{permission}")
            return False
        
        # 3. 检查数据权限
        allowed_roles = self.data_permissions.get(data_type, [])
        if role not in allowed_roles:
            print(f"【权限拒绝】用户 {user_id}（角色 {role}）无权访问数据：{data_type}")
            return False
        
        return True
    
    def _get_user_role(self, user_id: str) -> str:
        """获取用户角色（模拟）"""
        # 实际应该从数据库查询
        user_roles = {
            "user_001": "admin",
            "user_002": "user",
            "user_003": "guest"
        }
        return user_roles.get(user_id)
    
    def require_data_permission(self, data_type: str, permission: str):
        """
        数据权限检查装饰器
        
        Usage:
            @require_data_permission("user_data", "read")
            def get_user_data(user_id: str, target_user_id: str):
                ...
        """
        def decorator(func):
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                # 获取 user_id（假设第一个参数是 user_id）
                user_id = args[0] if args else kwargs.get("user_id")
                
                # 检查权限
                if not self.check_permission(user_id, data_type, permission):
                    raise PermissionError(f"用户 {user_id} 无权访问 {data_type}（需要 {permission} 权限）")
                
                # 有权限，执行函数
                return func(*args, **kwargs)
            return wrapper
        return decorator


# 使用示例
access_control = DataAccessControl()

# 检查权限
user_id = "user_002"
data_type = "financial_data"
permission = "read"

if access_control.check_permission(user_id, data_type, permission):
    print(f"✅ 用户 {user_id} 有权限访问 {data_type}")
else:
    print(f"❌ 用户 {user_id} 无权限访问 {data_type}")

# 使用装饰器
@access_control.require_data_permission("user_data", "read")
def get_user_data(user_id: str, target_user_id: str):
    """获取用户数据"""
    print(f"获取用户 {target_user_id} 的数据")
    return {"user_id": target_user_id, "name": "张三"}

# 调用
try:
    data = get_user_data("user_001", "user_002")  # user_001 是 admin，有权限
    print(f"✅ 获取成功：{data}")
except PermissionError as e:
    print(f"❌ 获取失败：{e}")

更精细的访问控制：基于属性的访问控制（ABAC）

class ABACDataAccessControl:
    """基于属性的访问控制（ABAC）"""
    def __init__(self):
        pass
    
    def check_permission(
        self,
        user_id: str,
        data_type: str,
        permission: str,
        resource_owner: str
    ) -> bool:
        """
        检查权限（基于属性）
        
        Args:
            user_id: 用户 ID
            data_type: 数据类型
            permission: 权限
            resource_owner: 资源所有者
        """
        # 规则 1：资源所有者有全部权限
        if user_id == resource_owner:
            return True
        
        # 规则 2：admin 有全部权限
        if self._get_user_role(user_id) == "admin":
            return True
        
        # 规则 3：其他人只有 read 权限
        if permission == "read":
            return True
        
        # 规则 4：其他操作拒绝
        print(f"【权限拒绝】用户 {user_id} 无权对 {data_type} 执行 {permission} 操作")
        return False
    
    def _get_user_role(self, user_id: str) -> str:
        """获取用户角色（模拟）"""
        user_roles = {
            "user_001": "admin",
            "user_002": "user",
            "user_003": "guest"
        }
        return user_roles.get(user_id)

优点：防止越权访问
缺点：增加代码复杂度

---

解决方案 4：数据脱敏（Data Masking）

原理

在输出前，脱敏敏感数据。

import re
from typing import Dict, Any


class DataMasker:
    """
    数据脱敏器
    
    脱敏对象：
    - 密码（password、passwd）
    - API Key（sk-xxx、api_key）
    - 身份证号（18 位）
    - 手机号（11 位）
    - 邮箱（xxx@xxx.com）
    - 银行卡号（16-19 位）
    """
    def __init__(self):
        # 脱敏规则
        self.mask_rules = {
            "password": r"(password|passwd|pwd)\s*[:=]\s*\S+",
            "api_key": r"(sk-|api_key|apikey)\s*[:=]\s*\S+",
            "id_card": r"\b\d{17}[\dXx]\b",
            "phone": r"\b1[3-9]\d{9}\b",
            "email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
            "bank_card": r"\b\d{16,19}\b"
        }
    
    def mask(self, text: str, mask_char: str = "*") -> str:
        """
        脱敏敏感数据
        
        Args:
            text: 原始文本
            mask_char: 脱敏字符（默认 *）
        
        Returns:
            脱敏后的文本
        """
        masked_text = text
        
        for info_type, pattern in self.mask_rules.items():
            # 查找所有匹配
            matches = re.finditer(pattern, masked_text, re.IGNORECASE)
            
            for match in matches:
                # 脱敏（保留前 3 后 4，中间用 * 替换）
                original = match.group(0)
                if len(original) > 7:
                    masked = original[:3] + mask_char * (len(original) - 7) + original[-4:]
                else:
                    masked = mask_char * len(original)
                
                masked_text = masked_text.replace(original, masked)
                print(f"【脱敏】{info_type}：{original} → {masked}")
        
        return masked_text
    
    def mask_dict(self, data: Dict, mask_char: str = "*") -> Dict:
        """
        脱敏字典中的敏感信息
        
        Args:
            data: 原始字典
            mask_char: 脱敏字符
        
        Returns:
            脱敏后的字典
        """
        masked_data = data.copy()
        
        # 敏感字段
        sensitive_fields = ["password", "api_key", "id_card", "phone", "email", "bank_card"]
        
        for field in sensitive_fields:
            if field in masked_data:
                # 脱敏
                original = str(masked_data[field])
                if len(original) > 7:
                    masked = original[:3] + mask_char * (len(original) - 7) + original[-4:]
                else:
                    masked = mask_char * len(original)
                
                masked_data[field] = masked
                print(f"【脱敏】{field}：{original} → {masked}")
        
        return masked_data


# 使用示例
masker = DataMasker()

# 脱敏文本
text = """
用户名：xiaoyuer
密码：mypassword123
API Key：sk-abc123def456
身份证号：110101199001011234
手机号：13800138000
邮箱：xiaoyuer@example.com
银行卡号：6222021234567890123
"""

masked_text = masker.mask(text)
print(f"原始文本：\n{text}")
print(f"脱敏后：\n{masked_text}")

# 脱敏字典
user_data = {
    "user_id": "user_001",
    "username": "xiaoyuer",
    "password": "mypassword123",
    "api_key": "sk-abc123def456",
    "email": "xiaoyuer@example.com"
}

masked_data = masker.mask_dict(user_data)
print(f"原始数据：{user_data}")
print(f"脱敏后：{masked_data}")

集成到 Agent

class AgentWithDataMasking:
    """带数据脱敏的 Agent"""
    def __init__(self, masker: DataMasker):
        self.masker = masker
        self.llm = ...  # LLM 实例
    
    def process(self, query: str) -> str:
        """处理用户问题（带数据脱敏）"""
        # 1. 调用 LLM
        response = self.llm.call(query)
        
        # 2. 脱敏
        masked_response = self.masker.mask(response)
        
        return masked_response


# 使用
masker = DataMasker()
agent = AgentWithDataMasking(masker=masker)

response = agent.process("查询我的信息")
print(f"响应：{response}")

优点：保护敏感信息，防止泄露
缺点：可能误伤正常内容

---

解决方案 5：安全数据处理规范（Secure Data Handling Practices）

原理

建立安全数据处理规范，防止人为失误。

from typing import Dict, Any, List
import json


class SecureDataHandling:
    """
    安全数据处理规范
    
    规范：
    1. 不记录敏感数据到日志
    2. 不发送敏感数据到第三方
    3. 定期清理敏感数据
    4. 使用安全的数据传输协议（HTTPS）
    5. 定期安全培训
    """
    def __init__(self):
        # 敏感字段
        self.sensitive_fields = ["password", "api_key", "id_card", "phone", "email"]
    
    def sanitize_for_logging(self, data: Dict) -> Dict:
        """
        清理敏感数据（用于日志记录）
        
        Args:
            data: 原始数据
        
        Returns:
            清理后的数据（敏感字段被移除或脱敏）
        """
        sanitized = data.copy()
        
        for field in self.sensitive_fields:
            if field in sanitized:                # 移除敏感字段
                del sanitized[field]
                print(f"【日志清理】移除敏感字段：{field}")
        
        return sanitized
    
    def sanitize_for_third_party(self, data: Dict) -> Dict:
        """
        清理敏感数据（用于发送给第三方）
        
        Args:
            data: 原始数据
        
        Returns:
            清理后的数据（敏感字段被移除或脱敏）
        """
        sanitized = data.copy()
        
        for field in self.sensitive_fields:
            if field in sanitized:
                # 脱敏（而不是移除）
                sanitized[field] = "***"
                print(f"【第三方清理】脱敏敏感字段：{field}")
        
        return sanitized
    
    def validate_https(self, url: str) -> bool:
        """
        验证 URL 是否使用 HTTPS
        
        Args:
            url: URL
        
        Returns:
            True: 使用 HTTPS
            False: 未使用 HTTPS
        """
        if not url.startswith("https://"):
            print(f"【安全警告】URL 未使用 HTTPS：{url}")
            return False
        return True
    
    def audit_data_access(self, user_id: str, data_type: str, action: str):
        """
        审计数据访问
        
        Args:
            user_id: 用户 ID
            data_type: 数据类型
            action: 操作（read/write/delete）
        """
        audit_log = {
            "timestamp": time.time(),
            "user_id": user_id,
            "data_type": data_type,
            "action": action
        }
        
        # 写入审计日志
        with open("data_access_audit.log", "a", encoding="utf-8") as f:
            f.write(json.dumps(audit_log, ensure_ascii=False) + "\n")
        
        print(f"【审计】{user_id} {action} {data_type}")


# 使用示例
handler = SecureDataHandling()

# 清理敏感数据（用于日志记录）
user_data = {
    "user_id": "user_001",
    "name": "张三",
    "password": "mypassword123",  # 敏感
    "api_key": "sk-abc123def456"  # 敏感
}

sanitized = handler.sanitize_for_logging(user_data)
print(f"原始数据：{user_data}")
print(f"清理后：{sanitized}")
# {'user_id': 'user_001', 'name': '张三'}

# 清理敏感数据（用于发送给第三方）
sanitized_for_third_party = handler.sanitize_for_third_party(user_data)
print(f"清理后（第三方）：{sanitized_for_third_party}")
# {'user_id': 'user_001', 'name': '张三', 'password': '***', 'api_key': '***'}

# 验证 HTTPS
url = "http://third-party.com/api"
if not handler.validate_https(url):
    print("❌ 禁止使用 HTTP 传输数据")

# 审计数据访问
handler.audit_data_access(user_id="user_001", data_type="user_data", action="read")

更先进的规范：用 LLM 检查违规行为

class LLMDataHandlingAuditor:
    """用 LLM 检查数据处理的违规行为"""
    def __init__(self, llm):
        self.llm = llm
    
    def audit_code(self, code: str) -> Dict:
        """用 LLM 检查代码中的违规行为"""
        prompt = f"""
请检查以下代码是否违反安全数据处理规范：

代码：
{code}

安全规范：
1. 不记录敏感数据到日志
2. 不发送敏感数据到第三方（除非脱敏）
3. 使用 HTTPS 传输数据
4. 不明文存储敏感数据

输出格式（JSON）：
```json
{{
  "violations": [
    {{
      "line": 1,
      "description": "违规描述"
    }}
  ],
  "is_safe": true/false
}}

“”"

    response = self.llm(prompt)
    
    try:
        result = json.loads(response)
        return result
    except json.JSONDecodeError:
        return {"violations": [], "is_safe": True}

**优点**：防止人为失误  
**缺点**：需要培训和执行

---

## 效果对比

| 方案 | 安全性 | 实现难度 | 性能影响 | 适用场景 |
|------|--------|---------|---------|---------|
| 数据最小化 | ⭐⭐⭐⭐ | ⭐⭐ | 低 | 所有 Agent |
| 数据加密 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | 中 | 敏感数据存储 |
| 访问控制 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | 低 | 多用户系统 |
| 数据脱敏 | ⭐⭐⭐⭐ | ⭐⭐ | 低 | 输出包含敏感数据 |
| 安全数据处理规范 | ⭐⭐⭐ | ⭐⭐⭐⭐ | 无 | 团队协作 |

---

## 避坑指南

### 1. 数据最小化不是越少越好

**错误做法**：
```python
# 过度最小化（缺少必需字段）
minimized = minimize_query("order_query", user_data)
# 问题：缺少 order_id，无法查询订单

正确做法：

# 合理最小化（保留必需字段）
minimized = minimize_query("order_query", user_data)
# 确保包含 order_id

2. 数据加密不是越复杂越好

错误做法：

# 使用自定义加密算法（不安全）
def my_encrypt(data: str) -> str:
    # 自定义加密（容易被破解）
    return data[::-1]  # 简单反转

正确做法：

# 使用标准加密算法
from cryptography.fernet import Fernet
encryptor = DataEncryptor()
encrypted = encryptor.encrypt(data)

3. 访问控制不是越严越好

错误做法：

# 太严格（影响正常使用）
@require_data_permission("user_data", "read")
def get_user_data(user_id: str):
    # 问题：连用户自己都无法读取自己的数据
    ...

正确做法：

# 合理权限（资源所有者有全部权限）
@require_data_permission("user_data", "read")
def get_user_data(user_id: str, target_user_id: str):
    # 如果用户是资源所有者，允许
    if user_id == target_user_id:
        return get_data(target_user_id)
    # 否则检查权限
    ...

---

延伸思考

1. 如何用 RL 优化数据安全策略？

方案：用强化学习（RL）学习最优安全策略。

# 伪代码
class DataSecurityOptimizationRL:
    """用强化学习优化数据安全策略"""
    def __init__(self):
        self.policy = self._init_policy()
    
    def optimize(self, security_events: List[Dict]) -> Dict:
        """优化安全策略"""
        # 用 RL 学习最优策略
        optimal_policy = self.policy.learn(security_events)
        return optimal_policy

2. 如何自动检测数据泄露？

方案：用异常检测（Anomaly Detection）发现数据泄露。

def detect_data_leak_anomalies(data_access_logs: List[Dict]) -> List[Dict]:
    """用异常检测发现数据泄露"""
    # 用 Isolation Forest 检测异常
    from sklearn.ensemble import IsolationForest
    model = IsolationForest()
    model.fit(data_access_logs)
    anomalies = model.predict(data_access_logs)
    return anomalies

3. 如何平衡数据安全性和用户体验？

方案：定义安全性-用户体验权衡函数。

def security_ux_tradeoff(security: float, ux: float) -> float:
    """
    安全性-用户体验权衡函数
    
    Returns:
        得分（越高越好）
    """
    # 简化：安全性权重 0.6，用户体验权重 0.4
    score = 0.6 * security + 0.4 * ux
    return score

---

总结

数据泄露是 Agent 开发的核心痛点，需要组合多种方案：

基础方案（必做） → 数据最小化 + 数据脱敏
进阶方案（推荐） → 数据加密 + 访问控制
高级方案（可选） → 安全数据处理规范（团队协作用）

关键原则：

1. 优先数据最小化，只收集必要数据
2. 加密存储和传输敏感数据
3. 做好访问控制，防止越权访问
4. 脱敏输出中的敏感数据
5. 建立安全数据处理规范，防止人为失误

---

下一篇预告：《Agent 开发实战：版本管理的 5 个解决方案》