HTTP proxy written in Go to replace Tinyproxy (#269)

2020-10-31 21:50:31 -04:00
parent 58da55da1e
commit 0c9f74ffa4
29 changed files with 623 additions and 680 deletions
--- a/internal/httpproxy/auth.go
+++ b/internal/httpproxy/auth.go
@@ -0,0 +1,33 @@
+package httpproxy
+
+import (
+	"encoding/base64"
+	"net/http"
+	"strings"
+)
+
+func isAuthorized(responseWriter http.ResponseWriter, request *http.Request,
+	username, password string) (authorized bool) {
+	basicAuth := request.Header.Get("Proxy-Authorization")
+	if len(basicAuth) == 0 {
+		responseWriter.WriteHeader(http.StatusProxyAuthRequired)
+		return false
+	}
+	b64UsernamePassword := strings.TrimPrefix(basicAuth, "Basic ")
+	b, err := base64.StdEncoding.DecodeString(b64UsernamePassword)
+	if err != nil {
+		responseWriter.WriteHeader(http.StatusUnauthorized)
+		return false
+	}
+	usernamePassword := strings.Split(string(b), ":")
+	const expectedFields = 2
+	if len(usernamePassword) != expectedFields {
+		responseWriter.WriteHeader(http.StatusBadRequest)
+		return false
+	}
+	if username != usernamePassword[0] && password != usernamePassword[1] {
+		responseWriter.WriteHeader(http.StatusUnauthorized)
+		return false
+	}
+	return true
+}
--- a/internal/httpproxy/handler.go
+++ b/internal/httpproxy/handler.go
@@ -0,0 +1,62 @@
+package httpproxy
+
+import (
+	"context"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/qdm12/golibs/logging"
+)
+
+func newHandler(ctx context.Context, wg *sync.WaitGroup,
+	client *http.Client, logger logging.Logger,
+	stealth, verbose bool, username, password string) http.Handler {
+	const relayTimeout = 10 * time.Second
+	return &handler{
+		ctx:          ctx,
+		wg:           wg,
+		client:       client,
+		logger:       logger,
+		relayTimeout: relayTimeout,
+		verbose:      verbose,
+		stealth:      stealth,
+		username:     username,
+		password:     password,
+	}
+}
+
+type handler struct {
+	ctx                context.Context
+	wg                 *sync.WaitGroup
+	client             *http.Client
+	logger             logging.Logger
+	relayTimeout       time.Duration
+	verbose, stealth   bool
+	username, password string
+}
+
+func (h *handler) ServeHTTP(responseWriter http.ResponseWriter, request *http.Request) {
+	if len(h.username) > 0 && !isAuthorized(responseWriter, request, h.username, h.password) {
+		h.logger.Info("%s unauthorized", request.RemoteAddr)
+		return
+	}
+	switch request.Method {
+	case http.MethodConnect:
+		h.handleHTTPS(responseWriter, request)
+	default:
+		h.handleHTTP(responseWriter, request)
+	}
+}
+
+// http://www.w3.org/Protocols/rfc2616/rfc2616-sec13.html
+var hopHeaders = [...]string{ //nolint:gochecknoglobals
+	"Connection",
+	"Keep-Alive",
+	"Proxy-Authenticate",
+	"Proxy-Authorization",
+	"Te", // canonicalized version of "TE"
+	"Trailers",
+	"Transfer-Encoding",
+	"Upgrade",
+}
--- a/internal/httpproxy/http.go
+++ b/internal/httpproxy/http.go
@@ -0,0 +1,74 @@
+package httpproxy
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+)
+
+func (h *handler) handleHTTP(responseWriter http.ResponseWriter, request *http.Request) {
+	switch request.URL.Scheme {
+	case "http", "https":
+	default:
+		h.logger.Warn("Unsupported scheme %q", request.URL.Scheme)
+		http.Error(responseWriter, "unsupported scheme", http.StatusBadRequest)
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(h.ctx, h.relayTimeout)
+	defer cancel()
+	request = request.WithContext(ctx)
+
+	request.RequestURI = ""
+
+	for _, key := range hopHeaders {
+		request.Header.Del(key)
+	}
+
+	if !h.stealth {
+		setForwardedHeaders(request)
+	}
+
+	response, err := h.client.Do(request)
+	if err != nil {
+		http.Error(responseWriter, "server error", http.StatusInternalServerError)
+		h.logger.Warn("cannot request %s for client %q: %s",
+			request.URL, request.RemoteAddr, err)
+		return
+	}
+	defer response.Body.Close()
+	if h.verbose {
+		h.logger.Info("%s %s %s %s", request.RemoteAddr, response.Status, request.Method, request.URL)
+	}
+
+	for _, key := range hopHeaders {
+		response.Header.Del(key)
+	}
+
+	targetHeaderPtr := responseWriter.Header()
+	for key, values := range response.Header {
+		for _, value := range values {
+			targetHeaderPtr.Add(key, value)
+		}
+	}
+
+	responseWriter.WriteHeader(response.StatusCode)
+	if _, err := io.Copy(responseWriter, response.Body); err != nil {
+		h.logger.Error("%s %s: body copy error: %s", request.RemoteAddr, request.URL, err)
+	}
+}
+
+func setForwardedHeaders(request *http.Request) {
+	clientIP, _, err := net.SplitHostPort(request.RemoteAddr)
+	if err != nil {
+		return
+	}
+	// keep existing proxy headers
+	if prior, ok := request.Header["X-Forwarded-For"]; ok {
+		clientIP = fmt.Sprintf("%s,%s", strings.Join(prior, ", "), clientIP)
+	}
+	request.Header.Set("X-Forwarded-For", clientIP)
+}
--- a/internal/httpproxy/https.go
+++ b/internal/httpproxy/https.go
@@ -0,0 +1,64 @@
+package httpproxy
+
+import (
+	"context"
+	"io"
+	"net"
+	"net/http"
+	"sync"
+)
+
+func (h *handler) handleHTTPS(responseWriter http.ResponseWriter, request *http.Request) {
+	dialer := net.Dialer{Timeout: h.relayTimeout}
+	destinationConn, err := dialer.DialContext(h.ctx, "tcp", request.Host)
+	if err != nil {
+		http.Error(responseWriter, err.Error(), http.StatusServiceUnavailable)
+		return
+	}
+
+	responseWriter.WriteHeader(http.StatusOK)
+
+	hijacker, ok := responseWriter.(http.Hijacker)
+	if !ok {
+		http.Error(responseWriter, "Hijacking not supported", http.StatusInternalServerError)
+		return
+	}
+	clientConnection, _, err := hijacker.Hijack()
+	if err != nil {
+		h.logger.Warn(err)
+		http.Error(responseWriter, err.Error(), http.StatusServiceUnavailable)
+		if err := destinationConn.Close(); err != nil {
+			h.logger.Error("closing destination connection: %s", err)
+		}
+		return
+	}
+
+	if h.verbose {
+		h.logger.Info("%s <-> %s", request.RemoteAddr, request.Host)
+	}
+
+	h.wg.Add(1)
+	ctx, cancel := context.WithCancel(h.ctx)
+	const transferGoroutines = 2
+	wg := &sync.WaitGroup{}
+	wg.Add(transferGoroutines)
+	go func() { // trigger cleanup when done
+		wg.Wait()
+		cancel()
+	}()
+	go func() { // cleanup
+		<-ctx.Done()
+		destinationConn.Close()
+		clientConnection.Close()
+		h.wg.Done()
+	}()
+	go transfer(destinationConn, clientConnection, wg)
+	go transfer(clientConnection, destinationConn, wg)
+}
+
+func transfer(destination io.WriteCloser, source io.ReadCloser, wg *sync.WaitGroup) {
+	_, _ = io.Copy(destination, source)
+	_ = source.Close()
+	_ = destination.Close()
+	wg.Done()
+}
--- a/internal/httpproxy/loop.go
+++ b/internal/httpproxy/loop.go
@@ -0,0 +1,139 @@
+package httpproxy
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"sync"
+
+	"github.com/qdm12/gluetun/internal/settings"
+	"github.com/qdm12/golibs/logging"
+)
+
+type Looper interface {
+	Run(ctx context.Context, wg *sync.WaitGroup)
+	Restart()
+	Start()
+	Stop()
+	GetSettings() (settings settings.HTTPProxy)
+	SetSettings(settings settings.HTTPProxy)
+}
+
+type looper struct {
+	client        *http.Client
+	settings      settings.HTTPProxy
+	settingsMutex sync.RWMutex
+	logger        logging.Logger
+	restart       chan struct{}
+	start         chan struct{}
+	stop          chan struct{}
+}
+
+func NewLooper(client *http.Client, logger logging.Logger,
+	settings settings.HTTPProxy) Looper {
+	return &looper{
+		client:   client,
+		settings: settings,
+		logger:   logger.WithPrefix("http proxy: "),
+		restart:  make(chan struct{}),
+		start:    make(chan struct{}),
+		stop:     make(chan struct{}),
+	}
+}
+
+func (l *looper) GetSettings() (settings settings.HTTPProxy) {
+	l.settingsMutex.RLock()
+	defer l.settingsMutex.RUnlock()
+	return l.settings
+}
+
+func (l *looper) SetSettings(settings settings.HTTPProxy) {
+	l.settingsMutex.Lock()
+	defer l.settingsMutex.Unlock()
+	l.settings = settings
+}
+
+func (l *looper) isEnabled() bool {
+	l.settingsMutex.RLock()
+	defer l.settingsMutex.RUnlock()
+	return l.settings.Enabled
+}
+
+func (l *looper) setEnabled(enabled bool) {
+	l.settingsMutex.Lock()
+	defer l.settingsMutex.Unlock()
+	l.settings.Enabled = enabled
+}
+
+func (l *looper) Restart() { l.restart <- struct{}{} }
+func (l *looper) Start()   { l.start <- struct{}{} }
+func (l *looper) Stop()    { l.stop <- struct{}{} }
+
+func (l *looper) Run(ctx context.Context, wg *sync.WaitGroup) {
+	defer wg.Done()
+	waitForStart := true
+	for waitForStart {
+		select {
+		case <-l.stop:
+			l.logger.Info("not started yet")
+		case <-l.start:
+			waitForStart = false
+		case <-l.restart:
+			waitForStart = false
+		case <-ctx.Done():
+			return
+		}
+	}
+	defer l.logger.Warn("loop exited")
+
+	for ctx.Err() == nil {
+		for !l.isEnabled() {
+			// wait for a signal to re-enable
+			select {
+			case <-l.stop:
+				l.logger.Info("already disabled")
+			case <-l.restart:
+				l.setEnabled(true)
+			case <-l.start:
+				l.setEnabled(true)
+			case <-ctx.Done():
+				return
+			}
+		}
+
+		settings := l.GetSettings()
+		address := fmt.Sprintf("0.0.0.0:%d", settings.Port)
+
+		server := New(ctx, address, l.logger, l.client, settings.Stealth, settings.Log, settings.User, settings.Password)
+
+		runCtx, runCancel := context.WithCancel(context.Background())
+		runWg := &sync.WaitGroup{}
+		runWg.Add(1)
+		go server.Run(runCtx, runWg)
+
+		stayHere := true
+		for stayHere {
+			select {
+			case <-ctx.Done():
+				l.logger.Warn("context canceled: exiting loop")
+				runCancel()
+				runWg.Wait()
+				return
+			case <-l.restart: // triggered restart
+				l.logger.Info("restarting")
+				runCancel()
+				runWg.Wait()
+				stayHere = false
+			case <-l.start:
+				l.logger.Info("already started")
+			case <-l.stop:
+				l.logger.Info("stopping")
+				runCancel()
+				runWg.Wait()
+				l.setEnabled(false)
+				stayHere = false
+			}
+		}
+		runCancel() // repetition for linter only
+	}
+}
--- a/internal/httpproxy/server.go
+++ b/internal/httpproxy/server.go
@@ -0,0 +1,55 @@
+package httpproxy
+
+import (
+	"context"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/qdm12/golibs/logging"
+)
+
+type Server interface {
+	Run(ctx context.Context, wg *sync.WaitGroup)
+}
+
+type server struct {
+	address    string
+	handler    http.Handler
+	logger     logging.Logger
+	internalWG *sync.WaitGroup
+}
+
+func New(ctx context.Context, address string,
+	logger logging.Logger, client *http.Client,
+	stealth, verbose bool, username, password string) Server {
+	wg := &sync.WaitGroup{}
+	return &server{
+		address:    address,
+		handler:    newHandler(ctx, wg, client, logger, stealth, verbose, username, password),
+		logger:     logger,
+		internalWG: wg,
+	}
+}
+
+func (s *server) Run(ctx context.Context, wg *sync.WaitGroup) {
+	defer wg.Done()
+	server := http.Server{Addr: s.address, Handler: s.handler}
+	go func() {
+		<-ctx.Done()
+		s.logger.Warn("context canceled: exiting loop")
+		defer s.logger.Warn("loop exited")
+		const shutdownGraceDuration = 2 * time.Second
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownGraceDuration)
+		defer cancel()
+		if err := server.Shutdown(shutdownCtx); err != nil {
+			s.logger.Error("failed shutting down: %s", err)
+		}
+	}()
+	s.logger.Info("listening on %s", s.address)
+	err := server.ListenAndServe()
+	if err != nil && ctx.Err() != context.Canceled {
+		s.logger.Error(err)
+	}
+	s.internalWG.Wait()
+}