Merge pull request #159 from buzzfeed/sso-proxy-transport-reset-deadline

sso_proxy: add reset deadline for http transports

Author: jphines

internal/proxy/oauthproxy.go

@@ -13,6 +13,7 @@ import (
 	"reflect"
 	"regexp"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/buzzfeed/sso/internal/pkg/aead"
@@ -145,14 +146,23 @@ func (u *UpstreamProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 }
 
 // upstreamTransport is used to ensure that upstreams cannot override the
-// security headers applied by sso_proxy
+// security headers applied by sso_proxy. Also includes functionality to rotate
+// http.Transport objects to ensure tcp connection reset deadlines are not exceeded
 type upstreamTransport struct {
-	transport *http.Transport
+	mux sync.Mutex
+
+	deadAfter     time.Time
+	resetDeadline time.Duration
+
+	transport          *http.Transport
+	insecureSkipVerify bool
 }
 
 // RoundTrip round trips the request and deletes security headers before returning the response.
 func (t *upstreamTransport) RoundTrip(req *http.Request) (*http.Response, error) {
-	resp, err := t.transport.RoundTrip(req)
+	transport := t.getTransport()
+
+	resp, err := transport.RoundTrip(req)
 	if err != nil {
 		logger := log.NewLogEntry()
 		logger.Error(err, "error in upstreamTransport RoundTrip")
@@ -164,9 +174,14 @@ func (t *upstreamTransport) RoundTrip(req *http.Request) (*http.Response, error)
 	return resp, err
 }
 
-func newUpstreamTransport(insecureSkipVerify bool) *upstreamTransport {
-	return &upstreamTransport{
-		transport: &http.Transport{
+// getTransport gets either a cached http transport or allocates a new one
+func (t *upstreamTransport) getTransport() *http.Transport {
+	t.mux.Lock()
+	defer t.mux.Unlock()
+
+	if t.transport == nil || time.Now().After(t.deadAfter) {
+		t.deadAfter = time.Now().Add(t.resetDeadline)
+		t.transport = &http.Transport{
 			Proxy: http.ProxyFromEnvironment,
 			DialContext: (&net.Dialer{
 				Timeout:   30 * time.Second,
@@ -176,17 +191,22 @@ func newUpstreamTransport(insecureSkipVerify bool) *upstreamTransport {
 			MaxIdleConns:          100,
 			IdleConnTimeout:       90 * time.Second,
 			TLSHandshakeTimeout:   10 * time.Second,
-			TLSClientConfig:       &tls.Config{InsecureSkipVerify: insecureSkipVerify},
+			TLSClientConfig:       &tls.Config{InsecureSkipVerify: t.insecureSkipVerify},
 			ExpectContinueTimeout: 1 * time.Second,
-		},
+		}
 	}
+
+	return t.transport
 }
 
 // NewReverseProxy creates a reverse proxy to a specified url.
 // It adds an X-Forwarded-Host header that is the request's host.
 func NewReverseProxy(to *url.URL, config *UpstreamConfig) *httputil.ReverseProxy {
 	proxy := httputil.NewSingleHostReverseProxy(to)
-	proxy.Transport = newUpstreamTransport(config.TLSSkipVerify)
+	proxy.Transport = &upstreamTransport{
+		resetDeadline:      config.ResetDeadline,
+		insecureSkipVerify: config.TLSSkipVerify,
+	}
 	director := proxy.Director
 	proxy.Director = func(req *http.Request) {
 		req.Header.Add("X-Forwarded-Host", req.Host)
@@ -203,7 +223,10 @@ func NewReverseProxy(to *url.URL, config *UpstreamConfig) *httputil.ReverseProxy
 // It adds an X-Forwarded-Host header to the the upstream's request.
 func NewRewriteReverseProxy(route *RewriteRoute, config *UpstreamConfig) *httputil.ReverseProxy {
 	proxy := &httputil.ReverseProxy{}
-	proxy.Transport = newUpstreamTransport(config.TLSSkipVerify)
+	proxy.Transport = &upstreamTransport{
+		resetDeadline:      config.ResetDeadline,
+		insecureSkipVerify: config.TLSSkipVerify,
+	}
 	proxy.Director = func(req *http.Request) {
 		// we do this to rewrite requests
 		rewritten := route.FromRegex.ReplaceAllString(req.Host, route.ToTemplate.Opaque)

internal/proxy/oauthproxy_test.go

@@ -454,7 +454,10 @@ func TestRoundTrip(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			req := httptest.NewRequest("GET", tc.url, nil)
-			ut := newUpstreamTransport(false)
+			ut := &upstreamTransport{
+				insecureSkipVerify: false,
+				resetDeadline:      time.Duration(1) * time.Minute,
+			}
 			resp, err := ut.RoundTrip(req)
 			if err == nil && tc.expectedError {
 				t.Errorf("expected error but error was nil")

internal/proxy/options.go

@@ -31,6 +31,7 @@ import (
 // ClientID - the OAuth Client ID: ie: "123456.apps.googleusercontent.com"
 // ClientSecret - The OAuth Client Secret
 // DefaultUpstreamTimeout - the default time period to wait for a response from an upstream
+// DefaultUpstreamTCPResetDeadline - the default time period to wait for a response from an upstream
 // TCPWriteTimeout - http server tcp write timeout
 // TCPReadTimeout - http server tcp read timeout
 // CookieName - name of the cookie
@@ -66,7 +67,8 @@ type Options struct {
 	ClientID     string `envconfig:"CLIENT_ID"`
 	ClientSecret string `envconfig:"CLIENT_SECRET"`
 
-	DefaultUpstreamTimeout time.Duration `envconfig:"DEFAULT_UPSTREAM_TIMEOUT" default:"10s"`
+	DefaultUpstreamTimeout          time.Duration `envconfig:"DEFAULT_UPSTREAM_TIMEOUT" default:"10s"`
+	DefaultUpstreamTCPResetDeadline time.Duration `envconfig:"DEFAULT_UPSTREAM_TCP_RESET_DEADLINE" default:"60s"`
 
 	TCPWriteTimeout time.Duration `envconfig:"TCP_WRITE_TIMEOUT" default:"30s"`
 	TCPReadTimeout  time.Duration `envconfig:"TCP_READ_TIMEOUT" default:"30s"`
@@ -110,15 +112,19 @@ type Options struct {
 // NewOptions returns a new options struct
 func NewOptions() *Options {
 	return &Options{
-		CookieName:             "_sso_proxy",
-		CookieSecure:           true,
-		CookieHTTPOnly:         true,
-		CookieExpire:           time.Duration(168) * time.Hour,
-		SkipAuthPreflight:      false,
-		RequestLogging:         true,
-		DefaultUpstreamTimeout: time.Duration(1) * time.Second,
-		DefaultAllowedGroups:   []string{},
-		PassAccessToken:        false,
+		CookieName:     "_sso_proxy",
+		CookieSecure:   true,
+		CookieHTTPOnly: true,
+		CookieExpire:   time.Duration(168) * time.Hour,
+
+		SkipAuthPreflight: false,
+		RequestLogging:    true,
+
+		DefaultUpstreamTimeout:          time.Duration(1) * time.Second,
+		DefaultUpstreamTCPResetDeadline: time.Duration(1) * time.Minute,
+
+		DefaultAllowedGroups: []string{},
+		PassAccessToken:      false,
 	}
 }
 
@@ -185,6 +191,7 @@ func (o *Options) Validate() error {
 		defaultUpstreamOptionsConfig := &OptionsConfig{
 			AllowedGroups: o.DefaultAllowedGroups,
 			Timeout:       o.DefaultUpstreamTimeout,
+			ResetDeadline: o.DefaultUpstreamTCPResetDeadline,
 		}
 
 		o.upstreamConfigs, err = loadServiceConfigs(rawBytes, o.Cluster, o.Scheme, templateVars, defaultUpstreamOptionsConfig)

internal/proxy/proxy_config.go

@@ -56,6 +56,7 @@ type UpstreamConfig struct {
 	PreserveHost          bool
 	HMACAuth              hmacauth.HmacAuth
 	Timeout               time.Duration
+	ResetDeadline         time.Duration
 	FlushInterval         time.Duration
 	HeaderOverrides       map[string]string
 	SkipRequestSigning    bool
@@ -85,6 +86,7 @@ type OptionsConfig struct {
 	TLSSkipVerify      bool              `yaml:"tls_skip_verify"`
 	PreserveHost       bool              `yaml:"preserve_host"`
 	Timeout            time.Duration     `yaml:"timeout"`
+	ResetDeadline      time.Duration     `yaml:"reset_deadline"`
 	FlushInterval      time.Duration     `yaml:"flush_interval"`
 	SkipRequestSigning bool              `yaml:"skip_request_signing"`
 }
@@ -382,6 +384,7 @@ func parseOptionsConfig(proxy *UpstreamConfig, defaultOpts *OptionsConfig) error
 
 	proxy.AllowedGroups = dst.AllowedGroups
 	proxy.Timeout = dst.Timeout
+	proxy.ResetDeadline = dst.ResetDeadline
 	proxy.FlushInterval = dst.FlushInterval
 	proxy.HeaderOverrides = dst.HeaderOverrides
 	proxy.TLSSkipVerify = dst.TLSSkipVerify