企业AI数据泄露防护实战指南：三十六计护数据

# ❌ 危险做法一：硬编码
openai_api_key = "sk-proj-xxxxxxxxxxxx"

# ❌ 危险做法二：.env文件提交到Git
# .env 文件被推送到公开仓库

# ❌ 危险做法三：Docker镜像内嵌
# Dockerfile 中 ENV OPENAI_API_KEY=sk-xxx

# ❌ 危险做法四：日志打印
print(f"Using API key: {api_key}")

# ❌ 危险做法五：前端暴露
# JavaScript代码中直接调用API

# ❌ 危险做法六：配置文件明文存储
# config.json 中 "api_key": "sk-xxx"

# ❌ 危险做法七：截图/文档中泄露
# 技术文档截图包含API密钥

#!/bin/bash
# 自动密钥轮换脚本
VAULT_ADDR="http://localhost:8200"
KEY_PATH="secret/data/ai-services/openai"

# 生成新密钥
NEW_KEY=$(openssl rand -hex 32)

# 写入Vault
vault kv put $KEY_PATH api_key="$NEW_KEY" \
  rotation_date=$(date +%Y-%m-%d) \
  created_by="auto-rotation"

# 通知服务更新
systemctl reload ai-gateway

echo "密钥轮换完成: $(date)"

#!/bin/bash
# Git仓库密钥扫描脚本

echo "🔍 扫描代码仓库中的AI密钥..."

# 使用TruffleHog扫描
trufflehog filesystem ./src \
  --only-verified \
  --filter-entropy=4.5

# 使用GitGuardian扫描
ggshield scan path ./src \
  --ignore-known-secrets \
  --minimum-severity "medium"

# 自定义正则扫描
grep -rnE '(sk-[a-zA-Z0-9]{20,}|AIza[a-zA-Z0-9_-]{35}|AKIA[0-9A-Z]{16})' ./src/

echo "✅ 扫描完成"

import re
from typing import Dict, List
import logging

logger = logging.getLogger("ai-data-masker")


class AIDataMasker:
    """AI对话实时脱敏网关"""

    SENSITIVE_PATTERNS = {
        "手机号": (r"1[3-9]\d{9}", lambda m: m.group(0)[:3] + "****" + m.group(0)[-4:]),
        "身份证": (r"\d{17}[\dXx]", lambda m: m.group(0)[:6] + "********" + m.group(0)[-4:]),
        "银行卡": (r"\d{16,19}", lambda m: "**** **** **** " + m.group(0)[-4:]),
        "邮箱": (r"[\w.-]+@[\w.-]+\.[\w]+", lambda m: m.group(0)[0] + "***@" + m.group(0).split("@")[1]),
        "API密钥": (r"sk-[a-zA-Z0-9]{20,}", lambda m: "sk-***REDACTED***"),
        "IP地址": (r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}", lambda m: "***.***.***.***"),
    }

    def __init__(self):
        self.mask_log = []

    def mask(self, text: str) -> tuple:
        """脱敏处理，返回(脱敏后文本, 脱敏记录)"""
        masked_text = text
        changes = []

        for data_type, (pattern, replacer) in self.SENSITIVE_PATTERNS.items():
            matches = list(re.finditer(pattern, masked_text))
            for match in matches:
                original = match.group(0)
                masked = replacer(match)
                masked_text = masked_text.replace(original, masked, 1)
                changes.append({
                    "type": data_type,
                    "original_length": len(original),
                    "masked": masked,
                })

        if changes:
            logger.warning(f"检测到{len(changes)}处敏感数据，已自动脱敏")

        return masked_text, changes


# 使用示例
masker = AIDataMasker()
user_input = "请帮我分析这份报告，联系邮箱zhangsan@company.com，手机138****8000，服务器IP 192.168.1.100"
masked, log = masker.mask(user_input)
print(f"脱敏后: {masked}")
# 输出: 请帮我分析这份报告，联系邮箱z***@company.com，手机138****8000，服务器IP ***.***.***.***

from enum import Enum
from dataclasses import dataclass


class DataSensitivity(Enum):
    PUBLIC = "公开"       # 可对外公开
    INTERNAL = "内部"     # 仅内部使用
    CONFIDENTIAL = "机密" # 需审批访问
    RESTRICTED = "绝密"   # 禁止AI处理


@dataclass
class DataAsset:
    name: str
    sensitivity: DataSensitivity
    ai_allowed: bool
    requires_masking: bool
    approval_required: bool
    owner: str


# 数据资产注册表
DATA_REGISTRY = {
    "客户信息表": DataAsset("客户信息表", DataSensitivity.RESTRICTED, False, True, True, "数据部门"),
    "产品文档": DataAsset("产品文档", DataSensitivity.INTERNAL, True, False, False, "产品部门"),
    "营销文案": DataAsset("营销文案", DataSensitivity.PUBLIC, True, False, False, "市场部门"),
    "财务报表": DataAsset("财务报表", DataSensitivity.CONFIDENTIAL, False, True, True, "财务部门"),
    "技术架构图": DataAsset("技术架构图", DataSensitivity.CONFIDENTIAL, True, True, True, "技术部门"),
    "API文档": DataAsset("API文档", DataSensitivity.INTERNAL, True, False, False, "开发部门"),
}


def check_ai_permission(data_name: str) -> dict:
    """检查数据是否允许输入AI"""
    asset = DATA_REGISTRY.get(data_name)
    if not asset:
        return {"status": "UNKNOWN", "message": f"未注册的数据资产: {data_name}"}

    return {
        "data_name": asset.name,
        "sensitivity": asset.sensitivity.value,
        "ai_allowed": asset.ai_allowed,
        "requires_masking": asset.requires_masking,
        "approval_required": asset.approval_required,
        "owner": asset.owner,
    }


# 使用示例
result = check_ai_permission("客户信息表")
print(result)
# 输出: ai_allowed=False, sensitivity=绝密, requires_masking=True

import hashlib
import time
from typing import Optional
from dataclasses import dataclass, field
from enum import Enum


class RiskLevel(Enum):
    LOW = "低风险"
    MEDIUM = "中风险"
    HIGH = "高风险"
    CRITICAL = "极高风险"


@dataclass
class DLPRule:
    """DLP检测规则"""
    name: str
    pattern: str
    risk_level: RiskLevel
    action: str  # allow, mask, block
    description: str


@dataclass
class DLPResult:
    """DLP检测结果"""
    allowed: bool
    risk_level: RiskLevel
    violations: list = field(default_factory=list)
    masked_content: Optional[str] = None
    timestamp: float = field(default_factory=time.time)


# 预定义规则库
DLP_RULES = [
    DLPRule("密钥检测", r"sk-[a-zA-Z0-9]{20,}", RiskLevel.CRITICAL, "block", "检测到API密钥"),
    DLPRule("身份证号", r"\d{17}[\dXx]", RiskLevel.HIGH, "mask", "检测到身份证号"),
    DLPRule("银行卡号", r"\d{16,19}", RiskLevel.HIGH, "mask", "检测到银行卡号"),
    DLPRule("手机号码", r"1[3-9]\d{9}", RiskLevel.MEDIUM, "mask", "检测到手机号"),
    DLPRule("邮箱地址", r"[\w.-]+@[\w.-]+\.[\w]+", RiskLevel.MEDIUM, "mask", "检测到邮箱"),
    DLPRule("内网IP", r"192\.168\.\d{1,3}\.\d{1,3}", RiskLevel.MEDIUM, "mask", "检测到内网IP"),
    DLPRule("密码字段", r"(password|passwd|pwd)\s*[:=]\s*\S+", RiskLevel.HIGH, "block", "检测到密码明文"),
]


class DLPGateway:
    """数据防泄露网关"""

    def __init__(self, rules: list = None):
        self.rules = rules or DLP_RULES
        self.audit_log = []

    def inspect(self, content: str, direction: str = "outbound") -> DLPResult:
        """
        检测内容是否包含敏感数据
        direction: outbound(发送给AI) / inbound(AI返回结果)
        """
        import re

        violations = []
        masked_content = content
        max_risk = RiskLevel.LOW
        blocked = False

        for rule in self.rules:
            matches = list(re.finditer(rule.pattern, masked_content, re.IGNORECASE))
            if matches:
                violations.append({
                    "rule": rule.name,
                    "risk": rule.risk_level.value,
                    "count": len(matches),
                    "action": rule.action,
                })

                if rule.risk_level.value > max_risk.value:
                    max_risk = rule.risk_level

                if rule.action == "block":
                    blocked = True
                elif rule.action == "mask":
                    for match in matches:
                        original = match.group(0)
                        masked = self._mask_value(original, rule.name)
                        masked_content = masked_content.replace(original, masked, 1)

        # 记录审计日志
        audit_entry = {
            "timestamp": time.time(),
            "direction": direction,
            "content_hash": hashlib.sha256(content.encode()).hexdigest()[:16],
            "violations": violations,
            "action": "BLOCKED" if blocked else "MASKED" if violations else "ALLOWED",
        }
        self.audit_log.append(audit_entry)

        return DLPResult(
            allowed=not blocked,
            risk_level=max_risk,
            violations=violations,
            masked_content=masked_content if not blocked else None,
        )

    @staticmethod
    def _mask_value(value: str, rule_name: str) -> str:
        """根据规则类型脱敏"""
        if "密钥" in rule_name:
            return "***API_KEY_REDACTED***"
        elif len(value) >= 8:
            return value[:3] + "****" + value[-3:]
        return "***"


# 使用示例
gateway = DLPGateway()

# 场景1：正常请求
result1 = gateway.inspect("请帮我写一段Python排序算法")
print(f"✅ 正常请求: allowed={result1.allowed}, risk={result1.risk_level.value}")

# 场景2：包含敏感数据
result2 = gateway.inspect("帮我分析这个用户的消费记录，手机号138****8000，银行卡622202123****890123")
print(f"⚠️ 敏感数据: allowed={result2.allowed}, masked={result2.masked_content}")

# 场景3：包含API密钥（直接阻断）
result3 = gateway.inspect("我的OpenAI密钥是sk-proj-abcdefghijklmnopqrstuvwxyz，帮我测试一下")
print(f"🚫 密钥泄露: allowed={result3.allowed}, violations={result3.violations}")

# Prometheus AI安全监控指标
# prometheus.yml
scrape_configs:
  - job_name: 'ai-security-metrics'
    static_configs:
      - targets: ['localhost:9090']
    metrics_path: '/metrics'
    scrape_interval: 15s

# 自定义指标
metric_relabel_configs:
  - source_labels: [__name__]
    regex: 'ai_(requests_total|blocked_total|masked_total|response_time_seconds)'
    action: keep

# Prometheus指标采集器
from prometheus_client import Counter, Histogram, Gauge, generate_latest

# 核心监控指标
AI_REQUESTS_TOTAL = Counter(
    'ai_requests_total',
    'AI请求总数',
    ['service', 'model', 'status']
)

AI_BLOCKED_TOTAL = Counter(
    'ai_blocked_total',
    '被DLP阻断的请求总数',
    ['rule', 'risk_level']
)

AI_MASKED_TOTAL = Counter(
    'ai_masked_total',
    '被脱敏处理的请求总数',
    ['data_type']
)

AI_RESPONSE_TIME = Histogram(
    'ai_response_time_seconds',
    'AI请求响应时间',
    ['service', 'model']
)

AI_ACTIVE_USERS = Gauge(
    'ai_active_users',
    '当前活跃AI用户数',
    ['department']
)


class SecurityMetricsCollector:
    """安全指标采集器"""

    @staticmethod
    def record_request(service: str, model: str, status: str):
        AI_REQUESTS_TOTAL.labels(
            service=service, model=model, status=status
        ).inc()

    @staticmethod
    def record_block(rule: str, risk_level: str):
        AI_BLOCKED_TOTAL.labels(
            rule=rule, risk_level=risk_level
        ).inc()

    @staticmethod
    def record_mask(data_type: str):
        AI_MASKED_TOTAL.labels(data_type=data_type).inc()

{
  "alerts": [
    {
      "name": "AI密钥泄露告警",
      "condition": "ai_blocked_total{rule=\"密钥检测\"} > 0",
      "severity": "critical",
      "message": "检测到API密钥尝试外泄，请立即排查！",
      "channels": ["钉钉群", "邮件", "短信"]
    },
    {
      "name": "异常调用频率告警",
      "condition": "rate(ai_requests_total[5m]) > 100",
      "severity": "warning",
      "message": "5分钟内AI调用超过100次，可能存在异常。",
      "channels": ["钉钉群"]
    },
    {
      "name": "脱敏量异常告警",
      "condition": "rate(ai_masked_total[1h]) > 50",
      "severity": "warning",
      "message": "过去1小时内大量数据被脱敏，员工AI使用习惯需关注。",
      "channels": ["邮件"]
    }
  ]
}

# 应急响应自动化脚本
class DataLeakageIncident:
    """数据泄露事件处理"""

    def __init__(self, incident_id, level, description):
        self.incident_id = incident_id
        self.level = level
        self.description = description
        self.timeline = []
        self.status = "OPEN"

    def auto_contain(self):
        """自动遏制措施"""
        actions = []

        # 1. 暂停涉事用户的AI访问权限
        actions.append(self._suspend_user_access())

        # 2. 联系AI服务商请求数据删除
        actions.append(self._request_data_deletion())

        # 3. 保存证据
        actions.append(self._preserve_evidence())

        # 4. 通知相关方
        actions.append(self._notify_stakeholders())

        return actions

    def _suspend_user_access(self):
        """暂停用户AI访问"""
        return {
            "action": "suspend_access",
            "status": "done",
            "detail": "已暂停涉事账号的AI服务访问权限"
        }

    def _request_data_deletion(self):
        """请求数据删除"""
        return {
            "action": "data_deletion_request",
            "status": "submitted",
            "detail": "已向AI服务商提交数据删除请求"
        }

    def _preserve_evidence(self):
        """保存证据"""
        return {
            "action": "evidence_preservation",
            "status": "done",
            "detail": "已保存对话日志、访问记录、网络流量"
        }

    def _notify_stakeholders(self):
        """通知相关方"""
        return {
            "action": "notification",
            "status": "done",
            "detail": "已通知安全负责人、业务负责人、数据所有者"
        }


# 触发应急响应
incident = DataLeakageIncident(
    incident_id="INC-2026-0509-001",
    level="P1",
    description="员工将客户名单输入ChatGPT进行分析"
)
actions = incident.auto_contain()
for action in actions:
    print(f"[{action['action']}] {action['detail']}")

# AI数据泄露事件报告模板

## 一、事件概述
- **事件编号**：INC-YYYY-MMDD-NNN
- **发生时间**：YYYY-MM-DD HH:MM
- **发现时间**：YYYY-MM-DD HH:MM
- **事件等级**：P0/P1/P2/P3
- **影响范围**：XX个用户/XX条数据/XX个系统

## 二、事件经过
1. 时间线描述（精确到分钟）
2. 攻击路径分析
3. 涉及的数据类型和数量

## 三、影响评估
- 数据敏感级别：绝密/机密/秘密/内部
- 受影响用户数：XX人
- 涉及个人信息：是/否
- 是否触发监管上报：是/否
- 预估损失：XX万元

## 四、处置措施
1. 遏制措施及时间
2. 根除措施及时间
3. 恢复措施及时间

## 五、根因分析
- 直接原因：
- 根本原因：
- 管理原因：

## 六、改进措施
| 序号 | 改进措施 | 责任人 | 完成时间 |
|:---:|---------|-------|---------|
| 1 | | | |
| 2 | | | |

## 七、附件
- 日志记录
- 证据截图
- 技术分析报告