performing-plc-firmware-security-analysis

执行 PLC 固件安全分析

何时使用

• 在作为 IEC 62443 组件安全评估（IEC 62443-4-2）的一部分评估 PLC 安全性时
• 在疑似遭受入侵或供应链攻击后验证固件完整性时
• 在关键基础设施中部署新 PLC 平台之前评估其安全性时
• 在授权实验室中对工业控制系统设备进行漏洞研究时
• 在响应怀疑 PLC 逻辑或固件被篡改的事件时

请勿在未获得明确授权且未采取安全控制措施的在线生产 PLC 上使用。固件提取和分析应在实验室设备或离线备份上进行。切勿将 PLC 固件上传至公共分析服务。授权在线测试流程请参阅 performing-ics-penetration-testing。

前置条件

• 具备目标 PLC 硬件或仿真环境的隔离实验室环境
• 目标平台的 PLC 编程软件（Siemens TIA Portal、Rockwell Studio 5000、Schneider EcoStruxure）
• 固件提取工具（binwalk、firmware-mod-kit、JTAG/SWD 调试器）
• 静态分析工具（支持 ARM/MIPS/PowerPC 的 Ghidra、IDA Pro、Binary Ninja）
• 理解 PLC 架构（实时操作系统、梯形逻辑执行、I/O 扫描）
• 用于完整性比较的已知良好固件参考副本

工作流

步骤 1：获取用于分析的 PLC 固件

通过授权方法提取或获取 PLC 固件。可以通过从供应商处下载、从实验室设备中提取，或从项目备份中获取来完成。

#!/usr/bin/env python3
"""PLC Firmware Acquisition and Integrity Verification.

Supports firmware extraction from project files, network downloads,
and binary image comparison against known-good baselines.
"""

import hashlib
import json
import os
import struct
import sys
import zipfile
from datetime import datetime
from pathlib import Path


class PLCFirmwareAcquisition:
    """Handles PLC firmware acquisition from various sources."""

    def __init__(self, output_dir="firmware_analysis"):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)
        self.manifest = {
            "acquisition_date": datetime.now().isoformat(),
            "firmware_samples": [],
        }

    def extract_from_siemens_project(self, project_path):
        """Extract firmware/program blocks from Siemens TIA Portal project.

        TIA Portal projects (.ap16/.ap17) are ZIP archives containing
        XML-encoded PLC program blocks and system configuration.
        """
        print(f"[*] Analyzing Siemens project: {project_path}")
        results = {"platform": "Siemens", "blocks": []}

        if zipfile.is_zipfile(project_path):
            with zipfile.ZipFile(project_path, "r") as zf:
                for info in zf.infolist():
                    # Program blocks are stored as XML in specific paths
                    if "ProgramBlocks" in info.filename or "SystemBlocks" in info.filename:
                        block_data = zf.read(info.filename)
                        block_hash = hashlib.sha256(block_data).hexdigest()

                        block_path = self.output_dir / info.filename.replace("/", "_")
                        block_path.write_bytes(block_data)

                        results["blocks"].append({
                            "name": info.filename,
                            "size": info.file_size,
                            "sha256": block_hash,
                            "extracted_to": str(block_path),
                        })
                        print(f"  [+] Extracted: {info.filename} ({info.file_size} bytes)")

        self.manifest["firmware_samples"].append(results)
        return results

    def extract_from_rockwell_project(self, acd_path):
        """Extract program data from Rockwell Studio 5000 ACD file.

        ACD files contain controller program, tags, and configuration.
        """
        print(f"[*] Analyzing Rockwell project: {acd_path}")
        results = {"platform": "Rockwell/Allen-Bradley", "blocks": []}

        with open(acd_path, "rb") as f:
            header = f.read(256)
            # ACD files have a specific signature
            if b"RSLogix" in header or b"Studio 5000" in header:
                f.seek(0)
                full_data = f.read()
                file_hash = hashlib.sha256(full_data).hexdigest()

                results["blocks"].append({
                    "name": os.path.basename(acd_path),
                    "size": len(full_data),
                    "sha256": file_hash,
                    "header_signature": header[:16].hex(),
                })
                print(f"  [+] Project hash: {file_hash}")

        self.manifest["firmware_samples"].append(results)
        return results

    def compute_firmware_hash(self, firmware_path):
        """Compute multiple hashes of a firmware image for integrity tracking."""
        data = Path(firmware_path).read_bytes()
        return {
            "file": str(firmware_path),
            "size": len(data),
            "md5": hashlib.md5(data).hexdigest(),
            "sha256": hashlib.sha256(data).hexdigest(),
            "sha512": hashlib.sha512(data).hexdigest(),
        }

    def compare_firmware_integrity(self, current_fw, baseline_fw):
        """Compare current firmware against known-good baseline."""
        current_hash = self.compute_firmware_hash(current_fw)
        baseline_hash = self.compute_firmware_hash(baseline_fw)

        match = current_hash["sha256"] == baseline_hash["sha256"]

        result = {
            "comparison_date": datetime.now().isoformat(),
            "current_firmware": current_hash,
            "baseline_firmware": baseline_hash,
            "integrity_match": match,
            "verdict": "PASS - Firmware matches baseline" if match else "FAIL - Firmware modified!",
        }

        if not match:
            # Find the offset where files diverge
            current_data = Path(current_fw).read_bytes()
            baseline_data = Path(baseline_fw).read_bytes()
            min_len = min(len(current_data), len(baseline_data))

            first_diff = None
            diff_count = 0
            for i in range(min_len):
                if current_data[i] != baseline_data[i]:
                    if first_diff is None:
                        first_diff = i
                    diff_count += 1

            result["first_difference_offset"] = f"0x{first_diff:08x}" if first_diff else None
            result["total_different_bytes"] = diff_count
            result["size_difference"] = len(current_data) - len(baseline_data)

        return result

    def save_manifest(self):
        """Save acquisition manifest."""
        manifest_path = self.output_dir / "acquisition_manifest.json"
        with open(manifest_path, "w") as f:
            json.dump(self.manifest, f, indent=2)
        print(f"\n[*] Manifest saved: {manifest_path}")


if __name__ == "__main__":
    acq = PLCFirmwareAcquisition()

    if len(sys.argv) < 2:
        print("Usage:")
        print("  python process.py extract-siemens <project.ap17>")
        print("  python process.py extract-rockwell <project.acd>")
        print("  python process.py compare <current.bin> <baseline.bin>")
        sys.exit(1)

    cmd = sys.argv[1]
    if cmd == "extract-siemens" and len(sys.argv) > 2:
        acq.extract_from_siemens_project(sys.argv[2])
    elif cmd == "extract-rockwell" and len(sys.argv) > 2:
        acq.extract_from_rockwell_project(sys.argv[2])
    elif cmd == "compare" and len(sys.argv) > 3:
        result = acq.compare_firmware_integrity(sys.argv[2], sys.argv[3])
        print(json.dumps(result, indent=2))
    else:
        print("Invalid command")
        sys.exit(1)

    acq.save_manifest()

步骤 2：对固件镜像执行静态分析

使用 binwalk 解包固件，使用 Ghidra 进行反汇编，以识别固件二进制文件中的安全问题。

# Step 2a: Unpack firmware image with binwalk
binwalk -e firmware.bin
# Output: _firmware.bin.extracted/

# Identify firmware components
binwalk firmware.bin
# Look for: file system images, compressed sections, bootloader, RTOS kernel

# Extract strings for credential and configuration analysis
strings -n 8 firmware.bin > firmware_strings.txt

# Search for hardcoded credentials
grep -iE "(password|passwd|pwd|secret|key|credential|login|admin|root)" firmware_strings.txt

# Search for network configuration
grep -iE "(http|ftp|telnet|ssh|snmp|modbus|192\.168|10\.|172\.)" firmware_strings.txt

# Search for debug/backdoor indicators
grep -iE "(debug|backdoor|test_mode|factory|service_port|hidden)" firmware_strings.txt

# Search for cryptographic material
grep -iE "(BEGIN RSA|BEGIN CERTIFICATE|AES|DES|private.key)" firmware_strings.txt

# Step 2b: Entropy analysis to detect encrypted/compressed sections
binwalk -E firmware.bin
# High entropy sections may contain encrypted payloads or compressed data

# Step 2c: Analyze with Ghidra (headless mode)
analyzeHeadless /tmp/ghidra_project PLC_FW \
  -import firmware.bin \
  -processor ARM:LE:32:Cortex \
  -postScript FindCryptoConstants.java \
  -postScript FindHardcodedStrings.java \
  -log /tmp/ghidra_analysis.log

步骤 3：分析 PLC 通信栈安全

检查 PLC 如何处理工业协议请求，重点关注认证绕过、数据包解析中的缓冲区溢出以及命令注入漏洞。

#!/usr/bin/env python3
"""PLC Protocol Security Analyzer.

Tests PLC protocol implementation for common vulnerabilities
including authentication bypass, malformed packet handling,
and function code access control.

WARNING: Only run against lab/test PLCs, never production systems.
"""

import socket
import struct
import sys
import time
from dataclasses import dataclass


@dataclass
class ProtocolTestResult:
    test_name: str
    target: str
    protocol: str
    result: str  # PASS, FAIL, ERROR
    severity: str
    detail: str


class ModbusSecurityTester:
    """Tests Modbus/TCP implementation security."""

    def __init__(self, target_ip, target_port=502):
        self.target = target_ip
        self.port = target_port
        self.results = []

    def _send_modbus(self, unit_id, func_code, data=b""):
        """Send a Modbus/TCP request and return response."""
        # MBAP Header: transaction_id(2) + protocol_id(2) + length(2) + unit_id(1)
        mbap = struct.pack(">HHHB", 0x0001, 0x0000, len(data) + 2, unit_id)
        pdu = struct.pack("B", func_code) + data

        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(5)
            sock.connect((self.target, self.port))
            sock.send(mbap + pdu)
            response = sock.recv(1024)
            sock.close()
            return response
        except Exception as e:
            return None

    def test_authentication_required(self):
        """Test if PLC requires authentication for read/write operations."""
        # Test unauthenticated read
        read_data = struct.pack(">HH", 0, 10)  # Read 10 registers from address 0
        response = self._send_modbus(1, 3, read_data)

        if response and len(response) > 8 and response[7] != 0x83:
            self.results.append(ProtocolTestResult(
                test_name="Modbus Authentication - Read",
                target=self.target,
                protocol="Modbus/TCP",
                result="FAIL",
                severity="high",
                detail="PLC accepts unauthenticated Modbus read commands. No authentication required.",
            ))

        # Test unauthenticated write
        write_data = struct.pack(">HH", 100, 0)  # Write 0 to register 100
        response = self._send_modbus(1, 6, write_data)

        if response and len(response) > 8 and response[7] != 0x86:
            self.results.append(ProtocolTestResult(
                test_name="Modbus Authentication - Write",
                target=self.target,
                protocol="Modbus/TCP",
                result="FAIL",
                severity="critical",
                detail="PLC accepts unauthenticated Modbus WRITE commands. Any host can modify registers.",
            ))

    def test_function_code_access_control(self):
        """Test if PLC restricts dangerous function codes."""
        dangerous_funcs = {
            8: "Diagnostics (can restart communications)",
            17: "Report Slave ID (information disclosure)",
            43: "Encapsulated Interface Transport (device identification)",
        }

        for fc, desc in dangerous_funcs.items():
            response = self._send_modbus(1, fc, b"\x00\x00")
            if response and len(response) > 8:
                error_code = response[7]
                if error_code != (fc | 0x80):  # Not an exception response
                    self.results.append(ProtocolTestResult(
                        test_name=f"Function Code Access - FC{fc}",
                        target=self.target,
                        protocol="Modbus/TCP",
                        result="FAIL",
                        severity="medium",
                        detail=f"PLC responds to FC{fc} ({desc}) without access control",
                    ))

    def test_invalid_unit_id(self):
        """Test PLC response to broadcast and invalid unit IDs."""
        # Broadcast (unit ID 0) - should be carefully handled
        read_data = struct.pack(">HH", 0, 1)
        response = self._send_modbus(0, 3, read_data)

        if response and len(response) > 8 and response[7] != 0x83:
            self.results.append(ProtocolTestResult(
                test_name="Broadcast Unit ID Handling",
                target=self.target,
                protocol="Modbus/TCP",
                result="FAIL",
                severity="high",
                detail="PLC responds to broadcast unit ID 0. This enables broadcast write attacks.",
            ))

    def test_malformed_packet_handling(self):
        """Test PLC resilience against malformed Modbus packets."""
        # Oversized length field
        malformed = struct.pack(">HHH", 0x0001, 0x0000, 0xFFFF) + b"\x01\x03\x00\x00\x00\x01"
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(5)
            sock.connect((self.target, self.port))
            sock.send(malformed)
            time.sleep(1)

            # Verify PLC is still responsive
            read_data = struct.pack(">HH", 0, 1)
            response = self._send_modbus(1, 3, read_data)
            sock.close()

            if response is None:
                self.results.append(ProtocolTestResult(
                    test_name="Malformed Packet - Oversized Length",
                    target=self.target,
                    protocol="Modbus/TCP",
                    result="FAIL",
                    severity="critical",
                    detail="PLC became unresponsive after receiving oversized length field. Possible DoS vulnerability.",
                ))
            else:
                self.results.append(ProtocolTestResult(
                    test_name="Malformed Packet - Oversized Length",
                    target=self.target,
                    protocol="Modbus/TCP",
                    result="PASS",
                    severity="info",
                    detail="PLC correctly handles oversized length field without crashing",
                ))
        except Exception as e:
            pass

    def run_all_tests(self):
        """Run all Modbus security tests."""
        print(f"\n{'='*60}")
        print(f"PLC MODBUS SECURITY ANALYSIS - {self.target}:{self.port}")
        print(f"{'='*60}")

        self.test_authentication_required()
        self.test_function_code_access_control()
        self.test_invalid_unit_id()
        self.test_malformed_packet_handling()

        for r in self.results:
            icon = "[FAIL]" if r.result == "FAIL" else "[PASS]"
            print(f"\n  {icon} {r.test_name}")
            print(f"    Severity: {r.severity}")
            print(f"    Detail: {r.detail}")

        return self.results


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python plc_protocol_tester.py <target_plc_ip> [port]")
        print("WARNING: Only use against lab/test PLCs!")
        sys.exit(1)

    target = sys.argv[1]
    port = int(sys.argv[2]) if len(sys.argv) > 2 else 502

    tester = ModbusSecurityTester(target, port)
    tester.run_all_tests()

关键概念

术语	定义
PLC 固件	运行在可编程逻辑控制器上的嵌入式软件，包括实时操作系统、协议栈和 I/O 驱动
梯形逻辑	PLC 的图形化编程语言，用于表示继电器逻辑电路，以程序块形式存储在 PLC 内存中
功能块	可复用的 PLC 编程元素，将逻辑与已定义的输入/输出封装在一起，可分析是否存在恶意修改
固件完整性	通过加密哈希比较验证 PLC 固件未被从供应商提供或批准的版本修改
IEC 62443-4-2	IEC 62443 标准中的组件安全要求，定义了包括 PLC 在内的 IACS 组件所需的安全能力
JTAG/SWD	用于固件提取和底层分析的硬件调试接口（联合测试行动组 / 串行线调试）

工具与系统

• Binwalk：固件分析工具，用于扫描、提取和分析嵌入式固件镜像
• Ghidra：NSA 开发的逆向工程框架，支持 PLC 中常见的 ARM、MIPS、PowerPC 架构
• EMUX/FIRMADYNE：固件仿真框架，用于嵌入式设备固件的动态分析
• PLCinject：用于分析 PLC 逻辑注入漏洞的研究工具（仅在授权实验室环境中使用）
• OpenPLC：开源 PLC 平台，可作为安全研究的测试目标

输出格式

PLC Firmware Security Analysis Report
=======================================
Device: [PLC Model and Firmware Version]
Analysis Date: YYYY-MM-DD
Methodology: Static + Dynamic Analysis

FIRMWARE INTEGRITY:
  SHA-256: [hash]
  Baseline Match: [Yes/No]
  Vendor Signature Valid: [Yes/No/Not Signed]

VULNERABILITIES FOUND:
  [PLC-001] [Severity] [Title]
    CWE: [CWE-ID]
    Detail: [Technical description]
    Impact: [Operational impact]
    Remediation: [Fix or mitigation]