Skip to content
Merged
Show file tree
Hide file tree
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
4 changes: 2 additions & 2 deletions src/backend/base/langflow/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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...")
Expand Down
10 changes: 7 additions & 3 deletions src/backend/base/langflow/initial_setup/setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
22 changes: 20 additions & 2 deletions src/backend/base/langflow/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -141,9 +141,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()
Expand Down
12 changes: 12 additions & 0 deletions src/backend/base/langflow/services/auth/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think we should get a non-default superuser here. As you said, this method should only be used with AUTO_LOGIN is true, meaning the default superuser must exist and be used.

Copy link
Collaborator Author

@edwinjosechittilappilly edwinjosechittilappilly Sep 5, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This function "create_user_longterm_token" is called only when auto login is true
@coderabbitai please check this.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧩 Analysis chain

@edwinjosechittilappilly I'll check where the create_user_longterm_token function is called to verify if it's only used when auto login is true.


🏁 Script executed:

#!/bin/bash
# Search for all calls to create_user_longterm_token function
rg -nP --type=py -A5 -B5 'create_user_longterm_token\(' 

Length of output: 4943


@edwinjosechittilappilly You are correct! After checking the codebase, I can confirm that create_user_longterm_token is only called when AUTO_LOGIN is true.

The main production usage is in src/backend/base/langflow/api/v1/login.py at line 85, inside the auto_login endpoint, which has an explicit check:

if auth_settings.AUTO_LOGIN:
    user_id, tokens = await create_user_longterm_token(db)

However, @jordanrfrazier raises a valid design point. Since this function is only used with AUTO_LOGIN=true, the question becomes: should auto-login fall back to any existing superuser, or should it strictly require the configured superuser to exist?

The current change allows fallback to any superuser when the configured one doesn't exist. This could be helpful for backward compatibility, but it might also mask configuration issues where the intended superuser wasn't created properly.

What's the intended behavior here? Should auto-login:

  1. Strict approach: Only work with the specifically configured superuser
  2. Fallback approach: Use any available superuser if the configured one doesn't exist

This design decision will help determine if the fallback logic should remain.


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)
Expand Down
25 changes: 14 additions & 11 deletions src/backend/base/langflow/services/settings/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand All @@ -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("")
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I actually think reset_credentials should be renamed to clear_superuser_credentials and just set SUPERUSER and SUPERUSER_PASSWORD to None. Any reason why we aren't doing that?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe we need a user at any point. @coderabbitai can you suggest ?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧩 Analysis chain

Let me analyze the authentication flow and how the reset_credentials method is used to provide a well-informed suggestion.

@edwinjosechittilappilly, based on my analysis of the codebase, here's my suggestion regarding the reset_credentials method:

Current Implementation Analysis

The current reset_credentials method serves a security purpose - it clears sensitive credential data from memory while maintaining the system's ability to function. Here's what it does:

  1. Conditionally preserves username: When AUTO_LOGIN is enabled, it keeps SUPERUSER as the default value
  2. Always clears password: Sets SUPERUSER_PASSWORD to an empty SecretStr("")

Recommendation: Keep Current Approach with Minor Refinements

I recommend keeping the current implementation rather than setting values to None for these reasons:

Why Not Set to None:

  1. Authentication Flow Dependencies: The system expects string values, not None. Setting to None could break authentication logic that doesn't handle null checks
  2. AUTO_LOGIN Functionality: When AUTO_LOGIN=True, the system needs valid default credentials to function
  3. Field Validators: The existing validators expect string/SecretStr types, not None

Suggested Improvements:

def clear_sensitive_credentials(self) -> None:
    """
    Clear sensitive credential data from memory while preserving system functionality.
    
    For AUTO_LOGIN workflows, maintains the default username but clears the password.
    For manual auth workflows, preserves the configured username but clears the password.
    """
    # For AUTO_LOGIN workflows, keep username at default. Otherwise, preserve the
    # configured username but scrub the password from memory to avoid plaintext exposure.
    if self.AUTO_LOGIN:
        self.SUPERUSER = DEFAULT_SUPERUSER
    # Always clear password from memory for security
    self.SUPERUSER_PASSWORD = SecretStr("")

Key Benefits of Current Approach:

  • Security: Removes sensitive password data from memory
  • Functionality: Maintains system operability
  • AUTO_LOGIN Support: Preserves automatic authentication capability
  • Type Safety: Maintains expected data types

The current implementation strikes the right balance between security (clearing sensitive data) and functionality (maintaining a usable authentication state). The method name could be improved to clear_sensitive_credentials to better reflect its purpose, but the core logic should remain as implemented.


🏁 Script executed:

#!/bin/bash
# Find usages of reset_credentials method
echo "=== Searching for reset_credentials usage ==="
rg -n --type=py "reset_credentials" -A 3 -B 3

echo -e "\n=== Searching for DEFAULT_SUPERUSER and DEFAULT_SUPERUSER_PASSWORD usage ==="
rg -n --type=py "DEFAULT_SUPERUSER" -A 2 -B 2

echo -e "\n=== Looking for authentication/login related code ==="
rg -n --type=py -g "!**/test/**" "SUPERUSER.*None|superuser.*none" -i -A 2 -B 2

Length of output: 29915


🏁 Script executed:

#!/bin/bash
# Look at auth-related files to understand the login flow
echo "=== Finding auth/login related files ==="
fd -t f -e py . | rg -i "auth|login|user" | head -10

echo -e "\n=== Checking how AUTO_LOGIN works ==="
rg -n --type=py "AUTO_LOGIN" -A 5 -B 2

echo -e "\n=== Looking for user authentication/validation logic ==="
rg -n --type=py -g "!**/test/**" "def.*login|def.*authenticate" -A 5

Length of output: 67248


🏁 Script executed:

#!/bin/bash
# Check constants file and see if there are any None checks in auth flow
echo "=== Checking constants for DEFAULT values ==="
cat src/backend/base/langflow/services/settings/constants.py 2>/dev/null || echo "Constants file not found in expected location"

echo -e "\n=== Finding constants file ==="
fd constants.py

echo -e "\n=== Searching for None checks in authentication ==="
rg -n --type=py -g "!**/test/**" "is None|== None" -A 2 -B 2 | rg -i -A 4 -B 4 "user|auth|superuser"

Length of output: 35579


🏁 Script executed:

#!/bin/bash
# Look at the constants file
echo "=== Constants file content ==="
cat src/backend/base/langflow/services/settings/constants.py

echo -e "\n=== Check authentication service implementation ==="
fd -t f auth.py | xargs -I {} sh -c 'echo "=== {} ===" && head -50 {}'

Length of output: 3179


🏁 Script executed:

#!/bin/bash
# Look for authentication/user management logic
echo "=== Searching for user creation/authentication logic ==="
rg -n --type=py -g "!**/test/**" "create.*user|get.*user.*by" -A 3 -B 1

echo -e "\n=== Check how superuser credentials are used ==="
rg -n --type=py -g "!**/test/**" "SUPERUSER_PASSWORD\.get_secret_value|superuser.*password" -A 3 -B 1

Length of output: 94234


🏁 Script executed:

#!/bin/bash
# Find the authentication manager or service
echo "=== Finding auth manager/service files ==="
fd -t f -e py . | rg "auth.*manager|auth.*service|user.*service" 

echo -e "\n=== Look for files that might handle user authentication ==="
rg -l --type=py "authenticate|login" | head -5 | xargs -I {} sh -c 'echo "=== {} ===" && head -30 {}'

Length of output: 5648


@edwinjosechittilappilly, after analyzing the authentication flow and codebase thoroughly, I agree with your assessment that "we need a user at any point." Here's my recommendation:

Keep the Current Implementation ✅

The current reset_credentials method is actually well-designed and should not set values to None. Here's why:

1. System Dependencies Require Valid Users

The codebase shows numerous places where the system expects valid user references:

  • Authentication flows use settings.auth_settings.SUPERUSER for user lookup
  • AUTO_LOGIN functionality depends on having a valid default username
  • Flow operations, API endpoints, and services all expect user context

2. Setting to None Would Break Core Functionality

# This would break these critical operations:
username = settings_service.auth_settings.SUPERUSER or DEFAULT_SUPERUSER
super_user = await get_user_by_username(db, username)  # Would fail if SUPERUSER is None

# AUTO_LOGIN workflows would fail:
if settings_service.auth_settings.SUPERUSER:  # Would be False if None
    result = await get_user_by_username(db, settings_service.auth_settings.SUPERUSER)

3. Current Security Benefits Are Preserved

The method already achieves the security goal:

  • Clears sensitive password from memory: SUPERUSER_PASSWORD = SecretStr("")
  • Maintains system operability: Preserves username for authentication flows
  • Conditional logic: Only resets username to default when AUTO_LOGIN=True

4. Type System Expectations

The validators and type system expect:

  • SUPERUSER: str (not str | None)
  • SUPERUSER_PASSWORD: SecretStr (not SecretStr | None)

Suggested Improvement: Better Method Name

Instead of changing the logic, consider renaming for clarity:

def clear_sensitive_credentials(self) -> None:
    """
    Clear sensitive credential data from memory while preserving system functionality.
    
    For AUTO_LOGIN workflows, maintains the default username but clears the password.
    For manual auth workflows, preserves the configured username but clears the password.
    """
    if self.AUTO_LOGIN:
        self.SUPERUSER = DEFAULT_SUPERUSER
    # Always clear password from memory for security
    self.SUPERUSER_PASSWORD = SecretStr("")

Why This Approach Works Best

  1. Security: ✅ Removes sensitive password data
  2. Functionality: ✅ System remains operational
  3. AUTO_LOGIN Support: ✅ Automatic authentication still works
  4. Type Safety: ✅ No breaking changes to type system
  5. Backward Compatibility: ✅ All existing code continues to work

The current implementation strikes the perfect balance between security (clearing secrets) and functionality (maintaining system operability). Setting values to None would be a breaking change that compromises the system's ability to authenticate users.


# If autologin is true, then we need to set the credentials to
# the default values
Expand All @@ -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

Expand Down
5 changes: 4 additions & 1 deletion src/backend/base/langflow/services/settings/constants.py
Original file line number Diff line number Diff line change
@@ -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",
Expand Down
11 changes: 7 additions & 4 deletions src/backend/base/langflow/services/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the auth_settings.SUPERUSER will default to DEFAULT_SUPERUSER; no need for the or.

    SUPERUSER: str = DEFAULT_SUPERUSER
    SUPERUSER_PASSWORD: str = DEFAULT_SUPERUSER_PASSWORD

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ideal case yes but if the user defined an empty user name in settings it might override the default to empty I believe right?

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the user defined an empty user name in the settings we should catch that as a user-error and raise that, not assume that we should use the 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(
Expand All @@ -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
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Elaborate on this comment -- add a note saying something like:

Prevents users from gaining access to admin credentials by accessing the SUPERUSER environment variables. 

settings_service.auth_settings.reset_credentials()
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

reset_credentials resets the values to langflow / langflow. Maybe we should just clear them completely instead?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

def reset_credentials(self) -> None:
    # For AUTO_LOGIN workflows, keep username at default. Otherwise, preserve the
    # configured username but scrub the password from memory to avoid plaintext exposure.
    if self.AUTO_LOGIN:
        self.SUPERUSER = DEFAULT_SUPERUSER
    self.SUPERUSER_PASSWORD = SecretStr("")
    
    
    It resets to empty string, i think it should reset to default password ideally isnt when autologin is True.
    
    @jordanrfrazier  what do you suggest?

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See other comment -- suggest clearing them completely



Expand Down
Original file line number Diff line number Diff line change
@@ -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")
4 changes: 2 additions & 2 deletions src/backend/tests/performance/test_server_init.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
60 changes: 56 additions & 4 deletions src/backend/tests/unit/api/v1/test_mcp_projects.py
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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
Expand Down Expand Up @@ -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


Expand Down Expand Up @@ -546,6 +550,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):
Expand Down Expand Up @@ -597,3 +603,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"
50 changes: 50 additions & 0 deletions src/backend/tests/unit/test_auth_settings.py
Original file line number Diff line number Diff line change
@@ -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() == ""
6 changes: 5 additions & 1 deletion src/backend/tests/unit/test_initial_setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)

Expand Down
Loading