数据科学在大数据领域的安全保障

    from Crypto.Cipher import AES
    from Crypto.Random import get_random_bytes
    import base64
    
    class AESCipher:
    def __init__(self, key=None):
        self.key = key if key else get_random_bytes(16)
        self.cipher = AES.new(self.key, AES.MODE_GCM)
        
    def encrypt(self, plaintext):
        ciphertext, tag = self.cipher.encrypt_and_digest(plaintext.encode())
        return base64.b64encode(self.cipher.nonce + tag + ciphertext).decode()
    
    def decrypt(self, encrypted):
        data = base64.b64decode(encrypted)
        nonce = data[:16]
        tag = data[16:32]
        ciphertext = data[32:]
        cipher = AES.new(self.key, AES.MODE_GCM, nonce=nonce)
        return cipher.decrypt_and_verify(ciphertext, tag).decode()
    
    # 使用示例
    cipher = AESCipher()
    encrypted = cipher.encrypt("敏感数据")
    print("加密结果:", encrypted)
    decrypted = cipher.decrypt(encrypted)
    print("解密结果:", decrypted)
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/EYF1mShcXyuI6lRHKogk4aedrDOG.png)

    class RBAC:
    def __init__(self):
        self.roles = {}
        self.users = {}
        
    def add_role(self, role_name, permissions):
        self.roles[role_name] = permissions
        
    def assign_role(self, user_id, role_name):
        if user_id not in self.users:
            self.users[user_id] = []
        self.users[user_id].append(role_name)
        
    def check_permission(self, user_id, permission):
        if user_id not in self.users:
            return False
        for role in self.users[user_id]:
            if permission in self.roles.get(role, []):
                return True
        return False
    
    # 使用示例
    rbac = RBAC()
    rbac.add_role('admin', ['read', 'write', 'delete'])
    rbac.add_role('user', ['read'])
    rbac.assign_role('user1', 'admin')
    print("user1有写权限:", rbac.check_permission('user1', 'write'))
    print("user1有执行权限:", rbac.check_permission('user1', 'execute'))
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/IsiUGOADeq4JpkrjB9NoxVlv5Yhg.png)

    import re
    
    def anonymize_name(name):
    if len(name) <= 1:
        return "*"
    return name[0] + "*"*(len(name)-1)
    
    def anonymize_email(email):
    parts = email.split('@')
    if len(parts) != 2:
        return email
    username = parts[0]
    domain = parts[1]
    if len(username) <= 2:
        masked_username = "*"*len(username)
    else:
        masked_username = username[0] + "*"*(len(username)-2) + username[-1]
    return f"{masked_username}@{domain}"
    
    def anonymize_phone(phone):
    return re.sub(r'(\d{3})\d{4}(\d{3})', r'\1****\2', phone)
    
    # 使用示例
    print("姓名脱敏:", anonymize_name("张三"))
    print("邮箱脱敏:", anonymize_email("zhangsan@example.com"))
    print("电话脱敏:", anonymize_phone("13812345678"))
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/n5JBu8R3arCcoMXv6QzxfUbS4py9.png)

    pip install pycryptodome pandas numpy scikit-learn
    
    
    bash

    import pandas as pd
    from Crypto.Cipher import AES
    from Crypto.Random import get_random_bytes
    import base64
    import numpy as np
    
    class SecureDataPipeline:
    def __init__(self, encryption_key=None):
        self.encryption_key = encryption_key or get_random_bytes(16)
        self.sensitive_columns = []
        
    def identify_sensitive_columns(self, df, regex_patterns):
        """识别包含敏感信息的列"""
        self.sensitive_columns = []
        for col in df.columns:
            if df[col].apply(lambda x: any(re.search(p, str(x)) for p in regex_patterns)).any():
                self.sensitive_columns.append(col)
        return self.sensitive_columns
    
    def encrypt_column(self, df, column_name):
        """加密指定列"""
        cipher = AES.new(self.encryption_key, AES.MODE_GCM)
        def encrypt_field(x):
            if pd.isna(x): return x
            nonce = get_random_bytes(16)
            cipher = AES.new(self.encryption_key, AES.MODE_GCM, nonce=nonce)
            ciphertext, tag = cipher.encrypt_and_digest(str(x).encode())
            return base64.b64encode(nonce + tag + ciphertext).decode()
        
        df[column_name] = df[column_name].apply(encrypt_field)
        return df
    
    def decrypt_column(self, df, column_name):
        """解密指定列"""
        def decrypt_field(x):
            if pd.isna(x): return x
            data = base64.b64decode(x)
            nonce = data[:16]
            tag = data[16:32]
            ciphertext = data[32:]
            cipher = AES.new(self.encryption_key, AES.MODE_GCM, nonce=nonce)
            return cipher.decrypt_and_verify(ciphertext, tag).decode()
        
        df[column_name] = df[column_name].apply(decrypt_field)
        return df
    
    def apply_differential_privacy(self, df, column_name, epsilon=1.0):
        """应用差分隐私"""
        if df[column_name].dtype.kind in 'biufc':  # 数值类型
            sensitivity = df[column_name].max() - df[column_name].min()
            scale = sensitivity / epsilon
            noise = np.random.laplace(0, scale, len(df))
            df[column_name] = df[column_name] + noise
        return df
    
    # 使用示例
    data = {
    'name': ['张三', '李四', '王五'],
    'email': ['zhangsan@test.com', 'lisi@test.com', 'wangwu@test.com'],
    'salary': [10000, 15000, 20000]
    }
    df = pd.DataFrame(data)
    
    pipeline = SecureDataPipeline()
    print("识别敏感列:", pipeline.identify_sensitive_columns(df, [r'@.*\.com$', r'^\D{2,4}$']))
    
    df_encrypted = pipeline.encrypt_column(df.copy(), 'email')
    print("\n加密后的数据:")
    print(df_encrypted)
    
    df_decrypted = pipeline.decrypt_column(df_encrypted.copy(), 'email')
    print("\n解密后的数据:")
    print(df_decrypted)
    
    df_dp = pipeline.apply_differential_privacy(df.copy(), 'salary', epsilon=0.5)
    print("\n应用差分隐私后的薪资:")
    print(df_dp)
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/JO2tLovEqnpiylKCU7G4fQk1Fg9m.png)

    from sklearn.ensemble import IsolationForest
    from sklearn.preprocessing import StandardScaler
    import numpy as np
    import pandas as pd
    
    class AccessAnomalyDetector:
    def __init__(self, contamination=0.01):
        self.model = IsolationForest(contamination=contamination)
        self.scaler = StandardScaler()
        self.features = ['access_count', 'data_volume', 'time_spread', 'resource_variety']
        
    def train(self, access_logs):
        """训练异常检测模型"""
        # 从日志中提取特征
        features = self._extract_features(access_logs)
        # 标准化特征
        features_scaled = self.scaler.fit_transform(features)
        # 训练模型
        self.model.fit(features_scaled)
        
    def predict(self, access_logs):
        """预测异常访问"""
        features = self._extract_features(access_logs)
        if len(features) == 0:
            return np.array([])
        features_scaled = self.scaler.transform(features)
        return self.model.predict(features_scaled)
    
    def _extract_features(self, logs):
        """从日志中提取特征"""
        if len(logs) == 0:
            return np.zeros((0, len(self.features)))
        
        # 计算每个用户/IP的特征
        logs['timestamp'] = pd.to_datetime(logs['timestamp'])
        features = []
        
        for entity, group in logs.groupby('user_id'):
            access_count = len(group)
            data_volume = group['data_size'].sum()
            
            time_diff = group['timestamp'].max() - group['timestamp'].min()
            time_spread = time_diff.total_seconds() / 3600 if access_count > 1 else 0
            
            resource_variety = len(group['resource_id'].unique())
            
            features.append([access_count, data_volume, time_spread, resource_variety])
        
        return np.array(features)
    
    # 使用示例
    # 模拟访问日志
    np.random.seed(42)
    normal_logs = pd.DataFrame({
    'user_id': np.random.choice(['user1', 'user2', 'user3'], 1000),
    'timestamp': pd.date_range('2023-01-01', periods=1000, freq='T'),
    'resource_id': np.random.choice(['res1', 'res2', 'res3', 'res4'], 1000),
    'data_size': np.random.randint(1, 100, 1000)
    })
    
    # 添加一些异常访问
    anomaly_logs = pd.DataFrame({
    'user_id': ['attacker1', 'attacker2'],
    'timestamp': ['2023-01-01 12:00', '2023-01-01 12:05'],
    'resource_id': ['res1', 'res2'],
    'data_size': [10000, 5000]
    })
    
    all_logs = pd.concat([normal_logs, anomaly_logs])
    
    # 训练和预测
    detector = AccessAnomalyDetector(contamination=0.01)
    detector.train(normal_logs)
    predictions = detector.predict(all_logs)
    
    # 标记异常
    all_logs['is_anomaly'] = np.where(predictions == -1, 1, 0)
    print("检测到的异常访问:")
    print(all_logs[all_logs['is_anomaly'] == 1])
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/5M2zs6gTbofuYyHQpkWGh8EXZcF4.png)