feat: add scopes to authorization (#1288)

Signed-off-by: Sidhant Kohli <[email protected]>

config/advanced-install/namespaced-numaflow-server.yaml

@@ -84,15 +84,21 @@ subjects:
 ---
 apiVersion: v1
 data:
+  rbac-conf.yaml: |
+    policy.default: role:readonly
+    # The scopes field controls which authentication scopes to examine during rbac enforcement.
+    # We can have multiple scopes, and the first scope that matches with the policy will be used.
+    # The default value is "groups", which means that the groups field of the user's token will be examined
+    # The other possible value is "email", which means that the email field of the user's token will be examined
+    # It can be provided as a comma-separated list, e.g "groups,email"
+    # policy.scopes: groups,email
   rbac-policy.csv: |
     # Policies go here
     p, role:admin, *, *, *
     p, role:readonly, *, *, GET
     # Groups go here
     # g, admin, role:admin
     # g, my-github-org:my-github-team, role:readonly
-  rbac.conf: |
-    policy.default: role:readonly
 kind: ConfigMap
 metadata:
   name: numaflow-server-rbac-config

config/advanced-install/numaflow-server.yaml

@@ -87,15 +87,21 @@ subjects:
 ---
 apiVersion: v1
 data:
+  rbac-conf.yaml: |
+    policy.default: role:readonly
+    # The scopes field controls which authentication scopes to examine during rbac enforcement.
+    # We can have multiple scopes, and the first scope that matches with the policy will be used.
+    # The default value is "groups", which means that the groups field of the user's token will be examined
+    # The other possible value is "email", which means that the email field of the user's token will be examined
+    # It can be provided as a comma-separated list, e.g "groups,email"
+    # policy.scopes: groups,email
   rbac-policy.csv: |
     # Policies go here
     p, role:admin, *, *, *
     p, role:readonly, *, *, GET
     # Groups go here
     # g, admin, role:admin
     # g, my-github-org:my-github-team, role:readonly
-  rbac.conf: |
-    policy.default: role:readonly
 kind: ConfigMap
 metadata:
   name: numaflow-server-rbac-config

config/base/numaflow-server/numaflow-server-rbac-config.yaml

@@ -10,5 +10,11 @@ data:
     # Groups go here
     # g, admin, role:admin
     # g, my-github-org:my-github-team, role:readonly
-  rbac.conf: |
+  rbac-conf.yaml: |
     policy.default: role:readonly
+    # The scopes field controls which authentication scopes to examine during rbac enforcement.
+    # We can have multiple scopes, and the first scope that matches with the policy will be used.
+    # The default value is "groups", which means that the groups field of the user's token will be examined
+    # The other possible value is "email", which means that the email field of the user's token will be examined
+    # It can be provided as a comma-separated list, e.g "groups,email"
+    # policy.scopes: groups,email

config/install.yaml

@@ -16388,15 +16388,21 @@ metadata:
 ---
 apiVersion: v1
 data:
+  rbac-conf.yaml: |
+    policy.default: role:readonly
+    # The scopes field controls which authentication scopes to examine during rbac enforcement.
+    # We can have multiple scopes, and the first scope that matches with the policy will be used.
+    # The default value is "groups", which means that the groups field of the user's token will be examined
+    # The other possible value is "email", which means that the email field of the user's token will be examined
+    # It can be provided as a comma-separated list, e.g "groups,email"
+    # policy.scopes: groups,email
   rbac-policy.csv: |
     # Policies go here
     p, role:admin, *, *, *
     p, role:readonly, *, *, GET
     # Groups go here
     # g, admin, role:admin
     # g, my-github-org:my-github-team, role:readonly
-  rbac.conf: |
-    policy.default: role:readonly
 kind: ConfigMap
 metadata:
   name: numaflow-server-rbac-config

config/namespace-install.yaml

@@ -16292,15 +16292,21 @@ metadata:
 ---
 apiVersion: v1
 data:
+  rbac-conf.yaml: |
+    policy.default: role:readonly
+    # The scopes field controls which authentication scopes to examine during rbac enforcement.
+    # We can have multiple scopes, and the first scope that matches with the policy will be used.
+    # The default value is "groups", which means that the groups field of the user's token will be examined
+    # The other possible value is "email", which means that the email field of the user's token will be examined
+    # It can be provided as a comma-separated list, e.g "groups,email"
+    # policy.scopes: groups,email
   rbac-policy.csv: |
     # Policies go here
     p, role:admin, *, *, *
     p, role:readonly, *, *, GET
     # Groups go here
     # g, admin, role:admin
     # g, my-github-org:my-github-team, role:readonly
-  rbac.conf: |
-    policy.default: role:readonly
 kind: ConfigMap
 metadata:
   name: numaflow-server-rbac-config

server/authz/consts.go

@@ -20,6 +20,10 @@ const (
 	// PolicyMapPath is the path to the policy map.
 	policyMapPath = "/etc/numaflow/rbac-policy.csv"
 
+	// rbacPropertiesPath is the path to the rbac properties file. It includes configuraion for authorization like
+	// scope, default policy etc.
+	rbacPropertiesPath = "/etc/numaflow/rbac-conf.yaml"
+
 	// Objects for the RBAC policy
 	ObjectAll      = "*"
 	ObjectPipeline = "pipeline"
@@ -32,4 +36,13 @@ const (
 
 	// MatchAll is a wildcard to match all patterns
 	MatchAll = "*"
+
+	// RbacProperties contain the different properties for RBAC configuration
+	RbacPropertyScopes        = "policy.scopes"
+	RbacPropertyDefaultPolicy = "policy.default"
+
+	// Auth scopes supported
+	ScopeGroup   = "groups"
+	ScopeEmail   = "email"
+	ScopeDefault = "default"
 )

server/authz/interface.go

@@ -1,12 +1,15 @@
 package authz
 
-import "github.com/gin-gonic/gin"
+import (
+	"github.com/gin-gonic/gin"
+
+	"github.com/numaproj/numaflow/server/authn"
+)
 
 type Authorizer interface {
 	// Authorize checks if a user is authorized to access the resource.
-	// c is the gin context.
-	// g is the list of groups the user belongs to.
 	// Authorize trusts that the user is already authenticated and directly uses the groups to authorize the user.
 	// please don't use gin to get the user information again.
-	Authorize(c *gin.Context, g []string) (bool, error)
+	// Authorize returns true if the user is authorized, otherwise false.
+	Authorize(c *gin.Context, userInfo *authn.UserInfo) bool
 }

server/authz/rbac.go

@@ -20,49 +20,119 @@ import (
 	_ "embed"
 	"fmt"
 	"path"
+	"strings"
 
 	"github.com/casbin/casbin/v2"
 	"github.com/casbin/casbin/v2/model"
 	fileadapter "github.com/casbin/casbin/v2/persist/file-adapter"
+	"github.com/fsnotify/fsnotify"
 	"github.com/gin-gonic/gin"
+	"github.com/spf13/viper"
+	"k8s.io/utils/strings/slices"
+
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+	"github.com/numaproj/numaflow/server/authn"
 )
 
 var (
 	//go:embed rbac-model.conf
 	rbacModel string
+	logger    = logging.NewLogger()
 )
 
 const (
 	emptyString = ""
 )
 
 type CasbinObject struct {
-	enforcer *casbin.Enforcer
+	enforcer      *casbin.Enforcer
+	userPermCount map[string]int
+	currentScopes []string
+	policyDefault string
+	configReader  *viper.Viper
 }
 
 func NewCasbinObject() (*CasbinObject, error) {
 	enforcer, err := getEnforcer()
 	if err != nil {
 		return nil, err
 	}
-	return &CasbinObject{
-		enforcer: enforcer,
-	}, nil
+	configReader := viper.New()
+	configReader.SetConfigFile(rbacPropertiesPath)
+	err = configReader.ReadInConfig()
+	if err != nil {
+		return nil, err
+	}
+	currentScopes := getRBACScopes(configReader)
+	logger.Infow("Auth Scopes", "scopes", currentScopes)
+	// Set the default policy for authorization.
+	policyDefault := getDefaultPolicy(configReader)
+	userPermCount := make(map[string]int)
+
+	cas := &CasbinObject{
+		enforcer:      enforcer,
+		userPermCount: userPermCount,
+		currentScopes: currentScopes,
+		policyDefault: policyDefault,
+		configReader:  configReader,
+	}
+
+	// Watch for changes in the config file.
+	configReader.WatchConfig()
+	configReader.OnConfigChange(func(in fsnotify.Event) {
+		cas.configFileReload(in)
+	})
+
+	return cas, nil
 }
 
-func (cas *CasbinObject) Authorize(c *gin.Context, groups []string) (bool, error) {
+// Authorize checks if a user is authorized to access the resource.
+// It returns true if the user is authorized, otherwise false.
+// It also returns the policy count of the user. The policy count is used to check if there are any policies defined
+// for the given user, if not we will allocate a default policy for the user.
+func (cas *CasbinObject) Authorize(c *gin.Context, userInfo *authn.UserInfo) bool {
+	// Get the scopes to check from the policy.
+	scopedList := getSubjectFromScope(cas.currentScopes, userInfo)
+	// Get the resource, object and action from the request.
 	resource := extractResource(c)
 	object := extractObject(c)
 	action := c.Request.Method
-	// Check if the user has permission for any of the groups.
-	for _, group := range groups {
-		// Get the user from the group. The group is in the format "group:role".
-		// Check if the user has permission using Casbin Enforcer.
-		if ok, _ := cas.enforcer.Enforce(group, resource, object, action); ok {
-			return true, nil
+	userHasPolicies := false
+	// Check for the given scoped list if the user is authorized using any of the subjects in the list.
+	for _, scopedSubject := range scopedList {
+		// Check if the user has permissions in the policy for the given scoped subject.
+		userHasPolicies = userHasPolicies || hasPermissionsDefined(cas.enforcer, scopedSubject, cas.userPermCount)
+		if ok := enforceCheck(cas.enforcer, scopedSubject, resource, object, action); ok {
+			return ok
+		}
+	}
+	// If the user does not have any policy defined, allocate a default policy for the user.
+	if !userHasPolicies {
+		logger.Infow("No policy defined for the user, allocating default policy",
+			"DefaultPolicy", cas.policyDefault)
+		ok := enforceCheck(cas.enforcer, cas.policyDefault, resource, object, action)
+		if ok {
+			return ok
 		}
 	}
-	return false, fmt.Errorf("user is not authorized to execute the requested action")
+	return false
+}
+
+// getSubjectFromScope returns the subjects in the request for the given scopes.
+// The scopes are the params used to check the authentication params to check if the
+// user is authorized to access the resource. For any new scope, add the scope to the
+// rbac properties file and add the scope to the cases below.
+func getSubjectFromScope(scopes []string, userInfo *authn.UserInfo) []string {
+	var scopedList []string
+	// If the scope is group, fetch the groups from the user identity token.
+	if slices.Contains(scopes, ScopeGroup) {
+		scopedList = append(scopedList, userInfo.IDTokenClaims.Groups...)
+	}
+	// If the scope is email, fetch the email from the user identity token.
+	if slices.Contains(scopes, ScopeEmail) {
+		scopedList = append(scopedList, userInfo.IDTokenClaims.Email)
+	}
+	return scopedList
 }
 
 // getEnforcer initializes the Casbin Enforcer with the model and policy.
@@ -153,3 +223,98 @@ func extractObject(c *gin.Context) string {
 	}
 	return emptyString
 }
+
+// getRbacProperty is used to read the rbacPropertiesPath file path and extract the policy provided as argument,
+func getRbacProperty(property string, config *viper.Viper) interface{} {
+	val := config.Get(property)
+	if val == nil {
+		return emptyString
+	}
+	return val
+}
+
+// getRBACScopes returns the scopes from the rbac properties file. If no scopes are provided, it returns Group as the
+// default scope. The scopes are used to determine the user identity token to be used for authorization.
+// If the scope is group, the user identity token will be the groups assigned to the user from the authentication
+// system. If the scope is email, the user identity token will be the email assigned to
+// the user from the authentication.
+// The scopes are provided as a comma separated list in the rbac properties file.
+// Example: policy.scopes=groups,email
+func getRBACScopes(config *viper.Viper) []string {
+	scopes := getRbacProperty(RbacPropertyScopes, config)
+	var retList []string
+	// If no scopes are provided, set Group as the default scope.
+	if scopes == emptyString {
+		retList = append(retList, ScopeGroup)
+		return retList
+	}
+	scopes = strings.Split(scopes.(string), ",")
+	for _, scope := range scopes.([]string) {
+		scope = strings.TrimSpace(scope)
+		retList = append(retList, scope)
+	}
+	return retList
+}
+
+// enforceCheck checks if the user has permission based on the Casbin model and policy.
+// It returns true if the user is authorized, otherwise false.
+func enforceCheck(enforcer *casbin.Enforcer, user, resource, object, action string) bool {
+	ok, _ := enforcer.Enforce(user, resource, object, action)
+	return ok
+}
+
+// configFileReload is used to reload the config file when it is changed. This is used to reload the policy without
+// restarting the server. The config file is in the format of yaml. The config file is read by viper.
+func (cas *CasbinObject) configFileReload(e fsnotify.Event) {
+	logger.Infow("RBAC conf file updated:", "fileName", e.Name)
+	err := cas.configReader.ReadInConfig()
+	if err != nil {
+		return
+	}
+	// update the scopes
+	newScopes := getRBACScopes(cas.configReader)
+	cas.currentScopes = newScopes
+	// update the default policy
+	cas.policyDefault = getDefaultPolicy(cas.configReader)
+	// clear the userPermCount cache
+	cas.userPermCount = make(map[string]int)
+	logger.Infow("Auth Scopes Updated", "scopes", cas.currentScopes)
+}
+
+// getDefaultPolicy returns the default policy from the rbac properties file. The default policy is used when the
+// requested resource is not present in the policy.
+// The default policy is provided in the rbac properties file in the format "policy.default: value"
+// Example: policy.default: deny
+func getDefaultPolicy(config *viper.Viper) string {
+	defaultPolicy := getRbacProperty(RbacPropertyDefaultPolicy, config)
+	return defaultPolicy.(string)
+}
+
+// hasPermissionsDefined checks if the user has permissions defined in the policy. It returns true if the user has
+// permissions in the policy and false if the user does not have permissions in the policy.
+// We have a cache userPermCount to store the count of permissions for a user. If the user has permissions in the
+// policy, we store the count in the cache and return based on the value.
+// If the user does not have permissions in the policy, we add it to the cache before returning
+func hasPermissionsDefined(enforcer *casbin.Enforcer, user string, userPermCount map[string]int) bool {
+	// check if user exists in userPermCount
+	if userPermCount == nil {
+		userPermCount = make(map[string]int)
+	}
+	val, ok := userPermCount[user]
+	// If the key exists
+	if ok {
+		// Return true if the user has permissions in the policy
+		// and false if the user does not have permissions in the policy.
+		return val > 0
+	}
+	// get the permissions for the user
+	cnt, err := enforcer.GetImplicitPermissionsForUser(user)
+	if err != nil {
+		logger.Errorw("Failed to get permissions for user", "user", user, "error", err)
+		return false
+	}
+	count := len(cnt)
+	// store the count in userPermCount
+	userPermCount[user] = count
+	return count > 0
+}