feat: Add asymmetric JWT authentication with RS256/RS512 support (#10755)

* add possibility to change jwt algorithm

* add docs related to jwt auth

* [autofix.ci] apply automated fixes

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

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

* ruff check fixes

* [autofix.ci] apply automated fixes

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

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

* convert to enum auth types

* [autofix.ci] apply automated fixes

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

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

* copilot gh suggestions

* ruff check

* [autofix.ci] apply automated fixes

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

* add validation to images read file compnent

* [autofix.ci] apply automated fixes

* ruff fixes

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

* [autofix.ci] apply automated fixes

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

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

* [autofix.ci] apply automated fixes

* revert: remove starter project changes

* revert: yarn lock

* [autofix.ci] apply automated fixes

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

* remove-docs-for-1.8-release

* remove starter projects

* [autofix.ci] apply automated fixes

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

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

* [autofix.ci] apply automated fixes

* improve logs message

* add missing dep in lfx

* [autofix.ci] apply automated fixes

* fix typo test error

* create helper functions

* [autofix.ci] apply automated fixes

* ruff checker

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes

---------

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>

Makefile

@@ -1,4 +1,4 @@
-.PHONY: all init format_backend format lint build run_backend dev help tests coverage clean_python_cache clean_npm_cache clean_frontend_build clean_all run_clic load_test_setup load_test_setup_basic load_test_list_flows load_test_run load_test_langflow_quick load_test_stress load_test_example load_test_clean load_test_remote_setup load_test_remote_run load_test_help
+.PHONY: all init format_backend format lint build run_backend dev help tests coverage clean_python_cache clean_npm_cache clean_frontend_build clean_all run_clic load_test_setup load_test_setup_basic load_test_list_flows load_test_run load_test_langflow_quick load_test_stress load_test_example load_test_clean load_test_remote_setup load_test_remote_run load_test_help docs docs_build docs_install
 
 # Configurations
 VERSION=$(shell grep "^version" pyproject.toml | sed 's/.*\"\(.*\)\"$$/\1/')
@@ -54,6 +54,7 @@ help: ## show basic help message with common commands
 	@echo "  $(GREEN)make format$(NC)              - Format all code (backend + frontend)"
 	@echo "  $(GREEN)make tests$(NC)               - Run all tests"
 	@echo "  $(GREEN)make build$(NC)               - Build the project"
+	@echo "  $(GREEN)make docs$(NC)                - Start documentation server (http://localhost:3030)"
 	@echo "  $(GREEN)make clean_all$(NC)           - Clean all caches and build artifacts"
 	@echo ''
 	@echo "$(GREEN)Specialized Help Commands:$(NC)"
@@ -996,6 +997,39 @@ help_advanced: ## show advanced and miscellaneous commands
 	@echo "$(GREEN)═══════════════════════════════════════════════════════════════════$(NC)"
 	@echo ''
 
+######################
+# DOCUMENTATION
+######################
+
+docs_port ?= 3030
+
+check_yarn:
+	@command -v yarn >/dev/null 2>&1 || { \
+		echo "$(RED)Error: yarn is not installed.$(NC)"; \
+		echo "$(YELLOW)The docs project requires yarn. Please install it:$(NC)"; \
+		echo "  brew install yarn"; \
+		echo "  # or"; \
+		echo "  npm install -g yarn"; \
+		exit 1; \
+	}
+
+docs_install: check_yarn ## install documentation dependencies
+	@echo "$(GREEN)Installing documentation dependencies...$(NC)"
+	@cd docs && yarn install
+
+docs: docs_install ## start documentation development server (default port 3030)
+	@echo "$(GREEN)Starting documentation server at http://localhost:$(docs_port)$(NC)"
+	@cd docs && yarn start --port $(docs_port)
+
+docs_build: docs_install ## build documentation for production
+	@echo "$(GREEN)Building documentation...$(NC)"
+	@cd docs && yarn build
+	@echo "$(GREEN)Documentation built successfully in docs/build/$(NC)"
+
+docs_serve: docs_build ## build and serve documentation locally
+	@echo "$(GREEN)Serving built documentation...$(NC)"
+	@cd docs && yarn serve --port $(docs_port)
+
 ######################
 # INCLUDE FRONTEND MAKEFILE
 ######################

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

@@ -44,6 +44,60 @@ AUTO_LOGIN_ERROR = (
 REFRESH_TOKEN_TYPE: Final[str] = "refresh"  # noqa: S105
 ACCESS_TOKEN_TYPE: Final[str] = "access"  # noqa: S105
 
+# JWT key configuration error messages
+PUBLIC_KEY_NOT_CONFIGURED_ERROR: Final[str] = (
+    "Server configuration error: Public key not configured for asymmetric JWT algorithm."
+)
+SECRET_KEY_NOT_CONFIGURED_ERROR: Final[str] = "Server configuration error: Secret key not configured."  # noqa: S105
+
+
+class JWTKeyError(HTTPException):
+    """Raised when JWT key configuration is invalid."""
+
+    def __init__(self, detail: str, *, include_www_authenticate: bool = True):
+        headers = {"WWW-Authenticate": "Bearer"} if include_www_authenticate else None
+        super().__init__(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail=detail,
+            headers=headers,
+        )
+
+
+def get_jwt_verification_key(settings_service: SettingsService) -> str:
+    """Get the appropriate key for JWT verification based on configured algorithm.
+
+    For asymmetric algorithms (RS256, RS512): returns public key
+    For symmetric algorithms (HS256): returns secret key
+    """
+    algorithm = settings_service.auth_settings.ALGORITHM
+
+    if algorithm.is_asymmetric():
+        verification_key = settings_service.auth_settings.PUBLIC_KEY
+        if not verification_key:
+            logger.error("Public key is not set in settings for RS256/RS512.")
+            raise JWTKeyError(PUBLIC_KEY_NOT_CONFIGURED_ERROR)
+        return verification_key
+
+    secret_key = settings_service.auth_settings.SECRET_KEY.get_secret_value()
+    if secret_key is None:
+        logger.error("Secret key is not set in settings.")
+        raise JWTKeyError(SECRET_KEY_NOT_CONFIGURED_ERROR)
+    return secret_key
+
+
+def get_jwt_signing_key(settings_service: SettingsService) -> str:
+    """Get the appropriate key for JWT signing based on configured algorithm.
+
+    For asymmetric algorithms (RS256, RS512): returns private key
+    For symmetric algorithms (HS256): returns secret key
+    """
+    algorithm = settings_service.auth_settings.ALGORITHM
+
+    if algorithm.is_asymmetric():
+        return settings_service.auth_settings.PRIVATE_KEY.get_secret_value()
+
+    return settings_service.auth_settings.SECRET_KEY.get_secret_value()
+
 
 # Source: https://github.com/mrtolkien/fastapi_simple_security/blob/master/fastapi_simple_security/security_api_key.py
 async def api_key_security(
@@ -172,20 +226,13 @@ async def get_current_user_by_jwt(
     if isinstance(token, Coroutine):
         token = await token
 
-    secret_key = settings_service.auth_settings.SECRET_KEY.get_secret_value()
-    if secret_key is None:
-        logger.error("Secret key is not set in settings.")
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            # Careful not to leak sensitive information
-            detail="Authentication failure: Verify authentication settings.",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
+    algorithm = settings_service.auth_settings.ALGORITHM
+    verification_key = get_jwt_verification_key(settings_service)
 
     try:
         with warnings.catch_warnings():
             warnings.simplefilter("ignore")
-            payload = jwt.decode(token, secret_key, algorithms=[settings_service.auth_settings.ALGORITHM])
+            payload = jwt.decode(token, verification_key, algorithms=[algorithm])
         user_id: UUID = payload.get("sub")  # type: ignore[assignment]
         token_type: str = payload.get("type")  # type: ignore[assignment]
 
@@ -364,10 +411,13 @@ def create_token(data: dict, expires_delta: timedelta):
     expire = datetime.now(timezone.utc) + expires_delta
     to_encode["exp"] = expire
 
+    algorithm = settings_service.auth_settings.ALGORITHM
+    signing_key = get_jwt_signing_key(settings_service)
+
     return jwt.encode(
         to_encode,
-        settings_service.auth_settings.SECRET_KEY.get_secret_value(),
-        algorithm=settings_service.auth_settings.ALGORITHM,
+        signing_key,
+        algorithm=algorithm,
     )
 
 
@@ -484,14 +534,17 @@ async def create_user_tokens(user_id: UUID, db: AsyncSession, *, update_last_log
 async def create_refresh_token(refresh_token: str, db: AsyncSession):
     settings_service = get_settings_service()
 
+    algorithm = settings_service.auth_settings.ALGORITHM
+    verification_key = get_jwt_verification_key(settings_service)
+
     try:
         # Ignore warning about datetime.utcnow
         with warnings.catch_warnings():
             warnings.simplefilter("ignore")
             payload = jwt.decode(
                 refresh_token,
-                settings_service.auth_settings.SECRET_KEY.get_secret_value(),
-                algorithms=[settings_service.auth_settings.ALGORITHM],
+                verification_key,
+                algorithms=[algorithm],
             )
         user_id: UUID = payload.get("sub")  # type: ignore[assignment]
         token_type: str = payload.get("type")  # type: ignore[assignment]

src/backend/tests/unit/test_auth_jwt_algorithms.py

@@ -0,0 +1,989 @@
+"""Comprehensive tests for JWT algorithm support (HS256, RS256, RS512).
+
+Tests cover:
+- AuthSettings configuration for each algorithm
+- RSA key generation and persistence
+- Token creation and verification
+- Error cases and edge cases
+- Authentication failure scenarios
+"""
+
+import tempfile
+from datetime import timedelta
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+from jose import JWTError, jwt
+from pydantic import SecretStr
+
+
+class TestAuthSettingsAlgorithms:
+    """Test AuthSettings configuration for different JWT algorithms."""
+
+    def test_default_algorithm_is_hs256(self):
+        """Default algorithm should be HS256 for backward compatibility (when not overridden by env)."""
+        from lfx.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # Explicitly set HS256 to test the setting works (env var may override default)
+            settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="HS256")
+            assert settings.ALGORITHM == "HS256"
+
+    def test_hs256_generates_secret_key(self):
+        """HS256 should generate a secret key automatically."""
+        from lfx.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="HS256")
+            secret_key = settings.SECRET_KEY.get_secret_value()
+            assert secret_key is not None
+            assert len(secret_key) >= 32
+
+    def test_rs256_generates_rsa_key_pair(self):
+        """RS256 should generate RSA key pair automatically."""
+        from lfx.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="RS256")
+
+            private_key = settings.PRIVATE_KEY.get_secret_value()
+            public_key = settings.PUBLIC_KEY
+
+            assert private_key is not None
+            assert public_key is not None
+            assert "-----BEGIN PRIVATE KEY-----" in private_key
+            assert "-----BEGIN PUBLIC KEY-----" in public_key
+
+    def test_rs512_generates_rsa_key_pair(self):
+        """RS512 should generate RSA key pair automatically."""
+        from lfx.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="RS512")
+
+            private_key = settings.PRIVATE_KEY.get_secret_value()
+            public_key = settings.PUBLIC_KEY
+
+            assert private_key is not None
+            assert public_key is not None
+            assert "-----BEGIN PRIVATE KEY-----" in private_key
+            assert "-----BEGIN PUBLIC KEY-----" in public_key
+
+    def test_rsa_keys_persisted_to_files(self):
+        """RSA keys should be persisted to files in CONFIG_DIR."""
+        from lfx.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="RS256")
+
+            private_key_path = Path(tmpdir) / "private_key.pem"
+            public_key_path = Path(tmpdir) / "public_key.pem"
+
+            assert private_key_path.exists()
+            assert public_key_path.exists()
+
+            # Verify file contents match settings
+            assert private_key_path.read_text() == settings.PRIVATE_KEY.get_secret_value()
+            assert public_key_path.read_text() == settings.PUBLIC_KEY
+
+    def test_rsa_keys_loaded_from_existing_files(self):
+        """RSA keys should be loaded from existing files."""
+        from lfx.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # First run - generate keys
+            settings1 = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="RS256")
+            original_private = settings1.PRIVATE_KEY.get_secret_value()
+            original_public = settings1.PUBLIC_KEY
+
+            # Second run - should load existing keys
+            settings2 = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="RS256")
+
+            assert settings2.PRIVATE_KEY.get_secret_value() == original_private
+            assert original_public == settings2.PUBLIC_KEY
+
+    def test_custom_private_key_derives_public_key(self):
+        """When custom private key is provided, public key should be derived."""
+        from lfx.services.settings.auth import AuthSettings
+        from lfx.services.settings.utils import generate_rsa_key_pair
+
+        custom_private, expected_public = generate_rsa_key_pair()
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            settings = AuthSettings(
+                CONFIG_DIR=tmpdir,
+                ALGORITHM="RS256",
+                PRIVATE_KEY=SecretStr(custom_private),
+            )
+
+            assert settings.PRIVATE_KEY.get_secret_value() == custom_private
+            assert expected_public == settings.PUBLIC_KEY
+
+    def test_no_config_dir_generates_keys_in_memory(self):
+        """Without CONFIG_DIR, keys should be generated in memory."""
+        from lfx.services.settings.auth import AuthSettings
+
+        settings = AuthSettings(CONFIG_DIR="", ALGORITHM="RS256")
+
+        assert settings.PRIVATE_KEY.get_secret_value() is not None
+        assert settings.PUBLIC_KEY is not None
+        assert "-----BEGIN PRIVATE KEY-----" in settings.PRIVATE_KEY.get_secret_value()
+
+    def test_hs256_does_not_generate_rsa_keys(self):
+        """HS256 should not trigger RSA key generation."""
+        from lfx.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="HS256")
+
+            private_key_path = Path(tmpdir) / "private_key.pem"
+            public_key_path = Path(tmpdir) / "public_key.pem"
+
+            # RSA key files should not be created for HS256
+            assert not private_key_path.exists()
+            assert not public_key_path.exists()
+
+    def test_invalid_algorithm_rejected(self):
+        """Invalid algorithm should be rejected by pydantic."""
+        from lfx.services.settings.auth import AuthSettings
+        from pydantic import ValidationError
+
+        with tempfile.TemporaryDirectory() as tmpdir, pytest.raises(ValidationError):
+            AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="INVALID")
+
+
+class TestRSAKeyGeneration:
+    """Test RSA key pair generation utility."""
+
+    def test_generate_rsa_key_pair_returns_valid_keys(self):
+        """Generated keys should be valid PEM format."""
+        from lfx.services.settings.utils import generate_rsa_key_pair
+
+        private_key, public_key = generate_rsa_key_pair()
+
+        assert "-----BEGIN PRIVATE KEY-----" in private_key
+        assert "-----END PRIVATE KEY-----" in private_key
+        assert "-----BEGIN PUBLIC KEY-----" in public_key
+        assert "-----END PUBLIC KEY-----" in public_key
+
+    def test_generated_keys_are_unique(self):
+        """Each call should generate unique keys."""
+        from lfx.services.settings.utils import generate_rsa_key_pair
+
+        private1, public1 = generate_rsa_key_pair()
+        private2, public2 = generate_rsa_key_pair()
+
+        assert private1 != private2
+        assert public1 != public2
+
+    def test_generated_keys_can_sign_and_verify(self):
+        """Generated keys should work for JWT signing and verification."""
+        from lfx.services.settings.utils import generate_rsa_key_pair
+
+        private_key, public_key = generate_rsa_key_pair()
+
+        # Sign a token
+        payload = {"sub": "test-user", "type": "access"}
+        token = jwt.encode(payload, private_key, algorithm="RS256")
+
+        # Verify the token
+        decoded = jwt.decode(token, public_key, algorithms=["RS256"])
+
+        assert decoded["sub"] == "test-user"
+        assert decoded["type"] == "access"
+
+
+class TestTokenCreation:
+    """Test JWT token creation with different algorithms."""
+
+    def _create_mock_settings_service(self, algorithm, tmpdir):
+        """Helper to create mock settings service."""
+        from lfx.services.settings.auth import AuthSettings
+
+        settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM=algorithm)
+
+        mock_service = MagicMock()
+        mock_service.auth_settings = settings
+        return mock_service
+
+    def test_create_token_hs256(self):
+        """Token creation with HS256 should use secret key."""
+        from langflow.services.auth.utils import create_token
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+
+            with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+                token = create_token(
+                    data={"sub": "user-123", "type": "access"},
+                    expires_delta=timedelta(hours=1),
+                )
+
+                assert token is not None
+                # Verify token header shows HS256
+                header = jwt.get_unverified_header(token)
+                assert header["alg"] == "HS256"
+
+    def test_create_token_rs256(self):
+        """Token creation with RS256 should use private key."""
+        from langflow.services.auth.utils import create_token
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("RS256", tmpdir)
+
+            with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+                token = create_token(
+                    data={"sub": "user-456", "type": "access"},
+                    expires_delta=timedelta(hours=1),
+                )
+
+                assert token is not None
+                # Verify token header shows RS256
+                header = jwt.get_unverified_header(token)
+                assert header["alg"] == "RS256"
+
+    def test_create_token_rs512(self):
+        """Token creation with RS512 should use private key."""
+        from langflow.services.auth.utils import create_token
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("RS512", tmpdir)
+
+            with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+                token = create_token(
+                    data={"sub": "user-789", "type": "access"},
+                    expires_delta=timedelta(hours=1),
+                )
+
+                assert token is not None
+                # Verify token header shows RS512
+                header = jwt.get_unverified_header(token)
+                assert header["alg"] == "RS512"
+
+    def test_token_contains_expiration(self):
+        """Created token should contain expiration claim."""
+        from langflow.services.auth.utils import create_token
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+
+            with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+                token = create_token(
+                    data={"sub": "user-123", "type": "access"},
+                    expires_delta=timedelta(hours=1),
+                )
+
+                # Decode without verification to check claims
+                claims = jwt.get_unverified_claims(token)
+                assert "exp" in claims
+
+
+class TestTokenVerification:
+    """Test JWT token verification with different algorithms."""
+
+    def _create_mock_settings_service(self, algorithm, tmpdir):
+        """Helper to create mock settings service."""
+        from lfx.services.settings.auth import AuthSettings
+
+        settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM=algorithm)
+
+        mock_service = MagicMock()
+        mock_service.auth_settings = settings
+        return mock_service
+
+    @pytest.mark.asyncio
+    async def test_verify_hs256_token_success(self):
+        """Valid HS256 token should be verified successfully."""
+        from langflow.services.auth.utils import create_token, get_current_user_by_jwt
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+
+            # Create a mock user
+            mock_user = MagicMock()
+            mock_user.id = "user-123"
+            mock_user.is_active = True
+
+            mock_db = AsyncMock()
+
+            with (
+                patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service),
+                patch("langflow.services.auth.utils.get_user_by_id", return_value=mock_user),
+            ):
+                token = create_token(
+                    data={"sub": "user-123", "type": "access"},
+                    expires_delta=timedelta(hours=1),
+                )
+
+                user = await get_current_user_by_jwt(token, mock_db)
+                assert user == mock_user
+
+    @pytest.mark.asyncio
+    async def test_verify_rs256_token_success(self):
+        """Valid RS256 token should be verified successfully."""
+        from langflow.services.auth.utils import create_token, get_current_user_by_jwt
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("RS256", tmpdir)
+
+            mock_user = MagicMock()
+            mock_user.id = "user-456"
+            mock_user.is_active = True
+
+            mock_db = AsyncMock()
+
+            with (
+                patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service),
+                patch("langflow.services.auth.utils.get_user_by_id", return_value=mock_user),
+            ):
+                token = create_token(
+                    data={"sub": "user-456", "type": "access"},
+                    expires_delta=timedelta(hours=1),
+                )
+
+                user = await get_current_user_by_jwt(token, mock_db)
+                assert user == mock_user
+
+    @pytest.mark.asyncio
+    async def test_verify_rs512_token_success(self):
+        """Valid RS512 token should be verified successfully."""
+        from langflow.services.auth.utils import create_token, get_current_user_by_jwt
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("RS512", tmpdir)
+
+            mock_user = MagicMock()
+            mock_user.id = "user-789"
+            mock_user.is_active = True
+
+            mock_db = AsyncMock()
+
+            with (
+                patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service),
+                patch("langflow.services.auth.utils.get_user_by_id", return_value=mock_user),
+            ):
+                token = create_token(
+                    data={"sub": "user-789", "type": "access"},
+                    expires_delta=timedelta(hours=1),
+                )
+
+                user = await get_current_user_by_jwt(token, mock_db)
+                assert user == mock_user
+
+
+class TestAuthenticationFailures:
+    """Test authentication failure scenarios."""
+
+    def _create_mock_settings_service(self, algorithm, tmpdir, **overrides):
+        """Helper to create mock settings service with optional overrides."""
+        from lfx.services.settings.auth import AuthSettings
+
+        settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM=algorithm)
+
+        # Apply overrides
+        for key, value in overrides.items():
+            object.__setattr__(settings, key, value)
+
+        mock_service = MagicMock()
+        mock_service.auth_settings = settings
+        return mock_service
+
+    @pytest.mark.asyncio
+    async def test_missing_public_key_rs256_raises_401(self):
+        """Missing public key for RS256 should raise 401."""
+        from langflow.services.auth.utils import get_current_user_by_jwt
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("RS256", tmpdir, PUBLIC_KEY="")
+
+            mock_db = AsyncMock()
+
+            with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+                with pytest.raises(HTTPException) as exc_info:
+                    await get_current_user_by_jwt("some-token", mock_db)
+
+                assert exc_info.value.status_code == 401
+                assert "Server configuration error" in exc_info.value.detail
+                assert "Public key not configured" in exc_info.value.detail
+
+    @pytest.mark.asyncio
+    async def test_missing_secret_key_hs256_raises_401(self):
+        """Missing secret key for HS256 should raise 401."""
+        from langflow.services.auth.utils import get_current_user_by_jwt
+        from lfx.services.settings.auth import JWTAlgorithm
+
+        # Create a fully mocked settings service without using AuthSettings
+        mock_auth_settings = MagicMock()
+        mock_auth_settings.ALGORITHM = JWTAlgorithm.HS256
+        mock_auth_settings.SECRET_KEY = MagicMock()
+        mock_auth_settings.SECRET_KEY.get_secret_value.return_value = None
+
+        mock_service = MagicMock()
+        mock_service.auth_settings = mock_auth_settings
+
+        mock_db = AsyncMock()
+
+        with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+            with pytest.raises(HTTPException) as exc_info:
+                await get_current_user_by_jwt("some-token", mock_db)
+
+            assert exc_info.value.status_code == 401
+            assert "Server configuration error" in exc_info.value.detail
+            assert "Secret key not configured" in exc_info.value.detail
+
+    @pytest.mark.asyncio
+    async def test_invalid_token_raises_401(self):
+        """Invalid token should raise 401."""
+        from langflow.services.auth.utils import get_current_user_by_jwt
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+            mock_db = AsyncMock()
+
+            with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+                with pytest.raises(HTTPException) as exc_info:
+                    await get_current_user_by_jwt("invalid-token-format", mock_db)
+
+                assert exc_info.value.status_code == 401
+                assert "Could not validate credentials" in exc_info.value.detail
+
+    @pytest.mark.asyncio
+    async def test_token_signed_with_wrong_key_raises_401(self):
+        """Token signed with different key should raise 401."""
+        from langflow.services.auth.utils import get_current_user_by_jwt
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+
+            # Create token with different secret
+            wrong_token = jwt.encode(
+                {"sub": "user-123", "type": "access"},
+                "different-secret-key",
+                algorithm="HS256",
+            )
+
+            mock_db = AsyncMock()
+
+            with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+                with pytest.raises(HTTPException) as exc_info:
+                    await get_current_user_by_jwt(wrong_token, mock_db)
+
+                assert exc_info.value.status_code == 401
+
+    @pytest.mark.asyncio
+    async def test_expired_token_raises_401(self):
+        """Expired token should raise 401."""
+        from langflow.services.auth.utils import create_token, get_current_user_by_jwt
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+            mock_db = AsyncMock()
+
+            with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+                # Create token that's already expired
+                token = create_token(
+                    data={"sub": "user-123", "type": "access"},
+                    expires_delta=timedelta(seconds=-10),  # Negative = already expired
+                )
+
+                with pytest.raises(HTTPException) as exc_info:
+                    await get_current_user_by_jwt(token, mock_db)
+
+                assert exc_info.value.status_code == 401
+                # jose library raises JWTError for expired tokens before our custom check
+                assert (
+                    "expired" in exc_info.value.detail.lower() or "could not validate" in exc_info.value.detail.lower()
+                )
+
+    @pytest.mark.asyncio
+    async def test_token_without_user_id_raises_401(self):
+        """Token without user ID should raise 401."""
+        from langflow.services.auth.utils import get_current_user_by_jwt
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+
+            # Create token without 'sub' claim
+            token = jwt.encode(
+                {"type": "access"},
+                mock_service.auth_settings.SECRET_KEY.get_secret_value(),
+                algorithm="HS256",
+            )
+
+            mock_db = AsyncMock()
+
+            with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+                with pytest.raises(HTTPException) as exc_info:
+                    await get_current_user_by_jwt(token, mock_db)
+
+                assert exc_info.value.status_code == 401
+                assert "Invalid token" in exc_info.value.detail
+
+    @pytest.mark.asyncio
+    async def test_token_without_type_raises_401(self):
+        """Token without type should raise 401."""
+        from langflow.services.auth.utils import get_current_user_by_jwt
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+
+            # Create token without 'type' claim
+            token = jwt.encode(
+                {"sub": "user-123"},
+                mock_service.auth_settings.SECRET_KEY.get_secret_value(),
+                algorithm="HS256",
+            )
+
+            mock_db = AsyncMock()
+
+            with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+                with pytest.raises(HTTPException) as exc_info:
+                    await get_current_user_by_jwt(token, mock_db)
+
+                assert exc_info.value.status_code == 401
+                assert "invalid" in exc_info.value.detail.lower()
+
+    @pytest.mark.asyncio
+    async def test_user_not_found_raises_401(self):
+        """Token for non-existent user should raise 401."""
+        from langflow.services.auth.utils import create_token, get_current_user_by_jwt
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+            mock_db = AsyncMock()
+
+            with (
+                patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service),
+                patch("langflow.services.auth.utils.get_user_by_id", return_value=None),
+            ):
+                token = create_token(
+                    data={"sub": "non-existent-user", "type": "access"},
+                    expires_delta=timedelta(hours=1),
+                )
+
+                with pytest.raises(HTTPException) as exc_info:
+                    await get_current_user_by_jwt(token, mock_db)
+
+                assert exc_info.value.status_code == 401
+                assert "User not found" in exc_info.value.detail
+
+    @pytest.mark.asyncio
+    async def test_inactive_user_raises_401(self):
+        """Token for inactive user should raise 401."""
+        from langflow.services.auth.utils import create_token, get_current_user_by_jwt
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+
+            mock_user = MagicMock()
+            mock_user.is_active = False
+
+            mock_db = AsyncMock()
+
+            with (
+                patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service),
+                patch("langflow.services.auth.utils.get_user_by_id", return_value=mock_user),
+            ):
+                token = create_token(
+                    data={"sub": "inactive-user", "type": "access"},
+                    expires_delta=timedelta(hours=1),
+                )
+
+                with pytest.raises(HTTPException) as exc_info:
+                    await get_current_user_by_jwt(token, mock_db)
+
+                assert exc_info.value.status_code == 401
+                assert "inactive" in exc_info.value.detail.lower()
+
+
+class TestRefreshTokenVerification:
+    """Test refresh token verification with different algorithms."""
+
+    def _create_mock_settings_service(self, algorithm, tmpdir):
+        """Helper to create mock settings service."""
+        from lfx.services.settings.auth import AuthSettings
+
+        settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM=algorithm)
+
+        mock_service = MagicMock()
+        mock_service.auth_settings = settings
+        return mock_service
+
+    @pytest.mark.asyncio
+    async def test_refresh_token_rs256_success(self):
+        """Valid RS256 refresh token should create new tokens."""
+        from langflow.services.auth.utils import create_refresh_token, create_token
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("RS256", tmpdir)
+
+            mock_user = MagicMock()
+            mock_user.id = "user-123"
+            mock_user.is_active = True
+
+            mock_db = AsyncMock()
+
+            with (
+                patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service),
+                patch("langflow.services.auth.utils.get_user_by_id", return_value=mock_user),
+            ):
+                # Create refresh token
+                refresh_token = create_token(
+                    data={"sub": "user-123", "type": "refresh"},
+                    expires_delta=timedelta(days=7),
+                )
+
+                # Use refresh token to get new tokens
+                new_tokens = await create_refresh_token(refresh_token, mock_db)
+
+                assert "access_token" in new_tokens
+                assert "refresh_token" in new_tokens
+                assert new_tokens.get("token_type") == "bearer"
+
+    @pytest.mark.asyncio
+    async def test_refresh_token_wrong_type_raises_401(self):
+        """Access token used as refresh token should raise 401."""
+        from langflow.services.auth.utils import create_refresh_token, create_token
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+            mock_db = AsyncMock()
+
+            with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+                # Create access token (not refresh)
+                access_token = create_token(
+                    data={"sub": "user-123", "type": "access"},
+                    expires_delta=timedelta(hours=1),
+                )
+
+                with pytest.raises(HTTPException) as exc_info:
+                    await create_refresh_token(access_token, mock_db)
+
+                assert exc_info.value.status_code == 401
+                assert "Invalid refresh token" in exc_info.value.detail
+
+
+class TestAlgorithmMismatch:
+    """Test scenarios where algorithm configuration changes."""
+
+    def test_hs256_token_fails_with_rs256_verification(self):
+        """Token created with HS256 should fail RS256 verification."""
+        from lfx.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # Create token with HS256
+            hs256_settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="HS256")
+            token = jwt.encode(
+                {"sub": "user-123", "type": "access"},
+                hs256_settings.SECRET_KEY.get_secret_value(),
+                algorithm="HS256",
+            )
+
+        with tempfile.TemporaryDirectory() as tmpdir2:
+            # Try to verify with RS256
+            rs256_settings = AuthSettings(CONFIG_DIR=tmpdir2, ALGORITHM="RS256")
+
+            with pytest.raises(JWTError):
+                jwt.decode(token, rs256_settings.PUBLIC_KEY, algorithms=["RS256"])
+
+    def test_rs256_token_fails_with_hs256_verification(self):
+        """Token created with RS256 should fail HS256 verification."""
+        from lfx.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # Create token with RS256
+            rs256_settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="RS256")
+            token = jwt.encode(
+                {"sub": "user-123", "type": "access"},
+                rs256_settings.PRIVATE_KEY.get_secret_value(),
+                algorithm="RS256",
+            )
+
+        with tempfile.TemporaryDirectory() as tmpdir2:
+            # Try to verify with HS256
+            hs256_settings = AuthSettings(CONFIG_DIR=tmpdir2, ALGORITHM="HS256")
+
+            with pytest.raises(JWTError):
+                jwt.decode(
+                    token,
+                    hs256_settings.SECRET_KEY.get_secret_value(),
+                    algorithms=["HS256"],
+                )
+
+
+class TestKeyPersistence:
+    """Test key persistence and file permissions."""
+
+    def test_secret_key_file_created(self):
+        """Secret key should be saved to file."""
+        from lfx.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="HS256")
+
+            secret_key_path = Path(tmpdir) / "secret_key"
+            assert secret_key_path.exists()
+
+    def test_rsa_key_files_created(self):
+        """RSA keys should be saved to files."""
+        from lfx.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="RS256")
+
+            private_key_path = Path(tmpdir) / "private_key.pem"
+            public_key_path = Path(tmpdir) / "public_key.pem"
+
+            assert private_key_path.exists()
+            assert public_key_path.exists()
+
+    def test_keys_reloaded_on_restart(self):
+        """Keys should be consistent across settings reloads."""
+        from lfx.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # First load
+            settings1 = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="RS256")
+            private1 = settings1.PRIVATE_KEY.get_secret_value()
+            public1 = settings1.PUBLIC_KEY
+
+            # Simulate restart
+            settings2 = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="RS256")
+            private2 = settings2.PRIVATE_KEY.get_secret_value()
+            public2 = settings2.PUBLIC_KEY
+
+            assert private1 == private2
+            assert public1 == public2
+
+
+class TestEdgeCases:
+    """Test edge cases and boundary conditions."""
+
+    def test_empty_config_dir_string(self):
+        """Empty CONFIG_DIR string should work (in-memory keys)."""
+        from lfx.services.settings.auth import AuthSettings
+
+        settings = AuthSettings(CONFIG_DIR="", ALGORITHM="RS256")
+        assert settings.PRIVATE_KEY.get_secret_value() is not None
+        assert settings.PUBLIC_KEY is not None
+
+    def test_token_with_extra_claims(self):
+        """Token with extra claims should still work."""
+        from langflow.services.auth.utils import get_current_user_by_jwt
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            from lfx.services.settings.auth import AuthSettings
+
+            settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="HS256")
+
+            token = jwt.encode(
+                {
+                    "sub": "user-123",
+                    "type": "access",
+                    "extra_claim": "some-value",
+                    "another": 123,
+                },
+                settings.SECRET_KEY.get_secret_value(),
+                algorithm="HS256",
+            )
+
+            mock_service = MagicMock()
+            mock_service.auth_settings = settings
+
+            mock_user = MagicMock()
+            mock_user.id = "user-123"
+            mock_user.is_active = True
+
+            mock_db = AsyncMock()
+
+            with (
+                patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service),
+                patch("langflow.services.auth.utils.get_user_by_id", return_value=mock_user),
+            ):
+                import asyncio
+
+                user = asyncio.get_event_loop().run_until_complete(get_current_user_by_jwt(token, mock_db))
+                assert user == mock_user
+
+    def test_very_long_user_id(self):
+        """Very long user ID should work."""
+        from langflow.services.auth.utils import create_token
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            from lfx.services.settings.auth import AuthSettings
+
+            settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM="HS256")
+
+            mock_service = MagicMock()
+            mock_service.auth_settings = settings
+
+            long_user_id = "a" * 1000
+
+            with patch("langflow.services.auth.utils.get_settings_service", return_value=mock_service):
+                token = create_token(
+                    data={"sub": long_user_id, "type": "access"},
+                    expires_delta=timedelta(hours=1),
+                )
+
+                claims = jwt.get_unverified_claims(token)
+                assert claims["sub"] == long_user_id
+
+
+class TestJWTKeyHelpers:
+    """Test JWT key helper functions."""
+
+    def _create_mock_settings_service(self, algorithm, tmpdir):
+        """Helper to create mock settings service."""
+        from lfx.services.settings.auth import AuthSettings
+
+        settings = AuthSettings(CONFIG_DIR=tmpdir, ALGORITHM=algorithm)
+
+        mock_service = MagicMock()
+        mock_service.auth_settings = settings
+        return mock_service
+
+    def test_get_jwt_verification_key_hs256_returns_secret_key(self):
+        """HS256 should return secret key for verification."""
+        from langflow.services.auth.utils import get_jwt_verification_key
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+
+            key = get_jwt_verification_key(mock_service)
+
+            assert key == mock_service.auth_settings.SECRET_KEY.get_secret_value()
+            assert len(key) >= 32
+
+    def test_get_jwt_verification_key_rs256_returns_public_key(self):
+        """RS256 should return public key for verification."""
+        from langflow.services.auth.utils import get_jwt_verification_key
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("RS256", tmpdir)
+
+            key = get_jwt_verification_key(mock_service)
+
+            assert key == mock_service.auth_settings.PUBLIC_KEY
+            assert "-----BEGIN PUBLIC KEY-----" in key
+
+    def test_get_jwt_verification_key_rs512_returns_public_key(self):
+        """RS512 should return public key for verification."""
+        from langflow.services.auth.utils import get_jwt_verification_key
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("RS512", tmpdir)
+
+            key = get_jwt_verification_key(mock_service)
+
+            assert key == mock_service.auth_settings.PUBLIC_KEY
+            assert "-----BEGIN PUBLIC KEY-----" in key
+
+    def test_get_jwt_verification_key_missing_public_key_raises_error(self):
+        """Missing public key for asymmetric algorithm should raise JWTKeyError."""
+        from langflow.services.auth.utils import JWTKeyError, get_jwt_verification_key
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("RS256", tmpdir)
+            object.__setattr__(mock_service.auth_settings, "PUBLIC_KEY", "")
+
+            with pytest.raises(JWTKeyError) as exc_info:
+                get_jwt_verification_key(mock_service)
+
+            assert exc_info.value.status_code == 401
+            assert "Public key not configured" in exc_info.value.detail
+
+    def test_get_jwt_verification_key_missing_secret_key_raises_error(self):
+        """Missing secret key for HS256 should raise JWTKeyError."""
+        from langflow.services.auth.utils import JWTKeyError, get_jwt_verification_key
+        from lfx.services.settings.auth import JWTAlgorithm
+
+        mock_auth_settings = MagicMock()
+        mock_auth_settings.ALGORITHM = JWTAlgorithm.HS256
+        mock_auth_settings.SECRET_KEY = MagicMock()
+        mock_auth_settings.SECRET_KEY.get_secret_value.return_value = None
+
+        mock_service = MagicMock()
+        mock_service.auth_settings = mock_auth_settings
+
+        with pytest.raises(JWTKeyError) as exc_info:
+            get_jwt_verification_key(mock_service)
+
+        assert exc_info.value.status_code == 401
+        assert "Secret key not configured" in exc_info.value.detail
+
+    def test_get_jwt_signing_key_hs256_returns_secret_key(self):
+        """HS256 should return secret key for signing."""
+        from langflow.services.auth.utils import get_jwt_signing_key
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+
+            key = get_jwt_signing_key(mock_service)
+
+            assert key == mock_service.auth_settings.SECRET_KEY.get_secret_value()
+
+    def test_get_jwt_signing_key_rs256_returns_private_key(self):
+        """RS256 should return private key for signing."""
+        from langflow.services.auth.utils import get_jwt_signing_key
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("RS256", tmpdir)
+
+            key = get_jwt_signing_key(mock_service)
+
+            assert key == mock_service.auth_settings.PRIVATE_KEY.get_secret_value()
+            assert "-----BEGIN PRIVATE KEY-----" in key
+
+    def test_get_jwt_signing_key_rs512_returns_private_key(self):
+        """RS512 should return private key for signing."""
+        from langflow.services.auth.utils import get_jwt_signing_key
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("RS512", tmpdir)
+
+            key = get_jwt_signing_key(mock_service)
+
+            assert key == mock_service.auth_settings.PRIVATE_KEY.get_secret_value()
+            assert "-----BEGIN PRIVATE KEY-----" in key
+
+    def test_verification_and_signing_keys_work_together_hs256(self):
+        """Verification and signing keys should work together for HS256."""
+        from langflow.services.auth.utils import get_jwt_signing_key, get_jwt_verification_key
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("HS256", tmpdir)
+
+            signing_key = get_jwt_signing_key(mock_service)
+            verification_key = get_jwt_verification_key(mock_service)
+
+            # For symmetric algorithms, both keys are the same
+            assert signing_key == verification_key
+
+            # Sign and verify a token
+            payload = {"sub": "test-user", "type": "access"}
+            token = jwt.encode(payload, signing_key, algorithm="HS256")
+            decoded = jwt.decode(token, verification_key, algorithms=["HS256"])
+
+            assert decoded["sub"] == "test-user"
+
+    def test_verification_and_signing_keys_work_together_rs256(self):
+        """Verification and signing keys should work together for RS256."""
+        from langflow.services.auth.utils import get_jwt_signing_key, get_jwt_verification_key
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_service = self._create_mock_settings_service("RS256", tmpdir)
+
+            signing_key = get_jwt_signing_key(mock_service)
+            verification_key = get_jwt_verification_key(mock_service)
+
+            # For asymmetric algorithms, keys are different
+            assert signing_key != verification_key
+
+            # Sign and verify a token
+            payload = {"sub": "test-user", "type": "access"}
+            token = jwt.encode(payload, signing_key, algorithm="RS256")
+            decoded = jwt.decode(token, verification_key, algorithms=["RS256"])
+
+            assert decoded["sub"] == "test-user"

src/backend/tests/unit/test_security_cors.py

@@ -8,6 +8,7 @@ from unittest.mock import MagicMock, patch
 import pytest
 from fastapi import HTTPException
 from fastapi.middleware.cors import CORSMiddleware
+from lfx.services.settings.auth import JWTAlgorithm
 from lfx.services.settings.base import Settings
 
 
@@ -290,7 +291,7 @@ class TestRefreshTokenSecurity:
 
             with patch("langflow.services.auth.utils.get_settings_service") as mock_settings:
                 mock_settings.return_value.auth_settings.SECRET_KEY.get_secret_value.return_value = "secret"
-                mock_settings.return_value.auth_settings.ALGORITHM = "HS256"
+                mock_settings.return_value.auth_settings.ALGORITHM = JWTAlgorithm.HS256
                 mock_settings.return_value.auth_settings.ACCESS_TOKEN_EXPIRE_SECONDS = 3600
                 mock_settings.return_value.auth_settings.REFRESH_TOKEN_EXPIRE_SECONDS = 604800
 

src/lfx/pyproject.toml

@@ -41,6 +41,7 @@ dependencies = [
     "validators>=0.34.0,<1.0.0",
     "filelock>=3.20.0",
     "pypdf>=5.1.0",
+    "cryptography>=43.0.0",
 ]
 
 [project.scripts]

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

@@ -1,14 +1,33 @@
 import secrets
+from enum import Enum
 from pathlib import Path
 from typing import Literal
 
 from passlib.context import CryptContext
-from pydantic import Field, SecretStr, field_validator
+from pydantic import Field, SecretStr, field_validator, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 from lfx.log.logger import logger
 from lfx.services.settings.constants import DEFAULT_SUPERUSER, DEFAULT_SUPERUSER_PASSWORD
-from lfx.services.settings.utils import read_secret_from_file, write_secret_to_file
+from lfx.services.settings.utils import (
+    derive_public_key_from_private,
+    generate_rsa_key_pair,
+    read_secret_from_file,
+    write_public_key_to_file,
+    write_secret_to_file,
+)
+
+
+class JWTAlgorithm(str, Enum):
+    """JWT signing algorithm options."""
+
+    HS256 = "HS256"
+    RS256 = "RS256"
+    RS512 = "RS512"
+
+    def is_asymmetric(self) -> bool:
+        """Return True if this algorithm uses asymmetric (public/private key) cryptography."""
+        return self in (JWTAlgorithm.RS256, JWTAlgorithm.RS512)
 
 
 class AuthSettings(BaseSettings):
@@ -16,10 +35,22 @@ class AuthSettings(BaseSettings):
     CONFIG_DIR: str
     SECRET_KEY: SecretStr = Field(
         default=SecretStr(""),
-        description="Secret key for JWT. If not provided, a random one will be generated.",
+        description="Secret key for JWT (used with HS256). If not provided, a random one will be generated.",
         frozen=False,
     )
-    ALGORITHM: str = "HS256"
+    PRIVATE_KEY: SecretStr = Field(
+        default=SecretStr(""),
+        description="RSA private key for JWT signing (RS256/RS512). Auto-generated if not provided.",
+        frozen=False,
+    )
+    PUBLIC_KEY: str = Field(
+        default="",
+        description="RSA public key for JWT verification (RS256/RS512). Derived from private key if not provided.",
+    )
+    ALGORITHM: JWTAlgorithm = Field(
+        default=JWTAlgorithm.HS256,
+        description="JWT signing algorithm. Use RS256 or RS512 for asymmetric signing (recommended for production).",
+    )
     ACCESS_TOKEN_EXPIRE_SECONDS: int = 60 * 60  # 1 hour
     REFRESH_TOKEN_EXPIRE_SECONDS: int = 60 * 60 * 24 * 7  # 7 days
 
@@ -138,3 +169,63 @@ class AuthSettings(BaseSettings):
             logger.debug("Saved secret key")
 
         return value if isinstance(value, SecretStr) else SecretStr(value).get_secret_value()
+
+    @model_validator(mode="after")
+    def setup_rsa_keys(self):
+        """Generate or load RSA keys when using RS256/RS512 algorithm."""
+        if not self.ALGORITHM.is_asymmetric():
+            return self
+
+        config_dir = self.CONFIG_DIR
+        private_key_value = self.PRIVATE_KEY.get_secret_value() if self.PRIVATE_KEY else ""
+
+        if not config_dir:
+            # No config dir - generate keys in memory if not provided
+            if not private_key_value:
+                logger.debug("No CONFIG_DIR provided, generating RSA keys in memory")
+                private_key_pem, public_key_pem = generate_rsa_key_pair()
+                object.__setattr__(self, "PRIVATE_KEY", SecretStr(private_key_pem))
+                object.__setattr__(self, "PUBLIC_KEY", public_key_pem)
+            elif not self.PUBLIC_KEY:
+                # Derive public key from private key
+                public_key_pem = derive_public_key_from_private(private_key_value)
+                object.__setattr__(self, "PUBLIC_KEY", public_key_pem)
+            return self
+
+        private_key_path = Path(config_dir) / "private_key.pem"
+        public_key_path = Path(config_dir) / "public_key.pem"
+
+        if private_key_value:
+            # Private key provided via env var - save it and derive public key
+            logger.debug("RSA private key provided")
+            write_secret_to_file(private_key_path, private_key_value)
+
+            if not self.PUBLIC_KEY:
+                public_key_pem = derive_public_key_from_private(private_key_value)
+                object.__setattr__(self, "PUBLIC_KEY", public_key_pem)
+                write_public_key_to_file(public_key_path, public_key_pem)
+        # No private key provided - load from file or generate
+        elif private_key_path.exists():
+            logger.debug("Loading RSA keys from files")
+            private_key_pem = read_secret_from_file(private_key_path)
+            object.__setattr__(self, "PRIVATE_KEY", SecretStr(private_key_pem))
+
+            if public_key_path.exists():
+                public_key_pem = public_key_path.read_text(encoding="utf-8")
+                object.__setattr__(self, "PUBLIC_KEY", public_key_pem)
+            else:
+                # Derive public key from private key
+                public_key_pem = derive_public_key_from_private(private_key_pem)
+                object.__setattr__(self, "PUBLIC_KEY", public_key_pem)
+                write_public_key_to_file(public_key_path, public_key_pem)
+        else:
+            # Generate new RSA key pair
+            logger.debug("Generating new RSA key pair")
+            private_key_pem, public_key_pem = generate_rsa_key_pair()
+            write_secret_to_file(private_key_path, private_key_pem)
+            write_public_key_to_file(public_key_path, public_key_pem)
+            object.__setattr__(self, "PRIVATE_KEY", SecretStr(private_key_pem))
+            object.__setattr__(self, "PUBLIC_KEY", public_key_pem)
+            logger.debug("RSA key pair generated and saved")
+
+        return self

src/lfx/src/lfx/services/settings/utils.py

@@ -1,9 +1,74 @@
 import platform
 from pathlib import Path
 
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.serialization import load_pem_private_key
+
 from lfx.log.logger import logger
 
 
+class RSAKeyError(Exception):
+    """Exception raised when RSA key operations fail."""
+
+
+def derive_public_key_from_private(private_key_pem: str) -> str:
+    """Derive a public key from a private key PEM string.
+
+    Args:
+        private_key_pem: The private key in PEM format.
+
+    Returns:
+        str: The public key in PEM format.
+
+    Raises:
+        RSAKeyError: If the private key is invalid or cannot be processed.
+    """
+    try:
+        private_key = load_pem_private_key(private_key_pem.encode(), password=None)
+        return (
+            private_key.public_key()
+            .public_bytes(
+                encoding=serialization.Encoding.PEM,
+                format=serialization.PublicFormat.SubjectPublicKeyInfo,
+            )
+            .decode("utf-8")
+        )
+    except Exception as e:
+        msg = f"Failed to derive public key from private key: {e}"
+        logger.error(msg)
+        raise RSAKeyError(msg) from e
+
+
+def generate_rsa_key_pair() -> tuple[str, str]:
+    """Generate an RSA key pair for RS256 JWT signing.
+
+    Returns:
+        tuple[str, str]: A tuple of (private_key_pem, public_key_pem) as strings.
+    """
+    private_key = rsa.generate_private_key(
+        public_exponent=65537,
+        key_size=2048,
+    )
+
+    private_key_pem = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
+    ).decode("utf-8")
+
+    public_key_pem = (
+        private_key.public_key()
+        .public_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PublicFormat.SubjectPublicKeyInfo,
+        )
+        .decode("utf-8")
+    )
+
+    return private_key_pem, public_key_pem
+
+
 def set_secure_permissions(file_path: Path) -> None:
     if platform.system() in {"Linux", "Darwin"}:  # Unix/Linux/Mac
         file_path.chmod(0o600)
@@ -38,3 +103,20 @@ def write_secret_to_file(path: Path, value: str) -> None:
 
 def read_secret_from_file(path: Path) -> str:
     return path.read_text(encoding="utf-8")
+
+
+def write_public_key_to_file(path: Path, value: str) -> None:
+    """Write a public key to file with appropriate permissions (0o644).
+
+    Public keys can be readable by others but should only be writable by owner.
+
+    Args:
+        path: The file path to write to.
+        value: The public key content.
+    """
+    path.write_text(value, encoding="utf-8")
+    try:
+        if platform.system() in {"Linux", "Darwin"}:
+            path.chmod(0o644)
+    except Exception:  # noqa: BLE001
+        logger.exception("Failed to set permissions on public key file")

uv.lock

@@ -1,5 +1,5 @@
 version = 1
-revision = 3
+revision = 2
 requires-python = ">=3.10, <3.14"
 resolution-markers = [
     "python_full_version >= '3.13' and platform_machine == 'arm64' and sys_platform == 'darwin'",
@@ -6402,6 +6402,7 @@ dependencies = [
     { name = "asyncer" },
     { name = "cachetools" },
     { name = "chardet" },
+    { name = "cryptography" },
     { name = "defusedxml" },
     { name = "docstring-parser" },
     { name = "emoji" },
@@ -6457,6 +6458,7 @@ requires-dist = [
     { name = "asyncer", specifier = ">=0.0.8,<1.0.0" },
     { name = "cachetools", specifier = ">=6.0.0" },
     { name = "chardet", specifier = ">=5.2.0,<6.0.0" },
+    { name = "cryptography", specifier = ">=43.0.0" },
     { name = "defusedxml", specifier = ">=0.7.1,<1.0.0" },
     { name = "docstring-parser", specifier = ">=0.16,<1.0.0" },
     { name = "emoji", specifier = ">=2.14.1,<3.0.0" },