package reverseProxy // Modified version of the original golang HTTP reverse proxy handler // And Vars in Gorilla/mux // Added support for Filter functions // Copyright 2011 The Go Authors. All rights reserved. // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. import ( "io" "log" "net" "net/http" "net/url" "strings" "sync" "time" "github.com/Sirupsen/logrus" "github.com/gorilla/context" "github.com/gorilla/mux" "github.com/coreos/clair/cmd/clairctl/database" "github.com/coreos/clair/cmd/clairctl/docker" "github.com/coreos/clair/cmd/clairctl/docker/httpclient" "github.com/wunderlist/moxy" ) // FilterFunc is a function that is called to process a proxy response // Since it has handle to the response object, it can manipulate the content type FilterFunc func(*http.Request, *http.Response) // onExitFlushLoop is a callback set by tests to detect the state of the // flushLoop() goroutine. var onExitFlushLoop func() // ReverseProxy is an HTTP Handler that takes an incoming request and // sends it to another server, proxying the response back to the // client. type ReverseProxy struct { // Director must be a function which modifies // the request into a new request to be sent // using Transport. Its response is then copied // back to the original client unmodified. Director func(*http.Request) // Filters must be an array of functions which modify // the response before the body is written Filters []FilterFunc // The transport used to perform proxy requests. // If nil, http.DefaultTransport is used. Transport http.RoundTripper // FlushInterval specifies the flush interval // to flush to the client while copying the // response body. // If zero, no periodic flushing is done. FlushInterval time.Duration // ErrorLog specifies an optional logger for errors // that occur when attempting to proxy the request. // If nil, logging goes to os.Stderr via the log package's // standard logger. ErrorLog *log.Logger } func copyHeader(dst, src http.Header) { for k, vv := range src { for _, v := range vv { dst.Add(k, v) } } } // Hop-by-hop headers. These are removed when sent to the backend. // http://www.w3.org/Protocols/rfc2616/rfc2616-sec13.html var hopHeaders = []string{ "Connection", "Keep-Alive", "Proxy-Authenticate", "Proxy-Authorization", "Te", // canonicalized version of "TE" "Trailers", "Transfer-Encoding", "Upgrade", } func (p *ReverseProxy) ServeHTTP(rw http.ResponseWriter, req *http.Request) { transport := p.Transport if transport == nil { transport = http.DefaultTransport } outreq := new(http.Request) *outreq = *req // includes shallow copies of maps, but okay context.Set(outreq, "in_req", req) p.Director(outreq) outreq.Proto = "HTTP/1.1" outreq.ProtoMajor = 1 outreq.ProtoMinor = 1 outreq.Close = false // Remove hop-by-hop headers to the backend. Especially // important is "Connection" because we want a persistent // connection, regardless of what the client sent to us. This // is modifying the same underlying map from req (shallow // copied above) so we only copy it if necessary. copiedHeaders := false for _, h := range hopHeaders { if outreq.Header.Get(h) != "" { if !copiedHeaders { outreq.Header = make(http.Header) copyHeader(outreq.Header, req.Header) copiedHeaders = true } outreq.Header.Del(h) } } if clientIP, _, err := net.SplitHostPort(req.RemoteAddr); err == nil { // If we aren't the first proxy retain prior // X-Forwarded-For information as a comma+space // separated list and fold multiple headers into one. if prior, ok := outreq.Header["X-Forwarded-For"]; ok { clientIP = strings.Join(prior, ", ") + ", " + clientIP } outreq.Header.Set("X-Forwarded-For", clientIP) } res, err := transport.RoundTrip(outreq) if err != nil { logrus.Errorf("http: proxy error: %v", err) rw.WriteHeader(http.StatusInternalServerError) return } defer res.Body.Close() for _, filterFn := range p.Filters { filterFn(req, res) } for _, h := range hopHeaders { res.Header.Del(h) } copyHeader(rw.Header(), res.Header) rw.WriteHeader(res.StatusCode) p.copyResponse(rw, res.Body) } func (p *ReverseProxy) copyResponse(dst io.Writer, src io.Reader) { if p.FlushInterval != 0 { if wf, ok := dst.(writeFlusher); ok { mlw := &maxLatencyWriter{ dst: wf, latency: p.FlushInterval, done: make(chan bool), } go mlw.flushLoop() defer mlw.stop() dst = mlw } } io.Copy(dst, src) } type writeFlusher interface { io.Writer http.Flusher } type maxLatencyWriter struct { dst writeFlusher latency time.Duration lk sync.Mutex // protects Write + Flush done chan bool } func (m *maxLatencyWriter) Write(p []byte) (int, error) { m.lk.Lock() defer m.lk.Unlock() return m.dst.Write(p) } func (m *maxLatencyWriter) flushLoop() { t := time.NewTicker(m.latency) defer t.Stop() for { select { case <-m.done: if onExitFlushLoop != nil { onExitFlushLoop() } return case <-t.C: m.lk.Lock() m.dst.Flush() m.lk.Unlock() } } } func (m *maxLatencyWriter) stop() { m.done <- true } // NewReverseProxy returns a new ReverseProxy that load-balances the proxy requests between multiple hosts defined by the RegistryMapping in the database // It also allows to define a chain of filter functions to process the outgoing response(s) func NewReverseProxy(filters []FilterFunc) *ReverseProxy { director := func(request *http.Request) { inr := context.Get(request, "in_req").(*http.Request) host, _ := database.GetRegistryMapping(mux.Vars(inr)["digest"]) out, _ := url.Parse(host) request.URL.Scheme = out.Scheme request.URL.Host = out.Host client := httpclient.Get() req, _ := http.NewRequest("HEAD", request.URL.String(), nil) resp, err := client.Do(req) if err != nil { logrus.Errorf("response error: %v", err) return } if resp.StatusCode == http.StatusUnauthorized { logrus.Info("pull from clair is unauthorized") docker.AuthenticateResponse(resp, request) } r, _ := http.NewRequest("GET", request.URL.String(), nil) r.Header.Set("Authorization", request.Header.Get("Authorization")) r.Header.Set("Accept-Encoding", request.Header.Get("Accept-Encoding")) *request = *r } return &ReverseProxy{ Transport: moxy.NewTransport(), Director: director, Filters: filters, } }