How to Build a Multi-threaded Port Scanner in Python

Learn how to create a fast, concurrent TCP port scanner in Python using threading and sockets to efficiently detect open ports on a target host.

Port scanning is a fundamental technique used by system administrators to verify security policies and by security professionals to identify open services on a network. While tools like Nmap exist, building your own scanner in Python is an excellent way to understand TCP/IP networking, socket programming, and concurrency.

This guide covers how to build a command-line port scanner that uses multi-threading to scan ranges of ports efficiently.

Understanding the Mechanism

A port scanner works by attempting to establish a TCP handshake with a specific port on a target IP.

Open Port: The target responds with a SYN-ACK packet. The connection succeeds.
Closed/Filtered Port: The target responds with RST (Reset) or drops the packet (Timeout).

Scanning ports one by one (sequentially) is incredibly slow. To make our tool practical, we will use threading to scan hundreds of ports simultaneously.

Step 1: Establishing TCP Connections

We use Python's built-in socket library. The key is to use connect_ex instead of connect.

The Connection Logic:

from socket import AF_INET, SOCK_STREAM, socket

def basic_check(ip, port):
    sock = socket(AF_INET, SOCK_STREAM)
    
    try:
        sock.connect((ip, port))
        return True
    except:
        return False
    finally:
        sock.close()

def robust_check(ip, port):
    with socket(AF_INET, SOCK_STREAM) as sock:
        sock.settimeout(1) 
        
        result = sock.connect_ex((ip, port))
        if result == 0:
            return True
        return False

Step 2: Implementing Concurrency with Queues

To handle threading safely, we use a queue.Queue. The main thread fills the queue with port numbers, and worker threads consume them.

from queue import Queue
import threading

def worker(target_ip, queue):
    while not queue.empty():
        port = queue.get()
        
        if tcp_test(port, target_ip):
            print(f"Opened Port: {port}")
            
        queue.task_done()

Step 3: Handling Command Line Arguments

We want to run the tool like this: python scanner.py 127.0.0.1 5000-9000. We use argparse to handle inputs cleanly.

import argparse

parser = argparse.ArgumentParser(description='TCP Port Scanner')
parser.add_argument('host', help='Target Host IP')
parser.add_argument('ports', help='Port range (e.g., 50-100)')
args = parser.parse_args()

start_port, end_port = map(int, args.ports.split('-'))

Complete Source Code

Here is the complete port_scanner.py script combining sockets, threading, and argument parsing.

import argparse
import threading
from queue import Queue
from socket import AF_INET, gethostbyname, socket, SOCK_STREAM
import sys

def tcp_test(port: int, target_ip: str) -> bool:
    """
    Attempts to connect to a specific port.
    Returns True if the port is open, False otherwise.
    """
    with socket(AF_INET, SOCK_STREAM) as sock:
        sock.settimeout(1)
        result = sock.connect_ex((target_ip, port))
        if result == 0:
            return True
        return False

def worker(target_ip: str, queue: Queue) -> None:
    """
    Thread worker function to process ports from the queue.
    """
    while not queue.empty():
        port = queue.get()
        if tcp_test(port, target_ip):
            print(f"Opened Port: {port}")
        queue.task_done()

def main(host: str, start_port: int, end_port: int) -> None:
    print(f"Scanning {host} from port {start_port} to {end_port}...")
    
    try:
        target_ip = gethostbyname(host)
    except Exception as e:
        print(f"Error resolving host: {e}")
        return

    queue = Queue()
    
    for port in range(start_port, end_port + 1):
        queue.put(port)

    thread_list = []
    for _ in range(100):
        t = threading.Thread(target=worker, args=(target_ip, queue,))
        t.daemon = True
        t.start()
        thread_list.append(t)

    queue.join()
    print("Scanning completed.")

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Multi-threaded TCP Port Scanner')
    parser.add_argument('host', help='Host to scan (e.g., 127.0.0.1)')
    parser.add_argument('ports', help='Port range to scan (e.g., 5000-9000)')
    
    args = parser.parse_args()

    try:
        start_port, end_port = map(int, args.ports.split('-'))
        main(args.host, start_port, end_port)
    except ValueError:
        print("Invalid port format. Use format: start-end (e.g., 20-80)")

Running the Scanner

To run the scanner, execute the following command in your terminal:

python port_scanner.py 127.0.0.1 8000-8100

Output:

Scanning 127.0.0.1 from port 8000 to 8100...
Opened Port: 8081
Scanning completed.

Conclusion

You have successfully built a multi-threaded network tool.

Sockets: You used connect_ex for non-raising connection tests.
Concurrency: You implemented a Queue based worker system to speed up the process significantly.
CLI: You created a user-friendly interface using argparse.