weave/lb/gateway_proxy.go

package lb

import (
	"context"
	"crypto/tls"
	"fmt"
	"log"
	"math/rand"
	"net"
	"net/http"
	"net/http/httputil"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	coreapi "k8s.io/api/core/v1"

	"github.com/rajch/weave/common"
)

// Proxy serves the HTTP and HTTPS listeners of the Gateways this node holds. TLS is terminated with the
// listener's certificate, chosen by SNI and swapped in place when its Secret changes (no restart, no rebind).
// Requests are routed by the attached HTTPRoutes in the order the spec prescribes and passed to a ready endpoint
// of the chosen backend, with the client's own address in X-Forwarded-For. Bodies are streamed, never limited.
type Proxy struct {
	mu      sync.Mutex
	servers map[string]*portServer // by "address:port"
	// listen binds an address (swappable in tests)
	listen    func(addr string) (net.Listener, error)
	transport http.RoundTripper
	certs     map[string]*tls.Certificate // parsed certificates, by Secret uid@resourceVersion
	rr        sync.Map                    // backend key -> *uint64, for round robin over its endpoints
}

func NewProxy() *Proxy {
	return &Proxy{
		servers: map[string]*portServer{},
		certs:   map[string]*tls.Certificate{},
		listen:  func(addr string) (net.Listener, error) { return net.Listen("tcp", addr) },
		transport: &http.Transport{
			DialContext:           (&net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}).DialContext,
			MaxIdleConns:          256,
			MaxIdleConnsPerHost:   32,
			IdleConnTimeout:       90 * time.Second,
			ExpectContinueTimeout: time.Second,
		},
	}
}

// portServer is one bound address:port and what it currently serves.
type portServer struct {
	addr  string
	https bool
	ln    net.Listener
	srv   *http.Server
	cfg   atomic.Pointer[portConfig]
}

// portConfig is everything one address:port serves: its listeners, most specific hostname first.
type portConfig struct {
	listeners []*liveListener
}

type liveListener struct {
	spec ServeListener
	cert *tls.Certificate
}

// Apply serves exactly these listeners: ports no longer listed are closed, new ones bound, and every served
// port's routes and certificates replaced at once. It returns, by "namespace/gateway/listener", the listeners
// it could not bind.
func (p *Proxy) Apply(listeners []ServeListener) map[string]error {
	p.mu.Lock()
	defer p.mu.Unlock()
	failed := map[string]error{}
	byAddr := map[string][]ServeListener{}
	for _, l := range listeners {
		a := net.JoinHostPort(l.Address.String(), strconv.Itoa(int(l.Port)))
		byAddr[a] = append(byAddr[a], l)
	}
	for a, s := range p.servers {
		if group, ok := byAddr[a]; !ok || group[0].HTTPS != s.https {
			s.close()
			delete(p.servers, a)
		}
	}
	used := map[string]bool{}
	for a, group := range byAddr {
		cfg := p.compile(group, used)
		s := p.servers[a]
		if s == nil {
			var err error
			if s, err = p.start(a, group[0].HTTPS); err != nil {
				for _, l := range group {
					failed[l.Gateway+"/"+l.Name] = err
				}
				continue
			}
			p.servers[a] = s
		}
		s.cfg.Store(cfg)
	}
	// forget certificates no listener uses any more
	for k := range p.certs {
		if !used[k] {
			delete(p.certs, k)
		}
	}
	return failed
}

// Close stops serving everything.
func (p *Proxy) Close() {
	p.Apply(nil)
}

func (p *Proxy) compile(group []ServeListener, used map[string]bool) *portConfig {
	cfg := &portConfig{}
	for _, l := range group {
		ll := &liveListener{spec: l}
		if l.HTTPS && l.Cert != nil {
			ll.cert = p.cert(l.Cert)
			used[certKey(l.Cert)] = true
		}
		cfg.listeners = append(cfg.listeners, ll)
	}
	// most specific first: exact hostnames, then wildcards (longest first), then the catch-all
	sort.SliceStable(cfg.listeners, func(i, j int) bool {
		return hostRank(cfg.listeners[i].spec.Hostname) > hostRank(cfg.listeners[j].spec.Hostname)
	})
	return cfg
}

func certKey(s *coreapi.Secret) string {
	return string(s.UID) + "@" + s.ResourceVersion + "/" + s.Namespace + "/" + s.Name
}

// cert parses a Secret's certificate once per version of the Secret.
func (p *Proxy) cert(s *coreapi.Secret) *tls.Certificate {
	k := certKey(s)
	if c, ok := p.certs[k]; ok {
		return c
	}
	c, err := tls.X509KeyPair(s.Data[coreapi.TLSCertKey], s.Data[coreapi.TLSPrivateKeyKey])
	if err != nil {
		common.Log.Warnf("lb: gateway certificate %s/%s: %v", s.Namespace, s.Name, err)
		return nil
	}
	p.certs[k] = &c
	return &c
}

// hostRank orders listener hostnames: exact (by length) above wildcards (by length) above none.
func hostRank(h string) int {
	switch {
	case h == "":
		return 0
	case strings.HasPrefix(h, "*."):
		return 1000 + len(h)
	}
	return 100000 + len(h)
}

func (p *Proxy) start(addr string, https bool) (*portServer, error) {
	ln, err := p.listen(addr)
	if err != nil {
		return nil, err
	}
	s := &portServer{addr: addr, https: https, ln: ln}
	s.cfg.Store(&portConfig{})
	s.srv = &http.Server{
		Handler:           http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { p.serve(s, w, r) }),
		ReadHeaderTimeout: 30 * time.Second,
		IdleTimeout:       2 * time.Minute,
		// handshakes from scanners and the like would flood the log
		ErrorLog: log.New(debugWriter{}, "lb: gateway "+addr+": ", 0),
	}
	if https {
		s.srv.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12, GetCertificate: s.certificate}
		go func() {
			if err := s.srv.ServeTLS(ln, "", ""); err != nil && err != http.ErrServerClosed {
				common.Log.Errorf("lb: gateway %s: %v", addr, err)
			}
		}()
	} else {
		go func() {
			if err := s.srv.Serve(ln); err != nil && err != http.ErrServerClosed {
				common.Log.Errorf("lb: gateway %s: %v", addr, err)
			}
		}()
	}
	common.Log.Infof("lb: gateway serving %s (%s)", addr, map[bool]string{true: "HTTPS", false: "HTTP"}[https])
	return s, nil
}

func (s *portServer) close() {
	srv := s.srv
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		_ = srv.Shutdown(ctx)
		_ = srv.Close()
	}()
	common.Log.Infof("lb: gateway no longer serving %s", s.addr)
}

type debugWriter struct{}

func (debugWriter) Write(b []byte) (int, error) {
	common.Log.Debugf("%s", strings.TrimRight(string(b), "\n"))
	return len(b), nil
}

// listenerFor picks the listener a name (SNI or Host) reaches on a port, most specific hostname first.
func (c *portConfig) listenerFor(name string) *liveListener {
	name = strings.ToLower(name)
	for _, l := range c.listeners {
		if l.spec.Hostname == "" || hostMatches(l.spec.Hostname, name) {
			return l
		}
	}
	return nil
}

// certificate answers a TLS handshake with the certificate of the listener its SNI names.
func (s *portServer) certificate(hello *tls.ClientHelloInfo) (*tls.Certificate, error) {
	cfg := s.cfg.Load()
	l := cfg.listenerFor(hello.ServerName)
	if l == nil || l.cert == nil {
		return nil, fmt.Errorf("no listener for %q", hello.ServerName)
	}
	return l.cert, nil
}

// candidate is one match of one rule that a request satisfies, with what decides its precedence.
type candidate struct {
	route   *ServeRoute
	rule    *ServeRule
	ruleIdx int
	// the matched hostname: characters if not a wildcard, and characters
	hostExact, hostLen int
	exact              bool
	prefixLen          int
	method             bool
	headers, query     int
	prefix             string // the path prefix matched (for ReplacePrefixMatch)
}

func (p *Proxy) serve(s *portServer, w http.ResponseWriter, r *http.Request) {
	cfg := s.cfg.Load()
	host := hostOnly(r.Host)
	name := host
	if r.TLS != nil && r.TLS.ServerName != "" {
		name = r.TLS.ServerName
	}
	l := cfg.listenerFor(name)
	if l == nil {
		http.Error(w, "no listener for this host", http.StatusNotFound)
		return
	}
	c := pick(l, host, r)
	if c == nil {
		http.Error(w, "no route", http.StatusNotFound)
		return
	}
	for _, f := range c.rule.Filters {
		if f.Type == filterRequestRedirect && f.RequestRedirect != nil {
			to, code := redirect(r, f.RequestRedirect, l.spec.Port, c.prefix)
			http.Redirect(w, r, to, code)
			return
		}
	}
	b := chooseBackend(c.rule.Backends)
	if b == nil {
		// no resolved backend to send it to: the spec's answer for unresolved backends
		http.Error(w, "no usable backend", http.StatusInternalServerError)
		return
	}
	if len(b.Endpoints) == 0 {
		http.Error(w, "no ready endpoint", http.StatusServiceUnavailable)
		return
	}
	n, _ := p.rr.LoadOrStore(b.Key, new(uint64))
	ep := b.Endpoints[int(atomic.AddUint64(n.(*uint64), 1)-1)%len(b.Endpoints)]
	filters := append(append([]HTTPRouteFilter(nil), c.rule.Filters...), b.Filters...)
	rp := &httputil.ReverseProxy{
		Transport:     p.transport,
		FlushInterval: -1, // stream as it comes (git, server-sent events)
		Rewrite: func(pr *httputil.ProxyRequest) {
			pr.SetURL(&url.URL{Scheme: "http", Host: ep.String()})
			pr.Out.Host = pr.In.Host // forwarded unmodified unless a filter says otherwise
			for _, f := range filters {
				switch {
				case f.Type == filterRequestHeaderModifier && f.RequestHeaderModifier != nil:
					modifyHeaders(pr.Out.Header, f.RequestHeaderModifier)
				case f.Type == filterURLRewrite && f.URLRewrite != nil:
					if f.URLRewrite.Hostname != nil {
						pr.Out.Host = *f.URLRewrite.Hostname
					}
					if f.URLRewrite.Path != nil {
						pr.Out.URL.Path = rewritePath(pr.In.URL.Path, f.URLRewrite.Path, c.prefix)
						pr.Out.URL.RawPath = ""
					}
				}
			}
			// After the filters, so none can change these (the model already refuses routes that try): only this
			// hop's view of the connection is passed on. Cleared first, as SetXForwarded appends to an
			// X-Forwarded-For already there.
			for _, h := range forwardingHeaders {
				pr.Out.Header.Del(h)
			}
			pr.SetXForwarded()
			if ip, _, err := net.SplitHostPort(pr.In.RemoteAddr); err == nil {
				pr.Out.Header.Set("X-Real-IP", ip)
			}
		},
		ModifyResponse: func(resp *http.Response) error {
			for _, f := range filters {
				if f.Type == filterResponseHeaderModifier && f.ResponseHeaderModifier != nil {
					modifyHeaders(resp.Header, f.ResponseHeaderModifier)
				}
			}
			return nil
		},
		ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) {
			common.Log.Debugf("lb: gateway %s -> %s: %v", s.addr, ep, err)
			w.WriteHeader(http.StatusBadGateway)
		},
	}
	rp.ServeHTTP(w, r)
}

// pick is the match that takes the request, by the spec's precedence: the most specific matching hostname, then
// an exact path, the longest prefix, a method, the most header matches, the most query matches, then the oldest
// route, its namespace/name, and the first rule.
func pick(l *liveListener, host string, r *http.Request) *candidate {
	var best *candidate
	for ri := range l.spec.Routes {
		route := &l.spec.Routes[ri]
		hostExact, hostLen, ok := routeHost(route, l.spec.Hostname, host)
		if !ok {
			continue
		}
		for i := range route.Rules {
			rule := &route.Rules[i]
			for _, m := range rule.Matches {
				c, ok := matches(m, r)
				if !ok {
					continue
				}
				c.route, c.rule, c.ruleIdx, c.hostExact, c.hostLen = route, rule, i, hostExact, hostLen
				if best == nil || better(c, *best) {
					cc := c
					best = &cc
				}
			}
		}
	}
	return best
}

func better(a, b candidate) bool {
	switch {
	case a.hostExact != b.hostExact:
		return a.hostExact > b.hostExact
	case a.hostLen != b.hostLen:
		return a.hostLen > b.hostLen
	case a.exact != b.exact:
		return a.exact
	case a.prefixLen != b.prefixLen:
		return a.prefixLen > b.prefixLen
	case a.method != b.method:
		return a.method
	case a.headers != b.headers:
		return a.headers > b.headers
	case a.query != b.query:
		return a.query > b.query
	case !a.route.Created.Equal(b.route.Created):
		return a.route.Created.Before(b.route.Created)
	case a.route.Key != b.route.Key:
		return a.route.Key < b.route.Key
	}
	return a.ruleIdx < b.ruleIdx
}

// routeHost reports whether a route answers this host on its listener, and how specifically.
func routeHost(route *ServeRoute, listenerHost, host string) (int, int, bool) {
	names := route.Hostnames
	if len(names) == 0 {
		if listenerHost == "" {
			return 0, 0, true
		}
		names = []string{listenerHost}
	}
	exact, length, ok := 0, 0, false
	for _, h := range names {
		if !hostMatches(h, host) {
			continue
		}
		ok = true
		e := 0
		if !strings.HasPrefix(h, "*.") {
			e = len(h)
		}
		if e > exact || (e == exact && len(h) > length) {
			exact, length = e, len(h)
		}
	}
	return exact, length, ok
}

func matches(m HTTPRouteMatch, r *http.Request) (candidate, bool) {
	var c candidate
	path, value := r.URL.Path, str(m.Path.Value)
	switch str(m.Path.Type) {
	case pathExact:
		if path != value {
			return c, false
		}
		c.exact = true
		c.prefixLen = len(value)
	default:
		trimmed := strings.TrimSuffix(value, "/")
		if trimmed != "" && path != trimmed && !strings.HasPrefix(path, trimmed+"/") {
			return c, false
		}
		c.prefixLen = len(value)
		c.prefix = value
	}
	if m.Method != nil {
		if r.Method != *m.Method {
			return c, false
		}
		c.method = true
	}
	for _, h := range m.Headers {
		if r.Header.Get(h.Name) != h.Value {
			return c, false
		}
	}
	c.headers = len(m.Headers)
	q := r.URL.Query()
	for _, qp := range m.QueryParams {
		if q.Get(qp.Name) != qp.Value {
			return c, false
		}
	}
	c.query = len(m.QueryParams)
	return c, true
}

// chooseBackend picks a backend by weight (nil when there is none to pick).
func chooseBackend(bs []ServeBackend) *ServeBackend {
	total := int32(0)
	for _, b := range bs {
		if b.Weight > 0 {
			total += b.Weight
		}
	}
	if total == 0 {
		return nil
	}
	n := rand.Int31n(total)
	for i := range bs {
		if bs[i].Weight <= 0 {
			continue
		}
		if n < bs[i].Weight {
			return &bs[i]
		}
		n -= bs[i].Weight
	}
	return nil
}

// redirect is where a RequestRedirect filter sends a request, and with which status.
func redirect(r *http.Request, f *HTTPRequestRedirectFilter, listenerPort int32, prefix string) (string, int) {
	scheme := "http"
	if r.TLS != nil {
		scheme = "https"
	}
	port := int(listenerPort)
	if f.Scheme != nil {
		// a new scheme brings its well-known port, unless a port is given
		scheme = *f.Scheme
		port = map[string]int{"http": 80, "https": 443}[scheme]
	}
	if f.Port != nil {
		port = int(*f.Port)
	}
	host := hostOnly(r.Host)
	if f.Hostname != nil {
		host = *f.Hostname
	}
	if !(scheme == "http" && port == 80) && !(scheme == "https" && port == 443) && port != 0 {
		host = net.JoinHostPort(host, strconv.Itoa(port))
	}
	path := r.URL.Path
	if f.Path != nil {
		path = rewritePath(path, f.Path, prefix)
	}
	code := http.StatusFound
	if f.StatusCode != nil {
		code = *f.StatusCode
	}
	return (&url.URL{Scheme: scheme, Host: host, Path: path, RawQuery: r.URL.RawQuery}).String(), code
}

// rewritePath applies a path modifier: the whole path, or the matched prefix replaced.
func rewritePath(path string, m *HTTPPathModifier, prefix string) string {
	switch m.Type {
	case "ReplaceFullPath":
		return str(m.ReplaceFullPath)
	case "ReplacePrefixMatch":
		rest := strings.TrimPrefix(path, strings.TrimSuffix(prefix, "/"))
		out := strings.TrimSuffix(str(m.ReplacePrefixMatch), "/") + rest
		if !strings.HasPrefix(out, "/") {
			out = "/" + out
		}
		return out
	}
	return path
}

func modifyHeaders(h http.Header, f *HTTPHeaderFilter) {
	for _, s := range f.Set {
		h.Set(s.Name, s.Value)
	}
	for _, a := range f.Add {
		h.Add(a.Name, a.Value)
	}
	for _, name := range f.Remove {
		h.Del(name)
	}
}

// hostOnly is a Host header without its port, lowercased (the spec ignores the port when matching).
func hostOnly(h string) string {
	if host, _, err := net.SplitHostPort(h); err == nil {
		h = host
	}
	return strings.ToLower(strings.TrimSuffix(h, "."))
}