Listen to this Post
Introduction:
In the realm of Operational Technology (OT) and Industrial Control Systems (ICS), network visibility is the first line of defense. Unlike traditional IT networks, OT environments prioritize availability and integrity over confidentiality, making blind spots exceptionally dangerous. Building a custom TCP port scanner from scratch is not just a Python exercise; it is a fundamental practice in understanding socket-level communication, network reconnaissance principles, and the architecture required to map out critical industrial assets safely.
Learning Objectives:
- Understand the mechanics of TCP socket communication and its application in network discovery.
- Implement threading in Python to optimize the performance of network scans.
- Develop structured reporting capabilities to document exposed services in an OT context.
You Should Know:
1. Building the Core TCP Socket Scanner
The foundation of any port scanner lies in the socket—the endpoint for communication between two machines over a network. In Python, the `socket` library allows us to create a client that attempts to connect to a target IP on a specified port. A successful connection indicates the port is open.
Let’s start with a basic, single-threaded version to understand the logic. This script attempts to connect to a target and checks the response.
import socket
def scan_port(target, port):
try:
Create a socket object (AF_INET for IPv4, SOCK_STREAM for TCP)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
Set a timeout to avoid hanging on closed ports
sock.settimeout(1)
Attempt to connect
result = sock.connect_ex((target, port))
connect_ex returns 0 if successful, otherwise an error code
if result == 0:
print(f"Port {port}: OPEN")
sock.close()
except socket.error:
pass
Example usage
target_ip = "192.168.1.1" Replace with target
for port in range(20, 1025):
scan_port(target_ip, port)
Step-by-Step:
- Step 1: Import the `socket` library.
- Step 2: Define a function `scan_port` that creates a TCP socket.
- Step 3: Use `settimeout(1)` to prevent the script from freezing while waiting for a non-responding port.
- Step 4: `connect_ex()` is used instead of `connect()` because it returns a numeric result (0 for success), allowing us to handle errors cleanly without exceptions.
- Step 5: Loop through a range of ports and call the function. In OT, common ports might include 102 (Siemens S7), 502 (Modbus), or 44818 (EtherNet/IP).
2. Implementing Threaded Architecture for Performance
Single-threaded scanning is impractically slow. Scanning 65,535 ports one by one could take hours. Threading allows us to execute multiple scan attempts concurrently, drastically reducing scan time. However, excessive threads can cause inaccurate results or network congestion; therefore, we use a threading pool.
import socket
import threading
from queue import Queue
target = "10.0.0.1" Example OT device IP
open_ports = []
queue = Queue()
def portscan(port):
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1)
result = sock.connect_ex((target, port))
if result == 0:
open_ports.append(port)
sock.close()
except:
pass
def threader():
while True:
worker = queue.get()
portscan(worker)
queue.task_done()
Create 300 threads
for x in range(300):
t = threading.Thread(target=threader)
t.daemon = True
t.start()
Put ports into the queue
for port in range(1, 65535):
queue.put(port)
queue.join()
print("Open Ports:", sorted(open_ports))
Step-by-Step:
- Step 1: Import `threading` and
queue. The queue will hold the list of ports to be scanned. - Step 2: Define the `portscan` function (similar to before) and a `threader` function that pulls a port from the queue and scans it.
- Step 3: Start 300 daemon threads. These threads will run in the background and execute the `threader` function.
- Step 4: Fill the queue with port numbers (1 to 65535). The threads automatically consume these tasks.
- Step 5: `queue.join()` blocks the main script until all tasks (ports) are processed.
3. Service Identification and Banner Grabbing
Knowing a port is open is useful, but identifying the service running on it is critical for OT security. Many industrial protocols do not require authentication, and grabbing a banner can reveal device make, model, and firmware version. We extend our scan to receive data after connection.
def grab_banner(sock):
try:
Send a generic probe; for specific protocols, you'd send specific payloads
sock.send(b"\r\n")
banner = sock.recv(1024).decode().strip()
return banner
except:
return "No Banner"
def portscan(port):
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2)
result = sock.connect_ex((target, port))
if result == 0:
banner = grab_banner(sock)
print(f"Port {port}: OPEN - Banner: {banner}")
sock.close()
except:
pass
Step-by-Step:
- Step 1: Create a `grab_banner` function that sends a newline command (
\r\n) and listens for a response. - Step 2: After a successful connection, call this function to retrieve and print the banner.
- Step 3: In OT environments, sending a generic probe might crash legacy devices. Always ensure you have authorization and are using safe scanning techniques.
4. Structured Reporting (TXT/CSV)
A penetration tester’s output is only as good as their report. For documentation and compliance, results must be saved in a structured format. We will modify the script to write findings to a CSV file, which can be imported into spreadsheets or reporting tools.
import csv
def save_results(target, open_ports_with_banners):
filename = f"scan_results_{target}.csv"
with open(filename, 'w', newline='') as file:
writer = csv.writer(file)
writer.writerow(["Port", "Banner"])
for port, banner in open_ports_with_banners:
writer.writerow([port, banner])
print(f"Results saved to {filename}")
Modify the main logic to store results
open_ports_with_banners = []
Inside portscan, append (port, banner) to the list
Step-by-Step:
- Step 1: Import the `csv` module.
- Step 2: Define `save_results` to create a CSV file with headers “Port” and “Banner”.
- Step 3: Replace the simple list `open_ports` with a list of tuples containing port numbers and their banners.
- Step 4: Call `save_results` after the scan completes.
5. OT-Specific Considerations: Ping Sweep and Network Segmentation
Before port scanning, it is essential to identify live hosts. A ping sweep (ICMP echo requests) helps map the network segment without aggressive probing. On Linux, this can be done with a simple bash script or using Python’s `ping` library.
Linux Command Line Example:
Ping sweep a /24 subnet
for i in {1..254}; do
ping -c 1 -W 1 192.168.1.$i | grep "64 bytes" | cut -d " " -f 4 | tr -d ":" &
done
Python Equivalent:
import subprocess
import ipaddress
def ping_host(ip):
try:
Using subprocess to call system ping
output = subprocess.run(['ping', '-c', '1', '-W', '1', str(ip)], stdout=subprocess.DEVNULL)
return output.returncode == 0
except:
return False
network = ipaddress.ip_network("192.168.1.0/24", strict=False)
for ip in network.hosts():
if ping_host(ip):
print(f"{ip} is alive")
Why this matters in OT: Many industrial devices are sensitive to port scans. Starting with a passive ICMP sweep identifies targets and reduces the risk of crashing a PLC (Programmable Logic Controller) with aggressive TCP traffic.
6. Analyzing Results with Shodan and Censys Integration
Once you have identified open ports and banners, cross-referencing this data with public intelligence databases like Shodan can provide context on whether the device is known to be vulnerable. While Shodan has an API, you can also manually query it.
Command Line Lookup:
Using curl to query Shodan CLI (requires API key) shodan host <target_ip>
This step is crucial in OT assessments to determine if exposed industrial protocols (like Modbus, DNP3, or BACnet) are also visible on the public internet, indicating a severe segmentation failure.
What Undercode Say:
- Foundational Skills Trump Tools: While commercial scanners like Nessus or Nmap are powerful, building a custom scanner ingrains a deep understanding of the TCP/IP stack and socket programming, which is essential when dealing with non-standard or legacy OT protocols that off-the-shelf tools might miss.
- Visibility is the Precursor to Security: In OT, you cannot protect what you cannot see. This exercise highlights the importance of network segmentation. If a Python script running on a laptop can discover and connect to a PLC, so can an attacker. The primary takeaway is that proper network segmentation and firewalls should render such scanning ineffective between zones.
- Performance vs. Safety: The threading code demonstrates speed optimization, but in OT, speed must be balanced with safety. A high-speed scan can cause a denial-of-service on older industrial equipment. Understanding the code allows the pentester to tune delays and concurrency to safe levels.
Prediction:
As IT/OT convergence accelerates, we will see a rise in cyber-physical attacks targeting exposed industrial protocols. The skills demonstrated in this post—specifically, the ability to craft custom network reconnaissance tools—will become mandatory for red teamers and blue teamers alike. Future OT security frameworks will likely mandate regular “discovery scans” using lightweight, custom tools rather than relying solely on agent-based visibility, ensuring that the network map is continuously validated against the reality of active, talking devices.
▶️ Related Video (80% Match):
🎯Let’s Practice For Free:
IT/Security Reporter URL:
Reported By: Onurkeskien Otsecurity – Hackers Feeds
Extra Hub: Undercode MoN
Basic Verification: Pass ✅
