feat: encrypt oauth auth settings at rest

src/backend/base/langflow/alembic/versions/0882f9657f22_encrypt_existing_mcp_auth_settings_.py

@@ -0,0 +1,122 @@
+"""Encrypt existing MCP auth_settings credentials
+
+Revision ID: 0882f9657f22
+Revises: 1cb603706752
+Create Date: 2025-08-21 20:11:26.504681
+
+"""
+import json
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+from sqlalchemy.engine.reflection import Inspector
+from langflow.utils import migration
+
+
+# revision identifiers, used by Alembic.
+revision: str = '0882f9657f22'
+down_revision: Union[str, None] = '1cb603706752'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Encrypt sensitive fields in existing auth_settings data."""
+    conn = op.get_bind()
+
+    # Import encryption utilities
+    try:
+        from langflow.services.auth.mcp_encryption import encrypt_auth_settings
+        from langflow.services.deps import get_settings_service
+
+        # Check if the folder table exists
+        inspector = sa.inspect(conn)
+        if 'folder' not in inspector.get_table_names():
+            return
+
+        # Query all folders with auth_settings
+        result = conn.execute(
+            sa.text("SELECT id, auth_settings FROM folder WHERE auth_settings IS NOT NULL")
+        )
+
+        # Encrypt auth_settings for each folder
+        for row in result:
+            folder_id = row.id
+            auth_settings = row.auth_settings
+
+            if auth_settings:
+                try:
+                    # Parse JSON if it's a string
+                    if isinstance(auth_settings, str):
+                        auth_settings_dict = json.loads(auth_settings)
+                    else:
+                        auth_settings_dict = auth_settings
+
+                    # Encrypt sensitive fields
+                    encrypted_settings = encrypt_auth_settings(auth_settings_dict)
+
+                    # Update the record with encrypted data
+                    if encrypted_settings:
+                        conn.execute(
+                            sa.text("UPDATE folder SET auth_settings = :auth_settings WHERE id = :id"),
+                            {"auth_settings": json.dumps(encrypted_settings), "id": folder_id}
+                        )
+                except Exception as e:
+                    # Log the error but continue with other records
+                    print(f"Warning: Failed to encrypt auth_settings for folder {folder_id}: {e}")
+
+    except ImportError as e:
+        # If encryption utilities are not available, skip the migration
+        print(f"Warning: Encryption utilities not available, skipping encryption migration: {e}")
+
+
+def downgrade() -> None:
+    """Decrypt sensitive fields in auth_settings data (for rollback)."""
+    conn = op.get_bind()
+
+    # Import decryption utilities
+    try:
+        from langflow.services.auth.mcp_encryption import decrypt_auth_settings
+        from langflow.services.deps import get_settings_service
+
+        # Check if the folder table exists
+        inspector = sa.inspect(conn)
+        if 'folder' not in inspector.get_table_names():
+            return
+
+        # Query all folders with auth_settings
+        result = conn.execute(
+            sa.text("SELECT id, auth_settings FROM folder WHERE auth_settings IS NOT NULL")
+        )
+
+        # Decrypt auth_settings for each folder
+        for row in result:
+            folder_id = row.id
+            auth_settings = row.auth_settings
+
+            if auth_settings:
+                try:
+                    # Parse JSON if it's a string
+                    if isinstance(auth_settings, str):
+                        auth_settings_dict = json.loads(auth_settings)
+                    else:
+                        auth_settings_dict = auth_settings
+
+                    # Decrypt sensitive fields
+                    decrypted_settings = decrypt_auth_settings(auth_settings_dict)
+
+                    # Update the record with decrypted data
+                    if decrypted_settings:
+                        conn.execute(
+                            sa.text("UPDATE folder SET auth_settings = :auth_settings WHERE id = :id"),
+                            {"auth_settings": json.dumps(decrypted_settings), "id": folder_id}
+                        )
+                except Exception as e:
+                    # Log the error but continue with other records
+                    print(f"Warning: Failed to decrypt auth_settings for folder {folder_id}: {e}")
+
+    except ImportError as e:
+        # If decryption utilities are not available, skip the migration
+        print(f"Warning: Decryption utilities not available, skipping decryption migration: {e}")

src/backend/base/langflow/api/v1/mcp_projects.py

@@ -8,7 +8,7 @@
 from ipaddress import ip_address
 from pathlib import Path
 from subprocess import CalledProcessError
-from typing import Annotated
+from typing import Annotated, Any
 from uuid import UUID
 
 from anyio import BrokenResourceError
@@ -39,6 +39,7 @@
 from langflow.base.mcp.constants import MAX_MCP_SERVER_NAME_LENGTH
 from langflow.base.mcp.util import sanitize_mcp_name
 from langflow.logging import logger
+from langflow.services.auth.mcp_encryption import decrypt_auth_settings, encrypt_auth_settings
 from langflow.services.database.models import Flow, Folder
 from langflow.services.database.models.api_key.crud import check_key, create_api_key
 from langflow.services.database.models.api_key.model import ApiKeyCreate
@@ -205,7 +206,7 @@ async def list_project_tools(
                 )
                 try:
                     tool = MCPSettings(
-                        id=str(flow.id),
+                        id=flow.id,
                         action_name=name,
                         action_description=description,
                         mcp_enabled=flow.mcp_enabled,
@@ -219,10 +220,14 @@ async def list_project_tools(
                     await logger.awarning(msg)
                     continue
 
-            # Get project-level auth settings
+            # Get project-level auth settings and decrypt sensitive fields
             auth_settings = None
             if project.auth_settings:
-                auth_settings = AuthSettings(**project.auth_settings)
+                from langflow.api.v1.schemas import AuthSettings
+
+                # Decrypt sensitive fields before returning
+                decrypted_settings = decrypt_auth_settings(project.auth_settings)
+                auth_settings = AuthSettings(**decrypted_settings) if decrypted_settings else None
 
     except Exception as e:
         msg = f"Error listing project tools: {e!s}"
@@ -336,11 +341,33 @@ async def update_project_mcp_settings(
             if not project:
                 raise HTTPException(status_code=404, detail="Project not found")
 
-            # Update project-level auth settings
-            if request.auth_settings:
-                project.auth_settings = request.auth_settings.model_dump(mode="json")
-            else:
-                project.auth_settings = None
+            # Update project-level auth settings with encryption
+            if "auth_settings" in request.model_fields_set:
+                if request.auth_settings is None:
+                    # Explicitly set to None - clear auth settings
+                    project.auth_settings = None
+                else:
+                    # Use python mode to get raw values without SecretStr masking
+                    auth_model = request.auth_settings
+                    auth_dict = auth_model.model_dump(mode="python", exclude_none=True)
+
+                    # Extract actual secret values before encryption
+                    from pydantic import SecretStr
+
+                    # Handle api_key if it's a SecretStr
+                    api_key_val = getattr(auth_model, "api_key", None)
+                    if isinstance(api_key_val, SecretStr):
+                        auth_dict["api_key"] = api_key_val.get_secret_value()
+
+                    # Handle oauth_client_secret if it's a SecretStr
+                    client_secret_val = getattr(auth_model, "oauth_client_secret", None)
+                    if isinstance(client_secret_val, SecretStr):
+                        auth_dict["oauth_client_secret"] = client_secret_val.get_secret_value()
+
+                    # Encrypt and store
+                    encrypted_settings = encrypt_auth_settings(auth_dict)
+                    project.auth_settings = encrypted_settings
+
             session.add(project)
 
             # Query flows in the project
@@ -458,7 +485,7 @@ async def install_mcp_config(
                 should_generate_api_key = not settings_service.auth_settings.AUTO_LOGIN
             elif project.auth_settings:
                 # When MCP_COMPOSER is enabled, only generate if auth_type is "apikey"
-                auth_settings = AuthSettings(**project.auth_settings)
+                auth_settings = AuthSettings(**project.auth_settings) if project.auth_settings else AuthSettings()
                 should_generate_api_key = auth_settings.auth_type == "apikey"
 
             if should_generate_api_key:
@@ -498,7 +525,7 @@ async def install_mcp_config(
                         stdout=asyncio.subprocess.PIPE,
                         stderr=asyncio.subprocess.PIPE,
                     )
-                    stdout, stderr = await proc.communicate()
+                    stdout, _ = await proc.communicate()
 
                     if proc.returncode == 0 and stdout.strip():
                         wsl_ip = stdout.decode().strip().split()[0]  # Get first IP address
@@ -508,8 +535,8 @@ async def install_mcp_config(
                 except OSError as e:
                     await logger.awarning("Failed to get WSL IP address: %s. Using default URL.", str(e))
 
-        # Build the base args for mcp-proxy
-        args = ["mcp-proxy"]
+        # Base args
+        args = ["mcp-composer"] if FEATURE_FLAGS.mcp_composer else ["mcp-proxy"]
 
         # Add authentication args based on MCP_COMPOSER feature flag and auth settings
         if not FEATURE_FLAGS.mcp_composer:
@@ -518,14 +545,22 @@ async def install_mcp_config(
             if generated_api_key:
                 args.extend(["--headers", "x-api-key", generated_api_key])
         elif project.auth_settings:
+            # Decrypt sensitive fields before using them
+            decrypted_settings = decrypt_auth_settings(project.auth_settings)
+            auth_settings = AuthSettings(**decrypted_settings) if decrypted_settings else AuthSettings()
+            args.extend(["--auth_type", auth_settings.auth_type])
+
             # When MCP_COMPOSER is enabled, only add headers if auth_type is "apikey"
             auth_settings = AuthSettings(**project.auth_settings)
             if auth_settings.auth_type == "apikey" and generated_api_key:
                 args.extend(["--headers", "x-api-key", generated_api_key])
             # If no auth_settings or auth_type is "none", don't add any auth headers
 
         # Add the SSE URL
-        args.append(sse_url)
+        if FEATURE_FLAGS.mcp_composer:
+            args.extend(["--sse-url", sse_url])
+        else:
+            args.append(sse_url)
 
         if os_type == "Windows":
             command = "cmd"
@@ -535,7 +570,7 @@ async def install_mcp_config(
         name = project.name
 
         # Create the MCP configuration
-        server_config = {
+        server_config: dict[str, Any] = {
             "command": command,
             "args": args,
         }

src/backend/base/langflow/api/v1/schemas.py

@@ -8,6 +8,7 @@
     BaseModel,
     ConfigDict,
     Field,
+    SecretStr,
     field_serializer,
     field_validator,
     model_serializer,
@@ -449,7 +450,7 @@ class AuthSettings(BaseModel):
     oauth_server_url: str | None = None
     oauth_callback_path: str | None = None
     oauth_client_id: str | None = None
-    oauth_client_secret: str | None = None
+    oauth_client_secret: SecretStr | None = None
     oauth_auth_url: str | None = None
     oauth_token_url: str | None = None
     oauth_mcp_scope: str | None = None

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

@@ -0,0 +1,104 @@
+"""MCP Authentication encryption utilities for secure credential storage."""
+
+from typing import Any
+
+from cryptography.fernet import InvalidToken
+from loguru import logger
+
+from langflow.services.auth import utils as auth_utils
+from langflow.services.deps import get_settings_service
+
+# Fields that should be encrypted when stored
+SENSITIVE_FIELDS = [
+    "oauth_client_secret",
+    "api_key",
+]
+
+
+def encrypt_auth_settings(auth_settings: dict[str, Any] | None) -> dict[str, Any] | None:
+    """Encrypt sensitive fields in auth_settings dictionary.
+
+    Args:
+        auth_settings: Dictionary containing authentication settings
+
+    Returns:
+        Dictionary with sensitive fields encrypted, or None if input is None
+    """
+    if auth_settings is None:
+        return None
+
+    settings_service = get_settings_service()
+    encrypted_settings = auth_settings.copy()
+
+    for field in SENSITIVE_FIELDS:
+        if encrypted_settings.get(field):
+            try:
+                # Only encrypt if the value is not already encrypted
+                # Try to decrypt first - if it fails, it's not encrypted
+                try:
+                    auth_utils.decrypt_api_key(encrypted_settings[field], settings_service)
+                    # If decrypt succeeds, it's already encrypted
+                    logger.debug(f"Field {field} is already encrypted")
+                except (ValueError, TypeError, KeyError, InvalidToken):
+                    # If decrypt fails, the value is plaintext and needs encryption
+                    encrypted_value = auth_utils.encrypt_api_key(encrypted_settings[field], settings_service)
+                    encrypted_settings[field] = encrypted_value
+                    logger.debug(f"Encrypted field {field}")
+            except (ValueError, TypeError, KeyError) as e:
+                logger.error(f"Failed to encrypt field {field}: {e}")
+                raise
+
+    return encrypted_settings
+
+
+def decrypt_auth_settings(auth_settings: dict[str, Any] | None) -> dict[str, Any] | None:
+    """Decrypt sensitive fields in auth_settings dictionary.
+
+    Args:
+        auth_settings: Dictionary containing encrypted authentication settings
+
+    Returns:
+        Dictionary with sensitive fields decrypted, or None if input is None
+    """
+    if auth_settings is None:
+        return None
+
+    settings_service = get_settings_service()
+    decrypted_settings = auth_settings.copy()
+
+    for field in SENSITIVE_FIELDS:
+        if decrypted_settings.get(field):
+            try:
+                decrypted_value = auth_utils.decrypt_api_key(decrypted_settings[field], settings_service)
+                decrypted_settings[field] = decrypted_value
+                logger.debug(f"Decrypted field {field}")
+            except (ValueError, TypeError, KeyError, InvalidToken) as e:
+                # If decryption fails, assume the value is already plaintext
+                # This handles backward compatibility with existing unencrypted data
+                logger.debug(f"Field {field} appears to be plaintext or decryption failed: {e}")
+                # Keep the original value
+
+    return decrypted_settings
+
+
+def is_encrypted(value: str) -> bool:
+    """Check if a value appears to be encrypted.
+
+    Args:
+        value: String value to check
+
+    Returns:
+        True if the value appears to be encrypted (base64 Fernet token)
+    """
+    if not value:
+        return False
+
+    settings_service = get_settings_service()
+    try:
+        # Try to decrypt - if it succeeds, it's encrypted
+        auth_utils.decrypt_api_key(value, settings_service)
+    except (ValueError, TypeError, KeyError, InvalidToken):
+        # If decryption fails, it's not encrypted
+        return False
+    else:
+        return True