From 309b834e3e943dcab61886f7ca9c9222cb630f71 Mon Sep 17 00:00:00 2001 From: anders-albert Date: Fri, 30 Jan 2026 07:20:42 +0100 Subject: [PATCH] feat; first pass on device code --- cognite/neat/_client/init/credentials.py | 29 +++++++++++++++++++++++- cognite/neat/_client/init/env_vars.py | 10 +++++++- 2 files changed, 37 insertions(+), 2 deletions(-) diff --git a/cognite/neat/_client/init/credentials.py b/cognite/neat/_client/init/credentials.py index 44054c761..2cc823dfb 100644 --- a/cognite/neat/_client/init/credentials.py +++ b/cognite/neat/_client/init/credentials.py @@ -1,17 +1,28 @@ from collections.abc import Callable -from cognite.client.credentials import CredentialProvider, OAuthClientCredentials, OAuthInteractive, Token +from cognite.client.credentials import ( + CredentialProvider, + OAuthClientCredentials, + OAuthDeviceCode, + OAuthInteractive, + Token, +) from cognite.neat._utils.text import humanize_collection from .env_vars import ClientEnvironmentVariables, LoginFlow +# This is the Cognite app registration in Entra for device code +# to be used with Neat. +NEAT_CLIENT_ENTRA_ID = "edf28ead-c06c-47f3-8521-288eb2893fee" + def get_credentials(env_vars: ClientEnvironmentVariables) -> CredentialProvider: options: dict[LoginFlow, Callable[[ClientEnvironmentVariables], CredentialProvider]] = { "client_credentials": create_client_credentials, "interactive": create_interactive_credentials, "token": create_token_credentials, + "device_code": create_device_code_credentials, } return options[env_vars.LOGIN_FLOW](env_vars) @@ -58,3 +69,19 @@ def create_token_credentials(env_vars: ClientEnvironmentVariables) -> Credential if not env_vars.CDF_TOKEN: raise ValueError("CDF_TOKEN environment variable must be set for token authentication.") return Token(env_vars.CDF_TOKEN) + + +def create_device_code_credentials(env_vars: ClientEnvironmentVariables) -> CredentialProvider: + client_id: str + if env_vars.PROVIDER == "entra_id": + client_id = NEAT_CLIENT_ENTRA_ID + elif env_vars.IDP_CLIENT_ID is None: + raise ValueError("IDP_CLIENT_ID environment variable must be set for device code authentication.") + else: + client_id = env_vars.IDP_CLIENT_ID + return OAuthDeviceCode( + authority_url=env_vars.idp_authority_url, + client_id=client_id, + scopes=env_vars.idp_scopes, + audience=env_vars.idp_audience, + ) diff --git a/cognite/neat/_client/init/env_vars.py b/cognite/neat/_client/init/env_vars.py index e1aa15ccc..3444915d2 100644 --- a/cognite/neat/_client/init/env_vars.py +++ b/cognite/neat/_client/init/env_vars.py @@ -11,7 +11,8 @@ else: from typing_extensions import Self -LoginFlow: TypeAlias = Literal["client_credentials", "interactive", "token"] + +LoginFlow: TypeAlias = Literal["client_credentials", "interactive", "device_code", "token"] AVAILABLE_LOGIN_FLOWS: tuple[LoginFlow, ...] = get_args(LoginFlow) Provider: TypeAlias = Literal["entra_id", "auth0", "cdf", "other"] AVAILABLE_PROVIDERS: tuple[Provider, ...] = get_args(Provider) @@ -85,6 +86,13 @@ def idp_audience(self) -> str: @property def idp_scopes(self) -> list[str]: + if self.PROVIDER == "entra_id" and self.LOGIN_FLOW == "device_code": + return [ + f"https://{self.CDF_CLUSTER}.cognitedata.com/IDENTITY", + f"https://{self.CDF_CLUSTER}.cognitedata.com/user_impersonation", + "profile", + "openid", + ] if self.IDP_SCOPES: return self.IDP_SCOPES.split(",") if self.PROVIDER == "auth0":