sink.go

package service

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"github.com/cortezaproject/corteza/server/pkg/actionlog"
	"github.com/cortezaproject/corteza/server/pkg/api"
	internalAuth "github.com/cortezaproject/corteza/server/pkg/auth"
	"github.com/cortezaproject/corteza/server/pkg/eventbus"
	"github.com/cortezaproject/corteza/server/system/service/event"
	"github.com/cortezaproject/corteza/server/system/types"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"
)

type (
	sink struct {
		signer     internalAuth.Signer
		actionlog  actionlog.Recorder
		eventbus   sinkEventDispatcher
		isMonolith bool
	}

	SinkRequestUrlParams struct {
		// Expect sink request to be of this method
		Method string `json:"mtd,omitempty"`

		// Origin is as an identifier, no validation of request params
		Origin string `json:"origin,omitempty"`

		// Optional, signature expiration
		Expires *time.Time `json:"exp,omitempty"`

		// When set it enables body processing (but limits it to that size!)
		MaxBodySize int64 `json:"mbs,omitempty"`

		// Acceptable content type
		ContentType string `json:"ct,omitempty"`

		Path string `json:"pt,omitempty"`

		// Should we put signature in the path (true)
		// or in query string (false, default)
		SignatureInPath bool `json:"sip,omitempty"`
	}

	sinkEventDispatcher interface {
		WaitFor(ctx context.Context, ev eventbus.Event) (err error)
	}
)

const (
	SinkContentTypeMail = "message/rfc822"

	// base url
	// we're using this for router, signature...
	SinkBaseURL = "/sink"

	// name of the parameter used for sink request signature
	SinkSignUrlParamName = "__sign"

	// delimiter between signature and payload
	SinkSignUrlParamDelimiter = "_"
)

func Sink() *sink {
	return &sink{
		actionlog:  DefaultActionlog,
		signer:     internalAuth.DefaultSigner,
		eventbus:   eventbus.Service(),
		isMonolith: true,
	}
}

// SignURL takes sink request parameters and generates signed URL
//
// With signed URL, external systems can make requests to sink subsystem
// and trigger scripts
func (svc sink) SignURL(srup SinkRequestUrlParams) (signedURL *url.URL, out SinkRequestUrlParams, err error) {
	var (
		params []byte
		sap    = &sinkActionProps{sinkParams: &srup}
		qs     = url.Values{}
	)

	err = func() error {
		// Append normalized path to the base URL
		srup.Path = svc.pathCleanup(srup.Path)
		path := svc.GetPath() + srup.Path

		srup.Method = strings.ToUpper(srup.Method)

		params, err = json.Marshal(srup)
		if err != nil {
			return SinkErrFailedToSign(sap).Wrap(err)
		}

		signature := svc.signer.Sign(0, params) + SinkSignUrlParamDelimiter + base64.StdEncoding.EncodeToString(params)

		if srup.SignatureInPath {
			// Optional, use path for sink signature
			path = fmt.Sprintf("%s/%s=%s", path, SinkSignUrlParamName, signature)
		} else {
			// By default put signature in query string
			qs.Set(SinkSignUrlParamName, signature)
		}

		signedURL = &url.URL{RawQuery: qs.Encode(), Path: path}
		return nil
	}()

	return signedURL, srup, svc.recordAction(context.Background(), sap, SinkActionSign, err)
}

func (svc sink) GetPath() string {
	path := ""

	if svc.isMonolith {
		path = "/system"
	}

	return path + SinkBaseURL
}

// pathCleanup removes base URL prefix and adds leading slash
func (svc sink) pathCleanup(p string) string {
	if len(p) > 0 {
		if pos := strings.Index(p, SinkBaseURL); pos > -1 {
			p = p[pos+len(SinkBaseURL):]
		}

		return "/" + strings.Trim(p, "/")
	}

	return ""
}

// ProcessRequest function is used directly in the HTTP controller
func (svc *sink) ProcessRequest(w http.ResponseWriter, r *http.Request) {
	var (
		ctx = r.Context()
		sap = &sinkActionProps{}
	)

	// capture error from request handling and process functions
	err := func() error {
		defer r.Body.Close()
		srup, err := svc.handleRequest(r)
		if err != nil {
			return err
		}

		var body io.Reader
		if srup.MaxBodySize > 0 {
			// Utilize body only when max-body-size limit is set
			body = http.MaxBytesReader(w, r.Body, srup.MaxBodySize)
		} else {
			body = http.MaxBytesReader(w, r.Body, 32<<10) // 32k limit
		}

		if err := svc.process(srup, w, r, body); err != nil {
			return SinkErrProcessingError(sap).Wrap(err)
		}

		return nil
	}()

	_ = svc.recordAction(ctx, sap, SinkActionRequest, err)
	if err != nil {
		// use standard facility for encoding errors for HTTP
		api.Send(w, r, err)
	}
}

// Verifies and extracts sink request params
func (svc sink) handleRequest(r *http.Request) (*SinkRequestUrlParams, error) {
	var (
		srup = &SinkRequestUrlParams{}
		sap  = &sinkActionProps{}
		qs   = r.URL.Query()

		signatureFoundInPath bool

		param string

		// this value is modified if signature is found in a path
		reqPath = r.URL.Path
	)

	// try to find a signature
	if _, has := qs[SinkSignUrlParamName]; has {
		// first, in a query string
		param = r.URL.Query().Get(SinkSignUrlParamName)
	} else if i := strings.Index(reqPath, SinkSignUrlParamName); i > -1 {
		// fallback to path, expecting signature to be at the end
		// offset string index by start of signature param name, length of param name, and = char
		param = reqPath[i+len(SinkSignUrlParamName)+1:]
		reqPath = reqPath[:i]

		// this is more for consistency and cleaner tests
		signatureFoundInPath = true
	}

	if len(param) == 0 {
		return nil, SinkErrMissingSignature(sap)
	}

	split := strings.SplitN(param, SinkSignUrlParamDelimiter, 2)
	if len(split) < 2 {
		return nil, SinkErrInvalidSignatureParam(sap)
	}

	params, err := base64.StdEncoding.DecodeString(split[1])
	if err != nil {
		return nil, SinkErrBadSinkParamEncoding(sap)
	}

	if !svc.signer.Verify(split[0], 0, params) {
		return nil, SinkErrInvalidSignature(sap)
	}

	if err = json.Unmarshal(params, srup); err != nil {
		// Impossible scenario :)
		// How can we have verified signature of an invalid JSON ?!
		return nil, SinkErrInvalidSinkRequestUrlParams(sap)
	}

	sap.setSinkParams(srup)

	if srup.SignatureInPath != signatureFoundInPath {
		return nil, SinkErrMisplacedSignature(sap)
	}

	if srup.Method != "" && srup.Method != r.Method {
		return nil, SinkErrInvalidHttpMethod(sap)
	}

	contentType := strings.ToLower(r.Header.Get("content-type"))
	if i := strings.Index(contentType, ";"); i > 0 {
		contentType = contentType[0 : i-1]
	}

	if srup.ContentType != "" {
		if strings.ToLower(srup.ContentType) != contentType {
			return nil, SinkErrInvalidContentType(sap)
		}
	}

	if srup.Path != "" {
		if srup.Path != svc.pathCleanup(reqPath) {
			return nil, SinkErrInvalidPath(sap)
		}
	}

	if srup.Expires != nil && srup.Expires.Before(time.Now()) {
		return nil, SinkErrSignatureExpired(sap)
	}

	if srup.MaxBodySize > 0 {
		// See if there is content length param and reject it right away
		if r.ContentLength > srup.MaxBodySize {
			return nil, SinkErrContentLengthExceedsMaxAllowedSize(sap)
		}
	}

	return srup, nil
}

// Processes sink request, casts it and forwards it to processor (depending on content type)
//
// Main reason for content-type & body to be passed separately (and not extracted from r param) is
// that:
// a) content type might be forced via sink params
// This is useful to enforce mail processing
// b) Max-body-size check might be limited via sink params
// and io.Reader that is passed is limited w/ io.LimitReader
//
func (svc *sink) process(srup *SinkRequestUrlParams, w http.ResponseWriter, r *http.Request, body io.Reader) error {
	var (
		err         error
		ctx         = r.Context()
		contentType = srup.ContentType
		sap         = &sinkActionProps{
			contentType: contentType,
		}
	)

	switch strings.ToLower(contentType) {
	case SinkContentTypeMail, "rfc822", "email", "mail":
		// this is handled by dedicated event that parses raw payload from HTTP request
		// as rfc882 message.
		var msg *types.MailMessage
		msg, err = types.NewMailMessage(body)
		if err != nil {
			return SinkErrFailedToCreateEvent(sap).Wrap(err)
		}

		sap.setMailHeader(&msg.Header)

		err = svc.eventbus.WaitFor(ctx, event.MailOnReceive(msg))
		if err != nil {
			return SinkErrFailedToProcess(sap).Wrap(err)
		}

	default:
		var (
			sr *types.SinkRequest

			// Predefine default response
			rsp = &types.SinkResponse{
				Status: http.StatusOK,
			}
		)

		// Sanitize URL
		sanitizedURL := r.URL

		// Step 1: removing sink sign url param
		sanitizedQuery := r.URL.Query()
		sanitizedQuery.Del(SinkSignUrlParamName)
		sanitizedURL.RawQuery = sanitizedQuery.Encode()

		// Step 2: remove prefix
		if i := strings.Index(sanitizedURL.Path, SinkBaseURL); i > -1 {
			sanitizedURL.Path = sanitizedURL.Path[i+len(SinkBaseURL):]
		}

		// Step 3: remove sink suffix if in path
		if srup.SignatureInPath {
			i := strings.Index(sanitizedURL.Path, SinkSignUrlParamName)
			if i > 0 {
				sanitizedURL.Path = sanitizedURL.Path[0 : i-1]
			}
		}

		r.URL = sanitizedURL
		r.RequestURI = sanitizedURL.String()

		sr, err = types.NewSinkRequest(r, body)
		if err != nil {
			return SinkErrFailedToCreateEvent(sap).Wrap(err)

		}

		sap.setUrl(sanitizedURL.String())

		err = svc.eventbus.WaitFor(ctx, event.SinkOnRequest(rsp, sr))
		if err != nil {
			return SinkErrFailedToProcess(sap).Wrap(err)
		}

		sap.setResponseStatus(rsp.Status)

		// Now write everything we've received from the script
		for k, vv := range rsp.Header {
			for _, v := range vv {
				w.Header().Add(k, v)
			}
		}

		w.WriteHeader(rsp.Status)

		var output []byte
		if bb, ok := rsp.Body.([]byte); ok {
			// Ok, handled
			output = bb
		} else if s, ok := rsp.Body.(string); ok {
			output = []byte(s)
		}

		if _, err = w.Write(output); err != nil {
			return SinkErrFailedToRespond(sap).Wrap(err)
		}
	}

	return nil
}