HomeAssistant_Integrations/modbusanalysis
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
constantin
2026-04-06 15:55:57 +03:00
commit f9310b340d
3 changed files with 1741 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modbus_decoder.py
+508
View File
@@ -0,0 +1,508 @@
#!/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()