feat: Add configurable API key validation source (db/env) (#10783)

* add xapikey to env authentication

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* remove-docs-for-1.8-release

* add fallback to db

* [autofix.ci] apply automated fixes

* test(api_key_source.py): enhance tests for check_key function to cover more scenarios and improve reliability

- Add tests to verify routing to environment and database based on API_KEY_SOURCE.
- Implement fallback logic tests when environment validation fails.
- Ensure correct behavior when both environment and database validations fail.
- Refactor existing tests to improve clarity and coverage of edge cases.

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes

* fix tests

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Mendon Kissling <59585235+mendonk@users.noreply.github.com>

.env.example

@@ -116,6 +116,21 @@ LANGFLOW_SUPERUSER=
 # Example: LANGFLOW_SUPERUSER_PASSWORD=123456
 LANGFLOW_SUPERUSER_PASSWORD=
 
+# API Key Source
+# Controls how API keys are validated for the x-api-key header
+# Values: db, env
+# - db (default): Validates against API keys stored in the database
+# - env: Validates against the LANGFLOW_API_KEY environment variable
+# Example: LANGFLOW_API_KEY_SOURCE=db
+LANGFLOW_API_KEY_SOURCE=
+
+# API Key (only used when LANGFLOW_API_KEY_SOURCE=env)
+# The API key to use for authentication when API_KEY_SOURCE is set to 'env'
+# This allows injecting a pre-defined API key via environment variables
+# (useful for Kubernetes Secrets, CI/CD pipelines, etc.)
+# Example: LANGFLOW_API_KEY=your-secure-api-key
+LANGFLOW_API_KEY=
+
 # Should store environment variables in the database
 # Values: true, false
 LANGFLOW_STORE_ENVIRONMENT_VARIABLES=

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

@@ -1,4 +1,5 @@
 import datetime
+import os
 import secrets
 from typing import TYPE_CHECKING
 from uuid import UUID
@@ -49,11 +50,29 @@ async def delete_api_key(session: AsyncSession, api_key_id: UUID) -> None:
 
 
 async def check_key(session: AsyncSession, api_key: str) -> User | None:
-    """Check if the API key is valid."""
+    """Check if the API key is valid.
+
+    Validates API keys based on the LANGFLOW_API_KEY_SOURCE setting:
+    - 'db': Validates against database-stored API keys (default)
+    - 'env': Validates against the LANGFLOW_API_KEY environment variable,
+             falls back to database if env validation fails
+    """
+    settings_service = get_settings_service()
+    api_key_source = settings_service.auth_settings.API_KEY_SOURCE
+
+    if api_key_source == "env":
+        user = await _check_key_from_env(session, api_key, settings_service)
+        if user is not None:
+            return user
+        # Fallback to database if env validation fails
+    return await _check_key_from_db(session, api_key, settings_service)
+
+
+async def _check_key_from_db(session: AsyncSession, api_key: str, settings_service) -> User | None:
+    """Validate API key against the database."""
     query: SelectOfScalar = select(ApiKey).options(selectinload(ApiKey.user)).where(ApiKey.api_key == api_key)
     api_key_object: ApiKey | None = (await session.exec(query)).first()
     if api_key_object is not None:
-        settings_service = get_settings_service()
         if settings_service.settings.disable_track_apikey_usage is not True:
             api_key_object.total_uses += 1
             api_key_object.last_used_at = datetime.datetime.now(datetime.timezone.utc)
@@ -61,3 +80,27 @@ async def check_key(session: AsyncSession, api_key: str) -> User | None:
             await session.flush()
         return api_key_object.user
     return None
+
+
+async def _check_key_from_env(session: AsyncSession, api_key: str, settings_service) -> User | None:
+    """Validate API key against the environment variable.
+
+    When API_KEY_SOURCE='env', the x-api-key header is validated against
+    LANGFLOW_API_KEY environment variable. If valid, returns the superuser for authorization.
+    """
+    from langflow.services.database.models.user.crud import get_user_by_username
+
+    env_api_key = os.getenv("LANGFLOW_API_KEY")
+    if not env_api_key:
+        return None
+
+    # Compare the provided API key with the environment variable
+    if api_key != env_api_key:
+        return None
+
+    # Return the superuser for authorization purposes
+    superuser_username = settings_service.auth_settings.SUPERUSER
+    user = await get_user_by_username(session, superuser_username)
+    if user and user.is_active:
+        return user
+    return None

src/backend/tests/unit/test_api_key_source.py

@@ -0,0 +1,579 @@
+"""Tests for API key validation with different sources (db and env).
+
+This module tests the check_key function behavior when:
+- API_KEY_SOURCE='db' (default): Validates against database-stored API keys
+- API_KEY_SOURCE='env': Validates against LANGFLOW_API_KEY environment variable
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+from uuid import uuid4
+
+import pytest
+from langflow.services.database.models.api_key.crud import (
+    _check_key_from_db,
+    _check_key_from_env,
+    check_key,
+)
+from langflow.services.database.models.user.model import User
+
+
+@pytest.fixture
+def mock_user():
+    """Create a mock active user."""
+    user = MagicMock(spec=User)
+    user.id = uuid4()
+    user.username = "testuser"
+    user.is_active = True
+    user.is_superuser = False
+    return user
+
+
+@pytest.fixture
+def mock_superuser():
+    """Create a mock active superuser."""
+    user = MagicMock(spec=User)
+    user.id = uuid4()
+    user.username = "langflow"
+    user.is_active = True
+    user.is_superuser = True
+    return user
+
+
+@pytest.fixture
+def mock_inactive_user():
+    """Create a mock inactive user."""
+    user = MagicMock(spec=User)
+    user.id = uuid4()
+    user.username = "inactive"
+    user.is_active = False
+    user.is_superuser = False
+    return user
+
+
+@pytest.fixture
+def mock_session():
+    """Create a mock async database session."""
+    return AsyncMock()
+
+
+@pytest.fixture
+def mock_settings_service_db():
+    """Create a mock settings service with API_KEY_SOURCE='db'."""
+    settings_service = MagicMock()
+    settings_service.auth_settings.API_KEY_SOURCE = "db"
+    settings_service.auth_settings.SUPERUSER = "langflow"
+    settings_service.settings.disable_track_apikey_usage = False
+    return settings_service
+
+
+@pytest.fixture
+def mock_settings_service_env():
+    """Create a mock settings service with API_KEY_SOURCE='env'."""
+    settings_service = MagicMock()
+    settings_service.auth_settings.API_KEY_SOURCE = "env"
+    settings_service.auth_settings.SUPERUSER = "langflow"
+    settings_service.settings.disable_track_apikey_usage = False
+    return settings_service
+
+
+# ============================================================================
+# check_key routing tests
+# ============================================================================
+
+
+class TestCheckKeyRouting:
+    """Tests for check_key routing based on API_KEY_SOURCE setting."""
+
+    @pytest.mark.asyncio
+    async def test_check_key_routes_to_db_by_default(self, mock_session, mock_settings_service_db):
+        """check_key should route to _check_key_from_db when API_KEY_SOURCE='db'."""
+        with (
+            patch(
+                "langflow.services.database.models.api_key.crud.get_settings_service",
+                return_value=mock_settings_service_db,
+            ),
+            patch(
+                "langflow.services.database.models.api_key.crud._check_key_from_db",
+                new_callable=AsyncMock,
+            ) as mock_db_check,
+            patch(
+                "langflow.services.database.models.api_key.crud._check_key_from_env",
+                new_callable=AsyncMock,
+            ) as mock_env_check,
+        ):
+            mock_db_check.return_value = None
+
+            await check_key(mock_session, "sk-test-key")
+
+            mock_db_check.assert_called_once()
+            mock_env_check.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_check_key_routes_to_env_when_configured_and_succeeds(self, mock_session, mock_settings_service_env):
+        """check_key should route to _check_key_from_env when API_KEY_SOURCE='env' and env succeeds."""
+        mock_user = MagicMock(spec=User)
+        with (
+            patch(
+                "langflow.services.database.models.api_key.crud.get_settings_service",
+                return_value=mock_settings_service_env,
+            ),
+            patch(
+                "langflow.services.database.models.api_key.crud._check_key_from_db",
+                new_callable=AsyncMock,
+            ) as mock_db_check,
+            patch(
+                "langflow.services.database.models.api_key.crud._check_key_from_env",
+                new_callable=AsyncMock,
+            ) as mock_env_check,
+        ):
+            mock_env_check.return_value = mock_user
+
+            result = await check_key(mock_session, "sk-test-key")
+
+            mock_env_check.assert_called_once()
+            mock_db_check.assert_not_called()
+            assert result == mock_user
+
+    @pytest.mark.asyncio
+    async def test_check_key_falls_back_to_db_when_env_fails(self, mock_session, mock_settings_service_env):
+        """check_key should fallback to _check_key_from_db when API_KEY_SOURCE='env' but env validation fails."""
+        mock_user = MagicMock(spec=User)
+        with (
+            patch(
+                "langflow.services.database.models.api_key.crud.get_settings_service",
+                return_value=mock_settings_service_env,
+            ),
+            patch(
+                "langflow.services.database.models.api_key.crud._check_key_from_db",
+                new_callable=AsyncMock,
+            ) as mock_db_check,
+            patch(
+                "langflow.services.database.models.api_key.crud._check_key_from_env",
+                new_callable=AsyncMock,
+            ) as mock_env_check,
+        ):
+            mock_env_check.return_value = None  # env validation fails
+            mock_db_check.return_value = mock_user  # db has the key
+
+            result = await check_key(mock_session, "sk-test-key")
+
+            mock_env_check.assert_called_once()
+            mock_db_check.assert_called_once()  # Should fallback to db
+            assert result == mock_user
+
+    @pytest.mark.asyncio
+    async def test_check_key_returns_none_when_both_env_and_db_fail(self, mock_session, mock_settings_service_env):
+        """check_key should return None when both env and db validation fail."""
+        with (
+            patch(
+                "langflow.services.database.models.api_key.crud.get_settings_service",
+                return_value=mock_settings_service_env,
+            ),
+            patch(
+                "langflow.services.database.models.api_key.crud._check_key_from_db",
+                new_callable=AsyncMock,
+            ) as mock_db_check,
+            patch(
+                "langflow.services.database.models.api_key.crud._check_key_from_env",
+                new_callable=AsyncMock,
+            ) as mock_env_check,
+        ):
+            mock_env_check.return_value = None  # env validation fails
+            mock_db_check.return_value = None  # db validation also fails
+
+            result = await check_key(mock_session, "sk-test-key")
+
+            mock_env_check.assert_called_once()
+            mock_db_check.assert_called_once()
+            assert result is None
+
+
+# ============================================================================
+# _check_key_from_db tests
+# ============================================================================
+
+
+class TestCheckKeyFromDb:
+    """Tests for database-based API key validation."""
+
+    @pytest.mark.asyncio
+    async def test_valid_key_returns_user(self, mock_session, mock_user, mock_settings_service_db):
+        """Valid API key should return the associated user."""
+        mock_api_key = MagicMock()
+        mock_api_key.user = mock_user
+        mock_api_key.total_uses = 0
+
+        mock_result = MagicMock()
+        mock_result.first.return_value = mock_api_key
+        mock_session.exec.return_value = mock_result
+
+        result = await _check_key_from_db(mock_session, "sk-valid-key", mock_settings_service_db)
+
+        assert result == mock_user
+        assert mock_api_key.total_uses == 1
+
+    @pytest.mark.asyncio
+    async def test_invalid_key_returns_none(self, mock_session, mock_settings_service_db):
+        """Invalid API key should return None."""
+        mock_result = MagicMock()
+        mock_result.first.return_value = None
+        mock_session.exec.return_value = mock_result
+
+        result = await _check_key_from_db(mock_session, "sk-invalid-key", mock_settings_service_db)
+
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_usage_tracking_increments(self, mock_session, mock_user, mock_settings_service_db):
+        """API key usage should be tracked when not disabled."""
+        mock_api_key = MagicMock()
+        mock_api_key.user = mock_user
+        mock_api_key.total_uses = 5
+
+        mock_result = MagicMock()
+        mock_result.first.return_value = mock_api_key
+        mock_session.exec.return_value = mock_result
+
+        await _check_key_from_db(mock_session, "sk-valid-key", mock_settings_service_db)
+
+        assert mock_api_key.total_uses == 6
+        mock_session.add.assert_called_once_with(mock_api_key)
+        mock_session.flush.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_usage_tracking_disabled(self, mock_session, mock_user, mock_settings_service_db):
+        """API key usage should not be tracked when disabled."""
+        mock_settings_service_db.settings.disable_track_apikey_usage = True
+
+        mock_api_key = MagicMock()
+        mock_api_key.user = mock_user
+        mock_api_key.total_uses = 5
+
+        mock_result = MagicMock()
+        mock_result.first.return_value = mock_api_key
+        mock_session.exec.return_value = mock_result
+
+        await _check_key_from_db(mock_session, "sk-valid-key", mock_settings_service_db)
+
+        assert mock_api_key.total_uses == 5  # Not incremented
+        mock_session.add.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_empty_key_returns_none(self, mock_session, mock_settings_service_db):
+        """Empty API key should return None."""
+        mock_result = MagicMock()
+        mock_result.first.return_value = None
+        mock_session.exec.return_value = mock_result
+
+        result = await _check_key_from_db(mock_session, "", mock_settings_service_db)
+
+        assert result is None
+
+
+# ============================================================================
+# _check_key_from_env tests
+# ============================================================================
+
+
+class TestCheckKeyFromEnv:
+    """Tests for environment variable-based API key validation."""
+
+    @pytest.mark.asyncio
+    async def test_valid_key_returns_superuser(
+        self, mock_session, mock_superuser, mock_settings_service_env, monkeypatch
+    ):
+        """Valid API key matching env var should return the superuser."""
+        monkeypatch.setenv("LANGFLOW_API_KEY", "sk-test-env-key")
+
+        with patch(
+            "langflow.services.database.models.user.crud.get_user_by_username",
+            new_callable=AsyncMock,
+        ) as mock_get_user:
+            mock_get_user.return_value = mock_superuser
+
+            result = await _check_key_from_env(mock_session, "sk-test-env-key", mock_settings_service_env)
+
+            assert result == mock_superuser
+            mock_get_user.assert_called_once_with(mock_session, "langflow")
+
+    @pytest.mark.asyncio
+    async def test_invalid_key_returns_none(self, mock_session, mock_settings_service_env, monkeypatch):
+        """Invalid API key not matching env var should return None."""
+        monkeypatch.setenv("LANGFLOW_API_KEY", "sk-test-env-key")
+
+        result = await _check_key_from_env(mock_session, "sk-wrong-key", mock_settings_service_env)
+
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_no_env_api_key_configured_returns_none(self, mock_session, mock_settings_service_env, monkeypatch):
+        """When LANGFLOW_API_KEY is not set, should return None."""
+        monkeypatch.delenv("LANGFLOW_API_KEY", raising=False)
+
+        result = await _check_key_from_env(mock_session, "sk-any-key", mock_settings_service_env)
+
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_empty_env_api_key_returns_none(self, mock_session, mock_settings_service_env, monkeypatch):
+        """When LANGFLOW_API_KEY is empty string, should return None."""
+        monkeypatch.setenv("LANGFLOW_API_KEY", "")
+
+        result = await _check_key_from_env(mock_session, "sk-any-key", mock_settings_service_env)
+
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_superuser_not_found_returns_none(self, mock_session, mock_settings_service_env, monkeypatch):
+        """When superuser doesn't exist in database, should return None."""
+        monkeypatch.setenv("LANGFLOW_API_KEY", "sk-test-env-key")
+
+        with patch(
+            "langflow.services.database.models.user.crud.get_user_by_username",
+            new_callable=AsyncMock,
+        ) as mock_get_user:
+            mock_get_user.return_value = None
+
+            result = await _check_key_from_env(mock_session, "sk-test-env-key", mock_settings_service_env)
+
+            assert result is None
+
+    @pytest.mark.asyncio
+    async def test_superuser_inactive_returns_none(
+        self, mock_session, mock_inactive_user, mock_settings_service_env, monkeypatch
+    ):
+        """When superuser is inactive, should return None."""
+        monkeypatch.setenv("LANGFLOW_API_KEY", "sk-test-env-key")
+
+        with patch(
+            "langflow.services.database.models.user.crud.get_user_by_username",
+            new_callable=AsyncMock,
+        ) as mock_get_user:
+            mock_get_user.return_value = mock_inactive_user
+
+            result = await _check_key_from_env(mock_session, "sk-test-env-key", mock_settings_service_env)
+
+            assert result is None
+
+    @pytest.mark.asyncio
+    async def test_case_sensitive_key_comparison(self, mock_session, mock_settings_service_env, monkeypatch):
+        """API key comparison should be case-sensitive."""
+        monkeypatch.setenv("LANGFLOW_API_KEY", "sk-Test-Key")
+
+        # Different case should not match
+        result = await _check_key_from_env(mock_session, "sk-test-key", mock_settings_service_env)
+        assert result is None
+
+        result = await _check_key_from_env(mock_session, "SK-TEST-KEY", mock_settings_service_env)
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_whitespace_in_key_not_trimmed(self, mock_session, mock_settings_service_env, monkeypatch):
+        """Whitespace in API key should not be trimmed."""
+        monkeypatch.setenv("LANGFLOW_API_KEY", "sk-test-key")
+
+        # Key with leading/trailing whitespace should not match
+        result = await _check_key_from_env(mock_session, " sk-test-key", mock_settings_service_env)
+        assert result is None
+
+        result = await _check_key_from_env(mock_session, "sk-test-key ", mock_settings_service_env)
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_special_characters_in_key(
+        self, mock_session, mock_superuser, mock_settings_service_env, monkeypatch
+    ):
+        """API key with special characters should work correctly."""
+        special_key = "sk-test!@#$%^&*()_+-=[]{}|;':\",./<>?"
+        monkeypatch.setenv("LANGFLOW_API_KEY", special_key)
+
+        with patch(
+            "langflow.services.database.models.user.crud.get_user_by_username",
+            new_callable=AsyncMock,
+        ) as mock_get_user:
+            mock_get_user.return_value = mock_superuser
+
+            result = await _check_key_from_env(mock_session, special_key, mock_settings_service_env)
+
+            assert result == mock_superuser
+
+    @pytest.mark.asyncio
+    async def test_unicode_in_key(self, mock_session, mock_superuser, mock_settings_service_env, monkeypatch):
+        """API key with unicode characters should work correctly."""
+        unicode_key = "sk-тест-キー-密钥"
+        monkeypatch.setenv("LANGFLOW_API_KEY", unicode_key)
+
+        with patch(
+            "langflow.services.database.models.user.crud.get_user_by_username",
+            new_callable=AsyncMock,
+        ) as mock_get_user:
+            mock_get_user.return_value = mock_superuser
+
+            result = await _check_key_from_env(mock_session, unicode_key, mock_settings_service_env)
+
+            assert result == mock_superuser
+
+    @pytest.mark.asyncio
+    async def test_very_long_key(self, mock_session, mock_superuser, mock_settings_service_env, monkeypatch):
+        """Very long API key should work correctly."""
+        long_key = "sk-" + "a" * 1000
+        monkeypatch.setenv("LANGFLOW_API_KEY", long_key)
+
+        with patch(
+            "langflow.services.database.models.user.crud.get_user_by_username",
+            new_callable=AsyncMock,
+        ) as mock_get_user:
+            mock_get_user.return_value = mock_superuser
+
+            result = await _check_key_from_env(mock_session, long_key, mock_settings_service_env)
+
+            assert result == mock_superuser
+
+
+# ============================================================================
+# Edge cases and error handling
+# ============================================================================
+
+
+class TestCheckKeyEdgeCases:
+    """Edge cases and error handling tests."""
+
+    @pytest.mark.asyncio
+    async def test_none_api_key_raises_or_returns_none(self, mock_session, mock_settings_service_db):
+        """Passing None as API key should be handled gracefully."""
+        mock_result = MagicMock()
+        mock_result.first.return_value = None
+        mock_session.exec.return_value = mock_result
+
+        # Should not raise, just return None
+        result = await _check_key_from_db(mock_session, None, mock_settings_service_db)
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_custom_superuser_name(self, mock_session, mock_superuser, mock_settings_service_env, monkeypatch):
+        """Should use custom superuser name from settings."""
+        monkeypatch.setenv("LANGFLOW_API_KEY", "sk-test-env-key")
+        mock_settings_service_env.auth_settings.SUPERUSER = "admin"
+        mock_superuser.username = "admin"
+
+        with patch(
+            "langflow.services.database.models.user.crud.get_user_by_username",
+            new_callable=AsyncMock,
+        ) as mock_get_user:
+            mock_get_user.return_value = mock_superuser
+
+            result = await _check_key_from_env(mock_session, "sk-test-env-key", mock_settings_service_env)
+
+            mock_get_user.assert_called_once_with(mock_session, "admin")
+            assert result == mock_superuser
+
+
+# ============================================================================
+# Integration-style tests (with real settings mocking)
+# ============================================================================
+
+
+class TestCheckKeyIntegration:
+    """Integration-style tests for the complete check_key flow."""
+
+    @pytest.mark.asyncio
+    async def test_full_flow_db_mode_valid_key(self, mock_session, mock_user):
+        """Full flow test: db mode with valid key."""
+        mock_api_key = MagicMock()
+        mock_api_key.user = mock_user
+        mock_api_key.total_uses = 0
+
+        mock_result = MagicMock()
+        mock_result.first.return_value = mock_api_key
+        mock_session.exec.return_value = mock_result
+
+        mock_settings = MagicMock()
+        mock_settings.auth_settings.API_KEY_SOURCE = "db"
+        mock_settings.settings.disable_track_apikey_usage = False
+
+        with patch(
+            "langflow.services.database.models.api_key.crud.get_settings_service",
+            return_value=mock_settings,
+        ):
+            result = await check_key(mock_session, "sk-valid-key")
+
+            assert result == mock_user
+
+    @pytest.mark.asyncio
+    async def test_full_flow_env_mode_valid_key(self, mock_session, mock_superuser, monkeypatch):
+        """Full flow test: env mode with valid key."""
+        monkeypatch.setenv("LANGFLOW_API_KEY", "sk-env-secret")
+
+        mock_settings = MagicMock()
+        mock_settings.auth_settings.API_KEY_SOURCE = "env"
+        mock_settings.auth_settings.SUPERUSER = "langflow"
+
+        with (
+            patch(
+                "langflow.services.database.models.api_key.crud.get_settings_service",
+                return_value=mock_settings,
+            ),
+            patch(
+                "langflow.services.database.models.user.crud.get_user_by_username",
+                new_callable=AsyncMock,
+            ) as mock_get_user,
+        ):
+            mock_get_user.return_value = mock_superuser
+
+            result = await check_key(mock_session, "sk-env-secret")
+
+            assert result == mock_superuser
+
+    @pytest.mark.asyncio
+    async def test_full_flow_env_mode_invalid_key_falls_back_to_db(self, mock_session, mock_user, monkeypatch):
+        """Full flow test: env mode with invalid key falls back to db."""
+        monkeypatch.setenv("LANGFLOW_API_KEY", "sk-correct-key")
+
+        # Setup mock for db fallback
+        mock_api_key = MagicMock()
+        mock_api_key.user = mock_user
+        mock_api_key.total_uses = 0
+
+        mock_result = MagicMock()
+        mock_result.first.return_value = mock_api_key
+        mock_session.exec.return_value = mock_result
+
+        mock_settings = MagicMock()
+        mock_settings.auth_settings.API_KEY_SOURCE = "env"
+        mock_settings.auth_settings.SUPERUSER = "langflow"
+        mock_settings.settings.disable_track_apikey_usage = False
+
+        with patch(
+            "langflow.services.database.models.api_key.crud.get_settings_service",
+            return_value=mock_settings,
+        ):
+            # Key doesn't match env, but exists in db
+            result = await check_key(mock_session, "sk-wrong-key")
+
+            # Should return user from db fallback
+            assert result == mock_user
+
+    @pytest.mark.asyncio
+    async def test_full_flow_env_mode_invalid_key_not_in_db(self, mock_session, monkeypatch):
+        """Full flow test: env mode with invalid key that's also not in db returns None."""
+        monkeypatch.setenv("LANGFLOW_API_KEY", "sk-correct-key")
+
+        # Setup mock for db - key not found
+        mock_result = MagicMock()
+        mock_result.first.return_value = None
+        mock_session.exec.return_value = mock_result
+
+        mock_settings = MagicMock()
+        mock_settings.auth_settings.API_KEY_SOURCE = "env"
+        mock_settings.auth_settings.SUPERUSER = "langflow"
+        mock_settings.settings.disable_track_apikey_usage = False
+
+        with patch(
+            "langflow.services.database.models.api_key.crud.get_settings_service",
+            return_value=mock_settings,
+        ):
+            # Key doesn't match env AND not in db
+            result = await check_key(mock_session, "sk-wrong-key")
+
+            # Should return None since both failed
+            assert result is None

src/backend/tests/unit/test_auth_settings.py

@@ -3,7 +3,7 @@ from pathlib import Path
 import pytest
 from lfx.services.settings.auth import AuthSettings
 from lfx.services.settings.constants import DEFAULT_SUPERUSER
-from pydantic import SecretStr
+from pydantic import SecretStr, ValidationError
 
 
 @pytest.mark.parametrize("auto_login", [True, False])
@@ -48,3 +48,68 @@ def test_auto_login_false_preserves_username_and_scrubs_password_on_reset(tmp_pa
     settings.reset_credentials()
     assert settings.SUPERUSER == "admin"
     assert settings.SUPERUSER_PASSWORD.get_secret_value() == ""
+
+
+# ============================================================================
+# API_KEY_SOURCE Settings Tests
+# ============================================================================
+
+
+class TestApiKeySourceSettings:
+    """Tests for API_KEY_SOURCE configuration setting."""
+
+    def test_api_key_source_default_is_db(self, tmp_path: Path):
+        """Default API_KEY_SOURCE should be 'db' for backward compatibility."""
+        cfg_dir = tmp_path.as_posix()
+        settings = AuthSettings(CONFIG_DIR=cfg_dir)
+        assert settings.API_KEY_SOURCE == "db"
+
+    def test_api_key_source_accepts_db(self, tmp_path: Path):
+        """API_KEY_SOURCE should accept 'db' value."""
+        cfg_dir = tmp_path.as_posix()
+        settings = AuthSettings(CONFIG_DIR=cfg_dir, API_KEY_SOURCE="db")
+        assert settings.API_KEY_SOURCE == "db"
+
+    def test_api_key_source_accepts_env(self, tmp_path: Path):
+        """API_KEY_SOURCE should accept 'env' value."""
+        cfg_dir = tmp_path.as_posix()
+        settings = AuthSettings(CONFIG_DIR=cfg_dir, API_KEY_SOURCE="env")
+        assert settings.API_KEY_SOURCE == "env"
+
+    def test_api_key_source_rejects_invalid_value(self, tmp_path: Path):
+        """API_KEY_SOURCE should reject invalid values."""
+        cfg_dir = tmp_path.as_posix()
+        with pytest.raises(ValidationError) as exc_info:
+            AuthSettings(CONFIG_DIR=cfg_dir, API_KEY_SOURCE="invalid")
+        assert "API_KEY_SOURCE" in str(exc_info.value)
+
+    def test_api_key_source_rejects_empty_string(self, tmp_path: Path):
+        """API_KEY_SOURCE should reject empty string."""
+        cfg_dir = tmp_path.as_posix()
+        with pytest.raises(ValidationError):
+            AuthSettings(CONFIG_DIR=cfg_dir, API_KEY_SOURCE="")
+
+
+class TestApiKeySourceEnvironmentVariables:
+    """Tests for API_KEY_SOURCE loaded from environment variables."""
+
+    def test_api_key_source_from_env_var(self, tmp_path: Path, monkeypatch):
+        """API_KEY_SOURCE should be loaded from LANGFLOW_API_KEY_SOURCE env var."""
+        cfg_dir = tmp_path.as_posix()
+        monkeypatch.setenv("LANGFLOW_API_KEY_SOURCE", "env")
+        settings = AuthSettings(CONFIG_DIR=cfg_dir)
+        assert settings.API_KEY_SOURCE == "env"
+
+    def test_explicit_value_overrides_env_var(self, tmp_path: Path, monkeypatch):
+        """Explicit parameter should override environment variable."""
+        cfg_dir = tmp_path.as_posix()
+        monkeypatch.setenv("LANGFLOW_API_KEY_SOURCE", "env")
+        settings = AuthSettings(CONFIG_DIR=cfg_dir, API_KEY_SOURCE="db")
+        assert settings.API_KEY_SOURCE == "db"
+
+    def test_invalid_api_key_source_from_env_var(self, tmp_path: Path, monkeypatch):
+        """Invalid API_KEY_SOURCE from env var should raise ValidationError."""
+        cfg_dir = tmp_path.as_posix()
+        monkeypatch.setenv("LANGFLOW_API_KEY_SOURCE", "invalid")
+        with pytest.raises(ValidationError):
+            AuthSettings(CONFIG_DIR=cfg_dir)

src/frontend/tests/extended/features/filterEdge-shard-1.spec.ts

@@ -1,8 +1,21 @@
+import type { Locator } from "@playwright/test";
+
 import { expect, test } from "../../fixtures";
 import { addLegacyComponents } from "../../utils/add-legacy-components";
 import { adjustScreenView } from "../../utils/adjust-screen-view";
 import { awaitBootstrapTest } from "../../utils/await-bootstrap-test";
 
+async function findVisibleElement(
+  elements: Locator[],
+): Promise<Locator | undefined> {
+  for (const element of elements) {
+    if (await element.isVisible()) {
+      return element;
+    }
+  }
+  return undefined;
+}
+
 test(
   "user must see on handle click the possibility connections - RetrievalQA",
   { tag: ["@release", "@api", "@components"] },
@@ -37,17 +50,14 @@ test(
     await page.mouse.down();
 
     await adjustScreenView(page);
-    let visibleElementHandle;
 
     const outputElements = await page
       .getByTestId("handle-retrievalqa-shownode-text-right")
       .all();
 
-    for (const element of outputElements) {
-      if (await element.isVisible()) {
-        visibleElementHandle = element;
-        break;
-      }
+    let visibleElementHandle = await findVisibleElement(outputElements);
+    if (!visibleElementHandle) {
+      throw new Error("Output handle not visible");
     }
 
     await visibleElementHandle.click({
@@ -123,11 +133,9 @@ test(
       .getByTestId("handle-retrievalqa-shownode-llm-left")
       .all();
 
-    for (const element of chainInputElements1) {
-      if (await element.isVisible()) {
-        visibleElementHandle = element;
-        break;
-      }
+    const llmHandle = await findVisibleElement(chainInputElements1);
+    if (llmHandle) {
+      visibleElementHandle = llmHandle;
     }
 
     await visibleElementHandle.blur();
@@ -142,11 +150,9 @@ test(
       .getByTestId("handle-retrievalqa-shownode-template-left")
       .all();
 
-    for (const element of rqaChainInputElements0) {
-      if (await element.isVisible()) {
-        visibleElementHandle = element;
-        break;
-      }
+    const templateHandle = await findVisibleElement(rqaChainInputElements0);
+    if (templateHandle) {
+      visibleElementHandle = templateHandle;
     }
 
     await visibleElementHandle.click();

src/lfx/src/lfx/services/settings/auth.py

@@ -27,6 +27,16 @@ class AuthSettings(BaseSettings):
     API_KEY_ALGORITHM: str = "HS256"
     API_V1_STR: str = "/api/v1"
 
+    # API Key Source Configuration
+    API_KEY_SOURCE: Literal["db", "env"] = Field(
+        default="db",
+        description=(
+            "Source for API key validation. "
+            "'db' validates against database-stored API keys (default behavior). "
+            "'env' validates against the LANGFLOW_API_KEY environment variable."
+        ),
+    )
+
     AUTO_LOGIN: bool = Field(
         default=True,  # TODO: Set to False in v2.0
         description=(