[pydantic issue] User token parsing fixed

Signed-off-by: jyejare <[email protected]>

Author: jyejare

infra/feast-operator/config/rbac/role.yaml

@@ -15,6 +15,12 @@ rules:
   - list
   - update
   - watch
+- apiGroups:
+  - authentication.k8s.io
+  resources:
+  - tokenreviews
+  verbs:
+  - create
 - apiGroups:
   - batch
   resources:
@@ -44,11 +50,13 @@ rules:
 - apiGroups:
   - ""
   resources:
+  - namespaces
   - pods
   - secrets
   verbs:
   - get
   - list
+  - watch
 - apiGroups:
   - ""
   resources:
@@ -84,8 +92,11 @@ rules:
 - apiGroups:
   - rbac.authorization.k8s.io
   resources:
+  - clusterrolebindings
+  - clusterroles
   - rolebindings
   - roles
+  - subjectaccessreviews
   verbs:
   - create
   - delete
@@ -104,32 +115,3 @@ rules:
   - list
   - update
   - watch
-# Token Access Review permissions for Feast server RBAC creation
-- apiGroups:
-  - authentication.k8s.io
-  resources:
-  - tokenreviews
-  verbs:
-  - create
-- apiGroups:
-  - rbac.authorization.k8s.io
-  resources:
-  - subjectaccessreviews
-  verbs:
-  - create
-- apiGroups:
-  - ""
-  resources:
-  - namespaces
-  verbs:
-  - get
-  - list
-  - watch
-- apiGroups:
-  - rbac.authorization.k8s.io
-  resources:
-  - clusterroles
-  - clusterrolebindings
-  verbs:
-  - get
-  - list

infra/feast-operator/internal/controller/authz/authz.go

@@ -37,13 +37,23 @@ func (authz *FeastAuthorization) deployKubernetesAuth() error {
 	if authz.isKubernetesAuth() {
 		authz.removeOrphanedRoles()
 
+		// Create namespace-scoped RBAC resources
 		if err := authz.createFeastRole(); err != nil {
 			return authz.setFeastKubernetesAuthCondition(err)
 		}
 		if err := authz.createFeastRoleBinding(); err != nil {
 			return authz.setFeastKubernetesAuthCondition(err)
 		}
 
+		// Create cluster-scoped RBAC resources (separate from namespace resources)
+		if err := authz.createFeastClusterRole(); err != nil {
+			return authz.setFeastKubernetesAuthCondition(err)
+		}
+		if err := authz.createFeastClusterRoleBinding(); err != nil {
+			return authz.setFeastKubernetesAuthCondition(err)
+		}
+
+		// Create custom auth roles
 		for _, roleName := range authz.Handler.FeatureStore.Status.Applied.AuthzConfig.KubernetesAuthz.Roles {
 			if err := authz.createAuthRole(roleName); err != nil {
 				return authz.setFeastKubernetesAuthCondition(err)
@@ -89,6 +99,80 @@ func (authz *FeastAuthorization) createFeastRole() error {
 	return nil
 }
 
+func (authz *FeastAuthorization) createFeastClusterRole() error {
+	logger := log.FromContext(authz.Handler.Context)
+	clusterRole := authz.initFeastClusterRole()
+	if op, err := controllerutil.CreateOrUpdate(authz.Handler.Context, authz.Handler.Client, clusterRole, controllerutil.MutateFn(func() error {
+		return authz.setFeastClusterRole(clusterRole)
+	})); err != nil {
+		return err
+	} else if op == controllerutil.OperationResultCreated || op == controllerutil.OperationResultUpdated {
+		logger.Info("Successfully reconciled", "ClusterRole", clusterRole.Name, "operation", op)
+	}
+
+	return nil
+}
+
+func (authz *FeastAuthorization) initFeastClusterRole() *rbacv1.ClusterRole {
+	clusterRole := &rbacv1.ClusterRole{
+		ObjectMeta: metav1.ObjectMeta{Name: authz.getFeastClusterRoleName()},
+	}
+	clusterRole.SetGroupVersionKind(rbacv1.SchemeGroupVersion.WithKind("ClusterRole"))
+	return clusterRole
+}
+
+func (authz *FeastAuthorization) setFeastClusterRole(clusterRole *rbacv1.ClusterRole) error {
+	clusterRole.Labels = authz.getLabels()
+	clusterRole.Rules = []rbacv1.PolicyRule{
+		{
+			APIGroups: []string{rbacv1.GroupName},
+			Resources: []string{"rolebindings"},
+			Verbs:     []string{"list"},
+		},
+	}
+	return nil
+}
+
+func (authz *FeastAuthorization) initFeastClusterRoleBinding() *rbacv1.ClusterRoleBinding {
+	clusterRoleBinding := &rbacv1.ClusterRoleBinding{
+		ObjectMeta: metav1.ObjectMeta{Name: authz.getFeastClusterRoleBindingName()},
+	}
+	clusterRoleBinding.SetGroupVersionKind(rbacv1.SchemeGroupVersion.WithKind("ClusterRoleBinding"))
+	return clusterRoleBinding
+}
+
+func (authz *FeastAuthorization) setFeastClusterRoleBinding(clusterRoleBinding *rbacv1.ClusterRoleBinding) error {
+	clusterRoleBinding.Labels = authz.getLabels()
+	clusterRoleBinding.Subjects = []rbacv1.Subject{
+		{
+			Kind:      "ServiceAccount",
+			Name:      authz.getFeastServiceAccountName(),
+			Namespace: authz.Handler.FeatureStore.Namespace,
+		},
+	}
+	clusterRoleBinding.RoleRef = rbacv1.RoleRef{
+		APIGroup: rbacv1.GroupName,
+		Kind:     "ClusterRole",
+		Name:     authz.getFeastClusterRoleName(),
+	}
+	return nil
+}
+
+// Create ClusterRoleBinding
+func (authz *FeastAuthorization) createFeastClusterRoleBinding() error {
+	logger := log.FromContext(authz.Handler.Context)
+	clusterRoleBinding := authz.initFeastClusterRoleBinding()
+	if op, err := controllerutil.CreateOrUpdate(authz.Handler.Context, authz.Handler.Client, clusterRoleBinding, controllerutil.MutateFn(func() error {
+		return authz.setFeastClusterRoleBinding(clusterRoleBinding)
+	})); err != nil {
+		return err
+	} else if op == controllerutil.OperationResultCreated || op == controllerutil.OperationResultUpdated {
+		logger.Info("Successfully reconciled", "ClusterRoleBinding", clusterRoleBinding.Name, "operation", op)
+	}
+
+	return nil
+}
+
 func (authz *FeastAuthorization) initFeastRole() *rbacv1.Role {
 	role := &rbacv1.Role{
 		ObjectMeta: metav1.ObjectMeta{Name: authz.getFeastRoleName(), Namespace: authz.Handler.FeatureStore.Namespace},
@@ -230,3 +314,23 @@ func (authz *FeastAuthorization) getFeastRoleName() string {
 func GetFeastRoleName(featureStore *feastdevv1alpha1.FeatureStore) string {
 	return services.GetFeastName(featureStore)
 }
+
+func (authz *FeastAuthorization) getFeastClusterRoleName() string {
+	return GetFeastClusterRoleName(authz.Handler.FeatureStore)
+}
+
+func GetFeastClusterRoleName(featureStore *feastdevv1alpha1.FeatureStore) string {
+	return services.GetFeastName(featureStore) + "-cluster"
+}
+
+func (authz *FeastAuthorization) getFeastClusterRoleBindingName() string {
+	return GetFeastClusterRoleBindingName(authz.Handler.FeatureStore)
+}
+
+func GetFeastClusterRoleBindingName(featureStore *feastdevv1alpha1.FeatureStore) string {
+	return services.GetFeastName(featureStore) + "-cluster-binding"
+}
+
+func (authz *FeastAuthorization) getFeastServiceAccountName() string {
+	return services.GetFeastName(authz.Handler.FeatureStore)
+}

infra/feast-operator/internal/controller/featurestore_controller.go

@@ -59,9 +59,10 @@ type FeatureStoreReconciler struct {
 // +kubebuilder:rbac:groups=feast.dev,resources=featurestores/finalizers,verbs=update
 // +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;create;update;watch;delete
 // +kubebuilder:rbac:groups=core,resources=services;configmaps;persistentvolumeclaims;serviceaccounts,verbs=get;list;create;update;watch;delete
-// +kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=roles;rolebindings,verbs=get;list;create;update;watch;delete
-// +kubebuilder:rbac:groups=core,resources=secrets;pods,verbs=get;list
+// +kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=roles;rolebindings;clusterroles;clusterrolebindings;subjectaccessreviews,verbs=get;list;create;update;watch;delete
+// +kubebuilder:rbac:groups=core,resources=secrets;pods;namespaces,verbs=get;list;watch
 // +kubebuilder:rbac:groups=core,resources=pods/exec,verbs=create
+// +kubebuilder:rbac:groups=authentication.k8s.io,resources=tokenreviews,verbs=create
 // +kubebuilder:rbac:groups=route.openshift.io,resources=routes,verbs=get;list;create;update;watch;delete
 // +kubebuilder:rbac:groups=batch,resources=cronjobs,verbs=get;list;watch;create;update;patch;delete
 

sdk/python/feast/permissions/auth/kubernetes_token_parser.py

@@ -32,46 +32,71 @@ def __init__(self):
 
     async def user_details_from_access_token(self, access_token: str) -> User:
         """
-        Extract the service account from the token and search the roles associated with it.
-        Also extract groups and namespaces using Token Access Review.
+        Extract user details from the token using Token Access Review.
+        Handles both service account tokens (JWTs) and user tokens (opaque tokens).
 
         Returns:
-            User: Current user, with associated roles, groups, and namespaces. The `username` is the `:` separated concatenation of `namespace` and `service account name`.
+            User: Current user, with associated roles, groups, and namespaces.
 
         Raises:
             AuthenticationError if any error happens.
         """
-        sa_namespace, sa_name = _decode_token(access_token)
-        current_user = f"{sa_namespace}:{sa_name}"
-        logger.info(
-            f"Request received from ServiceAccount: {sa_name} in namespace: {sa_namespace}"
+        # First, try to extract user information using Token Access Review
+        groups, namespaces = self._extract_groups_and_namespaces_from_token(
+            access_token
         )
 
-        intra_communication_base64 = os.getenv("INTRA_COMMUNICATION_BASE64")
-        if sa_name is not None and sa_name == intra_communication_base64:
-            return User(username=sa_name, roles=[], groups=[], namespaces=[])
-        else:
-            current_namespace = self._read_namespace_from_file()
+        # Try to determine if this is a service account or regular user
+        try:
+            # Attempt to decode as JWT (for service accounts)
+            sa_namespace, sa_name = _decode_token(access_token)
+            current_user = f"{sa_namespace}:{sa_name}"
             logger.info(
-                f"Looking for ServiceAccount roles of {sa_namespace}:{sa_name} in {current_namespace}"
+                f"Request received from ServiceAccount: {sa_name} in namespace: {sa_namespace}"
             )
 
-            # Get roles using existing method
-            roles = self.get_roles(
-                current_namespace=current_namespace,
-                service_account_namespace=sa_namespace,
-                service_account_name=sa_name,
-            )
-            logger.info(f"Roles: {roles}")
+            intra_communication_base64 = os.getenv("INTRA_COMMUNICATION_BASE64")
+            if sa_name is not None and sa_name == intra_communication_base64:
+                return User(username=sa_name, roles=[], groups=[], namespaces=[])
+            else:
+                current_namespace = self._read_namespace_from_file()
+                logger.info(
+                    f"Looking for ServiceAccount roles of {sa_namespace}:{sa_name} in {current_namespace}"
+                )
 
-            # Extract groups and namespaces using Token Access Review
-            groups, namespaces = self._extract_groups_and_namespaces_from_token(
-                access_token
-            )
-            logger.info(f"Groups: {groups}, Namespaces: {namespaces}")
+                # Get roles using existing method
+                roles = self.get_roles(
+                    current_namespace=current_namespace,
+                    service_account_namespace=sa_namespace,
+                    service_account_name=sa_name,
+                )
+                logger.info(f"Roles: {roles}")
+
+                return User(
+                    username=current_user,
+                    roles=roles,
+                    groups=groups,
+                    namespaces=namespaces,
+                )
+
+        except AuthenticationError as e:
+            # If JWT decoding fails, this is likely a user token
+            # Use Token Access Review to get user information
+            logger.info(f"Token is not a JWT (likely a user token): {e}")
+
+            # Get username from Token Access Review
+            username = self._get_username_from_token_review(access_token)
+            if not username:
+                raise AuthenticationError("Could not extract username from token")
+
+            logger.info(f"Request received from User: {username}")
+
+            # For user tokens, we don't have traditional roles, but we have groups and namespaces
+            # You might want to map groups to roles or use a different role assignment strategy
+            roles = []  # Users don't have traditional service account roles
 
             return User(
-                username=current_user, roles=roles, groups=groups, namespaces=namespaces
+                username=username, roles=roles, groups=groups, namespaces=namespaces
             )
 
     def _read_namespace_from_file(self):
@@ -294,6 +319,35 @@ def _extract_user_data_science_projects(self, username: str) -> list[str]:
 
         return user_namespaces
 
+    def _get_username_from_token_review(self, access_token: str) -> str:
+        """
+        Extract username from Token Access Review.
+
+        Args:
+            access_token: The access token to review
+
+        Returns:
+            str: The username from the token review, or empty string if not found
+        """
+        try:
+            token_review = client.V1TokenReview(
+                spec=client.V1TokenReviewSpec(token=access_token)
+            )
+
+            response = self.auth_v1.create_token_review(token_review)
+
+            if response.status.authenticated and response.status.user:
+                username = getattr(response.status.user, "username", "") or ""
+                logger.debug(f"Extracted username from Token Access Review: {username}")
+                return username
+            else:
+                logger.warning(f"Token Access Review failed: {response.status.error}")
+                return ""
+
+        except Exception as e:
+            logger.error(f"Failed to get username from Token Access Review: {e}")
+            return ""
+
     def _cluster_role_binding_grants_namespace_access(
         self, cluster_role_binding, namespace: str
     ) -> bool:

sdk/python/feast/permissions/auth_model.py

@@ -70,4 +70,4 @@ class KubernetesAuthConfig(AuthConfig):
     # Optional user token for users (not service accounts)
     user_token: Optional[str] = None
 
-    model_config = ConfigDict(extra="allow")
+    model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow")