【PythonAI】6.1.2 数据安全:工作中使用AI工具的数据隐私保护策略(2. 数据脱敏示例)
·
# mask_sensitive_data.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据脱敏工具类
支持手机号、身份证号、姓名、邮箱、银行卡号、地址等敏感信息脱敏
"""
import re
import hashlib
import random
import string
from typing import Any, Dict, List, Optional, Union
from datetime import datetime
import json
# 尝试导入numpy,如果没有则使用替代方案
try:
import numpy as np
HAS_NUMPY = True
except ImportError:
HAS_NUMPY = False
# 如果没有numpy,提供一个简单的替代实现
class SimpleRandom:
@staticmethod
def laplace(loc, scale, size):
"""简单的拉普拉斯分布随机数生成"""
u = random.random()
return loc - scale * (1 if u > 0.5 else -1) * (1 - 2 * abs(u - 0.5))
np = SimpleRandom()
class DataMasking:
"""数据脱敏工具类"""
# 常见敏感字段关键词
SENSITIVE_KEYWORDS = {
'phone': ['phone', 'mobile', 'tel', '电话', '手机', '联系电话'],
'id_card': ['id_card', 'idnumber', 'identity', '身份证', '证件号'],
'name': ['name', 'user_name', 'full_name', '姓名', '名称', '用户名'],
'email': ['email', 'mail', '邮箱', '电子邮件'],
'bank_card': ['bank_card', 'bankcard', 'credit_card', '银行卡', '信用卡', '卡号'],
'address': ['address', 'addr', '住址', '地址', '家庭住址'],
'password': ['password', 'pwd', 'passwd', '密码', '口令'],
'ip': ['ip', 'ip_address', 'ipv4', 'ipv6', 'IP地址'],
'license': ['license', 'plate', '车牌', '驾驶证'],
'social': ['social', 'wechat', 'qq', '微信', 'QQ', '社交账号']
}
@staticmethod
def mask_phone(phone: str, keep_prefix: int = 3, keep_suffix: int = 4) -> str:
"""
手机号脱敏
Args:
phone: 手机号
keep_prefix: 保留前几位
keep_suffix: 保留后几位
Returns:
脱敏后的手机号
"""
if not phone:
return phone
phone_str = str(phone)
# 检查是否包含国际区号
has_country_code = False
country_code = ""
# 处理86开头的手机号
if phone_str.startswith('86') and len(phone_str) > 11:
has_country_code = True
country_code = "86"
phone_str = phone_str[2:] # 去掉区号
elif phone_str.startswith('+86') and len(phone_str) > 12:
has_country_code = True
country_code = "+86"
phone_str = phone_str[3:] # 去掉区号
# 确保手机号长度足够
if len(phone_str) < keep_prefix + keep_suffix:
# 如果长度不够,返回全部隐藏
return '*' * len(phone_str)
# 脱敏处理
mask_length = len(phone_str) - keep_prefix - keep_suffix
masked = phone_str[:keep_prefix] + '*' * mask_length + phone_str[-keep_suffix:]
# 如果有国际区号,重新加上
if has_country_code:
masked = country_code + masked
return masked
@staticmethod
def mask_id_card(id_card: str, keep_prefix: int = 6, keep_suffix: int = 4) -> str:
"""
身份证号脱敏
Args:
id_card: 身份证号(15位或18位)
keep_prefix: 保留前几位
keep_suffix: 保留后几位
Returns:
脱敏后的身份证号
"""
if not id_card:
return id_card
id_card_str = str(id_card)
# 检查长度是否足够
if len(id_card_str) < keep_prefix + keep_suffix:
return '*' * len(id_card_str)
mask_length = len(id_card_str) - keep_prefix - keep_suffix
return id_card_str[:keep_prefix] + '*' * mask_length + id_card_str[-keep_suffix:]
@staticmethod
def mask_name(name: str, keep_prefix: int = 1, keep_suffix: int = 0) -> str:
"""
姓名脱敏
Args:
name: 姓名
keep_prefix: 保留前几位(默认保留姓)
keep_suffix: 保留后几位
Returns:
脱敏后的姓名
"""
if not name:
return "*"
name_str = str(name)
# 单字符姓名
if len(name_str) <= 1:
return "*"
# 如果保留位数超过姓名长度
if keep_prefix + keep_suffix >= len(name_str):
return name_str[0] + '*' * (len(name_str) - 1)
if keep_suffix > 0:
return name_str[:keep_prefix] + '*' * (len(name_str) - keep_prefix - keep_suffix) + name_str[-keep_suffix:]
else:
return name_str[:keep_prefix] + '*' * (len(name_str) - keep_prefix)
@staticmethod
def mask_email(email: str, keep_prefix: int = 3, keep_suffix: int = 3) -> str:
"""
邮箱脱敏
Args:
email: 邮箱地址
keep_prefix: 保留前缀的前几位
keep_suffix: 保留后缀的前几位
Returns:
脱敏后的邮箱
"""
if not email or '@' not in email:
return email
email_str = str(email)
parts = email_str.split('@', 1)
if len(parts) != 2:
return email_str
local_part, domain = parts
# 处理本地部分
if len(local_part) <= keep_prefix:
masked_local = '*' * len(local_part)
else:
mask_length = len(local_part) - keep_prefix - keep_suffix
if mask_length <= 0:
# 如果长度不够,只保留前缀
masked_local = local_part[:keep_prefix] + '*' * (len(local_part) - keep_prefix)
else:
masked_local = local_part[:keep_prefix] + '*' * mask_length + local_part[-keep_suffix:]
return f"{masked_local}@{domain}"
@staticmethod
def mask_bank_card(card_number: str, keep_prefix: int = 4, keep_suffix: int = 4) -> str:
"""
银行卡号脱敏
Args:
card_number: 银行卡号
keep_prefix: 保留前几位
keep_suffix: 保留后几位
Returns:
脱敏后的银行卡号
"""
if not card_number:
return card_number
# 移除空格
card_str = str(card_number).replace(' ', '')
# 检查长度
if len(card_str) < keep_prefix + keep_suffix:
return '*' * len(card_str)
mask_length = len(card_str) - keep_prefix - keep_suffix
masked = card_str[:keep_prefix] + '*' * mask_length + card_str[-keep_suffix:]
# 格式化输出(每4位一组)
return ' '.join([masked[i:i+4] for i in range(0, len(masked), 4)])
@staticmethod
def mask_address(address: str, keep_prefix: int = 6, keep_suffix: int = 4) -> str:
"""
地址脱敏
Args:
address: 详细地址
keep_prefix: 保留前几位字符
keep_suffix: 保留后几位字符
Returns:
脱敏后的地址
"""
if not address:
return address
address_str = str(address)
# 检查长度
if len(address_str) <= keep_prefix + keep_suffix:
return address_str[:keep_prefix] + '***'
mask_length = len(address_str) - keep_prefix - keep_suffix
return address_str[:keep_prefix] + '*' * mask_length + address_str[-keep_suffix:]
@staticmethod
def mask_ip(ip: str) -> str:
"""
IP地址脱敏
Args:
ip: IP地址
Returns:
脱敏后的IP地址
"""
if not ip:
return ip
ip_str = str(ip)
parts = ip_str.split('.')
if len(parts) == 4:
# IPv4: 保留前两段
return f"{parts[0]}.{parts[1]}.*.*"
elif ':' in ip_str:
# IPv6: 保留前4组
parts = ip_str.split(':')
if len(parts) >= 4:
return ':'.join(parts[:2]) + ':****:****'
else:
return ip_str
else:
return ip_str
@staticmethod
def mask_password(password: str) -> str:
"""
密码脱敏(全部替换为*)
Args:
password: 密码
Returns:
脱敏后的密码
"""
if not password:
return ""
return '*' * len(str(password))
@staticmethod
def mask_license(license: str, keep_prefix: int = 2, keep_suffix: int = 2) -> str:
"""
车牌号脱敏
Args:
license: 车牌号
keep_prefix: 保留前几位
keep_suffix: 保留后几位
Returns:
脱敏后的车牌号
"""
if not license:
return license
license_str = str(license)
# 检查长度
if len(license_str) <= keep_prefix + keep_suffix:
return license_str
mask_length = len(license_str) - keep_prefix - keep_suffix
return license_str[:keep_prefix] + '*' * mask_length + license_str[-keep_suffix:]
@staticmethod
def mask_social_account(account: str, keep_prefix: int = 2, keep_suffix: int = 2) -> str:
"""
社交账号脱敏(微信、QQ等)
Args:
account: 社交账号
keep_prefix: 保留前几位
keep_suffix: 保留后几位
Returns:
脱敏后的账号
"""
if not account:
return account
account_str = str(account)
# 检查长度
if len(account_str) <= keep_prefix + keep_suffix:
return account_str[0] + '*' * (len(account_str) - 1) if len(account_str) > 1 else '*'
mask_length = len(account_str) - keep_prefix - keep_suffix
return account_str[:keep_prefix] + '*' * mask_length + account_str[-keep_suffix:]
@staticmethod
def mask_date(date: str, keep_year: bool = True) -> str:
"""
日期脱敏(保留年份,隐藏月日)
Args:
date: 日期字符串(格式:YYYY-MM-DD)
keep_year: 是否保留年份
Returns:
脱敏后的日期
"""
if not date:
return date
date_str = str(date)
# 匹配常见日期格式
patterns = [
(r'(\d{4})-(\d{2})-(\d{2})', r'\1-**-**'), # YYYY-MM-DD
(r'(\d{4})/(\d{2})/(\d{2})', r'\1/**/**'), # YYYY/MM/DD
(r'(\d{4})年(\d{2})月(\d{2})日', r'\1年**月**日') # YYYY年MM月DD日
]
for pattern, replacement in patterns:
match = re.search(pattern, date_str)
if match:
if keep_year:
return re.sub(pattern, replacement, date_str)
else:
return "****-**-**"
return date_str
@staticmethod
def pseudonymize(data: str, salt: str = "secret_salt", length: int = 16) -> str:
"""
假名化(不可逆哈希)
Args:
data: 原始数据
salt: 盐值
length: 输出长度
Returns:
哈希后的假名
"""
if not data:
return ""
# 使用SHA256进行哈希
hash_obj = hashlib.sha256(f"{data}{salt}".encode())
return hash_obj.hexdigest()[:length]
@staticmethod
def tokenize(data: str, token_length: int = 32) -> str:
"""
令牌化(生成随机令牌替换原值)
Args:
data: 原始数据(用于生成确定性的令牌)
token_length: 令牌长度
Returns:
随机令牌
"""
if not data:
return ""
# 使用数据的哈希作为种子,生成确定性令牌
hash_obj = hashlib.md5(data.encode())
seed = int(hash_obj.hexdigest()[:8], 16)
random.seed(seed)
# 生成随机字符串
chars = string.ascii_letters + string.digits
token = ''.join(random.choice(chars) for _ in range(token_length))
# 重置随机种子
random.seed()
return token
@staticmethod
def mask_by_type(data: Any, data_type: str, **kwargs) -> Any:
"""
根据数据类型自动选择脱敏方法
Args:
data: 待脱敏数据
data_type: 数据类型(phone, id_card, name, email, bank_card, address, ip, password, license, social, date)
**kwargs: 传递给具体脱敏方法的参数
Returns:
脱敏后的数据
"""
mask_methods = {
'phone': DataMasking.mask_phone,
'id_card': DataMasking.mask_id_card,
'name': DataMasking.mask_name,
'email': DataMasking.mask_email,
'bank_card': DataMasking.mask_bank_card,
'address': DataMasking.mask_address,
'ip': DataMasking.mask_ip,
'password': DataMasking.mask_password,
'license': DataMasking.mask_license,
'social': DataMasking.mask_social_account,
'date': DataMasking.mask_date
}
if data_type in mask_methods:
return mask_methods[data_type](data, **kwargs)
else:
return data
@classmethod
def mask_sensitive_dict(cls, data: Dict[str, Any], custom_rules: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
"""
对字典中的敏感字段自动脱敏
Args:
data: 待脱敏的字典
custom_rules: 自定义脱敏规则,格式:{'字段名': '数据类型'}
Returns:
脱敏后的字典
"""
if not data:
return data
masked_data = {}
# 合并默认规则和自定义规则
rules = custom_rules or {}
for key, value in data.items():
key_lower = key.lower()
masked_value = value
# 优先使用自定义规则
if key in rules:
masked_value = cls.mask_by_type(value, rules[key])
else:
# 根据关键词自动识别
for data_type, keywords in cls.SENSITIVE_KEYWORDS.items():
if any(keyword in key_lower for keyword in keywords):
masked_value = cls.mask_by_type(value, data_type)
break
masked_data[key] = masked_value
return masked_data
@classmethod
def mask_sensitive_json(cls, json_str: str, custom_rules: Optional[Dict[str, str]] = None) -> str:
"""
对JSON字符串中的敏感字段自动脱敏
Args:
json_str: JSON字符串
custom_rules: 自定义脱敏规则
Returns:
脱敏后的JSON字符串
"""
try:
data = json.loads(json_str)
masked_data = cls.mask_sensitive_dict(data, custom_rules)
return json.dumps(masked_data, ensure_ascii=False, indent=2)
except json.JSONDecodeError:
return json_str
@classmethod
def mask_sensitive_list(cls, data_list: List[Dict[str, Any]], custom_rules: Optional[Dict[str, str]] = None) -> List[Dict[str, Any]]:
"""
对列表中的字典进行批量脱敏
Args:
data_list: 待脱敏的字典列表
custom_rules: 自定义脱敏规则
Returns:
脱敏后的字典列表
"""
return [cls.mask_sensitive_dict(item, custom_rules) for item in data_list]
class DataDesensitization:
"""数据去标识化(高级功能)"""
@staticmethod
def k_anonymity(data: List[Dict], quasi_identifiers: List[str], k: int = 3) -> List[Dict]:
"""
K-匿名化处理
Args:
data: 数据集
quasi_identifiers: 准标识符列表
k: 匿名化参数,每组至少k条记录
Returns:
匿名化后的数据集
"""
if not data:
return data
# 简化实现:对准标识符进行泛化处理
anonymized_data = []
for record in data:
anonymized_record = record.copy()
for qi in quasi_identifiers:
if qi in anonymized_record:
# 泛化处理:年龄分段、地区模糊等
value = anonymized_record[qi]
if isinstance(value, (int, float)) and qi == 'age':
# 年龄分段
age_range = (int(value) // 10) * 10
anonymized_record[qi] = f"{age_range}-{age_range + 9}"
elif isinstance(value, str) and ('address' in qi.lower() or 'city' in qi.lower()):
# 地址泛化(保留前3个字符)
anonymized_record[qi] = value[:3] + '***' if len(value) > 3 else '***'
anonymized_data.append(anonymized_record)
return anonymized_data
@staticmethod
def l_diversity(data: List[Dict], sensitive_attribute: str, l: int = 2) -> List[Dict]:
"""
L-多样性处理
Args:
data: 数据集
sensitive_attribute: 敏感属性
l: 多样性参数
Returns:
多样性处理后的数据集
"""
# 简化实现:确保每个等价类中敏感属性至少有l个不同值
return data
@staticmethod
def differential_privacy(data: List[float], epsilon: float = 0.1) -> List[float]:
"""
差分隐私处理(添加拉普拉斯噪声)
Args:
data: 数值数据列表
epsilon: 隐私预算
Returns:
添加噪声后的数据
"""
if not data:
return data
# 添加拉普拉斯噪声
scale = 1.0 / epsilon
if HAS_NUMPY:
noise = np.random.laplace(0, scale, len(data))
return [d + n for d, n in zip(data, noise)]
else:
# 如果没有numpy,使用简单实现
result = []
for d in data:
noise = np.laplace(0, scale, 1) if hasattr(np, 'laplace') else random.uniform(-scale, scale)
result.append(d + noise)
return result
# 单元测试
def run_tests():
"""运行单元测试"""
print("运行数据脱敏单元测试...")
print("=" * 50)
test_passed = True
# 测试手机号脱敏
try:
result1 = DataMasking.mask_phone("13800138000")
assert result1 == "138****8000", f"期望: 138****8000, 实际: {result1}"
result2 = DataMasking.mask_phone("8613800138000", keep_prefix=2, keep_suffix=4)
assert result2 == "86****8000", f"期望: 86****8000, 实际: {result2}"
result3 = DataMasking.mask_phone("+8613800138000", keep_prefix=3, keep_suffix=4)
assert result3 == "+86****8000", f"期望: +86****8000, 实际: {result3}"
result4 = DataMasking.mask_phone("12345")
assert result4 == "*****", f"期望: *****, 实际: {result4}"
print("✓ 手机号脱敏测试通过")
except AssertionError as e:
print(f"✗ 手机号脱敏测试失败: {e}")
test_passed = False
# 测试身份证脱敏
try:
result = DataMasking.mask_id_card("650102199001011234")
assert result == "650102********1234", f"期望: 650102********1234, 实际: {result}"
result2 = DataMasking.mask_id_card("12345")
assert result2 == "*****", f"期望: *****, 实际: {result2}"
print("✓ 身份证脱敏测试通过")
except AssertionError as e:
print(f"✗ 身份证脱敏测试失败: {e}")
test_passed = False
# 测试姓名脱敏
try:
assert DataMasking.mask_name("张三") == "张*"
assert DataMasking.mask_name("欧阳菲菲") == "欧***"
assert DataMasking.mask_name("李") == "*"
assert DataMasking.mask_name("") == "*"
print("✓ 姓名脱敏测试通过")
except AssertionError as e:
print(f"✗ 姓名脱敏测试失败: {e}")
test_passed = False
# 测试邮箱脱敏
try:
assert DataMasking.mask_email("zhangsan@example.com") == "zha***@example.com"
assert DataMasking.mask_email("a@b.com", keep_prefix=1) == "*@b.com"
assert DataMasking.mask_email("ab@c.com", keep_prefix=1, keep_suffix=1) == "a*@c.com"
print("✓ 邮箱脱敏测试通过")
except AssertionError as e:
print(f"✗ 邮箱脱敏测试失败: {e}")
test_passed = False
# 测试银行卡脱敏
try:
result = DataMasking.mask_bank_card("6228480012345678912")
assert "****" in result
print("✓ 银行卡脱敏测试通过")
except AssertionError as e:
print(f"✗ 银行卡脱敏测试失败: {e}")
test_passed = False
# 测试IP脱敏
try:
assert DataMasking.mask_ip("192.168.1.100") == "192.168.*.*"
assert DataMasking.mask_ip("2001:0db8:85a3:0000:0000:8a2e:0370:7334") == "2001:0db8:****:****"
print("✓ IP脱敏测试通过")
except AssertionError as e:
print(f"✗ IP脱敏测试失败: {e}")
test_passed = False
# 测试假名化
try:
pseudonym1 = DataMasking.pseudonymize("test", salt="salt1")
pseudonym2 = DataMasking.pseudonymize("test", salt="salt1")
assert pseudonym1 == pseudonym2
print("✓ 假名化测试通过(确定性)")
except AssertionError as e:
print(f"✗ 假名化测试失败: {e}")
test_passed = False
# 测试地址脱敏
try:
result = DataMasking.mask_address("北京市朝阳区建国路88号", keep_prefix=3, keep_suffix=2)
assert len(result) == len("北京市朝阳区建国路88号")
print("✓ 地址脱敏测试通过")
except Exception as e:
print(f"✗ 地址脱敏测试失败: {e}")
test_passed = False
# 测试日期脱敏
try:
assert DataMasking.mask_date("1990-01-01") == "1990-**-**"
assert DataMasking.mask_date("1990/01/01") == "1990/**/**"
assert DataMasking.mask_date("1990年01月01日") == "1990年**月**日"
print("✓ 日期脱敏测试通过")
except AssertionError as e:
print(f"✗ 日期脱敏测试失败: {e}")
test_passed = False
print("\n" + "=" * 50)
if test_passed:
print("所有测试通过!✓")
else:
print("部分测试失败!✗")
return test_passed
# 使用示例
def demo():
"""演示数据脱敏功能"""
print("\n" + "=" * 70)
print("数据脱敏工具演示")
print("=" * 70)
# 1. 基本脱敏示例
print("\n1. 基本脱敏示例:")
print("-" * 50)
sensitive_data = {
"姓名": "张三丰",
"手机号": "13800138000",
"身份证号": "650102199001011234",
"邮箱": "zhangsan@example.com",
"银行卡号": "6228480012345678912",
"地址": "北京市朝阳区建国路88号SOHO现代城A座1234室",
"IP地址": "192.168.1.100",
"密码": "MySecret123",
"车牌号": "京A12345",
"微信号": "zhangsan_2024",
"出生日期": "1990-01-01"
}
# 手动脱敏
masked_manual = {
"姓名": DataMasking.mask_name(sensitive_data["姓名"]),
"手机号": DataMasking.mask_phone(sensitive_data["手机号"]),
"身份证号": DataMasking.mask_id_card(sensitive_data["身份证号"]),
"邮箱": DataMasking.mask_email(sensitive_data["邮箱"]),
"银行卡号": DataMasking.mask_bank_card(sensitive_data["银行卡号"]),
"地址": DataMasking.mask_address(sensitive_data["地址"], keep_prefix=6, keep_suffix=4),
"IP地址": DataMasking.mask_ip(sensitive_data["IP地址"]),
"密码": DataMasking.mask_password(sensitive_data["密码"]),
"车牌号": DataMasking.mask_license(sensitive_data["车牌号"]),
"微信号": DataMasking.mask_social_account(sensitive_data["微信号"]),
"出生日期": DataMasking.mask_date(sensitive_data["出生日期"])
}
print("原始数据:")
for key, value in sensitive_data.items():
print(f" {key}: {value}")
print("\n脱敏后数据:")
for key, value in masked_manual.items():
print(f" {key}: {value}")
# 2. 自动识别脱敏
print("\n2. 自动识别脱敏:")
print("-" * 50)
mixed_data = {
"user_name": "李四",
"mobile_phone": "13912345678",
"email_address": "lisi@company.com",
"id_number": "11010119900307663X",
"home_address": "上海市浦东新区世纪大道100号",
"bank_card_no": "6212260200001234567"
}
masked_auto = DataMasking.mask_sensitive_dict(mixed_data)
print("原始数据:")
for key, value in mixed_data.items():
print(f" {key}: {value}")
print("\n自动脱敏后:")
for key, value in masked_auto.items():
print(f" {key}: {value}")
# 3. 国际手机号测试
print("\n3. 国际手机号脱敏:")
print("-" * 50)
international_phones = [
"8613800138000",
"+8613800138000",
"13800138000"
]
for phone in international_phones:
masked = DataMasking.mask_phone(phone)
print(f" {phone} -> {masked}")
if __name__ == "__main__":
# 运行单元测试
test_result = run_tests()
if test_result:
# 运行演示示例
demo()
# 实际使用示例
print("\n" + "=" * 70)
print("实际使用示例")
print("=" * 70)
# 创建脱敏工具实例
masker = DataMasking()
# 处理单个敏感信息
phone = "13800138000"
masked_phone = masker.mask_phone(phone)
print(f"原始手机号: {phone}")
print(f"脱敏手机号: {masked_phone}")
# 处理国际手机号
phone_international = "8613800138000"
masked_international = masker.mask_phone(phone_international, keep_prefix=2, keep_suffix=4)
print(f"国际手机号: {phone_international} -> {masked_international}")
# 处理JSON数据
json_str = '{"name":"张三","phone":"13800138000","email":"zhang@example.com"}'
masked_json = masker.mask_sensitive_json(json_str)
print(f"\n原始JSON: {json_str}")
print(f"脱敏JSON: {masked_json}")
# 处理字典数据
user_info = {
"user_id": 1001,
"name": "李四",
"phone": "13912345678",
"id_card": "11010119900307663X"
}
masked_user = masker.mask_sensitive_dict(user_info)
print(f"\n原始用户信息: {user_info}")
print(f"脱敏用户信息: {masked_user}")
else:
print("\n请修复测试失败的问题后重新运行。")
运行结果:
(ai_env) $ python3 mask_sensitive_data.py
运行数据脱敏单元测试...
==================================================
✗ 手机号脱敏测试失败: 期望: 86****8000, 实际: 8613*****8000
✓ 身份证脱敏测试通过
✓ 姓名脱敏测试通过
✗ 邮箱脱敏测试失败:
✓ 银行卡脱敏测试通过
✓ IP脱敏测试通过
✓ 假名化测试通过(确定性)
✓ 地址脱敏测试通过
✓ 日期脱敏测试通过
==================================================
部分测试失败!✗
请修复测试失败的问题后重新运行。
(ai_env) $
更多推荐


所有评论(0)