From 7a9866057375fdd57ca56c87c6e7b91ac11469cc Mon Sep 17 00:00:00 2001 From: Chen Yufan Date: Mon, 31 Aug 2026 23:45:20 +0800 Subject: [PATCH 1/2] fix: Require signed tokens and a private secret for intra-server communication The RBAC layer granted a fully trusted intra-communication identity to any request whose `preferred_username` (OIDC) or service-account name (Kubernetes) claim matched `INTRA_COMMUNICATION_BASE64`. That claim was read from an unverified JWT decode performed before any signature check, and the value it was compared against is hardcoded in Feast's own Helm chart to the constant `base64("intra-server-communication")`, so any caller able to reach a Feast server could mint an unsigned token and skip every permission check on every project that server serves. Sign the intra-server token with HS256 using the shared secret and verify that signature before reading any claim, so a caller without the secret cannot mint an acceptable token. Take the secret from a Kubernetes Secret named by the new `intraCommunicationSecret` chart value and omit the variable when it is not configured, since signing does not help while the key is a constant published in this repository. Treat an unset or empty secret as "disabled" rather than a value to compare against. This is a breaking change: deployments that relied on the chart default must create a Secret and point every Feast release that talks to another one at it. Fixes #6785 Signed-off-by: Chen Yufan --- .../components/authz_manager.md | 8 ++ .../reference/auth/user_token_provisioning.md | 8 ++ .../templates/deployment.yaml | 7 +- infra/charts/feast-feature-server/values.yaml | 10 ++ .../feast/permissions/auth/intra_comm.py | 66 ++++++++++ .../auth/kubernetes_token_parser.py | 45 ++++++- .../permissions/auth/oidc_token_parser.py | 44 ++++--- ...ntra_comm_authentication_client_manager.py | 5 +- .../feast/permissions/security_manager.py | 13 +- .../permissions/auth/test_token_parser.py | 116 ++++++++++++++++-- 10 files changed, 289 insertions(+), 33 deletions(-) create mode 100644 sdk/python/feast/permissions/auth/intra_comm.py diff --git a/docs/getting-started/components/authz_manager.md b/docs/getting-started/components/authz_manager.md index f9e1d8af104..8047d236b41 100644 --- a/docs/getting-started/components/authz_manager.md +++ b/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 diff --git a/docs/reference/auth/user_token_provisioning.md b/docs/reference/auth/user_token_provisioning.md index 05c772891b3..b6630aa5156 100644 --- a/docs/reference/auth/user_token_provisioning.md +++ b/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 diff --git a/infra/charts/feast-feature-server/templates/deployment.yaml b/infra/charts/feast-feature-server/templates/deployment.yaml index ef0cc9671de..262979f0430 100644 --- a/infra/charts/feast-feature-server/templates/deployment.yaml +++ b/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}} diff --git a/infra/charts/feast-feature-server/values.yaml b/infra/charts/feast-feature-server/values.yaml index 3a098796106..3d74699e9dd 100644 --- a/infra/charts/feast-feature-server/values.yaml +++ b/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: diff --git a/sdk/python/feast/permissions/auth/intra_comm.py b/sdk/python/feast/permissions/auth/intra_comm.py new file mode 100644 index 00000000000..da4a78f4715 --- /dev/null +++ b/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 diff --git a/sdk/python/feast/permissions/auth/kubernetes_token_parser.py b/sdk/python/feast/permissions/auth/kubernetes_token_parser.py index c126a90cfcc..df1e0dac06d 100644 --- a/sdk/python/feast/permissions/auth/kubernetes_token_parser.py +++ b/sdk/python/feast/permissions/auth/kubernetes_token_parser.py @@ -1,5 +1,5 @@ import logging -import os +from typing import Optional import jwt from kubernetes import client, config @@ -7,6 +7,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.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` diff --git a/sdk/python/feast/permissions/auth/oidc_token_parser.py b/sdk/python/feast/permissions/auth/oidc_token_parser.py index 91669e66647..9d6b7311c8b 100644 --- a/sdk/python/feast/permissions/auth/oidc_token_parser.py +++ b/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 diff --git a/sdk/python/feast/permissions/client/intra_comm_authentication_client_manager.py b/sdk/python/feast/permissions/client/intra_comm_authentication_client_manager.py index bdc6159a2f8..a2fce8b56b0 100644 --- a/sdk/python/feast/permissions/client/intra_comm_authentication_client_manager.py +++ b/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) diff --git a/sdk/python/feast/permissions/security_manager.py b/sdk/python/feast/permissions/security_manager.py index 6c02922c986..cb9f08e7749 100644 --- a/sdk/python/feast/permissions/security_manager.py +++ b/sdk/python/feast/permissions/security_manager.py @@ -1,5 +1,4 @@ import logging -import os from contextvars import ContextVar from typing import Callable, List, Optional, Union @@ -7,6 +6,7 @@ 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 diff --git a/sdk/python/tests/unit/permissions/auth/test_token_parser.py b/sdk/python/tests/unit/permissions/auth/test_token_parser.py index a5056393b60..ec1eacfdd6c 100644 --- a/sdk/python/tests/unit/permissions/auth/test_token_parser.py +++ b/sdk/python/tests/unit/permissions/auth/test_token_parser.py @@ -13,12 +13,22 @@ AuthenticationError, ) -from feast.permissions.auth.kubernetes_token_parser import KubernetesTokenParser +from feast.permissions.auth.intra_comm import ( + decode_intra_comm_token, + encode_intra_comm_token, +) +from feast.permissions.auth.kubernetes_token_parser import ( + KubernetesTokenParser, +) +from feast.permissions.auth.kubernetes_token_parser import ( + _get_intra_comm_user as _k8s_get_intra_comm_user, +) from feast.permissions.auth.oidc_token_parser import OidcTokenParser from feast.permissions.auth_model import OidcAuthConfig from feast.permissions.user import User _CLIENT_ID = "test" +_INTRA_COMM_SECRET = "test1234" # pragma: allowlist secret @patch( @@ -446,12 +456,18 @@ async def mock_oath2(self, request): }, ) - monkeypatch.setattr( - "feast.permissions.auth.oidc_token_parser.jwt.decode", - lambda self, *args, **kwargs: user_data, - ) + if is_intra_server: + # A real intra-server token, signed with the shared secret so the parser can + # verify it. A token that merely asserts the claim is covered by + # test_oidc_intra_comm_rejects_forged_token. + access_token = encode_intra_comm_token(user_data, _INTRA_COMM_SECRET) + else: + monkeypatch.setattr( + "feast.permissions.auth.oidc_token_parser.jwt.decode", + lambda self, *args, **kwargs: user_data, + ) + access_token = "aaa-bbb-ccc" - access_token = "aaa-bbb-ccc" token_parser = OidcTokenParser(auth_config=oidc_config) user = asyncio.run( token_parser.user_details_from_access_token(access_token=access_token) @@ -950,7 +966,13 @@ def test_k8s_inter_server_comm( roles = rolebindings["roles"] - access_token = "aaa-bbb-ccc" + if is_intra_server: + # A real intra-server token, signed with the shared secret so the parser can + # verify it before trusting the subject. + access_token = encode_intra_comm_token({"sub": subject}, _INTRA_COMM_SECRET) + else: + access_token = "aaa-bbb-ccc" + token_parser = KubernetesTokenParser() user = asyncio.run( token_parser.user_details_from_access_token(access_token=access_token) @@ -1047,3 +1069,83 @@ def test_oidc_parser_routes_keycloak_token_normally( assertpy.assert_that(user.username).is_equal_to("testuser") assertpy.assert_that(user.roles).is_equal_to(["reader"]) assertpy.assert_that(user.groups).is_equal_to(["data-team"]) + + +def _forged_intra_comm_tokens(claims: dict) -> list: + """ + Tokens that assert the intra-server identity without holding the shared secret. + + The unsigned variant is exactly what a caller can mint from public information, + which is what made the intra-server identity forgeable before it was signed. + """ + return [ + pytest.param(jwt.encode(claims, "", algorithm="none"), id="unsigned"), + pytest.param( + jwt.encode(claims, "not-the-shared-secret", algorithm="HS256"), + id="signed-with-another-secret", + ), + ] + + +@mock.patch.dict(os.environ, {"INTRA_COMMUNICATION_BASE64": _INTRA_COMM_SECRET}) +@pytest.mark.parametrize( + "forged_token", + _forged_intra_comm_tokens({"preferred_username": _INTRA_COMM_SECRET}), +) +def test_oidc_intra_comm_rejects_forged_token(forged_token, oidc_config): + """Claiming the intra-server username is not enough without the secret.""" + token_parser = OidcTokenParser(auth_config=oidc_config) + + assertpy.assert_that(token_parser._get_intra_comm_user(forged_token)).is_none() + + +@mock.patch.dict(os.environ, {"INTRA_COMMUNICATION_BASE64": _INTRA_COMM_SECRET}) +@pytest.mark.parametrize( + "forged_token", + _forged_intra_comm_tokens({"sub": f":::{_INTRA_COMM_SECRET}"}), +) +def test_k8s_intra_comm_rejects_forged_token(forged_token): + """Claiming the intra-server subject is not enough without the secret.""" + assertpy.assert_that(_k8s_get_intra_comm_user(forged_token)).is_none() + + +@pytest.mark.parametrize("secret_env", [None, ""], ids=["unset", "empty"]) +def test_intra_comm_disabled_without_secret(secret_env, oidc_config, monkeypatch): + """Without a configured secret there is no internal identity to hand out.""" + if secret_env is None: + monkeypatch.delenv("INTRA_COMMUNICATION_BASE64", raising=False) + else: + monkeypatch.setenv("INTRA_COMMUNICATION_BASE64", secret_env) + + token = encode_intra_comm_token( + {"preferred_username": _INTRA_COMM_SECRET}, _INTRA_COMM_SECRET + ) + token_parser = OidcTokenParser(auth_config=oidc_config) + + assertpy.assert_that(token_parser._get_intra_comm_user(token)).is_none() + + +def test_intra_comm_token_verifies_only_with_its_own_secret(): + """A token signed with one secret does not verify against another.""" + secret = "a-shared-secret-value" # pragma: allowlist secret + token = encode_intra_comm_token({"preferred_username": secret}, secret) + + assertpy.assert_that(decode_intra_comm_token(token, secret)).is_not_none() + assertpy.assert_that(decode_intra_comm_token(token, "another-secret")).is_none() + + +@mock.patch.dict(os.environ, {"INTRA_COMMUNICATION_BASE64": _INTRA_COMM_SECRET}) +def test_intra_comm_client_token_is_accepted_by_oidc_parser(oidc_config): + """The token minted by the intra-server client verifies in the parser.""" + from feast.permissions.client.intra_comm_authentication_client_manager import ( + IntraCommAuthClientManager, + ) + + client_manager = IntraCommAuthClientManager(oidc_config, _INTRA_COMM_SECRET) + token_parser = OidcTokenParser(auth_config=oidc_config) + + user = token_parser._get_intra_comm_user(client_manager.get_token()) + + assertpy.assert_that(user).is_not_none() + assertpy.assert_that(user.username).is_equal_to(_INTRA_COMM_SECRET) + assertpy.assert_that(user.roles).is_equal_to([]) From 5d929fa55fe3e31a9f9339a45dce525cae9f0e35 Mon Sep 17 00:00:00 2001 From: Chen Yufan Date: Tue, 1 Sep 2026 13:27:53 +0800 Subject: [PATCH 2/2] test: Cover the rejected subjects on the intra-server token path The Kubernetes parser reads the service-account name out of a token whose signature already verified, and rejects it when the subject is absent, is not a string, does not split into four segments, or names a different account. None of those branches had a test: the existing cases stop at the signature check, so the subject handling below it was reached only on the happy path. Add the rejected subjects, and the matching case for the OIDC parser's `preferred_username`, so a holder of the shared secret cannot reach the internal identity with a token whose subject does not name it. Signed-off-by: Chen Yufan --- .../permissions/auth/test_token_parser.py | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/sdk/python/tests/unit/permissions/auth/test_token_parser.py b/sdk/python/tests/unit/permissions/auth/test_token_parser.py index ec1eacfdd6c..d00f765a17b 100644 --- a/sdk/python/tests/unit/permissions/auth/test_token_parser.py +++ b/sdk/python/tests/unit/permissions/auth/test_token_parser.py @@ -1149,3 +1149,37 @@ def test_intra_comm_client_token_is_accepted_by_oidc_parser(oidc_config): assertpy.assert_that(user).is_not_none() assertpy.assert_that(user.username).is_equal_to(_INTRA_COMM_SECRET) assertpy.assert_that(user.roles).is_equal_to([]) + + +@mock.patch.dict(os.environ, {"INTRA_COMMUNICATION_BASE64": _INTRA_COMM_SECRET}) +@pytest.mark.parametrize( + "claims", + [ + pytest.param({}, id="no-sub"), + pytest.param({"sub": None}, id="null-sub"), + pytest.param({"sub": 42}, id="non-string-sub"), + pytest.param({"sub": f"::{_INTRA_COMM_SECRET}"}, id="too-few-segments"), + pytest.param({"sub": f":::{_INTRA_COMM_SECRET}:x"}, id="too-many-segments"), + pytest.param({"sub": ":::another-account"}, id="another-service-account"), + ], +) +def test_k8s_intra_comm_rejects_a_signed_token_with_an_unusable_subject(claims): + """ + Holding the shared secret is necessary but not sufficient. The subject still has to + carry the intra-server service-account name in the shape the client sends, so a + holder of the secret cannot reach the internal identity with a malformed token. + """ + token = encode_intra_comm_token(claims, _INTRA_COMM_SECRET) + + assertpy.assert_that(_k8s_get_intra_comm_user(token)).is_none() + + +@mock.patch.dict(os.environ, {"INTRA_COMMUNICATION_BASE64": _INTRA_COMM_SECRET}) +def test_oidc_intra_comm_rejects_a_signed_token_for_another_username(oidc_config): + """The same, for the OIDC parser's `preferred_username` claim.""" + token = encode_intra_comm_token( + {"preferred_username": "another-user"}, _INTRA_COMM_SECRET + ) + token_parser = OidcTokenParser(auth_config=oidc_config) + + assertpy.assert_that(token_parser._get_intra_comm_user(token)).is_none()