DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Support for HTTP errors, add sink impl.

Browse Source
This commit is contained in:
Denis Arh
2020-05-26 08:06:44 +02:00
parent 5f8fb8a294
commit e8b81396ef
6 changed files with 1336 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
codegen/v2/actionlog/actions.gen.go.tpl
View File

@@ -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 {

11
codegen/v2/actionlog/main.go
View File

@@ -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) {

172
system/service/sink.go
View File

@@ -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
}

929
system/service/sink_actions.gen.go Normal file
View File

@@ -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
}

93
system/service/sink_actions.yaml Normal file
View File

@@ -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

177
system/service/sink_test.go Normal file
View File

@@ -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)
}
})
}
}