performing-privacy-impact-assessment

执行隐私影响评估

适用场景

• 启动处理个人数据的新系统、产品或处理活动时
• 执行 GDPR 第 35 条规定的数据保护影响评估（DPIA）时
• 评估数据处理操作是否符合 CCPA/CPRA 时
• 开展符合 NIST 隐私框架的隐私风险评估时
• 绘制跨组织边界和第三方处理方的数据流图时
• 构建自动化隐私治理和评估流水线时
• 准备监管审计或证明问责义务履行情况时

前提条件

• 熟悉 GDPR、CCPA/CPRA 和 NIST 隐私框架的概念
• 能访问数据处理清单和系统架构文档
• 已安装所需依赖项的 Python 3.8+
• 已获得数据保护官（DPO）或隐私团队的适当授权
• 了解组织的数据流和第三方处理方关系

说明

阶段 1：数据清单和处理活动目录

建立完整的个人数据处理活动清单。每条处理活动记录（ROPA）都必须记录数据类别、法律依据、保留期限和涉及的数据主体。

from agent import PrivacyImpactAssessmentEngine

engine = PrivacyImpactAssessmentEngine()

# Register a processing activity for assessment
activity = engine.register_processing_activity(
    name="Customer Analytics Platform",
    description="Collects browsing behavior and purchase history for personalization",
    data_controller="Acme Corp",
    data_processor="CloudAnalytics Inc",
    data_categories=["browsing_history", "purchase_records", "ip_address", "device_id"],
    data_subjects=["customers", "website_visitors"],
    legal_basis="consent",
    retention_period_days=730,
    cross_border_transfer=True,
    transfer_destinations=["US", "IN"],
    automated_decision_making=True,
)
print(f"Registered activity: {activity['activity_id']}")

阶段 2：数据流映射

绘制从收集到删除的所有数据流，确定每个接触点、转换过程和存储位置。这有助于揭示数据在系统间流动时潜藏的隐私风险。

# Build the data flow map
flow_map = engine.map_data_flows(
    activity_id=activity["activity_id"],
    flows=[
        {
            "stage": "collection",
            "source": "Web browser cookie + form submission",
            "destination": "CDN edge server",
            "data_elements": ["ip_address", "device_id", "browsing_history"],
            "encryption_in_transit": True,
            "protocol": "TLS 1.3",
        },
        {
            "stage": "processing",
            "source": "CDN edge server",
            "destination": "Analytics data warehouse (US-East)",
            "data_elements": ["browsing_history", "purchase_records", "device_id"],
            "encryption_in_transit": True,
            "encryption_at_rest": True,
            "protocol": "mTLS",
        },
        {
            "stage": "storage",
            "source": "Analytics data warehouse",
            "destination": "S3 encrypted bucket",
            "data_elements": ["browsing_history", "purchase_records"],
            "encryption_at_rest": True,
            "retention_days": 730,
            "access_controls": "IAM role-based, MFA required",
        },
        {
            "stage": "sharing",
            "source": "Analytics data warehouse",
            "destination": "Third-party ML provider (IN)",
            "data_elements": ["browsing_history", "purchase_records"],
            "encryption_in_transit": True,
            "data_processing_agreement": True,
            "cross_border": True,
        },
        {
            "stage": "deletion",
            "source": "S3 bucket + data warehouse",
            "destination": "Secure erasure",
            "method": "Cryptographic erasure + lifecycle policy",
            "verification": "Automated deletion audit log",
        },
    ],
)
engine.render_data_flow_diagram(flow_map)

阶段 3：使用评分矩阵评估隐私风险

应用结构化风险评分方法，从多个隐私风险维度评估发生可能性和影响。该矩阵同时符合 NIST PRAM 和 ICO DPIA 风险评估方法。

# Run the risk assessment
risk_report = engine.assess_privacy_risks(
    activity_id=activity["activity_id"],
    assessment_type="full_dpia",
)

# Display risk matrix results
for risk in risk_report["risks"]:
    print(f"[{risk['severity']}] {risk['category']}: {risk['description']}")
    print(f"  Likelihood: {risk['likelihood']}/5 | Impact: {risk['impact']}/5 | Score: {risk['risk_score']}/25")
    print(f"  Mitigation: {risk['recommended_mitigation']}")

评估的风险类别包括：

1. 数据最小化 -- 超出既定目的过度收集数据
2. 目的限制 -- 未经法律依据将数据用于其他目的
3. 跨境传输 -- 未采取充分保障措施（SCC、BCR）即进行传输
4. 自动化决策 -- 未经人工监督或申诉机制进行画像分析
5. 数据主体权利 -- 无法满足访问、删除或可携权请求
6. 第三方风险 -- 处理方合规缺口、次级处理方链条
7. 安全控制措施 -- 加密、访问控制、违规响应方面的缺口
8. 保留期限 -- 超出必要期限或法律要求存储数据
9. 同意管理 -- 同意机制无效或含糊不清
10. 违规通知 -- 无法在 72 小时内（GDPR）发现违规并发出通知

阶段 4：GDPR 和 CCPA/CPRA 合规性检查

针对特定法规要求运行自动化合规检查。该引擎会依据 GDPR 条款级义务和 CCPA/CPRA 消费者权利要求，逐项核查处理活动。

# GDPR compliance check
gdpr_report = engine.check_gdpr_compliance(activity_id=activity["activity_id"])
print(f"GDPR Score: {gdpr_report['compliance_score']}/100")
for finding in gdpr_report["findings"]:
    print(f"  [{finding['status']}] Art.{finding['article']}: {finding['description']}")

# CCPA/CPRA compliance check
ccpa_report = engine.check_ccpa_compliance(activity_id=activity["activity_id"])
print(f"CCPA Score: {ccpa_report['compliance_score']}/100")
for finding in ccpa_report["findings"]:
    print(f"  [{finding['status']}] Sec.{finding['section']}: {finding['description']}")

阶段 5：生成整改计划和报告

生成优先级排序的整改计划，其中包含具体行动项、责任方和截止日期，并生成正式的 PIA/DPIA 报告文档。

# Generate remediation plan
remediation = engine.generate_remediation_plan(
    activity_id=activity["activity_id"],
    risk_report=risk_report,
    gdpr_report=gdpr_report,
    ccpa_report=ccpa_report,
)

for item in remediation["action_items"]:
    print(f"[{item['priority']}] {item['action']}")
    print(f"  Owner: {item['owner']} | Deadline: {item['deadline']}")
    print(f"  Addresses: {', '.join(item['addresses_risks'])}")

# Generate formal DPIA report
engine.generate_dpia_report(
    activity_id=activity["activity_id"],
    output_path="dpia_report_customer_analytics.json",
    format="json",
)
print("[+] DPIA report generated")

示例

快速筛查评估

使用 ICO 筛查清单判断是否需要进行完整的 DPIA：

engine = PrivacyImpactAssessmentEngine()

screening = engine.run_screening_checklist(
    uses_special_category_data=False,
    large_scale_processing=True,
    systematic_monitoring=True,
    automated_decision_making=True,
    cross_border_transfer=True,
    vulnerable_data_subjects=False,
    innovative_technology=True,
    denial_of_service_or_rights=False,
)
print(f"DPIA Required: {screening['dpia_required']}")
print(f"Triggers: {screening['triggers']}")
# Output: DPIA Required: True
# Triggers: ['large_scale_processing', 'systematic_monitoring',
#            'automated_decision_making', 'cross_border_transfer',
#            'innovative_technology']

对多个处理活动进行批量评估

engine = PrivacyImpactAssessmentEngine()

activities = [
    {"name": "Email Marketing", "data_categories": ["email", "name"],
     "legal_basis": "consent", "cross_border_transfer": False},
    {"name": "HR Analytics", "data_categories": ["employee_id", "performance_scores",
     "health_data"], "legal_basis": "legitimate_interest", "cross_border_transfer": True},
    {"name": "Fraud Detection", "data_categories": ["transaction_data", "ip_address",
     "device_fingerprint"], "legal_basis": "legitimate_interest",
     "automated_decision_making": True, "cross_border_transfer": False},
]

for act_def in activities:
    activity = engine.register_processing_activity(**act_def)
    risk = engine.assess_privacy_risks(activity_id=activity["activity_id"])
    print(f"{act_def['name']}: Overall Risk={risk['overall_risk_level']} "
          f"({risk['risk_count_by_severity']})")

NIST 隐私框架配置文件映射

engine = PrivacyImpactAssessmentEngine()

profile = engine.generate_nist_privacy_profile(
    activity_id=activity["activity_id"],
    target_tier="tier_3",  # Repeatable
)

for function_id, outcomes in profile["functions"].items():
    print(f"\n{function_id}:")
    for outcome in outcomes:
        status = "PASS" if outcome["implemented"] else "GAP"
        print(f"  [{status}] {outcome['subcategory']}: {outcome['description']}")