fix: enforce authentication for superuser cli command (#9152)
* Enforce authentication for superuser cli command * shorten security md * cleanup * use session_scope * re-add uvlock * [autofix.ci] apply automated fixes * ruff * update env example * [autofix.ci] apply automated fixes * better exception handling * [autofix.ci] apply automated fixes * update tests to not use mocks * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes * Remove old test * Catch exceptions for typer * Try output instead of stdout * Use xdist to run in serial * Separate create superuse * [autofix.ci] apply automated fixes * Ruff * [autofix.ci] apply automated fixes * lint --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
parent
ede849aaf7
commit
c188ec113c
8 changed files with 353 additions and 112 deletions
|
|
@ -1,9 +1,11 @@
|
|||
import socket
|
||||
import threading
|
||||
import time
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from langflow.__main__ import app
|
||||
import typer
|
||||
from langflow.__main__ import _create_superuser, app
|
||||
from langflow.services import deps
|
||||
|
||||
|
||||
|
|
@ -57,7 +59,89 @@ def test_components_path(runner, default_settings, tmp_path):
|
|||
assert str(temp_dir) in settings_service.settings.components_path
|
||||
|
||||
|
||||
def test_superuser(runner):
|
||||
result = runner.invoke(app, ["superuser"], input="admin\nadmin\n")
|
||||
assert result.exit_code == 0, result.stdout
|
||||
assert "Superuser created successfully." in result.stdout
|
||||
@pytest.mark.xdist_group(name="serial-superuser-tests")
|
||||
class TestSuperuserCommand:
|
||||
"""Deterministic tests for the superuser CLI command."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_additional_superuser_requires_auth_production(self, client, active_super_user): # noqa: ARG002
|
||||
"""Test additional superuser creation requires authentication in production."""
|
||||
# We already have active_super_user from the fixture, so we're not in first setup
|
||||
with (
|
||||
patch("langflow.services.deps.get_settings_service") as mock_settings,
|
||||
patch("langflow.__main__.get_settings_service") as mock_settings2,
|
||||
):
|
||||
# Configure settings for production mode (AUTO_LOGIN=False)
|
||||
mock_auth_settings = type("MockAuthSettings", (), {"AUTO_LOGIN": False, "ENABLE_SUPERUSER_CLI": True})()
|
||||
mock_settings.return_value.auth_settings = mock_auth_settings
|
||||
mock_settings2.return_value.auth_settings = mock_auth_settings
|
||||
|
||||
# Try to create a superuser without auth - should fail
|
||||
with pytest.raises(typer.Exit) as exc_info:
|
||||
await _create_superuser("newuser", "newpass", None)
|
||||
|
||||
assert exc_info.value.exit_code == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_additional_superuser_blocked_in_auto_login_mode(self, client, active_super_user): # noqa: ARG002
|
||||
"""Test additional superuser creation blocked when AUTO_LOGIN=true."""
|
||||
# We already have active_super_user from the fixture, so we're not in first setup
|
||||
with (
|
||||
patch("langflow.services.deps.get_settings_service") as mock_settings,
|
||||
patch("langflow.__main__.get_settings_service") as mock_settings2,
|
||||
):
|
||||
# Configure settings for AUTO_LOGIN mode
|
||||
mock_auth_settings = type("MockAuthSettings", (), {"AUTO_LOGIN": True, "ENABLE_SUPERUSER_CLI": True})()
|
||||
mock_settings.return_value.auth_settings = mock_auth_settings
|
||||
mock_settings2.return_value.auth_settings = mock_auth_settings
|
||||
|
||||
# Try to create a superuser - should fail
|
||||
with pytest.raises(typer.Exit) as exc_info:
|
||||
await _create_superuser("newuser", "newpass", None)
|
||||
|
||||
assert exc_info.value.exit_code == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cli_disabled_blocks_creation(self, client): # noqa: ARG002
|
||||
"""Test ENABLE_SUPERUSER_CLI=false blocks superuser creation."""
|
||||
with (
|
||||
patch("langflow.services.deps.get_settings_service") as mock_settings,
|
||||
patch("langflow.__main__.get_settings_service") as mock_settings2,
|
||||
):
|
||||
mock_auth_settings = type("MockAuthSettings", (), {"AUTO_LOGIN": True, "ENABLE_SUPERUSER_CLI": False})()
|
||||
mock_settings.return_value.auth_settings = mock_auth_settings
|
||||
mock_settings2.return_value.auth_settings = mock_auth_settings
|
||||
|
||||
# Try to create a superuser - should fail
|
||||
with pytest.raises(typer.Exit) as exc_info:
|
||||
await _create_superuser("admin", "password", None)
|
||||
|
||||
assert exc_info.value.exit_code == 1
|
||||
|
||||
@pytest.mark.skip(reason="Skip -- default superuser is created by initialize_services() function")
|
||||
@pytest.mark.asyncio
|
||||
async def test_auto_login_forces_default_credentials(self, client):
|
||||
"""Test AUTO_LOGIN=true forces default credentials."""
|
||||
# Since client fixture already creates default user, we need to test in a clean DB scenario
|
||||
# But that's why this test is skipped - the behavior is already handled by initialize_services
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failed_auth_token_validation(self, client, active_super_user): # noqa: ARG002
|
||||
"""Test failed superuser creation with invalid auth token."""
|
||||
# We already have active_super_user from the fixture, so we're not in first setup
|
||||
with (
|
||||
patch("langflow.services.deps.get_settings_service") as mock_settings,
|
||||
patch("langflow.__main__.get_settings_service") as mock_settings2,
|
||||
patch("langflow.__main__.get_current_user_by_jwt", side_effect=Exception("Invalid token")),
|
||||
patch("langflow.__main__.check_key", return_value=None),
|
||||
):
|
||||
# Configure settings for production mode (AUTO_LOGIN=False)
|
||||
mock_auth_settings = type("MockAuthSettings", (), {"AUTO_LOGIN": False, "ENABLE_SUPERUSER_CLI": True})()
|
||||
mock_settings.return_value.auth_settings = mock_auth_settings
|
||||
mock_settings2.return_value.auth_settings = mock_auth_settings
|
||||
|
||||
# Try to create a superuser with invalid token - should fail
|
||||
with pytest.raises(typer.Exit) as exc_info:
|
||||
await _create_superuser("newuser", "newpass", "invalid-token")
|
||||
|
||||
assert exc_info.value.exit_code == 1
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue