From 5e7fc110b21043c8804aa7767549a5ef82e2427f Mon Sep 17 00:00:00 2001 From: Akshay Gaikwad Date: Wed, 11 May 2022 18:47:11 +0530 Subject: [PATCH] Add IdP groups in Identities table The idp_groups is list of groups IdP user belongs to that is returning in the OIdC providers token response. The flow of Idp Group mapping is as follows: OIdC Provider (OP) return custom claim with groups in a token when authentication event | The value of custom claim is mapped to `idp_groups` of identity traint using JsonNet mapper. | On inserting/updating/deleting `identities` table, Postgresql sends a pg_notification with `PG_OPERATION,IDENTITY_ID,IDENTITY_TRAIN` as a payload. | The `pkg/service/user.UserService.UpdateIdpUserGroupPolicy` update the casbin policies for each notification based on payload received. --- _kratos/identity.schema.json | 7 ++++ main.go | 30 ++++++++++++- .../000038_pg_identities_triggers.down.sql | 2 + .../000038_pg_identities_triggers.up.sql | 23 ++++++++++ pkg/service/user.go | 42 +++++++++++++++++++ 5 files changed, 103 insertions(+), 1 deletion(-) create mode 100644 persistence/migrations/admindb/000038_pg_identities_triggers.down.sql create mode 100644 persistence/migrations/admindb/000038_pg_identities_triggers.up.sql diff --git a/_kratos/identity.schema.json b/_kratos/identity.schema.json index 09e1199..dd03208 100644 --- a/_kratos/identity.schema.json +++ b/_kratos/identity.schema.json @@ -40,6 +40,13 @@ "via": "email" } } + }, + "idp_groups": { + "title": "IDP groups", + "type": "array", + "items": { + "type": "string" + } } }, "required": [ diff --git a/main.go b/main.go index 150f8d5..38bf593 100644 --- a/main.go +++ b/main.go @@ -431,11 +431,11 @@ func run() { go runRelayPeerRPC(&wg, ctx) go runDebug(&wg, ctx) go runEventHandlers(&wg, ctx) + go runIdpGroupSync(&wg, ctx) <-ctx.Done() _log.Infow("shutting down, waiting for children to die") wg.Wait() - } func runAPI(wg *sync.WaitGroup, ctx context.Context) { @@ -675,6 +675,34 @@ func runDebug(wg *sync.WaitGroup, ctx context.Context) { s.Shutdown(ctx) } +func runIdpGroupSync(wg *sync.WaitGroup, ctx context.Context) { + defer wg.Done() + channel := "identities:changed" + ln := pgdriver.NewListener(db) +listen: + if err := ln.Listen(ctx, channel); err != nil { + _log.Errorf("error listening for notification on channel %q: %s", channel, err) + time.Sleep(2 * time.Second) + goto listen + } + + _log.Infof("Listening for notifications on channel %q", channel) + for n := range ln.Channel() { + _log.Info("A identities table notification received") + splitPl := strings.SplitN(n.Payload, ",", 3) + op := splitPl[0] + id := splitPl[1] + traits := splitPl[2] + err := us.UpdateIdpUserGroupPolicy(ctx, op, id, traits) + if err != nil { + _log.Warnf("Failed updating policy for IDP user with id %s: %s", id, err) + } else { + _log.Infof("Policies are updated successfully for IDP user with id %s", id) + } + } + <-ctx.Done() +} + func main() { setup() run() diff --git a/persistence/migrations/admindb/000038_pg_identities_triggers.down.sql b/persistence/migrations/admindb/000038_pg_identities_triggers.down.sql new file mode 100644 index 0000000..22e658d --- /dev/null +++ b/persistence/migrations/admindb/000038_pg_identities_triggers.down.sql @@ -0,0 +1,2 @@ +DROP FUNCTION IF EXISTS identities_after_change() CASCADE; +DROP TRIGGER IF EXISTS trigger_identities_update ON identities; diff --git a/persistence/migrations/admindb/000038_pg_identities_triggers.up.sql b/persistence/migrations/admindb/000038_pg_identities_triggers.up.sql new file mode 100644 index 0000000..3b57a95 --- /dev/null +++ b/persistence/migrations/admindb/000038_pg_identities_triggers.up.sql @@ -0,0 +1,23 @@ +CREATE OR REPLACE FUNCTION identities_after_change() RETURNS TRIGGER AS $$ + DECLARE + row RECORD; + output TEXT; + + BEGIN + IF (TG_OP = 'DELETE') THEN + row = OLD; + ELSE + row = NEW; + END IF; + + output = TG_OP || ',' || row.id || ',' || row.traits; + PERFORM pg_notify('identities:changed',output); + RETURN NULL; + END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER trigger_identities_update + AFTER INSERT OR UPDATE OR DELETE + ON identities + FOR EACH ROW + EXECUTE PROCEDURE identities_after_change(); diff --git a/pkg/service/user.go b/pkg/service/user.go index 3b6521e..08f190d 100644 --- a/pkg/service/user.go +++ b/pkg/service/user.go @@ -3,6 +3,7 @@ package service import ( "context" "database/sql" + "encoding/json" "fmt" "strings" "time" @@ -48,6 +49,8 @@ type UserService interface { List(context.Context, ...query.Option) (*userv3.UserList, error) // retrieve the cli config for the logged in user RetrieveCliConfig(ctx context.Context, req *userrpcv3.ApiKeyRequest) (*common.CliConfigDownloadData, error) + // Update UserGroup casbin for OIdC/Idp users + UpdateIdpUserGroupPolicy(context.Context, string, string, string) error } type userService struct { @@ -65,6 +68,7 @@ type userTraits struct { FirstName string LastName string Description string + IdpGroups []string `json:"idp_groups"` } // FIXME: find a better way to do this @@ -850,3 +854,41 @@ func (s *userService) RetrieveCliConfig(ctx context.Context, req *userrpcv3.ApiK return cliConfig, nil } + +func (s *userService) UpdateIdpUserGroupPolicy(ctx context.Context, op, id, traits string) error { + var userInfo userTraits + err := json.Unmarshal([]byte(traits), &userInfo) + if err != nil { + return fmt.Errorf("Encounterd error unmarshing payload to userInfo: %s", err) + } + switch op { + case "DELETE": + _, err = s.azc.DeleteUserGroups(ctx, &authzv1.UserGroup{Grp: "u:" + userInfo.Email}) + if err != nil { + return fmt.Errorf("error deleting UserGroups policy: %s", err) + } + case "UPDATE": + // delete old policies + _, err = s.azc.DeleteUserGroups(ctx, &authzv1.UserGroup{Grp: "u:" + userInfo.Email}) + if err != nil { + return fmt.Errorf("error deleting UserGroups policy: %s", err) + } + // create new policies + fallthrough + case "INSERT": + var ugs []*authzv1.UserGroup + for _, g := range utils.Unique(userInfo.IdpGroups) { + ugs = append(ugs, &authzv1.UserGroup{ + Grp: "g:" + g, + User: "u:" + userInfo.Email, + }) + } + _, err = s.azc.CreateUserGroups(ctx, &authzv1.UserGroups{UserGroups: ugs}) + if err != nil { + return fmt.Errorf("error creating UserGroups policy: %s", err) + } + default: + return fmt.Errorf("Unsupported %s operation in payload", op) + } + return nil +}