opera-proxy/handler/handler.go

package handler

import (
	"bufio"
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/Alexey71/opera-proxy/dialer"
	clog "github.com/Alexey71/opera-proxy/log"
)

const (
	COPY_BUF    = 128 * 1024
	BAD_REQ_MSG = "Bad Request\n"

	// Reduced idle pool: the proxy handler makes upstream connections per request,
	// not persistent keep-alive sessions. 10 total / 2 per host is plenty and
	// avoids leaking hundreds of idle goroutines/sockets under bursty traffic.
	TRANSPORT_MAX_IDLE_CONNS          = 10
	TRANSPORT_MAX_IDLE_CONNS_PER_HOST = 2
	TRANSPORT_IDLE_CONN_TIMEOUT       = 60 * time.Second
)

// copyBufPool reuses 128 KiB buffers for bidirectional data relay,
// avoiding per-connection heap allocations.
var copyBufPool = sync.Pool{
	New: func() any {
		b := make([]byte, COPY_BUF)
		return &b
	},
}

type ProxyHandler struct {
	logger        *clog.CondLogger
	dialer        dialer.ContextDialer
	httptransport http.RoundTripper
	fakeSNI       string
}

func NewProxyHandler(d dialer.ContextDialer, logger *clog.CondLogger, fakeSNI string) *ProxyHandler {
	httptransport := &http.Transport{
		MaxIdleConns:          TRANSPORT_MAX_IDLE_CONNS,
		MaxIdleConnsPerHost:   TRANSPORT_MAX_IDLE_CONNS_PER_HOST,
		IdleConnTimeout:       TRANSPORT_IDLE_CONN_TIMEOUT,
		TLSHandshakeTimeout:   10 * time.Second,
		ExpectContinueTimeout: 1 * time.Second,
		DialContext:           d.DialContext,
	}
	return &ProxyHandler{
		logger:        logger,
		dialer:        d,
		httptransport: httptransport,
		fakeSNI:       fakeSNI,
	}
}

func (s *ProxyHandler) HandleTunnel(wr http.ResponseWriter, req *http.Request) {
	ctx := req.Context()
	conn, err := s.dialer.DialContext(ctx, "tcp", req.RequestURI)
	if err != nil {
		s.logger.Error("Can't satisfy CONNECT request: %v", err)
		http.Error(wr, "Can't satisfy CONNECT request", http.StatusBadGateway)
		return
	}

	if req.ProtoMajor == 0 || req.ProtoMajor == 1 {
		// Upgrade client connection
		localconn, rw, err := hijack(wr)
		if err != nil {
			s.logger.Error("Can't hijack client connection: %v", err)
			http.Error(wr, "Can't hijack client connection", http.StatusInternalServerError)
			return
		}
		defer localconn.Close()

		// Inform client connection is built
		fmt.Fprintf(localconn, "HTTP/%d.%d 200 OK\r\n\r\n", req.ProtoMajor, req.ProtoMinor)

		clientReader := io.Reader(localconn)
		if rw != nil && rw.Reader.Buffered() > 0 {
			clientReader = io.MultiReader(rw.Reader, localconn)
		}
		proxy(req.Context(), localconn, clientReader, conn, s.fakeSNI)
	} else if req.ProtoMajor == 2 {
		wr.Header()["Date"] = nil
		wr.WriteHeader(http.StatusOK)
		flush(wr)
		proxyh2(req.Context(), req.Body, wr, conn, s.fakeSNI)
	} else {
		s.logger.Error("Unsupported protocol version: %s", req.Proto)
		http.Error(wr, "Unsupported protocol version.", http.StatusBadRequest)
		return
	}
}

func (s *ProxyHandler) HandleRequest(wr http.ResponseWriter, req *http.Request) {
	req.RequestURI = ""
	if req.ProtoMajor == 2 {
		req.URL.Scheme = "http" // We can't access :scheme pseudo-header, so assume http
		req.URL.Host = req.Host
	}
	resp, err := s.httptransport.RoundTrip(req)
	if err != nil {
		s.logger.Error("HTTP fetch error: %v", err)
		http.Error(wr, "Server Error", http.StatusInternalServerError)
		return
	}
	defer resp.Body.Close()
	s.logger.Info("%v %v %v %v", req.RemoteAddr, req.Method, req.URL, resp.Status)
	delHopHeaders(resp.Header)
	copyHeader(wr.Header(), resp.Header)
	wr.WriteHeader(resp.StatusCode)
	flush(wr)
	copyBody(wr, resp.Body)
}

func (s *ProxyHandler) ServeHTTP(wr http.ResponseWriter, req *http.Request) {
	s.logger.Info("Request: %v %v %v %v", req.RemoteAddr, req.Proto, req.Method, req.URL)

	isConnect := strings.ToUpper(req.Method) == "CONNECT"
	if (req.URL.Host == "" || req.URL.Scheme == "" && !isConnect) && req.ProtoMajor < 2 ||
		req.Host == "" && req.ProtoMajor == 2 {
		http.Error(wr, BAD_REQ_MSG, http.StatusBadRequest)
		return
	}
	delHopHeaders(req.Header)
	if isConnect {
		s.HandleTunnel(wr, req)
	} else {
		s.HandleRequest(wr, req)
	}
}

func proxy(ctx context.Context, left net.Conn, leftReader io.Reader, right net.Conn, fakeSNI string) {
	wg := sync.WaitGroup{}
	ltr := func(dst net.Conn, src io.Reader) {
		defer wg.Done()
		copyWithSNIRewrite(dst, src, fakeSNI)
		dst.Close()
	}
	rtl := func(dst, src net.Conn) {
		defer wg.Done()
		// Grab a pooled buffer for this copy direction.
		bufp := copyBufPool.Get().(*[]byte)
		defer copyBufPool.Put(bufp)
		io.CopyBuffer(dst, src, *bufp)
		dst.Close()
	}
	wg.Add(2)
	go ltr(right, leftReader)
	go rtl(left, right)
	groupdone := make(chan struct{})
	go func() {
		wg.Wait()
		groupdone <- struct{}{}
	}()
	select {
	case <-ctx.Done():
		left.Close()
		right.Close()
	case <-groupdone:
		return
	}
	<-groupdone
	return
}

func proxyh2(ctx context.Context, leftreader io.ReadCloser, leftwriter io.Writer, right net.Conn, fakeSNI string) {
	wg := sync.WaitGroup{}
	ltr := func(dst net.Conn, src io.Reader) {
		defer wg.Done()
		bufp := copyBufPool.Get().(*[]byte)
		defer copyBufPool.Put(bufp)
		io.CopyBuffer(dst, src, *bufp)
		copyWithSNIRewrite(dst, src, fakeSNI)
		dst.Close()
	}
	rtl := func(dst io.Writer, src io.Reader) {
		defer wg.Done()
		copyBody(dst, src)
	}
	wg.Add(2)
	go ltr(right, leftreader)
	go rtl(leftwriter, right)
	groupdone := make(chan struct{}, 1)
	go func() {
		wg.Wait()
		groupdone <- struct{}{}
	}()
	select {
	case <-ctx.Done():
		leftreader.Close()
		right.Close()
	case <-groupdone:
		return
	}
	<-groupdone
	return
}

// Hop-by-hop headers. These are removed when sent to the backend.
// http://www.w3.org/Protocols/rfc2616/rfc2616-sec13.html
var hopHeaders = []string{
	"Connection",
	"Keep-Alive",
	"Proxy-Authenticate",
	"Proxy-Connection",
	"Te", // canonicalized version of "TE"
	"Trailers",
	"Transfer-Encoding",
	"Upgrade",
}

func copyHeader(dst, src http.Header) {
	for k, vv := range src {
		for _, v := range vv {
			dst.Add(k, v)
		}
	}
}

func delHopHeaders(header http.Header) {
	for _, h := range hopHeaders {
		header.Del(h)
	}
}

func hijack(hijackable interface{}) (net.Conn, *bufio.ReadWriter, error) {
	hj, ok := hijackable.(http.Hijacker)
	if !ok {
		return nil, nil, errors.New("Connection doesn't support hijacking")
	}
	conn, rw, err := hj.Hijack()
	if err != nil {
		return nil, nil, err
	}
	var emptytime time.Time
	err = conn.SetDeadline(emptytime)
	if err != nil {
		conn.Close()
		return nil, nil, err
	}
	return conn, rw, nil
}

func flush(flusher interface{}) bool {
	f, ok := flusher.(http.Flusher)
	if !ok {
		return false
	}
	f.Flush()
	return true
}

func copyBody(wr io.Writer, body io.Reader) {
	// Use pooled buffer to avoid per-call allocation.
	bufp := copyBufPool.Get().(*[]byte)
	defer copyBufPool.Put(bufp)
	for {
		bread, read_err := body.Read(*bufp)
		var write_err error
		if bread > 0 {
			_, write_err = wr.Write((*bufp)[:bread])
			flush(wr)
		}
		if read_err != nil || write_err != nil {
			break
		}
	}
}