feat: client http proxy

app/client.example.yaml

@@ -27,3 +27,9 @@ socks5:
   # username: user
   # password: pass
   # disableUDP: true
+
+http:
+  listen: 127.0.0.1:8080
+  # username: user
+  # password: pass
+  # realm: my_private_realm

app/cmd/client.go

@@ -11,6 +11,7 @@ import (
 	"github.com/spf13/viper"
 	"go.uber.org/zap"
 
+	"github.com/apernet/hysteria/app/internal/http"
 	"github.com/apernet/hysteria/app/internal/socks5"
 	"github.com/apernet/hysteria/core/client"
 )
@@ -21,8 +22,11 @@ var clientCmd = &cobra.Command{
 	Run:   runClient,
 }
 
-var modeMap = map[string]func(*viper.Viper, client.Client) error{
+type modeFunc func(*viper.Viper, client.Client) error
+
+var modeMap = map[string]modeFunc{
 	"socks5": clientSOCKS5,
+	"http":   clientHTTP,
 }
 
 func init() {
@@ -48,17 +52,17 @@ func runClient(cmd *cobra.Command, args []string) {
 
 	var wg sync.WaitGroup
 	hasMode := false
-	for mode, f := range modeMap {
+	for mode, fn := range modeMap {
 		v := viper.Sub(mode)
 		if v != nil {
 			hasMode = true
 			wg.Add(1)
-			go func() {
+			go func(fn modeFunc) {
 				defer wg.Done()
-				if err := f(v, c); err != nil {
+				if err := fn(v, c); err != nil {
 					logger.Fatal("failed to run mode", zap.String("mode", mode), zap.Error(err))
 				}
-			}()
+			}(fn)
 		}
 	}
 	if !hasMode {
@@ -167,8 +171,8 @@ func clientSOCKS5(v *viper.Viper, c client.Client) error {
 	var authFunc func(username, password string) bool
 	username, password := v.GetString("username"), v.GetString("password")
 	if username != "" && password != "" {
-		authFunc = func(username, password string) bool {
+		authFunc = func(u, p string) bool {
-			return username == username && password == password
+			return u == username && p == password
 		}
 	}
 	s := socks5.Server{
@@ -181,6 +185,36 @@ func clientSOCKS5(v *viper.Viper, c client.Client) error {
 	return s.Serve(l)
 }
 
+func clientHTTP(v *viper.Viper, c client.Client) error {
+	listenAddr := v.GetString("listen")
+	if listenAddr == "" {
+		return configError{Field: "listen", Err: errors.New("listen address is empty")}
+	}
+	l, err := net.Listen("tcp", listenAddr)
+	if err != nil {
+		return configError{Field: "listen", Err: err}
+	}
+	var authFunc func(username, password string) bool
+	username, password := v.GetString("username"), v.GetString("password")
+	if username != "" && password != "" {
+		authFunc = func(u, p string) bool {
+			return u == username && p == password
+		}
+	}
+	realm := v.GetString("realm")
+	if realm == "" {
+		realm = "Hysteria"
+	}
+	h := http.Server{
+		HyClient:    c,
+		AuthFunc:    authFunc,
+		AuthRealm:   realm,
+		EventLogger: &httpLogger{},
+	}
+	logger.Info("HTTP proxy server listening", zap.String("addr", listenAddr))
+	return h.Serve(l)
+}
+
 func parseServerAddrString(addrStr string) (host, hostPort string) {
 	h, _, err := net.SplitHostPort(addrStr)
 	if err != nil {
@@ -215,3 +249,29 @@ func (l *socks5Logger) UDPError(addr net.Addr, err error) {
 		logger.Error("SOCKS5 UDP error", zap.String("addr", addr.String()), zap.Error(err))
 	}
 }
+
+type httpLogger struct{}
+
+func (l *httpLogger) ConnectRequest(addr net.Addr, reqAddr string) {
+	logger.Debug("HTTP CONNECT request", zap.String("addr", addr.String()), zap.String("reqAddr", reqAddr))
+}
+
+func (l *httpLogger) ConnectError(addr net.Addr, reqAddr string, err error) {
+	if err == nil {
+		logger.Debug("HTTP CONNECT closed", zap.String("addr", addr.String()), zap.String("reqAddr", reqAddr))
+	} else {
+		logger.Error("HTTP CONNECT error", zap.String("addr", addr.String()), zap.String("reqAddr", reqAddr), zap.Error(err))
+	}
+}
+
+func (l *httpLogger) HTTPRequest(addr net.Addr, reqURL string) {
+	logger.Debug("HTTP request", zap.String("addr", addr.String()), zap.String("reqURL", reqURL))
+}
+
+func (l *httpLogger) HTTPError(addr net.Addr, reqURL string, err error) {
+	if err == nil {
+		logger.Debug("HTTP closed", zap.String("addr", addr.String()), zap.String("reqURL", reqURL))
+	} else {
+		logger.Error("HTTP error", zap.String("addr", addr.String()), zap.String("reqURL", reqURL), zap.Error(err))
+	}
+}

app/internal/http/server.go

@@ -0,0 +1,295 @@
+package http
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/base64"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/apernet/hysteria/core/client"
+)
+
+const (
+	httpClientTimeout = 10 * time.Second
+)
+
+// Server is an HTTP server using a Hysteria client as outbound.
+type Server struct {
+	HyClient    client.Client
+	AuthFunc    func(username, password string) bool // nil = no authentication
+	AuthRealm   string
+	EventLogger EventLogger
+
+	httpClient *http.Client
+}
+
+type EventLogger interface {
+	ConnectRequest(addr net.Addr, reqAddr string)
+	ConnectError(addr net.Addr, reqAddr string, err error)
+	HTTPRequest(addr net.Addr, reqURL string)
+	HTTPError(addr net.Addr, reqURL string, err error)
+}
+
+func (s *Server) Serve(listener net.Listener) error {
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			return err
+		}
+		go s.dispatch(conn)
+	}
+}
+
+func (s *Server) dispatch(conn net.Conn) {
+	bufReader := bufio.NewReader(conn)
+	for {
+		req, err := http.ReadRequest(bufReader)
+		if err != nil {
+			// Connection error or invalid request
+			_ = conn.Close()
+			return
+		}
+		if s.AuthFunc != nil {
+			authOK := false
+			// Check the Proxy-Authorization header
+			pAuth := req.Header.Get("Proxy-Authorization")
+			if strings.HasPrefix(pAuth, "Basic ") {
+				userPass, err := base64.URLEncoding.DecodeString(pAuth[6:])
+				if err == nil {
+					userPassParts := strings.SplitN(string(userPass), ":", 2)
+					if len(userPassParts) == 2 {
+						authOK = s.AuthFunc(userPassParts[0], userPassParts[1])
+					}
+				}
+			}
+			if !authOK {
+				// Proxy authentication required
+				_ = sendProxyAuthRequired(conn, req, s.AuthRealm)
+				_ = conn.Close()
+				return
+			}
+		}
+		if req.Method == http.MethodConnect {
+			if bufReader.Buffered() > 0 {
+				// There is still data in the buffered reader.
+				// We need to get it out and put it into a cachedConn,
+				// so that handleConnect can read it.
+				data := make([]byte, bufReader.Buffered())
+				_, err := io.ReadFull(bufReader, data)
+				if err != nil {
+					// Read from buffer failed, is this possible?
+					_ = conn.Close()
+					return
+				}
+				cachedConn := &cachedConn{
+					Conn:   conn,
+					Buffer: *bytes.NewBuffer(data),
+				}
+				s.handleConnect(cachedConn, req)
+			} else {
+				// No data in the buffered reader, we can just pass the original connection.
+				s.handleConnect(conn, req)
+			}
+			// handleConnect will take over the connection,
+			// i.e. it will not return until the connection is closed.
+			// When it returns, there will be no more requests from this connection,
+			// so we simply exit the loop.
+			return
+		} else {
+			// handleRequest on the other hand handles one request at a time,
+			// and returns when the request is done. It returns a bool indicating
+			// whether the connection should be kept alive, but itself never closes
+			// the connection.
+			keepAlive := s.handleRequest(conn, req)
+			if !keepAlive {
+				_ = conn.Close()
+				return
+			}
+		}
+	}
+}
+
+// cachedConn is a net.Conn wrapper that first Read()s from a buffer,
+// and then from the underlying net.Conn when the buffer is drained.
+type cachedConn struct {
+	net.Conn
+	Buffer bytes.Buffer
+}
+
+func (c *cachedConn) Read(b []byte) (int, error) {
+	if c.Buffer.Len() > 0 {
+		n, err := c.Buffer.Read(b)
+		if err == io.EOF {
+			// Buffer is drained, hide it from the caller
+			err = nil
+		}
+		return n, err
+	}
+	return c.Conn.Read(b)
+}
+
+func (s *Server) handleConnect(conn net.Conn, req *http.Request) {
+	defer conn.Close()
+
+	port := req.URL.Port()
+	if port == "" {
+		// HTTP defaults to port 80
+		port = "80"
+	}
+	reqAddr := net.JoinHostPort(req.URL.Hostname(), port)
+
+	// Connect request & error log
+	if s.EventLogger != nil {
+		s.EventLogger.ConnectRequest(conn.RemoteAddr(), reqAddr)
+	}
+	var closeErr error
+	defer func() {
+		if s.EventLogger != nil {
+			s.EventLogger.ConnectError(conn.RemoteAddr(), reqAddr, closeErr)
+		}
+	}()
+
+	// Dial
+	rConn, err := s.HyClient.DialTCP(reqAddr)
+	if err != nil {
+		_ = sendSimpleResponse(conn, req, http.StatusBadGateway)
+		return
+	}
+	defer rConn.Close()
+
+	// Send 200 OK response and start relaying
+	_ = sendSimpleResponse(conn, req, http.StatusOK)
+	copyErrChan := make(chan error, 2)
+	go func() {
+		_, err := io.Copy(rConn, conn)
+		copyErrChan <- err
+	}()
+	go func() {
+		_, err := io.Copy(conn, rConn)
+		copyErrChan <- err
+	}()
+	closeErr = <-copyErrChan
+}
+
+func (s *Server) handleRequest(conn net.Conn, req *http.Request) bool {
+	// Some clients use Connection, some use Proxy-Connection
+	// https://www.oreilly.com/library/view/http-the-definitive/1565925092/re40.html
+	keepAlive := req.ProtoAtLeast(1, 1) &&
+		(strings.ToLower(req.Header.Get("Proxy-Connection")) == "keep-alive" ||
+			strings.ToLower(req.Header.Get("Connection")) == "keep-alive")
+	req.RequestURI = "" // Outgoing request should not have RequestURI
+
+	removeHopByHopHeaders(req.Header)
+	removeExtraHTTPHostPort(req)
+
+	if req.URL.Scheme == "" || req.URL.Host == "" {
+		_ = sendSimpleResponse(conn, req, http.StatusBadRequest)
+		return false
+	}
+
+	// Request & error log
+	if s.EventLogger != nil {
+		s.EventLogger.HTTPRequest(conn.RemoteAddr(), req.URL.String())
+	}
+	var closeErr error
+	defer func() {
+		if s.EventLogger != nil {
+			s.EventLogger.HTTPError(conn.RemoteAddr(), req.URL.String(), closeErr)
+		}
+	}()
+
+	if s.httpClient == nil {
+		s.initHTTPClient()
+	}
+
+	// Do the request and send the response back
+	resp, err := s.httpClient.Do(req)
+	if err != nil {
+		closeErr = err
+		_ = sendSimpleResponse(conn, req, http.StatusBadGateway)
+		return false
+	}
+
+	removeHopByHopHeaders(resp.Header)
+	if keepAlive {
+		resp.Header.Set("Connection", "keep-alive")
+		resp.Header.Set("Proxy-Connection", "keep-alive")
+		resp.Header.Set("Keep-Alive", "timeout=60")
+	}
+
+	closeErr = resp.Write(conn)
+	return closeErr == nil && keepAlive
+}
+
+func (s *Server) initHTTPClient() {
+	s.httpClient = &http.Client{
+		Transport: &http.Transport{
+			DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+				// HyClient doesn't support context for now
+				return s.HyClient.DialTCP(addr)
+			},
+		},
+		CheckRedirect: func(req *http.Request, via []*http.Request) error {
+			return http.ErrUseLastResponse
+		},
+		Timeout: httpClientTimeout,
+	}
+}
+
+func removeHopByHopHeaders(header http.Header) {
+	header.Del("Proxy-Connection") // Not in RFC but common
+	// https://www.ietf.org/rfc/rfc2616.txt
+	header.Del("Connection")
+	header.Del("Keep-Alive")
+	header.Del("Proxy-Authenticate")
+	header.Del("Proxy-Authorization")
+	header.Del("TE")
+	header.Del("Trailers")
+	header.Del("Transfer-Encoding")
+	header.Del("Upgrade")
+}
+
+func removeExtraHTTPHostPort(req *http.Request) {
+	host := req.Host
+	if host == "" {
+		host = req.URL.Host
+	}
+	if pHost, port, err := net.SplitHostPort(host); err == nil && port == "80" {
+		host = pHost
+	}
+	req.Host = host
+	req.URL.Host = host
+}
+
+// sendSimpleResponse sends a simple HTTP response with the given status code.
+func sendSimpleResponse(conn net.Conn, req *http.Request, statusCode int) error {
+	resp := &http.Response{
+		StatusCode: statusCode,
+		Status:     http.StatusText(statusCode),
+		Proto:      req.Proto,
+		ProtoMajor: req.ProtoMajor,
+		ProtoMinor: req.ProtoMinor,
+		Header:     http.Header{},
+	}
+	return resp.Write(conn)
+}
+
+// sendProxyAuthRequired sends a 407 Proxy Authentication Required response.
+func sendProxyAuthRequired(conn net.Conn, req *http.Request, realm string) error {
+	resp := &http.Response{
+		StatusCode: http.StatusProxyAuthRequired,
+		Status:     http.StatusText(http.StatusProxyAuthRequired),
+		Proto:      req.Proto,
+		ProtoMajor: req.ProtoMajor,
+		ProtoMinor: req.ProtoMinor,
+		Header:     http.Header{},
+	}
+	resp.Header.Set("Proxy-Authenticate", fmt.Sprintf("Basic realm=%q", realm))
+	return resp.Write(conn)
+}

app/internal/http/server_test.go

@@ -0,0 +1,58 @@
+package http
+
+import (
+	"errors"
+	"net"
+	"net/http"
+	"os/exec"
+	"testing"
+
+	"github.com/apernet/hysteria/core/client"
+)
+
+const (
+	testCertFile = "test.crt"
+	testKeyFile  = "test.key"
+)
+
+type mockEchoHyClient struct{}
+
+func (c *mockEchoHyClient) DialTCP(addr string) (net.Conn, error) {
+	return net.Dial("tcp", addr)
+}
+
+func (c *mockEchoHyClient) ListenUDP() (client.HyUDPConn, error) {
+	// Not implemented
+	return nil, errors.New("not implemented")
+}
+
+func (c *mockEchoHyClient) Close() error {
+	return nil
+}
+
+func TestServer(t *testing.T) {
+	// Start the server
+	s := &Server{
+		HyClient: &mockEchoHyClient{},
+	}
+	l, err := net.Listen("tcp", "127.0.0.1:18080")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer l.Close()
+	go s.Serve(l)
+
+	// Start a test HTTP & HTTPS server
+	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte("control is an illusion"))
+	})
+	go http.ListenAndServe("127.0.0.1:18081", nil)
+	go http.ListenAndServeTLS("127.0.0.1:18082", testCertFile, testKeyFile, nil)
+
+	// Run the Python test script
+	cmd := exec.Command("python", "server_test.py")
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("Failed to run test script: %v\n%s", err, out)
+	}
+}

app/internal/http/server_test.py

@@ -0,0 +1,26 @@
+import requests
+
+proxies = {
+    'http': 'http://127.0.0.1:18080',
+    'https': 'http://127.0.0.1:18080',
+}
+
+
+def test_http(it):
+    for i in range(it):
+        r = requests.get('http://127.0.0.1:18081', proxies=proxies)
+        print(r.status_code, r.text)
+        assert r.status_code == 200 and r.text == 'control is an illusion'
+
+
+def test_https(it):
+    for i in range(it):
+        r = requests.get('https://127.0.0.1:18082',
+                         proxies=proxies, verify=False)
+        print(r.status_code, r.text)
+        assert r.status_code == 200 and r.text == 'control is an illusion'
+
+
+if __name__ == '__main__':
+    test_http(10)
+    test_https(10)

app/internal/http/test.crt

@@ -0,0 +1,23 @@
+-----BEGIN CERTIFICATE-----
+MIIDwTCCAqmgAwIBAgIUMeefneiCXWS2ovxNN+fJcdrOIfAwDQYJKoZIhvcNAQEL
+BQAwcDELMAkGA1UEBhMCVFcxEzARBgNVBAgMClNvbWUtU3RhdGUxGTAXBgNVBAoM
+EFJhbmRvbSBTdHVmZiBMTEMxEjAQBgNVBAMMCWxvY2FsaG9zdDEdMBsGCSqGSIb3
+DQEJARYOcG9vcGVyQHNoaXQuY2MwHhcNMjMwNDI3MDAyMDQ1WhcNMzMwNDI0MDAy
+MDQ1WjBwMQswCQYDVQQGEwJUVzETMBEGA1UECAwKU29tZS1TdGF0ZTEZMBcGA1UE
+CgwQUmFuZG9tIFN0dWZmIExMQzESMBAGA1UEAwwJbG9jYWxob3N0MR0wGwYJKoZI
+hvcNAQkBFg5wb29wZXJAc2hpdC5jYzCCASIwDQYJKoZIhvcNAQEBBQADggEPADCC
+AQoCggEBAOU9/4AT/6fDKyEyZMMLFzUEVC8ZDJHoKZ+3g65ZFQLxRKqlEdhvOwq4
+ZsxYF0sceUPDAsdrT+km0l1jAvq6u82n6xQQ60HpKe6hOvDX7KS0dPcKa+nfEa0W
+DKamBB+TzxB2dBfBNS1oUU74nBb7ttpJiKnOpRJ0/J+CwslvhJzq04AUXC/W1CtW
+CbZBg1JjY0fCN+Oy1WjEqMtRSB6k5Ipk40a8NcsqReBOMZChR8elruZ09sIlA6tf
+jICOKToDVBmkjJ8m/GnxfV8MeLoK83M2VA73njsS6q9qe9KDVgIVQmifwi6JUb7N
+o0A6f2Z47AWJmvq4goHJtnQ3fyoeIsMCAwEAAaNTMFEwHQYDVR0OBBYEFPrBsm6v
+M29fKA3is22tK8yHYQaDMB8GA1UdIwQYMBaAFPrBsm6vM29fKA3is22tK8yHYQaD
+MA8GA1UdEwEB/wQFMAMBAf8wDQYJKoZIhvcNAQELBQADggEBAJvOwj0Tf8l9AWvf
+1ZLyW0K3m5oJAoUayjlLP9q7KHgJHWd4QXxg4ApUDo523m4Own3FwtN06KCMqlxc
+luDJi27ghRzZ8bpB9fUujikC1rs1oWYRz/K+JSO1VItan+azm9AQRj+nNepjUiT4
+FjvRif+inC4392tcKuwrqiUFmLIggtFZdsLeKUL+hRGCRjY4BZw0d1sjjPtyVNUD
+UMVO8pxlCV0NU4Nmt3vulD4YshAXM+Y8yX/vPRnaNGoRrbRgCg2VORRGaZVjQMHD
+OLMvqM7pFKnVg0uiSbQ3xbQJ8WeX620zKI0So2+kZt9HoI+46gd7BdNfl7mmd6K7
+ydYKuI8=
+-----END CERTIFICATE-----

app/internal/http/test.key

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEowIBAAKCAQEA5T3/gBP/p8MrITJkwwsXNQRULxkMkegpn7eDrlkVAvFEqqUR
+2G87CrhmzFgXSxx5Q8MCx2tP6SbSXWMC+rq7zafrFBDrQekp7qE68NfspLR09wpr
+6d8RrRYMpqYEH5PPEHZ0F8E1LWhRTvicFvu22kmIqc6lEnT8n4LCyW+EnOrTgBRc
+L9bUK1YJtkGDUmNjR8I347LVaMSoy1FIHqTkimTjRrw1yypF4E4xkKFHx6Wu5nT2
+wiUDq1+MgI4pOgNUGaSMnyb8afF9Xwx4ugrzczZUDveeOxLqr2p70oNWAhVCaJ/C
+LolRvs2jQDp/ZnjsBYma+riCgcm2dDd/Kh4iwwIDAQABAoIBABjiU/vJL/U8AFCI
+MdviNlCw+ZprM6wa8Xm+5/JjBR7epb+IT5mY6WXOgoon/c9PdfJfFswi3/fFGQy+
+FLK21nAKjEAPXho3fy/CHK3MIon2dMPkQ7aNWlPZkuH8H3J2DwIQeaWieW1GZ50U
+64yrIjwrw0P7hHuua0W9YfuPuWt29YpW5g6ilSRE0kdTzoB6TgMzlVRj6RWbxWLX
+erwYFesSpLPiQrozK2yywlQsvRV2AxTlf5woJyRTyCqcao5jNZOJJl0mqeGKNKbu
+1iYGtZl9aj1XIRxUt+JB2IMKNJasygIp+GRLUDCHKh8RVFwRlVaSNcWbfLDuyNWW
+T3lUEjECgYEA84mrs4TLuPfklsQM4WPBdN/2Ud1r0Zn/W8icHcVc/DCFXbcV4aPA
+g4yyyyEkyTac2RSbSp+rfUk/pJcG6CVjwaiRIPehdtcLIUP34EdIrwPrPT7/uWVA
+o/Hp1ANSILecknQXeE1qDlHVeGAq2k3vAQH2J0m7lMfar7QCBTMTMHcCgYEA8PkO
+Uj9+/LoHod2eb4raH29wntis31X5FX/C/8HlmFmQplxfMxpRckzDYQELdHvDggNY
+ZQo6pdE22MjCu2bk9AHa2ukMyieWm/mPe46Upr1YV2o5cWnfFFNa/LP2Ii/dWY5V
+rFNsHFnrnwcWymX7OKo0Xb8xYnKhKZJAFwSpXxUCgYBPMjXj6wtU20g6vwZxRT9k
+AnDXrmmhf7LK5jHefJAAcsbr8t3qwpWYMejypZSQ2nGnJkxZuBLMa0WHAJX+aCpI
+j8iiL+USAFxeNPwmswev4lZdVF9Uqtiad9DSYUIT4aHI/nejZ4lVnscMnjlRRIa0
+jS6/F/soJtW2zZLangFfgQKBgCOSAAUwDkSsCThhiGOasXv2bT9laI9HF4+O3m/2
+ZTfJ8Mo91GesuN0Qa77D8rbtFfz5FXFEw0d6zIfPir8y/xTtuSqbQCIPGfJIMl/g
+uhyq0oGE0pnlMOLFMyceQXTmb9wqYIchgVHmDBvbZgfWafEBXt1/vYB0v0ltpzw+
+menJAoGBAI0hx3+mrFgA+xJBEk4oexAlro1qbNWoR7BCmLQtd49jG3eZQu4JxWH2
+kh58AIXzLl0X9t4pfMYasYL6jBGvw+AqNdo2krpiL7MWEE8w8FP/wibzqmuloziB
+T7BZuCZjpcAM0IxLmQeeUK0LF0mihcqvssxveaet46mj7QoA7bGQ
+-----END RSA PRIVATE KEY-----