Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 28 additions & 1 deletion cognite/neat/_client/init/credentials.py
Original file line number Diff line number Diff line change
@@ -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)

Expand Down Expand Up @@ -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,
)
10 changes: 9 additions & 1 deletion cognite/neat/_client/init/env_vars.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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":
Expand Down
Loading