fix: update CORS configuration and add env vars to allow for user control (#9744)

* Update CORS configuration and add env vars to allow for user control

* Add unit test

* fix: add noqa comment for linting in create_refresh_token function

Added a noqa comment to suppress linting warnings for the token type check in the create_refresh_token function, ensuring code clarity while maintaining compliance with linting standards.

* refactor: enhance CORS tests with temporary directory usage and improve exception handling

Updated unit tests for CORS configuration to utilize a temporary directory for environment variables, ensuring isolation during tests. Improved exception handling in refresh token tests to raise HTTPException with appropriate status codes and messages. This enhances test reliability and clarity.

* fix: enhance CORS origin validation to ensure consistent list output

Updated the validate_cors_origins method to convert a single origin string into a list for consistency, in addition to handling comma-separated strings. This change improves the robustness of CORS configuration handling.

* refactor: update CORS test cases to ensure single origin is consistently treated as a list

Modified unit tests for CORS configuration to assert that single origin strings are converted to lists, enhancing consistency in CORS origin handling. This change aligns with recent updates to the CORS validation logic.

* re-add uvlock

* [autofix.ci] apply automated fixes

* feat: revert CORS changes and add deprecation warnings for v1.7

- Revert CORS behavior to maintain backward compatibility until v1.7
- Add CORS configuration fields with current defaults (*, True, *, *)
- Implement deprecation warnings for upcoming v1.7 security changes
- Preserve existing behavior while enabling early configuration adoption

* [autofix.ci] apply automated fixes

* Add back imports

* Fix tests

* [autofix.ci] apply automated fixes

* Rt;uff

* Ruff

---------

Co-authored-by: Gabriel Luiz Freitas Almeida <gabriel@langflow.org>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: italojohnny <italojohnnydosanjos@gmail.com>
Co-authored-by: Lucas Oliveira <62335616+lucaseduoli@users.noreply.github.com>

docker/build_and_push.Dockerfile

@@ -93,3 +93,4 @@ ENV LANGFLOW_HOST=0.0.0.0
 ENV LANGFLOW_PORT=7860
 
 CMD ["langflow", "run"]
+

src/backend/base/langflow/api/build.py

@@ -411,7 +411,7 @@ async def generate_flow_events(
         try:
             vertex_build_response: VertexBuildResponse = await _build_vertex(vertex_id, graph, event_manager)
         except asyncio.CancelledError as exc:
-            await logger.aerror(f"Build cancelled: {exc}")
+            await logger.ainfo(f"Build cancelled: {exc}")
             raise
 
         # send built event or error event

src/backend/base/langflow/main.py

@@ -117,6 +117,28 @@ async def load_bundles_with_error_handling():
         return [], []
 
 
+def warn_about_future_cors_changes(settings):
+    """Warn users about upcoming CORS security changes in version 1.7."""
+    # Check if using default (backward compatible) settings
+    using_defaults = settings.cors_origins == "*" and settings.cors_allow_credentials is True
+
+    if using_defaults:
+        logger.warning(
+            "DEPRECATION NOTICE: Starting in v1.7, CORS will be more restrictive by default. "
+            "Current behavior allows all origins (*) with credentials enabled. "
+            "Consider setting LANGFLOW_CORS_ORIGINS for production deployments. "
+            "See documentation for secure CORS configuration."
+        )
+
+    # Additional warning for potentially insecure configuration
+    if settings.cors_origins == "*" and settings.cors_allow_credentials:
+        logger.warning(
+            "SECURITY NOTICE: Current CORS configuration allows all origins with credentials. "
+            "In v1.7, credentials will be automatically disabled when using wildcard origins. "
+            "Specify exact origins in LANGFLOW_CORS_ORIGINS to use credentials securely."
+        )
+
+
 def get_lifespan(*, fix_migration=False, version=None):
     @asynccontextmanager
     async def lifespan(_app: FastAPI):
@@ -337,14 +359,24 @@ def create_app():
     )
 
     setup_sentry(app)
-    origins = ["*"]
 
+    settings = get_settings_service().settings
+
+    # Warn about future CORS changes
+    warn_about_future_cors_changes(settings)
+
+    # Configure CORS using settings (with backward compatible defaults)
+    origins = settings.cors_origins
+    if isinstance(origins, str) and origins != "*":
+        origins = [origins]
+
+    # Apply current CORS configuration (maintains backward compatibility)
     app.add_middleware(
         CORSMiddleware,
         allow_origins=origins,
-        allow_credentials=True,
-        allow_methods=["*"],
-        allow_headers=["*"],
+        allow_credentials=settings.cors_allow_credentials,
+        allow_methods=settings.cors_allow_methods,
+        allow_headers=settings.cors_allow_headers,
     )
     app.add_middleware(JavaScriptMIMETypeMiddleware)
 
@@ -393,7 +425,6 @@ def create_app():
 
         return await call_next(request)
 
-    settings = get_settings_service().settings
     if prome_port_str := os.environ.get("LANGFLOW_PROMETHEUS_PORT"):
         # set here for create_app() entry point
         prome_port = int(prome_port_str)

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

@@ -408,7 +408,7 @@ async def create_refresh_token(refresh_token: str, db: AsyncSession):
         user_id: UUID = payload.get("sub")  # type: ignore[assignment]
         token_type: str = payload.get("type")  # type: ignore[assignment]
 
-        if user_id is None or token_type == "":
+        if user_id is None or token_type != "refresh":  # noqa: S105
             raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token")
 
         user_exists = await get_user_by_id(db, user_id)
@@ -416,6 +416,10 @@ async def create_refresh_token(refresh_token: str, db: AsyncSession):
         if user_exists is None:
             raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token")
 
+        # Security: Check if user is still active
+        if not user_exists.is_active:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User account is inactive")
+
         return await create_user_tokens(user_id, db)
 
     except JWTError as e:

src/backend/base/langflow/services/settings/base.py

@@ -202,6 +202,18 @@ class Settings(BaseSettings):
     backend_only: bool = False
     """If set to True, Langflow will not serve the frontend."""
 
+    # CORS Settings
+    cors_origins: list[str] | str = "*"
+    """Allowed origins for CORS. Can be a list of origins or '*' for all origins.
+    Default is '*' for backward compatibility. In production, specify exact origins."""
+    cors_allow_credentials: bool = True
+    """Whether to allow credentials in CORS requests.
+    Default is True for backward compatibility. In v1.7, this will be changed to False when using wildcard origins."""
+    cors_allow_methods: list[str] | str = "*"
+    """Allowed HTTP methods for CORS requests."""
+    cors_allow_headers: list[str] | str = "*"
+    """Allowed headers for CORS requests."""
+
     # Telemetry
     do_not_track: bool = False
     """If set to True, Langflow will not track telemetry."""
@@ -290,6 +302,18 @@ class Settings(BaseSettings):
     update_starter_projects: bool = True
     """If set to True, Langflow will update starter projects."""
 
+    @field_validator("cors_origins", mode="before")
+    @classmethod
+    def validate_cors_origins(cls, value):
+        """Convert comma-separated string to list if needed."""
+        if isinstance(value, str) and value != "*":
+            if "," in value:
+                # Convert comma-separated string to list
+                return [origin.strip() for origin in value.split(",")]
+            # Convert single origin to list for consistency
+            return [value]
+        return value
+
     @field_validator("use_noop_database", mode="before")
     @classmethod
     def set_use_noop_database(cls, value):

src/backend/tests/unit/test_security_cors.py

@@ -0,0 +1,462 @@
+"""Unit tests for CORS security configuration."""
+
+import os
+import tempfile
+import warnings
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from langflow.services.settings.base import Settings
+
+
+class TestCORSConfiguration:
+    """Test CORS configuration and security validations."""
+
+    def test_default_cors_settings_current_behavior(self):
+        """Test current CORS settings behavior (warns about security implications)."""
+        with tempfile.TemporaryDirectory() as temp_dir, patch.dict(os.environ, {"LANGFLOW_CONFIG_DIR": temp_dir}):
+            settings = Settings()
+
+            # Current behavior: wildcard origins with credentials ENABLED (insecure)
+            assert settings.cors_origins == "*"
+            assert settings.cors_allow_credentials is True  # Currently defaults to True (insecure)
+            assert settings.cors_allow_methods == "*"
+            assert settings.cors_allow_headers == "*"
+
+            # Warn about CRITICAL security implications
+            warnings.warn(
+                "CRITICAL SECURITY WARNING: Current CORS configuration uses wildcard origins (*) "
+                "WITH CREDENTIALS ENABLED! This allows any website to make authenticated requests "
+                "to your Langflow instance and potentially steal user credentials. "
+                "This will be changed to more secure defaults in v1.7. "
+                "Please configure LANGFLOW_CORS_ORIGINS with specific domains for production use.",
+                UserWarning,
+                stacklevel=2,
+            )
+
+    @pytest.mark.skip(reason="Uncomment in v1.7 - represents future secure behavior")
+    def test_default_cors_settings_secure_future(self):
+        """Test future default CORS settings that will be secure (skip until v1.7)."""
+        # This test represents the behavior we want in v1.7
+        # with tempfile.TemporaryDirectory() as temp_dir, patch.dict(os.environ, {"LANGFLOW_CONFIG_DIR": temp_dir}):
+        #     settings = Settings()
+        #     # Future secure defaults:
+        #     assert settings.cors_origins == ["http://localhost:3000", "http://127.0.0.1:3000"]
+        #     assert settings.cors_allow_credentials is True
+        #     assert settings.cors_allow_methods == ["GET", "POST", "PUT", "DELETE"]
+        #     assert settings.cors_allow_headers == ["Content-Type", "Authorization"]
+
+    def test_cors_origins_string_to_list_conversion(self):
+        """Test comma-separated origins are converted to list."""
+        with (
+            tempfile.TemporaryDirectory() as temp_dir,
+            patch.dict(
+                os.environ,
+                {
+                    "LANGFLOW_CONFIG_DIR": temp_dir,
+                    "LANGFLOW_CORS_ORIGINS": "https://app1.example.com,https://app2.example.com",
+                },
+            ),
+        ):
+            settings = Settings()
+            assert settings.cors_origins == ["https://app1.example.com", "https://app2.example.com"]
+
+    def test_single_origin_converted_to_list(self):
+        """Test single origin is converted to list for consistency."""
+        with (
+            tempfile.TemporaryDirectory() as temp_dir,
+            patch.dict(
+                os.environ,
+                {
+                    "LANGFLOW_CONFIG_DIR": temp_dir,
+                    "LANGFLOW_CORS_ORIGINS": "https://app.example.com",
+                },
+            ),
+        ):
+            settings = Settings()
+            assert settings.cors_origins == ["https://app.example.com"]
+
+    def test_wildcard_with_credentials_allowed_current_behavior(self):
+        """Test that credentials are NOT disabled when using wildcard origins (current insecure behavior)."""
+        with (
+            tempfile.TemporaryDirectory() as temp_dir,
+            patch.dict(
+                os.environ,
+                {
+                    "LANGFLOW_CONFIG_DIR": temp_dir,
+                    "LANGFLOW_CORS_ORIGINS": "*",
+                    "LANGFLOW_CORS_ALLOW_CREDENTIALS": "true",
+                },
+            ),
+        ):
+            settings = Settings()
+            assert settings.cors_origins == "*"
+            # Current behavior: credentials are NOT prevented (INSECURE!)
+            assert settings.cors_allow_credentials is True
+
+            # Warn about the CRITICAL security implications
+            warnings.warn(
+                "CRITICAL SECURITY WARNING: Wildcard CORS origins (*) WITH CREDENTIALS ENABLED! "
+                "This is a severe security vulnerability that allows any website to make "
+                "authenticated requests and potentially steal user credentials. "
+                "This MUST be fixed in production! Configure specific origins immediately.",
+                UserWarning,
+                stacklevel=2,
+            )
+
+    def test_specific_origins_allow_credentials(self):
+        """Test that credentials work with specific origins."""
+        with (
+            tempfile.TemporaryDirectory() as temp_dir,
+            patch.dict(
+                os.environ,
+                {
+                    "LANGFLOW_CONFIG_DIR": temp_dir,
+                    "LANGFLOW_CORS_ORIGINS": "https://app.example.com",
+                    "LANGFLOW_CORS_ALLOW_CREDENTIALS": "true",
+                },
+            ),
+        ):
+            settings = Settings()
+            assert settings.cors_origins == ["https://app.example.com"]
+            assert settings.cors_allow_credentials is True
+
+    @patch("langflow.main.setup_sentry")  # Mock Sentry setup
+    @patch("langflow.main.get_settings_service")
+    def test_cors_middleware_configuration(self, mock_get_settings, mock_setup_sentry):
+        """Test that CORS middleware is configured correctly in the app."""
+        from langflow.main import create_app
+
+        # Mock settings
+        mock_settings = MagicMock()
+        mock_settings.settings.cors_origins = ["https://app.example.com"]
+        mock_settings.settings.cors_allow_credentials = True
+        mock_settings.settings.cors_allow_methods = ["GET", "POST"]
+        mock_settings.settings.cors_allow_headers = ["Content-Type"]
+        mock_settings.settings.prometheus_enabled = False
+        mock_settings.settings.mcp_server_enabled = False
+        mock_settings.settings.sentry_dsn = None  # Disable Sentry
+        mock_get_settings.return_value = mock_settings
+
+        # Create app
+        mock_setup_sentry.return_value = None  # Use the mock
+        app = create_app()
+
+        # Find CORS middleware
+        cors_middleware = None
+        for middleware in app.user_middleware:
+            if middleware.cls == CORSMiddleware:
+                cors_middleware = middleware
+                break
+
+        assert cors_middleware is not None
+        assert cors_middleware.kwargs["allow_origins"] == ["https://app.example.com"]
+        assert cors_middleware.kwargs["allow_credentials"] is True
+        assert cors_middleware.kwargs["allow_methods"] == ["GET", "POST"]
+        assert cors_middleware.kwargs["allow_headers"] == ["Content-Type"]
+
+    @patch("langflow.main.setup_sentry")  # Mock Sentry setup
+    @patch("langflow.main.get_settings_service")
+    @patch("langflow.main.logger")
+    def test_cors_wildcard_credentials_runtime_check_current_behavior(
+        self, mock_logger, mock_get_settings, mock_setup_sentry
+    ):
+        """Test runtime validation prevents wildcard with credentials (current behavior)."""
+        from langflow.main import create_app
+
+        # Mock settings with configuration that triggers current security measure
+        mock_settings = MagicMock()
+        mock_settings.settings.cors_origins = "*"
+        mock_settings.settings.cors_allow_credentials = True  # Gets disabled for security
+        mock_settings.settings.cors_allow_methods = "*"
+        mock_settings.settings.cors_allow_headers = "*"
+        mock_settings.settings.prometheus_enabled = False
+        mock_settings.settings.mcp_server_enabled = False
+        mock_settings.settings.sentry_dsn = None  # Disable Sentry
+        mock_get_settings.return_value = mock_settings
+
+        # Create app
+        mock_setup_sentry.return_value = None  # Use the mock
+        app = create_app()
+
+        # Check that warning was logged about deprecation/security
+        # The actual warning message is different from what we expected
+        warning_calls = [str(call) for call in mock_logger.warning.call_args_list]
+        # We expect warnings about the insecure configuration
+        assert any("DEPRECATION" in str(call) or "SECURITY" in str(call) for call in warning_calls), (
+            f"Expected security warning but got: {warning_calls}"
+        )
+
+        # Find CORS middleware and verify credentials are still allowed (current insecure behavior)
+        cors_middleware = None
+        for middleware in app.user_middleware:
+            if middleware.cls == CORSMiddleware:
+                cors_middleware = middleware
+                break
+
+        assert cors_middleware is not None
+        assert cors_middleware.kwargs["allow_origins"] == "*"
+        assert cors_middleware.kwargs["allow_credentials"] is True  # Current behavior: NOT disabled (insecure!)
+
+        # Warn about the security implications
+        warnings.warn(
+            "CRITICAL SECURITY WARNING: Current behavior allows wildcard origins WITH CREDENTIALS ENABLED! "
+            "This is a severe security vulnerability. Any website can make authenticated requests. "
+            "In v1.7, this will be changed to secure defaults with specific origins only.",
+            UserWarning,
+            stacklevel=2,
+        )
+
+
+class TestRefreshTokenSecurity:
+    """Test refresh token security improvements."""
+
+    @pytest.mark.asyncio
+    async def test_refresh_token_type_validation(self):
+        """Test that refresh token validates token type."""
+        from langflow.services.auth.utils import create_refresh_token
+
+        mock_db = MagicMock()
+
+        with patch("langflow.services.auth.utils.jwt.decode") as mock_decode:
+            # Test with wrong token type
+            mock_decode.return_value = {"sub": "user-123", "type": "access"}  # Wrong type
+
+            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"
+
+                with pytest.raises(HTTPException) as exc_info:
+                    await create_refresh_token("fake-token", mock_db)
+
+                assert exc_info.value.status_code == 401
+                assert "Invalid refresh token" in str(exc_info.value.detail)
+
+    @pytest.mark.asyncio
+    async def test_refresh_token_user_active_check(self):
+        """Test that inactive users cannot refresh tokens."""
+        from langflow.services.auth.utils import create_refresh_token
+
+        mock_db = MagicMock()
+        mock_user = MagicMock()
+        mock_user.is_active = False  # Inactive user
+
+        with patch("langflow.services.auth.utils.jwt.decode") as mock_decode:
+            mock_decode.return_value = {"sub": "user-123", "type": "refresh"}  # Correct type
+
+            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"
+
+                with patch("langflow.services.auth.utils.get_user_by_id") as mock_get_user:
+                    mock_get_user.return_value = mock_user
+
+                    with pytest.raises(HTTPException) as exc_info:
+                        await create_refresh_token("fake-token", mock_db)
+
+                    assert exc_info.value.status_code == 401
+                    assert "inactive" in str(exc_info.value.detail).lower()
+
+    @pytest.mark.asyncio
+    async def test_refresh_token_valid_flow(self):
+        """Test that valid refresh tokens work correctly."""
+        from langflow.services.auth.utils import create_refresh_token
+
+        mock_db = MagicMock()
+        mock_user = MagicMock()
+        mock_user.is_active = True  # Active user
+        mock_user.id = "user-123"
+
+        with patch("langflow.services.auth.utils.jwt.decode") as mock_decode:
+            mock_decode.return_value = {"sub": "user-123", "type": "refresh"}  # Correct type
+
+            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.ACCESS_TOKEN_EXPIRE_SECONDS = 3600
+                mock_settings.return_value.auth_settings.REFRESH_TOKEN_EXPIRE_SECONDS = 604800
+
+                with patch("langflow.services.auth.utils.get_user_by_id") as mock_get_user:
+                    mock_get_user.return_value = mock_user
+
+                    with patch("langflow.services.auth.utils.create_user_tokens") as mock_create_tokens:
+                        expected_access = "new-access-token"
+                        expected_refresh = "new-refresh-token"
+                        mock_create_tokens.return_value = {
+                            "access_token": expected_access,
+                            "refresh_token": expected_refresh,
+                        }
+
+                        result = await create_refresh_token("fake-token", mock_db)
+
+                        assert result["access_token"] == expected_access
+                        assert result["refresh_token"] == expected_refresh
+                        mock_create_tokens.assert_called_once_with("user-123", mock_db)
+
+    def test_refresh_token_samesite_setting_current_behavior(self):
+        """Test current refresh token SameSite settings (warns about security)."""
+        from langflow.services.settings.auth import AuthSettings
+
+        with tempfile.TemporaryDirectory() as temp_dir, patch.dict(os.environ, {"LANGFLOW_CONFIG_DIR": temp_dir}):
+            auth_settings = AuthSettings(CONFIG_DIR=temp_dir)
+            # Current behavior: refresh token uses 'none' (allows cross-site)
+            assert auth_settings.REFRESH_SAME_SITE == "none"  # Current: allows cross-site (less secure)
+            assert auth_settings.ACCESS_SAME_SITE == "lax"  # Access token is already lax (good)
+
+            # Warn about security implications
+            warnings.warn(
+                "SECURITY WARNING: Refresh tokens currently use SameSite=none which allows "
+                "cross-site requests. This should be changed to 'lax' or 'strict' in production. "
+                "In v1.7, this will default to 'lax' for better security.",
+                UserWarning,
+                stacklevel=2,
+            )
+
+    @pytest.mark.skip(reason="Uncomment in v1.7 - represents future secure SameSite behavior")
+    def test_refresh_token_samesite_setting_future_secure(self):
+        """Test future secure refresh token SameSite settings (skip until v1.7)."""
+        # Future secure behavior (uncomment in v1.7):
+        # from langflow.services.settings.auth import AuthSettings
+        # with tempfile.TemporaryDirectory() as temp_dir, patch.dict(os.environ, {"LANGFLOW_CONFIG_DIR": temp_dir}):
+        #     auth_settings = AuthSettings(CONFIG_DIR=temp_dir)
+        #     assert auth_settings.REFRESH_SAME_SITE == "lax"  # Secure default
+        #     assert auth_settings.ACCESS_SAME_SITE == "lax"
+
+
+class TestCORSIntegration:
+    """Integration tests for CORS with actual HTTP requests."""
+
+    @pytest.mark.asyncio
+    @patch("langflow.main.setup_sentry")  # Mock Sentry setup
+    async def test_cors_headers_in_response_current_behavior(self, mock_setup_sentry):
+        """Test that CORS headers are properly set in responses (current behavior)."""
+        from fastapi.testclient import TestClient
+
+        with patch("langflow.main.get_settings_service") as mock_get_settings:
+            mock_settings = MagicMock()
+            mock_settings.settings.cors_origins = ["https://app.example.com"]
+            mock_settings.settings.cors_allow_credentials = True
+            mock_settings.settings.cors_allow_methods = "*"
+            mock_settings.settings.cors_allow_headers = "*"
+            mock_settings.settings.prometheus_enabled = False
+            mock_settings.settings.mcp_server_enabled = False
+            mock_settings.settings.sentry_dsn = None  # Disable Sentry
+            mock_get_settings.return_value = mock_settings
+
+            from langflow.main import create_app
+
+            mock_setup_sentry.return_value = None  # Use the mock
+            app = create_app()
+            client = TestClient(app)
+
+            # Make OPTIONS request (CORS preflight)
+            response = client.options(
+                "/api/v1/version",
+                headers={
+                    "Origin": "https://app.example.com",
+                    "Access-Control-Request-Method": "GET",
+                },
+            )
+
+            assert response.status_code == 200
+            assert response.headers.get("access-control-allow-origin") == "https://app.example.com"
+            assert response.headers.get("access-control-allow-credentials") == "true"
+
+            # Warn that this is testing current behavior
+            warnings.warn(
+                "This test validates current CORS behavior. In v1.7, default origins will be more restrictive.",
+                UserWarning,
+                stacklevel=2,
+            )
+
+    @pytest.mark.skip(reason="Uncomment in v1.7 - represents future secure CORS blocking behavior")
+    async def test_cors_blocks_unauthorized_origin_future_secure(self):
+        """Test that future secure CORS configuration blocks unauthorized origins (skip until v1.7)."""
+        # This test represents the behavior we want in v1.7 with secure defaults
+
+    @pytest.mark.asyncio
+    @patch("langflow.main.setup_sentry")  # Mock Sentry setup
+    async def test_cors_blocks_unauthorized_origin_current_behavior(self, mock_setup_sentry):
+        """Test that CORS blocks requests from unauthorized origins."""
+        from fastapi.testclient import TestClient
+
+        with patch("langflow.main.get_settings_service") as mock_get_settings:
+            mock_settings = MagicMock()
+            mock_settings.settings.cors_origins = ["https://app.example.com"]
+            mock_settings.settings.cors_allow_credentials = True
+            mock_settings.settings.cors_allow_methods = "*"
+            mock_settings.settings.cors_allow_headers = "*"
+            mock_settings.settings.prometheus_enabled = False
+            mock_settings.settings.mcp_server_enabled = False
+            mock_settings.settings.sentry_dsn = None  # Disable Sentry
+            mock_get_settings.return_value = mock_settings
+
+            from langflow.main import create_app
+
+            mock_setup_sentry.return_value = None  # Use the mock
+            app = create_app()
+            client = TestClient(app)
+
+            # Make OPTIONS request from unauthorized origin
+            response = client.options(
+                "/api/v1/version",
+                headers={
+                    "Origin": "https://evil.com",
+                    "Access-Control-Request-Method": "GET",
+                },
+            )
+
+            assert response.status_code == 400  # CORS will block this
+
+            # Warn about current behavior implications
+            warnings.warn(
+                "This test shows current CORS behavior with specific origins. "
+                "Note that current default behavior uses wildcard origins (*) which would NOT block this. "
+                "In v1.7, secure defaults will be implemented to prevent unauthorized origins.",
+                UserWarning,
+                stacklevel=2,
+            )
+
+
+class TestFutureSecureCORSBehavior:
+    """Tests for future secure CORS behavior in v1.7 - currently skipped."""
+
+    @pytest.mark.skip(reason="Uncomment in v1.7 - represents future secure default CORS configuration")
+    def test_future_secure_defaults(self):
+        """Test that v1.7 will have secure CORS defaults."""
+        # Future secure behavior (uncomment in v1.7):
+        # with tempfile.TemporaryDirectory() as temp_dir, patch.dict(os.environ, {"LANGFLOW_CONFIG_DIR": temp_dir}):
+        #     settings = Settings()
+        #     # v1.7 secure defaults:
+        #     assert settings.cors_origins == ["http://localhost:3000", "http://127.0.0.1:3000", "http://localhost:7860"]
+        #     assert settings.cors_allow_credentials is True  # Safe with specific origins
+        #     assert settings.cors_allow_methods == ["GET", "POST", "PUT", "DELETE", "OPTIONS"]
+        #     assert settings.cors_allow_headers == ["Content-Type", "Authorization", "X-Requested-With"]
+
+    @pytest.mark.skip(reason="Uncomment in v1.7 - represents future secure wildcard rejection")
+    def test_future_wildcard_rejection(self):
+        """Test that v1.7 will warn about or reject wildcard origins in production."""
+        # Future behavior (uncomment in v1.7):
+        # with (
+        #     tempfile.TemporaryDirectory() as temp_dir,
+        #     patch.dict(
+        #         os.environ,
+        #         {
+        #             "LANGFLOW_CONFIG_DIR": temp_dir,
+        #             "LANGFLOW_CORS_ORIGINS": "*",
+        #         },
+        #     ),
+        # ):
+        #     # Should either warn strongly or reject wildcard in production mode
+        #     with pytest.warns(UserWarning, match="SECURITY WARNING.*wildcard.*production"):
+        #         settings = Settings()
+        #         # Or potentially: pytest.raises(ValueError, match="Wildcard origins not allowed in production")
+
+    @pytest.mark.skip(reason="Uncomment in v1.7 - represents future secure middleware configuration")
+    async def test_future_secure_middleware_config(self):
+        """Test that v1.7 middleware will use secure defaults."""
+        # Future secure middleware behavior (uncomment in v1.7):
+        # Test that the app creates middleware with secure defaults
+        # and properly validates origins in production mode