Skip to content

feat: add client credentials unattended auth flow #404

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open
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
128 changes: 128 additions & 0 deletions src/uipath/_cli/_auth/_client_credentials.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,128 @@
from typing import Optional
from urllib.parse import urlparse

import httpx

from .._utils._console import ConsoleLogger
from ._models import TokenData
from ._utils import parse_access_token, update_env_file

console = ConsoleLogger()


class ClientCredentialsService:
"""Service for client credentials authentication flow."""

def __init__(self, domain: str):
self.domain = domain

def get_token_url(self) -> str:
"""Get the token URL for the specified domain."""
if self.domain == "alpha":
return "https://alpha.uipath.com/identity_/connect/token"
elif self.domain == "staging":
return "https://staging.uipath.com/identity_/connect/token"
else: # cloud (default)
return "https://cloud.uipath.com/identity_/connect/token"

def extract_domain_from_base_url(self, base_url: str) -> str:
"""Extract domain from base URL.

Args:
base_url: The base URL to extract domain from

Returns:
The domain (alpha, staging, or cloud)
"""
try:
parsed = urlparse(base_url)
hostname = parsed.hostname

if hostname:
if "alpha.uipath.com" in hostname:
return "alpha"
elif "staging.uipath.com" in hostname:
return "staging"
elif "cloud.uipath.com" in hostname:
return "cloud"

# Default to cloud if we can't determine
return "cloud"
except Exception:
# Default to cloud if parsing fails
return "cloud"

def authenticate(
self, client_id: str, client_secret: str, scope: str = "OR.Execution"
) -> Optional[TokenData]:
"""Authenticate using client credentials flow.

Args:
client_id: The client ID for authentication
client_secret: The client secret for authentication
scope: The scope for the token (default: OR.Execution)

Returns:
Token data if successful, None otherwise
"""
token_url = self.get_token_url()

data = {
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
"scope": scope,
}

try:
with httpx.Client(timeout=30.0) as client:
response = client.post(token_url, data=data)

if response.status_code == 200:
token_data = response.json()
# Convert to our TokenData format
return {
"access_token": token_data["access_token"],
"token_type": token_data.get("token_type", "Bearer"),
"expires_in": token_data.get("expires_in", 3600),
"scope": token_data.get("scope", scope),
# Client credentials flow doesn't provide these, but we need them for compatibility
"refresh_token": "",
"id_token": "",
}
elif response.status_code == 400:
console.error("Invalid client credentials or request parameters.")
return None
elif response.status_code == 401:
console.error("Unauthorized: Invalid client credentials.")
return None
else:
console.error(
f"Authentication failed: {response.status_code} - {response.text}"
)
return None

except httpx.RequestError as e:
console.error(f"Network error during authentication: {e}")
return None
except Exception as e:
console.error(f"Unexpected error during authentication: {e}")
return None

def setup_environment(self, token_data: TokenData, base_url: str):
"""Setup environment variables for client credentials authentication.

Args:
token_data: The token data from authentication
base_url: The base URL for the UiPath instance
"""
parsed_access_token = parse_access_token(token_data["access_token"])

env_vars = {
"UIPATH_ACCESS_TOKEN": token_data["access_token"],
"UIPATH_URL": base_url,
"UIPATH_ORGANIZATION_ID": parsed_access_token.get("prt_id", ""),
"UIPATH_TENANT_ID": "",
}

update_env_file(env_vars)
63 changes: 61 additions & 2 deletions src/uipath/_cli/cli_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

from ..telemetry import track
from ._auth._auth_server import HTTPServer
from ._auth._client_credentials import ClientCredentialsService
from ._auth._oidc_utils import get_auth_config, get_auth_url
from ._auth._portal_service import PortalService, select_tenant
from ._auth._utils import update_auth_file, update_env_file
Expand Down Expand Up @@ -64,9 +65,67 @@ def set_port():
required=False,
help="Force new token",
)
@click.option(
"--client-id",
required=False,
help="Client ID for client credentials authentication (unattended mode)",
)
@click.option(
"--client-secret",
required=False,
help="Client secret for client credentials authentication (unattended mode)",
)
@click.option(
"--base-url",
required=False,
help="Base URL for the UiPath tenant instance (required for client credentials)",
)
@track
def auth(domain, force: None | bool = False):
"""Authenticate with UiPath Cloud Platform."""
def auth(
domain,
force: None | bool = False,
client_id: str = None,
client_secret: str = None,
base_url: str = None,
):
"""Authenticate with UiPath Cloud Platform.

Interactive mode (default): Opens browser for OAuth authentication.
Unattended mode: Use --client-id, --client-secret and --base-url for client credentials flow.
"""
# Check if client credentials are provided for unattended authentication
if client_id and client_secret:
if not base_url:
console.error(
"--base-url is required when using client credentials authentication."
)
return

with console.spinner("Authenticating with client credentials ..."):
# Create service instance
credentials_service = ClientCredentialsService(domain)

# If base_url is provided, extract domain from it to override the CLI domain parameter
if base_url:
extracted_domain = credentials_service.extract_domain_from_base_url(
base_url
)
credentials_service.domain = extracted_domain

token_data = credentials_service.authenticate(client_id, client_secret)

if token_data:
credentials_service.setup_environment(token_data, base_url)
console.success(
"Client credentials authentication successful.",
)
else:
console.error(
"Client credentials authentication failed. Please check your credentials.",
)
return

# Interactive authentication flow (existing logic)
with console.spinner("Authenticating with UiPath ..."):
portal_service = PortalService(domain)

Expand Down