From e8b81396effe09a582a3f265dd0b802af319a6cb Mon Sep 17 00:00:00 2001 From: Denis Arh Date: Tue, 26 May 2020 08:06:44 +0200 Subject: [PATCH] Support for HTTP errors, add sink impl. --- codegen/v2/actionlog/actions.gen.go.tpl | 23 + codegen/v2/actionlog/main.go | 11 + system/service/sink.go | 172 +++-- system/service/sink_actions.gen.go | 929 ++++++++++++++++++++++++ system/service/sink_actions.yaml | 93 +++ system/service/sink_test.go | 177 +++++ 6 files changed, 1336 insertions(+), 69 deletions(-) create mode 100644 system/service/sink_actions.gen.go create mode 100644 system/service/sink_actions.yaml create mode 100644 system/service/sink_test.go diff --git a/codegen/v2/actionlog/actions.gen.go.tpl b/codegen/v2/actionlog/actions.gen.go.tpl index 3de985db7..71a3047cb 100644 --- a/codegen/v2/actionlog/actions.gen.go.tpl +++ b/codegen/v2/actionlog/actions.gen.go.tpl @@ -9,6 +9,9 @@ import ( "strings" "errors" "time" +{{- if $.SupportHttpErrors }} + "net/http" +{{- end }} "github.com/cortezaproject/corteza-server/pkg/actionlog" {{- range $import := $.Import }} @@ -51,6 +54,10 @@ type ( wrap error props *{{ $.Service }}ActionProps + + {{ if $.SupportHttpErrors }} + httpStatusCode int + {{ end }} } {{ end }} ) @@ -255,6 +262,18 @@ func (e *{{ $.Service }}Error) LoggableAction() *actionlog.Action { Meta: e.props.serialize(), } } + +{{ if $.SupportHttpErrors }} +func (e *{{ $.Service }}Error) HttpResponse(w http.ResponseWriter) { + var code = e.httpStatusCode + if code == 0 { + code = http.StatusInternalServerError + } + + http.Error(w, e.message, code) +} +{{ end }} + {{ end }} {{ if $.Actions }} @@ -313,6 +332,10 @@ func {{ camelCase "" $.Service "Err" $e.Error }}(props ... *{{ $.Service }}Actio log: "{{ $e.Log }}", severity: {{ $e.SeverityConstName }}, props: func() *{{ $.Service }}ActionProps { if len(props) > 0 { return props[0] }; return nil}(), + + {{ if $e.HttpStatus }} + httpStatusCode: http.{{ $e.HttpStatus }}, + {{ end }} } if len(props) > 0 { diff --git a/codegen/v2/actionlog/main.go b/codegen/v2/actionlog/main.go index 1faf42b33..8090fc147 100644 --- a/codegen/v2/actionlog/main.go +++ b/codegen/v2/actionlog/main.go @@ -65,6 +65,9 @@ type ( // Error severity Severity string `yaml:"severity"` + + // HTTP Status code for this error + HttpStatus string `yaml:"httpStatus"` } ) @@ -132,6 +135,10 @@ func procDef(path, output string) { // Default severity for errors DefaultErrorSeverity string `yaml:"defaultErrorSeverity"` + // If at least one of the errors has HTTP status defined, + // add support for http errors + SupportHttpErrors bool + Props []*propsDef Actions []*actionDef Errors []*errorDef @@ -209,6 +216,10 @@ func procDef(path, output string) { if e.Severity == "" { e.Severity = tplData.DefaultErrorSeverity } + + if e.HttpStatus != "" { + tplData.SupportHttpErrors = true + } } checkHandle := func(s string) { diff --git a/system/service/sink.go b/system/service/sink.go index 621bbcf9c..2b5378efd 100644 --- a/system/service/sink.go +++ b/system/service/sink.go @@ -10,20 +10,17 @@ import ( "strings" "time" - "go.uber.org/zap" - "go.uber.org/zap/zapcore" - + "github.com/cortezaproject/corteza-server/pkg/actionlog" internalAuth "github.com/cortezaproject/corteza-server/pkg/auth" "github.com/cortezaproject/corteza-server/pkg/eventbus" - "github.com/cortezaproject/corteza-server/pkg/logger" "github.com/cortezaproject/corteza-server/system/service/event" "github.com/cortezaproject/corteza-server/system/types" ) type ( sink struct { - logger *zap.Logger signer internalAuth.Signer + actionlog actionlog.Recorder eventbus sinkEventDispatcher isMonolith bool } @@ -51,9 +48,6 @@ type ( ) const ( - ErrSinkContentTypeUnsupported serviceError = "SinkUnsupportedContentType" - ErrSinkContentProcessingFailed serviceError = "SinkProcessFailed" - SinkContentTypeMail = "message/rfc822" SinkSignUrlParamName = "__sign" @@ -62,32 +56,40 @@ const ( func Sink() *sink { return &sink{ - logger: DefaultLogger, + 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(surp SinkRequestUrlParams) (signedURL *url.URL, err error) { var ( params []byte + sap = &sinkActionProps{sinkParams: &surp} ) - params, err = json.Marshal(surp) - if err != nil { - return - } + err = func() error { + params, err = json.Marshal(surp) + if err != nil { + return SinkErrFailedToSign(sap).Wrap(err) + } - surp.Method = strings.ToUpper(surp.Method) + surp.Method = strings.ToUpper(surp.Method) - v := url.Values{} + v := url.Values{} - v.Set(SinkSignUrlParamName, svc.signer.Sign(0, params)+SinkSignUrlParamDelimiter+base64.StdEncoding.EncodeToString(params)) + v.Set(SinkSignUrlParamName, svc.signer.Sign(0, params)+SinkSignUrlParamDelimiter+base64.StdEncoding.EncodeToString(params)) - signedURL = &url.URL{RawQuery: v.Encode(), Path: svc.GetPath()} + signedURL = &url.URL{RawQuery: v.Encode(), Path: svc.GetPath()} + return nil + }() - return + return signedURL, svc.recordAction(context.Background(), sap, SinkActionSign, err) } func (svc sink) GetPath() string { @@ -100,44 +102,80 @@ func (svc sink) GetPath() string { return path + "/sink" } -// ProcessRequest handles sink request validation and processing -func (svc sink) ProcessRequest(w http.ResponseWriter, r *http.Request) { - defer r.Body.Close() +// ProcessRequest function is used directly in the HTTP controller +func (svc *sink) ProcessRequest(w http.ResponseWriter, r *http.Request) { + var ( + 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 { + //err.HttpResponse(w) + 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.ContentType, w, r, body); err != nil { + return SinkErrProcessingError(sap).Wrap(err) + } + + return nil + }() + + // record error or successful action with all sink action params + if err, ok := svc.recordAction(r.Context(), sap, SinkActionRequest, err).(*sinkError); ok && err != nil { + // in case of errors, write http response + err.HttpResponse(w) + } +} + +// Verifies and extracts sink request params +func (svc sink) handleRequest(r *http.Request) (*SinkRequestUrlParams, error) { + var ( + srup *SinkRequestUrlParams + sap = &sinkActionProps{} + ) param := r.URL.Query().Get(SinkSignUrlParamName) if len(param) == 0 { - http.Error(w, "missing sink signature parameter", http.StatusBadRequest) - return + return nil, SinkErrMissingSignature(sap) } split := strings.SplitN(param, SinkSignUrlParamDelimiter, 2) if len(split) < 2 { - http.Error(w, "invalid sink signature parameter", http.StatusUnauthorized) - return + return nil, SinkErrInvalidSignatureParam(sap) } params, err := base64.StdEncoding.DecodeString(split[1]) if err != nil { - http.Error(w, "bad encoding of sink parameters", http.StatusBadRequest) - return + return nil, SinkErrBadSinkParamEncoding(sap) } if !svc.signer.Verify(split[0], 0, params) { - http.Error(w, "invalid signature", http.StatusUnauthorized) - return + return nil, SinkErrInvalidSignature(sap) } - srup := &SinkRequestUrlParams{} - if err := json.Unmarshal(params, srup); err != nil { + srup = &SinkRequestUrlParams{} + if err = json.Unmarshal(params, srup); err != nil { // Impossible scenario :) // How can we have verified signature of an invalid JSON ?! - http.Error(w, "invalid sink request url params", http.StatusInternalServerError) - return + return nil, SinkErrInvalidSinkRequestUrlParams(sap) } + sap.setSinkParams(srup) + if srup.Method != "" && srup.Method != r.Method { - http.Error(w, "invalid method", http.StatusUnauthorized) - return + return nil, SinkErrInvalidHttpMethod(sap) } contentType := strings.ToLower(r.Header.Get("content-type")) @@ -147,33 +185,22 @@ func (svc sink) ProcessRequest(w http.ResponseWriter, r *http.Request) { if srup.ContentType != "" { if strings.ToLower(srup.ContentType) != contentType { - http.Error(w, "invalid content-type", http.StatusUnauthorized) - return + return nil, SinkErrInvalidContentType(sap) } } if srup.Expires != nil && srup.Expires.Before(time.Now()) { - http.Error(w, "signature expired", http.StatusGone) - return + return nil, SinkErrSignatureExpired(sap) } - var body io.Reader if srup.MaxBodySize > 0 { // See if there is content length param and reject it right away if r.ContentLength > srup.MaxBodySize { - http.Error(w, "content length exceeds max size limit", http.StatusRequestEntityTooLarge) + return nil, SinkErrContentLengthExceedsMaxAllowedSize(sap) } - - // 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(contentType, w, r, body); err != nil { - http.Error(w, "sink request process error", http.StatusInternalServerError) - return - } + return srup, nil } // Processes sink request, casts it and forwards it to processor (depending on content type) @@ -185,8 +212,14 @@ func (svc sink) ProcessRequest(w http.ResponseWriter, r *http.Request) { // 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(contentType string, w http.ResponseWriter, r *http.Request, body io.Reader) (err error) { - ctx := r.Context() +func (svc *sink) process(contentType string, w http.ResponseWriter, r *http.Request, body io.Reader) error { + var ( + err error + ctx = r.Context() + sap = &sinkActionProps{ + contentType: contentType, + } + ) switch strings.ToLower(contentType) { case SinkContentTypeMail, "rfc822", "email", "mail": @@ -195,10 +228,15 @@ func (svc *sink) process(contentType string, w http.ResponseWriter, r *http.Requ var msg *types.MailMessage msg, err = types.NewMailMessage(body) if err != nil { - return + return SinkErrFailedToCreateEvent(sap).Wrap(err) } - return svc.eventbus.WaitFor(ctx, event.MailOnReceive(msg)) + sap.setMailHeader(&msg.Header) + + err = svc.eventbus.WaitFor(ctx, event.MailOnReceive(msg)) + if err != nil { + return SinkErrFailedToProcess(sap).Wrap(err) + } default: var ( @@ -224,26 +262,26 @@ func (svc *sink) process(contentType string, w http.ResponseWriter, r *http.Requ sr, err = types.NewSinkRequest(r, body) if err != nil { - svc.log(ctx).Error("could create sink request event", zap.Error(err)) - return + return SinkErrFailedToCreateEvent(sap).Wrap(err) + } - ev := event.SinkOnRequest(rsp, sr) - err = svc.eventbus.WaitFor(ctx, ev) + sap.setUrl(sanitizedURL.String()) + + err = svc.eventbus.WaitFor(ctx, event.SinkOnRequest(rsp, sr)) if err != nil { - svc.log(ctx).Error("could not process event", zap.Error(err)) - return + return SinkErrFailedToProcess(sap).Wrap(err) } + sap.setResponseStatus(rsp.Status) + // Now write everything we've received from the script - //if err = rsp.Header.Write(w); err != nil { - // return - //} for k, vv := range rsp.Header { for _, v := range vv { w.Header().Add(k, v) } } + w.WriteHeader(rsp.Status) var output []byte @@ -255,13 +293,9 @@ func (svc *sink) process(contentType string, w http.ResponseWriter, r *http.Requ } if _, err = w.Write(output); err != nil { - return + return SinkErrFailedToRespond(sap).Wrap(err) } } - return -} - -func (svc sink) log(ctx context.Context, fields ...zapcore.Field) *zap.Logger { - return logger.AddRequestID(ctx, svc.logger).With(fields...) + return nil } diff --git a/system/service/sink_actions.gen.go b/system/service/sink_actions.gen.go new file mode 100644 index 000000000..ab0a739da --- /dev/null +++ b/system/service/sink_actions.gen.go @@ -0,0 +1,929 @@ +package service + +// This file is auto-generated from system/service/sink_actions.yaml +// + +import ( + "context" + "errors" + "fmt" + "net/http" + "strings" + "time" + + "github.com/cortezaproject/corteza-server/pkg/actionlog" + "github.com/cortezaproject/corteza-server/system/types" +) + +type ( + sinkActionProps struct { + url string + responseStatus int + contentType string + sinkParams *SinkRequestUrlParams + mailHeader *types.MailMessageHeader + } + + sinkAction struct { + timestamp time.Time + resource string + action string + log string + severity actionlog.Severity + + // prefix for error when action fails + errorMessage string + + props *sinkActionProps + } + + sinkError struct { + timestamp time.Time + error string + resource string + action string + message string + log string + severity actionlog.Severity + + wrap error + + props *sinkActionProps + + httpStatusCode int + } +) + +// ********************************************************************************************************************* +// ********************************************************************************************************************* +// Props methods +// setUrl updates sinkActionProps's url +// +// Allows method chaining +// +// This function is auto-generated. +// +func (p *sinkActionProps) setUrl(url string) *sinkActionProps { + p.url = url + return p +} + +// setResponseStatus updates sinkActionProps's responseStatus +// +// Allows method chaining +// +// This function is auto-generated. +// +func (p *sinkActionProps) setResponseStatus(responseStatus int) *sinkActionProps { + p.responseStatus = responseStatus + return p +} + +// setContentType updates sinkActionProps's contentType +// +// Allows method chaining +// +// This function is auto-generated. +// +func (p *sinkActionProps) setContentType(contentType string) *sinkActionProps { + p.contentType = contentType + return p +} + +// setSinkParams updates sinkActionProps's sinkParams +// +// Allows method chaining +// +// This function is auto-generated. +// +func (p *sinkActionProps) setSinkParams(sinkParams *SinkRequestUrlParams) *sinkActionProps { + p.sinkParams = sinkParams + return p +} + +// setMailHeader updates sinkActionProps's mailHeader +// +// Allows method chaining +// +// This function is auto-generated. +// +func (p *sinkActionProps) setMailHeader(mailHeader *types.MailMessageHeader) *sinkActionProps { + p.mailHeader = mailHeader + return p +} + +// serialize converts sinkActionProps to actionlog.Meta +// +// This function is auto-generated. +// +func (p sinkActionProps) serialize() actionlog.Meta { + var ( + m = make(actionlog.Meta) + str = func(i interface{}) string { return fmt.Sprintf("%v", i) } + ) + + m["url"] = str(p.url) + m["responseStatus"] = str(p.responseStatus) + m["contentType"] = str(p.contentType) + m["sinkParams"] = str(p.sinkParams) + if p.mailHeader != nil { + m["mailHeader.to"] = str(p.mailHeader.To) + m["mailHeader.CC"] = str(p.mailHeader.CC) + m["mailHeader.BCC"] = str(p.mailHeader.BCC) + m["mailHeader.from"] = str(p.mailHeader.From) + m["mailHeader.replyTo"] = str(p.mailHeader.ReplyTo) + m["mailHeader.raw"] = str(p.mailHeader.Raw) + } + + return m +} + +// tr translates string and replaces meta value placeholder with values +// +// This function is auto-generated. +// +func (p sinkActionProps) tr(in string, err error) string { + var pairs = []string{"{err}"} + + if err != nil { + for { + // Unwrap errors + ue := errors.Unwrap(err) + if ue == nil { + break + } + + err = ue + } + + pairs = append(pairs, err.Error()) + } else { + pairs = append(pairs, "nil") + } + pairs = append(pairs, "{url}", fmt.Sprintf("%v", p.url)) + pairs = append(pairs, "{responseStatus}", fmt.Sprintf("%v", p.responseStatus)) + pairs = append(pairs, "{contentType}", fmt.Sprintf("%v", p.contentType)) + pairs = append(pairs, "{sinkParams}", fmt.Sprintf("%v", p.sinkParams)) + + if p.mailHeader != nil { + pairs = append(pairs, "{mailHeader}", fmt.Sprintf("%v", p.mailHeader.To)) + pairs = append(pairs, "{mailHeader.to}", fmt.Sprintf("%v", p.mailHeader.To)) + pairs = append(pairs, "{mailHeader.CC}", fmt.Sprintf("%v", p.mailHeader.CC)) + pairs = append(pairs, "{mailHeader.BCC}", fmt.Sprintf("%v", p.mailHeader.BCC)) + pairs = append(pairs, "{mailHeader.from}", fmt.Sprintf("%v", p.mailHeader.From)) + pairs = append(pairs, "{mailHeader.replyTo}", fmt.Sprintf("%v", p.mailHeader.ReplyTo)) + pairs = append(pairs, "{mailHeader.raw}", fmt.Sprintf("%v", p.mailHeader.Raw)) + } + return strings.NewReplacer(pairs...).Replace(in) +} + +// ********************************************************************************************************************* +// ********************************************************************************************************************* +// Action methods + +// String returns loggable description as string +// +// This function is auto-generated. +// +func (a *sinkAction) String() string { + var props = &sinkActionProps{} + + if a.props != nil { + props = a.props + } + + return props.tr(a.log, nil) +} + +func (e *sinkAction) LoggableAction() *actionlog.Action { + return &actionlog.Action{ + Timestamp: e.timestamp, + Resource: e.resource, + Action: e.action, + Severity: e.severity, + Description: e.String(), + Meta: e.props.serialize(), + } +} + +// ********************************************************************************************************************* +// ********************************************************************************************************************* +// Error methods + +// String returns loggable description as string +// +// It falls back to message if log is not set +// +// This function is auto-generated. +// +func (e *sinkError) String() string { + var props = &sinkActionProps{} + + if e.props != nil { + props = e.props + } + + if e.wrap != nil && !strings.Contains(e.log, "{err}") { + // Suffix error log with {err} to ensure + // we log the cause for this error + e.log += ": {err}" + } + + return props.tr(e.log, e.wrap) +} + +// Error satisfies +// +// This function is auto-generated. +// +func (e *sinkError) Error() string { + var props = &sinkActionProps{} + + if e.props != nil { + props = e.props + } + + return props.tr(e.message, e.wrap) +} + +// Is fn for error equality check +// +// This function is auto-generated. +// +func (e *sinkError) Is(Resource error) bool { + t, ok := Resource.(*sinkError) + if !ok { + return false + } + + return t.resource == e.resource && t.error == e.error +} + +// Wrap wraps sinkError around another error +// +// This function is auto-generated. +// +func (e *sinkError) Wrap(err error) *sinkError { + e.wrap = err + return e +} + +// Unwrap returns wrapped error +// +// This function is auto-generated. +// +func (e *sinkError) Unwrap() error { + return e.wrap +} + +func (e *sinkError) LoggableAction() *actionlog.Action { + return &actionlog.Action{ + Timestamp: e.timestamp, + Resource: e.resource, + Action: e.action, + Severity: e.severity, + Description: e.String(), + Error: e.Error(), + Meta: e.props.serialize(), + } +} + +func (e *sinkError) HttpResponse(w http.ResponseWriter) { + var code = e.httpStatusCode + if code == 0 { + code = http.StatusInternalServerError + } + + http.Error(w, e.message, code) +} + +// ********************************************************************************************************************* +// ********************************************************************************************************************* +// Action constructors + +// SinkActionSign returns "system:sink.sign" error +// +// This function is auto-generated. +// +func SinkActionSign(props ...*sinkActionProps) *sinkAction { + a := &sinkAction{ + timestamp: time.Now(), + resource: "system:sink", + action: "sign", + log: "signed sink request URL", + severity: actionlog.Notice, + } + + if len(props) > 0 { + a.props = props[0] + } + + return a +} + +// SinkActionPreprocess returns "system:sink.preprocess" error +// +// This function is auto-generated. +// +func SinkActionPreprocess(props ...*sinkActionProps) *sinkAction { + a := &sinkAction{ + timestamp: time.Now(), + resource: "system:sink", + action: "preprocess", + log: "preprocess", + severity: actionlog.Info, + } + + if len(props) > 0 { + a.props = props[0] + } + + return a +} + +// SinkActionRequest returns "system:sink.request" error +// +// This function is auto-generated. +// +func SinkActionRequest(props ...*sinkActionProps) *sinkAction { + a := &sinkAction{ + timestamp: time.Now(), + resource: "system:sink", + action: "request", + log: "sink request processed", + severity: actionlog.Info, + } + + if len(props) > 0 { + a.props = props[0] + } + + return a +} + +// ********************************************************************************************************************* +// ********************************************************************************************************************* +// Error constructors + +// SinkErrGeneric returns "system:sink.generic" audit event as actionlog.Error +// +// +// This function is auto-generated. +// +func SinkErrGeneric(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "generic", + action: "error", + message: "failed to complete request due to internal error", + log: "{err}", + severity: actionlog.Error, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrFailedToSign returns "system:sink.failedToSign" audit event as actionlog.Error +// +// +// This function is auto-generated. +// +func SinkErrFailedToSign(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "failedToSign", + action: "error", + message: "could not sign request params: {err}", + log: "could not sign request params: {err}", + severity: actionlog.Error, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrFailedToCreateEvent returns "system:sink.failedToCreateEvent" audit event as actionlog.Error +// +// +// This function is auto-generated. +// +func SinkErrFailedToCreateEvent(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "failedToCreateEvent", + action: "error", + message: "failed to create sink event from request", + log: "failed to create sink event from request", + severity: actionlog.Error, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrFailedToProcess returns "system:sink.failedToProcess" audit event as actionlog.Error +// +// +// This function is auto-generated. +// +func SinkErrFailedToProcess(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "failedToProcess", + action: "error", + message: "failed to process request", + log: "failed to process request", + severity: actionlog.Error, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrFailedToRespond returns "system:sink.failedToRespond" audit event as actionlog.Error +// +// +// This function is auto-generated. +// +func SinkErrFailedToRespond(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "failedToRespond", + action: "error", + message: "failed to respond to request", + log: "failed to respond to request", + severity: actionlog.Error, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrMissingSignature returns "system:sink.missingSignature" audit event as actionlog.Alert +// +// +// This function is auto-generated. +// +func SinkErrMissingSignature(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "missingSignature", + action: "error", + message: "missing sink signature parameter", + log: "missing sink signature parameter", + severity: actionlog.Alert, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + + httpStatusCode: http.StatusBadRequest, + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrInvalidSignatureParam returns "system:sink.invalidSignatureParam" audit event as actionlog.Alert +// +// +// This function is auto-generated. +// +func SinkErrInvalidSignatureParam(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "invalidSignatureParam", + action: "error", + message: "invalid sink signature parameter", + log: "invalid sink signature parameter", + severity: actionlog.Alert, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + + httpStatusCode: http.StatusUnauthorized, + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrBadSinkParamEncoding returns "system:sink.badSinkParamEncoding" audit event as actionlog.Alert +// +// +// This function is auto-generated. +// +func SinkErrBadSinkParamEncoding(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "badSinkParamEncoding", + action: "error", + message: "bad encoding of sink parameters", + log: "bad encoding of sink parameters", + severity: actionlog.Alert, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + + httpStatusCode: http.StatusBadRequest, + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrInvalidSignature returns "system:sink.invalidSignature" audit event as actionlog.Alert +// +// +// This function is auto-generated. +// +func SinkErrInvalidSignature(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "invalidSignature", + action: "error", + message: "invalid signature", + log: "invalid signature", + severity: actionlog.Alert, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + + httpStatusCode: http.StatusUnauthorized, + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrInvalidSinkRequestUrlParams returns "system:sink.invalidSinkRequestUrlParams" audit event as actionlog.Alert +// +// +// This function is auto-generated. +// +func SinkErrInvalidSinkRequestUrlParams(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "invalidSinkRequestUrlParams", + action: "error", + message: "invalid sink request url params", + log: "invalid sink request url params", + severity: actionlog.Alert, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + + httpStatusCode: http.StatusInternalServerError, + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrInvalidHttpMethod returns "system:sink.invalidHttpMethod" audit event as actionlog.Alert +// +// +// This function is auto-generated. +// +func SinkErrInvalidHttpMethod(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "invalidHttpMethod", + action: "error", + message: "invalid HTTP method", + log: "invalid HTTP method", + severity: actionlog.Alert, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + + httpStatusCode: http.StatusUnauthorized, + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrInvalidContentType returns "system:sink.invalidContentType" audit event as actionlog.Alert +// +// +// This function is auto-generated. +// +func SinkErrInvalidContentType(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "invalidContentType", + action: "error", + message: "invalid content-type header", + log: "invalid content-type header", + severity: actionlog.Alert, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + + httpStatusCode: http.StatusUnauthorized, + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrSignatureExpired returns "system:sink.signatureExpired" audit event as actionlog.Alert +// +// +// This function is auto-generated. +// +func SinkErrSignatureExpired(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "signatureExpired", + action: "error", + message: "signature expired", + log: "signature expired", + severity: actionlog.Alert, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + + httpStatusCode: http.StatusGone, + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrContentLengthExceedsMaxAllowedSize returns "system:sink.contentLengthExceedsMaxAllowedSize" audit event as actionlog.Alert +// +// +// This function is auto-generated. +// +func SinkErrContentLengthExceedsMaxAllowedSize(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "contentLengthExceedsMaxAllowedSize", + action: "error", + message: "content length exceeds max size limit", + log: "content length exceeds max size limit", + severity: actionlog.Alert, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + + httpStatusCode: http.StatusRequestEntityTooLarge, + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// SinkErrProcessingError returns "system:sink.processingError" audit event as actionlog.Alert +// +// +// This function is auto-generated. +// +func SinkErrProcessingError(props ...*sinkActionProps) *sinkError { + var e = &sinkError{ + timestamp: time.Now(), + resource: "system:sink", + error: "processingError", + action: "error", + message: "sink request process error", + log: "sink request process error", + severity: actionlog.Alert, + props: func() *sinkActionProps { + if len(props) > 0 { + return props[0] + } + return nil + }(), + + httpStatusCode: http.StatusInternalServerError, + } + + if len(props) > 0 { + e.props = props[0] + } + + return e + +} + +// ********************************************************************************************************************* +// ********************************************************************************************************************* + +// recordAction is a service helper function wraps function that can return error +// +// context is used to enrich audit log entry with current user info, request ID, IP address... +// props are collected action/error properties +// action (optional) fn will be used to construct sinkAction struct from given props (and error) +// err is any error that occurred while action was happening +// +// Action has success and fail (error) state: +// - when recorded without an error (4th param), action is recorded as successful. +// - when an additional error is given (4th param), action is used to wrap +// the additional error +// +// This function is auto-generated. +// +func (svc sink) recordAction(ctx context.Context, props *sinkActionProps, action func(...*sinkActionProps) *sinkAction, err error) error { + var ( + ok bool + + // Return error + retError *sinkError + + // Recorder error + recError *sinkError + ) + + if err != nil { + if retError, ok = err.(*sinkError); !ok { + // got non-sink error, wrap it with SinkErrGeneric + retError = SinkErrGeneric(props).Wrap(err) + + // copy action to returning and recording error + retError.action = action().action + + // we'll use SinkErrGeneric for recording too + // because it can hold more info + recError = retError + } else if retError != nil { + // copy action to returning and recording error + retError.action = action().action + // start with copy of return error for recording + // this will be updated with tha root cause as we try and + // unwrap the error + recError = retError + + // find the original recError for this error + // for the purpose of logging + var unwrappedError error = retError + for { + if unwrappedError = errors.Unwrap(unwrappedError); unwrappedError == nil { + // nothing wrapped + break + } + + // update recError ONLY of wrapped error is of type sinkError + if unwrappedSinkError, ok := unwrappedError.(*sinkError); ok { + recError = unwrappedSinkError + } + } + + if retError.props == nil { + // set props on returning error if empty + retError.props = props + } + + if recError.props == nil { + // set props on recording error if empty + recError.props = props + } + } + } + + if svc.actionlog != nil { + if retError != nil { + // failed action, log error + svc.actionlog.Record(ctx, recError) + } else if action != nil { + // successful + svc.actionlog.Record(ctx, action(props)) + } + } + + if err == nil { + // retError not an interface and that WILL (!!) cause issues + // with nil check (== nil) when it is not explicitly returned + return nil + } + + return retError +} diff --git a/system/service/sink_actions.yaml b/system/service/sink_actions.yaml new file mode 100644 index 000000000..f44eba482 --- /dev/null +++ b/system/service/sink_actions.yaml @@ -0,0 +1,93 @@ +# List of loggable service actions + +resource: system:sink +service: sink + +# Default sensitivity for actions +defaultActionSeverity: info + +# default severity for errors +defaultErrorSeverity: alert + +import: + - github.com/cortezaproject/corteza-server/system/types + +props: + - name: url + - name: responseStatus + type: int + - name: contentType + - name: sinkParams + type: "*SinkRequestUrlParams" + - name: mailHeader + type: "*types.MailMessageHeader" + fields: [ to, CC, BCC, from, replyTo, raw ] + +actions: + - action: sign + log: "signed sink request URL" + severity: notice + + - action: preprocess + + - action: request + log: "sink request processed" + + +errors: + - error: failedToSign + message: "could not sign request params: {err}" + severity: error + + - error: failedToCreateEvent + message: "failed to create sink event from request" + severity: error + + - error: failedToProcess + message: "failed to process request" + severity: error + + - error: failedToRespond + message: "failed to respond to request" + severity: error + + - error: missingSignature + message: "missing sink signature parameter" + httpStatus: StatusBadRequest + + - error: invalidSignatureParam + message: "invalid sink signature parameter" + httpStatus: StatusUnauthorized + + - error: badSinkParamEncoding + message: "bad encoding of sink parameters" + httpStatus: StatusBadRequest + + - error: invalidSignature + message: "invalid signature" + httpStatus: StatusUnauthorized + + - error: invalidSinkRequestUrlParams + message: "invalid sink request url params" + httpStatus: StatusInternalServerError + + - error: invalidHttpMethod + message: "invalid HTTP method" + httpStatus: StatusUnauthorized + + - error: invalidContentType + message: "invalid content-type header" + httpStatus: StatusUnauthorized + + - error: signatureExpired + message: "signature expired" + httpStatus: StatusGone + + - error: contentLengthExceedsMaxAllowedSize + message: "content length exceeds max size limit" + httpStatus: StatusRequestEntityTooLarge + + - error: processingError + message: "sink request process error" + httpStatus: StatusInternalServerError + diff --git a/system/service/sink_test.go b/system/service/sink_test.go new file mode 100644 index 000000000..5261c9321 --- /dev/null +++ b/system/service/sink_test.go @@ -0,0 +1,177 @@ +package service + +import ( + "bytes" + "errors" + "io" + "net/http" + "reflect" + "strings" + "testing" + "time" + + internalAuth "github.com/cortezaproject/corteza-server/pkg/auth" +) + +func Test_sink_SignURL(t *testing.T) { + var ( + signer = internalAuth.HmacSigner("test") + + tests = []struct { + name string + surp SinkRequestUrlParams + wantSignedURL string + wantErr bool + }{ + { + name: "basic", + surp: SinkRequestUrlParams{ + Method: "POST", + Origin: "test", + Expires: nil, + MaxBodySize: 1024, + ContentType: "plain/text", + }, + wantSignedURL: "/sink?__sign=d8a8c5591acb0f5f6695ab6aa4a205a7066b3bf4_eyJtdGQiOiJQT1NUIiwib3JpZ2luIjoidGVzdCIsIm1icyI6MTAyNCwiY3QiOiJwbGFpbi90ZXh0In0%3D", + }, + } + ) + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + svc := sink{ + signer: signer, + } + + gotSignedURL, err := svc.SignURL(tt.surp) + if (err != nil) != tt.wantErr { + t.Errorf("SignURL() error = %v, wantErr %v", err, tt.wantErr) + return + } + if gotSignedURL.String() != tt.wantSignedURL { + t.Errorf("SignURL() gotSignedURL = %v, want %v", gotSignedURL, tt.wantSignedURL) + } + }) + } +} + +func Test_sink_handleRequest(t *testing.T) { + var ( + signer = internalAuth.HmacSigner("test") + svc = sink{signer: signer} + + signParams = SinkRequestUrlParams{ + Method: "POST", + Origin: "test", + Expires: nil, + MaxBodySize: 1024, + ContentType: "plain/text", + } + signedUrl, _ = svc.SignURL(signParams) + + signParamsNoPref = SinkRequestUrlParams{} + signedUrlNoPref, _ = svc.SignURL(signParamsNoPref) + + signParamsExp = SinkRequestUrlParams{Expires: &time.Time{}} + signedUrlExp, _ = svc.SignURL(signParamsExp) + ) + + var ( + tests = []struct { + name string + withMethod string + withURL string + withBody io.Reader + withHeader http.Header + wantParams *SinkRequestUrlParams + wantErr error + }{ + { + name: "missing signature", + withURL: "/sink", + wantErr: SinkErrMissingSignature(), + }, + { + name: "invalid signature", + withURL: "/sink?" + SinkSignUrlParamName + "=foo", + wantErr: SinkErrInvalidSignatureParam(), + }, + { + name: "invalid signature", + withURL: "/sink?__sign=foo_bar", + wantErr: SinkErrBadSinkParamEncoding(), + }, + { + name: "invalid signature", + withURL: "/sink?__sign=foo_eyJtdGQiOiJQT1NUIiwib3JpZ2luIjoidGVzdCIsIm1icyI6MTAyNCwiY3QiOiJwbGFpbi90ZXh0In0%3D", + wantErr: SinkErrInvalidSignature(), + }, + { + name: "any HTTP method (no pref. method)", + withMethod: "DELETE", + withURL: signedUrlNoPref.String(), + wantParams: &signParamsNoPref, + wantErr: nil, + }, + { + name: "invalid HTTP method (POST only)", + withMethod: "GET", + withURL: signedUrl.String(), + wantErr: SinkErrInvalidHttpMethod(), + }, + { + name: "invalid content type", + withMethod: "POST", + withHeader: map[string][]string{"content-type": {"foo/bar"}}, + withURL: signedUrl.String(), + wantErr: SinkErrInvalidContentType(), + }, + { + name: "valid content type", + withMethod: "POST", + withHeader: map[string][]string{"content-type": {"plain/text"}}, + withURL: signedUrl.String(), + wantErr: nil, + wantParams: &signParams, + }, + { + name: "expired", + withMethod: "POST", + withURL: signedUrlExp.String(), + wantErr: SinkErrSignatureExpired(), + }, + { + name: "content length exceeds", + withMethod: "POST", + withHeader: map[string][]string{"content-type": {"plain/text"}}, + withBody: bytes.NewBufferString(strings.Repeat(".", 1025)), + withURL: signedUrl.String(), + wantErr: SinkErrContentLengthExceedsMaxAllowedSize(), + }, + } + ) + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req, err := http.NewRequest(tt.withMethod, tt.withURL, tt.withBody) + if err != nil { + panic(err) + } + + for n, vv := range tt.withHeader { + for _, v := range vv { + req.Header.Add(n, v) + } + } + + got, err := svc.handleRequest(req) + if !errors.Is(err, tt.wantErr) { + t.Errorf("handleRequest() error = %v, wantErr %v", err, tt.wantErr) + return + } + if !reflect.DeepEqual(got, tt.wantParams) { + t.Errorf("handleRequest() params = %v, want %v", got, tt.wantParams) + } + }) + } +}