509 lines
22 KiB
Python
509 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Modbus TCP Traffic Decoder
|
|
Analyzes captured network traffic for Modbus communications,
|
|
specifically looking for temperature setpoint changes.
|
|
"""
|
|
|
|
import struct
|
|
import re
|
|
from typing import List, Dict, Tuple, Optional
|
|
|
|
class ModbusDecoder:
|
|
def __init__(self):
|
|
# Modbus function codes
|
|
self.function_codes = {
|
|
1: "Read Coils",
|
|
2: "Read Discrete Inputs",
|
|
3: "Read Holding Registers",
|
|
4: "Read Input Registers",
|
|
5: "Write Single Coil",
|
|
6: "Write Single Register",
|
|
15: "Write Multiple Coils",
|
|
16: "Write Multiple Registers"
|
|
}
|
|
|
|
# Common Modbus registers for HVAC/Temperature systems
|
|
self.hvac_registers = {
|
|
40001: "Temperature Setpoint",
|
|
40002: "Current Temperature",
|
|
40003: "Humidity Setpoint",
|
|
40004: "Current Humidity",
|
|
40005: "Operating Mode",
|
|
40006: "Fan Speed",
|
|
40007: "System Status"
|
|
}
|
|
|
|
def parse_hex_file(self, filename: str) -> List[Dict]:
|
|
"""Parse the hex dump file and extract packet data"""
|
|
packets = []
|
|
current_packet = []
|
|
packet_num = 0
|
|
|
|
print("=" * 80)
|
|
print("PARSING HEX DUMP FILE - LINE BY LINE ANALYSIS")
|
|
print("=" * 80)
|
|
|
|
with open(filename, 'r') as f:
|
|
lines = f.readlines()
|
|
|
|
for line_num, line in enumerate(lines, 1):
|
|
line = line.strip()
|
|
|
|
if not line:
|
|
if current_packet:
|
|
packet_num += 1
|
|
print(f"\n--- End of Packet {packet_num} ---")
|
|
parsed_packet = self._process_packet_lines(current_packet, packet_num)
|
|
if parsed_packet:
|
|
packets.append(parsed_packet)
|
|
self._analyze_packet_immediately(parsed_packet, packet_num)
|
|
else:
|
|
print(f"Packet {packet_num}: Failed to parse or not valid")
|
|
current_packet = []
|
|
continue
|
|
|
|
# Extract hex data from each line
|
|
if line.startswith(('0000', '0010', '0020', '0030', '0040', '0050')):
|
|
print(f"Line {line_num}: {line}")
|
|
|
|
# Parse the line structure
|
|
parts = line.split(' ')
|
|
if len(parts) >= 2:
|
|
offset = parts[0]
|
|
hex_part = parts[1].strip()
|
|
ascii_part = parts[2] if len(parts) > 2 else ''
|
|
|
|
# Extract hex bytes (remove spaces)
|
|
hex_bytes = hex_part.replace(' ', '')
|
|
current_packet.append(hex_bytes)
|
|
|
|
print(f" Offset: {offset}")
|
|
print(f" Hex Data: {hex_part}")
|
|
print(f" ASCII: {ascii_part}")
|
|
|
|
# Try to identify interesting patterns in this line
|
|
self._analyze_line_patterns(hex_bytes, offset)
|
|
|
|
# Process final packet if exists
|
|
if current_packet:
|
|
packet_num += 1
|
|
print(f"\n--- End of Packet {packet_num} ---")
|
|
parsed_packet = self._process_packet_lines(current_packet, packet_num)
|
|
if parsed_packet:
|
|
packets.append(parsed_packet)
|
|
self._analyze_packet_immediately(parsed_packet, packet_num)
|
|
|
|
return [p for p in packets if p is not None]
|
|
|
|
def _process_packet_lines(self, lines: List[str], packet_num: int) -> Optional[Dict]:
|
|
"""Process lines belonging to a single packet"""
|
|
try:
|
|
print(f" Processing packet {packet_num} with {len(lines)} lines of hex data")
|
|
|
|
# Combine all hex data
|
|
hex_data = ''.join(lines)
|
|
print(f" Total hex length: {len(hex_data)} characters ({len(hex_data)//2} bytes)")
|
|
|
|
# Convert to bytes
|
|
packet_bytes = bytes.fromhex(hex_data)
|
|
|
|
if len(packet_bytes) < 54: # Minimum for Ethernet + IP + TCP headers
|
|
print(f" Packet too short ({len(packet_bytes)} bytes) - skipping")
|
|
return None
|
|
|
|
return self._parse_packet(packet_bytes)
|
|
|
|
except Exception as e:
|
|
print(f" Error processing packet {packet_num}: {e}")
|
|
return None
|
|
|
|
def _analyze_line_patterns(self, hex_bytes: str, offset: str):
|
|
"""Analyze patterns in individual hex lines"""
|
|
try:
|
|
if len(hex_bytes) < 8:
|
|
return
|
|
|
|
# Convert to actual bytes for analysis
|
|
line_bytes = bytes.fromhex(hex_bytes)
|
|
|
|
# Look for common patterns
|
|
interesting_patterns = []
|
|
|
|
# Check for port 502 (Modbus)
|
|
if len(line_bytes) >= 2:
|
|
for i in range(len(line_bytes) - 1):
|
|
port = (line_bytes[i] << 8) | line_bytes[i+1]
|
|
if port == 502:
|
|
interesting_patterns.append(f"Modbus port 502 at position {i}")
|
|
|
|
# Check for Modbus device IDs (typically 1-247)
|
|
for i, byte_val in enumerate(line_bytes):
|
|
if 1 <= byte_val <= 247:
|
|
# Look ahead to see if next byte might be a function code
|
|
if i + 1 < len(line_bytes) and line_bytes[i + 1] in self.function_codes:
|
|
interesting_patterns.append(f"Possible Device ID {byte_val} with function {line_bytes[i + 1]}")
|
|
|
|
# Check for potential temperature values (various ranges)
|
|
if len(line_bytes) >= 2:
|
|
for i in range(len(line_bytes) - 1):
|
|
value = (line_bytes[i] << 8) | line_bytes[i+1]
|
|
# Check common temperature ranges
|
|
if 200 <= value <= 350: # 20.0-35.0°C range
|
|
temp = value / 10.0
|
|
interesting_patterns.append(f"Temp value: {value} = {temp}°C")
|
|
if abs(temp - 23.8) < 0.1:
|
|
interesting_patterns.append("*** 23.8°C SETPOINT FOUND! ***")
|
|
elif value == 238: # Direct match for 23.8
|
|
interesting_patterns.append("*** Direct 238 value (23.8°C) FOUND! ***")
|
|
|
|
# Check for Modbus function codes
|
|
for i, byte_val in enumerate(line_bytes):
|
|
if byte_val in self.function_codes:
|
|
interesting_patterns.append(f"Function {byte_val}: {self.function_codes[byte_val]}")
|
|
|
|
if interesting_patterns:
|
|
print(f" Patterns found:")
|
|
for pattern in interesting_patterns:
|
|
print(f" - {pattern}")
|
|
|
|
except Exception as e:
|
|
pass # Silently ignore parsing errors in pattern detection
|
|
|
|
def _analyze_packet_immediately(self, packet: Dict, packet_num: int):
|
|
"""Provide immediate analysis of a parsed packet"""
|
|
print(f"\n PACKET {packet_num} ANALYSIS:")
|
|
|
|
if packet.get('is_modbus'):
|
|
direction = "REQUEST" if packet.get('is_request') else "RESPONSE"
|
|
device_id = packet.get('modbus_unit_id', 'Unknown')
|
|
transaction_id = packet.get('modbus_trans_id', 'Unknown')
|
|
|
|
print(f" MODBUS {direction}")
|
|
print(f" Device ID: {device_id}")
|
|
print(f" Transaction ID: {transaction_id}")
|
|
print(f" Function: {packet.get('modbus_function_name', 'Unknown')} (Code {packet.get('modbus_function', '?')})")
|
|
|
|
# Detailed Modbus analysis with register values
|
|
if packet.get('modbus_function') == 6: # Write Single Register
|
|
addr = packet.get('register_address')
|
|
value = packet.get('register_value')
|
|
if addr is not None and value is not None:
|
|
temp = self.convert_to_temperature(value)
|
|
print(f" -> Device {device_id} WRITES to Register {addr}")
|
|
print(f" Raw Value: {value} (0x{value:04X})")
|
|
print(f" As Temperature: {temp}°C")
|
|
if abs(temp - 23.8) < 0.1:
|
|
print(f" *** FOUND 23.8°C SETPOINT! ***")
|
|
|
|
elif packet.get('modbus_function') == 16: # Write Multiple Registers
|
|
addr = packet.get('start_address')
|
|
values = packet.get('register_values', [])
|
|
if addr is not None and values:
|
|
print(f" -> Device {device_id} WRITES {len(values)} registers starting at {addr}")
|
|
for j, value in enumerate(values):
|
|
reg_addr = addr + j
|
|
temp = self.convert_to_temperature(value)
|
|
print(f" Register {reg_addr}: {value} (0x{value:04X}) = {temp}°C")
|
|
if abs(temp - 23.8) < 0.1:
|
|
print(f" *** FOUND 23.8°C SETPOINT at Register {reg_addr}! ***")
|
|
|
|
elif packet.get('modbus_function') == 3: # Read Holding Registers
|
|
if packet.get('is_request'):
|
|
addr = packet.get('start_address')
|
|
count = packet.get('num_registers')
|
|
if addr is not None and count is not None:
|
|
print(f" -> Device {device_id} REQUESTS to READ {count} registers starting from {addr}")
|
|
end_addr = addr + count - 1
|
|
print(f" Register range: {addr} to {end_addr}")
|
|
else:
|
|
values = packet.get('register_values', [])
|
|
if values:
|
|
print(f" -> Device {device_id} RESPONDS with {len(values)} register values:")
|
|
for j, value in enumerate(values):
|
|
temp = self.convert_to_temperature(value)
|
|
print(f" Register Value {j}: {value} (0x{value:04X}) = {temp}°C")
|
|
if abs(temp - 23.8) < 0.1:
|
|
print(f" *** Current value is 23.8°C! ***")
|
|
|
|
elif packet.get('modbus_function') == 4: # Read Input Registers
|
|
if packet.get('is_request'):
|
|
addr = packet.get('start_address')
|
|
count = packet.get('num_registers')
|
|
if addr is not None and count is not None:
|
|
print(f" -> Device {device_id} REQUESTS to READ {count} input registers starting from {addr}")
|
|
else:
|
|
values = packet.get('register_values', [])
|
|
if values:
|
|
print(f" -> Device {device_id} RESPONDS with {len(values)} input register values:")
|
|
for j, value in enumerate(values):
|
|
temp = self.convert_to_temperature(value)
|
|
print(f" Input Register {j}: {value} (0x{value:04X}) = {temp}°C")
|
|
|
|
elif packet.get('modbus_function') == 1: # Read Coils
|
|
if packet.get('is_request'):
|
|
addr = packet.get('start_address')
|
|
count = packet.get('num_registers')
|
|
if addr is not None and count is not None:
|
|
print(f" -> Device {device_id} REQUESTS to READ {count} coils starting from {addr}")
|
|
else:
|
|
print(f" -> Device {device_id} RESPONDS with coil status")
|
|
|
|
elif packet.get('modbus_function') == 5: # Write Single Coil
|
|
addr = packet.get('register_address')
|
|
value = packet.get('register_value')
|
|
if addr is not None and value is not None:
|
|
coil_state = "ON" if value == 0xFF00 else "OFF"
|
|
print(f" -> Device {device_id} WRITES coil {addr}: {coil_state} (0x{value:04X})")
|
|
|
|
else:
|
|
print(f" -> Device {device_id}: Function not fully decoded")
|
|
|
|
else:
|
|
print(f" -> Non-Modbus traffic (skipped)")
|
|
|
|
print(f" --- End Analysis Packet {packet_num} ---")
|
|
|
|
def _parse_packet(self, data: bytes) -> Optional[Dict]:
|
|
"""Parse individual packet data"""
|
|
try:
|
|
# Skip Ethernet header (14 bytes)
|
|
if len(data) < 14:
|
|
return None
|
|
|
|
ip_start = 14
|
|
|
|
# Parse IP header
|
|
if len(data) < ip_start + 20:
|
|
return None
|
|
|
|
ip_header = data[ip_start:ip_start + 20]
|
|
ip_info = struct.unpack('!BBHHHBBH4s4s', ip_header)
|
|
|
|
src_ip = '.'.join(map(str, ip_info[8]))
|
|
dst_ip = '.'.join(map(str, ip_info[9]))
|
|
protocol = ip_info[6]
|
|
|
|
if protocol != 6: # Not TCP
|
|
return None
|
|
|
|
# Parse TCP header
|
|
tcp_start = ip_start + 20
|
|
if len(data) < tcp_start + 20:
|
|
return None
|
|
|
|
tcp_header = data[tcp_start:tcp_start + 20]
|
|
tcp_info = struct.unpack('!HHLLBBHHH', tcp_header)
|
|
|
|
src_port = tcp_info[0]
|
|
dst_port = tcp_info[1]
|
|
|
|
# Check if this might be Modbus (port 502)
|
|
is_modbus = src_port == 502 or dst_port == 502
|
|
|
|
# Parse TCP data
|
|
tcp_data_start = tcp_start + 20
|
|
tcp_data = data[tcp_data_start:]
|
|
|
|
packet_info = {
|
|
'src_ip': src_ip,
|
|
'dst_ip': dst_ip,
|
|
'src_port': src_port,
|
|
'dst_port': dst_port,
|
|
'is_modbus': is_modbus,
|
|
'tcp_data': tcp_data,
|
|
'raw_data': data
|
|
}
|
|
|
|
# Parse Modbus if applicable
|
|
if is_modbus and len(tcp_data) >= 6:
|
|
modbus_info = self._parse_modbus(tcp_data)
|
|
if modbus_info:
|
|
packet_info.update(modbus_info)
|
|
|
|
return packet_info
|
|
|
|
except Exception as e:
|
|
return None
|
|
|
|
def _parse_modbus(self, data: bytes) -> Optional[Dict]:
|
|
"""Parse Modbus TCP data"""
|
|
try:
|
|
if len(data) < 6:
|
|
return None
|
|
|
|
# Modbus TCP header: Transaction ID (2), Protocol ID (2), Length (2)
|
|
trans_id, proto_id, length = struct.unpack('!HHH', data[:6])
|
|
|
|
if proto_id != 0: # Should be 0 for Modbus
|
|
return None
|
|
|
|
if len(data) < 6 + length:
|
|
return None
|
|
|
|
# Modbus PDU starts after the 6-byte header
|
|
modbus_data = data[6:6+length]
|
|
|
|
if len(modbus_data) < 2:
|
|
return None
|
|
|
|
unit_id = modbus_data[0]
|
|
function_code = modbus_data[1]
|
|
|
|
modbus_info = {
|
|
'modbus_trans_id': trans_id,
|
|
'modbus_unit_id': unit_id,
|
|
'modbus_function': function_code,
|
|
'modbus_function_name': self.function_codes.get(function_code, f'Unknown ({function_code})'),
|
|
'modbus_data': modbus_data[2:] if len(modbus_data) > 2 else b'',
|
|
|
|
}
|
|
|
|
# Parse specific function codes with enhanced register value handling
|
|
if function_code == 1 and modbus_info['is_request']: # Read Coils Request
|
|
if len(modbus_data) >= 6:
|
|
start_addr, num_coils = struct.unpack('!HH', modbus_data[2:6])
|
|
modbus_info['start_address'] = start_addr
|
|
modbus_info['num_registers'] = num_coils
|
|
|
|
elif function_code == 3 and modbus_info['is_request']: # Read Holding Registers Request
|
|
if len(modbus_data) >= 6:
|
|
start_addr, num_regs = struct.unpack('!HH', modbus_data[2:6])
|
|
modbus_info['start_address'] = start_addr
|
|
modbus_info['num_registers'] = num_regs
|
|
|
|
elif function_code == 4 and modbus_info['is_request']: # Read Input Registers Request
|
|
if len(modbus_data) >= 6:
|
|
start_addr, num_regs = struct.unpack('!HH', modbus_data[2:6])
|
|
modbus_info['start_address'] = start_addr
|
|
modbus_info['num_registers'] = num_regs
|
|
|
|
elif function_code in [3, 4] and not modbus_info['is_request']: # Read Registers Response
|
|
if len(modbus_data) >= 3:
|
|
byte_count = modbus_data[2]
|
|
if len(modbus_data) >= 3 + byte_count:
|
|
values = []
|
|
for i in range(0, byte_count, 2):
|
|
if i + 1 < byte_count:
|
|
val = struct.unpack('!H', modbus_data[3+i:3+i+2])[0]
|
|
values.append(val)
|
|
modbus_info['register_values'] = values
|
|
|
|
elif function_code == 5: # Write Single Coil
|
|
if len(modbus_data) >= 6:
|
|
coil_addr, coil_value = struct.unpack('!HH', modbus_data[2:6])
|
|
modbus_info['register_address'] = coil_addr
|
|
modbus_info['register_value'] = coil_value
|
|
|
|
elif function_code == 6: # Write Single Register
|
|
if len(modbus_data) >= 6:
|
|
reg_addr, reg_value = struct.unpack('!HH', modbus_data[2:6])
|
|
modbus_info['register_address'] = reg_addr
|
|
modbus_info['register_value'] = reg_value
|
|
|
|
elif function_code == 16 and modbus_info['is_request']: # Write Multiple Registers Request
|
|
if len(modbus_data) >= 7:
|
|
start_addr, num_regs, byte_count = struct.unpack('!HHB', modbus_data[2:7])
|
|
values = []
|
|
for i in range(0, byte_count, 2):
|
|
if 7 + i + 1 < len(modbus_data):
|
|
val = struct.unpack('!H', modbus_data[7+i:7+i+2])[0]
|
|
values.append(val)
|
|
modbus_info['start_address'] = start_addr
|
|
modbus_info['num_registers'] = num_regs
|
|
modbus_info['register_values'] = values
|
|
|
|
elif function_code == 16 and not modbus_info['is_request']: # Write Multiple Registers Response
|
|
if len(modbus_data) >= 6:
|
|
start_addr, num_regs = struct.unpack('!HH', modbus_data[2:6])
|
|
modbus_info['start_address'] = start_addr
|
|
modbus_info['num_registers'] = num_regs
|
|
|
|
return modbus_info
|
|
|
|
except Exception as e:
|
|
return None
|
|
|
|
def convert_to_temperature(self, raw_value: int) -> float:
|
|
"""Convert raw register value to temperature (assuming tenths of degrees)"""
|
|
# Many HVAC systems store temperature as tenths of degrees
|
|
# 238 = 23.8°C
|
|
return raw_value / 10.0
|
|
|
|
def analyze_packets(self, packets: List[Dict]) -> None:
|
|
"""Final summary analysis after real-time parsing"""
|
|
print("\n" + "=" * 80)
|
|
print("FINAL SUMMARY")
|
|
print("=" * 80)
|
|
|
|
devices = set()
|
|
modbus_packets = 0
|
|
temperature_changes = []
|
|
|
|
for i, packet in enumerate(packets):
|
|
if packet.get('is_modbus'):
|
|
modbus_packets += 1
|
|
devices.add(packet['src_ip'])
|
|
devices.add(packet['dst_ip'])
|
|
|
|
# Collect temperature changes for summary
|
|
if packet.get('modbus_function') == 6: # Write Single Register
|
|
value = packet.get('register_value')
|
|
if value:
|
|
temp = self.convert_to_temperature(value)
|
|
if abs(temp - 23.8) < 0.1:
|
|
temperature_changes.append({
|
|
'packet': i+1,
|
|
'temp': temp,
|
|
'raw_value': value,
|
|
'src': packet['src_ip'],
|
|
'dst': packet['dst_ip']
|
|
})
|
|
|
|
elif packet.get('modbus_function') == 16: # Write Multiple Registers
|
|
values = packet.get('register_values', [])
|
|
addr = packet.get('start_address', 0)
|
|
for j, value in enumerate(values):
|
|
temp = self.convert_to_temperature(value)
|
|
if abs(temp - 23.8) < 0.1:
|
|
temperature_changes.append({
|
|
'packet': i+1,
|
|
'temp': temp,
|
|
'raw_value': value,
|
|
'register': addr + j,
|
|
'src': packet['src_ip'],
|
|
'dst': packet['dst_ip']
|
|
})
|
|
|
|
print(f"Total packets parsed: {len(packets)}")
|
|
print(f"Modbus packets found: {modbus_packets}")
|
|
print(f"Unique devices: {', '.join(sorted(devices))}")
|
|
|
|
if temperature_changes:
|
|
print(f"\n*** TEMPERATURE SETPOINT CHANGES TO 23.8°C: {len(temperature_changes)} found ***")
|
|
for i, change in enumerate(temperature_changes, 1):
|
|
print(f" {i}. Packet {change['packet']}: {change['src']} -> {change['dst']}")
|
|
print(f" Temperature: {change['temp']}°C (Raw: {change['raw_value']})")
|
|
if 'register' in change:
|
|
print(f" Register: {change['register']}")
|
|
else:
|
|
print("\n*** NO EXACT 23.8°C TEMPERATURE CHANGES FOUND ***")
|
|
|
|
print("\nAnalysis complete!")
|
|
|
|
def main():
|
|
decoder = ModbusDecoder()
|
|
|
|
try:
|
|
packets = decoder.parse_hex_file('t3.txt')
|
|
print(f"Parsed {len(packets)} packets from capture file")
|
|
decoder.analyze_packets(packets)
|
|
|
|
except FileNotFoundError:
|
|
print("Error: t3.txt file not found!")
|
|
except Exception as e:
|
|
print(f"Error processing file: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|