Build Your Own TCP Scanner in PythonA TCP scanner is a program that probes one or more target IP addresses and ports to determine which TCP ports are open, closed, or filtered. Building your own TCP scanner in Python is an excellent way to learn about networking fundamentals, socket programming, port states, and responsible scanning practices. This article guides you through creating a flexible, efficient TCP scanner in Python — covering design choices, implementation, speed optimizations, and ethical considerations.
What this article covers
- Why build a TCP scanner
- TCP port states and scanning basics
- Design choices and features
- Implementation: synchronous and asynchronous scanners
- Performance optimizations (threading, asyncio, socket tuning)
- Interpreting results and handling errors
- Safety, legality, and ethics
- Extensions and next steps
Why build a TCP scanner
Building your own scanner helps you:
- Understand how TCP handshakes reveal port state.
- Learn socket programming and concurrency models in Python.
- Customize scanning behavior (time-outs, retries, rate-limiting).
- Integrate scanning into larger security tools or monitoring systems.
TCP port states and scanning basics
- Open — service responds to connection attempts (e.g., completes TCP handshake).
- Closed — host responds with RST (reset) indicating no process is listening.
- Filtered — probes time out or ICMP unreachable messages indicate filtering (firewall).
- Unfiltered — reachable but the scanner cannot determine open/closed state without additional techniques.
Common scanning methods:
- TCP connect scan: attempt a full TCP connection using the OS socket API. Simple and reliable; slower and more detectable.
- TCP SYN scan: send SYN packets and analyze replies (requires raw sockets / elevated privileges). Faster and stealthier but more complex.
- TCP ACK/FIN/NULL scans: used for firewall/stack fingerprinting; advanced techniques.
We’ll implement a TCP connect-style scanner (no raw sockets), which works cross-platform and doesn’t require root privileges.
Design choices and features
Core features to implement:
- Target input: single host, CIDR, list of hosts.
- Port range: single port, list, or ranges (e.g., 1–1024, 80,443).
- Concurrency: sequential, threaded, or async.
- Timeouts and retries.
- Rate limiting to avoid overwhelming targets.
- Output: console, CSV, or JSON.
- Optional banner grabbing (read server banner after connect).
- Error handling and logging.
We’ll show three implementations:
- Simple synchronous scanner (for learning).
- Threaded scanner (for speed).
- Asyncio-based scanner (modern, scalable).
Implementation — prerequisites
You’ll need Python 3.7+ (asyncio improvements in 3.7+). Standard library modules used:
- socket
- asyncio
- concurrent.futures (ThreadPoolExecutor)
- ipaddress
- argparse
- csv / json
- time
- ssl (optional for TLS banner)
Example environment:
- Python 3.11 on Linux/macOS/Windows
1) Simple synchronous scanner
This is the most straightforward version: loop through ports and attempt socket.connect_ex.
#!/usr/bin/env python3 import socket import argparse import ipaddress from typing import Iterable def scan_port(host: str, port: int, timeout: float = 1.0) -> bool: """Return True if port is open on host.""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.settimeout(timeout) result = s.connect_ex((host, port)) return result == 0 def scan_host(host: str, ports: Iterable[int], timeout: float = 1.0): open_ports = [] for port in ports: if scan_port(host, port, timeout): open_ports.append(port) return open_ports def parse_ports(port_str: str): ports = set() for part in port_str.split(','): part = part.strip() if '-' in part: a, b = part.split('-', 1) ports.update(range(int(a), int(b) + 1)) else: ports.add(int(part)) return sorted(p for p in ports if 0 < p < 65536) def main(): parser = argparse.ArgumentParser(description="Simple TCP connect scanner") parser.add_argument("target", help="Target IP or hostname") parser.add_argument("-p", "--ports", default="1-1024", help="Ports (e.g. 22,80,443,1000-2000)") parser.add_argument("-t", "--timeout", type=float, default=1.0) args = parser.parse_args() try: addr = socket.gethostbyname(args.target) except Exception as e: print(f"Failed to resolve {args.target}: {e}") return ports = parse_ports(args.ports) open_ports = scan_host(addr, ports, timeout=args.timeout) if open_ports: print(f"Open ports on {addr}: {', '.join(map(str, open_ports))}") else: print(f"No open ports found on {addr} in specified range.") if __name__ == "__main__": main() Notes:
- connect_ex returns 0 on success (open), errno otherwise.
- This version is slow for large ranges because it scans sequentially.
2) Threaded scanner (faster)
Use ThreadPoolExecutor to run many connections in parallel. Choose thread count carefully — too many threads can exhaust resources.
#!/usr/bin/env python3 import socket import argparse from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Iterable def scan_port(host: str, port: int, timeout: float = 1.0) -> tuple[int, bool]: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.settimeout(timeout) res = s.connect_ex((host, port)) return port, (res == 0) def scan_host_threaded(host: str, ports: Iterable[int], timeout: float = 1.0, workers: int = 100): open_ports = [] with ThreadPoolExecutor(max_workers=workers) as ex: futures = {ex.submit(scan_port, host, p, timeout): p for p in ports} for fut in as_completed(futures): port, is_open = fut.result() if is_open: open_ports.append(port) return sorted(open_ports) def main(): parser = argparse.ArgumentParser() parser.add_argument("target") parser.add_argument("-p", "--ports", default="1-1024") parser.add_argument("-t", "--timeout", type=float, default=1.0) parser.add_argument("-w", "--workers", type=int, default=200) args = parser.parse_args() try: addr = socket.gethostbyname(args.target) except Exception as e: print(f"Resolve error: {e}") return ports = parse_ports(args.ports) open_ports = scan_host_threaded(addr, ports, timeout=args.timeout, workers=args.workers) print(f"Open ports: {open_ports}") if __name__ == "__main__": main() Tips:
- For host scanning across many IPs, keep thread count per-host moderate and use higher-level orchestration.
- Use smaller timeouts when many ports are closed (saves time).
3) Asyncio-based scanner (scalable)
Asyncio with asyncio.open_connection is non-blocking and scales well for thousands of concurrent attempts.
#!/usr/bin/env python3 import asyncio import argparse import socket async def scan_port_async(host: str, port: int, timeout: float = 1.0): try: fut = asyncio.open_connection(host, port) reader, writer = await asyncio.wait_for(fut, timeout=timeout) writer.close() try: await writer.wait_closed() except Exception: pass return port, True except (asyncio.TimeoutError, ConnectionRefusedError, OSError): return port, False async def scan_host_async(host: str, ports, timeout=1.0, concurrency=1000): sem = asyncio.Semaphore(concurrency) results = [] async def sem_scan(p): async with sem: return await scan_port_async(host, p, timeout) tasks = [asyncio.create_task(sem_scan(p)) for p in ports] for coro in asyncio.as_completed(tasks): port, is_open = await coro if is_open: results.append(port) return sorted(results) def main(): parser = argparse.ArgumentParser() parser.add_argument("target") parser.add_argument("-p", "--ports", default="1-1024") parser.add_argument("-t", "--timeout", type=float, default=1.0) parser.add_argument("-c", "--concurrency", type=int, default=1000) args = parser.parse_args() try: addr = socket.gethostbyname(args.target) except Exception as e: print(f"Name resolution failed: {e}") return ports = parse_ports(args.ports) open_ports = asyncio.run(scan_host_async(addr, ports, timeout=args.timeout, concurrency=args.concurrency)) print(f"Open ports: {open_ports}") if __name__ == "__main__": main() Notes:
- asyncio.open_connection uses the same non-blocking sockets; it’s efficient but beware of OS limits on concurrent sockets.
- Use a semaphore to throttle concurrency.
Performance considerations and tuning
- Timeouts: shorter timeouts reduce wait on filtered ports but risk false negatives if network latency is high.
- Concurrency: more threads/coroutines increase speed until limited by CPU, memory, or OS socket limits.
- File descriptor limits: increase ulimit -n on Unix systems for large scans (e.g., 10000+).
- Port reuse: ephemeral ports and TIME_WAIT can limit throughput; using many concurrent connects triggers TIME_WAIT on client side. Keep concurrency moderate or use lower-level techniques (SYN scans) if required.
- Network and ethical limits: scan your own assets or get permission; never overwhelm targets.
Interpreting results and advanced detection
- Closed (RST) vs filtered (timeout): record both and consider retrying filtered ports with longer timeout or different probe type.
- Banner grabbing: after successful connect, attempt a small recv to read service banners (HTTP, SMTP, etc.). Many services will send banners; others require you to send protocol-specific bytes.
- Rate-limiting and backoff: implement exponential backoff for transient network errors.
Example banner grab snippet:
with socket.create_connection((host, port), timeout=2) as s: s.settimeout(2) try: data = s.recv(1024) banner = data.decode(errors="replace").strip() except socket.timeout: banner = "" Safety, legality, and ethics
- Always have explicit permission before scanning networks you do not own.
- Scanning without authorization can be illegal and may trigger intrusion detection and blocking.
- Use rate limiting, and avoid scanning critical infrastructure during business hours.
- Log your scans and provide contact info if scanning public ranges for research.
Extensions and next steps
- Implement SYN scan using raw sockets and pcap (requires root and deep networking knowledge).
- Add multi-host scanning (CIDR ranges) and parallelize across hosts.
- Add service detection (send protocol-specific probes).
- Export results to CSV/JSON and create a UI/dashboard.
- Add IPv6 support using socket.AF_INET6 and appropriate parsing.
Example real-world usage
- Quick port check: python tcp_scan.py example.com -p 22,80,443
- Wide range scan with concurrency: python tcp_scan.py 192.168.1.0/24 -p 1-1024 -c 500
- Integrate into monitoring: run periodic scans to detect service outages or unexpected open ports.
Building a TCP scanner in Python teaches valuable networking and security skills. Start with the simple connect scanner, move to threaded/async versions for scale, and always scan responsibly.
Leave a Reply