Role binding for token access review and groups identifications from namespace

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

Author: jyejare

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

@@ -105,6 +105,31 @@ func (authz *FeastAuthorization) setFeastRole(role *rbacv1.Role) error {
 			Resources: []string{"roles", "rolebindings"},
 			Verbs:     []string{"get", "list", "watch"},
 		},
+		{
+			APIGroups: []string{"authentication.k8s.io"},
+			Resources: []string{"tokenreviews"},
+			Verbs:     []string{"create"},
+		},
+		{
+			APIGroups: []string{rbacv1.GroupName},
+			Resources: []string{"subjectaccessreviews"},
+			Verbs:     []string{"create"},
+		},
+		{
+			APIGroups: []string{""},
+			Resources: []string{"namespaces"},
+			Verbs:     []string{"get", "list", "watch"},
+		},
+		{
+			APIGroups: []string{rbacv1.GroupName},
+			Resources: []string{"clusterroles"},
+			Verbs:     []string{"get", "list"},
+		},
+		{
+			APIGroups: []string{rbacv1.GroupName},
+			Resources: []string{"clusterrolebindings"},
+			Verbs:     []string{"get", "list"},
+		},
 	}
 
 	return controllerutil.SetControllerReference(authz.Handler.FeatureStore, role, authz.Handler.Scheme)

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

@@ -128,28 +128,41 @@ def _extract_groups_and_namespaces_from_token(
             token_review = client.V1TokenReview(
                 spec=client.V1TokenReviewSpec(token=access_token)
             )
-            groups = []
-            namespaces = []
+            groups: list[str] = []
+            namespaces: list[str] = []
 
             # Call Token Access Review API
             response = self.auth_v1.create_token_review(token_review)
 
             if response.status.authenticated:
                 # Extract groups and namespaces from the response
-                groups = response.status.groups
+                # Groups are in response.status.user.groups, not response.status.groups
+                if response.status.user and hasattr(response.status.user, "groups"):
+                    groups = response.status.user.groups or []
+                else:
+                    groups = []
 
                 # Extract namespaces from the user info
                 if response.status.user:
                     # For service accounts, the namespace is typically in the username
                     # For regular users, we might need to extract from groups or other fields
-                    username = response.status.user.get("username", "")
+                    username = getattr(response.status.user, "username", "") or ""
                     if ":" in username and username.startswith(
                         "system:serviceaccount:"
                     ):
                         # Extract namespace from service account username
                         parts = username.split(":")
                         if len(parts) >= 4:
-                            namespaces.append(parts[2])  # namespace is the 3rd part
+                            service_account_namespace = parts[
+                                2
+                            ]  # namespace is the 3rd part
+                            namespaces.append(service_account_namespace)
+
+                            # For service accounts, also extract groups that have access to this namespace
+                            namespace_groups = self._extract_namespace_access_groups(
+                                service_account_namespace
+                            )
+                            groups.extend(namespace_groups)
 
                     # Also check if there are namespace-specific groups
                     for group in groups:
@@ -170,6 +183,98 @@ def _extract_groups_and_namespaces_from_token(
             # We dont need to extract groups and namespaces from jwt decoding, not ideal for kubernetes auth
         return groups, namespaces
 
+    def _extract_namespace_access_groups(self, namespace: str) -> list[str]:
+        """
+        Extract groups that have access to a specific namespace by querying RoleBindings and ClusterRoleBindings.
+
+        Args:
+            namespace: The namespace to check for group access
+
+        Returns:
+            list[str]: List of groups that have access to the namespace
+        """
+        groups = []
+        try:
+            # Get RoleBindings in the namespace
+            role_bindings = self.rbac_v1.list_namespaced_role_binding(
+                namespace=namespace
+            )
+            for rb in role_bindings.items:
+                for subject in rb.subjects or []:
+                    if subject.kind == "Group":
+                        groups.append(subject.name)
+                        logger.debug(
+                            f"Found group {subject.name} in RoleBinding {rb.metadata.name}"
+                        )
+
+            # Get ClusterRoleBindings that might grant access to this namespace
+            cluster_role_bindings = self.rbac_v1.list_cluster_role_binding()
+            for crb in cluster_role_bindings.items:
+                # Check if this ClusterRoleBinding grants access to the namespace
+                if self._cluster_role_binding_grants_namespace_access(crb, namespace):
+                    for subject in crb.subjects or []:
+                        if subject.kind == "Group":
+                            groups.append(subject.name)
+                            logger.debug(
+                                f"Found group {subject.name} in ClusterRoleBinding {crb.metadata.name}"
+                            )
+
+            # Remove duplicates and sort
+            groups = sorted(list(set(groups)))
+            logger.info(
+                f"Found {len(groups)} groups with access to namespace {namespace}: {groups}"
+            )
+
+        except Exception as e:
+            logger.error(
+                f"Failed to extract namespace access groups for {namespace}: {e}"
+            )
+
+        return groups
+
+    def _cluster_role_binding_grants_namespace_access(
+        self, cluster_role_binding, namespace: str
+    ) -> bool:
+        """
+        Check if a ClusterRoleBinding grants access to a specific namespace.
+        This is a simplified check - in practice, you might need more sophisticated logic.
+
+        Args:
+            cluster_role_binding: The ClusterRoleBinding to check
+            namespace: The namespace to check access for
+
+        Returns:
+            bool: True if the ClusterRoleBinding likely grants access to the namespace
+        """
+        try:
+            # Get the ClusterRole referenced by this binding
+            cluster_role_name = cluster_role_binding.role_ref.name
+            cluster_role = self.rbac_v1.read_cluster_role(name=cluster_role_name)
+
+            # Check if the ClusterRole has rules that could grant access to the namespace
+            for rule in cluster_role.rules or []:
+                # Check if the rule applies to namespaces or has wildcard access
+                if (
+                    rule.resources
+                    and ("namespaces" in rule.resources or "*" in rule.resources)
+                    and rule.verbs
+                    and (
+                        "get" in rule.verbs or "list" in rule.verbs or "*" in rule.verbs
+                    )
+                ):
+                    return True
+
+                # Check if the rule has resourceNames that include our namespace
+                if rule.resource_names and namespace in rule.resource_names:
+                    return True
+
+        except Exception as e:
+            logger.debug(
+                f"Error checking ClusterRoleBinding {cluster_role_binding.metadata.name}: {e}"
+            )
+
+        return False
+
 
 def _decode_token(access_token: str) -> tuple[str, str]:
     """