CLI Authentication¶

This guide covers authentication for Flask CLI commands, supporting both human operators and automated service accounts.

Overview¶

CLI commands can be executed by:

Humans - Developers running commands locally or in one-off containers
Service accounts - Automated cron jobs and scheduled tasks

Both flows use Google OAuth to obtain an access token, but the mechanism differs. Authentication is handled by a chain of CliAuthContextProvider instances, mirroring the HTTP AuthContextProvider ⧉ pattern. When CLI auth providers are registered on the app, authenticate_command delegates to them. Apps that haven't migrated yet fall back to the legacy if/else cascade.

flowchart TD
    A[Flask Command] --> B{system_command?}
    B -->|Yes| C[Execute without auth]
    B -->|No| D{CLI auth providers<br/>registered?}
    D -->|Yes| E[run_cli_auth_provider_chain]
    E --> F{Exactly one<br/>will_handle = True?}
    F -->|Yes| G[provider.set_auth_context_from_cli]
    F -->|No| H[Fail: 0 or multiple matched]
    G --> I[authorize_command]
    D -->|No| J[Legacy fallback]
    J --> I
    I --> K{Has permission?}
    K -->|Yes| L[Execute command]
    K -->|No| M[Exit with error]

Hold "Alt" / "Option" to enable pan & zoom

CLI Auth Context Providers¶

CliAuthContextProvider is the CLI counterpart to AuthContextProvider. Where HTTP providers examine the request (will_handle_request / set_auth_context_from_request), CLI providers examine the environment (will_handle / set_auth_context_from_cli).

Base Class¶

from abc import ABC, abstractmethod
from shared.iam.auth_context.providers.cli import CliAuthContextProvider

class CliAuthContextProvider(ABC):
    def __init__(
        self,
        *auth_principal_types: type[AuthPrincipal],
        principal_attribute: str | None = None,
    ): ...

    @abstractmethod
    def will_handle(self) -> bool:
        """Whether this provider handles the current CLI context."""

    @abstractmethod
    def set_auth_context_from_cli(self) -> None:
        """Perform authentication and set auth context."""

    def resolve_and_set_auth_context(self, email: str) -> None:
        """Look up principal by email and call set_auth_context().

        When principal_attribute is set, resolves the principal
        via getattr(model, principal_attribute).
        """

Source: shared/iam/auth_context/providers/cli.py:26-76 ⧉

Built-in Providers¶

Provider	`will_handle()` condition	Auth method	Typical use
`DevCliAuthContextProvider`	Always True (only registered in dev)	AWS username → `{user}@alan.eu`	Dev/test
`CronCliAuthContextProvider`	`is_cronjob_mode()`	Service account credentials from Secrets Manager	Scheduled tasks
`EnvVarCliAuthContextProvider`	`FLASK_CLI_ACCESS_TOKEN` env var is set	Validates pre-set Google OAuth token	Alan CLI tool
`InteractiveCliAuthContextProvider`	In one-off container or dev mode, and no `FLASK_CLI_ACCESS_TOKEN`	Google Device Code Flow	Human operators

Source: shared/iam/auth_context/providers/cli.py:78-148 ⧉

Provider Chain¶

run_cli_auth_provider_chain(*providers) runs the chain, equivalent to the HTTP before_request auth provider chain:

Iterates providers, calls will_handle() on each
Collects eligible providers (those returning True)
Raises RuntimeError if zero or multiple match
Calls set_auth_context_from_cli() on the single match

flowchart LR
    A[Providers] --> B[will_handle?]
    B --> C{Eligible count}
    C -->|0| D[RuntimeError:<br/>No provider matched]
    C -->|1| E[set_auth_context_from_cli]
    C -->|2+| F[RuntimeError:<br/>Multiple matched]

Hold "Alt" / "Option" to enable pan & zoom

Source: shared/iam/auth_context/providers/cli.py:150-169 ⧉

Registration¶

Providers are registered in the app's create_app() via app.cli.register_auth_context_providers():

from shared.iam.auth_context.providers.cli import (
    CronCliAuthContextProvider,
    DevCliAuthContextProvider,
    EnvVarCliAuthContextProvider,
    InteractiveCliAuthContextProvider,
)

if is_development_mode():
    app.cli.register_auth_context_providers(
        DevCliAuthContextProvider(Alaner),
    )
else:
    app.cli.register_auth_context_providers(
        CronCliAuthContextProvider(Alaner),
        EnvVarCliAuthContextProvider(Alaner),
        InteractiveCliAuthContextProvider(Alaner),
    )

When the principal model differs from the auth principal, use principal_attribute to resolve via a nested attribute. For example, ca_api uses CaAlanEmployee but stores the auth principal on its .user attribute:

app.cli.register_auth_context_providers(
    CronCliAuthContextProvider(CaAlanEmployee, principal_attribute="user"),
    EnvVarCliAuthContextProvider(CaAlanEmployee, principal_attribute="user"),
    InteractiveCliAuthContextProvider(CaAlanEmployee, principal_attribute="user"),
)

Source: shared/cli/helpers/app_group.py:34-45 ⧉

Creating Custom Providers¶

Subclass CliAuthContextProvider and implement will_handle and set_auth_context_from_cli:

from shared.iam.auth_context.providers.cli import CliAuthContextProvider

class MyCustomCliAuthContextProvider(CliAuthContextProvider):
    def will_handle(self) -> bool:
        return some_custom_condition()

    def set_auth_context_from_cli(self) -> None:
        email = get_email_somehow()
        self.resolve_and_set_auth_context(email)

Parallels with HTTP providers

If you've created a custom AuthContextProvider for HTTP, the CLI equivalent follows the same pattern. Replace will_handle_request with will_handle, and set_auth_context_from_request with set_auth_context_from_cli.

Human Authentication (Interactive Flow)¶

Note

This flow is used by InteractiveCliAuthContextProvider (and the legacy fallback path).

When a human runs a command in production (one-off container or local with --auth), they authenticate via Google's Device Code Flow.

Flow Diagram¶

sequenceDiagram
    participant H as Human
    participant CLI as Flask CLI
    participant G as Google OAuth
    participant API as Alan Backend

    H->>CLI: flask my-command --auth
    CLI->>G: POST /device/code<br/>{client_id, scope: "email"}
    G-->>CLI: {verification_url, user_code, device_code}
    CLI->>H: Display URL and code

    H->>G: Open URL, enter code
    H->>G: Approve access

    loop Poll for completion
        CLI->>G: POST /token<br/>{device_code, grant_type}
        G-->>CLI: 428 (pending) or access_token
    end

    CLI->>API: Validate token
    API->>G: userinfo().get()
    G-->>API: {email}
    API-->>CLI: Authenticated as email
    CLI->>CLI: authorize_command()
    CLI->>H: Command executes

Hold "Alt" / "Option" to enable pan & zoom

What the User Sees¶

$ flask my-command --auth
Open the following URL in your browser and enter the verification code to authenticate
- Go to: https://www.google.com/device
- Enter code: ABCD-EFGH
Authenticated as: olivier@alan.com

Implementation¶

def get_interactive_user_access_token() -> str | None:
    oauth_client = json_secret_from_config("CLI_OAUTH_CLIENT_SECRET_NAME")

    # Request device code
    response = requests.post(
        "https://oauth2.googleapis.com/device/code",
        json={"client_id": oauth_client["installed"]["client_id"], "scope": "email"},
    )
    device_verification_info = response.json()

    # Display instructions
    click.secho("Open the following URL in your browser...", bold=True)
    click.echo(f"- Go to: {device_verification_info['verification_url']}")
    click.echo(f"- Enter code: {device_verification_info['user_code']}")

    # Poll for completion
    while not expired:
        token_response = requests.post(
            "https://oauth2.googleapis.com/token",
            json={
                "client_id": ...,
                "device_code": device_verification_info["device_code"],
                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
            },
        )
        if token_response.status_code != 428:  # Not pending
            return token_response.json()["access_token"]
        time.sleep(device_verification_info["interval"])

Source: shared/cli/helpers/auth.py:201-240 ⧉

Service Account Authentication (Cron Jobs)¶

Note

This flow is used by CronCliAuthContextProvider (and the legacy fallback path).

Scheduled commands authenticate using a dedicated service account, without human interaction.

Flow Diagram¶

sequenceDiagram
    participant Cron as Cron Scheduler
    participant CLI as Flask CLI
    participant AWS as Secrets Manager
    participant G as Google OAuth
    participant DB as Database

    Cron->>CLI: flask scheduled-command
    Note over CLI: is_cronjob_mode() = True

    CLI->>AWS: Fetch FLASK_CLI_SCHEDULER_<br/>SERVICE_ACCOUNT_CREDENTIALS
    AWS-->>CLI: JSON key file

    CLI->>CLI: Sign JWT assertion<br/>with private key
    CLI->>G: Exchange for access token
    G-->>CLI: access_token

    CLI->>G: userinfo().get()
    G-->>CLI: {email: "cron@...gserviceaccount.com"}

    CLI->>DB: Look up AlanEmployee<br/>by email
    DB-->>CLI: Service account principal

    CLI->>CLI: set_auth_context(principal)
    CLI->>CLI: authorize_command()
    Note over CLI: Check execute_scheduled_command<br/>permission
    CLI->>CLI: Command executes

Hold "Alt" / "Option" to enable pan & zoom

Implementation¶

def get_service_account_access_token(service_account_credentials: dict) -> str | None:
    from google.auth.transport.requests import Request
    from google.oauth2 import service_account

    credentials = service_account.Credentials.from_service_account_info(
        service_account_credentials,
        scopes=["https://www.googleapis.com/auth/userinfo.email"],
    )
    credentials.refresh(Request())
    return credentials.token

Source: shared/cli/helpers/auth.py:184-198 ⧉

The `@command` Decorator¶

All Flask commands use the @command decorator which wraps authentication and authorization. The authenticate_command wrapper checks for registered CLI auth context providers first, and falls back to the legacy path when none are registered.

Decorator Chain¶

flowchart LR
    A[command] --> B[log_command]
    B --> C[monitor_transactions]
    C --> D[authorize_command]
    D --> E[authenticate_command]
    E --> F[correlate_command]
    F --> G[Actual Command]

Hold "Alt" / "Option" to enable pan & zoom

Usage¶

from shared.cli.helpers.command import command
from shared.iam.abac.access_policy import enforce_policy
from shared.iam.abac.backoffice import BackofficePermissionAccessPolicy

@my_cli.command()  # Uses default authentication + default permission
def my_command():
    """Standard command requiring authentication in production."""
    pass

@my_cli.command()
@enforce_policy(BackofficePermissionAccessPolicy.permitted_for(EmployeePermission.manage_users))
def admin_command():
    """Command requiring specific permission via enforce_policy."""
    pass

When no explicit @enforce_policy is present, authorize_command applies a default BackofficePermissionAccessPolicy with execute_scheduled_command (cron) or execute_generic_command (manual).

Source: shared/cli/helpers/command.py ⧉

Deprecated: permitted_for and authenticate_actor

The permitted_for kwarg on @command() is deprecated — use @enforce_policy instead. Similarly, authenticate_actor() is deprecated — use CLI auth context providers. Both are still used by the legacy fallback path.

System Commands¶

Infrastructure commands that don't perform business actions can skip authentication:

from shared.cli.helpers.auth import system_command

@rq.command("worker")
@system_command
def start_worker():
    """System command - skips authentication."""
    pass

Source: shared/cli/helpers/auth.py:40-47 ⧉

Use Sparingly

Only use @system_command for infrastructure commands like runserver, worker, etc. Never for commands that modify data.

Permissions¶

Default Permissions¶

Context	Default Permission
Cron job (`is_cronjob_mode()`)	`execute_scheduled_command`
Manual execution	`execute_generic_command`
Flask shell	`execute_flask_shell`

Custom Permissions¶

Use @enforce_policy to require specific permissions. Access policies are much more flexible than the deprecated permitted_for — they support arbitrary authorization logic beyond simple permission checks (e.g. resource ownership, delegation context, compound conditions). See Access Control ⧉ for full documentation.

from shared.iam.abac.access_policy import enforce_policy
from shared.iam.abac.backoffice import BackofficePermissionAccessPolicy

@my_cli.command()
@enforce_policy(BackofficePermissionAccessPolicy.permitted_for(EmployeePermission.manage_users))
def dangerous_command():
    """Requires specific permission."""
    pass

Authentication Flow Decision Tree¶

flowchart TD
    A[Command Invoked] --> B{system_command?}
    B -->|Yes| C[Execute without auth]
    B -->|No| D{CLI auth providers<br/>registered?}
    D -->|Yes| E[run_cli_auth_provider_chain]
    E --> F{Exactly one<br/>will_handle?}
    F -->|Yes| G[provider.set_auth_context_from_cli]
    F -->|No| H[Fail: 0 or multiple matched]
    G --> I[authorize_command]
    D -->|No| J{is_production_mode<br/>OR --auth flag?}
    J -->|No| C
    J -->|Yes| K[Legacy if/else cascade]
    K --> I
    I --> L{Has permission?}
    L -->|Yes| M[Execute command]
    L -->|No| N[Exit with error]

Hold "Alt" / "Option" to enable pan & zoom

Legacy path

The right branch (legacy if/else cascade) only runs when no CliAuthContextProvider instances are registered on app.cli. All apps should migrate to the provider pattern.

Interactive Python Shell¶

The Flask shell (flask shell) and iPython sessions in production also require authentication.

Flow Diagram¶

sequenceDiagram
    participant H as Human
    participant IP as iPython
    participant Ext as iPython Extension
    participant G as Google OAuth
    participant DB as Database

    H->>IP: flask shell
    IP->>Ext: load_ipython_extension()
    Ext->>Ext: Push Flask app context

    alt Providers registered
        Ext->>Ext: run_cli_auth_provider_chain
        Note over Ext: Provider handles auth<br/>(e.g. Device Code Flow)
        Ext->>DB: Look up principal
        DB-->>Ext: Principal
        Ext->>Ext: Check execute_flask_shell<br/>permission
        alt Authorized
            Ext->>Ext: Configure CloudWatch logging
            Ext->>H: Shell ready<br/>"All commands are logged"
        else Not Authorized
            Ext->>H: Exit with error
        end
    else Legacy fallback (production)
        Ext->>G: Device Code Flow
        G-->>Ext: access_token
        Ext->>DB: Look up AlanEmployee
        DB-->>Ext: Principal
        Ext->>Ext: Check execute_flask_shell<br/>permission
        alt Authorized
            Ext->>Ext: Configure CloudWatch logging
            Ext->>H: Shell ready
        else Not Authorized
            Ext->>H: Exit with error
        end
    else Development Mode
        Ext->>Ext: Enable autoreload
        Ext->>H: Shell ready
    end

    loop Interactive Session
        H->>IP: Execute command
        IP->>Ext: pre_run_cell (log input)
        IP->>IP: Execute
        IP->>Ext: post_run_cell (log output)
    end

Hold "Alt" / "Option" to enable pan & zoom

How It Works¶

The interactive session is configured via an iPython extension that:

Pushes Flask app context - Makes current_app, db, etc. available
Authenticates the user - Uses CLI auth provider chain (or legacy OAuth flow as fallback)
Checks permissions - Requires execute_flask_shell permission
Configures audit logging - All commands logged to CloudWatch/Datadog
Sets transaction timeout - Prevents idle transactions from blocking the DB

Implementation¶

def _authenticate_interactive_session():
    providers = getattr(current_app.cli, "auth_context_providers", ())
    if providers:
        from shared.iam.auth_context.providers.cli import run_cli_auth_provider_chain

        run_cli_auth_provider_chain(*providers)

        authorized = BackofficePermissionAccessPolicy.permitted_for(
            EmployeePermission.execute_flask_shell
        ).do_evaluate()
        if not authorized:
            current_logger.error("Not authorized to run Flask shell")
            raise SystemExit

    elif is_production_mode():
        # Legacy path
        from shared.cli.helpers.auth import (
            authenticate_actor,
            get_interactive_user_access_token,
        )

        authenticate_actor(
            env.str("FLASK_CLI_ACCESS_TOKEN", None)
            or get_interactive_user_access_token()
        )
        # ... authorization check ...

Source: shared/cli/helpers/interactive_session.py:39-84 ⧉

Command Logging¶

In production, every command and its output is logged:

class IPythonCloudwatchLogger:
    def pre_run_cell(self, info):
        # Log the command being executed
        self.logger.info("interactive command", stdin=info.raw_cell)

    def post_run_cell(self, result):
        # Log stdout/stderr output
        self.logger.info("interactive command", stdout=..., stderr=...)

Source: shared/cli/helpers/interactive_session.py:120-187 ⧉

What the User Sees¶

$ flask shell

Configuring interactive Python session
Application: fr_api
This is a production environment - handle with care
DB sessions with idle transactions will timeout after 5min
Open the following URL in your browser and enter the verification code to authenticate
- Go to: https://www.google.com/device
- Enter code: ABCD-EFGH
Authenticated as: olivier@alan.com
All commands are logged
Logs are available in Datadog: https://app.datadoghq.eu/logs?query=...

In [1]: User.query.first()

Security Features¶

Feature	Description
Authentication	CLI auth provider chain (or legacy OAuth fallback)
Authorization	Requires `execute_flask_shell` permission
Audit logging	All commands + output sent to CloudWatch/Datadog
Session correlation	Unique `interactive_session_id` links all commands
Transaction timeout	5-minute idle timeout prevents DB blocking

Production Shell Access

Interactive shell access in production is a privileged operation. All commands are logged and auditable. Use with care.

Environment Variables¶

Variable	Description
`FLASK_CLI_ACCESS_TOKEN`	Pre-configured access token (skips interactive flow)
`FLASK_CLI_SCHEDULER_SERVICE_ACCOUNT_CREDENTIALS`	JSON credentials for cron service account
`CLI_OAUTH_CLIENT_SECRET_NAME`	Secret manager key for OAuth client

Using the Alan CLI

When running commands via the alan CLI tool (e.g., alan qovery run), the FLASK_CLI_ACCESS_TOKEN is automatically set for you. Instead of the device code flow (copy-paste a code), you get a simpler browser-based OAuth flow that opens automatically.

# This handles authentication automatically
alan qovery run fr_api -- flask my-command

The alan CLI authenticates you via browser, obtains the token, and injects it as FLASK_CLI_ACCESS_TOKEN before running your command.

Testing Commands Locally¶

With Authentication¶

# Force authentication in development
flask my-command --auth

Without Authentication¶

# Development mode skips auth by default
flask my-command

With a Specific Token¶

# Provide token via environment
FLASK_CLI_ACCESS_TOKEN=<token> flask my-command

CLI Authentication¶

Overview¶

CLI Auth Context Providers¶

Base Class¶

Built-in Providers¶

Provider Chain¶

Registration¶

Creating Custom Providers¶

Human Authentication (Interactive Flow)¶

Flow Diagram¶

What the User Sees¶

Implementation¶

Service Account Authentication (Cron Jobs)¶

Flow Diagram¶

Implementation¶

The @command Decorator¶

Decorator Chain¶

Usage¶

System Commands¶

Permissions¶

Default Permissions¶

Custom Permissions¶

Authentication Flow Decision Tree¶

Interactive Python Shell¶

Flow Diagram¶

How It Works¶

Implementation¶

Command Logging¶

What the User Sees¶

Security Features¶

Environment Variables¶

Testing Commands Locally¶

With Authentication¶

Without Authentication¶

With a Specific Token¶

The `@command` Decorator¶