Building a Proxy Server from Scratch: Python & Go Tutorial

Building your own proxy server teaches you more about networking than any tutorial. You will understand CONNECT tunneling, request rewriting, connection management, and the subtle differences between HTTP and HTTPS proxying — knowledge that makes you better at using any proxy service.

This guide walks through building a production-capable proxy server in both Python and Go.

What Our Proxy Will Support

HTTP forward proxying (request rewriting)
HTTPS tunneling (CONNECT method)
Basic authentication (username/password)
Access logging
Connection timeouts
Concurrent connections

Python Implementation

Step 1: Basic HTTP Proxy

import asyncio
import base64
from urllib.parse import urlparse

class AsyncProxy:
    """Async HTTP/HTTPS proxy server."""

    def __init__(self, host='0.0.0.0', port=8080, credentials=None):
        self.host = host
        self.port = port
        self.credentials = credentials  # {'user': 'pass'}
        self.request_count = 0

    async def start(self):
        server = await asyncio.start_server(
            self.handle_client,
            self.host,
            self.port,
        )
        print(f"Proxy server running on {self.host}:{self.port}")
        async with server:
            await server.serve_forever()

    async def handle_client(self, reader, writer):
        """Handle incoming client connection."""
        self.request_count += 1
        client_addr = writer.get_extra_info('peername')

        try:
            # Read the request line
            request_line = await asyncio.wait_for(
                reader.readline(), timeout=30
            )
            if not request_line:
                return

            request_str = request_line.decode('utf-8', errors='ignore').strip()

            # Read headers
            headers = {}
            while True:
                line = await asyncio.wait_for(reader.readline(), timeout=10)
                if line == b'\r\n' or line == b'\n' or not line:
                    break
                key, _, value = line.decode('utf-8', errors='ignore').partition(':')
                headers[key.strip().lower()] = value.strip()

            # Authenticate
            if self.credentials:
                if not self._authenticate(headers):
                    response = (
                        b"HTTP/1.1 407 Proxy Authentication Required\r\n"
                        b"Proxy-Authenticate: Basic realm=\"Proxy\"\r\n"
                        b"\r\n"
                    )
                    writer.write(response)
                    await writer.drain()
                    return

            # Parse request
            parts = request_str.split(' ')
            if len(parts) < 3:
                return

            method = parts[0]
            target = parts[1]

            # Log request
            print(f"[{self.request_count}] {client_addr} → {method} {target}")

            if method == 'CONNECT':
                await self._handle_connect(target, reader, writer, headers)
            else:
                await self._handle_http(method, target, reader, writer, headers)

        except asyncio.TimeoutError:
            pass
        except Exception as e:
            print(f"Error: {e}")
        finally:
            writer.close()
            try:
                await writer.wait_closed()
            except Exception:
                pass

    def _authenticate(self, headers):
        """Validate Proxy-Authorization header."""
        auth = headers.get('proxy-authorization', '')
        if not auth.startswith('Basic '):
            return False

        try:
            decoded = base64.b64decode(auth[6:]).decode()
            username, _, password = decoded.partition(':')
            return self.credentials.get(username) == password
        except Exception:
            return False

    async def _handle_http(self, method, target, client_reader, client_writer, headers):
        """Forward HTTP requests (non-CONNECT)."""
        parsed = urlparse(target)
        host = parsed.hostname
        port = parsed.port or 80
        path = parsed.path or '/'
        if parsed.query:
            path += f'?{parsed.query}'

        try:
            # Connect to target server
            target_reader, target_writer = await asyncio.wait_for(
                asyncio.open_connection(host, port),
                timeout=10
            )

            # Build forwarded request (relative path, not absolute)
            request = f"{method} {path} HTTP/1.1\r\n"
            request += f"Host: {host}\r\n"

            # Forward relevant headers (skip proxy-specific ones)
            skip_headers = {'proxy-authorization', 'proxy-connection'}
            for key, value in headers.items():
                if key not in skip_headers:
                    request += f"{key}: {value}\r\n"

            # Add Via header
            request += f"Via: 1.1 proxy\r\n"
            request += "\r\n"

            target_writer.write(request.encode())
            await target_writer.drain()

            # Read and forward body if present
            content_length = int(headers.get('content-length', 0))
            if content_length > 0:
                body = await client_reader.read(content_length)
                target_writer.write(body)
                await target_writer.drain()

            # Forward response back to client
            await self._relay_data(target_reader, client_writer)

            target_writer.close()

        except Exception as e:
            error_response = (
                f"HTTP/1.1 502 Bad Gateway\r\n"
                f"Content-Type: text/plain\r\n"
                f"\r\n"
                f"Proxy error: {str(e)}\r\n"
            )
            client_writer.write(error_response.encode())
            await client_writer.drain()

    async def _handle_connect(self, target, client_reader, client_writer, headers):
        """Handle HTTPS CONNECT tunneling."""
        host, _, port = target.partition(':')
        port = int(port) if port else 443

        try:
            # Connect to target
            target_reader, target_writer = await asyncio.wait_for(
                asyncio.open_connection(host, port),
                timeout=10
            )

            # Tell client the tunnel is established
            client_writer.write(b"HTTP/1.1 200 Connection Established\r\n\r\n")
            await client_writer.drain()

            # Bidirectional relay
            await asyncio.gather(
                self._pipe(client_reader, target_writer),
                self._pipe(target_reader, client_writer),
            )

        except Exception as e:
            error = f"HTTP/1.1 502 Bad Gateway\r\n\r\n{str(e)}\r\n"
            client_writer.write(error.encode())
            await client_writer.drain()

    async def _pipe(self, reader, writer):
        """Pipe data from reader to writer."""
        try:
            while True:
                data = await asyncio.wait_for(reader.read(65536), timeout=60)
                if not data:
                    break
                writer.write(data)
                await writer.drain()
        except (asyncio.TimeoutError, ConnectionError):
            pass

    async def _relay_data(self, reader, writer):
        """Relay all data from reader to writer."""
        try:
            while True:
                data = await asyncio.wait_for(reader.read(65536), timeout=30)
                if not data:
                    break
                writer.write(data)
                await writer.drain()
        except (asyncio.TimeoutError, ConnectionError):
            pass


# Run the proxy
if __name__ == '__main__':
    proxy = AsyncProxy(
        host='0.0.0.0',
        port=8080,
        credentials={'myuser': 'mypassword'}
    )
    asyncio.run(proxy.start())

Testing the Python Proxy

# Start the proxy
python proxy.py

# Test HTTP
curl -x http://myuser:mypassword@localhost:8080 http://httpbin.org/ip

# Test HTTPS
curl -x http://myuser:mypassword@localhost:8080 https://httpbin.org/ip

# Test with Python requests
python -c "
import requests
proxies = {'http': 'http://myuser:mypassword@localhost:8080',
           'https': 'http://myuser:mypassword@localhost:8080'}
r = requests.get('https://httpbin.org/ip', proxies=proxies)
print(r.json())
"

Go Implementation

package main

import (
    "encoding/base64"
    "fmt"
    "io"
    "log"
    "net"
    "net/http"
    "strings"
    "time"
)

type ProxyServer struct {
    Addr        string
    Username    string
    Password    string
    RequestLog  chan string
}

func NewProxyServer(addr, username, password string) *ProxyServer {
    return &ProxyServer{
        Addr:       addr,
        Username:   username,
        Password:   password,
        RequestLog: make(chan string, 1000),
    }
}

func (p *ProxyServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    // Authenticate
    if !p.authenticate(r) {
        w.Header().Set("Proxy-Authenticate", `Basic realm="Proxy"`)
        http.Error(w, "Authentication required", http.StatusProxyAuthRequired)
        return
    }

    // Log request
    log.Printf("%s %s %s", r.RemoteAddr, r.Method, r.URL)

    if r.Method == "CONNECT" {
        p.handleConnect(w, r)
    } else {
        p.handleHTTP(w, r)
    }
}

func (p *ProxyServer) authenticate(r *http.Request) bool {
    if p.Username == "" {
        return true
    }

    auth := r.Header.Get("Proxy-Authorization")
    if !strings.HasPrefix(auth, "Basic ") {
        return false
    }

    decoded, err := base64.StdEncoding.DecodeString(auth[6:])
    if err != nil {
        return false
    }

    parts := strings.SplitN(string(decoded), ":", 2)
    return len(parts) == 2 && parts[0] == p.Username && parts[1] == p.Password
}

func (p *ProxyServer) handleConnect(w http.ResponseWriter, r *http.Request) {
    // Connect to target
    targetConn, err := net.DialTimeout("tcp", r.Host, 10*time.Second)
    if err != nil {
        http.Error(w, err.Error(), http.StatusBadGateway)
        return
    }
    defer targetConn.Close()

    // Hijack the client connection
    hijacker, ok := w.(http.Hijacker)
    if !ok {
        http.Error(w, "Hijacking not supported", http.StatusInternalServerError)
        return
    }

    clientConn, _, err := hijacker.Hijack()
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    defer clientConn.Close()

    // Send 200 to client
    clientConn.Write([]byte("HTTP/1.1 200 Connection Established\r\n\r\n"))

    // Bidirectional copy
    done := make(chan struct{}, 2)
    go func() {
        io.Copy(targetConn, clientConn)
        done <- struct{}{}
    }()
    go func() {
        io.Copy(clientConn, targetConn)
        done <- struct{}{}
    }()
    <-done
}

func (p *ProxyServer) handleHTTP(w http.ResponseWriter, r *http.Request) {
    // Remove proxy-specific headers
    r.Header.Del("Proxy-Authorization")
    r.Header.Del("Proxy-Connection")

    // Forward request
    client := &http.Client{Timeout: 30 * time.Second}
    resp, err := client.Do(r)
    if err != nil {
        http.Error(w, err.Error(), http.StatusBadGateway)
        return
    }
    defer resp.Body.Close()

    // Copy response headers
    for key, values := range resp.Header {
        for _, value := range values {
            w.Header().Add(key, value)
        }
    }

    w.WriteHeader(resp.StatusCode)
    io.Copy(w, resp.Body)
}

func (p *ProxyServer) Start() error {
    log.Printf("Proxy server starting on %s", p.Addr)
    return http.ListenAndServe(p.Addr, p)
}

func main() {
    proxy := NewProxyServer(":8080", "myuser", "mypassword")
    log.Fatal(proxy.Start())
}

Adding Features

Request/Response Logging

import json
import datetime

class RequestLogger:
    def __init__(self, log_file='proxy_access.log'):
        self.log_file = log_file

    def log(self, method, target, status, latency_ms, bytes_transferred):
        entry = {
            'timestamp': datetime.datetime.utcnow().isoformat(),
            'method': method,
            'target': target,
            'status': status,
            'latency_ms': latency_ms,
            'bytes': bytes_transferred,
        }
        with open(self.log_file, 'a') as f:
            f.write(json.dumps(entry) + '\n')

IP Whitelisting

import ipaddress

class IPWhitelist:
    def __init__(self, allowed_ranges):
        self.allowed = [ipaddress.ip_network(r) for r in allowed_ranges]

    def is_allowed(self, client_ip):
        ip = ipaddress.ip_address(client_ip)
        return any(ip in network for network in self.allowed)

Security Hardening

# Production checklist:
# 1. Always require authentication
# 2. Use TLS for client-to-proxy (prevent credential sniffing)
# 3. Set strict timeouts (prevent slowloris)
# 4. Limit request sizes
# 5. Rate limit per user
# 6. Block private IP ranges (prevent SSRF)
# 7. Log all access

BLOCKED_NETWORKS = [
    '10.0.0.0/8',
    '172.16.0.0/12',
    '192.168.0.0/16',
    '127.0.0.0/8',
    '169.254.0.0/16',
]

def is_private_target(host):
    """Prevent SSRF attacks via proxy."""
    import socket
    try:
        ip = socket.gethostbyname(host)
        ip_obj = ipaddress.ip_address(ip)
        return any(
            ip_obj in ipaddress.ip_network(net)
            for net in BLOCKED_NETWORKS
        )
    except socket.gaierror:
        return True  # Block if DNS fails

Internal Links

Proxy Server Architecture — understand the design patterns behind this implementation
TCP/IP Proxy Internals — deeper look at socket-level operations
Self-Hosted Proxy Server Setup — deploy with Squid or HAProxy instead of building custom
Proxy Connection Pooling — add connection reuse to your proxy
Proxy Load Balancing — extend your proxy with load balancing

FAQ

Is it practical to use a custom-built proxy in production?

For personal or small-team use, yes. For production scraping infrastructure, established proxy servers (Squid, HAProxy, Nginx, 3proxy) are more battle-tested and performant. Building from scratch is valuable for learning and for specialized use cases where existing proxies do not meet your needs.

How do I add HTTPS support to the proxy itself?

Wrap the server socket with TLS using an SSL certificate. In Python, use ssl.create_default_context() with asyncio.start_server(). The client then connects to https://proxy:8080 instead of http://proxy:8080, encrypting the proxy credentials and CONNECT requests.

Can I modify HTTPS traffic in my proxy?

Only if you perform TLS interception (MITM). Generate a CA certificate, configure clients to trust it, and re-encrypt traffic on both sides. This is how mitmproxy and Charles Proxy work. Without MITM, CONNECT tunnel traffic is opaque encrypted bytes.

How do I handle WebSocket upgrades in my proxy?

After the HTTP Upgrade handshake, treat the connection like a CONNECT tunnel — bidirectional byte relay. The proxy does not need to understand WebSocket frames, just pipe data between client and server.

What are the performance limits of a Python proxy?

A Python asyncio proxy can handle roughly 1,000-5,000 concurrent connections depending on CPU and I/O patterns. For higher throughput, use Go (50,000+ connections) or C/Rust. Python is suitable for proxy development, testing, and moderate production loads.