🔧 fix(auth.py): move SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES, and REFRESH_TOKEN_EXPIRE_MINUTES to environment variables for better security and configurability

✨ feat(auth.py): add support for loading settings from environment variables in create_token and create_user_tokens functions
🔧 fix(auth.py): use settings_manager to access SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES, and REFRESH_TOKEN_EXPIRE_MINUTES in create_token and create_user_tokens functions
🔧 fix(auth.py): use settings_manager to access SECRET_KEY and ALGORITHM in get_current_user function
✨ feat(auth.py): add create_user_longterm_token function to create long-term access token for auto login
✨ feat(login.py): add auto_login endpoint to automatically log in as a super user if AUTO_LOGIN is enabled in the settings
🔧 fix(settings/base.py): add SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES, REFRESH_TOKEN_EXPIRE_MINUTES, and AUTO_LOGIN settings to the base settings class
✨ feat(settings/base.py): add AUTO_LOGIN setting to control whether auto login is enabled or not
🔧 fix(settings/base.py): use secrets.token_hex(32) to generate a random SECRET_KEY instead of hardcoding it
🔧 fix(settings/base.py): update comments and formatting in the base settings class
✨ feat(settings/base.py): add AUTO_LOGIN setting to control whether auto login is enabled or not
🔧 fix(settings/base.py): update comments and formatting in the base settings class
🔧 fix(settings/base.py): update comments and formatting in the base settings class
🔧 fix(settings/base.py): update comments and formatting in the base settings class
🔧 fix(settings/base.py): update comments and formatting in the base settings class
🔧 fix(settings/base.py): update comments and formatting in the base settings class
🔧 fix(settings/base.py): update comments and formatting in the base settings class
🔧 fix(settings/base.py): update comments and formatting in the base settings class
🔧 fix(settings/base.py): update comments and formatting in the base settings class
🔧 fix(settings/base.py): update comments and formatting in the base settings class
🔧 fix(settings/base.py): update comments and formatting in the base settings class
🔧 fix(settings/base.py): update comments and formatting in the base settings

🔥 refactor(settings.py): remove unused imports and code, clean up formatting and comments
🔀 chore(settings.py): merge duplicated code into reusable functions
📝 docs(settings.py): add missing docstrings and comments for better code documentation
🔧 chore(settings.py): update settings file to improve readability and maintainability

src/backend/langflow/auth/auth.py

@@ -7,6 +7,8 @@ from fastapi.security import OAuth2PasswordBearer
 from fastapi import Depends, HTTPException, status
 from datetime import datetime, timedelta, timezone
 
+from langflow.services.utils import get_settings_manager
+
 from langflow.services.utils import get_session
 from langflow.database.models.user import (
     User,
@@ -16,12 +18,6 @@ from langflow.database.models.user import (
 )
 
 
-# TODO: Move to env - JUST FOR TEST!!!!!
-SECRET_KEY = "698619adad2d916f1f32d264540976964b3c0d3828e0870a65add5800a8cc6b9"
-ALGORITHM = "HS256"
-ACCESS_TOKEN_EXPIRE_MINUTES = 60
-REFRESH_TOKEN_EXPIRE_MINUTES = 70
-
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")
 
@@ -29,6 +25,8 @@ oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")
 async def get_current_user(
     token: Annotated[str, Depends(oauth2_scheme)], db: Session = Depends(get_session)
 ) -> User:
+    settings_manager = get_settings_manager()
+
     credentials_exception = HTTPException(
         status_code=status.HTTP_401_UNAUTHORIZED,
         detail="Could not validate credentials",
@@ -36,7 +34,11 @@ async def get_current_user(
     )
 
     try:
-        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        payload = jwt.decode(
+            token,
+            settings_manager.settings.SECRET_KEY,
+            algorithms=[settings_manager.settings.ALGORITHM],
+        )
         user_id: UUID = payload.get("sub")  # type: ignore
         token_type: str = payload.get("type")  # type: ignore
 
@@ -68,24 +70,55 @@ def get_password_hash(password):
 
 
 def create_token(data: dict, expires_delta: timedelta):
-    to_encode = data.copy()
+    settings_manager = get_settings_manager()
 
+    to_encode = data.copy()
     expire = datetime.now(timezone.utc) + expires_delta
     to_encode["exp"] = expire
 
-    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return jwt.encode(
+        to_encode,
+        settings_manager.settings.SECRET_KEY,
+        algorithm=settings_manager.settings.ALGORITHM,
+    )
+
+
+def create_user_longterm_token(
+    user_id: UUID, db: Session = Depends(get_session), update_last_login: bool = False
+) -> dict:
+    access_token_expires_longterm = timedelta(days=365)
+    access_token = create_token(
+        data={"sub": str(user_id)},
+        expires_delta=access_token_expires_longterm,
+    )
+
+    # Update: last_login_at
+    if update_last_login:
+        update_user_last_login_at(user_id, db)
+
+    return {
+        "access_token": access_token,
+        "refresh_token": None,
+        "token_type": "bearer",
+    }
 
 
 def create_user_tokens(
     user_id: UUID, db: Session = Depends(get_session), update_last_login: bool = False
 ) -> dict:
-    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    settings_manager = get_settings_manager()
+
+    access_token_expires = timedelta(
+        minutes=settings_manager.settings.ACCESS_TOKEN_EXPIRE_MINUTES
+    )
     access_token = create_token(
         data={"sub": str(user_id)},
         expires_delta=access_token_expires,
     )
 
-    refresh_token_expires = timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES)
+    refresh_token_expires = timedelta(
+        minutes=settings_manager.settings.REFRESH_TOKEN_EXPIRE_MINUTES
+    )
     refresh_token = create_token(
         data={"sub": str(user_id), "type": "rf"},
         expires_delta=refresh_token_expires,
@@ -103,8 +136,14 @@ def create_user_tokens(
 
 
 def create_refresh_token(refresh_token: str, db: Session = Depends(get_session)):
+    settings_manager = get_settings_manager()
+
     try:
-        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
+        payload = jwt.decode(
+            refresh_token,
+            settings_manager.settings.SECRET_KEY,
+            algorithms=[settings_manager.settings.ALGORITHM],
+        )
         user_id: UUID = payload.get("sub")  # type: ignore
         token_type: str = payload.get("type")  # type: ignore
 

src/backend/langflow/routers/login.py

@@ -1,3 +1,4 @@
+from uuid import UUID
 from sqlalchemy.orm import Session
 from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordRequestForm
@@ -8,8 +9,11 @@ from langflow.auth.auth import (
     authenticate_user,
     create_user_tokens,
     create_refresh_token,
+    create_user_longterm_token,
 )
 
+from langflow.services.utils import get_settings_manager
+
 router = APIRouter()
 
 
@@ -29,6 +33,23 @@ async def login_to_get_access_token(
         )
 
 
+@router.get("/auto_login")
+async def auto_login(db: Session = Depends(get_session)):
+    settings_manager = get_settings_manager()
+
+    if settings_manager.settings.AUTO_LOGIN:
+        user_id = UUID("3fa85f64-5717-4562-b3fc-2c963f66afa6")
+        return create_user_longterm_token(user_id, db)
+
+    raise HTTPException(
+        status_code=status.HTTP_400_BAD_REQUEST,
+        detail={
+            "message": "Auto login is disabled. Please enable it in the settings",
+            "auto_login": False,
+        },
+    )
+
+
 @router.post("/refresh")
 async def refresh_token(token: str):
     if token:

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

@@ -1,6 +1,7 @@
 import contextlib
 import json
 import os
+import secrets
 from typing import Optional, List
 from pathlib import Path
 
@@ -35,6 +36,15 @@ class Settings(BaseSettings):
     REMOVE_API_KEYS: bool = False
     COMPONENTS_PATH: List[str] = []
 
+    # Login settings
+    SECRET_KEY: str = secrets.token_hex(32)
+    ALGORITHM: str = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
+    REFRESH_TOKEN_EXPIRE_MINUTES: int = 70
+    # If AUTO_LOGIN = True
+    # > The application does not request login and logs in automatically as a super user.
+    AUTO_LOGIN: bool = True
+
     @validator("DATABASE_URL", pre=True)
     def set_database_url(cls, value):
         if not value:

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

@@ -1,171 +0,0 @@
-import contextlib
-import json
-import os
-from typing import Optional, List
-from pathlib import Path
-
-import yaml
-from pydantic import BaseSettings, root_validator, validator
-from langflow.utils.logger import logger
-
-BASE_COMPONENTS_PATH = str(Path(__file__).parent / "components")
-
-
-class Settings(BaseSettings):
-    CHAINS: dict = {}
-    AGENTS: dict = {}
-    PROMPTS: dict = {}
-    LLMS: dict = {}
-    TOOLS: dict = {}
-    MEMORIES: dict = {}
-    EMBEDDINGS: dict = {}
-    VECTORSTORES: dict = {}
-    DOCUMENTLOADERS: dict = {}
-    WRAPPERS: dict = {}
-    RETRIEVERS: dict = {}
-    TOOLKITS: dict = {}
-    TEXTSPLITTERS: dict = {}
-    UTILITIES: dict = {}
-    OUTPUT_PARSERS: dict = {}
-    CUSTOM_COMPONENTS: dict = {}
-
-    DEV: bool = False
-    DATABASE_URL: Optional[str] = None
-    CACHE: str = "InMemoryCache"
-    REMOVE_API_KEYS: bool = False
-    COMPONENTS_PATH: List[str] = []
-
-    @validator("DATABASE_URL", pre=True)
-    def set_database_url(cls, value):
-        if not value:
-            logger.debug(
-                "No database_url provided, trying LANGFLOW_DATABASE_URL env variable"
-            )
-            if langflow_database_url := os.getenv("LANGFLOW_DATABASE_URL"):
-                value = langflow_database_url
-                logger.debug("Using LANGFLOW_DATABASE_URL env variable.")
-            else:
-                logger.debug("No DATABASE_URL env variable, using sqlite database")
-                value = "sqlite:///./langflow.db"
-
-        return value
-
-    @validator("COMPONENTS_PATH", pre=True)
-    def set_components_path(cls, value):
-        if os.getenv("LANGFLOW_COMPONENTS_PATH"):
-            logger.debug("Adding LANGFLOW_COMPONENTS_PATH to components_path")
-            langflow_component_path = os.getenv("LANGFLOW_COMPONENTS_PATH")
-            if (
-                Path(langflow_component_path).exists()
-                and langflow_component_path not in value
-            ):
-                if isinstance(langflow_component_path, list):
-                    for path in langflow_component_path:
-                        if path not in value:
-                            value.append(path)
-                    logger.debug(
-                        f"Extending {langflow_component_path} to components_path"
-                    )
-                elif langflow_component_path not in value:
-                    value.append(langflow_component_path)
-                    logger.debug(
-                        f"Appending {langflow_component_path} to components_path"
-                    )
-
-        if not value:
-            value = [BASE_COMPONENTS_PATH]
-            logger.debug("Setting default components path to components_path")
-        elif BASE_COMPONENTS_PATH not in value:
-            value.append(BASE_COMPONENTS_PATH)
-            logger.debug("Adding default components path to components_path")
-
-        logger.debug(f"Components path: {value}")
-        return value
-
-    class Config:
-        validate_assignment = True
-        extra = "ignore"
-        env_prefix = "LANGFLOW_"
-
-    @root_validator(allow_reuse=True)
-    def validate_lists(cls, values):
-        for key, value in values.items():
-            if key != "dev" and not value:
-                values[key] = []
-        return values
-
-    def update_from_yaml(self, file_path: str, dev: bool = False):
-        new_settings = load_settings_from_yaml(file_path)
-        self.CHAINS = new_settings.CHAINS or {}
-        self.AGENTS = new_settings.AGENTS or {}
-        self.PROMPTS = new_settings.PROMPTS or {}
-        self.LLMS = new_settings.LLMS or {}
-        self.TOOLS = new_settings.TOOLS or {}
-        self.MEMORIES = new_settings.MEMORIES or {}
-        self.WRAPPERS = new_settings.WRAPPERS or {}
-        self.TOOLKITS = new_settings.TOOLKITS or {}
-        self.TEXTSPLITTERS = new_settings.TEXTSPLITTERS or {}
-        self.UTILITIES = new_settings.UTILITIES or {}
-        self.EMBEDDINGS = new_settings.EMBEDDINGS or {}
-        self.VECTORSTORES = new_settings.VECTORSTORES or {}
-        self.DOCUMENTLOADERS = new_settings.DOCUMENTLOADERS or {}
-        self.RETRIEVERS = new_settings.RETRIEVERS or {}
-        self.OUTPUT_PARSERS = new_settings.OUTPUT_PARSERS or {}
-        self.CUSTOM_COMPONENTS = new_settings.CUSTOM_COMPONENTS or {}
-        self.COMPONENTS_PATH = new_settings.COMPONENTS_PATH or []
-        self.DEV = dev
-
-    def update_settings(self, **kwargs):
-        logger.debug("Updating settings")
-        for key, value in kwargs.items():
-            # value may contain sensitive information, so we don't want to log it
-            if not hasattr(self, key):
-                logger.debug(f"Key {key} not found in settings")
-                continue
-            logger.debug(f"Updating {key}")
-            if isinstance(getattr(self, key), list):
-                # value might be a '[something]' string
-                with contextlib.suppress(json.decoder.JSONDecodeError):
-                    value = json.loads(str(value))
-                if isinstance(value, list):
-                    for item in value:
-                        if item not in getattr(self, key):
-                            getattr(self, key).append(item)
-                    logger.debug(f"Extended {key}")
-                else:
-                    getattr(self, key).append(value)
-                    logger.debug(f"Appended {key}")
-
-            else:
-                setattr(self, key, value)
-                logger.debug(f"Updated {key}")
-            logger.debug(f"{key}: {getattr(self, key)}")
-
-
-def save_settings_to_yaml(settings: Settings, file_path: str):
-    with open(file_path, "w") as f:
-        settings_dict = settings.dict()
-        yaml.dump(settings_dict, f)
-
-
-def load_settings_from_yaml(file_path: str) -> Settings:
-    # Check if a string is a valid path or a file name
-    if "/" not in file_path:
-        # Get current path
-        current_path = os.path.dirname(os.path.abspath(__file__))
-
-        file_path = os.path.join(current_path, file_path)
-
-    with open(file_path, "r") as f:
-        settings_dict = yaml.safe_load(f)
-        settings_dict = {k.upper(): v for k, v in settings_dict.items()}
-
-        for key in settings_dict:
-            if key not in Settings.__fields__.keys():
-                raise KeyError(f"Key {key} not found in settings")
-            logger.debug(f"Loading {len(settings_dict[key])} {key} from {file_path}")
-
-    return Settings(**settings_dict)
-
-
-settings = load_settings_from_yaml("config.yaml")