simple_proxy.py

#!/usr/bin/env python3
"""
Simple Blocking HTTP Proxy

A minimal, single-threaded HTTP proxy with no caching.
Handles one request at a time sequentially.

For a non-blocking version with response caching, see proxy.py.

Usage:
    python simple_proxy.py
    python simple_proxy.py --port 9999
    python simple_proxy.py --upstream-port 8000
"""

import argparse
import socket
import sys


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="simple_proxy.py",
        description="Simple blocking HTTP proxy (no caching).",
    )
    p.add_argument("--port", type=int, default=8888,
                   help="Port to listen on (default: 8888)")
    p.add_argument("--upstream-port", type=int, default=80,
                   help="Port to connect to on upstream servers (default: 80)")
    return p.parse_args()


def parse_request(request: str):
    """Parse 'GET /<host>/<path> HTTP/x.x' -> (host, get_request_bytes).

    Raises ValueError on malformed input.
    """
    first_line = request.split("\r\n")[0]
    _method, path, *_ = first_line.split()
    segments = path.lstrip("/").split("/", 1)
    host = segments[0]
    endpoint = "/" + (segments[1] if len(segments) > 1 else "")
    if not host or ("." not in host and host != "localhost"):
        raise ValueError(f"invalid host: {host!r}")
    get_request = (
        b"GET " + endpoint.encode() +
        b" HTTP/1.0\r\nHost: " + host.encode() +
        b"\r\n\r\n"
    )
    return host, get_request


def recv_request(conn: socket.socket) -> bytes:
    """Read a complete HTTP request (terminated by \\r\\n\\r\\n)."""
    data = b""
    while b"\r\n\r\n" not in data:
        chunk = conn.recv(1024)
        if not chunk:
            break
        data += chunk
    return data


def recv_response(sock: socket.socket) -> bytes:
    """Read from sock until the server closes the connection."""
    data = b""
    while True:
        chunk = sock.recv(4096)
        if not chunk:
            break
        data += chunk
    return data


def main() -> None:
    args = parse_args()

    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind(("127.0.0.1", args.port))
    srv.listen(1)

    print(f"Simple proxy listening on 127.0.0.1:{args.port}")
    print(f"URL format: http://localhost:{args.port}/<hostname>/<path>")
    print("Press Ctrl-C to stop.")

    try:
        while True:
            conn, _addr = srv.accept()
            try:
                raw = recv_request(conn)
                if not raw:
                    continue
                host, get_req = parse_request(raw.decode(errors="replace"))
                upstream = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                upstream.connect((host, args.upstream_port))
                upstream.sendall(get_req)
                response = recv_response(upstream)
                upstream.close()
                conn.sendall(response)
            except Exception as exc:
                print(f"[error] {exc}", file=sys.stderr)
            finally:
                conn.close()
    except KeyboardInterrupt:
        print("\nStopped.")
    finally:
        srv.close()


if __name__ == "__main__":
    main()