CTO 视角下的数据安全：从“边界防御”到“数据血缘”的战略跃迁

@Component
public class SensitiveDataInterceptor {

    @Autowired
    private KmsClient kmsClient; // 对接云KMS或自建Vault

    public String encryptSensitive(String plainText) {
        // 1. 从KMS获取数据密钥（DEK），而非硬编码密钥
        String dek = kmsClient.getDataKey("user_phone_key");
        
        // 2. 使用GCM模式（提供完整性校验，防止篡改）
        SecretKey secretKey = new SecretKeySpec(dek.getBytes(), "AES");
        Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding");
        cipher.init(Cipher.ENCRYPT_MODE, secretKey);
        
        byte[] cipherText = cipher.doFinal(plainText.getBytes(StandardCharsets.UTF_8));
        // 3. 将IV（初始化向量）与密文拼接存储，便于解密
        return Base64.getEncoder().encodeToString(cipher.getIV()) + ":" 
             + Base64.getEncoder().encodeToString(cipherText);
    }
}

apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: default
  namespace: prod
spec:
  mtls:
    mode: STRICT  # 强制要求所有入站请求必须携带有效证书
---
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: db-policy
spec:
  host: postgres.prod.svc.cluster.local
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
    tls:
      mode: ISTIO_MUTUAL # 开启 Istio 自动颁发的 mTLS

package data_security

default allow = false

# 允许规则：仅在 9:00-18:00 内，且操作频率低于 100条/分钟
allow {
    input.method == "GET"
    input.path == "/api/export/sensitive"
    
    # 时间检查
    now := time.now_ns()
    hour := time.hour(now)
    hour >= 9
    hour <= 18
    
    # 频次检查（依赖外部计数器输入）
    input.rate_limit <= 100
}

# 强制告警：任何尝试访问 "backup" 表的行为直接拒绝
deny[msg] {
    input.sql contains "backup"
    msg = "高风险操作：禁止直接访问备份数据表，请通过审批流程"
}

import boto3
import random

def validate_backup_integrity(bucket, object_key):
    s3 = boto3.client('s3')
    # 1. 随机抽取备份文件中的 10KB 片段进行 MD5 比对
    response = s3.get_object(Bucket=bucket, Key=object_key, Range='bytes=0-10240')
    local_md5 = hashlib.md5(response['Body'].read()).hexdigest()
    
    # 2. 与备份时记录的元数据 MD5 比对
    metadata = s3.head_object(Bucket=bucket, Key=object_key)['Metadata']
    if metadata.get('original_md5') != local_md5:
        raise Exception(f"备份文件 {object_key} 已损坏或被篡改！")
    
    print(f"备份校验通过，RPO（恢复点目标）符合要求。")

# 该脚本应每月由混沌工程（Chaos Engineering）主动触发