"""S7comm Protocol Security Analyzer.
Analyzes Siemens S7comm protocol traffic to identify security
vulnerabilities, unauthorized access patterns, and potential
attack indicators against SIMATIC S7 PLCs.
"""
import struct
import sys
import json
from collections import defaultdict
from datetime import datetime
from typing import Dict, List, Optional
try:
from scapy.all import rdpcap, IP, TCP
except ImportError:
print("Install scapy: pip install scapy")
sys.exit(1)
S7_ROSCTR = {
0x01: "Job (Request)",
0x02: "Ack",
0x03: "Ack_Data (Response)",
0x07: "Userdata",
}
S7_FUNCTIONS = {
0x00: "CPU services",
0x04: "Read Variable",
0x05: "Write Variable",
0x1A: "Request Download (Program)",
0x1B: "Download Block",
0x1C: "Download Ended",
0x1D: "Start Upload (Read Program)",
0x1E: "Upload Block",
0x1F: "Upload Ended",
0x28: "PI Service (Start/Stop CPU)",
0x29: "PLC Stop",
0xF0: "Setup Communication",
}
CRITICAL_FUNCTIONS = {0x1A, 0x1B, 0x1C, 0x28, 0x29, 0x05}
PROGRAM_FUNCTIONS = {0x1A, 0x1B, 0x1C, 0x1D, 0x1E, 0x1F}
class S7commSecurityFinding:
"""Represents a security finding in S7comm traffic."""
def __init__(self, severity: str, finding_type: str, src_ip: str,
dst_ip: str, function: str, description: str,
cve: str = "", recommendation: str = ""):
self.timestamp = datetime.now().isoformat()
self.severity = severity
self.finding_type = finding_type
self.src_ip = src_ip
self.dst_ip = dst_ip
self.function = function
self.description = description
self.cve = cve
self.recommendation = recommendation
class S7commAnalyzer:
"""Analyzes S7comm protocol traffic for security vulnerabilities."""
def __init__(self):
self.findings: List[S7commSecurityFinding] = []
self.sessions: Dict[str, dict] = defaultdict(lambda: {
"packets": 0,
"functions_seen": set(),
"writes": 0,
"program_downloads": 0,
"cpu_commands": 0,
"first_seen": None,
"last_seen": None,
})
self.authorized_engineering: set = set()
self.packet_count = 0
def set_authorized_stations(self, ips: List[str]):
"""Set list of authorized engineering workstation IPs."""
self.authorized_engineering = set(ips)
def parse_s7comm(self, payload: bytes) -> Optional[dict]:
"""Parse S7comm protocol data from TCP payload."""
if len(payload) < 4:
return None
tpkt_version = payload[0]
if tpkt_version != 3:
return None
tpkt_length = struct.unpack(">H", payload[2:4])[0]
if len(payload) < 7:
return None
cotp_length = payload[4]
cotp_type = payload[5]
s7_offset = 4 + 1 + cotp_length
if len(payload) < s7_offset + 10:
return None
protocol_id = payload[s7_offset]
if protocol_id != 0x32:
return None
rosctr = payload[s7_offset + 1]
redundancy = struct.unpack(">H", payload[s7_offset + 2:s7_offset + 4])[0]
pdu_ref = struct.unpack(">H", payload[s7_offset + 4:s7_offset + 6])[0]
param_length = struct.unpack(">H", payload[s7_offset + 6:s7_offset + 8])[0]
data_length = struct.unpack(">H", payload[s7_offset + 8:s7_offset + 10])[0]
result = {
"rosctr": rosctr,
"rosctr_name": S7_ROSCTR.get(rosctr, f"Unknown (0x{rosctr:02x})"),
"pdu_ref": pdu_ref,
"param_length": param_length,
"data_length": data_length,
}
param_offset = s7_offset + 10
if rosctr in (0x01, 0x03) and param_length > 0 and len(payload) > param_offset:
func_code = payload[param_offset]
result["function_code"] = func_code
result["function_name"] = S7_FUNCTIONS.get(func_code, f"Unknown (0x{func_code:02x})")
return result
def analyze_packet(self, pkt):
"""Analyze a packet for S7comm security issues."""
self.packet_count += 1
if not pkt.haslayer(IP) or not pkt.haslayer(TCP):
return
tcp = pkt[TCP]
if tcp.dport != 102 and tcp.sport != 102:
return
payload = bytes(tcp.payload)
if not payload:
return
s7 = self.parse_s7comm(payload)
if not s7:
return
src_ip = pkt[IP].src
dst_ip = pkt[IP].dst
session_key = f"{src_ip}->{dst_ip}"
session = self.sessions[session_key]
session["packets"] += 1
if session["first_seen"] is None:
session["first_seen"] = float(pkt.time)
session["last_seen"] = float(pkt.time)
func_code = s7.get("function_code")
if func_code is not None:
session["functions_seen"].add(func_code)
if tcp.dport == 102 and func_code in CRITICAL_FUNCTIONS:
if self.authorized_engineering and src_ip not in self.authorized_engineering:
self.findings.append(S7commSecurityFinding(
severity="CRITICAL",
finding_type="UNAUTHORIZED_ENGINEERING_ACCESS",
src_ip=src_ip, dst_ip=dst_ip,
function=s7.get("function_name", "Unknown"),
description=(
f"Critical S7comm operation from unauthorized source {src_ip}. "
f"Function: {s7.get('function_name')}. Only authorized TIA Portal "
f"workstations should issue these commands."
),
recommendation="Block unauthorized sources at industrial firewall. Investigate source host for compromise.",
))
if func_code == 0x29:
session["cpu_commands"] += 1
self.findings.append(S7commSecurityFinding(
severity="CRITICAL",
finding_type="CPU_STOP_COMMAND",
src_ip=src_ip, dst_ip=dst_ip,
function="PLC CPU Stop (0x29)",
description=f"CPU STOP command sent to PLC at {dst_ip}. This halts PLC program execution.",
cve="MITRE T0881 - Service Stop",
recommendation="Verify if this is an authorized maintenance action. If not, isolate source immediately.",
))
if func_code in (0x1A, 0x1B, 0x1C):
session["program_downloads"] += 1
self.findings.append(S7commSecurityFinding(
severity="CRITICAL",
finding_type="PROGRAM_DOWNLOAD",
src_ip=src_ip, dst_ip=dst_ip,
function=s7.get("function_name", "Download"),
description=(
f"PLC program download operation to {dst_ip}. "
f"This modifies the running control logic on the PLC."
),
cve="MITRE T0843 - Program Download",
recommendation="Verify against change management records. Compare with known-good program backup.",
))
if func_code == 0x05:
session["writes"] += 1
if func_code in (0x1D, 0x1E, 0x1F):
self.findings.append(S7commSecurityFinding(
severity="HIGH",
finding_type="PROGRAM_UPLOAD_EXFILTRATION",
src_ip=src_ip, dst_ip=dst_ip,
function=s7.get("function_name", "Upload"),
description=f"PLC program upload (read) from {dst_ip}. Source {src_ip} is extracting PLC control logic.",
recommendation="Verify if this is authorized maintenance. Unauthorized uploads indicate reconnaissance.",
))
def check_known_vulnerabilities(self):
"""Check for known Siemens S7 vulnerabilities based on observed behavior."""
vuln_checks = [
{
"name": "S7-300/400 Replay Attack Vulnerability",
"cve": "CVE-2019-13945",
"description": "S7-300/400 PLCs lack integrity checks on S7comm sessions, allowing replay attacks",
"affected": "S7-300, S7-400 (all firmware versions)",
"severity": "HIGH",
},
{
"name": "S7CommPlus Integrity Bypass",
"cve": "Research finding (Biham et al.)",
"description": "S7CommPlusV3 integrity mechanism can be bypassed by attackers who can observe one legitimate session",
"affected": "S7-1200 (< V4.5), S7-1500 (< V2.9)",
"severity": "HIGH",
},
{
"name": "Unpatchable Hardware Root of Trust",
"cve": "CVE-2022-38773",
"description": "Hardware vulnerability allows bypassing protected boot and persistent firmware modification",
"affected": "S7-1500 (specific hardware revisions)",
"severity": "CRITICAL",
},
{
"name": "Remote DoS via Port 102",
"cve": "CVE-2019-10929",
"description": "Specially crafted packets on TCP port 102 can crash S7 PLCs remotely",
"affected": "S7-300, S7-400, S7-1200, S7-1500 (specific firmware)",
"severity": "HIGH",
},
]
return vuln_checks
def generate_report(self):
"""Generate comprehensive S7comm security analysis report."""
print(f"\n{'='*70}")
print("S7COMM PROTOCOL SECURITY ANALYSIS REPORT")
print(f"{'='*70}")
print(f"Analysis Time: {datetime.now().isoformat()}")
print(f"Packets Analyzed: {self.packet_count}")
print(f"S7comm Sessions: {len(self.sessions)}")
print(f"Security Findings: {len(self.findings)}")
print(f"\n--- SESSION SUMMARY ---")
for key, session in self.sessions.items():
funcs = [S7_FUNCTIONS.get(f, f"0x{f:02x}") for f in session["functions_seen"]]
print(f"\n {key}")
print(f" Packets: {session['packets']}")
print(f" Functions: {', '.join(funcs)}")
print(f" Writes: {session['writes']}")
print(f" Program Downloads: {session['program_downloads']}")
print(f" CPU Commands: {session['cpu_commands']}")
if self.findings:
print(f"\n--- SECURITY FINDINGS ---")
for f in self.findings:
print(f"\n [{f.severity}] {f.finding_type}")
print(f" Source: {f.src_ip} -> {f.dst_ip}")
print(f" Function: {f.function}")
print(f" Detail: {f.description}")
if f.cve:
print(f" Reference: {f.cve}")
if f.recommendation:
print(f" Action: {f.recommendation}")
print(f"\n--- KNOWN VULNERABILITY ASSESSMENT ---")
for vuln in self.check_known_vulnerabilities():
print(f"\n [{vuln['severity']}] {vuln['name']}")
print(f" CVE: {vuln['cve']}")
print(f" Affected: {vuln['affected']}")
print(f" Detail: {vuln['description']}")
if __name__ == "__main__":
analyzer = S7commAnalyzer()
analyzer.set_authorized_stations(["10.10.2.50", "10.10.2.51"])
if len(sys.argv) >= 2:
print(f"[*] Analyzing capture: {sys.argv[1]}")
packets = rdpcap(sys.argv[1])
for pkt in packets:
analyzer.analyze_packet(pkt)
analyzer.generate_report()
else:
print("Usage: python s7comm_analyzer.py <capture.pcap>")
print(" Analyzes S7comm traffic for security vulnerabilities")
