“

系列：S6 平台工程 · 第 12 篇 | 难度：高级 | 阅读时间：18 分钟

合规设计：多法规框架下的自动化合规引擎

#TL;DR

coomia-dip 的合规引擎支持 GDPR、HIPAA、PCI DSS、SOX 和中国《个人信息保护法》等多法规框架，通过声明式合规规则、自动化检查和持续监控，将合规要求转化为可执行的技术控制。引擎集成了数据分类、权限控制、脱敏、审计和数据保留五大子系统，提供合规仪表盘、差距分析和修复建议。本文从合规架构、规则引擎、自动化检查、报告生成到持续合规监控，完整解析这一企业级合规能力。

#1. 合规设计的核心挑战

#1.1 多法规并存

企业数据平台通常需要同时满足多个法规要求：

法规	适用范围	核心要求
GDPR	欧盟数据主体	数据最小化、同意、被遗忘权
HIPAA	美国健康数据	PHI 保护、访问控制、审计
PCI DSS	支付卡数据	卡号加密、访问限制
SOX	财务数据	审计追踪、内控
中国《个保法》	中国个人信息	告知同意、数据本地化

#1.2 合规即代码

coomia-dip 采用"Compliance-as-Code"理念，将合规要求编码为可执行的检查规则：

Code

法规条款 → 合规规则 → 技术控制 → 自动化检查 → 持续监控
   │           │           │           │           │
 人工解读    声明式定义   系统配置    定时执行    实时告警

#1.3 对标 Palantir Foundry

能力	Palantir Foundry	coomia-dip
多法规支持	部分内置	5+ 法规框架
合规检查	手动为主	自动化引擎
差距分析	咨询服务	内置分析
修复建议	无	自动生成
合规报告	手动生成	自动化报告

#2. 合规架构

#2.1 系统架构

Code

┌────────────────────────────────────────────────┐
│            Compliance Dashboard                 │
│     (合规仪表盘 + 差距分析 + 报告)               │
└───────────────────┬────────────────────────────┘
                    │
┌───────────────────▼────────────────────────────┐
│           Compliance Engine                     │
│  ┌──────────────┐  ┌────────────────────────┐  │
│  │ Rule Registry │  │ Check Executor         │  │
│  │ (规则注册表)   │  │ (检查执行器)            │  │
│  └──────┬───────┘  └──────────┬─────────────┘  │
│         │                     │                 │
│  ┌──────▼─────────────────────▼─────────────┐  │
│  │         Integration Layer                │  │
│  │  分类 │ 权限 │ 脱敏 │ 审计 │ 数据保留     │  │
│  └──────────────────────────────────────────┘  │
└────────────────────────────────────────────────┘

#2.2 合规规则模型

Python

class ComplianceRule(BaseModel):
    """合规规则定义"""

    rule_id: str = Field(description="规则唯一标识")
    regulation: Regulation = Field(description="所属法规")
    article: str = Field(description="法规条款")
    name: str = Field(description="规则名称")
    description: str = Field(description="规则描述")

    # 检查逻辑
    check_type: CheckType = Field(description="检查类型")
    check_config: dict = Field(description="检查配置")

    # 严重性
    severity: ComplianceSeverity = Field(description="严重性级别")

    # 适用条件
    applies_to: ApplicabilityFilter = Field(description="适用范围")

    # 修复建议
    remediation: str = Field(default="", description="修复建议")

    # 元数据
    enabled: bool = Field(default=True)
    last_updated: datetime = Field(default_factory=datetime.utcnow)


class Regulation(str, Enum):
    GDPR = "gdpr"
    HIPAA = "hipaa"
    PCI_DSS = "pci_dss"
    SOX = "sox"
    PIPL = "pipl"  # 中国《个人信息保护法》


class CheckType(str, Enum):
    CLASSIFICATION_CHECK = "classification_check"
    ACCESS_CONTROL_CHECK = "access_control_check"
    MASKING_CHECK = "masking_check"
    AUDIT_CHECK = "audit_check"
    RETENTION_CHECK = "retention_check"
    ENCRYPTION_CHECK = "encryption_check"
    CONSENT_CHECK = "consent_check"


class ComplianceSeverity(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"

#3. 内置合规规则

Python

GDPR_RULES = [
    ComplianceRule(
        rule_id="gdpr-001",
        regulation=Regulation.GDPR,
        article="Article 5(1)(c)",
        name="Data Minimization",
        description="个人数据必须仅包含处理目的所必需的数据",
        check_type=CheckType.CLASSIFICATION_CHECK,
        check_config={
            "require_classification": True,
            "require_purpose_tag": True,
            "max_retention_without_purpose": 365,
        },
        severity=ComplianceSeverity.HIGH,
        remediation="为所有包含 PII 的字段添加分类标记和用途说明",
    ),
    ComplianceRule(
        rule_id="gdpr-002",
        regulation=Regulation.GDPR,
        article="Article 17",
        name="Right to Erasure",
        description="必须能够删除数据主体的所有个人数据",
        check_type=CheckType.ACCESS_CONTROL_CHECK,
        check_config={
            "require_deletion_capability": True,
            "data_categories": ["pii"],
            "max_deletion_time_hours": 72,
        },
        severity=ComplianceSeverity.CRITICAL,
        remediation="确保所有 PII 数据支持通过 API 删除，并在 72 小时内完成",
    ),
    ComplianceRule(
        rule_id="gdpr-003",
        regulation=Regulation.GDPR,
        article="Article 30",
        name="Records of Processing",
        description="必须维护处理活动记录",
        check_type=CheckType.AUDIT_CHECK,
        check_config={
            "require_audit_trail": True,
            "event_types": ["data_access", "data_modification"],
            "min_retention_days": 365,
        },
        severity=ComplianceSeverity.HIGH,
        remediation="启用数据访问和数据变更的审计追踪，保留至少 1 年",
    ),
    ComplianceRule(
        rule_id="gdpr-004",
        regulation=Regulation.GDPR,
        article="Article 32",
        name="Security of Processing",
        description="个人数据必须加密存储和传输",
        check_type=CheckType.ENCRYPTION_CHECK,
        check_config={
            "require_encryption_at_rest": True,
            "require_encryption_in_transit": True,
            "min_classification": "B2",
        },
        severity=ComplianceSeverity.HIGH,
        remediation="对 B2 及以上分类的数据启用静态加密和传输加密",
    ),
]

#3.2 HIPAA 规则集

Python

HIPAA_RULES = [
    ComplianceRule(
        rule_id="hipaa-001",
        regulation=Regulation.HIPAA,
        article="164.312(a)(1)",
        name="Access Control",
        description="PHI 必须有访问控制机制",
        check_type=CheckType.ACCESS_CONTROL_CHECK,
        check_config={
            "data_categories": ["phi"],
            "require_rbac": True,
            "require_abac": True,
            "min_classification": "C1",
        },
        severity=ComplianceSeverity.CRITICAL,
    ),
    ComplianceRule(
        rule_id="hipaa-002",
        regulation=Regulation.HIPAA,
        article="164.312(b)",
        name="Audit Controls",
        description="PHI 访问必须有审计记录",
        check_type=CheckType.AUDIT_CHECK,
        check_config={
            "data_categories": ["phi"],
            "require_detailed_audit": True,
            "min_retention_days": 2190,  # 6 年
        },
        severity=ComplianceSeverity.CRITICAL,
    ),
    ComplianceRule(
        rule_id="hipaa-003",
        regulation=Regulation.HIPAA,
        article="164.312(e)(1)",
        name="Transmission Security",
        description="PHI 传输必须加密",
        check_type=CheckType.ENCRYPTION_CHECK,
        check_config={
            "data_categories": ["phi"],
            "require_tls": True,
            "min_tls_version": "1.2",
        },
        severity=ComplianceSeverity.CRITICAL,
    ),
]

#4. 合规检查执行器

#4.1 检查执行引擎

Python

class ComplianceCheckExecutor:
    """合规检查执行器"""

    def __init__(self):
        self._checkers: dict[CheckType, ComplianceChecker] = {
            CheckType.CLASSIFICATION_CHECK: ClassificationChecker(),
            CheckType.ACCESS_CONTROL_CHECK: AccessControlChecker(),
            CheckType.MASKING_CHECK: MaskingChecker(),
            CheckType.AUDIT_CHECK: AuditChecker(),
            CheckType.RETENTION_CHECK: RetentionChecker(),
            CheckType.ENCRYPTION_CHECK: EncryptionChecker(),
            CheckType.CONSENT_CHECK: ConsentChecker(),
        }

    async def run_check(
        self,
        rule: ComplianceRule,
        scope: CheckScope,
    ) -> CheckResult:
        """执行单条合规检查"""
        checker = self._checkers[rule.check_type]
        findings = await checker.check(rule.check_config, scope)

        return CheckResult(
            rule_id=rule.rule_id,
            regulation=rule.regulation,
            status=CheckStatus.PASS if not findings else CheckStatus.FAIL,
            findings=findings,
            checked_at=datetime.utcnow(),
            scope=scope,
        )

    async def run_full_assessment(
        self,
        regulation: Regulation | None = None,
    ) -> ComplianceAssessment:
        """执行完整合规评估"""
        rules = self._get_applicable_rules(regulation)
        results = []

        for rule in rules:
            if not rule.enabled:
                continue
            scope = self._determine_scope(rule)
            result = await self.run_check(rule, scope)
            results.append(result)

        return ComplianceAssessment(
            regulation=regulation,
            total_rules=len(results),
            passed=sum(1 for r in results if r.status == CheckStatus.PASS),
            failed=sum(1 for r in results if r.status == CheckStatus.FAIL),
            results=results,
            assessed_at=datetime.utcnow(),
            compliance_score=self._calculate_score(results),
        )

#4.2 分类检查器示例

Python

class ClassificationChecker(ComplianceChecker):
    """数据分类合规检查器"""

    async def check(
        self, config: dict, scope: CheckScope,
    ) -> list[ComplianceFinding]:
        findings = []

        for asset in await self._catalog.list_assets(scope=scope):
            for prop in asset.technical.properties:
                # 检查是否有分类标记
                if config.get("require_classification") and not prop.classification:
                    findings.append(ComplianceFinding(
                        severity=ComplianceSeverity.HIGH,
                        asset_id=asset.asset_id,
                        field_name=prop.name,
                        finding="字段缺少数据分类标记",
                        remediation="为该字段添加分类标记",
                    ))

                # 检查分类等级是否满足最低要求
                min_level = config.get("min_classification")
                if min_level and prop.classification:
                    if prop.classification.level.code < min_level:
                        findings.append(ComplianceFinding(
                            severity=ComplianceSeverity.MEDIUM,
                            asset_id=asset.asset_id,
                            field_name=prop.name,
                            finding=f"分类等级 {prop.classification.level.code} 低于要求 {min_level}",
                            remediation=f"将分类等级提升至 {min_level} 或以上",
                        ))

        return findings

#5. 差距分析

#5.1 差距分析引擎

Python

class GapAnalyzer:
    """合规差距分析引擎"""

    async def analyze_gaps(
        self,
        target_regulation: Regulation,
    ) -> GapAnalysisReport:
        """分析与目标法规的合规差距"""
        assessment = await self._executor.run_full_assessment(target_regulation)

        gaps = []
        for result in assessment.results:
            if result.status == CheckStatus.FAIL:
                rule = self._rule_registry.get(result.rule_id)
                gaps.append(ComplianceGap(
                    rule=rule,
                    findings=result.findings,
                    estimated_effort=self._estimate_effort(rule, result.findings),
                    priority=self._calculate_priority(rule, result.findings),
                ))

        # 按优先级排序
        gaps.sort(key=lambda g: g.priority, reverse=True)

        return GapAnalysisReport(
            regulation=target_regulation,
            compliance_score=assessment.compliance_score,
            total_gaps=len(gaps),
            critical_gaps=sum(1 for g in gaps if g.rule.severity == ComplianceSeverity.CRITICAL),
            gaps=gaps,
            remediation_plan=self._generate_plan(gaps),
        )

    def _estimate_effort(
        self, rule: ComplianceRule, findings: list[ComplianceFinding],
    ) -> EffortEstimate:
        """估算修复工作量"""
        base_effort = {
            CheckType.CLASSIFICATION_CHECK: timedelta(hours=2),
            CheckType.ACCESS_CONTROL_CHECK: timedelta(hours=8),
            CheckType.MASKING_CHECK: timedelta(hours=4),
            CheckType.AUDIT_CHECK: timedelta(hours=4),
            CheckType.RETENTION_CHECK: timedelta(hours=2),
            CheckType.ENCRYPTION_CHECK: timedelta(hours=16),
            CheckType.CONSENT_CHECK: timedelta(hours=8),
        }

        effort = base_effort.get(rule.check_type, timedelta(hours=4))
        multiplier = min(len(findings), 10) / 10 + 1

        return EffortEstimate(
            estimated_hours=effort.total_seconds() / 3600 * multiplier,
            complexity="high" if effort > timedelta(hours=8) else "medium",
        )

#6. 合规报告

#6.1 报告生成器

Python

class ComplianceReportGenerator:
    """合规报告生成器"""

    async def generate_report(
        self,
        regulation: Regulation,
        period: DateRange,
        report_type: ReportType = ReportType.FULL,
    ) -> ComplianceReport:
        """生成合规报告"""
        assessment = await self._executor.run_full_assessment(regulation)
        audit_summary = await self._audit_service.get_summary(period)
        gap_analysis = await self._gap_analyzer.analyze_gaps(regulation)

        return ComplianceReport(
            regulation=regulation,
            period=period,
            generated_at=datetime.utcnow(),
            executive_summary=self._build_executive_summary(
                assessment, gap_analysis,
            ),
            compliance_score=assessment.compliance_score,
            assessment_details=assessment,
            gap_analysis=gap_analysis,
            audit_summary=audit_summary,
            data_classification_summary=await self._get_classification_summary(),
            access_control_summary=await self._get_access_control_summary(),
            recommendations=self._generate_recommendations(gap_analysis),
        )

    def _build_executive_summary(
        self,
        assessment: ComplianceAssessment,
        gap_analysis: GapAnalysisReport,
    ) -> str:
        """生成执行摘要"""
        return (
            f"合规评分：{assessment.compliance_score:.1%}\n"
            f"检查规则：{assessment.total_rules} 条\n"
            f"通过：{assessment.passed} 条\n"
            f"未通过：{assessment.failed} 条\n"
            f"关键差距：{gap_analysis.critical_gaps} 项\n"
        )

#7. 持续合规监控

#7.1 定时检查调度

Python

class ComplianceMonitor:
    """持续合规监控"""

    SCHEDULE = {
        ComplianceSeverity.CRITICAL: timedelta(hours=1),
        ComplianceSeverity.HIGH: timedelta(hours=6),
        ComplianceSeverity.MEDIUM: timedelta(days=1),
        ComplianceSeverity.LOW: timedelta(days=7),
    }

    async def run_continuous_monitoring(self) -> None:
        """持续监控循环"""
        while True:
            for severity, interval in self.SCHEDULE.items():
                rules = self._get_rules_by_severity(severity)
                for rule in rules:
                    if self._is_due(rule, interval):
                        result = await self._executor.run_check(
                            rule, CheckScope.FULL,
                        )
                        if result.status == CheckStatus.FAIL:
                            await self._alert_service.send_alert(
                                ComplianceAlert(
                                    rule=rule,
                                    findings=result.findings,
                                    severity=severity,
                                ),
                            )
            await asyncio.sleep(300)  # 5 分钟检查间隔

#8. 测试策略

Python

class TestComplianceEngine:
    async def test_gdpr_classification_check(self):
        executor = ComplianceCheckExecutor()
        rule = GDPR_RULES[0]  # Data Minimization

        result = await executor.run_check(rule, CheckScope.FULL)
        assert result.status in (CheckStatus.PASS, CheckStatus.FAIL)

    async def test_hipaa_audit_check(self):
        executor = ComplianceCheckExecutor()
        rule = HIPAA_RULES[1]  # Audit Controls

        result = await executor.run_check(rule, CheckScope.FULL)
        if result.status == CheckStatus.FAIL:
            assert len(result.findings) > 0
            assert all(f.remediation for f in result.findings)

    async def test_gap_analysis(self):
        analyzer = GapAnalyzer(executor, rule_registry)
        report = await analyzer.analyze_gaps(Regulation.GDPR)
        assert report.compliance_score >= 0.0
        assert report.compliance_score <= 1.0

    async def test_report_generation(self):
        generator = ComplianceReportGenerator(executor, audit_service)
        report = await generator.generate_report(
            Regulation.GDPR,
            DateRange(start=datetime(2024, 1, 1), end=datetime(2024, 12, 31)),
        )
        assert report.executive_summary
        assert report.compliance_score is not None

#9. 生产最佳实践

#9.1 合规治理流程

初始差距分析 → 修复计划 → 实施 → 验证 → 持续监控
每季度执行一次完整合规评估
关键规则变更触发即时重新评估
合规报告自动归档，保留至少 7 年

#9.2 法规更新管理

订阅法规更新通知
评估新条款对现有规则的影响
新增或修改合规规则
执行差距分析确认影响范围
制定修复计划并跟踪执行

#9.3 跨法规协调

识别不同法规的重叠要求（如 GDPR 和 PIPL 的数据保护要求）
采用"最严格合规"策略：同一控制点采用最严格法规的要求
建立统一的合规控制矩阵

#10. 总结

coomia-dip 的合规引擎通过"Compliance-as-Code"理念，将多法规要求转化为自动化的技术控制。关键设计亮点：

多法规支持：内置 GDPR、HIPAA、PCI DSS、SOX、PIPL 规则集
自动化检查：7 类检查器覆盖分类、权限、脱敏、审计等维度
差距分析：自动识别合规差距并估算修复工作量
持续监控：按严重性分级的持续合规监控
报告生成：自动化合规报告满足审计要求

下一篇将深入探讨 coomia-dip SDK 的设计哲学。

合规设计：多法规框架下的自动化合规引擎

#TL;DR

#1. 合规设计的核心挑战

#1.1 多法规并存

#1.2 合规即代码

#1.3 对标 Palantir Foundry

#2. 合规架构

#2.1 系统架构

#2.2 合规规则模型

#3. 内置合规规则

#3.1 GDPR 规则集

#3.2 HIPAA 规则集

#4. 合规检查执行器

#4.1 检查执行引擎

#4.2 分类检查器示例

#5. 差距分析

#5.1 差距分析引擎

#6. 合规报告

#6.1 报告生成器

#7. 持续合规监控

#7.1 定时检查调度

#8. 测试策略

#9. 生产最佳实践

#9.1 合规治理流程

#9.2 法规更新管理

#9.3 跨法规协调

#10. 总结