Networking

Networking#

Python’s networking surface end-to-end, from raw packet manipulation up through WebSockets. The stdlib covers the basics (socket, ssl, urllib, asyncio); third-party libraries (httpx, websockets, paramiko, scapy, dnspython) cover the rest.

For the offensive / defensive operations these libraries support, see Operations. For the underlying Linux networking primitives, see Networking.

Packet manipulation (scapy)#

Below the socket abstraction. scapy lets the operator craft, send, sniff, and dissect packets at L2/L3/L4. Required for raw recon, fingerprinting, spoofing, custom protocol probes, and replay.

from scapy.all import IP, TCP, sr1, srp, send, sniff, Ether, ARP

# craft and send a single TCP SYN, return reply
reply = sr1(IP(dst="192.0.2.1") / TCP(dport=80, flags="S"),
            timeout=2, verbose=0)
reply.show()

# ARP scan a subnet (L2)
ans, _ = srp(Ether(dst="ff:ff:ff:ff:ff:ff") /
             ARP(pdst="192.0.2.0/24"),
             timeout=2, verbose=0)
for snd, rcv in ans:
    print(rcv.psrc, rcv.hwsrc)

# passive sniff
def handle(pkt):
    if pkt.haslayer(TCP):
        print(pkt.summary())
sniff(filter="tcp port 80", prn=handle, count=20)

Requires root (or CAP_NET_RAW) for raw sockets.

Sockets#

The stdlib socket module covers TCP, UDP, and Unix sockets.

TCP client#

import socket

with socket.create_connection(("example.com", 80), timeout=5) as s:
    s.sendall(b"GET / HTTP/1.0\r\nHost: example.com\r\n\r\n")
    data = b""
    while chunk := s.recv(4096):
        data += chunk
print(data[:200])

TCP server#

import socket

srv = socket.create_server(("0.0.0.0", 9000), reuse_port=True)
while True:
    conn, addr = srv.accept()
    with conn:
        data = conn.recv(4096)
        conn.sendall(b"hello, " + data)

UDP#

import socket

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.sendto(b"ping", ("192.0.2.1", 12345))
data, addr = s.recvfrom(1500)

Unix sockets#

import socket

srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
srv.bind("/tmp/op.sock")
srv.listen()

TLS#

ssl wraps a socket. SSLContext is the configuration object; reuse it across connections.

import socket, ssl

ctx = ssl.create_default_context()
ctx.minimum_version = ssl.TLSVersion.TLSv1_2

with socket.create_connection(("example.com", 443), timeout=5) as raw:
    with ctx.wrap_socket(raw, server_hostname="example.com") as s:
        print(s.version())
        cert = s.getpeercert()
        s.sendall(b"GET / HTTP/1.0\r\nHost: example.com\r\n\r\n")
        print(s.recv(2048)[:200])

For server-side, ctx.load_cert_chain(certfile, keyfile) and wrap_socket(sock, server_side=True).

HTTP clients#

Three weight classes.

urllib#

Stdlib, no install needed. Good for one-off scripts on minimal targets.

from urllib.request import Request, urlopen
req = Request("https://example.com/", headers={"User-Agent": "op/1"})
with urlopen(req, timeout=5) as r:
    print(r.status, r.read()[:200])

requests#

The classic sync client. Battle-tested ergonomic API.

import requests
r = requests.get("https://api.example.com/users/1", timeout=5)
r.raise_for_status()
user = r.json()

httpx#

Modern default, sync and async, HTTP/1.1 and HTTP/2.

import httpx, asyncio

r = httpx.get("https://example.com/", timeout=5)
r.raise_for_status()

async def main():
    async with httpx.AsyncClient(http2=True, timeout=5) as c:
        rs = await asyncio.gather(
            c.get("https://example.com/"),
            c.get("https://example.org/"),
        )
        for r in rs:
            print(r.status_code, r.url)

asyncio.run(main())

HTTP servers#

http.server#

Stdlib one-shot dev server. Never use in production.

$ python -m http.server 8000

WSGI / ASGI#

Production frameworks (Flask, Django, FastAPI) are WSGI or ASGI apps; they need a server like gunicorn or uvicorn to host. See Frameworks for the framework layer.

Bare ASGI app#

# app.py
async def app(scope, receive, send):
    assert scope["type"] == "http"
    await send({"type": "http.response.start",
                "status": 200,
                "headers": [(b"content-type", b"text/plain")]})
    await send({"type": "http.response.body", "body": b"hello"})

# serve with:  uvicorn app:app

DNS#

Stdlib gives getaddrinfo and gethostbyname for simple forward and reverse lookups. dnspython covers everything else.

import socket
socket.getaddrinfo("example.com", None)

import dns.resolver, dns.reversename

for rr in dns.resolver.resolve("example.com", "A"):
    print(rr.to_text())
for rr in dns.resolver.resolve("example.com", "MX"):
    print(rr.preference, rr.exchange)

rev = dns.reversename.from_address("192.0.2.1")
for rr in dns.resolver.resolve(rev, "PTR"):
    print(rr.to_text())

For zone transfers, EDNS, DNSSEC validation, see the dnspython docs.

SSH#

paramiko#

The classic synchronous SSH client.

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect("target.example.com", username="operator",
               key_filename="/home/operator/.ssh/id_ed25519",
               timeout=10)
stdin, stdout, stderr = client.exec_command("uname -a")
print(stdout.read().decode())
client.close()

asyncssh#

Async SSH client. Use when running commands against many hosts concurrently.

import asyncio, asyncssh

async def run(host):
    async with asyncssh.connect(host) as conn:
        result = await conn.run("uname -a", check=True)
        return host, result.stdout.strip()

async def main(hosts):
    return await asyncio.gather(*(run(h) for h in hosts))

asyncio.run(main(["a.example.com", "b.example.com"]))

Async I/O#

asyncio is the operator’s event loop for concurrent network work. The two patterns to know, gather for fan-out and Semaphore for bounded concurrency.

import asyncio, httpx

async def fetch(client, url):
    r = await client.get(url, timeout=5)
    return url, r.status_code

async def main(urls, limit=50):
    sem = asyncio.Semaphore(limit)
    async def bounded(c, u):
        async with sem:
            return await fetch(c, u)
    async with httpx.AsyncClient() as c:
        return await asyncio.gather(*(bounded(c, u) for u in urls))

results = asyncio.run(main(open("urls.txt").read().splitlines()))

Asyncio TCP server#

import asyncio

async def handle(reader, writer):
    data = await reader.read(4096)
    writer.write(b"echo: " + data)
    await writer.drain()
    writer.close()
    await writer.wait_closed()

async def main():
    server = await asyncio.start_server(handle, "0.0.0.0", 9000)
    async with server:
        await server.serve_forever()

asyncio.run(main())

WebSockets#

The websockets library is the default for both client and server.

import asyncio, websockets

async def client():
    async with websockets.connect("wss://stream.example.com/v1") as ws:
        await ws.send('{"subscribe": "trades"}')
        async for message in ws:
            print(message)

async def handle(ws):
    async for msg in ws:
        await ws.send(f"echo: {msg}")

async def server():
    async with websockets.serve(handle, "0.0.0.0", 9000):
        await asyncio.Future()

Safety#

Warning

SSRF. When a URL is user-supplied, urllib.request.urlopen accepts file://, ftp://, and any IP including 127.0.0.1 and metadata services (169.254.169.254 on AWS, GCP, Azure). Pin the scheme and resolve and validate the IP before fetching.

Warning

DNS rebinding. A URL that resolves to a public IP at parse time can resolve to 127.0.0.1 at connect time. Validate by reading the connected peer (not the input URL) before trusting the response.

Warning

TLS verify-off. Never pass verify=False / CERT_NONE in production code; it disables certificate verification and accepts any cert. Use verify=ca_bundle to pin a CA when needed.

References#

Contents