Performing Plc Firmware Security Analysis

#!/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()

# 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

#!/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()