Refactor: plain http proxy (#1443)

adapter/inbound/http.go

@@ -2,60 +2,20 @@ package inbound
 
 import (
 	"net"
-	"net/http"
-	"strings"
 
 	C "github.com/Dreamacro/clash/constant"
 	"github.com/Dreamacro/clash/context"
+	"github.com/Dreamacro/clash/transport/socks5"
 )
 
 // NewHTTP receive normal http request and return HTTPContext
-func NewHTTP(request *http.Request, conn net.Conn) *context.HTTPContext {
-	metadata := parseHTTPAddr(request)
+func NewHTTP(target string, source net.Addr, conn net.Conn) *context.ConnContext {
+	metadata := parseSocksAddr(socks5.ParseAddr(target))
+	metadata.NetWork = C.TCP
 	metadata.Type = C.HTTP
-	if ip, port, err := parseAddr(conn.RemoteAddr().String()); err == nil {
+	if ip, port, err := parseAddr(source.String()); err == nil {
 		metadata.SrcIP = ip
 		metadata.SrcPort = port
 	}
-	return context.NewHTTPContext(conn, request, metadata)
-}
-
-// RemoveHopByHopHeaders remove hop-by-hop header
-func RemoveHopByHopHeaders(header http.Header) {
-	// Strip hop-by-hop header based on RFC:
-	// http://www.w3.org/Protocols/rfc2616/rfc2616-sec13.html#sec13.5.1
-	// https://www.mnot.net/blog/2011/07/11/what_proxies_must_do
-
-	header.Del("Proxy-Connection")
-	header.Del("Proxy-Authenticate")
-	header.Del("Proxy-Authorization")
-	header.Del("TE")
-	header.Del("Trailers")
-	header.Del("Transfer-Encoding")
-	header.Del("Upgrade")
-
-	connections := header.Get("Connection")
-	header.Del("Connection")
-	if len(connections) == 0 {
-		return
-	}
-	for _, h := range strings.Split(connections, ",") {
-		header.Del(strings.TrimSpace(h))
-	}
-}
-
-// RemoveExtraHTTPHostPort remove extra host port (example.com:80 --> example.com)
-// It resolves the behavior of some HTTP servers that do not handle host:80 (e.g. baidu.com)
-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
+	return context.NewConnContext(conn, metadata)
 }

common/net/bufconn.go

@@ -1,4 +1,4 @@
-package mixed
+package net
 
 import (
 	"bufio"

context/http.go

@@ -1,47 +0,0 @@
-package context
-
-import (
-	"net"
-	"net/http"
-
-	C "github.com/Dreamacro/clash/constant"
-
-	"github.com/gofrs/uuid"
-)
-
-type HTTPContext struct {
-	id       uuid.UUID
-	metadata *C.Metadata
-	conn     net.Conn
-	req      *http.Request
-}
-
-func NewHTTPContext(conn net.Conn, req *http.Request, metadata *C.Metadata) *HTTPContext {
-	id, _ := uuid.NewV4()
-	return &HTTPContext{
-		id:       id,
-		metadata: metadata,
-		conn:     conn,
-		req:      req,
-	}
-}
-
-// ID implement C.ConnContext ID
-func (hc *HTTPContext) ID() uuid.UUID {
-	return hc.id
-}
-
-// Metadata implement C.ConnContext Metadata
-func (hc *HTTPContext) Metadata() *C.Metadata {
-	return hc.metadata
-}
-
-// Conn implement C.ConnContext Conn
-func (hc *HTTPContext) Conn() net.Conn {
-	return hc.conn
-}
-
-// Request return the http request struct
-func (hc *HTTPContext) Request() *http.Request {
-	return hc.req
-}

listener/http/client.go

@@ -0,0 +1,39 @@
+package http
+
+import (
+	"context"
+	"errors"
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/Dreamacro/clash/adapter/inbound"
+	C "github.com/Dreamacro/clash/constant"
+)
+
+func newClient(source net.Addr, in chan<- C.ConnContext) *http.Client {
+	return &http.Client{
+		Transport: &http.Transport{
+			// from http.DefaultTransport
+			MaxIdleConns:          100,
+			IdleConnTimeout:       90 * time.Second,
+			TLSHandshakeTimeout:   10 * time.Second,
+			ResponseHeaderTimeout: 10 * time.Second,
+			ExpectContinueTimeout: 1 * time.Second,
+			DialContext: func(context context.Context, network, address string) (net.Conn, error) {
+				if network != "tcp" && network != "tcp4" && network != "tcp6" {
+					return nil, errors.New("unsupported network " + network)
+				}
+
+				left, right := net.Pipe()
+
+				in <- inbound.NewHTTP(address, source, right)
+
+				return left, nil
+			},
+		},
+		CheckRedirect: func(req *http.Request, via []*http.Request) error {
+			return http.ErrUseLastResponse
+		},
+	}
+}

listener/http/hack.go

@@ -0,0 +1,10 @@
+package http
+
+import (
+	"bufio"
+	"net/http"
+	_ "unsafe"
+)
+
+//go:linkname ReadRequest net/http.readRequest
+func ReadRequest(b *bufio.Reader, deleteHostHeader bool) (req *http.Request, err error)

listener/http/proxy.go

@@ -0,0 +1,132 @@
+package http
+
+import (
+	"net"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/Dreamacro/clash/adapter/inbound"
+	"github.com/Dreamacro/clash/common/cache"
+	N "github.com/Dreamacro/clash/common/net"
+	C "github.com/Dreamacro/clash/constant"
+	authStore "github.com/Dreamacro/clash/listener/auth"
+	"github.com/Dreamacro/clash/log"
+)
+
+func HandleConn(c net.Conn, in chan<- C.ConnContext, cache *cache.Cache) {
+	client := newClient(c.RemoteAddr(), in)
+	defer client.CloseIdleConnections()
+
+	conn := N.NewBufferedConn(c)
+
+	keepAlive := true
+	trusted := cache == nil // disable authenticate if cache is nil
+
+	for keepAlive {
+		request, err := ReadRequest(conn.Reader(), false)
+		if err != nil {
+			break
+		}
+
+		request.RemoteAddr = conn.RemoteAddr().String()
+
+		keepAlive = strings.TrimSpace(strings.ToLower(request.Header.Get("Proxy-Connection"))) == "keep-alive"
+
+		var resp *http.Response
+
+		if !trusted {
+			resp = authenticate(request, cache)
+
+			trusted = resp == nil
+		}
+
+		if trusted {
+			if request.Method == http.MethodConnect {
+				resp = responseWith(200)
+				resp.Status = "Connection established"
+
+				if resp.Write(conn) != nil {
+					break // close connection
+				}
+
+				in <- inbound.NewHTTPS(request, conn)
+
+				return // hijack connection
+			}
+
+			host := request.Header.Get("Host")
+			if host != "" {
+				request.Host = host
+			}
+
+			request.RequestURI = ""
+
+			RemoveHopByHopHeaders(request.Header)
+			RemoveExtraHTTPHostPort(request)
+
+			if request.URL.Scheme == "" || request.URL.Host == "" {
+				resp = responseWith(http.StatusBadRequest)
+			} else {
+				resp, err = client.Do(request)
+				if err != nil {
+					resp = responseWith(http.StatusBadGateway)
+				}
+			}
+		}
+
+		RemoveHopByHopHeaders(resp.Header)
+
+		if keepAlive {
+			resp.Header.Set("Proxy-Connection", "keep-alive")
+			resp.Header.Set("Connection", "keep-alive")
+			resp.Header.Set("Keep-Alive", "timeout=4")
+		}
+
+		resp.Close = !keepAlive
+
+		err = resp.Write(conn)
+		if err != nil {
+			break // close connection
+		}
+	}
+
+	conn.Close()
+}
+
+func authenticate(request *http.Request, cache *cache.Cache) *http.Response {
+	authenticator := authStore.Authenticator()
+	if authenticator != nil {
+		credential := ParseBasicProxyAuthorization(request)
+		if credential == "" {
+			resp := responseWith(http.StatusProxyAuthRequired)
+			resp.Header.Set("Proxy-Authenticate", "Basic")
+			return resp
+		}
+
+		var authed interface{}
+		if authed = cache.Get(credential); authed == nil {
+			user, pass, err := DecodeBasicProxyAuthorization(credential)
+			authed = err == nil && authenticator.Verify(user, pass)
+			cache.Put(credential, authed, time.Minute)
+		}
+		if !authed.(bool) {
+			log.Infoln("Auth failed from %s", request.RemoteAddr)
+
+			return responseWith(http.StatusForbidden)
+		}
+	}
+
+	return nil
+}
+
+func responseWith(statusCode int) *http.Response {
+	return &http.Response{
+		StatusCode: statusCode,
+		Status:     http.StatusText(statusCode),
+		Proto:      "HTTP/1.1",
+		ProtoMajor: 1,
+		ProtoMinor: 1,
+		Header:     http.Header{},
+	}
+}

listener/http/server.go

@@ -1,44 +1,48 @@
 package http
 
 import (
-	"bufio"
-	"encoding/base64"
 	"net"
-	"net/http"
-	"strings"
 	"time"
 
-	"github.com/Dreamacro/clash/adapter/inbound"
 	"github.com/Dreamacro/clash/common/cache"
-	"github.com/Dreamacro/clash/component/auth"
 	C "github.com/Dreamacro/clash/constant"
-	authStore "github.com/Dreamacro/clash/listener/auth"
-	"github.com/Dreamacro/clash/log"
 )
 
 type Listener struct {
 	listener net.Listener
 	address  string
 	closed   bool
-	cache    *cache.Cache
 }
 
 func New(addr string, in chan<- C.ConnContext) (*Listener, error) {
+	return NewWithAuthenticate(addr, in, true)
+}
+
+func NewWithAuthenticate(addr string, in chan<- C.ConnContext, authenticate bool) (*Listener, error) {
 	l, err := net.Listen("tcp", addr)
 	if err != nil {
 		return nil, err
 	}
-	hl := &Listener{l, addr, false, cache.New(30 * time.Second)}
+
+	var c *cache.Cache
+	if authenticate {
+		c = cache.New(time.Second * 30)
+	}
+
+	hl := &Listener{
+		listener: l,
+		address:  addr,
+	}
 	go func() {
 		for {
-			c, err := hl.listener.Accept()
+			conn, err := hl.listener.Accept()
 			if err != nil {
 				if hl.closed {
 					break
 				}
 				continue
 			}
-			go HandleConn(c, in, hl.cache)
+			go HandleConn(conn, in, c)
 		}
 	}()
 
@@ -53,59 +57,3 @@ func (l *Listener) Close() {
 func (l *Listener) Address() string {
 	return l.address
 }
-
-func canActivate(loginStr string, authenticator auth.Authenticator, cache *cache.Cache) (ret bool) {
-	if result := cache.Get(loginStr); result != nil {
-		ret = result.(bool)
-		return
-	}
-	loginData, err := base64.StdEncoding.DecodeString(loginStr)
-	login := strings.Split(string(loginData), ":")
-	ret = err == nil && len(login) == 2 && authenticator.Verify(login[0], login[1])
-
-	cache.Put(loginStr, ret, time.Minute)
-	return
-}
-
-func HandleConn(conn net.Conn, in chan<- C.ConnContext, cache *cache.Cache) {
-	br := bufio.NewReader(conn)
-
-keepAlive:
-	request, err := http.ReadRequest(br)
-	if err != nil || request.URL.Host == "" {
-		conn.Close()
-		return
-	}
-
-	keepAlive := strings.TrimSpace(strings.ToLower(request.Header.Get("Proxy-Connection"))) == "keep-alive"
-	authenticator := authStore.Authenticator()
-	if authenticator != nil {
-		if authStrings := strings.Split(request.Header.Get("Proxy-Authorization"), " "); len(authStrings) != 2 {
-			conn.Write([]byte("HTTP/1.1 407 Proxy Authentication Required\r\nProxy-Authenticate: Basic\r\n\r\n"))
-			if keepAlive {
-				goto keepAlive
-			}
-			return
-		} else if !canActivate(authStrings[1], authenticator, cache) {
-			conn.Write([]byte("HTTP/1.1 403 Forbidden\r\n\r\n"))
-			log.Infoln("Auth failed from %s", conn.RemoteAddr().String())
-			if keepAlive {
-				goto keepAlive
-			}
-			conn.Close()
-			return
-		}
-	}
-
-	if request.Method == http.MethodConnect {
-		_, err := conn.Write([]byte("HTTP/1.1 200 Connection established\r\n\r\n"))
-		if err != nil {
-			conn.Close()
-			return
-		}
-		in <- inbound.NewHTTPS(request, conn)
-		return
-	}
-
-	in <- inbound.NewHTTP(request, conn)
-}

listener/http/utils.go

@@ -0,0 +1,74 @@
+package http
+
+import (
+	"encoding/base64"
+	"errors"
+	"net"
+	"net/http"
+	"strings"
+)
+
+// RemoveHopByHopHeaders remove hop-by-hop header
+func RemoveHopByHopHeaders(header http.Header) {
+	// Strip hop-by-hop header based on RFC:
+	// http://www.w3.org/Protocols/rfc2616/rfc2616-sec13.html#sec13.5.1
+	// https://www.mnot.net/blog/2011/07/11/what_proxies_must_do
+
+	header.Del("Proxy-Connection")
+	header.Del("Proxy-Authenticate")
+	header.Del("Proxy-Authorization")
+	header.Del("TE")
+	header.Del("Trailers")
+	header.Del("Transfer-Encoding")
+	header.Del("Upgrade")
+
+	connections := header.Get("Connection")
+	header.Del("Connection")
+	if len(connections) == 0 {
+		return
+	}
+	for _, h := range strings.Split(connections, ",") {
+		header.Del(strings.TrimSpace(h))
+	}
+}
+
+// RemoveExtraHTTPHostPort remove extra host port (example.com:80 --> example.com)
+// It resolves the behavior of some HTTP servers that do not handle host:80 (e.g. baidu.com)
+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
+}
+
+// ParseBasicProxyAuthorization parse header Proxy-Authorization and return base64-encoded credential
+func ParseBasicProxyAuthorization(request *http.Request) string {
+	value := request.Header.Get("Proxy-Authorization")
+	if !strings.HasPrefix(value, "Basic ") {
+		return ""
+	}
+
+	return value[6:] // value[len("Basic "):]
+}
+
+// DecodeBasicProxyAuthorization decode base64-encoded credential
+func DecodeBasicProxyAuthorization(credential string) (string, string, error) {
+	plain, err := base64.StdEncoding.DecodeString(credential)
+	if err != nil {
+		return "", "", err
+	}
+
+	login := strings.Split(string(plain), ":")
+	if len(login) != 2 {
+		return "", "", errors.New("invalid login")
+	}
+
+	return login[0], login[1], nil
+}

listener/mixed/mixed.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/Dreamacro/clash/common/cache"
+	N "github.com/Dreamacro/clash/common/net"
 	C "github.com/Dreamacro/clash/constant"
 	"github.com/Dreamacro/clash/listener/http"
 	"github.com/Dreamacro/clash/listener/socks"
@@ -51,7 +52,7 @@ func (l *Listener) Address() string {
 }
 
 func handleConn(conn net.Conn, in chan<- C.ConnContext, cache *cache.Cache) {
-	bufConn := NewBufferedConn(conn)
+	bufConn := N.NewBufferedConn(conn)
 	head, err := bufConn.Peek(1)
 	if err != nil {
 		return

tunnel/connection.go

@@ -1,93 +1,17 @@
 package tunnel
 
 import (
-	"bufio"
 	"errors"
 	"io"
 	"net"
-	"net/http"
-	"strings"
 	"time"
 
-	"github.com/Dreamacro/clash/adapter/inbound"
 	N "github.com/Dreamacro/clash/common/net"
 	"github.com/Dreamacro/clash/common/pool"
 	"github.com/Dreamacro/clash/component/resolver"
 	C "github.com/Dreamacro/clash/constant"
-	"github.com/Dreamacro/clash/context"
 )
 
-func handleHTTP(ctx *context.HTTPContext, outbound net.Conn) {
-	req := ctx.Request()
-	conn := ctx.Conn()
-
-	// make outbound close after inbound error or close
-	conn = &connLinker{conn, outbound}
-
-	inboundReader := bufio.NewReader(conn)
-	outboundReader := bufio.NewReader(outbound)
-
-	inbound.RemoveExtraHTTPHostPort(req)
-	host := req.Host
-	for {
-		keepAlive := strings.TrimSpace(strings.ToLower(req.Header.Get("Proxy-Connection"))) == "keep-alive"
-
-		req.RequestURI = ""
-		inbound.RemoveHopByHopHeaders(req.Header)
-		err := req.Write(outbound)
-		if err != nil {
-			break
-		}
-
-	handleResponse:
-		// resp will be closed after we call resp.Write()
-		// see https://golang.org/pkg/net/http/#Response.Write
-		resp, err := http.ReadResponse(outboundReader, req)
-		if err != nil {
-			break
-		}
-		inbound.RemoveHopByHopHeaders(resp.Header)
-
-		if resp.StatusCode == http.StatusContinue {
-			err = resp.Write(conn)
-			if err != nil {
-				break
-			}
-			goto handleResponse
-		}
-
-		// close conn when header `Connection` is `close`
-		if resp.Header.Get("Connection") == "close" {
-			keepAlive = false
-		}
-
-		if keepAlive {
-			resp.Header.Set("Proxy-Connection", "keep-alive")
-			resp.Header.Set("Connection", "keep-alive")
-			resp.Header.Set("Keep-Alive", "timeout=4")
-			resp.Close = false
-		} else {
-			resp.Close = true
-		}
-		err = resp.Write(conn)
-		if err != nil || resp.Close {
-			break
-		}
-
-		req, err = http.ReadRequest(inboundReader)
-		if err != nil {
-			break
-		}
-
-		inbound.RemoveExtraHTTPHostPort(req)
-		// Sometimes firefox just open a socket to process multiple domains in HTTP
-		// The temporary solution is close connection when encountering different HOST
-		if req.Host != host {
-			break
-		}
-	}
-}
-
 func handleUDPToRemote(packet C.UDPPacket, pc C.PacketConn, metadata *C.Metadata) error {
 	defer packet.Drop()
 
@@ -162,31 +86,3 @@ func relay(leftConn, rightConn net.Conn) {
 	rightConn.SetReadDeadline(time.Now())
 	<-ch
 }
-
-// connLinker make the two net.Conn correlated, for temporary resolution of leaks.
-// There is no better way to do this for now.
-type connLinker struct {
-	net.Conn
-	linker net.Conn
-}
-
-func (conn *connLinker) Read(b []byte) (n int, err error) {
-	n, err = conn.Conn.Read(b)
-	if err != nil {
-		conn.linker.Close()
-	}
-	return n, err
-}
-
-func (conn *connLinker) Write(b []byte) (n int, err error) {
-	n, err = conn.Conn.Write(b)
-	if err != nil {
-		conn.linker.Close()
-	}
-	return n, err
-}
-
-func (conn *connLinker) Close() error {
-	conn.linker.Close()
-	return conn.Conn.Close()
-}

tunnel/tunnel.go

@@ -289,12 +289,7 @@ func handleTCPConn(ctx C.ConnContext) {
 		log.Infoln("[TCP] %s --> %v doesn't match any rule using DIRECT", metadata.SourceAddress(), metadata.String())
 	}
 
-	switch c := ctx.(type) {
-	case *context.HTTPContext:
-		handleHTTP(c, remoteConn)
-	default:
 	handleSocket(ctx, remoteConn)
-	}
 }
 
 func shouldResolveIP(rule C.Rule, metadata *C.Metadata) bool {