fix: superuser credential handling and AUTO_LOGIN security (#9542)

* Refactor superuser credential handling for security

* [autofix.ci] apply automated fixes

* refactor: enhance superuser credential handling in setup process (#9574)

* [autofix.ci] apply automated fixes

* refactor: streamline superuser flow tests and enhance error handling (#9577)

* [autofix.ci] apply automated fixes

* None Super user is not allowed hence for a valid string resetting it to

* None Super user is not allowed hence for a valid string resetting it to ""

* use secret str for password everywhere

* [autofix.ci] apply automated fixes

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Jordan Frazier <[email protected]>

src/backend/base/langflow/__main__.py

@@ -669,7 +669,7 @@ async def _create_superuser(username: str, password: str, auth_token: str | None
     if settings_service.auth_settings.AUTO_LOGIN:
         # Force default credentials for AUTO_LOGIN mode
         username = DEFAULT_SUPERUSER
-        password = DEFAULT_SUPERUSER_PASSWORD
+        password = DEFAULT_SUPERUSER_PASSWORD.get_secret_value()
     else:
         # Production mode - prompt for credentials if not provided
         if not username:
@@ -697,7 +697,7 @@ async def _create_superuser(username: str, password: str, auth_token: str | None
             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}")
+        # Do not echo the default password to avoid exposing it in logs.
     # AUTO_LOGIN is false - production mode
     elif is_first_setup:
         typer.echo("No superusers found. Creating first superuser...")

src/backend/base/langflow/initial_setup/setup.py

@@ -1002,12 +1002,16 @@ async def create_or_update_starter_projects(all_types_dict: dict) -> None:
                 await logger.adebug(f"Successfully created {successfully_created_projects} starter projects")
 
 
-async def initialize_super_user_if_needed() -> None:
+async def initialize_auto_login_default_superuser() -> None:
     settings_service = get_settings_service()
     if not settings_service.auth_settings.AUTO_LOGIN:
         return
-    username = settings_service.auth_settings.SUPERUSER
-    password = settings_service.auth_settings.SUPERUSER_PASSWORD
+    # In AUTO_LOGIN mode, always use the default credentials for initial bootstrapping
+    # without persisting the password in memory after setup.
+    from langflow.services.settings.constants import DEFAULT_SUPERUSER, DEFAULT_SUPERUSER_PASSWORD
+
+    username = DEFAULT_SUPERUSER
+    password = DEFAULT_SUPERUSER_PASSWORD.get_secret_value()
     if not username or not password:
         msg = "SUPERUSER and SUPERUSER_PASSWORD must be set in the settings if AUTO_LOGIN is true."
         raise ValueError(msg)

src/backend/base/langflow/main.py

@@ -26,7 +26,7 @@
 from langflow.api.v1.mcp_projects import init_mcp_servers
 from langflow.initial_setup.setup import (
     create_or_update_starter_projects,
-    initialize_super_user_if_needed,
+    initialize_auto_login_default_superuser,
     load_bundles_from_urls,
     load_flows_from_directory,
     sync_flows_from_fs,
@@ -144,9 +144,27 @@ async def lifespan(_app: FastAPI):
             setup_llm_caching()
             await logger.adebug(f"LLM caching setup in {asyncio.get_event_loop().time() - current_time:.2f}s")
 
+            if get_settings_service().auth_settings.AUTO_LOGIN:
+                current_time = asyncio.get_event_loop().time()
+                await logger.adebug("Initializing default super user")
+                await initialize_auto_login_default_superuser()
+                await logger.adebug(
+                    f"Default super user initialized in {asyncio.get_event_loop().time() - current_time:.2f}s"
+                )
+
             current_time = asyncio.get_event_loop().time()
+            await logger.adebug("Loading bundles")
+            temp_dirs, bundles_components_paths = await load_bundles_with_error_handling()
+            get_settings_service().settings.components_path.extend(bundles_components_paths)
+            await logger.adebug(f"Bundles loaded in {asyncio.get_event_loop().time() - current_time:.2f}s")
+
+            current_time = asyncio.get_event_loop().time()
+            await logger.adebug("Caching types")
+            all_types_dict = await get_and_cache_all_types_dict(get_settings_service())
+            await logger.adebug(f"Types cached in {asyncio.get_event_loop().time() - current_time:.2f}s")
+
             await logger.adebug("Initializing super user")
-            await initialize_super_user_if_needed()
+            await initialize_auto_login_default_superuser()
             await logger.adebug(f"Super user initialized in {asyncio.get_event_loop().time() - current_time:.2f}s")
 
             current_time = asyncio.get_event_loop().time()

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

@@ -317,9 +317,21 @@ async def create_super_user(
 
 async def create_user_longterm_token(db: AsyncSession) -> tuple[UUID, dict]:
     settings_service = get_settings_service()
+    if not settings_service.auth_settings.AUTO_LOGIN:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST, detail="Auto login required to create a long-term token"
+        )
 
+    # Prefer configured username; fall back to default or any existing superuser
+    # NOTE: This user name cannot be a dynamic current user name since it is only used when autologin is True
     username = settings_service.auth_settings.SUPERUSER
     super_user = await get_user_by_username(db, username)
+    if not super_user:
+        from langflow.services.database.models.user.crud import get_all_superusers
+
+        superusers = await get_all_superusers(db)
+        super_user = superusers[0] if superusers else None
+
     if not super_user:
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Super user hasn't been created")
     access_token_expires_longterm = timedelta(days=365)

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

@@ -48,7 +48,8 @@ class AuthSettings(BaseSettings):
 
     NEW_USER_IS_ACTIVE: bool = False
     SUPERUSER: str = DEFAULT_SUPERUSER
-    SUPERUSER_PASSWORD: str = DEFAULT_SUPERUSER_PASSWORD
+    # Store password as SecretStr to prevent accidental plaintext exposure
+    SUPERUSER_PASSWORD: SecretStr = Field(default=DEFAULT_SUPERUSER_PASSWORD)
 
     REFRESH_SAME_SITE: Literal["lax", "strict", "none"] = "none"
     """The SameSite attribute of the refresh token cookie."""
@@ -71,8 +72,8 @@ class AuthSettings(BaseSettings):
     model_config = SettingsConfigDict(validate_assignment=True, extra="ignore", env_prefix="LANGFLOW_")
 
     def reset_credentials(self) -> None:
-        self.SUPERUSER = DEFAULT_SUPERUSER
-        self.SUPERUSER_PASSWORD = DEFAULT_SUPERUSER_PASSWORD
+        # Preserve the configured username but scrub the password from memory to avoid plaintext exposure.
+        self.SUPERUSER_PASSWORD = SecretStr("")
 
     # If autologin is true, then we need to set the credentials to
     # the default values
@@ -81,15 +82,17 @@ def reset_credentials(self) -> None:
     @field_validator("SUPERUSER", "SUPERUSER_PASSWORD", mode="before")
     @classmethod
     def validate_superuser(cls, value, info):
+        # When AUTO_LOGIN is enabled, force superuser to use default values.
         if info.data.get("AUTO_LOGIN"):
-            if value != DEFAULT_SUPERUSER:
-                value = DEFAULT_SUPERUSER
-                logger.debug("Resetting superuser to default value")
-            if info.data.get("SUPERUSER_PASSWORD") != DEFAULT_SUPERUSER_PASSWORD:
-                info.data["SUPERUSER_PASSWORD"] = DEFAULT_SUPERUSER_PASSWORD
-                logger.debug("Resetting superuser password to default value")
-
-            return value
+            logger.debug("Auto login is enabled, forcing superuser to use default values")
+            if info.field_name == "SUPERUSER":
+                if value != DEFAULT_SUPERUSER:
+                    logger.debug("Resetting superuser to default value")
+                return DEFAULT_SUPERUSER
+            if info.field_name == "SUPERUSER_PASSWORD":
+                if value != DEFAULT_SUPERUSER_PASSWORD.get_secret_value():
+                    logger.debug("Resetting superuser password to default value")
+                return DEFAULT_SUPERUSER_PASSWORD
 
         return value
 

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

@@ -1,5 +1,8 @@
+from pydantic import SecretStr
+
 DEFAULT_SUPERUSER = "langflow"
-DEFAULT_SUPERUSER_PASSWORD = "langflow"  # noqa: S105
+DEFAULT_SUPERUSER_PASSWORD = SecretStr("langflow")
+
 VARIABLES_TO_GET_FROM_ENVIRONMENT = [
     "COMPOSIO_API_KEY",
     "OPENAI_API_KEY",

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

@@ -72,18 +72,20 @@ async def setup_superuser(settings_service: SettingsService, session: AsyncSessi
     if settings_service.auth_settings.AUTO_LOGIN:
         await logger.adebug("AUTO_LOGIN is set to True. Creating default superuser.")
         username = DEFAULT_SUPERUSER
-        password = DEFAULT_SUPERUSER_PASSWORD
+        password = DEFAULT_SUPERUSER_PASSWORD.get_secret_value()
     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
+        # If AUTO_LOGIN is disabled, attempt to use configured credentials
+        # or fall back to default credentials if none are provided.
+        username = settings_service.auth_settings.SUPERUSER or DEFAULT_SUPERUSER
+        password = (settings_service.auth_settings.SUPERUSER_PASSWORD or DEFAULT_SUPERUSER_PASSWORD).get_secret_value()
 
     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)
+    is_default = (username == DEFAULT_SUPERUSER) and (password == DEFAULT_SUPERUSER_PASSWORD.get_secret_value())
 
     try:
         user = await get_or_create_super_user(
@@ -96,6 +98,7 @@ async def setup_superuser(settings_service: SettingsService, session: AsyncSessi
         msg = "Could not create superuser. Please create a superuser manually."
         raise RuntimeError(msg) from exc
     finally:
+        # Scrub credentials from in-memory settings after setup
         settings_service.auth_settings.reset_credentials()
 
 

src/backend/tests/integration/components/mcp/test_mcp_superuser_flow.py

@@ -0,0 +1,23 @@
+import pytest
+from langflow.services.auth.utils import create_user_longterm_token
+from langflow.services.deps import get_db_service, get_settings_service
+from langflow.services.utils import initialize_services
+
+
+@pytest.mark.asyncio
+async def test_mcp_longterm_token_headless_superuser_integration():
+    """Integration-style check that without explicit credentials, AUTO_LOGIN=false path.
+
+    Creates a headless superuser via initialize_services and allows minting a long-term token.
+    """
+    settings = get_settings_service()
+    settings.auth_settings.AUTO_LOGIN = False
+    settings.auth_settings.SUPERUSER = ""
+    settings.auth_settings.SUPERUSER_PASSWORD = ""
+
+    await initialize_services()
+
+    async with get_db_service().with_session() as session:
+        user_id, tokens = await create_user_longterm_token(session)
+        assert user_id is not None
+        assert tokens.get("access_token")

src/backend/tests/performance/test_server_init.py

@@ -43,11 +43,11 @@ def test_setup_llm_caching():
 
 async def test_initialize_super_user():
     """Benchmark super user initialization."""
-    from langflow.initial_setup.setup import initialize_super_user_if_needed
+    from langflow.initial_setup.setup import initialize_auto_login_default_superuser
     from langflow.services.utils import initialize_services
 
     await initialize_services(fix_migration=False)
-    await initialize_super_user_if_needed()
+    await initialize_auto_login_default_superuser()
     settings_service = get_settings_service()
     assert "test_performance.db" in settings_service.settings.database_url
 

src/backend/tests/unit/api/v1/test_mcp_projects.py

@@ -1,8 +1,9 @@
+import asyncio
 from unittest.mock import AsyncMock, MagicMock, patch
 from uuid import uuid4
 
 import pytest
-from fastapi import status
+from fastapi import HTTPException, status
 from httpx import AsyncClient
 from langflow.api.v1.mcp_projects import (
     get_project_mcp_server,
@@ -11,12 +12,14 @@
     project_mcp_servers,
     project_sse_transports,
 )
-from langflow.services.auth.utils import get_password_hash
+from langflow.services.auth.utils import create_user_longterm_token, get_password_hash
 from langflow.services.database.models.flow import Flow
 from langflow.services.database.models.folder import Folder
-from langflow.services.database.models.user import User
-from langflow.services.deps import session_scope
+from langflow.services.database.models.user.model import User
+from langflow.services.deps import get_db_service, get_settings_service, session_scope
+from langflow.services.utils import initialize_services
 from mcp.server.sse import SseServerTransport
+from sqlmodel import select
 
 # Mark all tests in this module as asyncio
 pytestmark = pytest.mark.asyncio
@@ -62,6 +65,7 @@ async def __aenter__(self):
         return (MagicMock(), MagicMock())
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
+        # No teardown required for this mock context manager in tests
         pass
 
 
@@ -601,6 +605,8 @@ async def test_project_sse_creation(user_test_project):
 
     assert sse_transport2 is sse_transport
     assert mcp_server2 is mcp_server
+    # Yield control to the event loop to satisfy async usage in this test
+    await asyncio.sleep(0)
 
 
 async def test_init_mcp_servers(user_test_project, other_test_project):
@@ -652,3 +658,49 @@ def mock_get_project_sse(project_id):
     with patch("langflow.api.v1.mcp_projects.get_project_sse", side_effect=mock_get_project_sse):
         # This should not raise any exception, as the error should be caught
         await init_mcp_servers()
+
+
+@pytest.mark.asyncio
+async def test_mcp_longterm_token_fails_without_superuser():
+    """When AUTO_LOGIN is false and no superuser exists, creating a long-term token should raise 400.
+
+    This simulates a clean DB with AUTO_LOGIN disabled and without provisioning a superuser.
+    """
+    settings_service = get_settings_service()
+    settings_service.auth_settings.AUTO_LOGIN = False
+
+    # Ensure no superuser exists in DB
+    async with get_db_service().with_session() as session:
+        result = await session.exec(select(User).where(User.is_superuser == True))  # noqa: E712
+        users = result.all()
+        for user in users:
+            await session.delete(user)
+        await session.commit()
+
+    # Now attempt to create long-term token -> expect HTTPException 400
+    async with get_db_service().with_session() as session:
+        with pytest.raises(HTTPException, match="Super user hasn't been created"):
+            await create_user_longterm_token(session)
+
+
+@pytest.mark.asyncio
+async def test_mcp_longterm_token_succeeds_with_headless_fallback():
+    """When AUTO_LOGIN is false and no credentials are provided.
+
+    The headless fallback should create an internal superuser so MCP can mint a token without raising 400.
+    """
+    settings_service = get_settings_service()
+    settings_service.auth_settings.AUTO_LOGIN = False
+    # Clear any configured credentials
+    settings_service.auth_settings.SUPERUSER = ""
+    settings_service.auth_settings.SUPERUSER_PASSWORD = ""  # SecretStr handled in service
+
+    # Re-initialize core services which now create the headless superuser fallback
+    await initialize_services()
+
+    # Should now be able to create a long-term token
+    async with get_db_service().with_session() as session:
+        user_id, tokens = await create_user_longterm_token(session)
+        assert user_id is not None
+        assert tokens.get("access_token")
+        assert tokens.get("token_type") == "bearer"

src/backend/tests/unit/test_auth_settings.py

@@ -0,0 +1,50 @@
+from pathlib import Path
+
+import pytest
+from langflow.services.settings.auth import AuthSettings
+from langflow.services.settings.constants import DEFAULT_SUPERUSER, DEFAULT_SUPERUSER_PASSWORD
+from pydantic import SecretStr
+
+
+@pytest.mark.parametrize("auto_login", [True, False])
+def test_superuser_password_is_secretstr(auto_login, tmp_path: Path):
+    cfg_dir = tmp_path.as_posix()
+    settings = AuthSettings(CONFIG_DIR=cfg_dir, AUTO_LOGIN=auto_login)
+    assert isinstance(settings.SUPERUSER_PASSWORD, SecretStr)
+
+
+def test_auto_login_true_forces_default_and_scrubs_password(tmp_path: Path):
+    cfg_dir = tmp_path.as_posix()
+    settings = AuthSettings(
+        CONFIG_DIR=cfg_dir,
+        AUTO_LOGIN=True,
+        SUPERUSER="custom",
+        SUPERUSER_PASSWORD=DEFAULT_SUPERUSER_PASSWORD + "_changed",
+    )
+    # Validator forces default username and scrubs password
+    assert settings.SUPERUSER == DEFAULT_SUPERUSER
+    assert isinstance(settings.SUPERUSER_PASSWORD, SecretStr)
+    assert settings.SUPERUSER_PASSWORD.get_secret_value() == ""
+
+    # reset_credentials keeps default username (AUTO_LOGIN on) and keeps password scrubbed
+    settings.reset_credentials()
+    assert settings.SUPERUSER == DEFAULT_SUPERUSER
+    assert settings.SUPERUSER_PASSWORD.get_secret_value() == ""
+
+
+def test_auto_login_false_preserves_username_and_scrubs_password_on_reset(tmp_path: Path):
+    cfg_dir = tmp_path.as_posix()
+    settings = AuthSettings(
+        CONFIG_DIR=cfg_dir,
+        AUTO_LOGIN=False,
+        SUPERUSER="admin",
+        SUPERUSER_PASSWORD="strongpass",  # noqa: S106
+    )
+    # Values preserved at init
+    assert settings.SUPERUSER == "admin"
+    assert settings.SUPERUSER_PASSWORD.get_secret_value() == "strongpass"
+
+    # After reset, username preserved, password scrubbed
+    settings.reset_credentials()
+    assert settings.SUPERUSER == "admin"
+    assert settings.SUPERUSER_PASSWORD.get_secret_value() == ""

src/backend/tests/unit/test_initial_setup.py

@@ -243,7 +243,11 @@ async def test_load_bundles_from_urls():
     async with session_scope() as session:
         await create_super_user(
             username=settings_service.auth_settings.SUPERUSER,
-            password=settings_service.auth_settings.SUPERUSER_PASSWORD,
+            password=(
+                settings_service.auth_settings.SUPERUSER_PASSWORD.get_secret_value()
+                if hasattr(settings_service.auth_settings.SUPERUSER_PASSWORD, "get_secret_value")
+                else settings_service.auth_settings.SUPERUSER_PASSWORD
+            ),
             db=session,
         )