diff --git a/pkg/apigw/apigw.go b/pkg/apigw/apigw.go new file mode 100644 index 000000000..cbdb02191 --- /dev/null +++ b/pkg/apigw/apigw.go @@ -0,0 +1,175 @@ +package apigw + +import ( + "context" + + "github.com/cortezaproject/corteza-server/system/types" + "github.com/go-chi/chi" + "go.uber.org/zap" +) + +type ( + storer interface { + SearchApigwRoutes(ctx context.Context, f types.RouteFilter) (types.RouteSet, types.RouteFilter, error) + SearchApigwFunctions(ctx context.Context, f types.FunctionFilter) (types.FunctionSet, types.FunctionFilter, error) + } + + apigw struct { + log *zap.Logger + reg *registry + routes []*route + dispatcher dispatcher + storer storer + reload chan bool + } +) + +var ( + // global service + apiGw *apigw +) + +func Service() *apigw { + return apiGw +} + +func Set(a *apigw) { + apiGw = a +} + +// Setup handles the singleton service +func Setup(opts interface{}, log *zap.Logger, dispatcher dispatcher, storer storer) { + if apiGw != nil { + return + } + + apiGw = New(opts, log, dispatcher, storer) +} + +func New(opts interface{}, logger *zap.Logger, dispatcher dispatcher, storer storer) *apigw { + reg := NewRegistry() + reg.Preload() + + return &apigw{ + log: logger, + dispatcher: dispatcher, + storer: storer, + reload: make(chan bool), + reg: reg, + } +} + +func (s *apigw) Reload(ctx context.Context) { + go func() { + s.reload <- true + }() +} + +func (s *apigw) loadRoutes(ctx context.Context) (rr []*route, err error) { + routes, _, err := s.storer.SearchApigwRoutes(ctx, types.RouteFilter{Enabled: true}) + + if err != nil { + return + } + + for _, r := range routes { + route := &route{ + ID: r.ID, + endpoint: r.Endpoint, + method: r.Method, + } + + rr = append(rr, route) + } + + return +} + +func (s *apigw) loadFunctions(ctx context.Context, route uint64) (ff []*types.Function, err error) { + ff, _, err = s.storer.SearchApigwFunctions(ctx, types.FunctionFilter{}) + return +} + +func (s *apigw) Router(ctx context.Context) func(r chi.Router) { + return func(r chi.Router) { + + routes, err := s.loadRoutes(ctx) + + if err != nil { + s.log.Error("could not load routes", zap.Error(err)) + return + } + + s.Init(ctx, routes...) + + for _, route := range s.routes { + r.Handle(route.endpoint, route) + } + + go func() { + for { + select { + case <-s.reload: + s.log.Debug("got reload signal") + + routes, err := s.loadRoutes(ctx) + + if err != nil { + s.log.Error("could not reload routes", zap.Error(err)) + return + } + + s.Init(ctx, routes...) + + for _, route := range s.routes { + r.Handle(route.endpoint, route) + } + + case <-ctx.Done(): + s.log.Debug("done! getting out") + return + } + } + }() + } +} + +// init all the routes +func (s *apigw) Init(ctx context.Context, route ...*route) { + s.routes = route + + s.log.Debug("initializing routes\n", zap.Int("num", len(s.routes))) + + for _, r := range s.routes { + r.pipe = &pl{} + regFuncs, err := s.loadFunctions(ctx, r.ID) + + if err != nil { + s.log.Error("could not load functions for route", zap.String("route", r.endpoint), zap.Error(err)) + continue + } + + r.pipe.ErrorHandler(errorHandler{ + name: "error handler expediter", + args: []string{}, + weight: 5, + step: 0, + }) + + for _, f := range regFuncs { + fc := functionHandler{} + + h, err := s.reg.Get(f.Ref) + + if err != nil { + s.log.Error("could not register function for route", zap.String("route", r.endpoint), zap.Error(err)) + continue + } + + fc.Merge(ctx, h.Meta(f)) + fc.SetHandler(h.Handler()) + + r.pipe.Add(fc, f.Params) + } + } +} diff --git a/pkg/apigw/apigw_test.go b/pkg/apigw/apigw_test.go new file mode 100644 index 000000000..debc459bf --- /dev/null +++ b/pkg/apigw/apigw_test.go @@ -0,0 +1,40 @@ +package apigw + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/cortezaproject/corteza-server/pkg/expr" + "github.com/cortezaproject/corteza-server/pkg/logger" + "github.com/cortezaproject/corteza-server/pkg/wfexec" + "github.com/stretchr/testify/require" +) + +func execFn(t *testing.T, r *http.Request, fn wfHandler) error { + var ( + req = require.New(t) + ctx = context.Background() + scope = &expr.Vars{} + graph = wfexec.NewGraph() + recorder = httptest.NewRecorder() + ) + + scope.Set("envelope", envelope{ + Request: r, + Writer: recorder, + }) + + step := wfexec.NewGenericStep(fn.self()) + + graph.AddStep(step) + + sess := wfexec.NewSession(ctx, graph, wfexec.SetLogger(logger.Default())) + + err := sess.Exec(ctx, step, scope) + + req.NoError(err) + + return sess.Wait(ctx) +} diff --git a/pkg/apigw/expediter.go b/pkg/apigw/expediter.go new file mode 100644 index 000000000..2a09081a3 --- /dev/null +++ b/pkg/apigw/expediter.go @@ -0,0 +1,52 @@ +package apigw + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + + "github.com/cortezaproject/corteza-server/pkg/expr" + "github.com/cortezaproject/corteza-server/pkg/wfexec" + "github.com/davecgh/go-spew/spew" +) + +type ( + redirectExpediterArgs struct { + Location string + } +) + +func redirectExpediter(c context.Context, params *expr.Vars) wfHandler { + var ( + clv = redirectExpediterArgs{} + ) + + return func(c context.Context, er *wfexec.ExecRequest) (r wfexec.ExecResponse, err error) { + params.Decode(&clv) + spew.Dump("redirect expediter fn()", clv) + e := er.Scope.GetValue()["envelope"] + ee := e.Get().(envelope) + + http.Redirect(ee.Writer, ee.Request, clv.Location, http.StatusTemporaryRedirect) + + r = &expr.Vars{} + return + } +} + +func expediterErrorFn(c context.Context, er *wfexec.ExecRequest) (r wfexec.ExecResponse, err error) { + // spew.Dump("expediter error fn()", er) + + e := er.Scope.GetValue()["error"] + values := er.Scope.GetValue()["writer"] + + writer := values.Get() + eValue := e.Get() + + fmt.Fprintf(writer.(*httptest.ResponseRecorder), fmt.Sprintf(`{"msg": "%s"}`, eValue)) + writer.(*httptest.ResponseRecorder).Code = http.StatusBadGateway + + r = &expr.Vars{} + return +} diff --git a/pkg/apigw/function.go b/pkg/apigw/function.go new file mode 100644 index 000000000..e064a8fa2 --- /dev/null +++ b/pkg/apigw/function.go @@ -0,0 +1,58 @@ +package apigw + +import ( + "context" + + "github.com/cortezaproject/corteza-server/system/types" +) + +type ( + Handler interface { + Handler() handlerFunc + Meta(f *types.Function) functionMeta + } + + handlerFunc func(context.Context, *scp, map[string]interface{}, functionHandler) error + + functionMeta struct { + step int + weight int + name string + label string + kind string + params map[string]interface{} + } + + functionHandler struct { + step int + weight int + name string + label string + kind string + handler handlerFunc + params map[string]interface{} + } +) + +func (ff functionHandler) Exec(ctx context.Context, scope *scp, params map[string]interface{}) error { + return ff.handler(ctx, scope, params, ff) +} + +func (ff *functionHandler) SetHandler(h handlerFunc) { + ff.handler = h +} + +func (ff *functionHandler) Merge(ctx context.Context, p functionMeta) { + ff.step = p.step + ff.kind = p.kind + ff.label = p.label + ff.name = p.name + ff.weight = p.weight + ff.params = p.params +} + +func (ff functionHandler) Weight() int { + // if there's gonna be more than 1000 funcs + // per step, we're doing something wrong + return ff.step*1000 + ff.weight +} diff --git a/pkg/apigw/matcher.go b/pkg/apigw/matcher.go new file mode 100644 index 000000000..fe0412020 --- /dev/null +++ b/pkg/apigw/matcher.go @@ -0,0 +1,42 @@ +package apigw + +import ( + "context" + "errors" + + "github.com/cortezaproject/corteza-server/pkg/expr" + "github.com/cortezaproject/corteza-server/pkg/wfexec" + "github.com/davecgh/go-spew/spew" +) + +type ( + authenticationOriginMatcherArgs struct { + Origin string + } +) + +func authenticationOriginMatcher(c context.Context, params *expr.Vars) wfHandler { + var ( + aomp = authenticationOriginMatcherArgs{} + ) + + return func(c context.Context, er *wfexec.ExecRequest) (r wfexec.ExecResponse, err error) { + + params.Decode(&aomp) + spew.Dump("authentication origin matcher fn()", aomp) + e := er.Scope.GetValue()["envelope"] + ee := e.Get().(envelope) + + origin := ee.Request.Header.Get("Origin") + + spew.Dump("input, real", aomp.Origin, origin) + + if aomp.Origin != origin { + err = errors.New("origin fail") + return + } + + r = &expr.Vars{} + return + } +} diff --git a/pkg/apigw/matcher_test.go b/pkg/apigw/matcher_test.go new file mode 100644 index 000000000..5a29e46e9 --- /dev/null +++ b/pkg/apigw/matcher_test.go @@ -0,0 +1,74 @@ +package apigw + +import ( + "context" + "net/http" + "testing" + + "github.com/cortezaproject/corteza-server/pkg/expr" + "github.com/stretchr/testify/require" +) + +func TestAuthenticationOriginMatcher(t *testing.T) { + type ( + tf struct { + name string + origin string + exp string + req *http.Request + } + ) + + var ( + ctx = context.Background() + + tcc = []tf{ + { + name: "fail on origin", + origin: "http://fail.ed", + exp: "workflow 0 step 0 execution failed: origin fail", + + req: &http.Request{ + Header: http.Header{ + "Origin": []string{ + "http://localhost", + }, + }, + }, + }, + { + name: "success on origin", + origin: "http://localhost", + exp: "", + + req: &http.Request{ + Header: http.Header{ + "Origin": []string{ + "http://localhost", + }, + }, + }, + }, + } + ) + + for _, tc := range tcc { + t.Run(tc.name, func(t *testing.T) { + var ( + req = require.New(t) + input = &expr.Vars{} + ) + + input.Set("origin", tc.origin) + + err := execFn(t, tc.req, authenticationOriginMatcher(ctx, input)) + + if tc.exp != "" { + req.EqualError(err, tc.exp) + } else { + req.NoError(err) + } + }) + } + +} diff --git a/pkg/apigw/processer.go b/pkg/apigw/processer.go new file mode 100644 index 000000000..ca92bf6bf --- /dev/null +++ b/pkg/apigw/processer.go @@ -0,0 +1,55 @@ +package apigw + +import ( + "context" + + "github.com/cortezaproject/corteza-server/pkg/expr" + "github.com/cortezaproject/corteza-server/pkg/wfexec" + "github.com/davecgh/go-spew/spew" +) + +func formDataProcesserFn(c context.Context, er *wfexec.ExecRequest) (r wfexec.ExecResponse, err error) { + type ( + formDataProcesserResponse struct { + Name string `json:"name"` + } + ) + + spew.Dump("step processer fn()") + + e := er.Scope.GetValue()["envelope"] + ee := e.Get() + + // ee.(envelope).Writer.WriteHeader(int(id)) + ee.(envelope).Writer.Write([]byte(`{"test":"foobar"}`)) + + e.Assign(ee) + + // req := values.Get() + // ww := wr.Get() + // writer := ww.(http.ResponseWriter) + + // formValue := req.(*http.Request).PostFormValue("name") + + // resp := formDataProcesserResponse{ + // // Name: fmt.Sprintf("AA %s AA", formValue), + // Name: "formValue", + // } + + // encoder := json.NewEncoder(writer) + // encoder.Encode(resp) + + // writer.(*httptest.ResponseRecorder).Header()["Content-Type"] = []string{"application/json"} + // writer.Header().Set("Content-Type", "application/json3") + + // spew.Dump(writer.(*httptest.ResponseRecorder).Header()) + // a, b := expr.NewKV(writer) + // spew.Dump("Aaaaaaaaaaa", a) + + vv := &expr.Vars{} + // vv.Set("writer", writer) + + r = vv + + return +} diff --git a/pkg/apigw/route.go b/pkg/apigw/route.go new file mode 100644 index 000000000..ddfa6e5c3 --- /dev/null +++ b/pkg/apigw/route.go @@ -0,0 +1,64 @@ +package apigw + +import ( + "context" + "net/http" + + "github.com/cortezaproject/corteza-server/pkg/expr" + "github.com/cortezaproject/corteza-server/pkg/logger" + "github.com/cortezaproject/corteza-server/pkg/wfexec" + "github.com/davecgh/go-spew/spew" +) + +type ( + route struct { + endpoint string + method string + graph *wfexec.Graph + steps []wfexec.Step + fns wfHandlerList + } +) + +func (r route) validate(req *http.Request) (err error) { + // if req.Method != r.method { + // err = errors.New("http method invalid") + // } + + return +} + +func (r route) ServeHTTP(w http.ResponseWriter, req *http.Request) { + if err := r.validate(req); err != nil { + spew.Dump("ERR", err) + return + } + + sess := wfexec.NewSession(context.Background(), r.graph, wfexec.SetLogger(logger.Default()), wfexec.SetHandler(func(ss wfexec.SessionStatus, s1 *wfexec.State, s2 *wfexec.Session) { + // spew.Dump("event handler here!", ss) + })) + + scope := &expr.Vars{} + + scope.Set("envelope", envelope{ + Request: req, + Writer: w, + }) + + if len(r.steps) == 0 { + // dont serve, do what? return default response? + return + } + + err := sess.Exec(context.Background(), r.steps[0], scope) + + // if err != nil { + // fmt.Fprintf(w, "no go, err on exec: %s", err) + // return + // } + + err = sess.Wait(context.Background()) + + if err != nil { + } +} diff --git a/pkg/apigw/validator.go b/pkg/apigw/validator.go new file mode 100644 index 000000000..d83b2de8e --- /dev/null +++ b/pkg/apigw/validator.go @@ -0,0 +1,40 @@ +package apigw + +import ( + "context" + "errors" + + "github.com/cortezaproject/corteza-server/pkg/expr" + "github.com/cortezaproject/corteza-server/pkg/wfexec" + "github.com/davecgh/go-spew/spew" +) + +type ( + contentLengthValidatorArgs struct { + Length int + } +) + +func contentLengthValidator(c context.Context, params *expr.Vars) wfHandler { + var ( + clv = contentLengthValidatorArgs{} + ) + + return func(c context.Context, er *wfexec.ExecRequest) (r wfexec.ExecResponse, err error) { + + params.Decode(&clv) + spew.Dump("body size validator fn()", clv) + e := er.Scope.GetValue()["envelope"] + ee := e.Get().(envelope) + + cl := ee.Request.ContentLength + + if clv.Length < int(cl) { + err = errors.New("content length overriden") + return + } + + r = &expr.Vars{} + return + } +} diff --git a/pkg/apigw/validator_test.go b/pkg/apigw/validator_test.go new file mode 100644 index 000000000..d74b4bb47 --- /dev/null +++ b/pkg/apigw/validator_test.go @@ -0,0 +1,63 @@ +package apigw + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/cortezaproject/corteza-server/pkg/expr" + "github.com/stretchr/testify/require" +) + +func TestContentLengthValidator(t *testing.T) { + type ( + tf struct { + name string + limit int + exp string + body string + } + ) + + var ( + ctx = context.Background() + + tcc = []tf{ + { + name: "fail on content length > limit", + limit: 10, + exp: "workflow 0 step 0 execution failed: content length overriden", + body: "A message that is 31 bytes long", + }, + { + name: "success on content length < limit", + limit: 10, + exp: "", + body: "Below 10", + }, + } + ) + + for _, tc := range tcc { + t.Run(tc.name, func(t *testing.T) { + var ( + req = require.New(t) + input = &expr.Vars{} + ) + + input.Set("length", tc.limit) + + r := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(tc.body)) + + err := execFn(t, r, contentLengthValidator(ctx, input)) + + if tc.exp != "" { + req.EqualError(err, tc.exp) + } else { + req.NoError(err) + } + }) + } +} diff --git a/system/service/route.go b/system/service/route.go new file mode 100644 index 000000000..c5980161c --- /dev/null +++ b/system/service/route.go @@ -0,0 +1,230 @@ +package service + +import ( + "context" + + "github.com/cortezaproject/corteza-server/pkg/actionlog" + "github.com/cortezaproject/corteza-server/pkg/apigw" + a "github.com/cortezaproject/corteza-server/pkg/auth" + "github.com/cortezaproject/corteza-server/store" + "github.com/cortezaproject/corteza-server/system/types" +) + +type ( + route struct { + actionlog actionlog.Recorder + store store.Storer + ac routeAccessController + } + + routeAccessController interface { + } +) + +func Route() *route { + return (&route{ + ac: DefaultAccessControl, + actionlog: DefaultActionlog, + store: DefaultStore, + }) +} + +func (svc *route) FindByID(ctx context.Context, ID uint64) (q *types.Route, err error) { + var ( + rProps = &routeActionProps{} + ) + + err = func() error { + if ID == 0 { + return RouteErrInvalidID() + } + + if q, err = store.LookupApigwRouteByID(ctx, svc.store, ID); err != nil { + return RouteErrInvalidID().Wrap(err) + } + + rProps.setRoute(q) + + // if !svc.ac.CanReadMessagebusQueue(ctx, q) { + // return QueueErrNotAllowedToRead(qProps) + // } + + return nil + }() + + return q, svc.recordAction(ctx, rProps, RouteActionLookup, err) +} + +func (svc *route) Create(ctx context.Context, new *types.Route) (q *types.Route, err error) { + var ( + qProps = &routeActionProps{new: new} + ) + + err = func() (err error) { + // if !svc.ac.CanCreateMessagebusQueue(ctx) { + // return QueueErrNotAllowedToCreate(qProps) + // } + + // Set new values after beforeCreate events are emitted + new.ID = nextID() + new.CreatedAt = *now() + new.CreatedBy = a.GetIdentityFromContext(ctx).Identity() + + // todo + new.Group = 0 + + if err = store.CreateApigwRoute(ctx, svc.store, new); err != nil { + return err + } + + q = new + + // send the signal to reload all routes + if new.Enabled { + apigw.Service().Reload(ctx) + } + + return nil + }() + + return q, svc.recordAction(ctx, qProps, RouteActionCreate, err) +} + +func (svc *route) Update(ctx context.Context, upd *types.Route) (q *types.Route, err error) { + var ( + qProps = &routeActionProps{update: upd} + qq *types.Route + e error + ) + + err = func() (err error) { + // if !svc.ac.CanUpdateMessagebusQueue(ctx, upd) { + // return QueueErrNotAllowedToUpdate(qProps) + // } + + if qq, e = store.LookupApigwRouteByID(ctx, svc.store, upd.ID); e != nil { + return RouteErrNotFound(qProps) + } + + if qq, e = store.LookupApigwRouteByEndpoint(ctx, svc.store, upd.Endpoint); e == nil && qq != nil { + return RouteErrExistsEndpoint(qProps) + } + + // Set new values after beforeCreate events are emitted + upd.UpdatedAt = now() + upd.CreatedAt = qq.CreatedAt + upd.UpdatedBy = a.GetIdentityFromContext(ctx).Identity() + + if err = store.UpdateApigwRoute(ctx, svc.store, upd); err != nil { + return + } + + q = upd + + // send the signal to reload all route + apigw.Service().Reload(ctx) + + return nil + }() + + return q, svc.recordAction(ctx, qProps, RouteActionUpdate, err) +} + +func (svc *route) DeleteByID(ctx context.Context, ID uint64) (err error) { + var ( + qProps = &routeActionProps{} + q *types.Route + ) + + err = func() (err error) { + if ID == 0 { + return RouteErrInvalidID() + } + + if q, err = store.LookupApigwRouteByID(ctx, svc.store, ID); err != nil { + return + } + + qProps.setRoute(q) + + // if !svc.ac.CanDeleteMessagebusQueue(ctx, q) { + // return QueueErrNotAllowedToDelete(qProps) + // } + + q.DeletedAt = now() + q.DeletedBy = a.GetIdentityFromContext(ctx).Identity() + + if err = store.UpdateApigwRoute(ctx, svc.store, q); err != nil { + return + } + + // send the signal to reload all queues + apigw.Service().Reload(ctx) + + return nil + }() + + return svc.recordAction(ctx, qProps, RouteActionDelete, err) +} + +func (svc *route) UndeleteByID(ctx context.Context, ID uint64) (err error) { + var ( + qProps = &routeActionProps{} + q *types.Route + ) + + err = func() (err error) { + if ID == 0 { + return RouteErrInvalidID() + } + + if q, err = store.LookupApigwRouteByID(ctx, svc.store, ID); err != nil { + return + } + + qProps.setRoute(q) + + // if !svc.ac.CanDeleteMessagebusQueue(ctx, q) { + // return QueueErrNotAllowedToDelete(qProps) + // } + + q.DeletedAt = nil + q.UpdatedBy = a.GetIdentityFromContext(ctx).Identity() + + if err = store.UpdateApigwRoute(ctx, svc.store, q); err != nil { + return + } + + // send the signal to reload all queues + apigw.Service().Reload(ctx) + + return nil + }() + + return svc.recordAction(ctx, qProps, RouteActionDelete, err) +} + +func (svc *route) Search(ctx context.Context, filter types.RouteFilter) (r types.RouteSet, f types.RouteFilter, err error) { + var ( + aProps = &routeActionProps{search: &filter} + ) + + // For each fetched item, store backend will check if it is valid or not + // filter.Check = func(res *messagebus.QueueSettings) (bool, error) { + // if !svc.ac.CanReadMessagebusQueue(ctx, res) { + // return false, nil + // } + + // return true, nil + // } + + err = func() error { + if r, f, err = store.SearchApigwRoutes(ctx, svc.store, filter); err != nil { + return err + } + + return nil + }() + + return r, f, svc.recordAction(ctx, aProps, RouteActionSearch, err) +} diff --git a/system/types/route.go b/system/types/route.go new file mode 100644 index 000000000..fc6e61251 --- /dev/null +++ b/system/types/route.go @@ -0,0 +1,44 @@ +package types + +import ( + "time" + + "github.com/cortezaproject/corteza-server/pkg/filter" +) + +type ( + RouteMeta struct{} + + Route struct { + ID uint64 `json:"routeID,string"` + Endpoint string `json:"endpoint"` + Method string `json:"method"` + Debug bool `json:"debug"` + Enabled bool `json:"enabled"` + Group uint64 `json:"group"` + + CreatedAt time.Time `json:"createdAt,omitempty"` + CreatedBy uint64 `json:"createdBy,string" ` + UpdatedAt *time.Time `json:"updatedAt,omitempty"` + UpdatedBy uint64 `json:"updatedBy,string,omitempty" ` + DeletedAt *time.Time `json:"deletedAt,omitempty"` + DeletedBy uint64 `json:"deletedBy,string,omitempty" ` + } + + RouteFilter struct { + Route string `json:"route"` + Group string `json:"group"` + Enabled bool `json:"enabled"` + + Deleted filter.State `json:"deleted"` + + // Check fn is called by store backend for each resource found function can + // modify the resource and return false if store should not return it + // + // Store then loads additional resources to satisfy the paging parameters + Check func(*Route) (bool, error) `json:"-"` + + filter.Sorting + filter.Paging + } +)