From c0d1594eb3d9840e498b808b6c3868d085e75443 Mon Sep 17 00:00:00 2001 From: Denis Arh Date: Mon, 30 Dec 2019 17:20:39 +0100 Subject: [PATCH] Refactors corredor initialization, run-as logic & trigger execution --- compose/app.go | 8 +- corteza/corteza.go | 12 ++- messaging/app.go | 6 +- monolith/app.go | 22 ++++- pkg/corredor/authtoken.go | 25 ----- pkg/corredor/conn_test.go | 3 +- pkg/corredor/service.go | 187 +++++++++++++++++++++++++++----------- pkg/corredor/util.go | 34 +------ system/app.go | 6 +- system/service/user.go | 41 +++++++-- 10 files changed, 208 insertions(+), 136 deletions(-) delete mode 100644 pkg/corredor/authtoken.go diff --git a/compose/app.go b/compose/app.go index 2a51bfeb8..9a72009e5 100644 --- a/compose/app.go +++ b/compose/app.go @@ -39,6 +39,9 @@ func (app *App) Setup(log *zap.Logger, opts *app.Options) (err error) { event.ComposeOnTimestamp(), ) + // @todo Wire in cross-service JWT maker for Corredor + corredor.Service().SetUserFinder(nil) + return } @@ -63,7 +66,7 @@ func (app *App) Initialize(ctx context.Context) (err error) { return } - // Initialize external authentication (from default settings) + // Initialize external authenpkg/corredor/conn_test.go:62:12:tication (from default settings) external.Init() return } @@ -75,9 +78,6 @@ func (app *App) Activate(ctx context.Context) (err error) { service.Watchers(ctx) - // Wire in cross service JWT maker for Corredor - corredor.Service().SetJwtMaker(corredor.CrossServiceAuthTokenMaker()) - return } diff --git a/corteza/corteza.go b/corteza/corteza.go index 3b0bfce9d..77b54e53b 100644 --- a/corteza/corteza.go +++ b/corteza/corteza.go @@ -55,11 +55,14 @@ func (app *App) Setup(log *zap.Logger, opts *app.Options) (err error) { scheduler.Setup(log, eventbus.Service(), 0) + if err = corredor.Setup(log, opts.Corredor); err != nil { + return err + } + return } func (app *App) Initialize(ctx context.Context) (err error) { - defer sentry.Recover() _, err = db.TryToConnect(ctx, app.log, app.opt.DB) @@ -67,6 +70,10 @@ func (app *App) Initialize(ctx context.Context) (err error) { return errors.Wrap(err, "could not connect to database") } + if err = corredor.Service().Connect(ctx); err != nil { + return + } + return } @@ -75,9 +82,6 @@ func (app *App) Upgrade(ctx context.Context) error { } func (app *App) Activate(ctx context.Context) (err error) { - if err = corredor.Start(ctx, app.log, app.opt.Corredor); err != nil { - return err - } // Start scheduler scheduler.Service().Start(ctx) diff --git a/messaging/app.go b/messaging/app.go index 1f81ef36b..b96b11c0c 100644 --- a/messaging/app.go +++ b/messaging/app.go @@ -47,6 +47,9 @@ func (app *App) Setup(log *zap.Logger, opts *app.Options) (err error) { PingPeriod: opts.Websocket.PingPeriod, }) + // @todo Wire in cross-service JWT maker for Corredor + corredor.Service().SetUserFinder(nil) + return } @@ -81,9 +84,6 @@ func (app *App) Activate(ctx context.Context) (err error) { service.Watchers(ctx) websocket.Watch(ctx) - // Wire in cross service JWT maker for Corredor - corredor.Service().SetJwtMaker(corredor.CrossServiceAuthTokenMaker()) - return } diff --git a/monolith/app.go b/monolith/app.go index 245179cd9..641239d13 100644 --- a/monolith/app.go +++ b/monolith/app.go @@ -15,6 +15,7 @@ import ( "github.com/cortezaproject/corteza-server/pkg/app" "github.com/cortezaproject/corteza-server/pkg/corredor" "github.com/cortezaproject/corteza-server/system" + systemService "github.com/cortezaproject/corteza-server/system/service" ) type ( @@ -33,13 +34,19 @@ func (monolith *App) Setup(log *zap.Logger, opts *app.Options) (err error) { // This will alter the auth settings provision procedure system.IsMonolith = true - return app.RunSetup( + err = app.RunSetup( log, opts, monolith.System, monolith.Compose, monolith.Messaging, ) + + if err != nil { + return + } + + return } func (monolith *App) Upgrade(ctx context.Context) (err error) { @@ -52,12 +59,20 @@ func (monolith *App) Upgrade(ctx context.Context) (err error) { } func (monolith App) Initialize(ctx context.Context) (err error) { - return app.RunInitialize( + err = app.RunInitialize( ctx, monolith.System, monolith.Compose, monolith.Messaging, ) + + if err != nil { + return + } + + corredor.Service().SetUserFinder(systemService.DefaultUser) + + return } func (monolith *App) Activate(ctx context.Context) (err error) { @@ -72,9 +87,6 @@ func (monolith *App) Activate(ctx context.Context) (err error) { return } - // Override JWT maker for Corredor with internal - corredor.Service().SetJwtMaker(corredor.InternalAuthTokenMaker()) - return } diff --git a/pkg/corredor/authtoken.go b/pkg/corredor/authtoken.go deleted file mode 100644 index 3e410515a..000000000 --- a/pkg/corredor/authtoken.go +++ /dev/null @@ -1,25 +0,0 @@ -package corredor - -// Authentication token maker must be able to convert user's handle or email -// into valid authentication token with short expiration - -// Used by non-system services -func CrossServiceAuthTokenMaker() AuthTokenMaker { - return func(user string) (s string, err error) { - panic("not implemented") - return "", nil - } -} - -// InternalAuthTokenMaker used by system or by all services when running in monolith mode -func InternalAuthTokenMaker() AuthTokenMaker { - return func(user string) (s string, err error) { - panic("not implemented") - // @todo implementation - // - // DefaultUser.FindByAny(user) - // auth.TokenEncoder - - return "", nil - } -} diff --git a/pkg/corredor/conn_test.go b/pkg/corredor/conn_test.go index e8328b02d..dca01b134 100644 --- a/pkg/corredor/conn_test.go +++ b/pkg/corredor/conn_test.go @@ -58,8 +58,7 @@ func TestNewConnection(t *testing.T) { grpcClientConn, err := NewConnection(ctx, opt, dbgLog) a.NoError(err) - // Go and - NewService(grpcClientConn, nil, dbgLog, opt) + // NewService(dbgLog, opt) grpcClientConn.WaitForStateChange(ctx, connectivity.Ready) grpcServer.GracefulStop() diff --git a/pkg/corredor/service.go b/pkg/corredor/service.go index 221f4d7cf..8cf9edf81 100644 --- a/pkg/corredor/service.go +++ b/pkg/corredor/service.go @@ -13,6 +13,7 @@ import ( "github.com/cortezaproject/corteza-server/pkg/app/options" "github.com/cortezaproject/corteza-server/pkg/auth" "github.com/cortezaproject/corteza-server/pkg/eventbus" + "github.com/cortezaproject/corteza-server/system/types" ) type ( @@ -22,24 +23,27 @@ type ( opt options.CorredorOpt // list of all registered triggers - // + // map[] registered map[string][]uintptr // list of all registered onManual triggers & scripts - // map[script-name][resource]bool - manual map[string]map[string]bool + // map[][] = + manual map[string]map[string]string // Combined list of client and server scripts sScripts ScriptSet cScripts ScriptSet + conn *grpc.ClientConn + ssClient ServerScriptsClient csClient ClientScriptsClient log *zap.Logger - eventbus eventRegistrator - jwtMaker AuthTokenMaker + eventRegistry eventRegistry + authTokenMaker authTokenMaker + users userFinder } Event interface { @@ -53,12 +57,18 @@ type ( Decode(map[string][]byte) error } - eventRegistrator interface { + eventRegistry interface { Register(h eventbus.Handler, ops ...eventbus.TriggerRegOp) uintptr Unregister(ptrs ...uintptr) } - AuthTokenMaker func(user string) (string, error) + userFinder interface { + FindByAny(interface{}) (*types.User, error) + } + + authTokenMaker interface { + Encode(auth.Identifiable) string + } ) const onManualEventType = "onManual" @@ -73,38 +83,58 @@ func Service() *service { } // Start connects to Corredor & initialize service -func Start(ctx context.Context, logger *zap.Logger, opt options.CorredorOpt) (err error) { +func Setup(logger *zap.Logger, opt options.CorredorOpt) (err error) { if gCorredor != nil { // Prevent multiple initializations return } - var ( - conn *grpc.ClientConn - ) - - if conn, err = NewConnection(ctx, opt, logger); err != nil { - return - } - - gCorredor = NewService(conn, eventbus.Service(), logger, opt) + gCorredor = NewService(logger, opt) return } -func NewService(conn *grpc.ClientConn, er eventRegistrator, logger *zap.Logger, opt options.CorredorOpt) *service { +func NewService(logger *zap.Logger, opt options.CorredorOpt) *service { return &service{ - ssClient: NewServerScriptsClient(conn), - csClient: NewClientScriptsClient(conn), - log: logger.Named("corredor"), + log: logger.Named("corredor"), + opt: opt, + registered: make(map[string][]uintptr), - manual: make(map[string]map[string]bool), - eventbus: er, - opt: opt, + manual: make(map[string]map[string]string), + + authTokenMaker: auth.DefaultJwtHandler, + eventRegistry: eventbus.Service(), } } -func (svc *service) SetJwtMaker(fn AuthTokenMaker) { - svc.jwtMaker = fn +func (svc *service) Connect(ctx context.Context) (err error) { + if err = svc.connect(ctx); err != nil { + return + } + + svc.ssClient = NewServerScriptsClient(svc.conn) + svc.csClient = NewClientScriptsClient(svc.conn) + + return +} + +func (svc *service) connect(ctx context.Context) (err error) { + if svc.conn, err = NewConnection(ctx, svc.opt, svc.log); err != nil { + return + } + + return +} + +func (svc *service) SetEventRegistry(er eventRegistry) { + svc.eventRegistry = er +} + +func (svc *service) SetAuthTokenMaker(atm authTokenMaker) { + svc.authTokenMaker = atm +} + +func (svc *service) SetUserFinder(uf userFinder) { + svc.users = uf } func (svc *service) Load(ctx context.Context) { @@ -140,21 +170,24 @@ func (svc service) ExecOnManual(ctx context.Context, script string, event Event) var ( res = event.ResourceType() evt = event.EventType() + + ok bool + runAs string ) if onManualEventType != evt { return errors.Errorf("triggered event type is not onManual (%q)", evt) } - if _, ok := svc.manual[script]; !ok { + if _, ok = svc.manual[script]; !ok { return errors.Errorf("unregistered onManual script %q", script) } - if _, ok := svc.manual[script][res]; !ok { + if runAs, ok = svc.manual[script][res]; !ok { return errors.Errorf("unregistered onManual script %q for resource %q", script, res) } - return svc.exec(ctx, script, event) + return svc.exec(ctx, script, runAs, event) } func (svc *service) loadServerScripts(ctx context.Context) { @@ -163,12 +196,8 @@ func (svc *service) loadServerScripts(ctx context.Context) { rsp *ServerScriptListResponse ) - if svc.jwtMaker == nil { - // @todo - // return errors.New("can not load corredor scripts without jwt maker") - } - svc.log.Debug("reloading server scripts") + rsp, err = svc.ssClient.List(ctx, &ServerScriptListRequest{}, grpc.WaitForReady(true)) if err != nil { svc.log.Error("could not load corredor server scripts", zap.Error(err)) @@ -185,13 +214,13 @@ func (svc *service) registerServerScripts(ss ...*ServerScript) { // Remove all previously registered triggers for _, ptrs := range svc.registered { if len(ptrs) > 0 { - svc.eventbus.Unregister(ptrs...) + svc.eventRegistry.Unregister(ptrs...) } } // Reset indexes svc.registered = make(map[string][]uintptr) - svc.manual = make(map[string]map[string]bool) + svc.manual = make(map[string]map[string]string) for i, script := range ss { svc.sScripts[i] = &Script{ @@ -232,7 +261,7 @@ func (svc *service) registerTriggers(script *ServerScript) []uintptr { for i := range script.Triggers { // We're modifying trigger in the loop, // so let's make a copy we can play with - trigger := *script.Triggers[i] + trigger := script.Triggers[i] if len(trigger.Events) == 0 { // We've removed the last event @@ -242,7 +271,7 @@ func (svc *service) registerTriggers(script *ServerScript) []uintptr { continue } - if ops, err = makeTriggerOpts(&trigger); err != nil { + if ops, err = makeTriggerOpts(trigger); err != nil { log.Warn( "could not make trigger options", zap.Error(err), @@ -251,7 +280,15 @@ func (svc *service) registerTriggers(script *ServerScript) []uintptr { continue } - ptr := svc.eventbus.Register(makeEventHandler(svc, script.Name, trigger.RunAs), ops...) + ptr := svc.eventRegistry.Register(func(ctx context.Context, ev eventbus.Event) (err error) { + // Is this compatible event? + if cce, ok := ev.(Event); ok { + // Can only work with corteza compatible events + return svc.exec(ctx, script.Name, trigger.RunAs, cce) + } + + return nil + }, ops...) ptrs = append(ptrs, ptr) } @@ -262,7 +299,7 @@ func (svc *service) registerTriggers(script *ServerScript) []uintptr { // // It does not do any constraints checking - this is the responsibility of the // individual event implemntation -func (svc service) exec(ctx context.Context, script string, event Event) (err error) { +func (svc service) exec(ctx context.Context, script string, runAs string, event Event) (err error) { var ( rsp *ExecResponse @@ -271,7 +308,6 @@ func (svc service) exec(ctx context.Context, script string, event Event) (err er log = svc.log.With( zap.String("script", script), - zap.Stringer("runAs", auth.GetIdentityFromContext(ctx)), zap.String("event", event.EventType()), zap.String("resource", event.ResourceType()), ) @@ -296,17 +332,61 @@ func (svc service) exec(ctx context.Context, script string, event Event) (err er req.Args[key] = string(encodedEvent[key]) } + // Resolve/expand invoker user details from the context + invoker, err := svc.users.FindByAny(ctx) + if err != nil { + return err + } + + log = log.With(zap.Stringer("invoker", invoker)) + if err = encodeArguments(req.Args, "invoker", invoker); err != nil { + return + } + + if len(runAs) > 0 { + if !svc.opt.RunAsEnabled { + return errors.New("could not make runner context, run-as disabled") + } + + var definer auth.Identifiable + + // Run this script as defined user (definer) + // + // We search for the defined (run-as) user, + // assign it to authUser argument and make an + // authentication token for it + definer, err = svc.users.FindByAny(runAs) + if err != nil { + return err + } + + log = log.With(zap.Stringer("run-as", definer)) + + // current (authenticated) user + if err = encodeArguments(req.Args, "authUser", definer); err != nil { + return + } + + if err = encodeArguments(req.Args, "authToken", svc.authTokenMaker.Encode(definer)); err != nil { + return + } + + } else { + // Run script with the same user that invoked it + + // current (authenticated) user + if err = encodeArguments(req.Args, "authUser", invoker); err != nil { + return + } + + if err = encodeArguments(req.Args, "authToken", svc.authTokenMaker.Encode(invoker)); err != nil { + return + } + } + // //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// // Additional (string) arguments - // pass security credentials - if err = encodeArguments(req.Args, "authUser", auth.GetIdentityFromContext(ctx)); err != nil { - return - } - if err = encodeArguments(req.Args, "jwt", auth.GetJwtFromContext(ctx)); err != nil { - return - } - // basic event/event info if err = encodeArguments(req.Args, "event", event.EventType()); err != nil { return @@ -332,23 +412,28 @@ func (svc service) exec(ctx context.Context, script string, event Event) (err er ) if err != nil { + // See if this was a "soft abort" + // + // This means, we do not make any logs of this just + // tell the caller that the call was aborted s := status.Convert(err) if s != nil && s.Code() == codes.Aborted { // Special care for errors with Aborted code msg := s.Message() if len(msg) == 0 { + // No extra message, fallback to "aborted" msg = "Aborted" } return errors.New(msg) } - log.Debug("corredor responded with error", zap.Error(err)) + log.Warn("corredor responded with error", zap.Error(err)) return errors.New("failed to execute corredor script") } - log.Debug("corredor responded", zap.Any("result", rsp.Result)) + log.Info("executed", zap.Any("result", rsp.Result)) // //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// diff --git a/pkg/corredor/util.go b/pkg/corredor/util.go index 555405061..7dd72f36d 100644 --- a/pkg/corredor/util.go +++ b/pkg/corredor/util.go @@ -1,11 +1,9 @@ package corredor import ( - "context" "encoding/json" "fmt" - "github.com/cortezaproject/corteza-server/pkg/auth" "github.com/cortezaproject/corteza-server/pkg/eventbus" "github.com/cortezaproject/corteza-server/pkg/slice" ) @@ -29,19 +27,19 @@ func popOnManualEventType(trigger *Trigger) (found bool) { // pluckManualTriggers removes all manual triggers from the list of script's triggers // // and returns a hash map with resources from these manual triggers -func pluckManualTriggers(script *ServerScript) map[string]bool { +func pluckManualTriggers(script *ServerScript) map[string]string { var ( - hash = make(map[string]bool) + hash = make(map[string]string) ) for i := range script.Triggers { // We're modifying trigger in the loop, // so let's make a copy we can play with - trigger := *script.Triggers[i] + trigger := script.Triggers[i] - if popOnManualEventType(&trigger) { + if popOnManualEventType(trigger) { for _, res := range trigger.Resources { - hash[res] = true + hash[res] = trigger.RunAs } } } @@ -72,28 +70,6 @@ func makeTriggerOpts(t *Trigger) (oo []eventbus.TriggerRegOp, err error) { return } -// makes event-handler callback -func makeEventHandler(svc *service, script string, runAs string) eventbus.Handler { - return func(ctx context.Context, ev eventbus.Event) error { - // Is this compatible event? - - if ce, ok := ev.(Event); ok { - if len(runAs) > 0 { - jwt, err := svc.jwtMaker(runAs) - if err != nil { - return err - } - - ctx = auth.SetJwtToContext(ctx, jwt) - } - - return svc.exec(ctx, script, ce) - } - - return nil - } -} - // encode adds entry (with json encoded value) to hash map // used to prepare data for transmission func encodeArguments(args map[string]string, key string, val interface{}) (err error) { diff --git a/system/app.go b/system/app.go index 94e775a42..2dbf92f1c 100644 --- a/system/app.go +++ b/system/app.go @@ -42,6 +42,9 @@ func (app *App) Setup(log *zap.Logger, opts *app.Options) (err error) { event.SystemOnTimestamp(), ) + // Wire in cross-service JWT maker for Corredor + corredor.Service().SetUserFinder(service.DefaultUser) + return } @@ -78,9 +81,6 @@ func (app *App) Activate(ctx context.Context) (err error) { service.Watchers(ctx) - // Wire in internal JWT maker for Corredor - corredor.Service().SetJwtMaker(corredor.InternalAuthTokenMaker()) - return } diff --git a/system/service/user.go b/system/service/user.go index b7052a19c..4e8b4eac7 100644 --- a/system/service/user.go +++ b/system/service/user.go @@ -46,6 +46,7 @@ type ( eventbus eventDispatcher user repository.UserRepository + role repository.RoleRepository credentials repository.CredentialsRepository // @todo wire this with settings (privacy.mask.email) @@ -85,7 +86,7 @@ type ( FindByEmail(email string) (*types.User, error) FindByHandle(handle string) (*types.User, error) FindByID(id uint64) (*types.User, error) - FindByAny(any string) (*types.User, error) + FindByAny(identifier interface{}) (*types.User, error) Find(types.UserFilter) (types.UserSet, types.UserFilter, error) Create(input *types.User) (*types.User, error) @@ -143,6 +144,7 @@ func (svc user) With(ctx context.Context) UserService { subscription: svc.subscription, user: repository.User(ctx, db), + role: repository.Role(ctx, db), credentials: repository.Credentials(ctx, db), privacyMaskEmail: svc.privacyMaskEmail, @@ -170,18 +172,37 @@ func (svc user) FindByHandle(handle string) (*types.User, error) { return svc.proc(svc.user.FindByHandle(handle)) } -func (svc user) FindByAny(any string) (*types.User, error) { - return svc.proc(func() (*types.User, error) { - if id, _ := strconv.ParseUint(any, 10, 64); id > 0 { - return svc.user.FindByID(id) - } +// FindByAny finds user by given identifier (context, id, handle, email) +func (svc user) FindByAny(identifier interface{}) (u *types.User, err error) { + if ctx, ok := identifier.(context.Context); ok { + identifier = internalAuth.GetIdentityFromContext(ctx).Identity() + } - if strings.Contains(any, "@") { - return svc.user.FindByEmail(any) + if ID, ok := identifier.(uint64); ok { + u, err = svc.FindByID(ID) + } else if strIdentifier, ok := identifier.(string); ok { + if ID, _ := strconv.ParseUint(strIdentifier, 10, 64); ID > 0 { + u, err = svc.FindByID(ID) + } else if strings.Contains(strIdentifier, "@") { + u, err = svc.FindByEmail(strIdentifier) + } else { + u, err = svc.FindByHandle(strIdentifier) } + } else { + err = ErrInvalidID + } - return svc.user.FindByHandle(any) - }()) + if err != nil { + return + } + + rr, _, err := svc.role.Find(types.RoleFilter{MemberID: u.ID}) + if err != nil { + return nil, err + } + + u.SetRoles(rr.IDs()) + return } func (svc user) proc(u *types.User, err error) (*types.User, error) {