120_物联网安全：从设备到云端的全方位防护策略与实践指南

# 固件提取与分析示例（Python）
import os
import subprocess
import tempfile
import shutil

# 固件提取函数
def extract_firmware(firmware_path, output_dir):
    """使用binwalk提取固件内容"""
    print(f"正在提取固件: {firmware_path}")
    
    # 确保输出目录存在
    os.makedirs(output_dir, exist_ok=True)
    
    # 使用binwalk提取固件
    try:
        subprocess.run(["binwalk", "-e", firmware_path, "-C", output_dir], check=True)
        print("固件提取成功")
        return True
    except subprocess.CalledProcessError as e:
        print(f"固件提取失败: {e}")
        return False

# 固件分析函数
def analyze_firmware(firmware_path):
    """对固件进行基本分析"""
    # 创建临时目录
    temp_dir = tempfile.mkdtemp()
    
    try:
        # 提取固件
        if extract_firmware(firmware_path, temp_dir):
            # 列出提取的文件
            print("\n提取的文件结构:")
            for root, dirs, files in os.walk(temp_dir):
                level = root.replace(temp_dir, '').count(os.sep)
                indent = ' ' * 4 * level
                print(f"{indent}{os.path.basename(root)}/")
                subindent = ' ' * 4 * (level + 1)
                for file in files:
                    file_path = os.path.join(root, file)
                    file_size = os.path.getsize(file_path)
                    print(f"{subindent}{file} ({file_size} bytes)")
                    
                    # 检查是否为可执行文件
                    if subprocess.run(["file", file_path], capture_output=True, text=True).stdout.find("executable") != -1:
                        print(f"{subindent}  [可执行文件]")
                    
                    # 检查是否包含字符串
                    strings_output = subprocess.run(["strings", file_path], capture_output=True, text=True).stdout
                    if strings_output:
                        # 寻找可能的敏感信息
                        interesting_patterns = ["password", "key", "secret", "admin", "root"]
                        for pattern in interesting_patterns:
                            if pattern in strings_output.lower():
                                print(f"{subindent}  [发现潜在敏感信息: {pattern}]")
                                break
    finally:
        # 清理临时目录
        shutil.rmtree(temp_dir)

# 主函数
def main():
    firmware_path = input("请输入固件文件路径: ")
    if os.path.exists(firmware_path):
        analyze_firmware(firmware_path)
    else:
        print("固件文件不存在")

if __name__ == "__main__":
    main()

# 嵌入式Linux系统安全加固脚本示例

# 1. 移除不必要的服务
systemctl disable telnet
rpm -e --nodeps telnet-server

# 2. 配置防火墙
iptables -F
iptables -P INPUT DROP
iptables -P FORWARD DROP
iptables -P OUTPUT ACCEPT
iptables -A INPUT -i lo -j ACCEPT
iptables -A INPUT -m state --state ESTABLISHED,RELATED -j ACCEPT
iptables -A INPUT -p tcp --dport 22 -j ACCEPT  # 仅允许SSH
iptables-save > /etc/iptables/rules.v4

# 3. 禁用不必要的账户
for user in $(cat /etc/passwd | grep bash | cut -d: -f1); do
    if [ "$user" != "root" ] && [ "$user" != "admin" ]; then
        usermod -s /sbin/nologin $user
    fi
done

# 4. 密码策略设置
echo "password    required    pam_cracklib.so minlen=12 difok=3 ucredit=-1 lcredit=-1 dcredit=-1 ocredit=-1" >> /etc/pam.d/common-password
echo "password    required    pam_unix.so sha512 shadow use_authtok" >> /etc/pam.d/common-password

# 5. 定期更新设置
echo "0 3 * * * root apt-get update && apt-get upgrade -y" > /etc/cron.d/security-updates

# MQTT客户端安全连接示例（Python）
import paho.mqtt.client as mqtt
import ssl
import time

# 连接参数
BROKER = "mqtt.example.com"
PORT = 8883
CLIENT_ID = "iot_device_secure"
USERNAME = "device1"
PASSWORD = "secure_password123"

# TLS/SSL配置
CA_CERTS = "/path/to/ca.crt"
CLIENT_CERT = "/path/to/client.crt"
CLIENT_KEY = "/path/to/client.key"

# 回调函数
def on_connect(client, userdata, flags, rc):
    print(f"已连接，返回码: {rc}")
    # 连接成功后订阅主题
    client.subscribe("devices/commands/#")

def on_message(client, userdata, msg):
    print(f"收到消息: {msg.topic} {msg.payload}")

def on_publish(client, userdata, mid):
    print(f"消息已发布，消息ID: {mid}")

# 创建MQTT客户端
client = mqtt.Client(client_id=CLIENT_ID, clean_session=True)

# 设置回调函数
client.on_connect = on_connect
client.on_message = on_message
client.on_publish = on_publish

# 设置用户名和密码
client.username_pw_set(username=USERNAME, password=PASSWORD)

# 配置TLS/SSL
client.tls_set(
    ca_certs=CA_CERTS,
    certfile=CLIENT_CERT,
    keyfile=CLIENT_KEY,
    cert_reqs=ssl.CERT_REQUIRED,
    tls_version=ssl.PROTOCOL_TLSv1_2,
    ciphers=None
)

# 启用证书验证
client.tls_insecure_set(False)

try:
    # 连接到MQTT代理
    print(f"正在连接到 {BROKER}:{PORT}...")
    client.connect(BROKER, PORT, keepalive=60)
    
    # 启动循环处理
    client.loop_start()
    
    # 定期发布消息
    while True:
        # 模拟传感器数据
        temperature = 25.5
        humidity = 60.0
        
        # 发布消息
        client.publish(
            topic="devices/sensors/temperature",
            payload=f"{{\"value\": {temperature}, \"timestamp\": {int(time.time())}}}",
            qos=1,  # 至少一次传递
            retain=True  # 保留最后一条消息
        )
        
        client.publish(
            topic="devices/sensors/humidity",
            payload=f"{{\"value\": {humidity}, \"timestamp\": {int(time.time())}}}",
            qos=1,
            retain=True
        )
        
        time.sleep(60)  # 每分钟发布一次
        
except KeyboardInterrupt:
    print("程序被用户中断")
except Exception as e:
    print(f"发生错误: {e}")
finally:
    # 断开连接
    client.loop_stop()
    client.disconnect()
    print("已断开连接")

# 物联网数据加密示例（Python）
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import padding
import os
import json
import base64
import hashlib
import hmac

# 生成加密密钥
def generate_key(password, salt=None):
    """使用PBKDF2从密码生成密钥"""
    if salt is None:
        salt = os.urandom(16)
    # 使用PBKDF2从密码派生密钥
    key = hashlib.pbkdf2_hmac(
        'sha256',
        password.encode(),
        salt,
        100000,  # 迭代次数
        dklen=32  # 密钥长度（256位）
    )
    return key, salt

# AES加密函数
def encrypt_aes(data, key):
    """使用AES-CBC模式加密数据"""
    # 生成随机IV
    iv = os.urandom(16)
    
    # 创建填充器
    padder = padding.PKCS7(algorithms.AES.block_size).padder()
    # 对数据进行填充
    padded_data = padder.update(data) + padder.finalize()
    
    # 创建加密器
    cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
    encryptor = cipher.encryptor()
    
    # 加密数据
    encrypted_data = encryptor.update(padded_data) + encryptor.finalize()
    
    # 返回IV和加密数据的组合
    return iv + encrypted_data

# AES解密函数
def decrypt_aes(encrypted_data, key):
    """使用AES-CBC模式解密数据"""
    # 提取IV
    iv = encrypted_data[:16]
    # 提取加密的数据
    ciphertext = encrypted_data[16:]
    
    # 创建解密器
    cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
    decryptor = cipher.decryptor()
    
    # 解密数据
    padded_data = decryptor.update(ciphertext) + decryptor.finalize()
    
    # 创建去填充器
    unpadder = padding.PKCS7(algorithms.AES.block_size).unpadder()
    # 去除填充
    data = unpadder.update(padded_data) + unpadder.finalize()
    
    return data

# 生成消息认证码
def generate_hmac(data, key):
    """使用HMAC-SHA256生成消息认证码"""
    h = hmac.new(key, data, hashlib.sha256)
    return h.digest()

# 验证消息认证码
def verify_hmac(data, mac, key):
    """验证消息认证码"""
    calculated_mac = generate_hmac(data, key)
    # 使用时间恒定比较避免时序攻击
    return hmac.compare_digest(mac, calculated_mac)

# 安全数据传输封装
def secure_data_transfer(sensor_data, password):
    """安全封装传感器数据用于传输"""
    # 将传感器数据转换为JSON字符串
    json_data = json.dumps(sensor_data).encode()
    
    # 生成加密密钥
    encryption_key, salt = generate_key(password)
    
    # 加密数据
    encrypted_data = encrypt_aes(json_data, encryption_key)
    
    # 生成HMAC用于完整性验证
    hmac_key, _ = generate_key(password + "hmac", salt)
    mac = generate_hmac(encrypted_data, hmac_key)
    
    # 组合所有数据：salt + encrypted_data + mac
    combined_data = salt + encrypted_data + mac
    
    # 使用base64编码以便于传输
    encoded_data = base64.b64encode(combined_data)
    
    return encoded_data

# 安全数据接收解析
def secure_data_receive(encoded_data, password):
    """接收并解密安全传输的数据"""
    # 解码base64数据
    combined_data = base64.b64decode(encoded_data)
    
    # 提取salt、encrypted_data和mac
    salt = combined_data[:16]
    encrypted_data = combined_data[16:-32]  # HMAC长度为32字节
    mac = combined_data[-32:]
    
    # 生成HMAC密钥并验证消息完整性
    hmac_key, _ = generate_key(password + "hmac", salt)
    if not verify_hmac(encrypted_data, mac, hmac_key):
        raise ValueError("消息完整性验证失败，数据可能被篡改")
    
    # 生成解密密钥并解密数据
    decryption_key, _ = generate_key(password, salt)
    decrypted_data = decrypt_aes(encrypted_data, decryption_key)
    
    # 解析JSON数据
    sensor_data = json.loads(decrypted_data.decode())
    
    return sensor_data

# 物联网数据匿名化示例（Python）
import json
import random
import hashlib
from datetime import datetime, timedelta

# 数据匿名化函数
def anonymize_data(sensor_data):
    """对物联网传感器数据进行匿名化处理"""
    # 创建数据副本以避免修改原始数据
    anonymized = sensor_data.copy()
    
    # 1. 设备ID哈希处理
    if "device_id" in anonymized:
        # 使用加盐哈希替换设备ID
        salt = "device_salt_2023"  # 在实际应用中应该使用随机盐
        hashed_id = hashlib.sha256((anonymized["device_id"] + salt).encode()).hexdigest()
        anonymized["device_id"] = f"device_{hashed_id[:8]}"
    
    # 2. 时间模糊化
    if "timestamp" in anonymized:
        # 将时间戳调整到最近的小时
        dt = datetime.fromtimestamp(anonymized["timestamp"])
        # 保持小时精度，模糊分钟和秒
        dt_rounded = dt.replace(minute=0, second=0, microsecond=0)
        # 添加随机的小时偏移（-1到+1小时）
        hour_offset = random.randint(-1, 1)
        dt_rounded = dt_rounded + timedelta(hours=hour_offset)
        anonymized["timestamp"] = int(dt_rounded.timestamp())
    
    # 3. 位置模糊化
    if "location" in anonymized:
        location = anonymized["location"]
        if "latitude" in location and "longitude" in location:
            # 对位置进行模糊处理，精度降低到约100米
            # 纬度约0.001度 ≈ 111米
            # 经度在赤道约0.001度 ≈ 111米，但随纬度增加而减小
            lat = location["latitude"]
            lon = location["longitude"]
            
            # 保留4位小数（约11米精度），然后添加随机偏移（-0.005到+0.005度）
            lat_base = round(lat, 4)
            lon_base = round(lon, 4)
            
            # 添加随机偏移（最大约500米）
            lat_offset = random.uniform(-0.005, 0.005)
            lon_offset = random.uniform(-0.005, 0.005)
            
            anonymized["location"] = {
                "latitude": round(lat_base + lat_offset, 6),
                "longitude": round(lon_base + lon_offset, 6),
                "accuracy": "approximate"  # 标记位置为近似值
            }
    
    # 4. 敏感数据删除
    # 删除可能的敏感字段
    sensitive_fields = ["serial_number", "user_id", "ip_address", "mac_address"]
    for field in sensitive_fields:
        if field in anonymized:
            del anonymized[field]
    
    # 5. 数值范围化
    # 对于某些数值型数据，可以将其转换为范围
    if "temperature" in anonymized:
        temp = anonymized["temperature"]
        # 将温度转换为范围，保留整数部分
        anonymized["temperature"] = {
            "range": f"{int(temp)}-{int(temp)+1}",
            "unit": "celsius"
        }
    
    return anonymized

# 物联网设备证书认证示例（Python）
import ssl
import socket
import json
import time
import os
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend

# 证书生成函数
def generate_certificate_pair(device_id, ca_key_path, ca_cert_path, output_dir="./"):
    """为物联网设备生成证书和私钥，并使用CA签名"""
    # 确保输出目录存在
    os.makedirs(output_dir, exist_ok=True)
    
    # 加载CA私钥
    with open(ca_key_path, "rb") as key_file:
        ca_key = serialization.load_pem_private_key(
            key_file.read(),
            password=None,
            backend=default_backend()
        )
    
    # 加载CA证书
    with open(ca_cert_path, "rb") as cert_file:
        ca_cert = x509.load_pem_x509_certificate(
            cert_file.read(),
            backend=default_backend()
        )
    
    # 生成设备私钥
    device_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=2048,
        backend=default_backend()
    )
    
    # 构建证书请求
    csr = x509.CertificateSigningRequestBuilder()
    csr = csr.subject_name(x509.Name([
        x509.NameAttribute(NameOID.COUNTRY_NAME, u"CN"),
        x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, u"Beijing"),
        x509.NameAttribute(NameOID.LOCALITY_NAME, u"Beijing"),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, u"IoT Security"),
        x509.NameAttribute(NameOID.COMMON_NAME, device_id),
    ]))
    csr = csr.add_extension(
        x509.SubjectAlternativeName([x509.DNSName(device_id)]),
        critical=False,
    )
    # 签署证书请求
    csr = csr.sign(device_key, hashes.SHA256(), backend=default_backend())
    
    # 使用CA签发证书
    certificate = x509.CertificateBuilder()
    certificate = certificate.subject_name(csr.subject)
    certificate = certificate.issuer_name(ca_cert.subject)
    certificate = certificate.public_key(csr.public_key())
    certificate = certificate.serial_number(x509.random_serial_number())
    certificate = certificate.not_valid_before(time.time())
    certificate = certificate.not_valid_after(
        # 证书有效期为1年
        time.time() + (365 * 24 * 60 * 60)
    )
    for extension in csr.extensions:
        certificate = certificate.add_extension(extension.value, extension.critical)
    
    # 使用CA私钥签名证书
    certificate = certificate.sign(ca_key, hashes.SHA256(), backend=default_backend())
    
    # 将设备私钥保存到文件
    device_key_path = os.path.join(output_dir, f"{device_id}_key.pem")
    with open(device_key_path, "wb") as key_file:
        key_file.write(
            device_key.private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.TraditionalOpenSSL,
                encryption_algorithm=serialization.NoEncryption()
            )
        )
    
    # 将设备证书保存到文件
    device_cert_path = os.path.join(output_dir, f"{device_id}_cert.pem")
    with open(device_cert_path, "wb") as cert_file:
        cert_file.write(certificate.public_bytes(serialization.Encoding.PEM))
    
    return device_key_path, device_cert_path

# 安全服务器实现
class SecureIoTServer:
    def __init__(self, host, port, ca_cert_path):
        self.host = host
        self.port = port
        self.ca_cert_path = ca_cert_path
        self.connected_devices = {}
    
    def handle_client(self, client_socket, client_address):
        """处理客户端连接"""
        try:
            # 获取客户端证书
            cert = client_socket.getpeercert()
            if cert:
                # 解析证书信息
                subject = dict(x[0] for x in cert['subject'])
                common_name = subject.get('commonName', 'Unknown')
                print(f"设备 {common_name} 已连接，地址: {client_address}")
                
                # 记录已连接设备
                self.connected_devices[common_name] = {
                    "address": client_address,
                    "connected_at": time.time()
                }
                
                # 向设备发送欢迎消息
                welcome_msg = json.dumps({
                    "status": "connected",
                    "message": f"欢迎，设备 {common_name}！",
                    "timestamp": int(time.time())
                })
                client_socket.send(welcome_msg.encode())
                
                # 接收并处理数据
                while True:
                    data = client_socket.recv(4096)
                    if not data:
                        break
                    
                    try:
                        # 解析接收到的JSON数据
                        payload = json.loads(data.decode())
                        print(f"从设备 {common_name} 接收数据: {payload}")
                        
                        # 处理数据...
                        # 在实际应用中，这里应该根据设备类型和数据内容进行相应的处理
                        
                        # 发送确认消息
                        response = json.dumps({
                            "status": "success",
                            "message": "数据已接收并处理",
                            "timestamp": int(time.time())
                        })
                        client_socket.send(response.encode())
                        
                    except json.JSONDecodeError:
                        print(f"从设备 {common_name} 接收无效的JSON数据")
                        error_response = json.dumps({
                            "status": "error",
                            "message": "无效的JSON数据",
                            "timestamp": int(time.time())
                        })
                        client_socket.send(error_response.encode())
                
                # 设备断开连接
                print(f"设备 {common_name} 已断开连接")
                if common_name in self.connected_devices:
                    del self.connected_devices[common_name]
                
            else:
                print(f"未收到客户端证书，拒绝连接: {client_address}")
                client_socket.close()
                
        except ssl.SSLError as e:
            print(f"SSL错误: {e}")
        except Exception as e:
            print(f"处理客户端连接时发生错误: {e}")
        finally:
            client_socket.close()

# 固件分析CTF挑战解题脚本
import os
import subprocess
import tempfile
import shutil
import re

def solve_firmware_challenge(firmware_path):
    """解决固件分析CTF挑战"""
    # 创建临时目录
    temp_dir = tempfile.mkdtemp()
    
    try:
        print(f"正在分析固件: {firmware_path}")
        
        # 第一步：使用binwalk提取固件
        print("\n[+] 步骤1: 提取固件内容")
        subprocess.run(["binwalk", "-e", firmware_path, "-C", temp_dir], check=True)
        print("固件提取成功")
        
        # 第二步：分析文件系统结构
        print("\n[+] 步骤2: 分析文件系统结构")
        root_dirs = []
        for item in os.listdir(temp_dir):
            item_path = os.path.join(temp_dir, item)
            if os.path.isdir(item_path):
                # 检查是否为根文件系统
                if os.path.exists(os.path.join(item_path, "bin")) or \
                   os.path.exists(os.path.join(item_path, "etc")) or \
                   os.path.exists(os.path.join(item_path, "usr")):
                    root_dirs.append(item_path)
        
        if not root_dirs:
            print("未找到明显的根文件系统，尝试列出所有目录")
            for root, dirs, files in os.walk(temp_dir):
                level = root.replace(temp_dir, '').count(os.sep)
                indent = ' ' * 4 * level
                print(f"{indent}{os.path.basename(root)}/")
        else:
            print(f"找到 {len(root_dirs)} 个可能的根文件系统")
            for root_dir in root_dirs:
                print(f"- {os.path.basename(root_dir)}")
        
        # 第三步：寻找flag
        print("\n[+] 步骤3: 寻找flag")
        
        # 3.1 在文本文件中搜索flag
        print("\n在文本文件中搜索flag...")
        flag_pattern = re.compile(r'flag{[^}]+}|FLAG{[^}]+}|ctf{[^}]+}|CTF{[^}]+}')
        found_flags = []
        
        for root, dirs, files in os.walk(temp_dir):
            for file in files:
                file_path = os.path.join(root, file)
                
                # 检查是否为文本文件
                try:
                    file_type = subprocess.run(["file", file_path], capture_output=True, text=True).stdout
                    if "text" in file_type:
                        # 读取文件内容并搜索flag
                        with open(file_path, 'r', errors='ignore') as f:
                            content = f.read()
                            matches = flag_pattern.findall(content)
                            for match in matches:
                                print(f"在文件 {file_path} 中找到: {match}")
                                found_flags.append((file_path, match))
                except Exception as e:
                    continue
        
        # 3.2 在二进制文件中搜索字符串
        print("\n在二进制文件中搜索flag相关字符串...")
        for root, dirs, files in os.walk(temp_dir):
            for file in files:
                file_path = os.path.join(root, file)
                
                # 检查是否为二进制文件
                try:
                    file_type = subprocess.run(["file", file_path], capture_output=True, text=True).stdout
                    if "executable" in file_type or "ELF" in file_type:
                        # 使用strings命令提取字符串
                        strings_output = subprocess.run(["strings", file_path], capture_output=True, text=True).stdout
                        matches = flag_pattern.findall(strings_output)
                        for match in matches:
                            print(f"在二进制文件 {file_path} 中找到: {match}")
                            found_flags.append((file_path, match))
                except Exception as e:
                    continue
        
        # 3.3 检查配置文件
        print("\n检查配置文件...")
        config_extensions = [".conf", ".cfg", ".config", ".ini", ".xml", ".json", ".yaml", ".yml"]
        for root, dirs, files in os.walk(temp_dir):
            for file in files:
                if any(file.endswith(ext) for ext in config_extensions):
                    file_path = os.path.join(root, file)
                    print(f"检查配置文件: {file_path}")
                    try:
                        with open(file_path, 'r', errors='ignore') as f:
                            content = f.read()
                            matches = flag_pattern.findall(content)
                            for match in matches:
                                print(f"在配置文件 {file_path} 中找到: {match}")
                                found_flags.append((file_path, match))
                    except Exception as e:
                        continue
        
        # 总结找到的flag
        if found_flags:
            print("\n[+] 找到的flag总结:")
            for i, (file_path, flag) in enumerate(found_flags, 1):
                print(f"{i}. {flag} (来自: {file_path})")
        else:
            print("\n[-] 未找到明显的flag，可能需要更深入的分析")
        
    except Exception as e:
        print(f"分析过程中发生错误: {e}")
    finally:
        # 清理临时目录
        shutil.rmtree(temp_dir)

# 主函数
def main():
    firmware_path = input("请输入固件文件路径: ")
    if os.path.exists(firmware_path):
        solve_firmware_challenge(firmware_path)
    else:
        print("固件文件不存在")

if __name__ == "__main__":
    main()