物联网安全最佳实践：保护您的IOT设备和数据

// ESP32 Secure Boot 配置
#include "esp_secure_boot.h"
#include "esp_flash_encrypt.h"
 
typedef struct {
    uint8_t signature[64];      // RSA-2048签名
    uint32_t version;           // 固件版本
    uint32_t timestamp;         // 构建时间戳
    uint8_t hash[32];          // SHA-256固件哈希
} firmware_header_t;
 
bool verify_firmware_signature(const uint8_t* firmware_data, size_t length) {
    firmware_header_t* header = (firmware_header_t*)firmware_data;
    
    // 验证固件签名
    if (!esp_secure_boot_verify_signature(
        firmware_data + sizeof(firmware_header_t),
        length - sizeof(firmware_header_t),
        header->signature)) {
        ESP_LOGE("SECURITY", "固件签名验证失败");
        return false;
    }
    
    // 验证固件哈希
    uint8_t calculated_hash[32];
    esp_sha256(firmware_data + sizeof(firmware_header_t), 
               length - sizeof(firmware_header_t), 
               calculated_hash);
    
    if (memcmp(calculated_hash, header->hash, 32) != 0) {
        ESP_LOGE("SECURITY", "固件哈希验证失败");
        return false;
    }
    
    return true;
}
 
void secure_ota_update() {
    // 下载固件更新
    const esp_partition_t* ota_partition = esp_ota_get_next_update_partition(NULL);
    esp_ota_handle_t ota_handle;
    
    esp_err_t err = esp_ota_begin(ota_partition, OTA_SIZE_UNKNOWN, &ota_handle);
    if (err != ESP_OK) {
        ESP_LOGE("OTA", "OTA开始失败");
        return;
    }
    
    // 在写入前验证每个数据块
    uint8_t ota_write_data[1024];
    while (download_firmware_chunk(ota_write_data, sizeof(ota_write_data))) {
        // 写入数据到OTA分区
        err = esp_ota_write(ota_handle, ota_write_data, sizeof(ota_write_data));
        if (err != ESP_OK) {
            esp_ota_abort(ota_handle);
            return;
        }
    }
    
    // 完成OTA更新前进行最终验证
    err = esp_ota_end(ota_handle);
    if (err == ESP_OK) {
        err = esp_ota_set_boot_partition(ota_partition);
        if (err == ESP_OK) {
            ESP_LOGI("OTA", "安全OTA更新成功，重启应用新固件");
            esp_restart();
        }
    }
}

import cryptography
from cryptography.x509 import load_pem_x509_certificate
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa, padding
import jwt
import uuid
from datetime import datetime, timedelta
 
class DeviceAuthenticationManager:
    def __init__(self, ca_cert_path: str, ca_private_key_path: str):
        # 加载CA证书和私钥
        with open(ca_cert_path, 'rb') as f:
            self.ca_cert = load_pem_x509_certificate(f.read())
            
        with open(ca_private_key_path, 'rb') as f:
            self.ca_private_key = serialization.load_pem_private_key(
                f.read(), password=None
            )
    
    def generate_device_certificate(self, device_id: str, device_type: str) -> tuple:
        """为设备生成证书和私钥"""
        # 生成设备私钥
        device_private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048,
        )
        
        # 创建证书签名请求 (CSR)
        subject = x509.Name([
            x509.NameAttribute(x509.NameOID.COMMON_NAME, device_id),
            x509.NameAttribute(x509.NameOID.ORGANIZATION_NAME, "IoT Devices"),
            x509.NameAttribute(x509.NameOID.ORGANIZATIONAL_UNIT_NAME, device_type),
        ])
        
        # 构建设备证书
        cert = x509.CertificateBuilder().subject_name(
            subject
        ).issuer_name(
            self.ca_cert.subject
        ).public_key(
            device_private_key.public_key()
        ).serial_number(
            int(uuid.uuid4().hex, 16)
        ).not_valid_before(
            datetime.utcnow()
        ).not_valid_after(
            datetime.utcnow() + timedelta(days=365)
        ).add_extension(
            x509.SubjectAlternativeName([
                x509.DNSName(f"{device_id}.iot.local"),
                x509.IPAddress("127.0.0.1"),
            ]),
            critical=False,
        ).add_extension(
            x509.KeyUsage(
                digital_signature=True,
                key_encipherment=True,
                data_encipherment=False,
                key_agreement=False,
                key_cert_sign=False,
                crl_sign=False,
                content_commitment=False,
                encipher_only=False,
                decipher_only=False
            ),
            critical=True,
        ).sign(self.ca_private_key, hashes.SHA256())
        
        return cert, device_private_key
    
    def generate_device_token(self, device_id: str, device_cert, 
                            expires_hours: int = 24) -> str:
        """生成设备JWT访问令牌"""
        payload = {
            'device_id': device_id,
            'iss': 'iot-auth-service',
            'sub': device_id,
            'aud': 'iot-platform',
            'exp': datetime.utcnow() + timedelta(hours=expires_hours),
            'iat': datetime.utcnow(),
            'cert_fingerprint': self._get_cert_fingerprint(device_cert)
        }
        
        # 使用CA私钥签名JWT
        token = jwt.encode(payload, self.ca_private_key, algorithm='RS256')
        return token
    
    def verify_device_token(self, token: str) -> dict:
        """验证设备令牌"""
        try:
            # 使用CA公钥验证JWT
            payload = jwt.decode(
                token, 
                self.ca_cert.public_key(), 
                algorithms=['RS256'],
                audience='iot-platform'
            )
            return {'valid': True, 'payload': payload}
        except jwt.InvalidTokenError as e:
            return {'valid': False, 'error': str(e)}
    
    def _get_cert_fingerprint(self, cert) -> str:
        """获取证书指纹"""
        return cert.fingerprint(hashes.SHA256()).hex()
 
# 使用示例
auth_manager = DeviceAuthenticationManager('/certs/ca.crt', '/certs/ca.key')
 
# 为新设备生成证书
device_cert, device_key = auth_manager.generate_device_certificate(
    device_id='sensor_001', 
    device_type='temperature_sensor'
)
 
# 生成访问令牌
access_token = auth_manager.generate_device_token('sensor_001', device_cert)

const mqtt = require('mqtt');
const fs = require('fs');
const crypto = require('crypto');
 
class SecureMQTTClient {
  constructor(brokerHost, options = {}) {
    this.brokerHost = brokerHost;
    this.clientId = options.clientId || this.generateClientId();
    
    // TLS选项配置
    this.tlsOptions = {
      port: 8883,
      protocol: 'mqtts',
      
      // 证书配置
      ca: fs.readFileSync(options.caFile || './certs/ca.crt'),
      cert: fs.readFileSync(options.certFile || './certs/client.crt'),
      key: fs.readFileSync(options.keyFile || './certs/client.key'),
      
      // 安全选项
      rejectUnauthorized: true,
      secureProtocol: 'TLSv1_2_method',
      ciphers: 'ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM:DHE+CHACHA20:!aNULL:!MD5:!DSS',
      
      // 连接选项
      clientId: this.clientId,
      clean: true,
      connectTimeout: 10000,
      reconnectPeriod: 5000,
    };
    
    this.client = null;
    this.messageQueue = [];
  }
  
  generateClientId() {
    const timestamp = Date.now();
    const random = crypto.randomBytes(4).toString('hex');
    return `secure_iot_${timestamp}_${random}`;
  }
  
  async connect() {
    return new Promise((resolve, reject) => {
      this.client = mqtt.connect(`mqtts://${this.brokerHost}`, this.tlsOptions);
      
      this.client.on('connect', () => {
        console.log('安全MQTT连接已建立');
        this.processMessageQueue();
        resolve(true);
      });
      
      this.client.on('error', (error) => {
        console.error('MQTT连接错误:', error);
        reject(error);
      });
      
      this.client.on('close', () => {
        console.log('MQTT连接已关闭');
      });
      
      this.setupMessageHandler();
    });
  }
  
  setupMessageHandler() {
    this.client.on('message', (topic, message) => {
      try {
        // 验证消息完整性
        const payload = JSON.parse(message.toString());
        
        if (this.verifyMessageIntegrity(topic, payload)) {
          this.handleSecureMessage(topic, payload);
        } else {
          console.warn('消息完整性验证失败:', topic);
        }
      } catch (error) {
        console.error('消息处理错误:', error);
      }
    });
  }
  
  verifyMessageIntegrity(topic, payload) {
    // 验证消息时间戳（防重放攻击）
    const messageTime = new Date(payload.timestamp);
    const currentTime = new Date();
    const timeDiff = Math.abs(currentTime - messageTime);
    
    if (timeDiff > 300000) { // 5分钟时间窗口
      console.warn('消息时间戳过期:', topic);
      return false;
    }
    
    // 验证消息签名（如果存在）
    if (payload.signature) {
      return this.verifyMessageSignature(payload);
    }
    
    return true;
  }
  
  verifyMessageSignature(payload) {
    // 实现消息签名验证逻辑
    // 这里简化处理，实际应用中需要完整的签名验证
    return true;
  }
  
  publishSecureMessage(topic, data, options = {}) {
    const securePayload = {
      ...data,
      timestamp: new Date().toISOString(),
      client_id: this.clientId,
      nonce: crypto.randomBytes(16).toString('hex')
    };
    
    // 添加消息签名（可选）
    if (options.signed) {
      securePayload.signature = this.signMessage(securePayload);
    }
    
    const qos = options.qos || 1;
    const retain = options.retain || false;
    
    if (this.client && this.client.connected) {
      this.client.publish(topic, JSON.stringify(securePayload), { qos, retain });
    } else {
      // 连接断开时将消息加入队列
      this.messageQueue.push({ topic, payload: securePayload, options });
    }
  }
  
  signMessage(payload) {
    // 简化的消息签名实现
    const messageString = JSON.stringify(payload);
    return crypto.createHash('sha256').update(messageString).digest('hex');
  }
  
  processMessageQueue() {
    while (this.messageQueue.length > 0) {
      const { topic, payload, options } = this.messageQueue.shift();
      this.client.publish(topic, JSON.stringify(payload), options);
    }
  }
}

import base64
import json
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import os
 
class IoTMessageEncryption:
    def __init__(self, password: bytes, salt: bytes = None):
        if salt is None:
            salt = os.urandom(16)
        
        # 使用PBKDF2从密码派生密钥
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=100000,
        )
        
        key = base64.urlsafe_b64encode(kdf.derive(password))
        self.cipher = Fernet(key)
        self.salt = salt
    
    def encrypt_message(self, data: dict) -> str:
        """加密消息数据"""
        try:
            # 序列化数据
            json_data = json.dumps(data, ensure_ascii=False)
            
            # 加密
            encrypted_data = self.cipher.encrypt(json_data.encode('utf-8'))
            
            # Base64编码便于传输
            return base64.b64encode(encrypted_data).decode('utf-8')
        except Exception as e:
            raise Exception(f"加密失败: {e}")
    
    def decrypt_message(self, encrypted_data: str) -> dict:
        """解密消息数据"""
        try:
            # Base64解码
            encrypted_bytes = base64.b64decode(encrypted_data.encode('utf-8'))
            
            # 解密
            decrypted_data = self.cipher.decrypt(encrypted_bytes)
            
            # 反序列化
            return json.loads(decrypted_data.decode('utf-8'))
        except Exception as e:
            raise Exception(f"解密失败: {e}")
 
class SecureIoTDevice:
    def __init__(self, device_id: str, encryption_password: str):
        self.device_id = device_id
        self.encryption = IoTMessageEncryption(encryption_password.encode())
        self.mqtt_client = None
        
    def setup_secure_connection(self, broker_host: str):
        """建立安全MQTT连接"""
        import paho.mqtt.client as mqtt
        
        self.mqtt_client = mqtt.Client(client_id=self.device_id)
        
        # SSL/TLS配置
        context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
        context.check_hostname = False
        context.verify_mode = ssl.CERT_REQUIRED
        
        self.mqtt_client.tls_set_context(context)
        
        # 设置回调
        self.mqtt_client.on_connect = self._on_connect
        self.mqtt_client.on_message = self._on_secure_message
        
        # 连接到broker
        self.mqtt_client.connect(broker_host, 8883, 60)
        self.mqtt_client.loop_start()
    
    def _on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            print(f"设备 {self.device_id} 安全连接成功")
            # 订阅设备控制主题
            client.subscribe(f"control/{self.device_id}/+", qos=2)
        else:
            print(f"连接失败: {rc}")
    
    def _on_secure_message(self, client, userdata, msg):
        """处理加密消息"""
        try:
            # 解密消息
            decrypted_data = self.encryption.decrypt_message(msg.payload.decode())
            
            # 验证消息来源和完整性
            if self._verify_message_source(decrypted_data):
                self._handle_control_command(msg.topic, decrypted_data)
            else:
                print(f"消息验证失败: {msg.topic}")
        except Exception as e:
            print(f"安全消息处理失败: {e}")
    
    def publish_encrypted_data(self, topic: str, data: dict):
        """发布加密数据"""
        # 添加安全元数据
        secure_data = {
            **data,
            'device_id': self.device_id,
            'timestamp': datetime.now().isoformat(),
            'nonce': os.urandom(16).hex()
        }
        
        # 加密并发布
        encrypted_payload = self.encryption.encrypt_message(secure_data)
        self.mqtt_client.publish(topic, encrypted_payload, qos=1)
    
    def _verify_message_source(self, data: dict) -> bool:
        """验证消息来源"""
        # 检查时间戳防重放攻击
        message_time = datetime.fromisoformat(data.get('timestamp', ''))
        time_diff = (datetime.now() - message_time).total_seconds()
        
        if time_diff > 300:  # 5分钟时间窗口
            return False
            
        # 检查nonce防重复
        if hasattr(self, 'processed_nonces'):
            if data.get('nonce') in self.processed_nonces:
                return False
            self.processed_nonces.add(data.get('nonce'))
        else:
            self.processed_nonces = {data.get('nonce')}
            
        return True

#!/bin/bash
 
# 创建IoT设备专用网络命名空间
sudo ip netns add iot_secure_network
 
# 配置虚拟网络接口
sudo ip link add veth_iot type veth peer name veth_iot_peer
sudo ip link set veth_iot_peer netns iot_secure_network
 
# 在命名空间内配置网络
sudo ip netns exec iot_secure_network ip addr add 192.168.100.1/24 dev veth_iot_peer
sudo ip netns exec iot_secure_network ip link set veth_iot_peer up
sudo ip netns exec iot_secure_network ip link set lo up
 
# 配置iptables防火墙规则
sudo iptables -N IOT_SECURITY_CHAIN
 
# 只允许必要的端口和协议
sudo iptables -A IOT_SECURITY_CHAIN -p tcp --dport 8883 -j ACCEPT  # MQTTS
sudo iptables -A IOT_SECURITY_CHAIN -p tcp --dport 443 -j ACCEPT   # HTTPS
sudo iptables -A IOT_SECURITY_CHAIN -p udp --dport 123 -j ACCEPT   # NTP
sudo iptables -A IOT_SECURITY_CHAIN -p udp --dport 53 -j ACCEPT    # DNS
 
# 阻止其他所有连接
sudo iptables -A IOT_SECURITY_CHAIN -j DROP
 
# 应用规则到IoT网络接口
sudo iptables -A FORWARD -i veth_iot -j IOT_SECURITY_CHAIN

import asyncio
import json
import time
from collections import defaultdict, deque
from typing import Dict, List, Set
import statistics
 
class IoTIntrusionDetection:
    def __init__(self):
        # 设备行为基线
        self.device_baselines = defaultdict(dict)
        
        # 异常检测阈值
        self.thresholds = {
            'max_message_rate': 100,    # 每分钟最大消息数
            'max_payload_size': 1024,   # 最大载荷大小
            'max_topic_diversity': 20,  # 最大主题多样性
            'min_message_interval': 1,  # 最小消息间隔(秒)
        }
        
        # 设备活动历史
        self.device_activities = defaultdict(lambda: deque(maxlen=1000))
        self.suspicious_activities = []
        
    def learn_device_baseline(self, device_id: str, activities: List[Dict]):
        """学习设备正常行为基线"""
        message_intervals = []
        payload_sizes = []
        topics = set()
        
        for i, activity in enumerate(activities[1:], 1):
            prev_activity = activities[i-1]
            interval = activity['timestamp'] - prev_activity['timestamp']
            message_intervals.append(interval)
            payload_sizes.append(len(activity.get('payload', '')))
            topics.add(activity['topic'])
        
        # 计算基线统计信息
        baseline = {
            'avg_message_interval': statistics.mean(message_intervals),
            'std_message_interval': statistics.stdev(message_intervals),
            'avg_payload_size': statistics.mean(payload_sizes),
            'std_payload_size': statistics.stdev(payload_sizes),
            'typical_topics': topics,
            'message_rate_per_minute': len(activities) / (activities[-1]['timestamp'] - activities[0]['timestamp']) * 60
        }
        
        self.device_baselines[device_id] = baseline
        print(f"设备 {device_id} 基线学习完成: {baseline}")
    
    def analyze_message(self, device_id: str, topic: str, payload: str, timestamp: float) -> Dict:
        """分析单个消息的安全性"""
        activity = {
            'device_id': device_id,
            'topic': topic,
            'payload': payload,
            'timestamp': timestamp,
            'payload_size': len(payload)
        }
        
        self.device_activities[device_id].append(activity)
        
        # 异常检测
        anomalies = []
        
        # 1. 检测消息频率异常
        anomalies.extend(self._detect_frequency_anomaly(device_id))
        
        # 2. 检测载荷大小异常  
        anomalies.extend(self._detect_payload_size_anomaly(device_id, len(payload)))
        
        # 3. 检测主题异常
        anomalies.extend(self._detect_topic_anomaly(device_id, topic))
        
        # 4. 检测时间模式异常
        anomalies.extend(self._detect_timing_anomaly(device_id, timestamp))
        
        if anomalies:
            self._handle_security_alert(device_id, anomalies, activity)
        
        return {
            'device_id': device_id,
            'anomaly_count': len(anomalies),
            'anomalies': anomalies,
            'risk_level': self._calculate_risk_level(anomalies)
        }
    
    def _detect_frequency_anomaly(self, device_id: str) -> List[str]:
        """检测消息频率异常"""
        activities = list(self.device_activities[device_id])
        if len(activities) < 10:
            return []
        
        # 计算最近1分钟的消息数
        current_time = time.time()
        recent_messages = [a for a in activities if current_time - a['timestamp'] <= 60]
        
        if len(recent_messages) > self.thresholds['max_message_rate']:
            return [f"消息频率异常: {len(recent_messages)}/分钟 (阈值: {self.thresholds['max_message_rate']})"]
        
        return []
    
    def _detect_payload_size_anomaly(self, device_id: str, payload_size: int) -> List[str]:
        """检测载荷大小异常"""
        anomalies = []
        
        # 检查绝对大小限制
        if payload_size > self.thresholds['max_payload_size']:
            anomalies.append(f"载荷过大: {payload_size} bytes (阈值: {self.thresholds['max_payload_size']})")
        
        # 检查相对于基线的异常
        baseline = self.device_baselines.get(device_id)
        if baseline:
            avg_size = baseline['avg_payload_size']
            std_size = baseline['std_payload_size']
            
            # 3倍标准差规则
            if payload_size > avg_size + 3 * std_size:
                anomalies.append(f"载荷大小异常: {payload_size} (基线: {avg_size:.1f}±{std_size:.1f})")
        
        return anomalies
    
    def _detect_topic_anomaly(self, device_id: str, topic: str) -> List[str]:
        """检测主题异常"""
        baseline = self.device_baselines.get(device_id)
        if not baseline:
            return []
        
        # 检查是否为设备典型主题
        typical_topics = baseline['typical_topics']
        
        if topic not in typical_topics:
            return [f"异常主题访问: {topic}"]
        
        return []
    
    def _detect_timing_anomaly(self, device_id: str, timestamp: float) -> List[str]:
        """检测时间模式异常"""
        activities = list(self.device_activities[device_id])
        if len(activities) < 2:
            return []
        
        # 检查消息间隔异常
        last_activity = activities[-2]
        interval = timestamp - last_activity['timestamp']
        
        baseline = self.device_baselines.get(device_id)
        if baseline:
            avg_interval = baseline['avg_message_interval']
            std_interval = baseline['std_message_interval']
            
            # 检查异常快速或缓慢的消息间隔
            if interval < avg_interval - 3 * std_interval:
                return [f"消息间隔过短: {interval:.2f}s (基线: {avg_interval:.2f}±{std_interval:.2f})"]
            elif interval > avg_interval + 5 * std_interval:
                return [f"消息间隔过长: {interval:.2f}s"]
        
        return []
    
    def _calculate_risk_level(self, anomalies: List[str]) -> str:
        """计算风险等级"""
        if not anomalies:
            return 'low'
        elif len(anomalies) == 1:
            return 'medium' 
        else:
            return 'high'
    
    def _handle_security_alert(self, device_id: str, anomalies: List[str], activity: Dict):
        """处理安全告警"""
        alert = {
            'alert_id': f"alert_{int(time.time())}_{device_id}",
            'device_id': device_id,
            'timestamp': datetime.now().isoformat(),
            'anomalies': anomalies,
            'activity': activity,
            'risk_level': self._calculate_risk_level(anomalies)
        }
        
        self.suspicious_activities.append(alert)
        
        # 根据风险等级采取不同行动
        risk_level = alert['risk_level']
        if risk_level == 'high':
            self._block_device(device_id)
            self._send_urgent_notification(alert)
        elif risk_level == 'medium':
            self._increase_monitoring(device_id)
            self._send_warning_notification(alert)
        
        print(f"安全告警 [{risk_level.upper()}]: 设备 {device_id}")
        for anomaly in anomalies:
            print(f"  - {anomaly}")
    
    def _block_device(self, device_id: str):
        """阻断可疑设备"""
        # 实现设备阻断逻辑
        print(f"已阻断可疑设备: {device_id}")
    
    def get_security_report(self) -> Dict:
        """生成安全报告"""
        total_devices = len(self.device_baselines)
        total_alerts = len(self.suspicious_activities)
        
        # 按风险等级统计
        risk_stats = defaultdict(int)
        for alert in self.suspicious_activities:
            risk_stats[alert['risk_level']] += 1
        
        return {
            'monitored_devices': total_devices,
            'total_alerts': total_alerts,
            'risk_distribution': dict(risk_stats),
            'recent_alerts': self.suspicious_activities[-10:],  # 最近10个告警
        }
 
# 使用示例
ids = IoTIntrusionDetection()
 
# 监控设备消息
def monitor_device_message(device_id, topic, payload):
    result = ids.analyze_message(device_id, topic, payload, time.time())
    
    if result['anomaly_count'] > 0:
        print(f"检测到 {result['anomaly_count']} 个异常，风险等级: {result['risk_level']}")

const express = require('express');
const rateLimit = require('express-rate-limit');
const helmet = require('helmet');
const jwt = require('jsonwebtoken');
const crypto = require('crypto');
 
const app = express();
 
// 安全头部设置
app.use(helmet({
  contentSecurityPolicy: {
    directives: {
      defaultSrc: ["'self'"],
      scriptSrc: ["'self'", "'unsafe-inline'"],
      styleSrc: ["'self'", "'unsafe-inline'"],
      imgSrc: ["'self'", "data:", "https:"],
    },
  },
  hsts: {
    maxAge: 31536000,
    includeSubDomains: true,
    preload: true
  }
}));
 
// 速率限制
const deviceApiLimiter = rateLimit({
  windowMs: 15 * 60 * 1000, // 15分钟
  max: 1000, // 限制每个设备每15分钟最多1000次请求
  message: {
    error: 'Too many requests from this device',
    code: 'RATE_LIMIT_EXCEEDED'
  },
  keyGenerator: (req) => {
    return req.headers['x-device-id'] || req.ip;
  }
});
 
// 设备认证中间件
const deviceAuth = (req, res, next) => {
  const token = req.headers['authorization']?.replace('Bearer ', '');
  const deviceId = req.headers['x-device-id'];
  
  if (!token || !deviceId) {
    return res.status(401).json({ 
      error: 'Missing authentication credentials',
      code: 'AUTH_REQUIRED'
    });
  }
  
  try {
    // 验证JWT令牌
    const payload = jwt.verify(token, process.env.JWT_SECRET);
    
    // 验证设备ID匹配
    if (payload.device_id !== deviceId) {
      return res.status(401).json({
        error: 'Device ID mismatch',
        code: 'DEVICE_ID_MISMATCH'
      });
    }
    
    // 验证设备是否在白名单中
    if (!isDeviceWhitelisted(deviceId)) {
      return res.status(403).json({
        error: 'Device not authorized',
        code: 'DEVICE_NOT_AUTHORIZED'
      });
    }
    
    req.deviceId = deviceId;
    req.deviceInfo = payload;
    next();
    
  } catch (error) {
    return res.status(401).json({
      error: 'Invalid token',
      code: 'INVALID_TOKEN',
      details: error.message
    });
  }
};
 
// 请求签名验证中间件
const verifyRequestSignature = (req, res, next) => {
  const signature = req.headers['x-signature'];
  const timestamp = req.headers['x-timestamp'];
  const deviceId = req.deviceId;
  
  if (!signature || !timestamp) {
    return res.status(400).json({
      error: 'Missing signature or timestamp',
      code: 'SIGNATURE_REQUIRED'
    });
  }
  
  // 检查时间戳（防重放攻击）
  const requestTime = parseInt(timestamp);
  const currentTime = Math.floor(Date.now() / 1000);
  
  if (Math.abs(currentTime - requestTime) > 300) { // 5分钟窗口
    return res.status(400).json({
      error: 'Request timestamp too old',
      code: 'TIMESTAMP_EXPIRED'
    });
  }
  
  // 验证请求签名
  const deviceSecret = getDeviceSecret(deviceId);
  const expectedSignature = crypto
    .createHmac('sha256', deviceSecret)
    .update(`${req.method}${req.originalUrl}${timestamp}${JSON.stringify(req.body)}`)
    .digest('hex');
  
  if (signature !== expectedSignature) {
    return res.status(401).json({
      error: 'Invalid request signature',
      code: 'INVALID_SIGNATURE'
    });
  }
  
  next();
};
 
// IoT设备数据接收API
app.post('/api/v1/device/data', 
  deviceApiLimiter,
  express.json({ limit: '1mb' }),
  deviceAuth,
  verifyRequestSignature,
  async (req, res) => {
    try {
      const { deviceId } = req;
      const sensorData = req.body;
      
      // 数据验证
      if (!validateSensorData(sensorData)) {
        return res.status(400).json({
          error: 'Invalid sensor data format',
          code: 'INVALID_DATA_FORMAT'
        });
      }
      
      // 存储数据到安全数据库
      await storeSensorDataSecurely(deviceId, sensorData);
      
      // 触发实时处理
      await processRealTimeData(deviceId, sensorData);
      
      res.json({
        status: 'success',
        message: 'Data received and processed',
        timestamp: new Date().toISOString()
      });
      
    } catch (error) {
      console.error('设备数据处理错误:', error);
      res.status(500).json({
        error: 'Internal server error',
        code: 'PROCESSING_ERROR'
      });
    }
  }
);
 
function isDeviceWhitelisted(deviceId) {
  // 实现设备白名单检查逻辑
  const whitelistedDevices = process.env.WHITELISTED_DEVICES?.split(',') || [];
  return whitelistedDevices.includes(deviceId);
}
 
function getDeviceSecret(deviceId) {
  // 实现设备密钥获取逻辑
  // 在生产环境中应从安全的密钥管理服务获取
  return process.env[`DEVICE_SECRET_${deviceId}`] || 'default_secret';
}

from sqlalchemy import create_engine, Column, String, DateTime, Text, LargeBinary
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from cryptography.fernet import Fernet
import json
import base64
 
Base = declarative_base()
 
class EncryptedIoTData(Base):
    __tablename__ = 'encrypted_iot_data'
    
    id = Column(String, primary_key=True)
    device_id = Column(String, nullable=False, index=True)
    encrypted_payload = Column(LargeBinary, nullable=False)
    data_hash = Column(String, nullable=False)  # 用于完整性验证
    timestamp = Column(DateTime, nullable=False, index=True)
    
class SecureIoTDataManager:
    def __init__(self, database_url: str, encryption_key: bytes):
        self.engine = create_engine(database_url)
        self.Session = sessionmaker(bind=self.engine)
        self.cipher = Fernet(encryption_key)
        
        # 创建表
        Base.metadata.create_all(self.engine)
    
    def store_device_data(self, device_id: str, data: dict) -> str:
        """安全存储设备数据"""
        session = self.Session()
        
        try:
            # 序列化数据
            json_data = json.dumps(data, ensure_ascii=False)
            
            # 加密数据
            encrypted_data = self.cipher.encrypt(json_data.encode('utf-8'))
            
            # 计算数据哈希
            data_hash = hashlib.sha256(json_data.encode('utf-8')).hexdigest()
            
            # 生成唯一ID
            record_id = f"{device_id}_{int(time.time() * 1000)}"
            
            # 存储到数据库
            encrypted_record = EncryptedIoTData(
                id=record_id,
                device_id=device_id,
                encrypted_payload=encrypted_data,
                data_hash=data_hash,
                timestamp=datetime.now()
            )
            
            session.add(encrypted_record)
            session.commit()
            
            return record_id
            
        except Exception as e:
            session.rollback()
            raise Exception(f"数据存储失败: {e}")
        finally:
            session.close()
    
    def retrieve_device_data(self, device_id: str, start_time: datetime, 
                           end_time: datetime) -> List[dict]:
        """检索并解密设备数据"""
        session = self.Session()
        
        try:
            # 查询加密数据
            encrypted_records = session.query(EncryptedIoTData).filter(
                EncryptedIoTData.device_id == device_id,
                EncryptedIoTData.timestamp >= start_time,
                EncryptedIoTData.timestamp <= end_time
            ).order_by(EncryptedIoTData.timestamp).all()
            
            decrypted_data = []
            
            for record in encrypted_records:
                try:
                    # 解密数据
                    decrypted_bytes = self.cipher.decrypt(record.encrypted_payload)
                    json_data = decrypted_bytes.decode('utf-8')
                    
                    # 验证数据完整性
                    calculated_hash = hashlib.sha256(json_data.encode('utf-8')).hexdigest()
                    if calculated_hash != record.data_hash:
                        print(f"数据完整性验证失败: {record.id}")
                        continue
                    
                    # 解析JSON
                    data = json.loads(json_data)
                    data['_metadata'] = {
                        'record_id': record.id,
                        'timestamp': record.timestamp.isoformat()
                    }
                    
                    decrypted_data.append(data)
                    
                except Exception as e:
                    print(f"数据解密失败 {record.id}: {e}")
                    continue
            
            return decrypted_data
            
        finally:
            session.close()

import asyncio
import websockets
import json
from datetime import datetime
from typing import Dict, List, Callable
 
class SecurityEventMonitor:
    def __init__(self):
        self.event_handlers: Dict[str, List[Callable]] = {
            'device_authentication_failed': [],
            'unusual_traffic_pattern': [],
            'unauthorized_access_attempt': [],
            'data_integrity_violation': [],
            'suspicious_payload_detected': []
        }
        
        self.active_incidents = {}
        self.security_metrics = {
            'total_events': 0,
            'blocked_attempts': 0,
            'false_positives': 0,
            'response_times': []
        }
    
    def register_event_handler(self, event_type: str, handler: Callable):
        """注册安全事件处理器"""
        if event_type in self.event_handlers:
            self.event_handlers[event_type].append(handler)
    
    async def emit_security_event(self, event_type: str, event_data: Dict):
        """发出安全事件"""
        self.security_metrics['total_events'] += 1
        
        event = {
            'id': f"event_{int(time.time() * 1000)}",
            'type': event_type,
            'timestamp': datetime.now().isoformat(),
            'data': event_data,
            'severity': self._calculate_severity(event_type, event_data)
        }
        
        # 执行注册的处理器
        start_time = time.time()
        
        for handler in self.event_handlers.get(event_type, []):
            try:
                await handler(event)
            except Exception as e:
                print(f"事件处理器错误: {e}")
        
        response_time = time.time() - start_time
        self.security_metrics['response_times'].append(response_time)
        
        # 创建安全事件（如果严重）
        if event['severity'] >= 7:
            await self._create_security_incident(event)
    
    def _calculate_severity(self, event_type: str, event_data: Dict) -> int:
        """计算事件严重程度 (1-10)"""
        severity_map = {
            'device_authentication_failed': 6,
            'unusual_traffic_pattern': 5,
            'unauthorized_access_attempt': 8,
            'data_integrity_violation': 9,
            'suspicious_payload_detected': 7
        }
        
        base_severity = severity_map.get(event_type, 5)
        
        # 根据事件数据调整严重程度
        if event_data.get('repeated_attempts', 0) > 5:
            base_severity += 2
        
        if event_data.get('high_privilege_target', False):
            base_severity += 1
            
        return min(10, base_severity)
    
    async def _create_security_incident(self, event: Dict):
        """创建安全事件"""
        incident_id = f"incident_{event['id']}"
        
        incident = {
            'id': incident_id,
            'title': f"安全事件: {event['type']}",
            'description': f"检测到{event['type']}类型的安全威胁",
            'severity': event['severity'],
            'status': 'open',
            'created_at': datetime.now().isoformat(),
            'events': [event],
            'affected_devices': self._extract_affected_devices(event)
        }
        
        self.active_incidents[incident_id] = incident
        
        # 发送告警通知
        await self._send_security_alert(incident)
    
    async def _send_security_alert(self, incident: Dict):
        """发送安全告警"""
        alert_message = {
            'type': 'security_alert',
            'incident_id': incident['id'],
            'severity': incident['severity'],
            'title': incident['title'],
            'timestamp': incident['created_at'],
            'affected_devices': incident['affected_devices']
        }
        
        # 发送到WebSocket客户端
        await self._broadcast_to_security_dashboard(alert_message)
        
        # 发送邮件通知（高严重性事件）
        if incident['severity'] >= 8:
            await self._send_email_alert(incident)
    
    async def _broadcast_to_security_dashboard(self, message: Dict):
        """广播到安全监控仪表板"""
        # WebSocket实现
        message_json = json.dumps(message)
        
        # 假设有WebSocket连接池
        for websocket in getattr(self, 'websocket_connections', []):
            try:
                await websocket.send(message_json)
            except websockets.exceptions.ConnectionClosed:
                pass

device_security_checklist:
  hardware:
    - 禁用调试接口
    - 启用安全启动
    - 硬件随机数生成器
    - 防篡改保护
    
  firmware:
    - 代码签名验证
    - 安全OTA更新
    - 最小权限原则
    - 定期安全更新
    
  communication:
    - TLS/DTLS加密
    - 证书绑定
    - 消息完整性校验
    - 防重放攻击
    
  authentication:
    - 强设备标识
    - PKI证书管理
    - 访问控制列表
    - 密钥轮换策略
    
  monitoring:
    - 实时威胁检测
    - 异常行为分析  
    - 安全日志审计
    - 事件响应流程

interface GDPRCompliantDataProcessor {
  processPersonalData(deviceId: string, data: any): Promise<void>;
  anonymizeData(data: any): any;
  deleteUserData(userId: string): Promise<void>;
  exportUserData(userId: string): Promise<any>;
}
 
class IoTGDPRProcessor implements GDPRCompliantDataProcessor {
  async processPersonalData(deviceId: string, data: any): Promise<void> {
    // 数据分类
    const personalData = this.extractPersonalData(data);
    const technicalData = this.extractTechnicalData(data);
    
    // 个人数据加密存储
    if (Object.keys(personalData).length > 0) {
      await this.storeEncryptedPersonalData(deviceId, personalData);
    }
    
    // 技术数据可以正常处理
    await this.processTechnicalData(deviceId, technicalData);
  }
  
  anonymizeData(data: any): any {
    // 实现数据匿名化
    const anonymized = { ...data };
    
    // 移除直接标识符
    delete anonymized.user_id;
    delete anonymized.email;
    delete anonymized.phone;
    
    // 泛化准直接标识符
    if (anonymized.location) {
      anonymized.location = this.generalizeLocation(anonymized.location);
    }
    
    if (anonymized.timestamp) {
      anonymized.timestamp = this.generalizeTimestamp(anonymized.timestamp);
    }
    
    return anonymized;
  }
  
  async deleteUserData(userId: string): Promise<void> {
    // 实现用户数据删除（被遗忘权）
    await this.deleteFromPersonalDataStore(userId);
    await this.anonymizeHistoricalData(userId);
    await this.updateDataRetentionPolicies(userId);
  }
}