Groups and Namespaces authorization

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

Author: jyejare

examples/operator-rbac-openshift-tls/permissions_with_groups_namespaces.py

@@ -0,0 +1,110 @@
+# Example permissions configuration with groups and namespaces support
+# This demonstrates how to use the new group-based and namespace-based policies
+# in addition to the existing role-based policies
+
+from feast.feast_object import ALL_RESOURCE_TYPES
+from feast.permissions.action import READ, AuthzedAction, ALL_ACTIONS
+from feast.permissions.permission import Permission
+from feast.permissions.policy import RoleBasedPolicy, GroupBasedPolicy, NamespaceBasedPolicy, CombinedGroupNamespacePolicy
+
+# Define K8s roles (existing functionality)
+admin_roles = ["feast-writer"]  # Full access (can create, update, delete) Feast Resources
+user_roles = ["feast-reader"]   # Read-only access on Feast Resources
+
+# Define groups for different teams
+data_team_groups = ["data-team", "ml-engineers"]
+dev_team_groups = ["dev-team", "developers"]
+admin_groups = ["feast-admins", "platform-admins"]
+
+# Define namespaces for different environments
+prod_namespaces = ["production", "prod"]
+staging_namespaces = ["staging", "dev"]
+test_namespaces = ["test", "testing"]
+
+# Role-based permissions (existing functionality)
+# - Grants read and describing Feast objects access
+user_perm = Permission(
+    name="feast_user_permission",
+    types=ALL_RESOURCE_TYPES,
+    policy=RoleBasedPolicy(roles=user_roles),
+    actions=[AuthzedAction.DESCRIBE] + READ  # Read access (READ_ONLINE, READ_OFFLINE) + describe other Feast Resources.
+)
+
+# Admin permissions (existing functionality)
+# - Grants full control over all resources
+admin_perm = Permission(
+    name="feast_admin_permission",
+    types=ALL_RESOURCE_TYPES,
+    policy=RoleBasedPolicy(roles=admin_roles),
+    actions=ALL_ACTIONS  # Full permissions: CREATE, UPDATE, DELETE, READ, WRITE
+)
+
+# Group-based permissions (new functionality)
+# - Grants read access to data team members
+data_team_perm = Permission(
+    name="data_team_read_permission",
+    types=ALL_RESOURCE_TYPES,
+    policy=GroupBasedPolicy(groups=data_team_groups),
+    actions=[AuthzedAction.DESCRIBE] + READ
+)
+
+# - Grants full access to admin groups
+admin_group_perm = Permission(
+    name="admin_group_permission",
+    types=ALL_RESOURCE_TYPES,
+    policy=GroupBasedPolicy(groups=admin_groups),
+    actions=ALL_ACTIONS
+)
+
+# Namespace-based permissions (new functionality)
+# - Grants read access to production namespace users
+prod_read_perm = Permission(
+    name="production_read_permission",
+    types=ALL_RESOURCE_TYPES,
+    policy=NamespaceBasedPolicy(namespaces=prod_namespaces),
+    actions=[AuthzedAction.DESCRIBE] + READ
+)
+
+# - Grants full access to staging namespace users
+staging_full_perm = Permission(
+    name="staging_full_permission",
+    types=ALL_RESOURCE_TYPES,
+    policy=NamespaceBasedPolicy(namespaces=staging_namespaces),
+    actions=ALL_ACTIONS
+)
+
+# Combined permissions (using combined policy type)
+# - Grants read access to dev team members in test namespaces
+dev_test_perm = Permission(
+    name="dev_test_permission",
+    types=ALL_RESOURCE_TYPES,
+    policy=CombinedGroupNamespacePolicy(groups=dev_team_groups, namespaces=test_namespaces),
+    actions=[AuthzedAction.DESCRIBE] + READ
+)
+
+# - Grants full access to data team members in staging namespaces
+data_staging_perm = Permission(
+    name="data_staging_permission",
+    types=ALL_RESOURCE_TYPES,
+    policy=CombinedGroupNamespacePolicy(groups=data_team_groups, namespaces=staging_namespaces),
+    actions=ALL_ACTIONS
+)
+
+# # Export all permissions
+# permissions = [
+#     # Role-based permissions (existing)
+#     user_perm,
+#     admin_perm,
+    
+#     # Group-based permissions (new)
+#     data_team_perm,
+#     admin_group_perm,
+    
+#     # Namespace-based permissions (new)
+#     prod_read_perm,
+#     staging_full_perm,
+    
+#     # Combined permissions
+#     dev_test_perm,
+#     data_staging_perm,
+# ]

protos/feast/core/Policy.proto

@@ -14,10 +14,22 @@ message Policy {
 
   oneof policy_type {
     RoleBasedPolicy role_based_policy = 3;
+    GroupBasedPolicy group_based_policy = 4;
+    NamespaceBasedPolicy namespace_based_policy = 5;
   }
 }
 
 message RoleBasedPolicy {
   // List of roles in this policy.
   repeated string roles = 1;
 }
+
+message GroupBasedPolicy {
+  // List of groups in this policy.
+  repeated string groups = 1;
+}
+
+message NamespaceBasedPolicy {
+  // List of namespaces in this policy.
+  repeated string namespaces = 1;
+}

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

@@ -28,13 +28,15 @@ def __init__(self):
         config.load_incluster_config()
         self.v1 = client.CoreV1Api()
         self.rbac_v1 = client.RbacAuthorizationV1Api()
+        self.auth_v1 = client.AuthenticationV1Api()
 
     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.
 
         Returns:
-            User: Current user, with associated roles. The `username` is the `:` separated concatenation of `namespace` and `service account name`.
+            User: Current user, with associated roles, groups, and namespaces. The `username` is the `:` separated concatenation of `namespace` and `service account name`.
 
         Raises:
             AuthenticationError if any error happens.
@@ -47,20 +49,30 @@ async def user_details_from_access_token(self, access_token: str) -> User:
 
         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=[])
+            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}"
             )
+
+            # 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)
+            # 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}")
+
+            return User(
+                username=current_user, roles=roles, groups=groups, namespaces=namespaces
+            )
 
     def _read_namespace_from_file(self):
         try:
@@ -99,6 +111,65 @@ def get_roles(
 
         return list(roles)
 
+    def _extract_groups_and_namespaces_from_token(
+        self, access_token: str
+    ) -> tuple[list[str], list[str]]:
+        """
+        Extract groups and namespaces from the token using Kubernetes Token Access Review.
+
+        Args:
+            access_token: The JWT token to analyze
+
+        Returns:
+            tuple[list[str], list[str]]: A tuple containing (groups, namespaces)
+        """
+        try:
+            # Create TokenReview object
+            token_review = client.V1TokenReview(
+                spec=client.V1TokenReviewSpec(token=access_token)
+            )
+            groups = []
+            namespaces = []
+
+            # 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
+
+                # 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", "")
+                    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
+
+                    # Also check if there are namespace-specific groups
+                    for group in groups:
+                        if group.startswith("system:serviceaccounts:"):
+                            # Extract namespace from service account group
+                            parts = group.split(":")
+                            if len(parts) >= 3:
+                                namespaces.append(parts[2])
+
+                logger.debug(
+                    f"Token Access Review successful. Groups: {groups}, Namespaces: {namespaces}"
+                )
+            else:
+                logger.warning(f"Token Access Review failed: {response.status.error}")
+
+        except Exception as e:
+            logger.error(f"Failed to perform Token Access Review: {e}")
+            # We dont need to extract groups and namespaces from jwt decoding, not ideal for kubernetes auth
+        return groups, namespaces
+
 
 def _decode_token(access_token: str) -> tuple[str, str]:
     """

sdk/python/feast/permissions/auth_model.py

@@ -67,4 +67,5 @@ class NoAuthConfig(AuthConfig):
 
 
 class KubernetesAuthConfig(AuthConfig):
-    pass
+    # Optional user token for users (not service accounts)
+    user_token: Optional[str] = None

sdk/python/feast/permissions/client/kubernetes_auth_client_manager.py

@@ -24,6 +24,11 @@ def get_token(self):
 
             return jwt.encode(payload, "")
 
+        # Check if user token is provided in config (for external users)
+        if hasattr(self.auth_config, "user_token") and self.auth_config.user_token:
+            logger.info("Using user token from configuration")
+            return self.auth_config.user_token
+
         try:
             token = self._read_token_from_file()
             return token