DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Refactors corredor initialization, run-as logic & trigger execution

Browse Source
This commit is contained in:
Denis Arh
2020-01-18 07:05:34 +01:00
parent 5eac027541
commit c0d1594eb3
10 changed files with 208 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
compose/app.go
+4 -4
View File
@@ -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
}
corteza/corteza.go
+8 -4
View File
@@ -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)
messaging/app.go
+3 -3
View File
@@ -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
}
monolith/app.go
+17 -5
View File
@@ -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
}
pkg/corredor/authtoken.go
-25
View File
@@ -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
}
}
pkg/corredor/conn_test.go
+1 -2
View File
@@ -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()
pkg/corredor/service.go
+136 -51
View File
@@ -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[<script-name>]
registered map[string][]uintptr
// list of all registered onManual triggers & scripts
// map[script-name][resource]bool
manual map[string]map[string]bool
// map[<script-name>][<resource>] = <run-as>
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))
// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// //// ////
pkg/corredor/util.go
+5 -29
View File
@@ -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) {
system/app.go
+3 -3
View File
@@ -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
}
system/service/user.go
+31 -10
View File
@@ -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) {