人工智能与物联网的安全与隐私保护

    // 安全世界代码 (trusted_app.c)
    #include <tee_internal_api.h>
    #include <tee_internal_api_extensions.h>
    
    #define SECURE_DATA_SIZE 256
    
    // 安全存储结构体
    typedef struct {
    uint8_t encrypted_data[SECURE_DATA_SIZE];
    uint32_t access_counter;
    } secure_store_t;
    
    // 安全服务函数
    TEE_Result secure_process_data(uint32_t param_types, TEE_Param params[4]) {
    static secure_store_t secure_data = {0};
    uint8_t *input = params[0].memref.buffer;
    uint32_t size = params[0].memref.size;
    
    if (size > SECURE_DATA_SIZE)
        return TEE_ERROR_SHORT_BUFFER;
    
    // 模拟加密处理
    for (int i = 0; i < size; i++) {
        secure_data.encrypted_data[i] = input[i] ^ 0xAA; // 简单XOR加密
    }
    secure_data.access_counter++;
    
    // 返回处理结果
    params[1].value.a = secure_data.access_counter;
    return TEE_SUCCESS;
    }
    
    // 非安全世界调用代码 (main.c)
    #include <stdio.h>
    #include <tee_client_api.h>
    
    int main() {
    TEEC_Context ctx;
    TEEC_Session sess;
    TEEC_Operation op = {0};
    TEEC_UUID uuid = {0x12345678, 0x8765, 0x4321, 
                      {0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08}};
    
    // 初始化TEE上下文
    TEEC_InitializeContext(NULL, &ctx);
    
    // 打开安全会话
    TEEC_OpenSession(&ctx, &sess, &uuid, TEEC_LOGIN_PUBLIC, NULL, NULL, NULL);
    
    uint8_t data[] = "Sensitive IoT data";
    uint32_t count;
    
    op.paramTypes = TEEC_PARAM_TYPES(TEEC_MEMREF_TEMP_INPUT, TEEC_VALUE_OUTPUT, 
                                    TEEC_NONE, TEEC_NONE);
    op.params[0].tmpref.buffer = data;
    op.params[0].tmpref.size = sizeof(data);
    
    // 调用安全服务
    TEEC_InvokeCommand(&sess, 1, &op, NULL);
    count = op.params[1].value.a;
    
    printf("Data processed in TEE, access count: %u\n", count);
    
    // 清理资源
    TEEC_CloseSession(&sess);
    TEEC_FinalizeContext(&ctx);
    return 0;
    }

    from cryptography.hazmat.primitives import hashes
    from cryptography.hazmat.primitives.asymmetric import ec
    from cryptography.hazmat.primitives import serialization
    from cryptography.hazmat.backends import default_backend
    from cryptography.x509 import load_pem_x509_certificate
    import os
    
    class DeviceIdentity:
    def __init__(self, device_id):
        self.device_id = device_id
        self.private_key = None
        self.certificate = None
        
    def generate_keys(self):
        # 生成ECDSA P-256密钥对
        self.private_key = ec.generate_private_key(ec.SECP256R1(), default_backend())
        
        # 保存私钥
        pem = self.private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        )
        with open(f"{self.device_id}_private.pem", "wb") as f:
            f.write(pem)
            
        return pem
    
    def sign_data(self, data):
        if not self.private_key:
            raise ValueError("Private key not initialized")
        
        # 使用SHA-256进行签名
        signature = self.private_key.sign(
            data,
            ec.ECDSA(hashes.SHA256())
        )
        return signature
    
    def verify_certificate(self, cert_path, ca_cert_path):
        # 加载设备证书
        with open(cert_path, "rb") as f:
            cert = load_pem_x509_certificate(f.read(), default_backend())
            
        # 加载CA证书
        with open(ca_cert_path, "rb") as f:
            ca_cert = load_pem_x509_certificate(f.read(), default_backend())
            
        # 验证证书链
        ca_pubkey = ca_cert.public_key()
        ca_pubkey.verify(
            cert.signature,
            cert.tbs_certificate_bytes,
            ec.ECDSA(hashes.SHA256())
        )
        self.certificate = cert
        return True
    
    # 安全启动验证流程示例
    def secure_boot_verify(image_path, signature_path, pubkey_path):
    # 加载固件镜像
    with open(image_path, "rb") as f:
        firmware = f.read()
    
    # 加载签名
    with open(signature_path, "rb") as f:
        signature = f.read()
    
    # 加载公钥
    with open(pubkey_path, "rb") as f:
        pubkey = serialization.load_pem_public_key(
            f.read(),
            backend=default_backend()
        )
    
    # 验证签名
    try:
        pubkey.verify(
            signature,
            firmware,
            ec.ECDSA(hashes.SHA256())
        )
        print("Secure boot verification passed")
        return True
    except Exception as e:
        print(f"Verification failed: {str(e)}")
        return False
    
    # 使用示例
    device = DeviceIdentity("sensor_node_001")
    device.generate_keys()
    
    # 模拟安全启动
    secure_boot_verify("firmware.bin", "firmware.sig", "manufacturer_pubkey.pem")

    from socket import socket, AF_INET, SOCK_DGRAM
    from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
    from cryptography.hazmat.primitives import padding
    from cryptography.hazmat.backends import default_backend
    import os
    import hmac
    import hashlib
    
    class DTLSConnection:
    def __init__(self, psk, psk_id):
        self.psk = psk
        self.psk_id = psk_id
        self.seq_num = 0
        self.session_keys = None
        
    def derive_keys(self, client_random, server_random):
        # 简化版的密钥派生函数
        key_material = hmac.new(self.psk, client_random + server_random, hashlib.sha256).digest()
        self.session_keys = {
            'client_key': key_material[:16],
            'server_key': key_material[16:32],
            'iv': key_material[32:48]
        }
        
    def encrypt_message(self, plaintext):
        # AES-CBC加密
        padder = padding.PKCS7(128).padder()
        padded_data = padder.update(plaintext) + padder.finalize()
        
        cipher = Cipher(
            algorithms.AES(self.session_keys['client_key']),
            modes.CBC(self.session_keys['iv']),
            backend=default_backend()
        )
        encryptor = cipher.encryptor()
        ciphertext = encryptor.update(padded_data) + encryptor.finalize()
        
        # 添加HMAC
        mac = hmac.new(self.session_keys['client_key'], ciphertext, hashlib.sha256).digest()[:16]
        
        return self.seq_num.to_bytes(4, 'big') + ciphertext + mac
    
    def decrypt_message(self, ciphertext):
        # 拆分数据包
        seq_num = int.from_bytes(ciphertext[:4], 'big')
        encrypted_data = ciphertext[4:-16]
        received_mac = ciphertext[-16:]
        
        # 验证MAC
        expected_mac = hmac.new(self.session_keys['server_key'], encrypted_data, hashlib.sha256).digest()[:16]
        if not hmac.compare_digest(received_mac, expected_mac):
            raise ValueError("MAC verification failed")
        
        # 解密数据
        cipher = Cipher(
            algorithms.AES(self.session_keys['server_key']),
            modes.CBC(self.session_keys['iv']),
            backend=default_backend()
        )
        decryptor = cipher.decryptor()
        padded_plaintext = decryptor.update(encrypted_data) + decryptor.finalize()
        
        # 去除填充
        unpadder = padding.PKCS7(128).unpadder()
        plaintext = unpadder.update(padded_plaintext) + unpadder.finalize()
        
        return plaintext
    
    # 模拟DTLS握手和数据传输
    def simulate_dtls_session():
    # 共享预共享密钥
    psk = os.urandom(32)
    client = DTLSConnection(psk, b"client1")
    server = DTLSConnection(psk, b"server1")
    
    # 模拟握手交换随机数
    client_random = os.urandom(32)
    server_random = os.urandom(32)
    
    client.derive_keys(client_random, server_random)
    server.derive_keys(client_random, server_random)
    
    # 客户端发送加密消息
    message = b"Sensor data: temperature=25.6C, humidity=60%"
    encrypted_msg = client.encrypt_message(message)
    
    # 服务器接收并解密
    try:
        decrypted = server.decrypt_message(encrypted_msg)
        print(f"Decrypted message: {decrypted.decode()}")
    except ValueError as e:
        print(f"Decryption failed: {str(e)}")
    
    simulate_dtls_session()

    import socket
    import random
    import time
    from threading import Thread
    from queue import Queue
    import zlib
    
    class TrafficObfuscator:
    def __init__(self, real_host, real_port, proxy_host, proxy_port):
        self.real_addr = (real_host, real_port)
        self.proxy_addr = (proxy_host, proxy_port)
        self.packet_queue = Queue()
        self.dummy_patterns = [
            b"\x00"*16, b"\xFF"*8 + b"\x00"*8,
            b"HEARTBEAT", b"PING"+bytes([random.randint(0,255)])
        ]
        
    def create_dummy_packet(self):
        # 生成伪装的虚拟数据包
        pattern = random.choice(self.dummy_patterns)
        if random.random() > 0.7:
            compressed = zlib.compress(pattern)
            return compressed + bytes([random.randint(0, 255)])
        return pattern
    
    def sender_thread(self):
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        while True:
            # 发送真实数据
            if not self.packet_queue.empty():
                real_data = self.packet_queue.get()
                sock.sendto(real_data, self.proxy_addr)
            
            # 随机发送虚拟数据
            if random.random() > 0.8:
                dummy = self.create_dummy_packet()
                sock.sendto(dummy, self.proxy_addr)
            
            time.sleep(random.uniform(0.1, 0.5))
    
    def receiver_thread(self):
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.bind(self.real_addr)
        
        while True:
            data, addr = sock.recvfrom(1024)
            try:
                # 尝试解压缩
                decompressed = zlib.decompress(data)
                if decompressed in self.dummy_patterns:
                    continue  # 丢弃虚拟数据包
                self.packet_queue.put(decompressed)
            except:
                self.packet_queue.put(data)
    
    def start(self):
        Thread(target=self.sender_thread, daemon=True).start()
        Thread(target=self.receiver_thread, daemon=True).start()
        print(f"Traffic obfuscator running between {self.real_addr} and {self.proxy_addr}")
    
    # 使用示例
    obfuscator = TrafficObfuscator(
    real_host="192.168.1.100", real_port=5000,
    proxy_host="10.0.0.1", proxy_port=6000
    )
    obfuscator.start()
    
    # 模拟发送真实数据
    while True:
    sensor_data = b"Temp: " + str(random.randint(20, 30)).encode() + b"C"
    obfuscator.packet_queue.put(sensor_data)
    time.sleep(2)

    import numpy as np
    from scipy import stats
    import matplotlib.pyplot as plt
    
    class DifferentialPrivacy:
    def __init__(self, epsilon=1.0, sensitivity=1.0):
        self.epsilon = epsilon
        self.sensitivity = sensitivity
        
    def laplace_noise(self, size=1):
        scale = self.sensitivity / self.epsilon
        return np.random.laplace(0, scale, size)
    
    def privatize_count(self, true_count):
        return true_count + self.laplace_noise()[0]
    
    def privatize_histogram(self, bins):
        noise = self.laplace_noise(size=len(bins))
        return np.round(bins + noise).astype(int)
    
    def privatize_average(self, data, true_mean=None):
        if true_mean is None:
            true_mean = np.mean(data)
        scale = self.sensitivity / (self.epsilon * len(data))
        return true_mean + np.random.laplace(0, scale)
    
    # 模拟智能电表数据收集
    def simulate_smart_meter():
    # 真实用电数据 (kWh)
    true_usage = np.array([2.5, 3.1, 2.8, 3.5, 4.2, 5.0, 6.1, 
                          7.3, 6.8, 5.5, 4.3, 3.2, 2.9, 3.1])
    
    dp = DifferentialPrivacy(epsilon=0.5, sensitivity=1.0)
    
    # 原始平均值
    true_avg = np.mean(true_usage)
    
    # 差分隐私处理后的平均值
    private_avg = dp.privatize_average(true_usage)
    
    # 生成直方图数据
    hist, bin_edges = np.histogram(true_usage, bins=5)
    private_hist = dp.privatize_histogram(hist)
    
    # 可视化结果
    plt.figure(figsize=(12, 5))
    
    plt.subplot(1, 2, 1)
    plt.bar(range(len(true_usage)), true_usage, alpha=0.7, label='真实数据')
    plt.axhline(true_avg, color='r', linestyle='--', label=f'真实平均: {true_avg:.2f}')
    plt.axhline(private_avg, color='g', linestyle=':', label=f'隐私平均: {private_avg:.2f}')
    plt.title("用电量时间序列")
    plt.legend()
    
    plt.subplot(1, 2, 2)
    plt.bar(bin_edges[:-1], hist, width=0.8, alpha=0.5, label='真实直方图')
    plt.bar(bin_edges[:-1], private_hist, width=0.4, alpha=0.5, label='隐私直方图')
    plt.title("用电量分布")
    plt.legend()
    
    plt.tight_layout()
    plt.show()
    
    simulate_smart_meter()

    from phe import paillier
    import numpy as np
    import time
    
    class HomomorphicProcessor:
    def __init__(self, key_size=2048):
        self.public_key, self.private_key = paillier.generate_paillier_keypair(n_length=key_size)
        
    def encrypt_array(self, data):
        return [self.public_key.encrypt(float(x)) for x in data]
    
    def decrypt_array(self, encrypted_data):
        return [self.private_key.decrypt(x) for x in encrypted_data]
    
    def secure_aggregate(self, encrypted_data_list):
        # 安全聚合多个客户端的加密数据
        if not encrypted_data_list:
            return None
            
        aggregated = encrypted_data_list[0].copy()
        for ed in encrypted_data_list[1:]:
            aggregated = [a + b for a, b in zip(aggregated, ed)]
        return aggregated
    
    def secure_average(self, encrypted_data_list):
        # 计算加密数据的平均值
        aggregated = self.secure_aggregate(encrypted_data_list)
        count = len(encrypted_data_list)
        return [x / count for x in aggregated]
    
    # 模拟医疗物联网数据聚合
    def simulate_medical_iot():
    # 模拟5家医院的敏感医疗数据
    hospital_data = [
        np.random.normal(37.0, 0.5, 10),  # 体温数据
        np.random.normal(36.8, 0.6, 10),
        np.random.normal(37.2, 0.4, 10),
        np.random.normal(36.9, 0.7, 10),
        np.random.normal(37.1, 0.3, 10)
    ]
    
    processor = HomomorphicProcessor()
    
    # 各医院加密数据
    encrypted_data = [processor.encrypt_array(data) for data in hospital_data]
    
    # 云端安全聚合
    start_time = time.time()
    encrypted_avg = processor.secure_average(encrypted_data)
    processing_time = time.time() - start_time
    
    # 解密结果
    final_avg = processor.decrypt_array(encrypted_avg)
    
    # 对比明文计算
    plain_avg = np.mean(hospital_data, axis=0)
    
    print(f"安全计算耗时: {processing_time:.4f}秒")
    print("明文平均值:", np.round(plain_avg, 2))
    print("加密计算平均值:", np.round(final_avg, 2))
    print("最大差异:", np.max(np.abs(np.array(plain_avg) - np.array(final_avg))))
    
    simulate_medical_iot()