fix: Require signed tokens and a private secret for intra-server communication

docs/getting-started/components/authz_manager.md

@@ -214,6 +214,14 @@ The client supports multiple token source modes. The SDK resolves tokens in the
 5. **`FEAST_OIDC_TOKEN`** — default fallback environment variable
 6. **Kubernetes service account token** — read from `/var/run/secrets/kubernetes.io/serviceaccount/token` when running inside a pod
 
+The intra-communication token is not a user credential. It is a shared secret that
+Feast servers sign their calls to each other with, and a caller able to produce it is
+trusted by the RBAC layer without any further permission check. Give it a value that is
+private to the installation and carries at least 32 bytes of entropy, and point every
+Feast release that has to reach another one at that same value (`intraCommunicationSecret`
+in the Helm chart). Leaving it unset disables intra-server communication rather than
+falling back to a default.
+
 **Token passthrough** (for use with external token providers like [kube-authkit](https://github.com/opendatahub-io/kube-authkit)):
 ```yaml
 project: my-project

docs/reference/auth/user_token_provisioning.md

@@ -258,6 +258,14 @@ The system checks for tokens in this order:
 3. **Service account token**: `/var/run/secrets/kubernetes.io/serviceaccount/token` (for pods)
 4. **Environment variable**: `LOCAL_K8S_TOKEN`
 
+The intra-communication token is not a user credential. It is a shared secret that
+Feast servers sign their calls to each other with, and a caller able to produce it is
+trusted by the RBAC layer without any further permission check. Give it a value that is
+private to the installation and carries at least 32 bytes of entropy, and point every
+Feast release that has to reach another one at that same value (`intraCommunicationSecret`
+in the Helm chart). Leaving it unset disables intra-server communication rather than
+falling back to a default.
+
 
 ## Troubleshooting
 

infra/charts/feast-feature-server/templates/deployment.yaml

@@ -43,8 +43,13 @@ spec:
           env:
             - name: FEATURE_STORE_YAML_BASE64
               value: {{ .Values.feature_store_yaml_base64 }}
+            {{- if .Values.intraCommunicationSecret.name }}
             - name: INTRA_COMMUNICATION_BASE64
-              value: {{ "intra-server-communication" | b64enc }}
+              valueFrom:
+                secretKeyRef:
+                  name: {{ .Values.intraCommunicationSecret.name }}
+                  key: {{ .Values.intraCommunicationSecret.key }}
+            {{- end }}
             {{- with .Values.extraEnvs}}
               {{- toYaml . | nindent 12 }}
             {{- end}}

infra/charts/feast-feature-server/values.yaml

@@ -29,6 +29,16 @@ feature_store_yaml_base64: ""
 # feast_mode -- Feast supported deployment modes - online (default), offline, ui and registry
 feast_mode: "online"
 
+# intraCommunicationSecret -- Kubernetes Secret holding the shared secret that Feast
+# servers present to each other instead of an end-user credential. Every Feast release
+# that has to call another one (a feature server and the registry server it reads from,
+# for example) must point at the same Secret, and the value has to stay private to the
+# installation: a caller able to produce it is trusted by the RBAC layer without any
+# further check. Intra-server communication stays disabled while `name` is empty.
+intraCommunicationSecret:
+  name: ""
+  key: "INTRA_COMMUNICATION_BASE64"
+
 # commandArgs -- Override the default command arguments for complete control over CLI options
 # If not specified, falls back to legacy behavior based on feast_mode
 # Example for UI mode with custom options:

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

@@ -0,0 +1,66 @@
+import logging
+import os
+from typing import Any, Optional
+
+import jwt
+
+logger = logging.getLogger(__name__)
+
+INTRA_COMMUNICATION_ENV_VAR = "INTRA_COMMUNICATION_BASE64"
+INTRA_COMMUNICATION_ALGORITHM = "HS256"
+
+
+def get_intra_comm_secret() -> Optional[str]:
+    """
+    Return the shared secret used to authenticate Feast intra-server communication.
+
+    Returns:
+        Optional[str]: the configured secret, or `None` when intra-server communication
+            is not configured, which covers both an unset and an empty environment
+            variable. Callers must treat `None` as "intra-server communication is
+            disabled" and never fall back to a default, since a well-known value would
+            let any caller assume the internal identity.
+    """
+    return os.getenv(INTRA_COMMUNICATION_ENV_VAR) or None
+
+
+def encode_intra_comm_token(claims: dict[str, Any], secret: str) -> str:
+    """
+    Sign an intra-server communication token with the shared secret.
+
+    Args:
+        claims: the claims identifying the internal caller.
+        secret: the shared secret, used as the signing key.
+
+    Returns:
+        str: the signed token.
+    """
+    return jwt.encode(claims, secret, algorithm=INTRA_COMMUNICATION_ALGORITHM)
+
+
+def decode_intra_comm_token(access_token: str, secret: str) -> Optional[dict[str, Any]]:
+    """
+    Return the claims of an intra-server communication token whose signature verifies.
+
+    A token that is not signed with the shared secret is not an intra-server
+    communication token. Ordinary user tokens issued by the identity provider reach
+    this function too and must fall through to the regular authentication path, so an
+    unverifiable token is reported as `None` rather than raised.
+
+    Args:
+        access_token: the raw bearer token.
+        secret: the shared secret, used as the verification key.
+
+    Returns:
+        Optional[dict[str, Any]]: the verified claims, or `None` when the token is not a
+            valid intra-server communication token.
+    """
+    try:
+        return jwt.decode(
+            access_token,
+            secret,
+            algorithms=[INTRA_COMMUNICATION_ALGORITHM],
+            options={"verify_aud": False},
+        )
+    except jwt.InvalidTokenError:
+        return None

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

@@ -1,12 +1,16 @@
 import logging
-import os
+from typing import Optional
 
 import jwt
 from kubernetes import client, config
 from starlette.authentication import (
     AuthenticationError,
 )
 
+from feast.permissions.auth.intra_comm import (
+    decode_intra_comm_token,
+    get_intra_comm_secret,
+)
 from feast.permissions.auth.token_parser import TokenParser
 from feast.permissions.user import User
 
@@ -55,9 +59,9 @@ async def user_details_from_access_token(self, access_token: str) -> User:
                 f"Request received from ServiceAccount: {sa_name} in namespace: {sa_namespace}"
             )
 
-            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=[])
+            intra_comm_user = _get_intra_comm_user(access_token)
+            if intra_comm_user is not None:
+                return intra_comm_user
             else:
                 current_namespace = self._read_namespace_from_file()
                 logger.info(
@@ -453,6 +457,39 @@ def _cluster_role_binding_grants_namespace_access(
         return False
 
 
+def _get_intra_comm_user(access_token: str) -> Optional[User]:
+    """
+    Return the intra-server communication user for a token that proves knowledge of
+    the shared secret.
+
+    The signature is verified against the shared secret before the subject is read,
+    so a caller that does not hold the secret cannot assume the internal identity.
+
+    Returns:
+        Optional[User]: the internal user, or `None` when the token is not an
+            intra-server communication token.
+    """
+    intra_communication_base64 = get_intra_comm_secret()
+    if not intra_communication_base64:
+        return None
+
+    claims = decode_intra_comm_token(access_token, intra_communication_base64)
+    if claims is None:
+        return None
+
+    subject = claims.get("sub")
+    if not isinstance(subject, str):
+        return None
+
+    parts = subject.split(":")
+    if len(parts) == 4 and parts[3] == intra_communication_base64:
+        return User(
+            username=intra_communication_base64, roles=[], groups=[], namespaces=[]
+        )
+
+    return None
+
+
 def _decode_token(access_token: str) -> tuple[str, str]:
     """
     The `sub` portion of the decoded token includes the service account name in the format: `system:serviceaccount:NAMESPACE:SA_NAME`

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

@@ -12,6 +12,10 @@
     AuthenticationError,
 )
 
+from feast.permissions.auth.intra_comm import (
+    decode_intra_comm_token,
+    get_intra_comm_secret,
+)
 from feast.permissions.auth.token_parser import TokenParser
 from feast.permissions.auth_model import OidcAuthConfig
 from feast.permissions.oidc_service import OIDCDiscoveryService
@@ -194,8 +198,9 @@ async def user_details_from_access_token(self, access_token: str) -> User:
         Validate the access token then decode it to extract the user credentials,
         roles, and groups.
 
-        A single unverified decode is performed upfront for lightweight routing:
-        intra-server communication, Kubernetes SA tokens (identified by the
+        Intra-server communication tokens are verified against the shared secret
+        before any claim on them is trusted. Every other token is routed with a
+        single unverified decode: Kubernetes SA tokens (identified by the
         ``kubernetes.io`` claim), or standard OIDC/Keycloak JWKS validation.
 
         Returns:
@@ -209,7 +214,7 @@ async def user_details_from_access_token(self, access_token: str) -> User:
         except jwt.exceptions.DecodeError as e:
             raise AuthenticationError(f"Failed to decode token: {e}")
 
-        user = self._get_intra_comm_user(unverified)
+        user = self._get_intra_comm_user(access_token)
         if user:
             return user
 
@@ -310,16 +315,27 @@ async def _validate_k8s_sa_token_and_extract_namespace(
         return User(username=username, roles=[], groups=[], namespaces=namespaces)
 
     @staticmethod
-    def _get_intra_comm_user(decoded_token: dict) -> Optional[User]:
-        intra_communication_base64 = os.getenv("INTRA_COMMUNICATION_BASE64")
-
-        if intra_communication_base64:
-            if "preferred_username" in decoded_token:
-                preferred_username: str = decoded_token["preferred_username"]
-                if (
-                    preferred_username is not None
-                    and preferred_username == intra_communication_base64
-                ):
-                    return User(username=preferred_username, roles=[])
+    def _get_intra_comm_user(access_token: str) -> Optional[User]:
+        """
+        Return the intra-server communication user for a token that proves knowledge
+        of the shared secret.
+
+        The signature is verified against the shared secret before the claim is read,
+        so a caller that does not hold the secret cannot assume the internal identity.
+
+        Returns:
+            Optional[User]: the internal user, or `None` when the token is not an
+                intra-server communication token.
+        """
+        intra_communication_base64 = get_intra_comm_secret()
+        if not intra_communication_base64:
+            return None
+
+        claims = decode_intra_comm_token(access_token, intra_communication_base64)
+        if claims is None:
+            return None
+
+        if claims.get("preferred_username") == intra_communication_base64:
+            return User(username=intra_communication_base64, roles=[])
 
         return None

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

@@ -1,8 +1,7 @@
 import logging
 
-import jwt
-
 from feast.permissions.auth.auth_type import AuthType
+from feast.permissions.auth.intra_comm import encode_intra_comm_token
 from feast.permissions.auth_model import AuthConfig
 from feast.permissions.client.auth_client_manager import AuthenticationClientManager
 
@@ -29,4 +28,4 @@ def get_token(self):
                 f"No Auth client manager implemented for the auth type:{self.auth_config.type}"
             )
 
-        return jwt.encode(payload, "", algorithm="none")
+        return encode_intra_comm_token(payload, self.intra_communication_base64)

sdk/python/feast/permissions/security_manager.py

@@ -1,12 +1,12 @@
 import logging
-import os
 from contextvars import ContextVar
 from typing import Callable, List, Optional, Union
 
 from feast.errors import FeastObjectNotFoundException
 from feast.feast_object import FeastObject
 from feast.infra.registry.base_registry import BaseRegistry
 from feast.permissions.action import AuthzedAction
+from feast.permissions.auth.intra_comm import get_intra_comm_secret
 from feast.permissions.enforcer import enforce_policy
 from feast.permissions.permission import Permission
 from feast.permissions.user import User
@@ -246,7 +246,7 @@ def no_security_manager():
 
 
 def is_auth_necessary(sm: Optional[SecurityManager]) -> bool:
-    intra_communication_base64 = os.getenv("INTRA_COMMUNICATION_BASE64")
+    intra_communication_base64 = get_intra_comm_secret()
 
     # If no security manager, no auth is necessary
     if sm is None:
@@ -256,8 +256,13 @@ def is_auth_necessary(sm: Optional[SecurityManager]) -> bool:
     if sm.current_user is None:
         return True
 
-    # If user is intra-communication, no auth is necessary
-    if sm.current_user.username == intra_communication_base64:
+    # If user is intra-communication, no auth is necessary. The secret must be
+    # configured: without it there is no internal identity to recognize, and a
+    # comparison against an unset value would let a blank username skip every check.
+    if (
+        intra_communication_base64
+        and sm.current_user.username == intra_communication_base64
+    ):
         return False
 
     # Otherwise, auth is necessary