This script implements a very basic HTTP/HTTPS proxy.

httpproxy.py

#!/usr/bin/env python3
#
# This script implements a very basic HTTP/HTTPS 1.1 proxy.
#
# Author: Joseph Coffland
# Copyright: 2025, Cauldron Development Oy
# License: GPL 3+

import socket, _thread, select, argparse


class ConnectionHandler:
  def __init__(self, connection, address, timeout, verbose):
    self.client  = connection
    self.timeout = timeout
    self.verbose = verbose
    self.buffer  = b''
    self.method, self.path, self.protocol = self._read_header()

    self.log(f'New connection {self.method}, {self.path}, {self.protocol}')

    if self.method == b'CONNECT':
      self._connect(self.path)
      self.client.sendall(b'HTTP/1.1 200 Connection established\n\n')
      self._transfer()

    else:
      self.path = self.path[7:]
      i = self.path.find(b'/')
      host, path = self.path[:i], self.path[i:]

      self._connect(host)
      self.target.sendall(
        b'%s %s %s\n' % (self.method, path, self.protocol) + self.buffer)
      self._transfer()

    self.client.close()


  def log(self, *args):
    if self.verbose: print(*args)


  def _read_header(self):
    while True:
      self.buffer += self.client.recv(8192)
      end = self.buffer.find(b'\n')
      if end != -1:
        header = self.buffer[:end + 1].split()
        self.buffer = self.buffer[end + 1:]
        return header


  def _connect(self, path):
    i = path.find(b':')
    if i != -1: host, port = path[:i], int(path[i + 1:])
    else: host, port = path, 80

    (family, _, _, _, addr) = socket.getaddrinfo(host, port)[0]
    self.target = socket.socket(family)
    self.target.connect(addr)


  def _transfer(self):
    fds = [self.client, self.target]

    while True:
      sent = False
      recv, _, err = select.select(fds, [], fds, self.timeout)

      if not err and recv:
        for fd in recv:
          data = fd.recv(65536)

          if data:
            out = self.target if fd is self.client else self.client
            out.sendall(data)
            sent = True

      if not sent: break


def main(args):
  sock = socket.socket(socket.AF_INET6 if args.ipv6 else socket.AF_INET)
  sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  sock.bind((args.bind, args.port))

  print(f'Serving HTTP proxy on {args.bind}:{args.port}')
  sock.listen(0)

  while True:
    _thread.start_new_thread(
      ConnectionHandler, sock.accept() + (args.timeout, args.verbose))


parser = argparse.ArgumentParser(description = 'A very basic HTTP(S) 1.1 proxy server')

parser.add_argument('--verbose', '-v', action = 'store_true',
                    help = 'verbose output')
parser.add_argument('--bind', '-b', default = '127.0.0.1',
                    help = 'address to bind')
parser.add_argument('--port', '-p', default = 12347, type = int,
                    help = 'port to bind')
parser.add_argument('--ipv6', '-6', action = 'store_true', help = 'use IPv6')
parser.add_argument('--timeout', '-t', default = 60, type = int,
                    help = 'timeout in seconds')

main(parser.parse_args())