mux.go

/*
 * Copyright (c) 2017, MegaEase
 * All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http:https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package httpserver

import (
	"fmt"
	"io"
	"net"
	"net/http"
	"reflect"
	"regexp"
	"strings"
	"sync/atomic"

	lru "github.com/hashicorp/golang-lru"
	"github.com/megaease/easegress/pkg/object/globalfilter"
	"github.com/megaease/easegress/pkg/protocols/httpprot"

	"github.com/megaease/easegress/pkg/context"
	"github.com/megaease/easegress/pkg/logger"
	"github.com/megaease/easegress/pkg/object/autocertmanager"
	"github.com/megaease/easegress/pkg/protocols/httpprot/httpstat"
	"github.com/megaease/easegress/pkg/supervisor"
	"github.com/megaease/easegress/pkg/tracing"
	"github.com/megaease/easegress/pkg/util/fasttime"
	"github.com/megaease/easegress/pkg/util/ipfilter"
	"github.com/megaease/easegress/pkg/util/readers"
	"github.com/megaease/easegress/pkg/util/stringtool"
)

type (
	mux struct {
		httpStat *httpstat.HTTPStat
		topN     *httpstat.TopN

		inst atomic.Value // *muxInstance
	}

	muxInstance struct {
		superSpec *supervisor.Spec
		spec      *Spec
		httpStat  *httpstat.HTTPStat
		topN      *httpstat.TopN

		muxMapper context.MuxMapper

		cache *lru.ARCCache

		tracer       *tracing.Tracer
		ipFilter     *ipfilter.IPFilter
		ipFilterChan *ipfilter.IPFilters

		rules []*muxRule
	}

	muxRule struct {
		ipFilter      *ipfilter.IPFilter
		ipFilterChain *ipfilter.IPFilters

		host       string
		hostRegexp string
		hostRE     *regexp.Regexp
		paths      []*MuxPath
	}

	// MuxPath describes httpserver's path
	MuxPath struct {
		ipFilter      *ipfilter.IPFilter
		ipFilterChain *ipfilter.IPFilters

		path              string
		pathPrefix        string
		pathRegexp        string
		pathRE            *regexp.Regexp
		methods           []string
		rewriteTarget     string
		backend           string
		headers           []*Header
		clientMaxBodySize int64
		matchAllHeader    bool
	}

	route struct {
		code int
		path *MuxPath
	}
)

var (
	notFound         = &route{code: http.StatusNotFound}
	forbidden        = &route{code: http.StatusForbidden}
	methodNotAllowed = &route{code: http.StatusMethodNotAllowed}
	badRequest       = &route{code: http.StatusBadRequest}
)

// newIPFilterChain returns nil if the number of final filters is zero.
func newIPFilterChain(parentIPFilters *ipfilter.IPFilters, childSpec *ipfilter.Spec) *ipfilter.IPFilters {
	var ipFilters *ipfilter.IPFilters
	if parentIPFilters != nil {
		ipFilters = ipfilter.NewIPFilters(parentIPFilters.Filters()...)
	} else {
		ipFilters = ipfilter.NewIPFilters()
	}

	if childSpec != nil {
		ipFilters.Append(ipfilter.New(childSpec))
	}

	if len(ipFilters.Filters()) == 0 {
		return nil
	}

	return ipFilters
}

func newIPFilter(spec *ipfilter.Spec) *ipfilter.IPFilter {
	if spec == nil {
		return nil
	}

	return ipfilter.New(spec)
}

func allowIP(ipFilter *ipfilter.IPFilter, ip string) bool {
	if ipFilter == nil {
		return true
	}

	return ipFilter.Allow(ip)
}

func (mi *muxInstance) getRouteFromCache(req *httpprot.Request) *route {
	if mi.cache != nil {
		key := stringtool.Cat(req.Host(), req.Method(), req.Path())
		if value, ok := mi.cache.Get(key); ok {
			return value.(*route)
		}
	}
	return nil
}

func (mi *muxInstance) putRouteToCache(req *httpprot.Request, r *route) {
	if mi.cache != nil {
		key := stringtool.Cat(req.Host(), req.Method(), req.Path())
		mi.cache.Add(key, r)
	}
}

func newMuxRule(parentIPFilters *ipfilter.IPFilters, rule *Rule, paths []*MuxPath) *muxRule {
	var hostRE *regexp.Regexp

	if rule.HostRegexp != "" {
		var err error
		hostRE, err = regexp.Compile(rule.HostRegexp)
		// defensive programming
		if err != nil {
			logger.Errorf("BUG: compile %s failed: %v", rule.HostRegexp, err)
		}
	}

	return &muxRule{
		ipFilter:      newIPFilter(rule.IPFilter),
		ipFilterChain: newIPFilterChain(parentIPFilters, rule.IPFilter),

		host:       rule.Host,
		hostRegexp: rule.HostRegexp,
		hostRE:     hostRE,
		paths:      paths,
	}
}

func (mr *muxRule) match(r *httpprot.Request) bool {
	if mr.host == "" && mr.hostRE == nil {
		return true
	}

	host := r.Host()
	if h, _, err := net.SplitHostPort(host); err == nil {
		host = h
	}

	if mr.host != "" && mr.host == host {
		return true
	}
	if mr.hostRE != nil && mr.hostRE.MatchString(host) {
		return true
	}

	return false
}

func newMuxPath(parentIPFilters *ipfilter.IPFilters, path *Path) *MuxPath {
	var pathRE *regexp.Regexp
	if path.PathRegexp != "" {
		var err error
		pathRE, err = regexp.Compile(path.PathRegexp)
		// defensive programming
		if err != nil {
			logger.Errorf("BUG: compile %s failed: %v", path.PathRegexp, err)
		}
	}

	for _, p := range path.Headers {
		p.initHeaderRoute()
	}

	return &MuxPath{
		ipFilter:      newIPFilter(path.IPFilter),
		ipFilterChain: newIPFilterChain(parentIPFilters, path.IPFilter),

		path:              path.Path,
		pathPrefix:        path.PathPrefix,
		pathRegexp:        path.PathRegexp,
		pathRE:            pathRE,
		rewriteTarget:     path.RewriteTarget,
		methods:           path.Methods,
		backend:           path.Backend,
		headers:           path.Headers,
		clientMaxBodySize: path.ClientMaxBodySize,
		matchAllHeader:    path.MatchAllHeader,
	}
}

func (mp *MuxPath) matchPath(r *httpprot.Request) bool {
	if mp.path == "" && mp.pathPrefix == "" && mp.pathRE == nil {
		return true
	}

	path := r.Path()
	if mp.path != "" && mp.path == path {
		return true
	}
	if mp.pathPrefix != "" && strings.HasPrefix(path, mp.pathPrefix) {
		return true
	}
	if mp.pathRE != nil {
		return mp.pathRE.MatchString(path)
	}

	return false
}

func (mp *MuxPath) rewrite(r *httpprot.Request) {
	if mp.rewriteTarget == "" {
		return
	}

	path := r.Path()

	if mp.path != "" && mp.path == path {
		r.SetPath(mp.rewriteTarget)
		return
	}

	if mp.pathPrefix != "" && strings.HasPrefix(path, mp.pathPrefix) {
		path = mp.rewriteTarget + path[len(mp.pathPrefix):]
		r.SetPath(path)
		return
	}

	// sure (mp.pathRE != nil && mp.pathRE.MatchString(path)) is true
	path = mp.pathRE.ReplaceAllString(path, mp.rewriteTarget)
	r.SetPath(path)
}

func (mp *MuxPath) matchMethod(r *httpprot.Request) bool {
	if len(mp.methods) == 0 {
		return true
	}

	return stringtool.StrInSlice(r.Method(), mp.methods)
}

func (mp *MuxPath) matchHeaders(r *httpprot.Request) bool {
	if mp.matchAllHeader {
		for _, h := range mp.headers {
			v := r.HTTPHeader().Get(h.Key)
			if len(h.Values) > 0 && !stringtool.StrInSlice(v, h.Values) {
				return false
			}

			if h.Regexp != "" && !h.headerRE.MatchString(v) {
				return false
			}
		}
	} else {
		for _, h := range mp.headers {
			v := r.HTTPHeader().Get(h.Key)
			if stringtool.StrInSlice(v, h.Values) {
				return true
			}

			if h.Regexp != "" && h.headerRE.MatchString(v) {
				return true
			}
		}
	}

	return mp.matchAllHeader
}

func newMux(httpStat *httpstat.HTTPStat, topN *httpstat.TopN, mapper context.MuxMapper) *mux {
	m := &mux{
		httpStat: httpStat,
		topN:     topN,
	}

	m.inst.Store(&muxInstance{
		spec:      &Spec{},
		tracer:    tracing.NoopTracer,
		muxMapper: mapper,
		httpStat:  httpStat,
		topN:      topN,
	})

	return m
}

func (m *mux) reload(superSpec *supervisor.Spec, muxMapper context.MuxMapper) {
	spec := superSpec.ObjectSpec().(*Spec)

	tracer := tracing.NoopTracer
	oldInst := m.inst.Load().(*muxInstance)
	if !reflect.DeepEqual(oldInst.spec.Tracing, spec.Tracing) {
		defer func() {
			err := oldInst.tracer.Close()
			if err != nil {
				logger.Errorf("close tracing failed: %v", err)
			}
		}()
		tracer0, err := tracing.New(spec.Tracing)
		if err != nil {
			logger.Errorf("create tracing failed: %v", err)
		} else {
			tracer = tracer0
		}
	} else if oldInst.tracer != nil {
		tracer = oldInst.tracer
	}

	inst := &muxInstance{
		superSpec:    superSpec,
		spec:         spec,
		muxMapper:    muxMapper,
		httpStat:     m.httpStat,
		topN:         m.topN,
		ipFilter:     newIPFilter(spec.IPFilter),
		ipFilterChan: newIPFilterChain(nil, spec.IPFilter),
		rules:        make([]*muxRule, len(spec.Rules)),
		tracer:       tracer,
	}

	if spec.CacheSize > 0 {
		arc, err := lru.NewARC(int(spec.CacheSize))
		if err != nil {
			logger.Errorf("BUG: new arc cache failed: %v", err)
		}
		inst.cache = arc
	}

	for i := 0; i < len(inst.rules); i++ {
		specRule := spec.Rules[i]

		ruleIPFilterChain := newIPFilterChain(inst.ipFilterChan, specRule.IPFilter)

		paths := make([]*MuxPath, len(specRule.Paths))
		for j := 0; j < len(paths); j++ {
			paths[j] = newMuxPath(ruleIPFilterChain, specRule.Paths[j])
		}

		// NOTE: Given the parent ipFilters not its own.
		inst.rules[i] = newMuxRule(inst.ipFilterChan, specRule, paths)
	}

	m.inst.Store(inst)
}

func (m *mux) ServeHTTP(stdw http.ResponseWriter, stdr *http.Request) {
	// HTTP-01 challenges requires HTTP server to listen on port 80, but we
	// don't know which HTTP server listen on this port (consider there's an
	// nginx sitting in front of Easegress), so all HTTP servers need to
	// handle HTTP-01 challenges.
	if strings.HasPrefix(stdr.URL.Path, "/.well-known/acme-challenge/") {
		autocertmanager.HandleHTTP01Challenge(stdw, stdr)
		return
	}

	// Forward to the current muxInstance to handle the request.
	m.inst.Load().(*muxInstance).serveHTTP(stdw, stdr)
}

func buildFailureResponse(ctx *context.Context, statusCode int) *httpprot.Response {
	resp, _ := httpprot.NewResponse(nil)
	resp.SetStatusCode(statusCode)
	ctx.SetResponse(context.DefaultNamespace, resp)
	return resp
}

func (mi *muxInstance) sendResponse(ctx *context.Context, stdw http.ResponseWriter) (int, uint64) {
	var resp *httpprot.Response
	if v := ctx.GetResponse(context.DefaultNamespace); v == nil {
		logger.Errorf("%s: response is nil", mi.superSpec.Name())
		resp = buildFailureResponse(ctx, http.StatusServiceUnavailable)
	} else if r, ok := v.(*httpprot.Response); !ok {
		logger.Errorf("%s: expect an HTTP response", mi.superSpec.Name())
		resp = buildFailureResponse(ctx, http.StatusServiceUnavailable)
	} else {
		resp = r
	}

	// Send the response
	header := stdw.Header()
	for k, v := range resp.HTTPHeader() {
		header[k] = v
	}
	stdw.WriteHeader(resp.StatusCode())
	respBodySize, _ := io.Copy(stdw, resp.GetPayload())

	return resp.StatusCode(), uint64(respBodySize) + uint64(resp.MetaSize())
}

func (mi *muxInstance) serveHTTP(stdw http.ResponseWriter, stdr *http.Request) {
	// Replace the body of the original request with a ByteCountReader, so
	// that we can calculate the actual request size.
	body := readers.NewByteCountReader(stdr.Body)
	stdr.Body = body

	startAt := fasttime.Now()
	span := mi.tracer.NewSpanWithStart(mi.superSpec.Name(), startAt)
	ctx := context.New(span)
	ctx.SetData("HTTP_RESPONSE_WRITER", stdw)

	// httpprot.NewRequest never returns an error.
	req, _ := httpprot.NewRequest(stdr)

	// Calculate the meta size now, as everything could be modified.
	reqMetaSize := req.MetaSize()
	ctx.SetRequest(context.DefaultNamespace, req)

	// get topN here, as the path could be modified later.
	topN := mi.topN.Stat(req.Path())

	defer func() {
		metric, _ := ctx.GetData("HTTP_METRIC").(*httpstat.Metric)

		if metric == nil {
			statusCode, respSize := mi.sendResponse(ctx, stdw)
			ctx.Finish()

			// Drain off the body if it has not been, so that we can get the
			// correct body size.
			io.Copy(io.Discard, body)

			metric = &httpstat.Metric{
				StatusCode: statusCode,
				ReqSize:    uint64(reqMetaSize) + uint64(body.BytesRead()),
				RespSize:   respSize,
			}
		} else { // hijacked, websocket and etc.
			ctx.Finish()
		}

		metric.Duration = fasttime.Since(startAt)
		topN.Stat(metric)
		mi.httpStat.Stat(metric)

		span.Finish()

		// Write access log.
		logger.LazyHTTPAccess(func() string {
			// log format:
			//
			// [$startTime]
			// [$remoteAddr $realIP $method $requestURL $proto $statusCode]
			// [$contextDuration $readBytes $writeBytes]
			// [$tags]
			const logFmt = "[%s] [%s %s %s %s %s %d] [%v rx:%dB tx:%dB] [%s]"
			return fmt.Sprintf(logFmt,
				fasttime.Format(startAt, fasttime.RFC3339Milli),
				stdr.RemoteAddr, req.RealIP(), stdr.Method, stdr.RequestURI,
				stdr.Proto, metric.StatusCode, metric.Duration, metric.ReqSize,
				metric.RespSize, ctx.Tags())
		})
	}()

	route := mi.search(req)
	if route.code != 0 {
		logger.Debugf("%s: status code of result route for [%s %s]: %d", mi.superSpec.Name(), req.Method(), req.RequestURI, route.code)
		buildFailureResponse(ctx, route.code)
		return
	}

	handler, ok := mi.muxMapper.GetHandler(route.path.backend)
	if !ok {
		logger.Debugf("%s: backend(Pipeline) %q for [%s %s] not found", mi.superSpec.Name(), req.Method(), req.RequestURI, route.path.backend)
		buildFailureResponse(ctx, http.StatusServiceUnavailable)
		return
	}
	logger.Debugf("%s: the matched backend(Pipeline) for [%s %s] is %q", mi.superSpec.Name(), req.Method(), req.RequestURI, route.path.backend)

	route.path.rewrite(req)
	if mi.spec.XForwardedFor {
		appendXForwardedFor(req)
	}

	maxBodySize := route.path.clientMaxBodySize
	if maxBodySize == 0 {
		maxBodySize = mi.spec.ClientMaxBodySize
	}
	err := req.FetchPayload(maxBodySize)
	if err == httpprot.ErrRequestEntityTooLarge {
		logger.Debugf("%s: %s", mi.superSpec.Name(), err.Error())
		buildFailureResponse(ctx, http.StatusRequestEntityTooLarge)
		return
	}
	if err != nil {
		logger.Debugf("%s: failed to read request body: %v", mi.superSpec.Name(), err)
		buildFailureResponse(ctx, http.StatusBadRequest)
		return
	}

	// global filter
	globalFilter := mi.getGlobalFilter()
	if globalFilter == nil {
		handler.Handle(ctx)
	} else {
		globalFilter.Handle(ctx, handler)
	}
}

func (mi *muxInstance) search(req *httpprot.Request) *route {
	headerMismatch, methodMismatch := false, false

	ip := req.RealIP()

	// The key of the cache is req.Host + req.Method + req.URL.Path,
	// and if a path is cached, we are sure it does not contain any
	// headers.
	r := mi.getRouteFromCache(req)
	if r != nil {
		if r.code != 0 {
			return r
		}
		if r.path.ipFilterChain == nil {
			return r
		}
		if r.path.ipFilterChain.Allow(ip) {
			return r
		}
		return forbidden
	}

	if !allowIP(mi.ipFilter, ip) {
		return forbidden
	}

	for _, host := range mi.rules {
		if !host.match(req) {
			continue
		}

		if !allowIP(host.ipFilter, ip) {
			return forbidden
		}

		for _, path := range host.paths {
			if !path.matchPath(req) {
				continue
			}

			if !path.matchMethod(req) {
				methodMismatch = true
				continue
			}

			// The path can be put into the cache if it has no headers.
			if len(path.headers) == 0 {
				r = &route{code: 0, path: path}
				mi.putRouteToCache(req, r)
			} else if !path.matchHeaders(req) {
				headerMismatch = true
				continue
			}

			if !allowIP(path.ipFilter, ip) {
				return forbidden
			}

			return &route{code: 0, path: path}
		}
	}

	if headerMismatch {
		return badRequest
	}

	if methodMismatch {
		mi.putRouteToCache(req, methodNotAllowed)
		return methodNotAllowed
	}

	mi.putRouteToCache(req, notFound)
	return notFound
}

func appendXForwardedFor(r *httpprot.Request) {
	const xForwardedFor = "X-Forwarded-For"

	v := r.HTTPHeader().Get(xForwardedFor)
	ip := r.RealIP()

	if v == "" {
		r.Header().Add(xForwardedFor, ip)
		return
	}

	if !strings.Contains(v, ip) {
		v = stringtool.Cat(v, ",", ip)
		r.Header().Set(xForwardedFor, v)
	}
}

func (mi *muxInstance) getGlobalFilter() *globalfilter.GlobalFilter {
	if mi.spec.GlobalFilter == "" {
		return nil
	}
	globalFilter, ok := mi.superSpec.Super().GetBusinessController(mi.spec.GlobalFilter)
	if globalFilter == nil || !ok {
		return nil
	}
	globalFilterInstance, ok := globalFilter.Instance().(*globalfilter.GlobalFilter)
	if !ok {
		return nil
	}
	return globalFilterInstance
}

func (mi *muxInstance) close() {
	if err := mi.tracer.Close(); err != nil {
		logger.Errorf("%s close tracer failed: %v", mi.superSpec.Name(), err)
	}
}

func (m *mux) close() {
	m.inst.Load().(*muxInstance).close()
}