From 05b2092e07f9d10b3803d8fb9775d2f87dc58590 Mon Sep 17 00:00:00 2001 From: sergeyfrolov Date: Fri, 17 Aug 2018 17:01:52 -0400 Subject: [PATCH] Add http.Transport + performance improvements (#47) Add http.Transport + performance improvements --- forwardproxy.go | 201 +++++++++++++++++++++++---------------- httpclient/httpclient.go | 28 ++++-- probe_resist_test.go | 41 ++++---- setup.go | 69 ++++++++++---- 4 files changed, 211 insertions(+), 128 deletions(-) diff --git a/forwardproxy.go b/forwardproxy.go index 4a16220..c421590 100644 --- a/forwardproxy.go +++ b/forwardproxy.go @@ -18,13 +18,17 @@ package forwardproxy import ( "bufio" + "bytes" "context" "crypto/subtle" + "encoding/base64" "errors" "fmt" "io" + "io/ioutil" "net" "net/http" + "net/url" "strconv" "strings" "sync" @@ -51,13 +55,13 @@ type ForwardProxy struct { probeResistDomain string probeResistEnabled bool - dialTimeout time.Duration // for initial tcp connection - responseTimeout *time.Duration // for getting response (affects GET requests only) + dialTimeout time.Duration // for initial tcp connection - // overridden dial and dialContext allow to redirect requests to upstream proxy - dial func(network, address string) (net.Conn, error) + httpTransport http.Transport + + // overridden dialContext allow to redirect requests to upstream proxy dialContext func(ctx context.Context, network, address string) (net.Conn, error) - upstream string // address of upstream proxy + upstream *url.URL // address of upstream proxy aclRules []aclRule whitelistedPorts []int @@ -99,25 +103,25 @@ func (fp *ForwardProxy) portIsAllowed(port string) bool { return isAllowed } -// Copies data r1->w1 and r2->w2, flushes as needed, and returns when both streams are done. -func dualStream(w1 io.Writer, r1 io.Reader, w2 io.Writer, r2 io.Reader) error { - errChan := make(chan error) - - stream := func(w io.Writer, r io.Reader) { +// Copies data target->clientReader and clientWriter->target, and flushes as needed +// Returns when clientWriter-> target stream is done. +// Caddy should finish writing target -> clientReader. +func dualStream(target net.Conn, clientReader io.ReadCloser, clientWriter io.Writer) error { + stream := func(w io.Writer, r io.Reader) error { + // copy bytes from r to w buf := bufferPool.Get().([]byte) buf = buf[0:cap(buf)] _, _err := flushingIoCopy(w, r, buf) - errChan <- _err + if closeWriter, ok := w.(interface { + CloseWrite() error + }); ok { + closeWriter.CloseWrite() + } + return _err } - go stream(w1, r1) - go stream(w2, r2) - err1 := <-errChan - err2 := <-errChan - if err1 != nil { - return err1 - } - return err2 + go stream(target, clientReader) + return stream(clientWriter, target) } // Hijacks the connection from ResponseWriter, writes the response and proxies data between targetConn @@ -158,7 +162,7 @@ func serveHijack(w http.ResponseWriter, targetConn net.Conn) (int, error) { return http.StatusInternalServerError, errors.New("failed to send response to client: " + err.Error()) } - return 0, dualStream(targetConn, clientConn, clientConn, targetConn) + return 0, dualStream(targetConn, clientConn, clientConn) } // Returns nil error on successful credentials check. @@ -237,56 +241,37 @@ func (fp *ForwardProxy) servePacFile(w http.ResponseWriter) (int, error) { return 0, nil } -// bool indicates whether it was rejected as "Forbidden" -// TODO: after custom status code-based errors are implemented package-wide, remove the bool -func (fp *ForwardProxy) dialRequestedAddress(r *http.Request) (net.Conn, error, bool) { - var err error +// dialContextCheckACL enforces Access Control List and calls fp.DialContext +func (fp *ForwardProxy) dialContextCheckACL(ctx context.Context, network, hostPort string) (net.Conn, *ProxyError) { var conn net.Conn - hostPort := r.URL.Host - if hostPort == "" { - hostPort = r.Host + if network != "tcp" && network != "tcp4" && network != "tcp6" { + return nil, &ProxyError{S: "Network " + network + " is not supported", Code: http.StatusBadRequest} } + host, port, err := net.SplitHostPort(hostPort) if err != nil { - if r.Method == http.MethodConnect { - return nil, err, false - } - // for other methods, try implicit port 80 - hostPort = net.JoinHostPort(hostPort, "80") - host, port, err = net.SplitHostPort(hostPort) - if err != nil { - return nil, err, false - } + return nil, &ProxyError{S: err.Error(), Code: http.StatusBadRequest} } - if fp.upstream != "" { - // if upstreaming -- do not resolve locally nor check acl - if fp.dialContext != nil && !fp.hideIP { - ctxHeader := make(http.Header) - for k, v := range r.Header { - if kL := strings.ToLower(k); kL == "forwarded" || kL == "x-forwarded-for" { - ctxHeader[k] = v - } - } - ctxHeader.Add("Forwarded", "for=\""+r.RemoteAddr+"\"") - ctx := context.WithValue(context.Background(), httpclient.ContextKeyHeader{}, ctxHeader) - conn, err = fp.dialContext(ctx, "tcp", hostPort) - } else { - conn, err = fp.dial("tcp", hostPort) + if fp.upstream != nil { + // if upstreaming -- do not resolve locally nor check acl + conn, err = fp.dialContext(ctx, network, hostPort) + if err != nil { + return conn, &ProxyError{S: err.Error(), Code: http.StatusBadGateway} } - return conn, err, false + return conn, nil } if !fp.portIsAllowed(port) { - return nil, errors.New("port " + port + " is not allowed"), true + return nil, &ProxyError{S: "port " + port + " is not allowed", Code: http.StatusForbidden} } // in case IP was provided, net.LookupIP will simply return it IPs, err := net.LookupIP(host) if err != nil { - return nil, errors.New(fmt.Sprintf("Lookup of %s failed: %v", - host, err)), false + return nil, &ProxyError{S: fmt.Sprintf("Lookup of %s failed: %v", host, err), + Code: http.StatusBadGateway} } // This is net.Dial's default behavior: if the host resolves to multiple IP addresses, @@ -296,12 +281,12 @@ func (fp *ForwardProxy) dialRequestedAddress(r *http.Request) (net.Conn, error, continue } - conn, err = fp.dial("tcp", hostPort) + conn, err = fp.dialContext(ctx, network, net.JoinHostPort(ip.String(), port)) if err == nil { - return conn, err, false + return conn, nil } } - return nil, errors.New("No allowed IP addresses for " + host), true + return nil, &ProxyError{S: "No allowed IP addresses for " + host, Code: http.StatusForbidden} } func (fp *ForwardProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) (int, error) { @@ -335,20 +320,17 @@ func (fp *ForwardProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) (int, return http.StatusHTTPVersionNotSupported, errors.New("Unsupported HTTP major version: " + strconv.Itoa(r.ProtoMajor)) } - targetConn, err, forbidden := fp.dialRequestedAddress(r) - if forbidden { - return http.StatusForbidden, err + ctx := context.Background() + if !fp.hideIP { + ctxHeader := make(http.Header) + for k, v := range r.Header { + if kL := strings.ToLower(k); kL == "forwarded" || kL == "x-forwarded-for" { + ctxHeader[k] = v + } + } + ctxHeader.Add("Forwarded", "for=\""+r.RemoteAddr+"\"") + ctx = context.WithValue(ctx, httpclient.ContextKeyHeader{}, ctxHeader) } - if err != nil { - // failed, but not because it's forbidden - return http.StatusBadGateway, errors.New(fmt.Sprintf("dial %s failed: %v", r.URL.Host, err)) - } - if targetConn == nil { - // safest to check both error and targetConn afterwards, in case fp.dial (potentially unstable - // from x/net/proxy) misbehaves and returns both nil or both non-nil - return http.StatusForbidden, errors.New("hostname " + r.URL.Hostname() + " is not allowed") - } - defer targetConn.Close() if r.Method == http.MethodConnect { if r.ProtoMajor == 2 { @@ -357,6 +339,21 @@ func (fp *ForwardProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) (int, } } + hostPort := r.URL.Host + if hostPort == "" { + hostPort = r.Host + } + targetConn, err := fp.dialContextCheckACL(ctx, "tcp", hostPort) + if err != nil { + return err.SplitCodeError() + } + if targetConn == nil { + // safest to check both error and targetConn afterwards, in case fp.dial (potentially unstable + // from x/net/proxy) misbehaves and returns both nil or both non-nil + return http.StatusForbidden, errors.New("hostname " + r.URL.Hostname() + " is not allowed") + } + defer targetConn.Close() + switch r.ProtoMajor { case 1: // http1: hijack the whole flow return serveHijack(w, targetConn) @@ -368,7 +365,7 @@ func (fp *ForwardProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) (int, } w.WriteHeader(http.StatusOK) wFlusher.Flush() - return 0, dualStream(targetConn, r.Body, w, targetConn) + return 0, dualStream(targetConn, r.Body, w) default: panic("There was a check for http version, yet it's incorrect") } @@ -382,6 +379,9 @@ func (fp *ForwardProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) (int, if r.URL.Host == "" { r.URL.Host = r.Host } + r.Proto = "HTTP/1.1" + r.ProtoMajor = 1 + r.ProtoMinor = 1 r.RequestURI = "" removeHopByHop(r.Header) @@ -395,21 +395,61 @@ func (fp *ForwardProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) (int, r.Header.Add("Via", strconv.Itoa(r.ProtoMajor)+"."+strconv.Itoa(r.ProtoMinor)+" caddy") } - if fp.responseTimeout != nil { - targetConn.SetDeadline(time.Now().Add(*fp.responseTimeout)) - } - + var err error var response *http.Response - err = r.Write(targetConn) - if err != nil { - return http.StatusBadGateway, errors.New("failed to write http request: " + err.Error()) + if fp.upstream == nil { + // non-upstream request uses httpTransport to reuse connections + if r.Body != nil && + (r.Method == "GET" || r.Method == "HEAD" || r.Method == "OPTIONS" || r.Method == "TRACE") { + // make sure request is idempotent and could be retried by saving the Body + // None of those methods are supposed to have body, + // but we still need to copy the r.Body, even if it's empty + rBodyBuf, err := ioutil.ReadAll(r.Body) + if err != nil { + return http.StatusBadRequest, errors.New("failed to read request Body: " + err.Error()) + } + r.GetBody = func() (io.ReadCloser, error) { + return ioutil.NopCloser(bytes.NewReader(rBodyBuf)), nil + } + r.Body, _ = r.GetBody() + } + response, err = fp.httpTransport.RoundTrip(r) + } else { + // Upstream requests don't interact well with Transport: connections could always be + // reused, but Transport thinks they go to different Hosts, so it spawns tons of + // useless connections. + // Just use dialContext, which will multiplex via single connection, if http/2 + if creds := fp.upstream.User.String(); creds != "" { + // set upstream credentials for the request, if needed + r.Header.Set("Proxy-Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(creds))) + } + if r.URL.Port() == "" { + r.URL.Host = net.JoinHostPort(r.URL.Host, "80") + } + upsConn, err := fp.dialContext(ctx, "tcp", r.URL.Host) + if err != nil { + return http.StatusBadGateway, errors.New("failed to dial upstream: " + err.Error()) + } + err = r.Write(upsConn) + if err != nil { + return http.StatusBadGateway, errors.New("failed to write http request: " + err.Error()) + } + response, err = http.ReadResponse(bufio.NewReader(upsConn), r) + if err != nil { + return http.StatusBadGateway, errors.New("failed to read http response: " + err.Error()) + } + } + r.Body.Close() + if response != nil { + defer response.Body.Close() } - response, err = http.ReadResponse(bufio.NewReader(targetConn), r) if err != nil { + if p, ok := err.(*ProxyError); ok { + return p.SplitCodeError() + } return http.StatusBadGateway, errors.New("failed to read http response: " + err.Error()) } - // TODO?: check 301 and 302 redirects against ACL and follow them return 0, forwardResponse(w, response) } } @@ -429,7 +469,6 @@ func forwardResponse(w http.ResponseWriter, response *http.Response) error { buf := bufferPool.Get().([]byte) buf = buf[0:cap(buf)] _, err := io.CopyBuffer(w, response.Body, buf) - response.Body.Close() return err } diff --git a/httpclient/httpclient.go b/httpclient/httpclient.go index 38ce596..5a6ee48 100644 --- a/httpclient/httpclient.go +++ b/httpclient/httpclient.go @@ -23,7 +23,6 @@ import ( "encoding/base64" "errors" "io" - "io/ioutil" "net" "net/http" "net/url" @@ -129,7 +128,7 @@ func (c *HTTPConnectDialer) DialContext(ctx context.Context, network, address st req.ProtoMajor = 2 req.ProtoMinor = 0 pr, pw := io.Pipe() - req.Body = ioutil.NopCloser(pr) + req.Body = pr resp, err := h2clientConn.RoundTrip(req) if err != nil { @@ -170,17 +169,23 @@ func (c *HTTPConnectDialer) DialContext(ctx context.Context, network, address st if c.EnableH2ConnReuse { c.cacheH2Mu.Lock() + unlocked := false if c.cachedH2ClientConn != nil && c.cachedH2RawConn != nil { if c.cachedH2ClientConn.CanTakeNewRequest() { - proxyConn, err := connectHttp2(c.cachedH2RawConn, c.cachedH2ClientConn) + rc := c.cachedH2RawConn + cc := c.cachedH2ClientConn + c.cacheH2Mu.Unlock() + unlocked = true + proxyConn, err := connectHttp2(rc, cc) if err == nil { - c.cacheH2Mu.Unlock() return proxyConn, err } // else: carry on and try again } } - c.cacheH2Mu.Unlock() + if !unlocked { + c.cacheH2Mu.Unlock() + } } var err error @@ -269,7 +274,18 @@ func (h *http2Conn) Write(p []byte) (n int, err error) { } func (h *http2Conn) Close() error { - h.out.Close() h.in.Close() + return h.out.Close() +} + +func (h *http2Conn) CloseConn() error { return h.Conn.Close() } + +func (h *http2Conn) CloseWrite() error { + return h.in.Close() +} + +func (h *http2Conn) CloseRead() error { + return h.out.Close() +} diff --git a/probe_resist_test.go b/probe_resist_test.go index 4ebf365..a2558d6 100644 --- a/probe_resist_test.go +++ b/probe_resist_test.go @@ -4,7 +4,7 @@ import ( "bytes" "errors" "fmt" - "io" + "io/ioutil" "net/http" "strings" "testing" @@ -360,36 +360,27 @@ func responsesAreEqual(res1, res2 *http.Response) error { } } // Compare bodies - buf1 := make([]byte, 2048) - buf2 := make([]byte, 2048) - var n1, n2 int - var err1, err2 error + buf1, err1 := ioutil.ReadAll(res1.Body) + buf2, err2 := ioutil.ReadAll(res2.Body) + n1 := len(buf1) + n2 := len(buf2) makeBodyError := func(s string) error { return errors.New(fmt.Sprintf("Bodies are different: %s. n1 = %d, n2 = %d. err1 = %v, err2 = %v. buf1 = %s, buf2 = %s", s, n1, n2, err1, err2, buf1[:n1], buf2[:n2])) } - for { - n1, err1 = res1.Body.Read(buf1[:]) - n2, err2 = res2.Body.Read(buf2[:n1]) - buf1 = removeAddressesByte(buf1[:n1]) - buf2 = removeAddressesByte(buf2[:n1]) - for i := range buf1 { - if buf1[i] != buf2[i] { - return makeBodyError(fmt.Sprintf("Mismatched character %d", i)) - } - } - if err1 == io.EOF && err2 == io.EOF { - break - } - if err1 == io.EOF && err2 == nil { - _n, _ := res2.Body.Read(buf2[n1:]) - n2 += _n - return makeBodyError("Body 2 is longer") - } - if err1 != nil || err2 != nil { - return makeBodyError("Unexpected Read errors") + if n2 != n1 { + return makeBodyError("Body sizes are different") + } + buf1 = removeAddressesByte(buf1[:n1]) + buf2 = removeAddressesByte(buf2[:n1]) + for i := range buf1 { + if buf1[i] != buf2[i] { + return makeBodyError(fmt.Sprintf("Mismatched character %d", i)) } } + if err1 != nil || err2 != nil { + return makeBodyError("Unexpected Read errors") + } return nil } diff --git a/setup.go b/setup.go index e5bfacb..95a400d 100644 --- a/setup.go +++ b/setup.go @@ -20,8 +20,10 @@ import ( "crypto/tls" "encoding/base64" "errors" + "fmt" "log" "net" + "net/http" "net/url" "os" "strconv" @@ -40,6 +42,12 @@ func setup(c *caddy.Controller) error { fp := &ForwardProxy{ dialTimeout: time.Second * 20, hostname: httpserver.GetConfig(c).Host(), port: httpserver.GetConfig(c).Port(), + httpTransport: http.Transport{ + Proxy: http.ProxyFromEnvironment, + MaxIdleConns: 50, + IdleConnTimeout: 60 * time.Second, + TLSHandshakeTimeout: 10 * time.Second, + }, } c.Next() // skip the directive name @@ -137,8 +145,7 @@ func setup(c *caddy.Controller) error { if timeout < 0 { return c.Err("response_timeout cannot be negative.") } - responseTimeout := time.Duration(timeout) * time.Second - fp.responseTimeout = &responseTimeout + fp.httpTransport.ResponseHeaderTimeout = time.Duration(timeout) * time.Second case "dial_timeout": if len(args) != 1 { return c.ArgErr() @@ -155,7 +162,14 @@ func setup(c *caddy.Controller) error { if len(args) != 1 { return c.ArgErr() } - fp.upstream = args[0] + if fp.upstream != nil { + return c.Err("upstream directive specified more than once") + } + var err error + fp.upstream, err = url.Parse(args[0]) + if err != nil { + return c.Err("failed to parse upstream address: " + err.Error()) + } case "acl": if len(args) != 0 { return c.Err("acl should be only subdirective on the line") @@ -223,7 +237,7 @@ func setup(c *caddy.Controller) error { } } - if fp.upstream != "" && (fp.aclRules != nil || len(fp.whitelistedPorts) != 0) { + if fp.upstream != nil && (fp.aclRules != nil || len(fp.whitelistedPorts) != 0) { return c.Err("upstream subdirective is incompatible with acl/ports subdirectives") } @@ -258,16 +272,17 @@ func setup(c *caddy.Controller) error { KeepAlive: 30 * time.Second, DualStack: true, } - - fp.dial = dialer.Dial - - if fp.upstream != "" { - upstreamURL, err := url.Parse(fp.upstream) + fp.dialContext = dialer.DialContext + fp.httpTransport.DialContext = func(ctx context.Context, network string, address string) (net.Conn, error) { + conn, err := fp.dialContextCheckACL(ctx, network, address) if err != nil { - return errors.New("failed to parse upstream address: " + err.Error()) + return conn, err } + return conn, nil + } - if !isLocalhost(upstreamURL.Hostname()) && upstreamURL.Scheme != "https" { + if fp.upstream != nil { + if !isLocalhost(fp.upstream.Hostname()) && fp.upstream.Scheme != "https" { return errors.New("insecure schemes are only allowed to localhost upstreams") } @@ -275,12 +290,12 @@ func setup(c *caddy.Controller) error { // CONNECT request is proxied as-is, so we don't care about target url, but it could be // useful in future to implement policies of choosing between multiple upstream servers. // Given dialer is not used, since it's the same dialer provided by us. - d, err := httpclient.NewHTTPConnectDialer(upstreamURL.String()) + d, err := httpclient.NewHTTPConnectDialer(fp.upstream.String()) if err != nil { return nil, err } d.Dialer = *dialer - if isLocalhost(upstreamURL.Hostname()) && upstreamURL.Scheme == "https" { + if isLocalhost(fp.upstream.Hostname()) && fp.upstream.Scheme == "https" { // disabling verification helps with testing the package and setups // either way, it's impossible to have a legit TLS certificate for "127.0.0.1" log.Println("Localhost upstream detected, disabling verification of TLS certificate") @@ -297,15 +312,21 @@ func setup(c *caddy.Controller) error { proxy.RegisterDialerType("https", registerHTTPDialer) proxy.RegisterDialerType("http", registerHTTPDialer) - newDialer, err := proxy.FromURL(upstreamURL, dialer) + upstreamDialer, err := proxy.FromURL(fp.upstream, dialer) if err != nil { return errors.New("failed to create proxy to upstream: " + err.Error()) } - fp.dial = newDialer.Dial - if ctxDialer, ok := newDialer.(interface { + + if ctxDialer, ok := upstreamDialer.(interface { DialContext(ctx context.Context, network, address string) (net.Conn, error) }); ok { + // upstreamDialer has DialContext - use it fp.dialContext = ctxDialer.DialContext + } else { + // upstreamDialer does not have DialContext - ignore the context :( + fp.dialContext = func(ctx context.Context, network string, address string) (net.Conn, error) { + return upstreamDialer.Dial(network, address) + } } } @@ -370,3 +391,19 @@ func isValidDomainLite(domain string) error { } return nil } + +type ProxyError struct { + S string + Code int +} + +func (e *ProxyError) Error() string { + return fmt.Sprintf("[%v] %s", e.Code, e.S) +} + +func (e *ProxyError) SplitCodeError() (int, error) { + if e == nil { + return 200, nil + } + return e.Code, errors.New(e.S) +}