163_边缘计算与物联网安全：从设备安全到边缘防护的实战指南 [特殊字符]

table
  设备层威胁 | 网络层威胁 | 应用层威胁
  • 设备篡改
• 固件劫持
• 物理访问
• 侧信道攻击 | • 中间人攻击
• DDoS攻击
• 协议漏洞利用
• 网络嗅探 | • 恶意代码执行
• 数据泄露
• 权限提升
• API滥用

# 物联网设备固件分析工具示例
import os
import re
import subprocess
import hashlib
import shutil
import tempfile
from datetime import datetime

def extract_firmware(firmware_file, output_dir):
    """提取固件文件内容"""
    print(f"开始提取固件: {firmware_file}")
    
    # 创建输出目录
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)
    
    # 尝试不同的提取方法
    extract_methods = [
        ("binwalk自动提取", f"binwalk -e --directory={output_dir} {firmware_file}"),
        ("固件解压", f"unzip -o {firmware_file} -d {output_dir}"),
        ("7z解压", f"7z x {firmware_file} -o{output_dir}")
    ]
    
    success = False
    for name, cmd in extract_methods:
        print(f"尝试 {name}...")
        try:
            result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
            if result.returncode == 0 and os.listdir(output_dir):
                print(f"{name}成功")
                success = True
                break
        except Exception as e:
            print(f"{name}失败: {e}")
    
    if not success:
        print("所有提取方法均失败")
    
    return success

def analyze_firmware_security(output_dir):
    """分析固件安全问题"""
    vulnerabilities = {
        "default_credentials": [],
        "hardcoded_secrets": [],
        "dangerous_commands": [],
        "insecure_permissions": [],
        "weak_algorithms": []
    }
    
    # 1. 搜索默认凭据
    print("\n1. 搜索默认凭据...")
    cred_patterns = [
        r'(password|passwd|pwd|secret|key|token)\s*[=:]\s*[\'\"]([^\'\"]+)[\'"]',
        r'admin:[\'\"]?(password|admin|123456|default)[\'\"]?',
        r'default\s+(password|username)'
    ]
    
    for root, _, files in os.walk(output_dir):
        for file in files:
            file_path = os.path.join(root, file)
            if os.path.isfile(file_path) and os.path.getsize(file_path) < 10 * 1024 * 1024:  # 限制文件大小
                try:
                    with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
                        content = f.read()
                        for pattern in cred_patterns:
                            matches = re.findall(pattern, content, re.IGNORECASE)
                            for match in matches:
                                vulnerabilities["default_credentials"].append({
                                    "file": os.path.relpath(file_path, output_dir),
                                    "match": str(match)
                                })
                except Exception as e:
                    pass
    
    # 2. 检查危险命令和权限
    print("\n2. 检查危险命令和权限...")
    for root, _, files in os.walk(output_dir):
        for file in files:
            file_path = os.path.join(root, file)
            if os.path.isfile(file_path):
                # 检查setuid/setgid文件
                try:
                    stats = os.stat(file_path)
                    if (stats.st_mode & 0o4000) or (stats.st_mode & 0o2000):
                        vulnerabilities["insecure_permissions"].append({
                            "file": os.path.relpath(file_path, output_dir),
                            "permission": oct(stats.st_mode)[-3:]
                        })
                except:
                    pass
                
                # 检查脚本中的危险命令
                if file.endswith(('.sh', '.py', '.cgi', '.php')):
                    try:
                        with open(file_path, 'r', errors='ignore') as f:
                            content = f.read()
                            dangerous_patterns = [
                                r'system\s*\(',
                                r'exec\s*\(',
                                r'eval\s*\(',
                                r'popen\s*\(',
                                r'os\.system',
                                r'subprocess\.call',
                                r'rm\s+-rf',
                                r'chmod\s+777'
                            ]
                            for pattern in dangerous_patterns:
                                if re.search(pattern, content):
                                    vulnerabilities["dangerous_commands"].append({
                                        "file": os.path.relpath(file_path, output_dir),
                                        "pattern": pattern
                                    })
                    except:
                        pass
    
    return vulnerabilities

def generate_security_report(vulnerabilities, output_file):
    """生成安全分析报告"""
    print(f"\n生成安全分析报告: {output_file}")
    
    with open(output_file, 'w', encoding='utf-8') as f:
        f.write("=" * 80 + "\n")
        f.write("物联网设备固件安全分析报告\n")
        f.write("=" * 80 + "\n")
        f.write(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
        
        # 汇总统计
        total_vulns = sum(len(v) for v in vulnerabilities.values())
        f.write(f"发现的安全问题总数: {total_vulns}\n\n")
        
        # 详细漏洞信息
        if vulnerabilities["default_credentials"]:
            f.write("## 1. 默认凭据和硬编码密钥\n")
            for vuln in vulnerabilities["default_credentials"]:
                f.write(f"- 文件: {vuln['file']}\n  匹配内容: {vuln['match']}\n")
            f.write("\n")
        
        if vulnerabilities["dangerous_commands"]:
            f.write("## 2. 危险命令和函数\n")
            for vuln in vulnerabilities["dangerous_commands"]:
                f.write(f"- 文件: {vuln['file']}\n  危险模式: {vuln['pattern']}\n")
            f.write("\n")
        
        if vulnerabilities["insecure_permissions"]:
            f.write("## 3. 不安全的文件权限\n")
            for vuln in vulnerabilities["insecure_permissions"]:
                f.write(f"- 文件: {vuln['file']}\n  权限: {vuln['permission']}\n")
            f.write("\n")
        
        # 安全建议
        f.write("## 安全建议\n")
        f.write("1. 移除所有硬编码的凭据和密钥\n")
        f.write("2. 实施安全的密码策略和密钥管理\n")
        f.write("3. 避免使用危险的系统命令函数\n")
        f.write("4. 审查并修复不安全的文件权限\n")
        f.write("5. 确保固件更新机制安全可靠\n")
        f.write("6. 实施固件签名和验证机制\n")
    
    print(f"报告已保存至: {output_file}")

def calculate_firmware_hash(firmware_file):
    """计算固件文件的哈希值"""
    hashes = {"MD5": hashlib.md5(), "SHA1": hashlib.sha1(), "SHA256": hashlib.sha256()}
    
    try:
        with open(firmware_file, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                for h in hashes.values():
                    h.update(chunk)
        
        result = {name: h.hexdigest() for name, h in hashes.items()}
        print(f"\n固件哈希值:")
        for name, hash_value in result.items():
            print(f"{name}: {hash_value}")
        
        return result
    except Exception as e:
        print(f"计算哈希值失败: {e}")
        return None

# 使用示例
if __name__ == "__main__":
    # 示例固件文件路径
    # firmware_file = "./sample_firmware.bin"
    # temp_dir = tempfile.mkdtemp()
    # output_dir = os.path.join(temp_dir, "extracted")
    # report_file = "./firmware_security_report.txt"
    
    print("物联网设备固件安全分析工具")
    print("==========================\n")
    
    print("注意: 此工具仅用于授权的安全测试和研究")
    print("请确保您有合法授权后再使用此工具分析固件\n")

# 边缘网络入侵检测系统示例
import socket
import struct
import threading
import time
from collections import defaultdict, deque
import re

def mac_addr(mac_raw):
    """格式化MAC地址"""
    return ':'.join(['%02x' % (b,) for b in mac_raw])

def ip_addr(ip_raw):
    """格式化IP地址"""
    return '.'.join([str(b) for b in ip_raw])

def tcp_flags(flags):
    """解析TCP标志"""
    flag_str = []
    if flags & 0x01: flag_str.append('FIN')
    if flags & 0x02: flag_str.append('SYN')
    if flags & 0x04: flag_str.append('RST')
    if flags & 0x08: flag_str.append('PSH')
    if flags & 0x10: flag_str.append('ACK')
    if flags & 0x20: flag_str.append('URG')
    if flags & 0x40: flag_str.append('ECE')
    if flags & 0x80: flag_str.append('CWR')
    return '|'.join(flag_str)

class EdgeNetworkIDS:
    def __init__(self, interface='eth0', capture_filter='', alert_threshold=5):
        self.interface = interface
        self.capture_filter = capture_filter
        self.alert_threshold = alert_threshold
        self.is_running = False
        self.packets_count = defaultdict(int)
        self.syn_count = defaultdict(int)
        self.ip_connections = defaultdict(set)
        self.detected_attacks = []
        self.packet_history = deque(maxlen=1000)
        self.lock = threading.Lock()
    
    def start(self):
        """启动IDS监控"""
        print(f"启动边缘网络入侵检测系统...")
        print(f"监控接口: {self.interface}")
        
        self.is_running = True
        self.sniffer_thread = threading.Thread(target=self._sniffer)
        self.analyzer_thread = threading.Thread(target=self._analyzer)
        
        self.sniffer_thread.daemon = True
        self.analyzer_thread.daemon = True
        
        self.sniffer_thread.start()
        self.analyzer_thread.start()
        
        print("IDS系统已启动")
    
    def stop(self):
        """停止IDS监控"""
        print("停止边缘网络入侵检测系统...")
        self.is_running = False
        if hasattr(self, 'sniffer_thread'):
            self.sniffer_thread.join(1.0)
        if hasattr(self, 'analyzer_thread'):
            self.analyzer_thread.join(1.0)
        print("IDS系统已停止")
    
    def _sniffer(self):
        """数据包捕获线程"""
        try:
            # 创建原始套接字
            conn = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.ntohs(3))
            conn.bind((self.interface, 0))
            
            print(f"开始捕获网络数据包...")
            
            while self.is_running:
                raw_data, addr = conn.recvfrom(65536)
                self._process_packet(raw_data, addr)
        
        except Exception as e:
            print(f"数据包捕获错误: {e}")
    
    def _process_packet(self, raw_data, addr):
        """处理捕获的数据包"""
        # 记录数据包历史
        with self.lock:
            self.packet_history.append((time.time(), raw_data, addr))
        
        # 解析以太网帧
        eth_length = 14
        eth_header = raw_data[:eth_length]
        eth = struct.unpack('!6s6sH', eth_header)
        dest_mac = mac_addr(eth[0])
        src_mac = mac_addr(eth[1])
        eth_proto = socket.ntohs(eth[2])
        
        # 更新统计
        with self.lock:
            self.packets_count[src_mac] += 1
        
        # 只处理IP数据包
        if eth_proto == 8:
            self._process_ip_packet(raw_data[eth_length:], src_mac)
    
    def _process_ip_packet(self, ip_data, src_mac):
        """处理IP数据包"""
        ip_header = struct.unpack('!BBHHHBBH4s4s', ip_data[:20])
        ttl = ip_header[5]
        proto = ip_header[6]
        src_ip = ip_addr(ip_header[8])
        dest_ip = ip_addr(ip_header[9])
        
        # 更新IP连接统计
        with self.lock:
            self.ip_connections[src_ip].add(dest_ip)
        
        # 处理TCP数据包
        if proto == 6:
            tcp_header = struct.unpack('!HHLLBBHHH', ip_data[20:40])
            src_port = tcp_header[0]
            dest_port = tcp_header[1]
            flags = tcp_header[5]
            
            # 检测SYN扫描
            if (flags & 0x02) and not (flags & 0x10):
                with self.lock:
                    self.syn_count[src_ip] += 1
    
    def _analyzer(self):
        """威胁分析线程"""
        while self.is_running:
            self._detect_attacks()
            time.sleep(5)  # 每5秒分析一次
    
    def _detect_attacks(self):
        """检测常见攻击模式"""
        current_time = time.time()
        
        with self.lock:
            # 1. 检测SYN扫描
            for ip, count in self.syn_count.items():
                if count > self.alert_threshold:
                    attack = {
                        "time": current_time,
                        "type": "SYN_SCAN",
                        "source": ip,
                        "details": f"检测到SYN扫描: {count}个SYN包"
                    }
                    if attack not in self.detected_attacks:
                        self.detected_attacks.append(attack)
                        print(f"⚠️  告警: {attack['details']}")
            
            # 2. 检测异常连接数
            for ip, connections in self.ip_connections.items():
                if len(connections) > 50:  # 阈值可调整
                    attack = {
                        "time": current_time,
                        "type": "EXCESSIVE_CONNECTIONS",
                        "source": ip,
                        "details": f"检测到过多连接: {len(connections)}个目标"
                    }
                    if attack not in self.detected_attacks:
                        self.detected_attacks.append(attack)
                        print(f"⚠️  告警: {attack['details']}")
            
            # 3. 检测流量异常
            for mac, count in self.packets_count.items():
                if count > 1000:  # 阈值可调整
                    attack = {
                        "time": current_time,
                        "type": "EXCESSIVE_TRAFFIC",
                        "source": mac,
                        "details": f"检测到异常流量: {count}个数据包"
                    }
                    if attack not in self.detected_attacks:
                        self.detected_attacks.append(attack)
                        print(f"⚠️  告警: {attack['details']}")
            
            # 定期重置计数
            if len(self.detected_attacks) > 100:
                self.detected_attacks = self.detected_attacks[-50:]
            
            # 每60秒重置部分统计
            if hasattr(self, '_last_reset'):
                if current_time - self._last_reset > 60:
                    self.syn_count.clear()
                    self._last_reset = current_time
            else:
                self._last_reset = current_time
    
    def get_report(self):
        """生成安全报告"""
        with self.lock:
            report = {
                "total_packets": sum(self.packets_count.values()),
                "unique_devices": len(self.packets_count),
                "detected_attacks": len(self.detected_attacks),
                "attack_details": self.detected_attacks.copy()
            }
        return report

# 使用示例
if __name__ == "__main__":
    ids = EdgeNetworkIDS(interface='eth0', alert_threshold=10)
    try:
        ids.start()
        print("按Ctrl+C停止监控...")
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        ids.stop()
        report = ids.get_report()
        print(f"\n监控报告:")
        print(f"总数据包数: {report['total_packets']}")
        print(f"唯一设备数: {report['unique_devices']}")
        print(f"检测到的攻击: {report['detected_attacks']}")

边缘计算平台安全架构
┌────────────────────────────────────────────────────────┐
│                    边缘计算平台                         │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐     │
│  │  容器运行时  │  │  编排系统   │  │  安全监控   │     │
│  │  (Docker)   │  │  (K3s)      │  │  (Falco)    │     │
│  └─────────────┘  └─────────────┘  └─────────────┘     │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐     │
│  │  资源隔离   │  │  身份认证   │  │  访问控制   │     │
│  │  (Cgroups)  │  │  (OAuth)    │  │  (RBAC)     │     │
│  └─────────────┘  └─────────────┘  └─────────────┘     │
└────────────────────────────────────────────────────────┘

# 边缘计算容器安全扫描工具示例
import os
import sys
import json
import subprocess
import re
import requests
from datetime import datetime

def check_docker_installed():
    """检查Docker是否安装"""
    try:
        subprocess.run(["docker", "--version"], check=True, capture_output=True)
        return True
    except (subprocess.SubprocessError, FileNotFoundError):
        print("错误: Docker未安装或不可用")
        return False

def list_images():
    """列出系统中的Docker镜像"""
    try:
        result = subprocess.run(["docker", "images", "--format", "{{.Repository}}:{{.Tag}}"], 
                              capture_output=True, text=True, check=True)
        images = [img for img in result.stdout.strip().split('\n') if img]
        return images
    except subprocess.SubprocessError as e:
        print(f"列出镜像失败: {e}")
        return []

def scan_image(image_name):
    """扫描单个镜像的安全问题"""
    vulnerabilities = []
    
    print(f"\n开始扫描镜像: {image_name}")
    
    # 1. 使用Docker内置命令检查镜像历史
    try:
        print("\n1. 检查镜像历史...")
        result = subprocess.run(["docker", "history", "--no-trunc", image_name], 
                              capture_output=True, text=True, check=True)
        
        # 检查可疑的命令
        suspicious_commands = [
            r'ADD\s+http',
            r'curl\s+.*\|\s*sh',
            r'wget\s+.*\|\s*sh',
            r'chmod\s+777',
            r'rm\s+-rf\s+/tmp',
            r'passwd\s+-d'
        ]
        
        for line in result.stdout.split('\n'):
            for pattern in suspicious_commands:
                if re.search(pattern, line, re.IGNORECASE):
                    vulnerabilities.append({
                        "type": "SUSPICIOUS_COMMAND",
                        "severity": "HIGH",
                        "description": f"发现可疑命令: {line.strip()}"
                    })
    except subprocess.SubprocessError as e:
        print(f"检查镜像历史失败: {e}")
    
    # 2. 检查镜像中是否包含敏感文件
    try:
        print("\n2. 检查敏感文件...")
        sensitive_files = [
            '/etc/shadow', '/etc/passwd', '/root/.ssh/id_rsa',
            '/home/*/.ssh/id_rsa', '/var/www/html/.env',
            '/app/.env', '.git/config'
        ]
        
        for sensitive_file in sensitive_files:
            cmd = f"docker run --rm {image_name} find / -name '{sensitive_file.split('/')[-1]}' 2>/dev/null || echo''"
            result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
            if result.stdout.strip():
                vulnerabilities.append({
                    "type": "SENSITIVE_FILE",
                    "severity": "CRITICAL" if sensitive_file.endswith('id_rsa') else "HIGH",
                    "description": f"发现敏感文件: {sensitive_file}"
                })
    except subprocess.SubprocessError as e:
        print(f"检查敏感文件失败: {e}")
    
    # 3. 检查SUID/SGID文件
    try:
        print("\n3. 检查SUID/SGID文件...")
        cmd = f"docker run --rm {image_name} find / -type f \( -perm -4000 -o -perm -2000 \) 2>/dev/null || echo''"
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
        
        suid_files = result.stdout.strip().split('\n')
        for suid_file in suid_files:
            if suid_file and not any(x in suid_file for x in ['/bin/', '/usr/bin/']):
                vulnerabilities.append({
                    "type": "SUID_FILE",
                    "severity": "HIGH",
                    "description": f"发现非标准SUID/SGID文件: {suid_file}"
                })
    except subprocess.SubprocessError as e:
        print(f"检查SUID/SGID文件失败: {e}")
    
    # 4. 检查漏洞（模拟，实际可集成Trivy、Clair等工具）
    print("\n4. 模拟漏洞扫描...")
    # 这里可以集成实际的漏洞扫描工具API
    
    return vulnerabilities

def generate_scan_report(scan_results, output_file):
    """生成扫描报告"""
    print(f"\n生成安全扫描报告: {output_file}")
    
    total_images = len(scan_results)
    total_vulnerabilities = sum(len(vulns) for vulns in scan_results.values())
    
    severity_count = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0}
    
    for vulns in scan_results.values():
        for vuln in vulns:
            severity = vuln.get("severity", "LOW").upper()
            if severity in severity_count:
                severity_count[severity] += 1
    
    with open(output_file, 'w', encoding='utf-8') as f:
        f.write("=" * 80 + "\n")
        f.write("边缘计算容器安全扫描报告\n")
        f.write("=" * 80 + "\n")
        f.write(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
        
        # 汇总统计
        f.write("## 扫描结果汇总\n")
        f.write(f"扫描镜像总数: {total_images}\n")
        f.write(f"发现的漏洞总数: {total_vulnerabilities}\n")
        f.write("漏洞严重级别分布:\n")
        for severity, count in severity_count.items():
            f.write(f"  - {severity}: {count}\n")
        f.write("\n")
        
        # 详细漏洞信息
        f.write("## 详细漏洞信息\n")
        for image, vulns in scan_results.items():
            if vulns:
                f.write(f"\n### 镜像: {image}\n")
                f.write(f"发现漏洞数量: {len(vulns)}\n")
                for i, vuln in enumerate(vulns, 1):
                    f.write(f"\n{i}. **{vuln.get('type', 'Unknown')}** ({vuln.get('severity', 'LOW')})\n")
                    f.write(f"   描述: {vuln.get('description', 'No description')}\n")
        
        # 安全建议
        f.write("\n## 安全建议\n")
        f.write("1. 使用最小化基础镜像\n")
        f.write("2. 定期更新镜像中的软件包\n")
        f.write("3. 避免在镜像中包含敏感信息\n")
        f.write("4. 使用多阶段构建减小镜像体积\n")
        f.write("5. 实施容器运行时安全策略\n")
        f.write("6. 使用只读文件系统运行容器\n")
        f.write("7. 限制容器的网络和系统调用权限\n")
    
    print(f"报告已保存至: {output_file}")

def docker_security_hardening_check():
    """检查Docker守护进程安全配置"""
    print("\n检查Docker守护进程安全配置...")
    
    issues = []
    
    # 检查Docker是否以非root用户运行（通过docker.sock权限判断）
    try:
        result = subprocess.run(["ls", "-la", "/var/run/docker.sock"], 
                              capture_output=True, text=True, check=True)
        if "root" in result.stdout and "docker" not in result.stdout:
            issues.append("Docker socket权限配置不当，仅root可访问")
    except:
        pass
    
    # 检查是否启用了内容信任
    try:
        result = subprocess.run(["docker", "trust", "inspect", "--pretty", "registry:2"], 
                              capture_output=True, text=True)
        if "Error" in result.stdout:
            issues.append("未启用Docker内容信任")
    except:
        issues.append("未启用Docker内容信任")
    
    # 检查Docker版本是否为最新
    try:
        result = subprocess.run(["docker", "--version"], capture_output=True, text=True)
        print(f"当前Docker版本: {result.stdout.strip()}")
    except:
        pass
    
    if issues:
        print("发现的Docker安全配置问题:")
        for issue in issues:
            print(f"  - {issue}")
    else:
        print("未发现明显的Docker安全配置问题")

# 使用示例
if __name__ == "__main__":
    print("边缘计算容器安全扫描工具")
    print("======================\n")
    
    if not check_docker_installed():
        sys.exit(1)
    
    # 选择扫描方式
    print("1. 扫描所有镜像")
    print("2. 扫描特定镜像")
    
    choice = input("请选择扫描方式 (1/2): ")
    
    scan_results = {}
    
    if choice == '1':
        images = list_images()
        print(f"\n发现 {len(images)} 个镜像:")
        for i, img in enumerate(images, 1):
            print(f"  {i}. {img}")
        
        for image in images:
            vulns = scan_image(image)
            scan_results[image] = vulns
    
    elif choice == '2':
        image_name = input("请输入要扫描的镜像名称: ")
        vulns = scan_image(image_name)
        scan_results[image_name] = vulns
    
    else:
        print("无效选择")
        sys.exit(1)
    
    # 生成报告
    report_file = "container_security_scan_report.txt"
    generate_scan_report(scan_results, report_file)
    
    # 检查Docker安全配置
    docker_security_hardening_check()

边缘数据安全威胁矩阵
┌─────────────┬─────────────┬─────────────┐
│  数据泄露    │  数据篡改    │  隐私侵犯    │
├─────────────┼─────────────┼─────────────┤
│  1. 传输过程  │  1. 中间人    │  1. 未授权访问 │
│  2. 存储环节  │  2. 注入攻击  │  2. 数据收集  │
│  3. API暴露   │  3. 恶意软件   │  3. 行为分析  │
└─────────────┴─────────────┴─────────────┘

# 边缘数据加密传输工具示例
import os
import sys
import socket
import argparse
import json
import base64
import hashlib
import time
import threading
import random
from datetime import datetime

# 简化版加密模块（实际应用中应使用专业加密库如cryptography）
class SimpleEncryption:
    @staticmethod
    def _simple_xor(data, key):
        """简单XOR加密（仅用于演示）"""
        result = bytearray()
        for i in range(len(data)):
            result.append(data[i] ^ key[i % len(key)])
        return bytes(result)
    
    @staticmethod
    def generate_key(password, salt=None):
        """基于密码生成加密密钥"""
        if salt is None:
            salt = os.urandom(16)
        key = hashlib.pbkdf2_hmac('sha256', password.encode(), salt, 100000)
        return key, salt
    
    @staticmethod
    def encrypt(data, password):
        """加密数据"""
        if isinstance(data, str):
            data = data.encode()
        key, salt = SimpleEncryption.generate_key(password)
        encrypted = SimpleEncryption._simple_xor(data, key)
        # 返回base64编码的盐和加密数据
        return base64.b64encode(salt + encrypted).decode()
    
    @staticmethod
    def decrypt(encrypted_data, password):
        """解密数据"""
        # 解码base64数据
        raw_data = base64.b64decode(encrypted_data)
        salt = raw_data[:16]
        encrypted = raw_data[16:]
        key, _ = SimpleEncryption.generate_key(password, salt)
        decrypted = SimpleEncryption._simple_xor(encrypted, key)
        return decrypted.decode()

class SecureDataTransmitter:
    def __init__(self, host='0.0.0.0', port=9000, password='secure_password'):
        self.host = host
        self.port = port
        self.password = password
        self.server_socket = None
        self.is_running = False
        self.connections = []
        self.lock = threading.Lock()
        
    def start_server(self):
        """启动加密数据服务器"""
        try:
            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self.server_socket.bind((self.host, self.port))
            self.server_socket.listen(5)
            self.is_running = True
            
            print(f"安全数据服务器启动在 {self.host}:{self.port}")
            print(f"加密模式: 基于密码的XOR加密 (演示用)")
            print(f"按Ctrl+C停止服务器...")
            
            while self.is_running:
                try:
                    client_socket, addr = self.server_socket.accept()
                    print(f"新连接: {addr}")
                    
                    with self.lock:
                        self.connections.append((client_socket, addr))
                    
                    # 为每个客户端创建一个处理线程
                    client_thread = threading.Thread(target=self._handle_client, 
                                                    args=(client_socket, addr))
                    client_thread.daemon = True
                    client_thread.start()
                except socket.error as e:
                    if not self.is_running:  # 如果是主动关闭则不报错
                        break
                    print(f"接受连接错误: {e}")
        except Exception as e:
            print(f"服务器启动失败: {e}")
            self.stop_server()
    
    def _handle_client(self, client_socket, addr):
        """处理客户端连接"""
        try:
            while self.is_running:
                # 接收数据
                data = client_socket.recv(4096)
                if not data:
                    break
                
                # 解密数据
                try:
                    encrypted_message = data.decode()
                    decrypted_message = SimpleEncryption.decrypt(encrypted_message, self.password)
                    print(f"[{addr}] 收到: {decrypted_message}")
                    
                    # 解析消息
                    message = json.loads(decrypted_message)
                    
                    # 处理消息并响应
                    response = self._process_message(message)
                    
                    # 加密响应
                    encrypted_response = SimpleEncryption.encrypt(json.dumps(response), self.password)
                    client_socket.send(encrypted_response.encode())
                    print(f"[{addr}] 已响应")
                    
                except Exception as e:
                    print(f"[{addr}] 处理数据错误: {e}")
                    # 发送错误响应
                    error_response = {"status": "error", "message": "处理数据时出错"}
                    encrypted_error = SimpleEncryption.encrypt(json.dumps(error_response), self.password)
                    client_socket.send(encrypted_error.encode())
        
        except Exception as e:
            print(f"[{addr}] 连接错误: {e}")
        finally:
            print(f"[{addr}] 连接关闭")
            client_socket.close()
            
            with self.lock:
                for i, (sock, a) in enumerate(self.connections):
                    if a == addr:
                        self.connections.pop(i)
                        break
    
    def _process_message(self, message):
        """处理接收到的消息"""
        message_type = message.get("type", "unknown")
        
        if message_type == "ping":
            return {"status": "ok", "message": "pong", "timestamp": time.time()}
        
        elif message_type == "data_upload":
            data_id = message.get("data_id", str(random.randint(1000, 9999)))
            # 这里可以添加数据验证、存储等逻辑
            return {"status": "ok", "message": "数据已接收", "data_id": data_id}
        
        elif message_type == "request_data":
            # 模拟数据查询
            return {"status": "ok", "data": {"sample": "这是模拟的加密数据", "timestamp": time.time()}}
        
        else:
            return {"status": "error", "message": f"未知消息类型: {message_type}"}
    
    def stop_server(self):
        """停止服务器"""
        print("正在停止服务器...")
        self.is_running = False
        
        # 关闭所有客户端连接
        with self.lock:
            for client_socket, addr in self.connections:
                try:
                    client_socket.close()
                except:
                    pass
            self.connections.clear()
        
        # 关闭服务器套接字
        if self.server_socket:
            try:
                self.server_socket.close()
            except:
                pass
        
        print("服务器已停止")
    
    def send_data(self, target_host, target_port, data):
        """发送加密数据到目标服务器"""
        try:
            # 创建套接字
            client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            client_socket.connect((target_host, target_port))
            
            # 加密数据
            encrypted_data = SimpleEncryption.encrypt(json.dumps(data), self.password)
            
            # 发送数据
            client_socket.send(encrypted_data.encode())
            
            # 接收响应
            response_data = client_socket.recv(4096)
            if response_data:
                decrypted_response = SimpleEncryption.decrypt(response_data.decode(), self.password)
                response = json.loads(decrypted_response)
                print(f"服务器响应: {response}")
                return response
            
        except Exception as e:
            print(f"发送数据失败: {e}")
            return {"status": "error", "message": str(e)}
        finally:
            if 'client_socket' in locals():
                client_socket.close()

def main():
    """主函数"""
    parser = argparse.ArgumentParser(description='边缘数据加密传输工具')
    parser.add_argument('--mode', choices=['server', 'client'], required=True, 
                       help='运行模式: server(服务器) 或 client(客户端)')
    parser.add_argument('--host', default='0.0.0.0', help='主机地址')
    parser.add_argument('--port', type=int, default=9000, help='端口号')
    parser.add_argument('--password', default='secure_password', help='加密密码')
    args = parser.parse_args()
    
    transmitter = SecureDataTransmitter(host=args.host, port=args.port, password=args.password)
    
    if args.mode == 'server':
        try:
            transmitter.start_server()
        except KeyboardInterrupt:
            print("\n接收到停止信号")
        finally:
            transmitter.stop_server()
    
    elif args.mode == 'client':
        # 示例客户端操作
        print("边缘数据加密传输客户端")
        print(f"连接到服务器: {args.host}:{args.port}")
        
        # 发送ping消息
        ping_response = transmitter.send_data(args.host, args.port, {"type": "ping", "timestamp": time.time()})
        print(f"Ping响应: {ping_response}")
        
        # 发送数据上传消息
        data_response = transmitter.send_data(args.host, args.port, {
            "type": "data_upload",
            "timestamp": time.time(),
            "sensor_id": "edge_sensor_001",
            "value": 23.5,
            "unit": "celsius"
        })
        print(f"数据上传响应: {data_response}")
        
        # 请求数据
        request_response = transmitter.send_data(args.host, args.port, {
            "type": "request_data",
            "timestamp": time.time(),
            "sensor_id": "edge_sensor_001"
        })
        print(f"数据请求响应: {request_response}")

if __name__ == "__main__":
    print("边缘数据加密传输工具 - 演示版")
    print("注意: 此工具使用简化的加密方法，仅用于演示目的")
    print("生产环境中请使用专业的加密库和协议\n")
    
    main()

边缘安全防御纵深
┌───────────────────────────────────────────────────────────┐
│                       多层防御体系                          │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  物理安全    │  │  网络安全    │  │  应用安全    │        │
│  │  - 门禁     │  │  - 防火墙    │  │  - 代码审计   │        │
│  │  - 监控     │  │  - IDS/IPS   │  │  - 权限控制   │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  数据安全    │  │  身份认证    │  │  事件响应    │        │
│  │  - 加密     │  │  - MFA      │  │  - 告警      │        │
│  │  - 脱敏     │  │  - 零信任    │  │  - 恢复      │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
└───────────────────────────────────────────────────────────┘

# 边缘设备安全运维自动化工具示例
import os
import sys
import json
import yaml
import paramiko
import time
import threading
import schedule
import argparse
import subprocess
from datetime import datetime
from collections import defaultdict

def load_config(config_file='config.yaml'):
    """加载配置文件"""
    try:
        with open(config_file, 'r', encoding='utf-8') as f:
            config = yaml.safe_load(f)
        return config
    except Exception as e:
        print(f"加载配置文件失败: {e}")
        return None

def save_config(config, config_file='config.yaml'):
    """保存配置文件"""
    try:
        with open(config_file, 'w', encoding='utf-8') as f:
            yaml.dump(config, f, default_flow_style=False, allow_unicode=True)
        return True
    except Exception as e:
        print(f"保存配置文件失败: {e}")
        return False

class EdgeDeviceManager:
    def __init__(self, config):
        self.config = config
        self.devices = config.get('devices', [])
        self.results = defaultdict(list)
        self.lock = threading.Lock()
        self.concurrency = config.get('concurrency', 5)
        self.semaphore = threading.BoundedSemaphore(self.concurrency)
    
    def ssh_connect(self, device):
        """SSH连接到设备"""
        try:
            client = paramiko.SSHClient()
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            
            print(f"连接设备: {device['name']} ({device['ip']})")
            
            if 'key_file' in device:
                key = paramiko.RSAKey.from_private_key_file(device['key_file'])
                client.connect(hostname=device['ip'], username=device['username'], pkey=key, 
                             port=device.get('port', 22), timeout=10)
            else:
                client.connect(hostname=device['ip'], username=device['username'], password=device['password'],
                             port=device.get('port', 22), timeout=10)
            
            return client
        except Exception as e:
            error_msg = f"连接失败: {e}"
            print(f"{device['name']}: {error_msg}")
            return None, error_msg
    
    def execute_command(self, client, command):
        """在设备上执行命令"""
        try:
            stdin, stdout, stderr = client.exec_command(command, timeout=60)
            output = stdout.read().decode()
            error = stderr.read().decode()
            return output, error
        except Exception as e:
            return "", f"命令执行失败: {e}"
    
    def check_device_security(self, device):
        """检查设备安全状态"""
        with self.semaphore:
            device_name = device['name']
            print(f"\n开始安全检查: {device_name}")
            
            # 连接设备
            client, error = self.ssh_connect(device)
            if not client:
                with self.lock:
                    self.results[device_name].append({
                        "timestamp": datetime.now().isoformat(),
                        "check": "connection",
                        "status": "fail",
                        "details": error
                    })
                return
            
            try:
                # 1. 检查系统更新
                print(f"[{device_name}] 检查系统更新...")
                update_check_cmd = "apt list --upgradable 2>/dev/null || yum check-update 2>/dev/null || echo 'Unknown package manager'"
                output, error = self.execute_command(client, update_check_cmd)
                
                update_count = 0
                if "Unknown" not in output:
                    # 简单计算可更新包数量
                    update_lines = [line for line in output.split('\n') if line.strip() and not line.startswith('Listing') and not line.startswith('Loaded')]
                    update_count = len(update_lines)
                
                with self.lock:
                    self.results[device_name].append({
                        "timestamp": datetime.now().isoformat(),
                        "check": "system_updates",
                        "status": "warning" if update_count > 0 else "ok",
                        "details": f"可更新包数量: {update_count}"
                    })
                
                # 2. 检查开放端口
                print(f"[{device_name}] 检查开放端口...")
                port_cmd = "netstat -tuln || ss -tuln"
                output, error = self.execute_command(client, port_cmd)
                
                open_ports = []
                for line in output.split('\n'):
                    if 'LISTEN' in line:
                        # 简化解析，实际应该更严谨
                        parts = line.split()
                        if len(parts) >= 4:
                            # 不同系统输出格式可能不同
                            if ':' in parts[-2]:  # netstat格式
                                port_info = parts[-2].split(':')[-1]
                                open_ports.append(port_info)
                            else:  # 其他格式
                                open_ports.append(parts[-1].split(':')[-1])
                
                # 检查常见危险端口
                dangerous_ports = ['21', '23', '111', '135', '139', '445', '1433', '3306', '3389']
                risky_ports = [port for port in open_ports if port in dangerous_ports]
                
                with self.lock:
                    self.results[device_name].append({
                        "timestamp": datetime.now().isoformat(),
                        "check": "open_ports",
                        "status": "warning" if risky_ports else "ok",
                        "details": f"开放端口: {', '.join(open_ports)}, 危险端口: {', '.join(risky_ports) if risky_ports else '无'}"
                    })
                
                # 3. 检查用户账户
                print(f"[{device_name}] 检查用户账户...")
                user_cmd = "cat /etc/passwd | grep '0:0'"  # 查找具有root权限的用户
                output, error = self.execute_command(client, user_cmd)
                
                root_users = [line.split(':')[0] for line in output.split('\n') if line.strip()]
                
                with self.lock:
                    self.results[device_name].append({
                        "timestamp": datetime.now().isoformat(),
                        "check": "user_accounts",
                        "status": "warning" if len(root_users) > 1 else "ok",
                        "details": f"具有root权限的用户: {', '.join(root_users)}"
                    })
                
                # 4. 检查防火墙状态
                print(f"[{device_name}] 检查防火墙状态...")
                firewall_cmds = [
                    "systemctl status firewalld 2>/dev/null | grep Active",
                    "ufw status 2>/dev/null | grep Status",
                    "/etc/init.d/iptables status 2>/dev/null | grep Tables"
                ]
                
                firewall_active = False
                for cmd in firewall_cmds:
                    output, error = self.execute_command(client, cmd)
                    if "active" in output.lower() or "enabled" in output.lower() or "yes" in output.lower():
                        firewall_active = True
                        break
                
                with self.lock:
                    self.results[device_name].append({
                        "timestamp": datetime.now().isoformat(),
                        "check": "firewall",
                        "status": "ok" if firewall_active else "warning",
                        "details": "防火墙已激活" if firewall_active else "防火墙未激活"
                    })
                
                # 5. 检查系统日志中的安全事件
                print(f"[{device_name}] 检查安全事件...")
                log_cmd = "grep -i 'failed password\|authentication failure' /var/log/auth.log /var/log/secure 2>/dev/null | tail -20"
                output, error = self.execute_command(client, log_cmd)
                
                failed_attempts = len(output.split('\n')) if output else 0
                
                with self.lock:
                    self.results[device_name].append({
                        "timestamp": datetime.now().isoformat(),
                        "check": "security_events",
                        "status": "warning" if failed_attempts > 10 else "ok",
                        "details": f"登录失败尝试: {failed_attempts}"
                    })
                    
            except Exception as e:
                with self.lock:
                    self.results[device_name].append({
                        "timestamp": datetime.now().isoformat(),
                        "check": "general",
                        "status": "error",
                        "details": f"检查过程中出错: {e}"
                    })
            finally:
                client.close()
            
            print(f"{device_name} 安全检查完成")
    
    def run_all_checks(self):
        """对所有设备运行安全检查"""
        print(f"开始对 {len(self.devices)} 台设备进行安全检查")
        print(f"并发数: {self.concurrency}")
        print("=" * 60)
        
        threads = []
        for device in self.devices:
            thread = threading.Thread(target=self.check_device_security, args=(device,))
            threads.append(thread)
            thread.start()
        
        # 等待所有线程完成
        for thread in threads:
            thread.join()
        
        print("\n" + "=" * 60)
        print("所有设备安全检查完成")
    
    def generate_report(self, output_file='edge_security_report.json'):
        """生成安全检查报告"""
        report = {
            "generated_at": datetime.now().isoformat(),
            "total_devices": len(self.devices),
            "results": dict(self.results)
        }
        
        try:
            with open(output_file, 'w', encoding='utf-8') as f:
                json.dump(report, f, ensure_ascii=False, indent=2)
            print(f"\n安全报告已保存至: {output_file}")
            
            # 生成简化的控制台摘要
            self._print_summary()
            
            return True
        except Exception as e:
            print(f"保存报告失败: {e}")
            return False
    
    def _print_summary(self):
        """打印安全检查摘要"""
        print("\n安全检查摘要:")
        print("-" * 60)
        
        for device_name, checks in self.results.items():
            failed_checks = [check for check in checks if check['status'] in ['fail', 'error']]
            warning_checks = [check for check in checks if check['status'] == 'warning']
            
            print(f"设备: {device_name}")
            print(f"  失败项: {len(failed_checks)}")
            for check in failed_checks:
                print(f"    - {check['check']}: {check['details']}")
            
            print(f"  警告项: {len(warning_checks)}")
            for check in warning_checks:
                print(f"    - {check['check']}: {check['details']}")
            
            print()
    
    def deploy_update(self, device, update_script):
        """部署安全更新"""
        # 此处可以实现安全更新部署逻辑
        pass
    
    def setup_scheduled_checks(self):
        """设置定期安全检查"""
        # 根据配置设置定期任务
        schedule_config = self.config.get('scheduled_checks', {})
        
        if schedule_config.get('daily', False):
            schedule.every().day.at(schedule_config.get('daily_time', '00:00')).do(self.run_all_checks)
            schedule.every().day.at(schedule_config.get('daily_time', '00:00')).do(self.generate_report)
        
        if schedule_config.get('weekly', False):
            day_of_week = schedule_config.get('weekly_day', 'monday')
            time_of_day = schedule_config.get('weekly_time', '00:00')
            getattr(schedule.every(), day_of_week).at(time_of_day).do(self.run_all_checks)
            getattr(schedule.every(), day_of_week).at(time_of_day).do(self.generate_report)
        
        print("已设置定期安全检查")
        
        # 运行调度器
        try:
            while True:
                schedule.run_pending()
                time.sleep(60)
        except KeyboardInterrupt:
            print("\n定期检查已停止")

def generate_sample_config(output_file='config.yaml'):
    """生成示例配置文件"""
    sample_config = {
        'concurrency': 5,
        'devices': [
            {
                'name': 'edge_gateway_001',
                'ip': '192.168.1.100',
                'port': 22,
                'username': 'admin',
                'password': 'your_password_here'  # 实际使用中应避免明文密码
            },
            {
                'name': 'edge_gateway_002',
                'ip': '192.168.1.101',
                'port': 22,
                'username': 'admin',
                'key_file': '/path/to/ssh_key.pem'  # 推荐使用密钥认证
            }
        ],
        'scheduled_checks': {
            'daily': False,
            'daily_time': '00:00',
            'weekly': True,
            'weekly_day': 'monday',
            'weekly_time': '00:00'
        }
    }
    
    save_config(sample_config, output_file)
    print(f"示例配置文件已生成: {output_file}")

def main():
    parser = argparse.ArgumentParser(description='边缘设备安全运维自动化工具')
    parser.add_argument('--config', default='config.yaml', help='配置文件路径')
    parser.add_argument('--generate-config', action='store_true', help='生成示例配置文件')
    parser.add_argument('--check', action='store_true', help='运行安全检查')
    parser.add_argument('--schedule', action='store_true', help='启动定期检查')
    parser.add_argument('--report', default='edge_security_report.json', help='报告输出文件')
    args = parser.parse_args()
    
    if args.generate_config:
        generate_sample_config(args.config)
        return
    
    # 加载配置
    config = load_config(args.config)
    if not config:
        print("无法继续，配置文件加载失败")
        sys.exit(1)
    
    # 创建设备管理器
    manager = EdgeDeviceManager(config)
    
    if args.check:
        manager.run_all_checks()
        manager.generate_report(args.report)
    elif args.schedule:
        manager.setup_scheduled_checks()
    else:
        parser.print_help()

if __name__ == "__main__":
    print("边缘设备安全运维自动化工具")
    print("========================\n")
    main()

1. 物联网设备架构分析与固件安全评估
2. 设备固件逆向工程与漏洞挖掘
3. 边缘节点安全配置与通信协议加固
4. 边缘-云安全通信实现与数据保护
5. 物联网安全监控与入侵检测系统部署