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

@@ -40,6 +40,13 @@
               "via": "email"
             }
           }
+        },
+        "idp_groups": {
+          "title": "IDP groups",
+          "type": "array",
+          "items": {
+              "type": "string"
+          }
         }
       },
       "required": [

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()

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;

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();

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