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

@@ -37,6 +37,7 @@
 )
 from langflow.base.mcp.constants import MAX_MCP_SERVER_NAME_LENGTH
 from langflow.base.mcp.util import sanitize_mcp_name
+from langflow.services.auth.mcp_encryption import decrypt_auth_settings, encrypt_auth_settings
 from langflow.services.database.models import Flow, Folder
 from langflow.services.deps import get_settings_service, session_scope
 
@@ -119,12 +120,14 @@
                     logger.warning(msg)
                     continue
 
-            # Get project-level auth settings
+            # Get project-level auth settings and decrypt sensitive fields
             auth_settings = None
             if project.auth_settings:
                 from langflow.api.v1.schemas import AuthSettings
 
-                auth_settings = AuthSettings(**project.auth_settings)
+                # Decrypt sensitive fields before returning
+                decrypted_settings = decrypt_auth_settings(project.auth_settings)
+                auth_settings = AuthSettings(**decrypted_settings)
 
     except Exception as e:
         msg = f"Error listing project tools: {e!s}"
@@ -248,9 +251,12 @@
             if not project:
                 raise HTTPException(status_code=404, detail="Project not found")
 
-            # Update project-level auth settings
+            # Update project-level auth settings with encryption
             if request.auth_settings:
-                project.auth_settings = request.auth_settings.model_dump(mode="json")
+                # Encrypt sensitive fields before storing
+                auth_dict = request.auth_settings.model_dump(mode="json")
+                encrypted_settings = encrypt_auth_settings(auth_dict)
+                project.auth_settings = encrypted_settings
             else:
                 project.auth_settings = None
             session.add(project)
@@ -396,8 +402,31 @@
                         sse_url = sse_url.replace(f"http://{host}:{port}", f"http://{wsl_ip}:{port}")
                 except OSError as e:
                     logger.warning("Failed to get WSL IP address: %s. Using default URL.", str(e))
-        else:
-            args = ["mcp-proxy", sse_url]
+
+        args = ["mcp-proxy", "--sse-url", sse_url]
+
+        oauth_env = None
+
+        if project.auth_settings:
+            from langflow.api.v1.schemas import AuthSettings
+
+            # Decrypt sensitive fields before using them
+            decrypted_settings = decrypt_auth_settings(project.auth_settings)
+            auth_settings = AuthSettings(**decrypted_settings)
+            args.extend(["--auth_type", auth_settings.auth_type])
+
+            oauth_env = {
+                "OAUTH_HOST": auth_settings.oauth_host,
+                "OAUTH_PORT": auth_settings.oauth_port,
+                "OAUTH_SERVER_URL": auth_settings.oauth_server_url,
+                "OAUTH_CALLBACK_PATH": auth_settings.oauth_callback_path,
+                "OAUTH_CLIENT_ID": auth_settings.oauth_client_id,
+                "OAUTH_CLIENT_SECRET": auth_settings.oauth_client_secret,
+                "OAUTH_AUTH_URL": auth_settings.oauth_auth_url,
+                "OAUTH_TOKEN_URL": auth_settings.oauth_token_url,
+                "OAUTH_MCP_SCOPE": auth_settings.oauth_mcp_scope,
+                "OAUTH_PROVIDER_SCOPE": auth_settings.oauth_provider_scope,
+            }
 
         if os_type == "Windows":
             command = "cmd"

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

@@ -0,0 +1,107 @@
+"""MCP Authentication encryption utilities for secure credential storage."""
+
+from typing import Any
+
+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",
+    "oauth_token",
+    "bearer_token",
+    "password",
+    "api_key",
+    "access_token",
+    "refresh_token",
+]
+
+
+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 Exception:
+                    # 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 Exception 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 Exception 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
+
+    try:
+        settings_service = get_settings_service()
+        # Try to decrypt - if it succeeds, it's encrypted
+        auth_utils.decrypt_api_key(value, settings_service)
+        return True
+    except Exception:
+        # If decryption fails, it's not encrypted
+        return False

src/backend/tests/unit/api/v1/test_mcp_projects.py

@@ -444,6 +444,73 @@
         assert updated_flow.mcp_enabled is False
 
 
+async def test_update_project_auth_settings_encryption(
+    client: AsyncClient, user_test_project, test_flow_for_update, logged_in_headers
+):
+    """Test that sensitive auth_settings fields are encrypted when stored."""
+    # Create settings with sensitive data
+    json_payload = {
+        "settings": [
+            {
+                "id": str(test_flow_for_update.id),
+                "action_name": "test_action",
+                "action_description": "Test description",
+                "mcp_enabled": True,
+                "name": test_flow_for_update.name,
+                "description": test_flow_for_update.description,
+            }
+        ],
+        "auth_settings": {
+            "auth_type": "oauth",
+            "oauth_host": "localhost",
+            "oauth_port": "3000",
+            "oauth_server_url": "http://localhost:3000",
+            "oauth_callback_path": "/callback",
+            "oauth_client_id": "test-client-id",
+            "oauth_client_secret": "super-secret-password-123",
+            "oauth_auth_url": "https://oauth.example.com/auth",
+            "oauth_token_url": "https://oauth.example.com/token",
+            "oauth_mcp_scope": "read write",
+            "oauth_provider_scope": "user:email",
+        },
+    }
+
+    # Send the update request
+    response = await client.patch(
+        f"/api/v1/mcp/project/{user_test_project.id}",
+        json=json_payload,
+        headers=logged_in_headers,
+    )
+    assert response.status_code == 200
+
+    # Verify the sensitive data is encrypted in the database
+    async with session_scope() as session:
+        updated_project = await session.get(Folder, user_test_project.id)
+        assert updated_project is not None
+        assert updated_project.auth_settings is not None
+
+        # Check that sensitive field is encrypted (not plaintext)
+        stored_secret = updated_project.auth_settings.get("oauth_client_secret")
+        assert stored_secret is not None
+        assert stored_secret != "super-secret-password-123"  # Should be encrypted
+
+        # The encrypted value should be a base64-like string (Fernet token)
+        assert len(stored_secret) > 50  # Encrypted values are longer
+
+    # Now test that the GET endpoint decrypts the data correctly
+    response = await client.get(
+        f"/api/v1/mcp/project/{user_test_project.id}",
+        headers=logged_in_headers,
+    )
+    assert response.status_code == 200
+    data = response.json()
+
+    # The decrypted value should match the original
+    assert data["auth_settings"]["oauth_client_secret"] == "super-secret-password-123"
+    assert data["auth_settings"]["oauth_client_id"] == "test-client-id"
+    assert data["auth_settings"]["auth_type"] == "oauth"
+
+
 async def test_project_sse_creation(user_test_project):
     """Test that SSE transport and MCP server are correctly created for a project."""
     # Test getting an SSE transport for the first time