Skip to content
Closed
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
6 changes: 6 additions & 0 deletions .changes/unreleased/added-20260630-223439.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
kind: added
body: Add ``--use-device-code`` flag to ``auth login`` for device code authentication flow
time: 2026-06-30T22:34:39.210039642+02:00
custom:
Author: iemejia
AuthorLink: https://github.com/iemejia
20 changes: 19 additions & 1 deletion src/fabric_cli/commands/auth/fab_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
def init(args: Namespace) -> Any:
auth_options = [
"Interactive with a web browser",
"Device code",
"Service principal authentication with secret",
"Service principal authentication with certificate",
"Service principal authentication with federated credential",
Expand All @@ -27,7 +28,16 @@ def init(args: Namespace) -> Any:
# Clean up stale context files when logging in
Context().cleanup_context_files(cleanup_all_stale=True, cleanup_current=False)

if args.identity:
if args.use_device_code:
auth = FabAuth()
auth.set_access_mode("user", args.tenant)
with auth.device_code_flow():
auth.get_access_token(scope=fab_constant.SCOPE_FABRIC_DEFAULT)
auth.get_access_token(scope=fab_constant.SCOPE_ONELAKE_DEFAULT)
auth.get_access_token(scope=fab_constant.SCOPE_AZURE_DEFAULT)
Context().context = auth.get_tenant()

elif args.identity:
FabAuth().set_access_mode("managed_identity")
FabAuth().set_managed_identity(args.username)
FabAuth().get_access_token(scope=fab_constant.SCOPE_FABRIC_DEFAULT)
Expand Down Expand Up @@ -73,6 +83,14 @@ def init(args: Namespace) -> Any:
FabAuth().get_access_token(scope=fab_constant.SCOPE_ONELAKE_DEFAULT)
FabAuth().get_access_token(scope=fab_constant.SCOPE_AZURE_DEFAULT)
Context().context = FabAuth().get_tenant()
elif selected_auth == "Device code":
auth = FabAuth()
auth.set_access_mode("user", args.tenant)
with auth.device_code_flow():
auth.get_access_token(scope=fab_constant.SCOPE_FABRIC_DEFAULT)
auth.get_access_token(scope=fab_constant.SCOPE_ONELAKE_DEFAULT)
auth.get_access_token(scope=fab_constant.SCOPE_AZURE_DEFAULT)
Context().context = auth.get_tenant()
elif selected_auth.startswith("Service principal authentication"):
fab_logger.log_warning(
"Ensure tenant setting is enabled for Service Principal auth"
Expand Down
90 changes: 83 additions & 7 deletions src/fabric_cli/core/fab_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import os
import uuid
from binascii import hexlify
from contextlib import contextmanager
from typing import Any, NamedTuple, Optional

import jwt
Expand Down Expand Up @@ -50,11 +51,31 @@ def __init__(self):
# Reset the auth info
self.app: msal.ClientApplication = None
self._auth_info = {}
self._use_device_code = False

# Load the auth info and environment variables
self._load_auth()
self._load_env()

@contextmanager
def device_code_flow(self):
"""Context manager to enable device code authentication flow.

Temporarily enables device code flow for token acquisition within
the context block. Automatically resets the flag on exit, even if
an exception occurs.

Usage:
auth = FabAuth()
with auth.device_code_flow():
auth.get_access_token(scope=...)
"""
self._use_device_code = True
try:
yield
finally:
self._use_device_code = False

def _save_auth(self):
from fabric_cli.utils.fab_secure_io import write_restricted_file

Expand Down Expand Up @@ -270,6 +291,54 @@ def _get_app(self) -> msal.ClientApplication:
)
return self.app

def _acquire_token_by_device_code(self, scopes: list[str]) -> Optional[dict]:
"""
Acquire a token using the device code flow.

This initiates a device code flow, prints the user code and verification
URL to the console, and waits for the user to complete authentication
on another device. If an account already exists in the cache (from a
prior device code flow in the same session), attempts a silent refresh
first to avoid prompting the user multiple times for different scopes.

Args:
scopes: The scopes for which to request the token

Returns:
Full MSAL result dictionary containing access_token, expires_on, etc.,
or None if the device flow returns no result.

Raises:
FabricCLIError: When device code flow initiation or token acquisition fails
Comment on lines +307 to +312

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed — added a blank line before Raises: section.

"""
# If we already have an account from a prior device code auth in this
# session, try silent acquisition with force_refresh to use the refresh
# token for a different resource scope.
app = self._get_app()
accounts = app.get_accounts()
if accounts:
token = app.acquire_token_silent(
scopes=scopes, account=accounts[0], force_refresh=True
)
if token and token.get("access_token"):
return token

flow = app.initiate_device_flow(scopes=scopes)
if "user_code" not in flow:
raise FabricCLIError(
ErrorMessages.Auth.device_code_flow_failed(
flow.get("error_description", "Unknown error")
),
con.ERROR_AUTHENTICATION_FAILED,
)

# Display the device code message to the user
utils_ui.print_grey(flow["message"])

# Wait for the user to authenticate
token = app.acquire_token_by_device_flow(flow)
return token

def _get_access_token_from_env_vars_if_exist(self, scope):
if "FAB_TOKEN" in os.environ and "FAB_TOKEN_ONELAKE" in os.environ:
match scope:
Expand Down Expand Up @@ -494,14 +563,21 @@ def acquire_token(self, scope: list[str], interactive_renew=True) -> dict:
scopes=scope, account=account
)

if token is None and interactive_renew:
token = self._get_app().acquire_token_interactive(
scopes=scope,
prompt="select_account",
parent_window_handle=msal.PublicClientApplication.CONSOLE_WINDOW_HANDLE,
)
if (token is None or token.get("error")) and interactive_renew:
if self._use_device_code:
token = self._acquire_token_by_device_code(scope)
else:
token = self._get_app().acquire_token_interactive(
scopes=scope,
prompt="select_account",
parent_window_handle=msal.PublicClientApplication.CONSOLE_WINDOW_HANDLE,
)
if token is not None and "id_token_claims" in token:
self.set_tenant(token.get("id_token_claims")["tid"])
id_token_claims = token.get("id_token_claims")
if id_token_claims:
tid = id_token_claims.get("tid")
if tid:
self.set_tenant(tid)

if token and token.get("error"):
fab_logger.log_debug(
Expand Down
4 changes: 4 additions & 0 deletions src/fabric_cli/errors/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -119,3 +119,7 @@ def cert_read_failed(error: str) -> str:
@staticmethod
def only_supported_with_user_authentication() -> str:
return "This operation is only supported with user authentication"

@staticmethod
def device_code_flow_failed(error_description: str) -> str:
return f"Failed to initiate device code flow: {error_description}"
14 changes: 11 additions & 3 deletions src/fabric_cli/parsers/fab_auth_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,8 @@ def register_parser(subparsers: _SubParsersAction) -> None:
"$ auth login\n",
"# command_line mode",
"$ fab auth login\n",
"# command_line mode using device code auth",
"$ fab auth login --use-device-code\n",
"# command_line mode using service principal auth",
"$ fab auth login -u <client_id> -p <client_secret> --tenant <tenant_id>\n",
"# command_line mode using system assigned managed identity auth",
Expand Down Expand Up @@ -84,9 +86,15 @@ def register_parser(subparsers: _SubParsersAction) -> None:
required=False,
help="Federated token that can be used for OIDC token exchange. Optional, only for service principal auth",
)
login_parser.add_argument(
"--use-device-code",
required=False,
action="store_true",
help="Use device code authentication flow. Useful when browser access is not available.",
)

login_parser.usage = f"{utils_error_parser.get_usage_prog(login_parser)}"
login_parser.set_defaults(func=lazy_command(_auth_module_path, 'init'))
login_parser.set_defaults(func=lazy_command(_auth_module_path, "init"))

# Subcommand for 'logout'
logout_examples = [
Expand All @@ -104,7 +112,7 @@ def register_parser(subparsers: _SubParsersAction) -> None:
)

logout_parser.usage = f"{utils_error_parser.get_usage_prog(logout_parser)}"
logout_parser.set_defaults(func=lazy_command(_auth_module_path, 'logout'))
logout_parser.set_defaults(func=lazy_command(_auth_module_path, "logout"))

# Subcommand for 'status'
status_examples = [
Expand All @@ -121,7 +129,7 @@ def register_parser(subparsers: _SubParsersAction) -> None:
)

status_parser.usage = f"{utils_error_parser.get_usage_prog(status_parser)}"
status_parser.set_defaults(func=lazy_command(_auth_module_path, 'status'))
status_parser.set_defaults(func=lazy_command(_auth_module_path, "status"))


def show_help(args: Namespace) -> None:
Expand Down
Loading
Loading