Networks and the Web in Python

def application(data):
    return "GET /index.html | " + data

def transport(segment, src_port, dst_port, seq):
    return "TCP %d>%d seq=%d | " % (src_port, dst_port, seq) + segment

def internet(packet, src_ip, dst_ip):
    return "IP %s>%s ttl=64 | " % (src_ip, dst_ip) + packet

def link(frame, src_mac, dst_mac):
    return "ETH %s>%s | " % (src_mac, dst_mac) + frame

# going down the stack at the sender
message = application("Host: example.org")
segment = transport(message, 50312, 80, 1)
packet = internet(segment, "192.168.1.20", "203.0.113.7")
frame = link(packet, "aa:01", "bb:02")
print(frame)
print()

# going up the stack at the receiver
for layer in ("link", "internet", "transport", "application"):
    header, _, frame = frame.partition(" | ")
    print("the", layer, "layer reads:", header)
print("the program receives:", frame)

import random

message = "Packets can arrive out of order, so every one carries a sequence number."
size = 14
packets = [{"seq": n, "data": message[i:i + size]} for n, i in enumerate(range(0, len(message), size))]

arrived = list(packets)
random.Random(3).shuffle(arrived)                       # the network scrambles them
print("arrival order:", [p["seq"] for p in arrived])

rebuilt = "".join(p["data"] for p in sorted(arrived, key=lambda p: p["seq"]))
print(rebuilt == message)

servers = {
    "root": {"org": "the .org server"},
    "the .org server": {"example.org": "the example.org name server"},
    "the example.org name server": {"www.example.org": "203.0.113.7"},
}

def resolve(name):
    server = "root"
    while True:
        answers = servers[server]
        match = next(key for key in answers if name == key or name.endswith("." + key))
        reply = answers[match]
        print(server, "says:", match, "->", reply)
        if reply[0].isdigit():
            return reply
        server = reply

print("address:", resolve("www.example.org"))

def to_int(dotted):
    a, b, c, d = (int(part) for part in dotted.split("."))
    return (a << 24) | (b << 16) | (c << 8) | d

def to_dotted(number):
    return ".".join(str((number >> shift) & 255) for shift in (24, 16, 8, 0))

ip = to_int("192.168.5.130")
mask = to_int("255.255.255.192")             # also written /26: 26 one bits

network = ip & mask                           # keep only the network part
broadcast = network | (~mask & 0xFFFFFFFF)    # set every host bit to 1
host_bits = 32 - bin(mask).count("1")

print(bin(ip)[2:].zfill(32), "address")
print(bin(mask)[2:].zfill(32), "mask")
print(bin(network)[2:].zfill(32), "address AND mask")
print()
print("network address:  ", to_dotted(network))
print("broadcast address:", to_dotted(broadcast))
print("host bits:", host_bits, "-> usable hosts:", 2 ** host_bits - 2)

def same_network(a, b):
    return to_int(a) & mask == to_int(b) & mask

print(same_network("192.168.5.130", "192.168.5.140"))
print(same_network("192.168.5.130", "192.168.5.20"))

import ipaddress

network = ipaddress.ip_network("192.168.5.130/26", strict=False)
print(network, network.netmask, network.broadcast_address, network.num_addresses - 2)
print()

for subnet in ipaddress.ip_network("192.168.5.0/24").subnets(new_prefix=26):
    hosts = list(subnet.hosts())
    print(subnet, "hosts", hosts[0], "to", hosts[-1])

print()
print(ipaddress.ip_address("10.4.3.2").is_private, ipaddress.ip_address("8.8.8.8").is_private)

def checksum(data):
    return sum(data.encode()) % 256

def make_channel(plan):
    count = [0]
    def channel(packet):
        count[0] += 1
        fate = plan.get(count[0])
        if fate == "lost":
            return None
        if fate == "damaged":
            seq, data, check = packet
            return (seq, "X" + data[1:], check)
        return packet
    return channel, count

def send_all(words, channel):
    delivered = []
    for seq, data in enumerate(words):
        packet = (seq, data, checksum(data))
        while True:
            print("send", seq, data)
            got = channel(packet)
            if got is None:
                print("   no ACK before the timeout: send again")
            elif checksum(got[1]) != got[2]:
                print("   arrived damaged: checksum does not match, discard it")
            else:
                print("   ACK", seq)
                delivered.append(got[1])
                break
    return delivered

channel, count = make_channel({2: "lost", 4: "damaged", 6: "lost"})
delivered = send_all(["one", "two", "three", "four"], channel)
print(delivered)
print(count[0], "transmissions to deliver", len(delivered), "packets")

import math

packets = 100
for window in (1, 4, 16):
    print("window of", window, "->", math.ceil(packets / window), "round trips")

from urllib.parse import urlparse, parse_qs

url = "https://shop.example.org:8443/products/shoes?colour=red&size=7#reviews"
parts = urlparse(url)
print("scheme:  ", parts.scheme)
print("host:    ", parts.hostname)
print("port:    ", parts.port)
print("path:    ", parts.path)
print("query:   ", parse_qs(parts.query))
print("fragment:", parts.fragment)

import json

scores = [{"name": "Ana", "score": 120}, {"name": "Raj", "score": 95}]

def home(request):
    return 200, "text/html", "<h1>Leaderboard</h1>"

def list_scores(request):
    return 200, "application/json", json.dumps(scores)

def add_score(request):
    try:
        entry = json.loads(request["body"])
        entry = {"name": str(entry["name"]), "score": int(entry["score"])}
    except (ValueError, KeyError, TypeError):
        return 400, "application/json", json.dumps({"error": "send a name and a score"})
    scores.append(entry)
    return 201, "application/json", json.dumps(entry)

ROUTES = {("GET", "/"): home, ("GET", "/api/scores"): list_scores, ("POST", "/api/scores"): add_score}
REASONS = {200: "OK", 201: "Created", 400: "Bad Request", 404: "Not Found", 405: "Method Not Allowed"}

def parse_request(text):
    head, _, body = text.partition("\r\n\r\n")
    lines = head.split("\r\n")
    method, path, version = lines[0].split(" ")
    headers = dict(line.split(": ", 1) for line in lines[1:])
    return {"method": method, "path": path, "headers": headers, "body": body}

def serve(raw):
    request = parse_request(raw)
    handler = ROUTES.get((request["method"], request["path"]))
    if handler is not None:
        status, kind, body = handler(request)
    elif any(path == request["path"] for method, path in ROUTES):
        status, kind, body = 405, "text/plain", "That method is not allowed here"
    else:
        status, kind, body = 404, "text/plain", "Nothing at " + request["path"]
    return ("HTTP/1.1 " + str(status) + " " + REASONS[status] + "\r\n"
            "Content-Type: " + kind + "\r\n"
            "Content-Length: " + str(len(body)) + "\r\n\r\n" + body)

def ask(method, path, body=""):
    lines = [method + " " + path + " HTTP/1.1", "Host: scores.example.org"]
    if body:
        lines += ["Content-Type: application/json", "Content-Length: " + str(len(body))]
    reply = serve("\r\n".join(lines) + "\r\n\r\n" + body)
    status_line, _, rest = reply.partition("\r\n")
    print(method, path, "->", status_line, "|", rest.split("\r\n\r\n")[1])

ask("GET", "/")
ask("GET", "/api/scores")
ask("POST", "/api/scores", '{"name": "Mia", "score": 150}')
ask("POST", "/api/scores", "not json")
ask("GET", "/api/scores")
ask("GET", "/nothing")
ask("DELETE", "/api/scores")

import secrets

sessions = {}

def log_in(user):
    token = secrets.token_hex(8)                  # long and unpredictable
    sessions[token] = user
    return "Set-Cookie: session=" + token

header = log_in("amy")
token = header.split("=")[1]
print("token length:", len(token))
print("who is that?", sessions.get(token))
print("a guessed token:", sessions.get("0123456789abcdef"))

# server.py: turns whatever it is sent into upper case
import socket

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:     # TCP
    server.bind(("127.0.0.1", 5000))        # this computer, port 5000
    server.listen()
    print("waiting for a client...")
    connection, address = server.accept()
    with connection:
        print("connected to", address)
        while True:
            data = connection.recv(1024)
            if not data:
                break
            connection.sendall(data.upper())

# client.py
import socket

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
    client.connect(("127.0.0.1", 5000))
    client.sendall(b"hello server")
    print(client.recv(1024))                 # b'HELLO SERVER'

# web.py
import json
from http.server import BaseHTTPRequestHandler, HTTPServer

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/api/hello":
            body = json.dumps({"message": "hello from Python"}).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
        else:
            self.send_error(404)

HTTPServer(("127.0.0.1", 8000), Handler).serve_forever()

# call.py
import json
from urllib.request import urlopen

with urlopen("http://127.0.0.1:8000/api/hello") as response:
    print(response.status)
    print(json.load(response))

pip install flask
python app.py
curl http://127.0.0.1:5000/api/scores

# app.py
from flask import Flask, jsonify, request

app = Flask(__name__)
scores = [{"name": "Ana", "score": 120}]

@app.get("/api/scores")
def list_scores():
    return jsonify(scores)

@app.post("/api/scores")
def add_score():
    entry = request.get_json()
    scores.append(entry)
    return jsonify(entry), 201

if __name__ == "__main__":
    app.run(debug=True)