Skip to content
Open
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
172 changes: 164 additions & 8 deletions src/workato_platform_cli/cli/utils/exception_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,78 @@ def _get_output_mode() -> str:
return "table"


def _get_required_permissions(ctx: click.Context) -> list[str]:
"""Get required Workato permissions for a CLI command.

Args:
ctx: Click context object

Returns:
List of required permission strings
"""
# Map CLI command names to required Workato permissions
permission_map = {
"init": [
"Projects → Projects & folders",
"Projects → Connections",
"Projects → Recipes",
"Projects → Recipe Versions",
"Projects → Recipe lifecycle management",
"Projects → Export manifests",
"Tools → Collections & endpoints",
"Admin → Workspace details",
],
"recipes": [
"Projects → Export manifests",
"Projects → Recipes",
],
"pull": [
"Projects → Recipe lifecycle management",
"Projects → Export manifests",
],
"push": [
"Projects → Recipe lifecycle management",
],
"api-clients": [
"Tools → Clients & access profiles",
],
"api-collections": [
"Tools → Collections & endpoints",
],
"assets": [
"Projects → Export manifests",
],
"connections": [
"Projects → Connections",
],
"connectors": [
"Tools → Connector SDKs",
"Tools → Connectors",
],
"data-tables": [
"Tools → Data tables",
],
"properties": [
"Tools → Environment properties",
],
"workspace": [
"Admin → Workspace details",
],
}

# Get command name from context
# For nested commands like "workato recipes list", check parent context
command_name = ctx.info_name
if ctx.parent and ctx.parent.info_name and ctx.parent.info_name not in ("workato",):
# If parent is not root, use parent's name (the command group)
command_name = ctx.parent.info_name

if not command_name:
return []

return permission_map.get(command_name, [])


def _handle_client_error(
e: BadRequestException | UnprocessableEntityException,
) -> None:
Expand Down Expand Up @@ -233,24 +305,108 @@ def _handle_client_error(


def _handle_auth_error(e: UnauthorizedException) -> None:
"""Handle 401 Unauthorized errors."""
"""Handle 401 Unauthorized errors.

For 'workato init' command: Shows both authentication and authorization
possibilities since we cannot distinguish between them at this stage.

For other commands: Shows only authorization error, assuming the token
was already validated during initialization. A 401 error is treated as
missing permissions.

TODO: This is a temporary solution. The API should return distinct error
codes for authentication vs authorization failures. Track progress at:
https://github.com/workato-devs/workato-platform-cli-issues/issues/106
"""
output_mode = _get_output_mode()
ctx = click.get_current_context(silent=True)

# Check if this is the init command (top-level init, not subcommands)
is_init_command = False
if ctx and ctx.command:
is_init_command = (
ctx.command.name == "init"
and ctx.parent
and ctx.parent.info_name == "workato"
)

# Get required permissions for this command
required_permissions = []
if ctx:
required_permissions = _get_required_permissions(ctx)

if output_mode == "json":
if is_init_command:
error_msg = (
"Authentication failed - invalid or missing API token "
"or insufficient permissions"
)
else:
error_msg = "Authorization failed - insufficient permissions"

error_data = {
"status": "error",
"error": "Authentication failed - invalid or missing API token",
"error": error_msg,
"error_code": "UNAUTHORIZED",
}
click.echo(json.dumps(error_data))
return

click.echo("❌ Authentication failed")
click.echo(" Your API token may be invalid")
click.echo("💡 Please check your authentication:")
click.echo(" • Verify your API token is correct")
click.echo(" • Run 'workato profiles list' to check your profile")
click.echo(" • Run 'workato profiles use' to update your credentials")
command_info = f" (command: {ctx.command_path})" if ctx else ""

if is_init_command:
# For init command: Show both possibilities since we can't distinguish
click.echo(f"❌ Authentication failed{command_info}")
click.echo(" This could be due to:")
click.echo(" • Invalid or expired API token")
click.echo(" • API client lacking required permissions")
click.echo()

# Show required permissions if available
if required_permissions:
click.echo("🔐 Required permissions for this command:")
for permission in required_permissions:
click.echo(f" • {permission}")
click.echo()

click.echo("🔧 To resolve:")
click.echo(" • Verify your API token is correct")
click.echo(
" • Ensure your API client has all required permissions in Workato"
)
click.echo(" • Run 'workato profiles list' to check your profile")
click.echo(" • Run 'workato profiles use' to update your credentials")
click.echo()
click.echo("📚 Learn more about authentication and permissions")
click.echo(" https://docs.workato.com/en/platform-cli.html#authentication")
else:
# For non-init commands: Show only authorization error
# Assumption: Token validity was already verified during init when
# the profile was created. If they're using an existing profile and
# getting 401, it's an authorization issue (missing permissions).
click.echo(f"❌ Authorization failed{command_info}")
click.echo(
" Your API client lacks the required permissions for this operation"
)
click.echo()

# Show required permissions if available
if required_permissions:
click.echo("🔐 Required permissions for this command:")
for permission in required_permissions:
click.echo(f" • {permission}")
click.echo()

click.echo("🔧 To resolve:")
click.echo(" • Update your API client permissions in Workato")
click.echo(" • Ensure the permissions listed above are enabled")
click.echo(
" • Run 'workato profiles use' if you need to switch to "
"a different API client"
)
click.echo()
click.echo("📚 Learn more about permissions required for API client")
click.echo(" https://docs.workato.com/en/platform-cli.html#authentication")


def _handle_forbidden_error(e: ForbiddenException) -> None:
Expand Down