Niko/forwardproxy
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add http.Transport + performance improvements (#47)

Browse source 
Add http.Transport + performance improvements
This commit is contained in:
sergeyfrolov 2018-08-17 17:01:52 -04:00 • committed by GitHub
parent 4c991a9635
commit 05b2092e07
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 211 additions and 128 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

201
forwardproxy.go
View file

@ -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
}

28
httpclient/httpclient.go
View file

@ -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()
}

41
probe_resist_test.go
View file

@ -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
}

69
setup.go
View file

@ -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)
}