1. Introduction

In cybersecurity, network reconnaissance is the initial phase of any security audit. Security teams cannot protect assets they do not know exist. A port scanner serves as the foundational scanning engine to identify active hosts, open ports, and vulnerable services listening across network nodes.

I built this project to move beyond simply executing black-box network tools like Nmap. My goal was to understand the precise mechanics of socket handshakes, manage concurrent execution threads, and write a script to inspect and secure target hosts.

Cybersecurity Problem Addressed: Attack Surface Management. Rogue services, shadow IT, or forgotten open ports (such as database listeners or remote desktop ports) are major vectors for initial compromise. By mapping out open ports and banner versions, teams can identify vulnerabilities and enforce defensive firewalls.

2. Objectives

The objective of this project was to develop a professional-grade command-line tool that accomplishes the following:

3. Technologies Used

The scanner is written purely in Python to avoid third-party dependencies:

4. How It Works

The architecture follows a pipeline designed for optimal speeds and reliable banner harvesting:

Step-by-step workflow:

  1. Argument Parsing: The tool takes target IP/domains, ports range, timeout settings, and output options from the terminal.
  2. Hostname Resolution: Resolves target domains into IPv4 addresses using DNS lookups.
  3. Multi-threaded Scanning Loop: Submits scanning jobs across a ThreadPool. Each worker attempts to initialize a TCP socket connection (socket.AF_INET, socket.SOCK_STREAM).
  4. Banner Grabbing: If connect_ex() returns 0, the socket sends an introductory handshake bytes string (e.g., b"Hello\r\n") and reads the responsive banner payload.
  5. Output & Export: Output tables are printed in colored ANSI terminal layouts and written to a JSON file if configured.

5. Implementation

The script is split into two modules: scanner.py (the scanning engine class) and main.py (the command-line interface helper).

scanner.py 
import socket
import concurrent.futures
from typing import List, Dict, Any, Optional

COMMON_PORTS: Dict[int, str] = {
    21: "FTP", 22: "SSH", 23: "Telnet", 25: "SMTP", 53: "DNS",
    80: "HTTP", 110: "POP3", 143: "IMAP", 443: "HTTPS", 445: "SMB",
    3306: "MySQL", 3389: "RDP", 8080: "HTTP-Proxy"
}

class PortScanner:
    def __init__(self, target: str, timeout: float = 1.0, grab_banner: bool = True):
        self.target = target
        self.timeout = timeout
        self.grab_banner = grab_banner
        self.target_ip = self._resolve_target(target)

    def _resolve_target(self, target: str) -> str:
        try:
            return socket.gethostbyname(target)
        except socket.gaierror:
            raise ValueError(f"Could not resolve hostname: {target}")

    def scan_port(self, port: int) -> Optional[Dict[str, Any]]:
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(self.timeout)
                result = s.connect_ex((self.target_ip, port))
                if result == 0:
                    service = COMMON_PORTS.get(port, "Unknown")
                    banner = ""
                    if self.grab_banner:
                        try:
                            s.sendall(b"Hello\r\n")
                            banner = s.recv(1024).decode('utf-8', errors='ignore').strip()
                        except Exception:
                            pass
                    return {
                        "port": port,
                        "status": "open",
                        "service": service,
                        "banner": banner
                    }
        except Exception:
            pass
        return None

    def scan_range(self, start_port: int, end_port: int, threads: int = 100) -> List[Dict[str, Any]]:
        results = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=threads) as executor:
            futures = {executor.submit(self.scan_port, port): port for port in range(start_port, end_port + 1)}
            for future in concurrent.futures.as_completed(futures):
                res = future.result()
                if res:
                    results.append(res)
        return sorted(results, key=lambda x: x["port"])
main.py 
import argparse
import sys
import json
import time
from scanner import PortScanner

def main():
    parser = argparse.ArgumentParser(description="Multi-threaded Port Scanner")
    parser.add_argument("target", help="Target host")
    parser.add_argument("-sp", "--start-port", type=int, default=1)
    parser.add_argument("-ep", "--end-port", type=int, default=1024)
    parser.add_argument("-t", "--threads", type=int, default=100)
    parser.add_argument("--timeout", type=float, default=1.0)
    parser.add_argument("--no-banner", action="store_true")
    parser.add_argument("-o", "--output")

    args = parser.parse_args()

    try:
        scanner = PortScanner(args.target, timeout=args.timeout, grab_banner=not args.no_banner)
        print(f"[*] Target Host: {args.target} ({scanner.target_ip})")
        
        start_time = time.time()
        results = scanner.scan_range(args.start_port, args.end_port, threads=args.threads)
        elapsed = time.time() - start_time
        
        print("\nPORT      STATE     SERVICE    BANNER")
        print("-" * 50)
        for res in results:
            print(f"{res['port']:<10}open      {res['service']:<10} {res['banner']}")
            
        print(f"\n[*] Scan completed in {elapsed:.2f} seconds. Found {len(results)} open ports.")
        
        if args.output:
            with open(args.output, "w") as f:
                json.dump(results, f, indent=4)
    except Exception as e:
        print(f"[-] Error: {e}")

if __name__ == "__main__":
    main()

6. Testing

The tool was verified against an active testing virtual machine in a lab sandbox: 

python main.py 127.0.0.1 -sp 1 -ep 1024 -t 100 -o scan_results.json

Terminal Scan Logs

[*] Target Host: 127.0.0.1 (127.0.0.1)
[*] Port Range : 1 - 1024
[*] Threads    : 100
[*] Scanning in progress...

PORT      STATE     SERVICE    BANNER
--------------------------------------------------
22/tcp    open      SSH        SSH-2.0-OpenSSH_8.9p1 Ubuntu-3
80/tcp    open      HTTP       Apache/2.4.52 (Ubuntu)
443/tcp   open      HTTPS      
--------------------------------------------------
[*] Scan completed in 1.45 seconds.
[*] Total open ports found: 3
[+] Results saved to scan_results.json

7. Security Use Case

How does building a custom scanner align with industry operations?

8. Challenges

The Problem: When I initially integrated banner grabbing, threads started hanging. If a port was open but the service did not respond to the probe payload (e.g., firewall drop or non-interactive service), the thread block lasted until the timeout expired, which delayed the entire scan pipeline.

The Solution: I configured separate timeout boundaries. I set socket timeouts strictly to 0.5s for connections, and encapsulated the socket read process in a robust try-except wrapper. By limiting the bytes buffer size and applying strict read deadlines, I ensured that slow/silent services did not interrupt the execution flow of the main script.

9. Improvements

Future additions to make the utility more robust:

10. Conclusion

Building this tool manually enhanced my knowledge of standard socket connections and the low-level TCP handshake sequence. It demonstrated how easy it is to spin up custom network probes, and highlighted the importance of securing open services and maintaining strong firewalls.

11. GitHub

The complete project repository containing source codes, documentation, and configuration files is hosted publicly on GitHub:

Port Scanner Repository