49：L构建IoT安全：蓝队的物联网防御

import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import LabelEncoder

class DeviceAuthenticator:
    def __init__(self):
        self.model = RandomForestClassifier(n_estimators=100, random_state=42)
        self.label_encoder = LabelEncoder()
    
    def train_model(self, training_data):
        """训练设备认证模型"""
        # 特征提取
        X = training_data[['device_type', 'network_pattern', 'behavior_pattern', 'physical_features']]
        y = training_data['authentic']
        
        # 编码分类特征
        for col in ['device_type', 'network_pattern', 'behavior_pattern']:
            X[col] = self.label_encoder.fit_transform(X[col])
        
        # 训练模型
        self.model.fit(X, y)
    
    def authenticate(self, device_data):
        """认证设备"""
        # 特征提取
        features = {
            'device_type': device_data['type'],
            'network_pattern': device_data['network_pattern'],
            'behavior_pattern': device_data['behavior_pattern'],
            'physical_features': device_data['physical_features']
        }
        
        # 编码特征
        for key in ['device_type', 'network_pattern', 'behavior_pattern']:
            if features[key] in self.label_encoder.classes_:
                features[key] = self.label_encoder.transform([features[key]])[0]
            else:
                features[key] = -1
        
        # 预测
        X = np.array(list(features.values())).reshape(1, -1)
        prediction = self.model.predict(X)[0]
        confidence = self.model.predict_proba(X)[0][1]
        
        return {
            'authentic': bool(prediction),
            'confidence': confidence
        }

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest

class AnomalyDetector:
    def __init__(self):
        self.models = {}
    
    def train_model(self, device_type, training_data):
        """训练异常检测模型"""
        # 特征提取
        X = training_data[['network_traffic', 'device_status', 'battery_level', 'temperature']]
        
        # 训练模型
        model = IsolationForest(contamination=0.1, random_state=42)
        model.fit(X)
        
        # 保存模型
        self.models[device_type] = model
    
    def detect_anomaly(self, device_type, device_data):
        """检测设备异常"""
        if device_type not in self.models:
            return {"error": "Model not found for device type"}
        
        # 特征提取
        X = np.array([[device_data['network_traffic'],
                       device_data['device_status'],
                       device_data['battery_level'],
                       device_data['temperature']]])
        
        # 预测异常
        model = self.models[device_type]
        anomaly = model.predict(X)[0]
        score = model.score_samples(X)[0]
        
        return {
            'anomaly': bool(anomaly == -1),
            'anomaly_score': score
        }

import requests
import json

class VulnerabilityManager:
    def __init__(self):
        self.cve_api_url = 'https://services.nvd.nist.gov/rest/json/cves/1.0'
        self.firmware_repo = {}
    
    def scan_vulnerabilities(self, device_info):
        """扫描设备漏洞"""
        vulnerabilities = []
        
        device_type = device_info['type']
        firmware_version = device_info['firmware_version']
        
        # 查询CVE数据库
        cves = self._query_cve(device_type, firmware_version)
        
        if cves:
            for cve in cves:
                vulnerabilities.append({
                    'cve_id': cve['id'],
                    'severity': cve.get('severity', 'Unknown'),
                    'description': cve.get('description', ''),
                    'cvss_score': cve.get('cvss_score', 0)
                })
        
        return vulnerabilities
    
    def check_firmware_update(self, device_info):
        """检查固件更新"""
        device_type = device_info['type']
        current_version = device_info['firmware_version']
        
        # 检查是否有可用的固件更新
        if device_type in self.firmware_repo:
            latest_version = self.firmware_repo[device_type]['latest_version']
            if latest_version > current_version:
                return {
                    'update_available': True,
                    'latest_version': latest_version,
                    'release_notes': self.firmware_repo[device_type]['release_notes']
                }
        
        return {'update_available': False}
    
    def _query_cve(self, device_type, firmware_version):
        """查询CVE数据库"""
        # 这里是查询CVE数据库的逻辑
        # 实际实现中，需要根据具体的CVE API进行调整
        params = {
            'keyword': device_type,
            'versionStartIncluding': firmware_version,
            'versionEndIncluding': firmware_version
        }
        
        try:
            response = requests.get(self.cve_api_url, params=params)
            if response.status_code == 200:
                data = response.json()
                return data.get('result', {}).get('CVE_Items', [])
        except Exception as e:
            print(f"Error querying CVE database: {e}")
        
        return []