“

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

7 级数据分类体系：从公开到绝密的安全标记框架

#TL;DR

coomia-dip 实现了 7 级数据分类体系（A1-公开、A2-内部公开、B1-内部敏感、B2-机密、C1-高度机密、C2-受限、D-绝密），为权限控制、动态脱敏和合规审计提供统一的元数据基础。分类标记嵌入 Ontology Schema，通过 Protobuf 在 gRPC 调用链中自动传播。本文从分类设计理念、技术实现、策略引擎集成到运维管理，完整阐述这一企业级数据分类框架。

#1. 数据分类的必要性

#1.1 无分类的混乱

在缺乏统一数据分类体系的平台中，安全策略难以一致实施：

权限粒度不足：无法区分"普通客户信息"和"客户身份证号"的访问级别
脱敏策略模糊：开发人员凭经验判断哪些字段需要脱敏，导致保护不一致
合规风险：无法证明对不同敏感度的数据采取了相应的保护措施
数据治理盲区：元数据目录中缺少分类维度，无法按敏感度检索和管理数据

#1.2 7 级分类的设计动机

coomia-dip 的 7 级分类体系设计考量：

等级	标识	名称	说明	典型数据
1	A1	公开	可公开发布的数据	产品目录、公告
2	A2	内部公开	组织内部可自由访问	内部知识库、流程文档
3	B1	内部敏感	需按需访问	业务报告、项目数据
4	B2	机密	受限范围访问	财务数据、合同
5	C1	高度机密	严格访问控制	客户 PII、医疗记录
6	C2	受限	最小必要访问	密钥材料、安全配置
7	D	绝密	仅名义持有人	国防/监管核心数据

#1.3 对标 Palantir Foundry

能力	Palantir Foundry	coomia-dip
分类层级	多层 Marking	7 级分类（A1-D）
自动标记	部分支持	基于规则引擎自动标记
传播机制	内置但不透明	Protobuf 显式传播
继承规则	就高原则	可配置继承策略
与脱敏集成	内置	ABAC 驱动映射

#2. 分类模型设计

#2.1 分类等级枚举

Python

from enum import IntEnum

class ClassificationLevel(IntEnum):
    """数据分类等级 - 7 级体系"""

    PUBLIC = 1           # A1 - 公开
    INTERNAL_PUBLIC = 2  # A2 - 内部公开
    INTERNAL_SENSITIVE = 3  # B1 - 内部敏感
    CONFIDENTIAL = 4     # B2 - 机密
    HIGHLY_CONFIDENTIAL = 5  # C1 - 高度机密
    RESTRICTED = 6       # C2 - 受限
    TOP_SECRET = 7       # D  - 绝密

    @property
    def code(self) -> str:
        """获取分类代码"""
        return {
            1: "A1", 2: "A2", 3: "B1", 4: "B2",
            5: "C1", 6: "C2", 7: "D",
        }[self.value]

    @property
    def requires_encryption_at_rest(self) -> bool:
        """是否需要静态加密"""
        return self.value >= 4  # B2 及以上

    @property
    def requires_audit_on_access(self) -> bool:
        """是否需要访问审计"""
        return self.value >= 3  # B1 及以上

    @property
    def max_retention_days(self) -> int | None:
        """最大保留天数（合规要求）"""
        return {
            5: 365 * 3,   # C1: 3 年
            6: 365 * 5,   # C2: 5 年
            7: 365 * 10,  # D: 10 年
        }.get(self.value)

#2.2 分类标记模型

Python

class ClassificationLabel(BaseModel):
    """数据分类标记"""

    level: ClassificationLevel = Field(description="分类等级")
    category: DataCategory = Field(description="数据类别")
    sub_category: str | None = Field(default=None, description="子类别")

    # 标记元数据
    labeled_by: str = Field(description="标记人/系统")
    labeled_at: datetime = Field(description="标记时间")
    review_status: ReviewStatus = Field(default=ReviewStatus.PENDING)
    reviewed_by: str | None = Field(default=None)
    reviewed_at: datetime | None = Field(default=None)

    # 标记来源
    source: LabelSource = Field(description="标记来源")
    confidence: float = Field(default=1.0, ge=0.0, le=1.0, description="置信度")

    # 合规标签
    regulatory_tags: list[str] = Field(
        default_factory=list,
        description="关联法规标签（如 GDPR, HIPAA）",
    )


class DataCategory(str, Enum):
    """数据类别"""
    PII = "pii"                    # 个人身份信息
    PHI = "phi"                    # 受保护健康信息
    PCI = "pci"                    # 支付卡信息
    FINANCIAL = "financial"        # 财务数据
    BUSINESS = "business"          # 业务数据
    TECHNICAL = "technical"        # 技术数据
    OPERATIONAL = "operational"    # 运营数据


class LabelSource(str, Enum):
    """标记来源"""
    MANUAL = "manual"              # 人工标记
    RULE_ENGINE = "rule_engine"    # 规则引擎自动标记
    ML_CLASSIFIER = "ml_classifier"  # ML 分类器
    INHERITED = "inherited"        # 继承自上游
    SCHEMA_DEFAULT = "schema_default"  # Schema 默认值

#2.3 Schema 集成

分类标记嵌入 Ontology Schema 的属性定义中：

Python

class PropertyDefinition(BaseModel):
    """Ontology 属性定义 - 包含分类标记"""

    name: str
    display_name: str
    data_type: DataType
    description: str = ""

    # 数据分类
    classification: ClassificationLabel | None = Field(
        default=None,
        description="属性的数据分类标记",
    )

    # 分类约束
    min_classification: ClassificationLevel | None = Field(
        default=None,
        description="此属性允许的最低分类等级",
    )

    # 字段级安全策略
    field_security: FieldSecurityPolicy | None = Field(default=None)


class ObjectTypeDefinition(BaseModel):
    """对象类型定义 - 包含对象级分类"""

    api_name: str
    display_name: str
    description: str

    # 对象级分类（所有属性的最高分类等级）
    effective_classification: ClassificationLevel = Field(
        default=ClassificationLevel.PUBLIC,
    )

    properties: dict[str, PropertyDefinition]

    def compute_effective_classification(self) -> ClassificationLevel:
        """计算对象的有效分类等级（取所有属性的最高值）"""
        max_level = ClassificationLevel.PUBLIC
        for prop in self.properties.values():
            if prop.classification and prop.classification.level > max_level:
                max_level = prop.classification.level
        return max_level

#3. 自动分类引擎

#3.1 基于规则的自动分类

Python

class RuleBasedClassifier:
    """基于规则的自动分类引擎"""

    def __init__(self):
        self._rules: list[ClassificationRule] = []
        self._load_default_rules()

    def _load_default_rules(self):
        """加载默认分类规则"""
        self._rules = [
            # PII 规则
            ClassificationRule(
                name="pii_id_number",
                pattern=r"(身份证|id_card|ssn|social_security)",
                field_type_match=["string"],
                classification=ClassificationLevel.HIGHLY_CONFIDENTIAL,
                category=DataCategory.PII,
                confidence=0.95,
            ),
            ClassificationRule(
                name="pii_phone",
                pattern=r"(手机|电话|phone|mobile|tel)",
                field_type_match=["string"],
                classification=ClassificationLevel.CONFIDENTIAL,
                category=DataCategory.PII,
                confidence=0.90,
            ),
            ClassificationRule(
                name="pii_email",
                pattern=r"(邮箱|email|mail)",
                field_type_match=["string"],
                classification=ClassificationLevel.CONFIDENTIAL,
                category=DataCategory.PII,
                confidence=0.90,
            ),
            ClassificationRule(
                name="pii_name",
                pattern=r"(姓名|真名|real_name|full_name)",
                field_type_match=["string"],
                classification=ClassificationLevel.INTERNAL_SENSITIVE,
                category=DataCategory.PII,
                confidence=0.85,
            ),
            # 金融数据规则
            ClassificationRule(
                name="financial_account",
                pattern=r"(银行卡|账号|bank_card|account_number)",
                field_type_match=["string"],
                classification=ClassificationLevel.HIGHLY_CONFIDENTIAL,
                category=DataCategory.PCI,
                confidence=0.95,
            ),
            ClassificationRule(
                name="financial_salary",
                pattern=r"(薪资|工资|salary|compensation)",
                field_type_match=["decimal", "float", "integer"],
                classification=ClassificationLevel.CONFIDENTIAL,
                category=DataCategory.FINANCIAL,
                confidence=0.90,
            ),
            # 健康数据规则
            ClassificationRule(
                name="phi_diagnosis",
                pattern=r"(诊断|病历|diagnosis|medical_record)",
                field_type_match=["string", "text"],
                classification=ClassificationLevel.HIGHLY_CONFIDENTIAL,
                category=DataCategory.PHI,
                confidence=0.95,
            ),
        ]

    def classify_field(
        self,
        field_name: str,
        field_type: str,
        description: str = "",
    ) -> ClassificationLabel | None:
        """对字段进行自动分类"""
        best_match: ClassificationRule | None = None
        best_confidence = 0.0

        text_to_match = f"{field_name} {description}".lower()

        for rule in self._rules:
            if field_type not in rule.field_type_match:
                continue
            if re.search(rule.pattern, text_to_match, re.IGNORECASE):
                if rule.confidence > best_confidence:
                    best_match = rule
                    best_confidence = rule.confidence

        if best_match is None:
            return None

        return ClassificationLabel(
            level=best_match.classification,
            category=best_match.category,
            labeled_by="rule_engine",
            labeled_at=datetime.utcnow(),
            source=LabelSource.RULE_ENGINE,
            confidence=best_match.confidence,
        )

#3.2 分类继承与传播

当数据在 Ontology 中流转时，分类标记需要正确传播：

Python

class ClassificationPropagator:
    """分类标记传播器"""

    class PropagationPolicy(str, Enum):
        HIGH_WATER_MARK = "high_water_mark"  # 就高原则（默认）
        EXPLICIT_ONLY = "explicit_only"      # 仅显式标记
        INHERIT_SOURCE = "inherit_source"    # 继承源标记

    def propagate_on_transform(
        self,
        source_classifications: list[ClassificationLabel],
        transform_type: str,
        policy: PropagationPolicy = PropagationPolicy.HIGH_WATER_MARK,
    ) -> ClassificationLabel:
        """数据变换时传播分类标记"""
        if policy == self.PropagationPolicy.HIGH_WATER_MARK:
            # 就高原则：取所有源的最高分类等级
            max_level = max(c.level for c in source_classifications)
            categories = set(c.category for c in source_classifications)
            regulatory_tags = set()
            for c in source_classifications:
                regulatory_tags.update(c.regulatory_tags)

            return ClassificationLabel(
                level=max_level,
                category=categories.pop() if len(categories) == 1 else DataCategory.BUSINESS,
                labeled_by="propagation_engine",
                labeled_at=datetime.utcnow(),
                source=LabelSource.INHERITED,
                confidence=min(c.confidence for c in source_classifications),
                regulatory_tags=list(regulatory_tags),
            )

        elif policy == self.PropagationPolicy.EXPLICIT_ONLY:
            raise ValueError("Explicit policy requires manual classification")

        elif policy == self.PropagationPolicy.INHERIT_SOURCE:
            return source_classifications[0].model_copy(
                update={
                    "source": LabelSource.INHERITED,
                    "labeled_at": datetime.utcnow(),
                }
            )

#4. Protobuf 传播协议

#4.1 分类标记 Protobuf 定义

PROTOBUF

syntax = "proto3";

package onto.classification.v1;

enum ClassificationLevel {
    CLASSIFICATION_LEVEL_UNSPECIFIED = 0;
    PUBLIC = 1;
    INTERNAL_PUBLIC = 2;
    INTERNAL_SENSITIVE = 3;
    CONFIDENTIAL = 4;
    HIGHLY_CONFIDENTIAL = 5;
    RESTRICTED = 6;
    TOP_SECRET = 7;
}

enum DataCategory {
    DATA_CATEGORY_UNSPECIFIED = 0;
    PII = 1;
    PHI = 2;
    PCI = 3;
    FINANCIAL = 4;
    BUSINESS = 5;
    TECHNICAL = 6;
    OPERATIONAL = 7;
}

message ClassificationLabel {
    ClassificationLevel level = 1;
    DataCategory category = 2;
    string sub_category = 3;
    string labeled_by = 4;
    google.protobuf.Timestamp labeled_at = 5;
    string source = 6;
    double confidence = 7;
    repeated string regulatory_tags = 8;
}

message FieldClassification {
    string field_name = 1;
    ClassificationLabel label = 2;
}

message ObjectClassification {
    string object_type = 1;
    ClassificationLevel effective_level = 2;
    repeated FieldClassification field_classifications = 3;
}

#4.2 gRPC 元数据传播

分类信息通过 gRPC metadata 在服务间传播：

Python

class ClassificationInterceptor(grpc.aio.UnaryUnaryClientInterceptor):
    """gRPC 客户端拦截器 - 传播分类上下文"""

    async def intercept_unary_unary(self, continuation, client_call_details, request):
        # 从当前上下文获取最高分类等级
        current_classification = classification_context.get()

        if current_classification:
            metadata = list(client_call_details.metadata or [])
            metadata.append((
                "x-classification-level",
                str(current_classification.level.value),
            ))
            metadata.append((
                "x-classification-category",
                current_classification.category.value,
            ))

            new_details = client_call_details._replace(metadata=metadata)
            return await continuation(new_details, request)

        return await continuation(client_call_details, request)

#5. 与安全子系统的集成

#5.1 权限控制集成

分类等级直接影响 ABAC 策略评估：

Python

class ClassificationBasedAccessControl:
    """基于分类等级的访问控制"""

    # 用户安全许可 → 可访问的最高分类等级
    CLEARANCE_MAP = {
        "public_user": ClassificationLevel.PUBLIC,
        "internal_user": ClassificationLevel.INTERNAL_SENSITIVE,
        "confidential_user": ClassificationLevel.CONFIDENTIAL,
        "secret_user": ClassificationLevel.HIGHLY_CONFIDENTIAL,
        "top_secret_user": ClassificationLevel.TOP_SECRET,
    }

    def check_access(
        self,
        user_clearance: str,
        resource_classification: ClassificationLevel,
    ) -> AccessDecision:
        max_allowed = self.CLEARANCE_MAP.get(user_clearance, ClassificationLevel.PUBLIC)

        if resource_classification <= max_allowed:
            return AccessDecision.ALLOW
        else:
            return AccessDecision.DENY

#5.2 脱敏集成

分类等级决定脱敏模式的选择（详见 S6-06）：

Python

CLASSIFICATION_MASKING_MAP = {
    ClassificationLevel.PUBLIC: MaskingMode.NONE,
    ClassificationLevel.INTERNAL_PUBLIC: MaskingMode.NONE,
    ClassificationLevel.INTERNAL_SENSITIVE: MaskingMode.PARTIAL,
    ClassificationLevel.CONFIDENTIAL: MaskingMode.PARTIAL,
    ClassificationLevel.HIGHLY_CONFIDENTIAL: MaskingMode.HASH,
    ClassificationLevel.RESTRICTED: MaskingMode.FULL,
    ClassificationLevel.TOP_SECRET: MaskingMode.FULL,
}

#5.3 审计集成

高分类等级的数据访问自动触发审计记录：

Python

class ClassificationAuditTrigger:
    """分类驱动的审计触发器"""

    AUDIT_THRESHOLDS = {
        ClassificationLevel.INTERNAL_SENSITIVE: AuditLevel.BASIC,
        ClassificationLevel.CONFIDENTIAL: AuditLevel.STANDARD,
        ClassificationLevel.HIGHLY_CONFIDENTIAL: AuditLevel.DETAILED,
        ClassificationLevel.RESTRICTED: AuditLevel.COMPREHENSIVE,
        ClassificationLevel.TOP_SECRET: AuditLevel.FORENSIC,
    }

    async def on_data_access(
        self,
        classification: ClassificationLevel,
        access_event: DataAccessEvent,
    ) -> None:
        audit_level = self.AUDIT_THRESHOLDS.get(classification)
        if audit_level:
            await self._audit_service.record(
                event=access_event,
                level=audit_level,
                classification=classification,
            )

#6. 分类管理工作流

#6.1 初始分类流程

Code

Schema 注册 → 自动分类引擎 → 人工审核 → 标记生效
     │              │              │          │
     │         规则匹配         安全团队     ABAC 策略
     │         ML 分类器        合规团队     脱敏策略
     │                                      审计策略

#6.2 分类变更流程

Python

class ClassificationChangeRequest(BaseModel):
    """分类变更请求"""

    id: str
    object_type: str
    field_name: str
    current_classification: ClassificationLevel
    requested_classification: ClassificationLevel
    reason: str
    requester: str
    requested_at: datetime

    # 审批流程
    approvers: list[str] = Field(default_factory=list)
    approval_status: ApprovalStatus = Field(default=ApprovalStatus.PENDING)

    @validator("requested_classification")
    def validate_downgrade(cls, v, values):
        """降级分类需要额外审批"""
        current = values.get("current_classification")
        if current and v < current:
            # 降级需要安全团队审批
            values["approvers"] = ["security_team_lead", "compliance_officer"]
        return v

#6.3 定期审查

Python

class ClassificationReviewScheduler:
    """分类审查调度器"""

    REVIEW_INTERVALS = {
        ClassificationLevel.TOP_SECRET: timedelta(days=90),
        ClassificationLevel.RESTRICTED: timedelta(days=180),
        ClassificationLevel.HIGHLY_CONFIDENTIAL: timedelta(days=365),
        ClassificationLevel.CONFIDENTIAL: timedelta(days=365 * 2),
    }

    async def schedule_reviews(self) -> list[ReviewTask]:
        """生成待审查任务列表"""
        tasks = []
        for obj_type in await self._schema_registry.list_object_types():
            for field_name, prop in obj_type.properties.items():
                if prop.classification is None:
                    continue
                interval = self.REVIEW_INTERVALS.get(prop.classification.level)
                if interval is None:
                    continue
                if datetime.utcnow() - prop.classification.labeled_at > interval:
                    tasks.append(ReviewTask(
                        object_type=obj_type.api_name,
                        field_name=field_name,
                        current_classification=prop.classification,
                        due_date=prop.classification.labeled_at + interval,
                    ))
        return tasks

#7. 测试策略

#7.1 分类正确性测试

Python

class TestClassification:
    def test_auto_classify_pii_fields(self):
        classifier = RuleBasedClassifier()
        result = classifier.classify_field("id_card_number", "string", "用户身份证号")
        assert result is not None
        assert result.level == ClassificationLevel.HIGHLY_CONFIDENTIAL
        assert result.category == DataCategory.PII

    def test_propagation_high_water_mark(self):
        propagator = ClassificationPropagator()
        sources = [
            make_label(ClassificationLevel.INTERNAL_SENSITIVE),
            make_label(ClassificationLevel.CONFIDENTIAL),
        ]
        result = propagator.propagate_on_transform(sources, "join")
        assert result.level == ClassificationLevel.CONFIDENTIAL

    def test_object_effective_classification(self):
        obj = ObjectTypeDefinition(
            api_name="Customer",
            display_name="客户",
            description="客户对象",
            properties={
                "name": PropertyDefinition(
                    name="name", display_name="姓名", data_type=DataType.STRING,
                    classification=make_label(ClassificationLevel.INTERNAL_SENSITIVE),
                ),
                "id_card": PropertyDefinition(
                    name="id_card", display_name="身份证", data_type=DataType.STRING,
                    classification=make_label(ClassificationLevel.HIGHLY_CONFIDENTIAL),
                ),
            },
        )
        assert obj.compute_effective_classification() == ClassificationLevel.HIGHLY_CONFIDENTIAL

#8. 生产最佳实践

#8.1 分类治理原则

默认保守：未分类字段默认为 B1（内部敏感），而非公开
就高原则：数据组合或变换后，取各源数据的最高分类等级
定期审查：C1 及以上每年至少审查一次，D 级别每季度审查
变更审计：所有分类变更都记录审计日志，降级变更需要双重审批

#8.2 法规映射

法规	相关数据类别	建议最低分类
GDPR	PII	B2（机密）
HIPAA	PHI	C1（高度机密）
PCI DSS	PCI	C1（高度机密）
中国《个保法》	PII	B2（机密）
SOX	财务	B2（机密）

#8.3 运维建议

新增 Ontology 对象类型时自动触发分类引擎
分类结果需要人工审核后才能作为 ABAC 策略输入
建立分类仪表盘，监控各等级数据的分布和访问模式
对 C2/D 级别数据的分类变更设置告警通知

#9. 总结

coomia-dip 的 7 级数据分类体系通过精细化的分类等级划分，为安全子系统提供了统一的元数据基础。关键设计亮点：

7 级精细度：覆盖从公开到绝密的完整敏感度谱系
自动分类：规则引擎 + ML 分类器实现高置信度自动标记
传播机制：分类标记随数据流转自动传播，确保一致性
系统集成：与权限控制、动态脱敏、审计三大子系统深度集成
合规对齐：内置 GDPR、HIPAA、PCI DSS 等法规映射

下一篇将深入探讨 coomia-dip 的审计追踪系统及其 13 类审计事件设计。