fix: enforce authentication for superuser cli command

.env.example

@@ -79,12 +79,16 @@ LANGFLOW_REMOVE_API_KEYS=
 # LANGFLOW_REDIS_CACHE_EXPIRE (default: 3600)
 LANGFLOW_CACHE_TYPE=
 
-# Set AUTO_LOGIN to false if you want to disable auto login
+# Set LANGFLOW_AUTO_LOGIN to false if you want to disable auto login
 # and use the login form to login. LANGFLOW_SUPERUSER and LANGFLOW_SUPERUSER_PASSWORD
 # must be set if AUTO_LOGIN is set to false
 # Values: true, false
 LANGFLOW_AUTO_LOGIN=
 
+# SET LANGFLOW_ENABLE_SUPERUSER_CLI to false to disable
+# superuser creation via the CLI
+LANGFLOW_ENABLE_SUPERUSER_CLI=
+
 # Superuser username
 # Example: LANGFLOW_SUPERUSER=admin
 LANGFLOW_SUPERUSER=
@@ -111,4 +115,4 @@ LANGFLOW_STORE_ENVIRONMENT_VARIABLES=
 
 # Value must finish with slash /
 #BACKEND_URL=http://localhost:7860/
-BACKEND_URL=
+BACKEND_URL=

SECURITY.md

@@ -59,4 +59,36 @@ Setting `LANGFLOW_SKIP_AUTH_AUTO_LOGIN=true` and `LANGFLOW_AUTO_LOGIN=true` skip
 
 `LANGFLOW_SKIP_AUTH_AUTO_LOGIN=true` is the default behavior, so users do not need to change existing workflows in 1.5. To update your workflows to require authentication, set `LANGFLOW_SKIP_AUTH_AUTO_LOGIN=false`.
 
-For more information, see [API keys and authentication](https://docs.langflow.org/api-keys-and-authentication).
+For more information, see [API keys and authentication](https://docs.langflow.org/api-keys-and-authentication).
+
+## Security Configuration Guidelines
+
+### Superuser Creation Security
+
+The `langflow superuser` CLI command can present a privilege escalation risk if not properly secured.
+
+#### Security Measures
+
+1. **Authentication Required in Production**
+   - When `LANGFLOW_AUTO_LOGIN=false`, superuser creation requires authentication
+   - Use `--auth-token` parameter with a valid superuser API key or JWT token
+
+2. **Disable CLI Superuser Creation**
+   - Set `LANGFLOW_ENABLE_SUPERUSER_CLI=false` to disable the command entirely
+   - Strongly recommended for production environments
+
+3. **Secure AUTO_LOGIN Setting**
+   - Default is `true` for <=1.5. This may change in a future release.
+   - When `true`, creates default superuser `langflow/langflow` - **ONLY USE IN DEVELOPMENT**
+
+#### Production Security Configuration
+
+```bash
+# Recommended production settings
+export LANGFLOW_AUTO_LOGIN=false
+export LANGFLOW_ENABLE_SUPERUSER_CLI=false
+export LANGFLOW_SUPERUSER="<your-superuser-username>"
+export LANGFLOW_SUPERUSER_PASSWORD="<your-superuser-password>"
+export LANGFLOW_DATABASE_URL="<your-production-database-url>" # e.g. "postgresql+psycopg://langflow:[email protected]:5432/langflow"
+export LANGFLOW_SECRET_KEY="your-strong-random-secret-key"
+```

src/backend/base/langflow/__main__.py

@@ -15,7 +15,9 @@
 import httpx
 import typer
 from dotenv import load_dotenv
+from fastapi import HTTPException
 from httpx import HTTPError
+from jose import JWTError
 from multiprocess import cpu_count
 from multiprocess.context import Process
 from packaging import version as pkg_version
@@ -29,9 +31,9 @@
 from langflow.initial_setup.setup import get_or_create_default_folder
 from langflow.logging.logger import configure, logger
 from langflow.main import setup_app
-from langflow.services.database.utils import session_getter
+from langflow.services.auth.utils import check_key, get_current_user_by_jwt
 from langflow.services.deps import get_db_service, get_settings_service, session_scope
-from langflow.services.settings.constants import DEFAULT_SUPERUSER
+from langflow.services.settings.constants import DEFAULT_SUPERUSER, DEFAULT_SUPERUSER_PASSWORD
 from langflow.services.utils import initialize_services
 from langflow.utils.version import fetch_latest_version, get_version_info
 from langflow.utils.version import is_pre_release as langflow_is_pre_release
@@ -632,41 +634,138 @@ def print_banner(host: str, port: int, protocol: str) -> None:
 
 @app.command()
 def superuser(
-    username: str = typer.Option(..., prompt=True, help="Username for the superuser."),
-    password: str = typer.Option(..., prompt=True, hide_input=True, help="Password for the superuser."),
+    username: str = typer.Option(
+        None, help="Username for the superuser. Defaults to 'langflow' when AUTO_LOGIN is enabled."
+    ),
+    password: str = typer.Option(
+        None, help="Password for the superuser. Defaults to 'langflow' when AUTO_LOGIN is enabled."
+    ),
     log_level: str = typer.Option("error", help="Logging level.", envvar="LANGFLOW_LOG_LEVEL"),
+    auth_token: str = typer.Option(
+        None, help="Authentication token of existing superuser.", envvar="LANGFLOW_SUPERUSER_TOKEN"
+    ),
 ) -> None:
-    """Create a superuser."""
+    """Create a superuser.
+
+    When AUTO_LOGIN is enabled, uses default credentials.
+    In production mode, requires authentication.
+    """
     configure(log_level=log_level)
-    db_service = get_db_service()
 
-    async def _create_superuser():
-        await initialize_services()
-        async with session_getter(db_service) as session:
-            from langflow.services.auth.utils import create_super_user
-
-            if await create_super_user(db=session, username=username, password=password):
-                # Verify that the superuser was created
-                from langflow.services.database.models.user.model import User
-
-                stmt = select(User).where(User.username == username)
-                user: User = (await session.exec(stmt)).first()
-                if user is None or not user.is_superuser:
-                    typer.echo("Superuser creation failed.")
-                    return
-                # Now create the first folder for the user
-                result = await get_or_create_default_folder(session, user.id)
-                if result:
-                    typer.echo("Default folder created successfully.")
-                else:
-                    msg = "Could not create default folder."
-                    raise RuntimeError(msg)
-                typer.echo("Superuser created successfully.")
+    asyncio.run(_create_superuser(username, password, auth_token))
 
-            else:
+
+async def _create_superuser(username: str, password: str, auth_token: str | None):
+    """Create a superuser."""
+    await initialize_services()
+
+    settings_service = get_settings_service()
+    # Check if superuser creation via CLI is enabled
+    if not settings_service.auth_settings.ENABLE_SUPERUSER_CLI:
+        typer.echo("Error: Superuser creation via CLI is disabled.")
+        typer.echo("Set LANGFLOW_ENABLE_SUPERUSER_CLI=true to enable this feature.")
+        raise typer.Exit(1)
+
+    if settings_service.auth_settings.AUTO_LOGIN:
+        # Force default credentials for AUTO_LOGIN mode
+        username = DEFAULT_SUPERUSER
+        password = DEFAULT_SUPERUSER_PASSWORD
+    else:
+        # Production mode - prompt for credentials if not provided
+        if not username:
+            username = typer.prompt("Username")
+        if not password:
+            password = typer.prompt("Password", hide_input=True)
+
+    from langflow.services.database.models.user.crud import get_all_superusers
+
+    existing_superusers = []
+    async with session_scope() as session:
+        # Note that the default superuser is created by the initialize_services() function,
+        # but leaving this check here in case we change that behavior
+        existing_superusers = await get_all_superusers(session)
+    is_first_setup = len(existing_superusers) == 0
+
+    # If AUTO_LOGIN is true, only allow default superuser creation
+    if settings_service.auth_settings.AUTO_LOGIN:
+        if not is_first_setup:
+            typer.echo("Error: Cannot create additional superusers when AUTO_LOGIN is enabled.")
+            typer.echo("AUTO_LOGIN mode is for development with only the default superuser.")
+            typer.echo("To create additional superusers:")
+            typer.echo("1. Set LANGFLOW_AUTO_LOGIN=false")
+            typer.echo("2. Run this command again with --auth-token")
+            raise typer.Exit(1)
+
+        typer.echo(f"AUTO_LOGIN enabled. Creating default superuser '{username}'...")
+        typer.echo(f"Note: Default credentials are {DEFAULT_SUPERUSER}/{DEFAULT_SUPERUSER_PASSWORD}")
+    # AUTO_LOGIN is false - production mode
+    elif is_first_setup:
+        typer.echo("No superusers found. Creating first superuser...")
+    else:
+        # Authentication is required in production mode
+        if not auth_token:
+            typer.echo("Error: Creating a superuser requires authentication.")
+            typer.echo("Please provide --auth-token with a valid superuser API key or JWT token.")
+            typer.echo("To get a token, use: `uv run langflow api_key`")
+            raise typer.Exit(1)
+
+        # Validate the auth token
+        try:
+            auth_user = None
+            async with session_scope() as session:
+                # Try JWT first
+                user = None
+                try:
+                    user = await get_current_user_by_jwt(auth_token, session)
+                except (JWTError, HTTPException):
+                    # Try API key
+                    api_key_result = await check_key(session, auth_token)
+                    if api_key_result and hasattr(api_key_result, "is_superuser"):
+                        user = api_key_result
+                auth_user = user
+
+            if not auth_user or not auth_user.is_superuser:
+                typer.echo(
+                    "Error: Invalid token or insufficient privileges. Only superusers can create other superusers."
+                )
+                raise typer.Exit(1)
+        except typer.Exit:
+            raise  # Re-raise typer.Exit without wrapping
+        except Exception as e:  # noqa: BLE001
+            typer.echo(f"Error: Authentication failed - {e!s}")
+            raise typer.Exit(1) from None
+
+    # Auth complete, create the superuser
+    async with session_scope() as session:
+        from langflow.services.auth.utils import create_super_user
+
+        if await create_super_user(db=session, username=username, password=password):
+            # Verify that the superuser was created
+            from langflow.services.database.models.user.model import User
+
+            stmt = select(User).where(User.username == username)
+            created_user: User = (await session.exec(stmt)).first()
+            if created_user is None or not created_user.is_superuser:
                 typer.echo("Superuser creation failed.")
+                return
+            # Now create the first folder for the user
+            result = await get_or_create_default_folder(session, created_user.id)
+            if result:
+                typer.echo("Default folder created successfully.")
+            else:
+                msg = "Could not create default folder."
+                raise RuntimeError(msg)
 
-    asyncio.run(_create_superuser())
+            # Log the superuser creation for audit purposes
+            logger.warning(
+                f"SECURITY AUDIT: New superuser '{username}' created via CLI command"
+                + (" by authenticated user" if auth_token else " (first-time setup)")
+            )
+            typer.echo("Superuser created successfully.")
+
+        else:
+            logger.error(f"SECURITY AUDIT: Failed attempt to create superuser '{username}' via CLI")
+            typer.echo("Superuser creation failed.")
 
 
 # command to copy the langflow database from the cache to the current directory
@@ -749,6 +848,7 @@ async def aapi_key():
         settings_service = get_settings_service()
         auth_settings = settings_service.auth_settings
         if not auth_settings.AUTO_LOGIN:
+            # TODO: Allow non-auto-login users to create API keys via CLI
             typer.echo("Auto login is disabled. API keys cannot be created through the CLI.")
             return None
 

src/backend/base/langflow/services/database/models/user/crud.py

@@ -60,3 +60,10 @@ async def update_user_last_login_at(user_id: UUID, db: AsyncSession):
         return await update_user(user, user_data, db)
     except Exception as e:  # noqa: BLE001
         logger.error(f"Error updating user last login at: {e!s}")
+
+
+async def get_all_superusers(db: AsyncSession) -> list[User]:
+    """Get all superuser accounts from the database."""
+    stmt = select(User).where(User.is_superuser == True)  # noqa: E712
+    result = await db.exec(stmt)
+    return list(result.all())

src/backend/base/langflow/services/settings/auth.py

@@ -27,12 +27,25 @@ class AuthSettings(BaseSettings):
     API_KEY_ALGORITHM: str = "HS256"
     API_V1_STR: str = "/api/v1"
 
-    AUTO_LOGIN: bool = True
+    AUTO_LOGIN: bool = Field(
+        default=True,  # TODO: Set to False in v1.6
+        description=(
+            "Enable automatic login with default credentials. "
+            "SECURITY WARNING: This bypasses authentication and should only be used in development environments. "
+            "Set to False in production."
+        ),
+    )
     """If True, the application will attempt to log in automatically as a super user."""
     skip_auth_auto_login: bool = True
     """If True, the application will skip authentication when AUTO_LOGIN is enabled.
     This will be removed in v1.6"""
 
+    ENABLE_SUPERUSER_CLI: bool = Field(
+        default=True,
+        description="Allow creation of superusers via CLI. Set to False in production for security.",
+    )
+    """If True, allows creation of superusers via the CLI 'langflow superuser' command."""
+
     NEW_USER_IS_ACTIVE: bool = False
     SUPERUSER: str = DEFAULT_SUPERUSER
     SUPERUSER_PASSWORD: str = DEFAULT_SUPERUSER_PASSWORD

src/backend/base/langflow/services/utils.py

@@ -68,15 +68,20 @@ async def get_or_create_super_user(session: AsyncSession, username, password, is
     return await create_super_user(username, password, db=session)
 
 
-async def setup_superuser(settings_service, session: AsyncSession) -> None:
+async def setup_superuser(settings_service: SettingsService, session: AsyncSession) -> None:
     if settings_service.auth_settings.AUTO_LOGIN:
         logger.debug("AUTO_LOGIN is set to True. Creating default superuser.")
+        username = DEFAULT_SUPERUSER
+        password = DEFAULT_SUPERUSER_PASSWORD
     else:
         # Remove the default superuser if it exists
         await teardown_superuser(settings_service, session)
+        username = settings_service.auth_settings.SUPERUSER
+        password = settings_service.auth_settings.SUPERUSER_PASSWORD
 
-    username = settings_service.auth_settings.SUPERUSER
-    password = settings_service.auth_settings.SUPERUSER_PASSWORD
+    if not username or not password:
+        msg = "Username and password must be set"
+        raise ValueError(msg)
 
     is_default = (username == DEFAULT_SUPERUSER) and (password == DEFAULT_SUPERUSER_PASSWORD)