Networking with Sockets

Use Python's socket module to build TCP/UDP clients and servers, scan ports, grab banners, and understand low-level network communication.

Medium 65m 3 tasks

Prerequisites: Object-Oriented Python

Learning Objectives

→ Create TCP and UDP sockets and understand their differences
→ Connect to remote hosts and send/receive data
→ Build a simple TCP echo server
→ Grab service banners from open ports
→ Handle socket timeouts and connection errors

The Socket Module

A socket is an endpoint for network communication. Python's socket module wraps the POSIX socket API.

import socket

# Address families
# AF_INET  = IPv4
# AF_INET6 = IPv6
# SOCK_STREAM = TCP (reliable, ordered)
# SOCK_DGRAM  = UDP (unreliable, fast)

TCP Client

import socket

def tcp_connect(host, port, message=b"HEAD / HTTP/1.0\r\n\r\n"):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(3)
    try:
        s.connect((host, port))
        s.sendall(message)
        response = s.recv(4096)
        return response.decode(errors="replace")
    except socket.timeout:
        return None
    except ConnectionRefusedError:
        return None
    finally:
        s.close()

result = tcp_connect("example.com", 80)
if result:
    print(result[:200])

Banner Grabbing

import socket

def grab_banner(host, port, timeout=2):
    try:
        s = socket.socket()
        s.settimeout(timeout)
        s.connect((host, port))
        # Some services send a banner immediately
        banner = s.recv(1024).decode(errors="replace").strip()
        s.close()
        return banner
    except Exception:
        return None

for port in [21, 22, 25, 80]:
    banner = grab_banner("scanme.nmap.org", port)
    if banner:
        print(f"Port {port}: {banner[:60]}")

Port Scanner with Threading

import socket
import threading

results = {}
lock = threading.Lock()

def check_port(host, port):
    try:
        s = socket.socket()
        s.settimeout(0.5)
        s.connect((host, port))
        s.close()
        with lock:
            results[port] = "open"
    except OSError:
        with lock:
            results[port] = "closed"

host = "127.0.0.1"
threads = []
for port in range(1, 1025):
    t = threading.Thread(target=check_port, args=(host, port))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

open_ports = sorted(p for p, s in results.items() if s == "open")
print("Open:", open_ports)

TCP Server

import socket

def run_echo_server(host="0.0.0.0", port=9999):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(5)
    print(f"Listening on {host}:{port}")

    try:
        while True:
            client, addr = server.accept()
            print(f"Connection from {addr}")
            data = client.recv(1024)
            client.sendall(data)   # echo back
            client.close()
    except KeyboardInterrupt:
        server.close()

# run_echo_server()

UDP Client

import socket

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(2)
try:
    s.sendto(b"ping", ("8.8.8.8", 53))
    data, addr = s.recvfrom(1024)
    print(f"Response from {addr}: {data[:20]}")
except socket.timeout:
    print("No response")
finally:
    s.close()

socket.create_connection (shortcut)

import socket

# Combines getaddrinfo + connect in one call
with socket.create_connection(("example.com", 80), timeout=3) as s:
    s.sendall(b"GET / HTTP/1.0\r\nHost: example.com\r\n\r\n")
    print(s.recv(512).decode())

Mini Project – Multi-port Banner Grabber

import socket, json

def scan_and_grab(host, ports, timeout=1.5):
    report = {"host": host, "services": {}}
    for port in ports:
        try:
            with socket.create_connection((host, port), timeout=timeout) as s:
                try:
                    banner = s.recv(1024).decode(errors="replace").strip()[:100]
                except socket.timeout:
                    banner = ""
                report["services"][port] = banner or "(no banner)"
        except OSError:
            report["services"][port] = "closed"
    return report

r = scan_and_grab("scanme.nmap.org", [22, 80])
print(json.dumps(r, indent=2))

Use socket.create_connection to connect to example.com:80, send an HTTP GET request, and print the first 200 bytes of the response.

✦ Answer the questions to complete this task

What socket type is used for TCP?

What method sets the socket timeout in seconds?

Implement the banner grabber function. Test it against port 22 of scanme.nmap.org. Print the banner if received.

✦ Answer the questions to complete this task

What does recv(1024) return?

What exception is raised when a socket times out?

Implement the threaded port scanner. Use threading.Lock to safely update the shared results dict. Scan localhost ports 1-1024 and print all open ports.

✦ Answer the questions to complete this task

Why is a Lock needed in the threaded scanner?