Skip to content

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.

flowchart TD
    A[Flask Command] --> B{is_production_mode?}
    B -->|No| C[Skip Auth<br/>Dev Mode]
    B -->|Yes| D{is_cronjob_mode?}
    D -->|Yes| E[Service Account Flow]
    D -->|No| F{FLASK_CLI_ACCESS_TOKEN<br/>env var set?}
    F -->|Yes| G[Use provided token]
    F -->|No| H{one_off_container<br/>or dev_mode?}
    H -->|Yes| I[Interactive Flow]
    H -->|No| J[Fail]
    E --> K[authenticate_actor]
    G --> K
    I --> K
    K --> L[authorize_command]
Hold "Alt" / "Option" to enable pan & zoom

Human Authentication (Interactive Flow)

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)

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.

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

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

@my_cli.command(permitted_for={EmployeePermission.admin_action})
def admin_command():
    """Command requiring specific permission."""
    pass

Source: shared/cli/helpers/command.py

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

@my_cli.command(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{is_production_mode<br/>OR --auth flag?}
    D -->|No| C
    D -->|Yes| E{is_cronjob_mode?}
    E -->|Yes| F[get_service_account_access_token]
    E -->|No| G{FLASK_CLI_ACCESS_TOKEN?}
    G -->|Yes| H[Use env var token]
    G -->|No| I{one_off_container<br/>OR dev_mode?}
    I -->|Yes| J[get_interactive_user_access_token]
    I -->|No| K[Fail: No auth method]
    F --> L[authenticate_actor]
    H --> L
    J --> L
    L --> M[authorize_command]
    M --> N{Has permission?}
    N -->|Yes| O[Execute command]
    N -->|No| P[Exit with error]
Hold "Alt" / "Option" to enable pan & zoom

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 Production Mode
        Ext->>G: Device Code Flow<br/>(same as CLI)
        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<br/>"All commands are logged"
        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:

  1. Pushes Flask app context - Makes current_app, db, etc. available
  2. Authenticates the user - Same OAuth flow as CLI commands
  3. Checks permissions - Requires execute_flask_shell permission
  4. Configures audit logging - All commands logged to CloudWatch/Datadog
  5. Sets transaction timeout - Prevents idle transactions from blocking the DB

Implementation

def _authenticate_interactive_session():
    if is_production_mode():
        from shared.cli.helpers.auth import (
            authenticate_actor,
            get_interactive_user_access_token,
        )

        # Same OAuth flow as CLI commands
        authenticate_actor(
            env.str("FLASK_CLI_ACCESS_TOKEN", None)
            or get_interactive_user_access_token()
        )

        # Check shell-specific permission
        authorized = current_auth_context.has_backoffice_permissions(
            EmployeePermission.execute_flask_shell
        )
        if not authorized:
            current_logger.error("Not authorized to run Flask shell")
            raise SystemExit

Source: shared/cli/helpers/interactive_session.py:38-64

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:100-167

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 OAuth flow same as CLI commands
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