Agent 隐私与数据边界：什么代码不能进模型上下文

数据流向：
  本地代码库 → 上下文包生成器 → 模型提供商 API → 模型推理 → 返回结果

风险点：
  ① 上下文包可能包含不该发送的内容
  ② 传输过程可能被截获
  ③ 模型提供商可能记录/缓存/训练
  ④ 返回结果可能泄露给其他用户

# data-classification.yaml
classification_levels:
  
  L1_public:
    name: "公开数据"
    description: "可以发送给任何模型"
    examples:
      - "开源项目代码"
      - "公开的 API 文档"
      - "技术博客内容"
    handling: "无需特殊处理"
  
  L2_internal:
    name: "内部数据"
    description: "可以发送给商业模型，但不应发送给免费模型"
    examples:
      - "业务代码（非核心逻辑）"
      - "内部工具代码"
      - "测试数据（脱敏后）"
    handling: "正常发送，记录审计日志"
  
  L3_confidential:
    name: "机密数据"
    description: "需要审批才能发送，或必须脱敏后发送"
    examples:
      - "核心业务逻辑"
      - "数据库 Schema"
      - "内部 API 设计"
      - "配置文件（含环境变量）"
    handling: "审批后发送 / 脱敏后发送"
  
  L4_restricted:
    name: "受限数据"
    description: "默认不发送，特殊场景需高级审批"
    examples:
      - "客户个人信息（PII）"
      - "支付相关代码"
      - "安全相关代码（加密、认证）"
      - "生产环境日志"
    handling: "高级审批 + 脱敏 + 审计"
  
  L5_prohibited:
    name: "禁止数据"
    description: "绝对不发送给任何外部模型"
    examples:
      - "API 密钥、Token、密码"
      - "私钥文件（SSH、TLS）"
      - "客户原始数据（未脱敏）"
      - "核心算法（竞争优势）"
      - "合规数据（HIPAA、GDPR 约束）"
    handling: "本地模型处理 / 完全不处理"

# app/privacy/scanner.py
"""
数据边界扫描器。
扫描上下文包中的文件，标记敏感内容。
"""
import re
from pathlib import Path
from dataclasses import dataclass

@dataclass
class ScanResult:
    file_path: str
    classification: str     # L1-L5
    sensitive_items: list   # 发现的敏感项
    action: str             # allow / redact / block
    reason: str

class PrivacyScanner:
    def __init__(self, rules_path: str):
        self.rules = self._load_rules(rules_path)
    
    def scan_file(self, file_path: str) -> ScanResult:
        """扫描单个文件"""
        content = Path(file_path).read_text()
        
        sensitive_items = []
        max_level = "L1_public"
        
        # 1. 扫描密钥模式
        for pattern in self.rules["secret_patterns"]:
            matches = re.findall(pattern["regex"], content)
            if matches:
                sensitive_items.extend([{
                    "type": pattern["type"],
                    "count": len(matches),
                    "level": pattern["level"],
                } for _ in matches])
                max_level = self._max_level(max_level, pattern["level"])
        
        # 2. 扫描 PII 模式
        for pattern in self.rules["pii_patterns"]:
            matches = re.findall(pattern["regex"], content)
            if matches:
                sensitive_items.extend([{
                    "type": pattern["type"],
                    "count": len(matches),
                    "level": pattern["level"],
                } for _ in matches])
                max_level = self._max_level(max_level, pattern["level"])
        
        # 3. 检查文件路径规则
        for rule in self.rules["path_rules"]:
            if re.match(rule["pattern"], file_path):
                max_level = self._max_level(max_level, rule["level"])
        
        # 4. 决定处理方式
        action = self._decide_action(max_level)
        
        return ScanResult(
            file_path=file_path,
            classification=max_level,
            sensitive_items=sensitive_items,
            action=action,
            reason=f"Classification: {max_level}",
        )
    
    def _load_rules(self, rules_path: str) -> dict:
        import yaml
        with open(rules_path) as f:
            return yaml.safe_load(f)
    
    def _max_level(self, current: str, new: str) -> str:
        levels = ["L1_public", "L2_internal", "L3_confidential", 
                  "L4_restricted", "L5_prohibited"]
        return new if levels.index(new) > levels.index(current) else current
    
    def _decide_action(self, level: str) -> str:
        actions = {
            "L1_public": "allow",
            "L2_internal": "allow",
            "L3_confidential": "redact",
            "L4_restricted": "block",
            "L5_prohibited": "block",
        }
        return actions.get(level, "block")

# 扫描规则配置
SCANNER_RULES = """
secret_patterns:
  - type: "aws_access_key"
    regex: "AKIA[0-9A-Z]{16}"
    level: "L5_prohibited"
  
  - type: "github_token"
    regex: "ghp_[a-zA-Z0-9]{36}"
    level: "L5_prohibited"
  
  - type: "generic_api_key"
    regex: "(?i)(api[_-]?key|secret|password)\\\\s*[:=]\\\\s*['\\\\\"]?[a-zA-Z0-9]{16,}"
    level: "L4_restricted"
  
  - type: "private_key"
    regex: "-----BEGIN (RSA |EC |DSA )?PRIVATE KEY-----"
    level: "L5_prohibited"

pii_patterns:
  - type: "email"
    regex: "[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\\\.[a-zA-Z]{2,}"
    level: "L3_confidential"
  
  - type: "phone"
    regex: "\\\\b\\\\d{3}[-.]?\\\\d{3}[-.]?\\\\d{4}\\\\b"
    level: "L3_confidential"
  
  - type: "ssn"
    regex: "\\\\b\\\\d{3}-\\\\d{2}-\\\\d{4}\\\\b"
    level: "L4_restricted"
  
  - type: "credit_card"
    regex: "\\\\b\\\\d{4}[- ]?\\\\d{4}[- ]?\\\\d{4}[- ]?\\\\d{4}\\\\b"
    level: "L4_restricted"

path_rules:
  - pattern: ".*\\\\.(env|pem|key|p12|pfx)$"
    level: "L5_prohibited"
  
  - pattern: ".*/config/(prod|production)/.*"
    level: "L4_restricted"
  
  - pattern: ".*/(test|tests|spec)/.*"
    level: "L2_internal"
  
  - pattern: ".*\\\\.md$"
    level: "L2_internal"
"""

# redaction-strategies.yaml
strategies:
  
  # 不可逆脱敏（替换为占位符）
  irreversible:
    - type: "api_key"
      method: "replace"
      replacement: "***API_KEY***"
      use_case: "密钥、Token"
    
    - type: "email"
      method: "replace"
      replacement: "user@example.com"
      use_case: "邮件地址"
    
    - type: "phone"
      method: "replace"
      replacement: "123-456-7890"
      use_case: "电话号码"
    
    - type: "number"
      method: "mask_middle"
      keep_prefix: 3
      keep_suffix: 4
      use_case: "账号、订单号"
      example: "ORD-123***-4567"
  
  # 可逆脱敏（加密，需要时可解密）
  reversible:
    - type: "customer_id"
      method: "encrypt"
      algorithm: "AES-256"
      key_source: "KMS"
      use_case: "需要回溯客户数据时"
    
    - type: "ip_address"
      method: "hash"
      algorithm: "SHA-256"
      salt: "per-session"
      use_case: "日志分析时需要关联同一 IP"

# app/privacy/context_filter.py
"""
上下文包过滤器。
在发送给模型之前，过滤和脱敏上下文包中的内容。
"""

class ContextFilter:
    def __init__(self, scanner: PrivacyScanner, redactor: Redactor):
        self.scanner = scanner
        self.redactor = redactor
    
    def filter_context_pack(self, files: list, policy: str) -> dict:
        """
        过滤上下文包。
        policy: "strict" / "moderate" / "relaxed"
        """
        filtered = []
        blocked = []
        redacted = []
        
        for file_path in files:
            scan_result = self.scanner.scan_file(file_path)
            
            if policy == "strict":
                # 严格模式：L3+ 都阻止
                if scan_result.classification in ["L3_confidential", "L4_restricted", "L5_prohibited"]:
                    blocked.append({
                        "file": file_path,
                        "reason": scan_result.classification,
                    })
                    continue
            
            elif policy == "moderate":
                # 中等模式：L4+ 阻止，L3 脱敏
                if scan_result.classification in ["L4_restricted", "L5_prohibited"]:
                    blocked.append({
                        "file": file_path,
                        "reason": scan_result.classification,
                    })
                    continue
                elif scan_result.classification == "L3_confidential":
                    content = self.redactor.redact_file(file_path)
                    filtered.append({
                        "path": file_path,
                        "content": content,
                        "redacted": True,
                    })
                    redacted.append(file_path)
                    continue
            
            # L1-L2 或 relaxed 模式：直接通过
            filtered.append({
                "path": file_path,
                "content": Path(file_path).read_text(),
                "redacted": False,
            })
        
        return {
            "files": filtered,
            "blocked": blocked,
            "redacted": redacted,
            "policy": policy,
        }

# privacy-routing.yaml
routing:
  # 根据数据分级路由到不同模型
  
  - condition: "data_classification in ['L1_public', 'L2_internal']"
    model: "cloud-sonnet"
    reason: "非敏感数据可以用云端模型"
  
  - condition: "data_classification == 'L3_confidential'"
    model: "cloud-sonnet"
    pre_processing: "redact"
    reason: "脱敏后发送到云端"
  
  - condition: "data_classification in ['L4_restricted', 'L5_prohibited']"
    model: "local-llama3-70b"
    reason: "敏感数据只用本地模型"
    fallback: "skip_task"  # 没有本地模型就跳过
  
  # 特殊场景
  - condition: "task.involves_customer_pii"
    model: "local-only"
    note: "涉及客户 PII 的任务永远不出云"
  
  - condition: "task.involves_production_data"
    model: "local-only"
    note: "生产数据永远不出云"