🇪🇺 GDPR 合规

如果你的 AI 产品有欧盟用户,GDPR 不是可选项。2024 年意大利对 AI 公司的罚款表明,监管机构正在积极执法。

⚠️ 罚款规模:GDPR 违规最高可罚全球年营业额的 4% 或 2000 万欧元(取较高者)。对中小公司来说,一次严重违规可能就是灭顶之灾。

GDPR 核心原则(AI 产品视角)

原则传统含义AI 产品影响
合法性、公正性、透明性必须有合法基础处理数据必须告知用户数据如何用于 AI 训练/推理
目的限制只能为特定目的处理数据对话数据不能用于训练模型(除非单独同意)
数据最小化只收集必要的数据不要存储完整对话——只存必要元数据
准确性确保数据准确AI 输出的"幻觉"问题可能与准确性原则冲突
存储限制不能永久保存数据对话历史必须有保留期限和自动删除
完整性 & 保密性安全保障加密、访问控制、数据隔离
可问责性能证明合规数据处理记录、DPIA、审计日志

数据处理者义务

💡 关键区分:如果你用 OpenAI/Anthropic 的 API,你是数据处理者 (Controller),API 提供商是处理者 (Processor)。你需要与 API 提供商签署数据处理协议 (DPA)。
## 数据处理者 (Controller) 的义务

### 1. 确定合法基础
AI 产品常见的合法基础:
├── 同意 (Consent) — 用户明确同意数据处理
│   └── 必须可以随时撤回
│   └── 必须具体明确(不能一揽子同意)
├── 合同履行 (Contract) — 提供服务必须的数据处理
│   └── 如:存储对话历史以维持对话连续性
├── 合法利益 (Legitimate Interest) — 业务需要
│   └── 如:防止滥用、安全监控
│   └── 必须进行利益平衡测试
└── 法律义务 (Legal Obligation) — 法律要求
    └── 如:反洗钱、内容审核留存

### 2. 透明度义务
- 隐私政策必须包含 AI 数据处理说明
- 说明数据如何被 LLM 处理
- 说明是否用于模型训练(OpenAI API 默认不训练)
- 说明数据保留期限
- 说明用户权利

### 3. 数据处理协议 (DPA)
- 与 OpenAI 签署: https://openai.com/policies/dpa
- 与 Anthropic 签署: 通过企业计划
- DPA 必须包含:
  - 处理主题和期限
  - 处理性质和目的
  - 数据类型和主体类别
  - 处理者义务

用户数据导出实现

GDPR 第 20 条赋予用户数据可携带权——用户可以要求导出其所有数据。

import json
import zipfile
from datetime import datetime
from pathlib import Path

class DataExportService:
    """GDPR 数据导出服务"""
    
    def __init__(self, db_session, storage_client):
        self.db = db_session
        self.storage = storage_client
    
    async def export_user_data(self, user_id: str, tenant_id: str) -> str:
        """导出用户所有数据为 ZIP 文件"""
        export_data = {
            'export_info': {
                'user_id': user_id,
                'tenant_id': tenant_id,
                'export_date': datetime.utcnow().isoformat(),
                'format_version': '1.0',
                'regulation': 'GDPR Article 20 - Right to data portability'
            },
            'profile': await self._export_profile(user_id, tenant_id),
            'conversations': await self._export_conversations(user_id, tenant_id),
            'documents': await self._export_documents(user_id, tenant_id),
            'usage_logs': await self._export_usage_logs(user_id, tenant_id),
            'settings': await self._export_settings(user_id, tenant_id),
        }
        
        # 生成 ZIP 文件
        export_path = f"/tmp/gdpr_export_{user_id}_{datetime.now().strftime('%Y%m%d')}.zip"
        with zipfile.ZipFile(export_path, 'w', zipfile.ZIP_DEFLATED) as zf:
            # 完整 JSON
            zf.writestr('complete_export.json', json.dumps(export_data, indent=2, ensure_ascii=False))
            
            # 分类文件
            for key, value in export_data.items():
                if key != 'export_info' and value:
                    zf.writestr(f'{key}.json', json.dumps(value, indent=2, ensure_ascii=False))
        
        # 上传到临时存储(7天后自动删除)
        download_url = await self.storage.upload_temp(export_path, expires_in_days=7)
        return download_url
    
    async def _export_conversations(self, user_id: str, tenant_id: str):
        """导出对话历史"""
        conversations = self.db.query(Conversation).filter(
            Conversation.tenant_id == tenant_id,
            Conversation.user_id == user_id
        ).all()
        
        result = []
        for conv in conversations:
            messages = self.db.query(Message).filter(
                Message.conversation_id == conv.id,
                Message.tenant_id == tenant_id
            ).all()
            
            result.append({
                'id': conv.id,
                'title': conv.title,
                'created_at': conv.created_at,
                'messages': [{
                    'role': msg.role,
                    'content': msg.content,
                    'created_at': msg.created_at,
                    'model': msg.model if hasattr(msg, 'model') else None,
                    'tokens_used': msg.token_count if hasattr(msg, 'token_count') else None
                } for msg in messages]
            })
        return result

# API 端点
"""
@app.post("/api/gdpr/export")
async def request_data_export(request: Request):
    user_id = request.state.user_id
    tenant_id = request.state.tenant_id
    
    export_service = DataExportService(db, storage)
    download_url = await export_service.export_user_data(user_id, tenant_id)
    
    # 通知用户
    await send_email(
        to=request.state.user_email,
        subject="Your Data Export is Ready",
        body=f"Download link: {download_url} (expires in 7 days)"
    )
    
    return {"status": "processing", "estimated_time": "5-10 minutes"}
"""

用户数据删除实现

GDPR 第 17 条赋予用户删除权("被遗忘权")——用户可以要求删除其所有个人数据。

from datetime import datetime, timedelta

class DataDeletionService:
    """GDPR 数据删除服务"""
    
    def __init__(self, db_session, vector_store, storage_client):
        self.db = db_session
        self.vector_store = vector_store
        self.storage = storage_client
    
    async def delete_user_data(self, user_id: str, tenant_id: str, 
                                immediate: bool = False) -> dict:
        """删除用户所有数据
        
        Args:
            immediate: True = 立即删除; False = 30天宽限期
        """
        deletion_date = datetime.utcnow() if immediate else datetime.utcnow() + timedelta(days=30)
        
        result = {
            'user_id': user_id,
            'deletion_type': 'immediate' if immediate else 'scheduled',
            'scheduled_date': deletion_date.isoformat(),
            'items_deleted': {}
        }
        
        if immediate:
            result['items_deleted'] = await self._execute_deletion(user_id, tenant_id)
        else:
            # 30天宽限期:先标记,30天后自动删除
            await self._schedule_deletion(user_id, tenant_id, deletion_date)
            result['status'] = 'scheduled'
        
        return result
    
    async def _execute_deletion(self, user_id: str, tenant_id: str) -> dict:
        """执行实际删除"""
        deleted = {}
        
        # 1. 删除对话消息
        msg_count = self.db.query(Message).filter(
            Message.tenant_id == tenant_id,
            Message.user_id == user_id
        ).delete()
        deleted['messages'] = msg_count
        
        # 2. 删除对话
        conv_count = self.db.query(Conversation).filter(
            Conversation.tenant_id == tenant_id,
            Conversation.user_id == user_id
        ).delete()
        deleted['conversations'] = conv_count
        
        # 3. 删除向量数据库中的 Embeddings
        vector_ids = self.vector_store.list_ids(
            tenant_id=tenant_id,
            filter={'user_id': user_id}
        )
        if vector_ids:
            self.vector_store.delete(ids=vector_ids)
        deleted['embeddings'] = len(vector_ids)
        
        # 4. 删除上传的文件
        files = self.storage.list_user_files(user_id, tenant_id)
        for file in files:
            self.storage.delete(file['path'])
        deleted['files'] = len(files)
        
        # 5. 删除用户配置
        self.db.query(UserSettings).filter(
            UserSettings.user_id == user_id,
            UserSettings.tenant_id == tenant_id
        ).delete()
        deleted['settings'] = 1
        
        # 6. 删除日志中的 PII(不是删除日志,而是脱敏)
        # 审计日志通常需要保留,但要移除个人身份信息
        self._anonymize_logs(user_id, tenant_id)
        deleted['logs_anonymized'] = True
        
        # 7. 提交删除确认
        self.db.commit()
        
        return deleted
    
    def _anonymize_logs(self, user_id: str, tenant_id: str):
        """脱敏日志中的用户信息(而非删除日志)"""
        # 将用户标识替换为匿名 ID
        anonymous_id = f"deleted_user_{hashlib.sha256(user_id.encode()).hexdigest()[:8]}"
        # 批量更新日志表
        self.db.query(AuditLog).filter(
            AuditLog.user_id == user_id,
            AuditLog.tenant_id == tenant_id
        ).update({'user_id': anonymous_id, 'user_ip': None, 'user_email': None})
    
    async def _schedule_deletion(self, user_id: str, tenant_id: str, 
                                  deletion_date: datetime):
        """安排30天后的删除"""
        scheduled = ScheduledDeletion(
            user_id=user_id,
            tenant_id=tenant_id,
            scheduled_date=deletion_date,
            status='pending'
        )
        self.db.add(scheduled)
        self.db.commit()

# API 端点
"""
@app.delete("/api/gdpr/data")
async def request_data_deletion(request: Request):
    user_id = request.state.user_id
    tenant_id = request.state.tenant_id
    
    deletion_service = DataDeletionService(db, vector_store, storage)
    result = await deletion_service.delete_user_data(user_id, tenant_id)
    
    return result

@app.post("/api/gdpr/data/cancel-deletion")
async def cancel_deletion(request: Request):
    # 30天宽限期内可以取消删除
    ...
"""

Cookie 同意



<div id="cookie-consent" class="cookie-banner" style="display:none">
  <h3>🍪 Cookie & 数据处理同意</h3>
  <p>我们使用以下类型的 Cookie 和数据处理:</p>
  
  <div class="cookie-options">
    <label>
      <input type="checkbox" id="essential" checked disabled>
      <strong>必要 Cookie</strong> — 会话管理、安全(始终启用)
    </label>
    
    <label>
      <input type="checkbox" id="functional">
      <strong>功能 Cookie</strong> — 记住偏好设置
    </label>
    
    <label>
      <input type="checkbox" id="analytics">
      <strong>分析 Cookie</strong> — 使用统计(匿名化)
    </label>
    
    <label>
      <input type="checkbox" id="ai_training">
      <strong>AI 训练数据</strong> — 允许对话数据用于模型改进
    </label>
  </div>
  
  <div class="cookie-actions">
    <button onclick="acceptAll()">全部接受</button>
    <button onclick="acceptSelected()">仅接受选中</button>
    <button onclick="rejectOptional()">仅必要</button>
  </div>
  
  <p class="cookie-notice">
    您可以随时在设置中更改偏好。根据 GDPR,您有权访问、导出、删除您的数据。
    <a href="/privacy">隐私政策</a> | <a href="/gdpr/rights">您的权利</a>
  </p>
</div>

<script>
function acceptSelected() {
  const consent = {
    essential: true,
    functional: document.getElementById('functional').checked,
    analytics: document.getElementById('analytics').checked,
    ai_training: document.getElementById('ai_training').checked,
    timestamp: new Date().toISOString(),
    version: '1.0'
  };
  
  localStorage.setItem('cookie_consent', JSON.stringify(consent));
  document.getElementById('cookie-consent').style.display = 'none';
  
  // 根据同意设置初始化服务
  if (consent.analytics) initAnalytics();
  if (consent.ai_training) enableAITrainingOptIn();
  
  // 发送同意记录到服务器(用于合规审计)
  fetch('/api/gdpr/consent', {
    method: 'POST',
    headers: {'Content-Type': 'application/json'},
    body: JSON.stringify(consent)
  });
}
</script>

数据跨境传输

## 数据跨境传输方案

### EU → US 传输
1. EU-US Data Privacy Framework (DPF)
   - 2023年欧盟-美国数据隐私框架生效
   - 美国公司需通过 DPF 认证
   - OpenAI 已通过 DPF 认证
   - Anthropic 已通过 DPF 认证

2. 标准合同条款 (SCCs)
   - 如果提供商未通过 DPF
   - 使用欧盟委员会批准的 SCCs 模板
   - 需要进行传输影响评估 (TIA)

3. 补充措施
   - 加密传输(TLS 1.3)
   - 加密存储(AES-256)
   - 零数据保留协议(ZDR)
   - Anthropic 提供 Zero Data Retention 选项

### EU → China 传输
   - 中国不在欧盟"充分保护"名单内
   - 需要 SCCs + 补充措施
   - 同时需要满足中国数据出境安全评估

### API 提供商的合规情况

| 提供商 | DPF 认证 | DPA | ZDR | 数据驻留选项 |
|--------|---------|-----|-----|------------|
| OpenAI | ✅ | ✅ | API默认不训练 | 美国 |
| Anthropic | ✅ | 企业版 | ✅ | 美国 |
| Azure OpenAI | ✅ | ✅ | ✅ | 多区域 |
| AWS Bedrock | ✅ | ✅ | ✅ | 多区域 |

### 推荐做法
对于有严格数据驻留要求的客户:
1. 使用 Azure OpenAI(可在 EU 区域部署)
2. 或使用 AWS Bedrock(可在 EU 区域部署)
3. 签署 DPA + 启用 ZDR
4. 进行 DPIA(数据保护影响评估)

DPIA(数据保护影响评估)模板

# 数据保护影响评估 (DPIA)
## AI 对话助手产品

### 1. 处理描述
- **处理目的**: 提供 AI 对话服务
- **数据类型**: 用户输入文本、对话历史、用户偏好
- **数据主体**: 终端用户
- **处理者**: [OpenAI/Anthropic/Azure OpenAI]
- **数据流向**: 用户 → 应用服务器 → LLM API → 应用服务器 → 用户

### 2. 必要性和比例性
- 存储对话历史:必要(维持对话连续性)
- 发送至 LLM API:必要(核心功能)
- 记录 IP 地址:必要(安全防护)
- 用于模型训练:不必要(应默认关闭)

### 3. 风险评估
| 风险 | 可能性 | 影响 | 缓解措施 |
|------|--------|------|---------|
| 对话数据泄露 | 中 | 高 | 加密存储+传输、数据隔离 |
| LLM 输出含 PII | 中 | 高 | 输出过滤、PII 脱敏 |
| 数据被用于训练 | 低 | 高 | API 默认不训练、ZDR |
| 跨境传输截获 | 低 | 高 | TLS 1.3、ZDR |
| 系统提示词提取 | 中 | 中 | 输入验证、输出过滤 |

### 4. 缓解措施
- [列出所有技术和组织措施]

### 5. 残余风险
- [列出所有无法完全缓解的风险]
- [说明风险是否可接受]

### 6. DPO 意见
- [数据保护官的审查意见]

### 7. 审查计划
- 至少每年审查一次
- 重大变更时重新评估

GDPR 合规检查清单

## GDPR 合规检查清单

### 法律基础
[ ] 确定每类数据处理的合法基础
[ ] 获取用户明确同意(如用于 AI 训练)
[ ] 同意可以随时撤回
[ ] 同意记录可审计

### 透明度
[ ] 隐私政策包含 AI 处理说明
[ ] 说明数据发送到哪个 LLM 提供商
[ ] 说明数据保留期限
[ ] 说明用户权利
[ ] Cookie 同意横幅(可按类别选择)

### 数据主体权利
[ ] 数据访问权(用户可查看自己的数据)
[ ] 数据导出权(JSON/CSV 格式导出)
[ ] 数据删除权(30天内完成删除)
[ ] 数据更正权
[ ] 反对自动化决策权
[ ] 数据可携带权

### 数据处理协议
[ ] 与 OpenAI/Anthropic 签署 DPA
[ ] 与所有子处理者签署 DPA
[ ] 维护处理者清单

### 数据安全
[ ] 传输加密(TLS 1.3)
[ ] 存储加密(AES-256)
[ ] 访问控制(最小权限)
[ ] 数据隔离(多租户)
[ ] 日志审计

### 跨境传输
[ ] 确认 API 提供商的 DPF 认证状态
[ ] 或签署 SCCs
[ ] 进行传输影响评估

### DPIA
[ ] 完成 DPIA(AI 产品通常需要)
[ ] 指定 DPO(数据保护官)
[ ] 与监管机构协商(如高风险)

### 数据泄露
[ ] 72小时内通知监管机构
[ ] 通知受影响用户
[ ] 泄露响应流程

← 上一篇:数据隔离 | 下一篇:中国合规 → | 返回首页