HTTP/HTTPS Protocol Implementation

# raw_http_client.py
import socket

def build_http_request(
    method: str,
    host: str,
    path: str,
    headers: dict | None = None,
    body: str | None = None
) -> bytes:
    """Build a raw HTTP/1.1 request as bytes."""

    body_bytes = body.encode("utf-8") if body else b""

    base_headers = {
        "Host": host,
        "Connection": "close",
        "User-Agent": "PythonRawHTTP/1.0",
    }

    if body_bytes:
        base_headers["Content-Length"] = str(len(body_bytes))
        base_headers["Content-Type"] = "application/json"

    if headers:
        base_headers.update(headers)

    # Build request line + headers
    request_lines = [f"{method} {path} HTTP/1.1"]
    for key, value in base_headers.items():
        request_lines.append(f"{key}: {value}")

    # Headers end with a blank line (\r\n\r\n)
    request = "\r\n".join(request_lines) + "\r\n\r\n"
    return request.encode("utf-8") + body_bytes

# Print what the request looks like on the wire
request = build_http_request("GET", "httpbin.org", "/get")
print("Raw HTTP request bytes:")
print(repr(request[:150]))
print()
print("Formatted:")
print(request.decode())

# raw_http_send.py, continued
import socket

def send_raw_http(host: str, port: int, request: bytes) -> tuple[dict, str]:
    """Send a raw HTTP request and parse the response."""
    with socket.create_connection((host, port), timeout=10) as sock:
        sock.sendall(request)

        # Receive the full response
        response_bytes = b""
        while True:
            chunk = sock.recv(4096)
            if not chunk:
                break
            response_bytes += chunk

    # Split headers and body at \r\n\r\n
    header_section, _, body = response_bytes.partition(b"\r\n\r\n")
    header_lines = header_section.decode("utf-8").split("\r\n")

    # Parse status line
    status_line = header_lines[0]
    print(f"Status: {status_line}")

    # Parse headers into dict
    headers = {}
    for line in header_lines[1:]:
        if ": " in line:
            key, _, value = line.partition(": ")
            headers[key.lower()] = value

    return headers, body.decode("utf-8")

# Make a GET request
request = build_http_request("GET", "httpbin.org", "/get")
headers, body = send_raw_http("httpbin.org", 80, request)

print(f"Content-Type: {headers.get('content-type')}")
print(f"Body (first 200 chars):\n{body[:200]}")

# minimal_http_server.py
import socket
import json
from datetime import datetime, timezone

HOST = "127.0.0.1"
PORT = 8080

ROUTES = {
    "/": ("200 OK", "text/plain", "Welcome to minimal HTTP server!"),
    "/health": ("200 OK", "application/json",
                json.dumps({"status": "ok", "version": "1.0"})),
}

def parse_request(raw: bytes) -> dict:
    """Parse an HTTP request into method, path, headers, body."""
    text, _, body = raw.partition(b"\r\n\r\n")
    lines = text.decode("utf-8").split("\r\n")
    method, path, _ = lines[0].split(" ", 2)
    headers = {}
    for line in lines[1:]:
        if ": " in line:
            k, _, v = line.partition(": ")
            headers[k.lower()] = v
    return {"method": method, "path": path, "headers": headers, "body": body}

def build_response(status: str, content_type: str, body: str) -> bytes:
    body_bytes = body.encode("utf-8")
    response = (
        f"HTTP/1.1 {status}\r\n"
        f"Content-Type: {content_type}\r\n"
        f"Content-Length: {len(body_bytes)}\r\n"
        f"Date: {datetime.now(timezone.utc).strftime('%a, %d %b %Y %H:%M:%S GMT')}\r\n"
        f"\r\n"
    )
    return response.encode("utf-8") + body_bytes

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((HOST, PORT))
    server.listen(5)
    print(f"Server at http://{HOST}:{PORT}")

    while True:
        conn, addr = server.accept()
        with conn:
            raw = conn.recv(4096)
            if not raw:
                continue
            req = parse_request(raw)
            print(f"{req['method']} {req['path']}")

            if req["path"] in ROUTES:
                status, ct, body = ROUTES[req["path"]]
            else:
                status, ct, body = "404 Not Found", "text/plain", "Not found"

            conn.sendall(build_response(status, ct, body))

# https_client.py
import socket
import ssl

def https_get(host: str, path: str) -> str:
    """Make an HTTPS GET request using ssl.wrap_socket."""
    # Create SSL context with certificate verification
    context = ssl.create_default_context()

    with socket.create_connection((host, 443), timeout=10) as raw_sock:
        # Wrap the socket with TLS
        with context.wrap_socket(raw_sock, server_hostname=host) as tls_sock:
            # Inspect the TLS connection
            cert = tls_sock.getpeercert()
            cipher = tls_sock.cipher()
            print(f"TLS version: {tls_sock.version()}")
            print(f"Cipher: {cipher[0]}")
            print(f"Server CN: {cert['subject'][0][0][1]}")

            # Send HTTP request over TLS
            request = (
                f"GET {path} HTTP/1.1\r\n"
                f"Host: {host}\r\n"
                f"Connection: close\r\n"
                f"\r\n"
            ).encode()
            tls_sock.sendall(request)

            # Receive response
            response = b""
            while chunk := tls_sock.recv(4096):
                response += chunk

    header_section, _, body = response.partition(b"\r\n\r\n")
    status = header_section.split(b"\r\n")[0].decode()
    return status, body.decode()[:200]

status, body = https_get("httpbin.org", "/get")
print(f"\nStatus: {status}")
print(f"Body preview: {body}")