diff --git a/src/backend/langflow/services/auth/__init__.py b/src/backend/langflow/services/auth/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/backend/langflow/services/auth/factory.py b/src/backend/langflow/services/auth/factory.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/backend/langflow/services/auth/service.py b/src/backend/langflow/services/auth/service.py new file mode 100644 index 000000000..5e25aa207 --- /dev/null +++ b/src/backend/langflow/services/auth/service.py @@ -0,0 +1,228 @@ +from uuid import UUID +from typing import Annotated +from jose import JWTError, jwt +from langflow.services.base import Service +from langflow.services.database.models.user import ( + User, + get_user_by_id, + get_user_by_username, +) +from sqlalchemy.orm import Session +from passlib.context import CryptContext +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, get_session + +from langflow.services.database.models.user import ( + update_user_last_login_at, +) + + +pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") +oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login") + + +class AuthManager(Service): + name = "auth_manager" + + def __init__(self): + self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") + self.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", + headers={"WWW-Authenticate": "Bearer"}, + ) + + try: + payload = jwt.decode( + token, + settings_manager.auth_settings.SECRET_KEY, + algorithms=[settings_manager.auth_settings.ALGORITHM], + ) + user_id: UUID = payload.get("sub") # type: ignore + token_type: str = payload.get("type") # type: ignore + + if user_id is None or token_type: + raise credentials_exception + except JWTError as e: + raise credentials_exception from e + + user = get_user_by_id(db, user_id) # type: ignore + if user is None: + raise credentials_exception + return user + + +async def get_current_active_user( + current_user: Annotated[User, Depends(get_current_user)] +): + if not current_user.is_active: + raise HTTPException(status_code=400, detail="Inactive user") + return current_user + + +def verify_password(plain_password, hashed_password): + return pwd_context.verify(plain_password, hashed_password) + + +def get_password_hash(password): + return pwd_context.hash(password) + + +def create_token(data: dict, expires_delta: timedelta): + 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, + settings_manager.auth_settings.SECRET_KEY, + algorithm=settings_manager.auth_settings.ALGORITHM, + ) + + +def create_super_user(db: Session = Depends(get_session)) -> User: + settings_manager = get_settings_manager() + + super_user = get_user_by_username( + db, settings_manager.auth_settings.FIRST_SUPERUSER + ) + + if not super_user: + super_user = User( + username=settings_manager.auth_settings.FIRST_SUPERUSER, + password=get_password_hash( + settings_manager.auth_settings.FIRST_SUPERUSER_PASSWORD + ), + is_superuser=True, + is_active=True, + last_login_at=None, + ) + + db.add(super_user) + db.commit() + db.refresh(super_user) + + return super_user + + +def create_user_longterm_token(db: Session = Depends(get_session)) -> dict: + super_user = create_super_user(db) + + access_token_expires_longterm = timedelta(days=365) + access_token = create_token( + data={"sub": str(super_user.id)}, + expires_delta=access_token_expires_longterm, + ) + + # Update: last_login_at + update_user_last_login_at(super_user.id, db) + + return { + "access_token": access_token, + "refresh_token": None, + "token_type": "bearer", + } + + +def create_user_api_key(user_id: UUID) -> dict: + access_token = create_token( + data={"sub": str(user_id), "role": "api_key"}, + expires_delta=timedelta(days=365 * 2), + ) + + return {"api_key": access_token} + + +def get_user_id_from_token(token: str) -> UUID: + try: + user_id = jwt.get_unverified_claims(token)["sub"] + return UUID(user_id) + except (KeyError, JWTError, ValueError): + return UUID(int=0) + + +def create_user_tokens( + user_id: UUID, db: Session = Depends(get_session), update_last_login: bool = False +) -> dict: + settings_manager = get_settings_manager() + + access_token_expires = timedelta( + minutes=settings_manager.auth_settings.ACCESS_TOKEN_EXPIRE_MINUTES + ) + access_token = create_token( + data={"sub": str(user_id)}, + expires_delta=access_token_expires, + ) + + refresh_token_expires = timedelta( + minutes=settings_manager.auth_settings.REFRESH_TOKEN_EXPIRE_MINUTES + ) + refresh_token = create_token( + data={"sub": str(user_id), "type": "rf"}, + expires_delta=refresh_token_expires, + ) + + # Update: last_login_at + if update_last_login: + update_user_last_login_at(user_id, db) + + return { + "access_token": access_token, + "refresh_token": refresh_token, + "token_type": "bearer", + } + + +def create_refresh_token(refresh_token: str, db: Session = Depends(get_session)): + settings_manager = get_settings_manager() + + try: + payload = jwt.decode( + refresh_token, + settings_manager.auth_settings.SECRET_KEY, + algorithms=[settings_manager.auth_settings.ALGORITHM], + ) + user_id: UUID = payload.get("sub") # type: ignore + token_type: str = payload.get("type") # type: ignore + + if user_id is None or token_type is None: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token" + ) + + return create_user_tokens(user_id, db) + + except JWTError as e: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid refresh token", + ) from e + + +def authenticate_user( + username: str, password: str, db: Session = Depends(get_session) +) -> User | None: + user = get_user_by_username(db, username) + + if not user: + return None + + if not user.is_active: + if not user.last_login_at: + raise HTTPException(status_code=400, detail="Waiting for approval") + raise HTTPException(status_code=400, detail="Inactive user") + + return user if verify_password(password, user.password) else None diff --git a/src/backend/langflow/services/settings/auth.py b/src/backend/langflow/services/settings/auth.py new file mode 100644 index 000000000..d144b4347 --- /dev/null +++ b/src/backend/langflow/services/settings/auth.py @@ -0,0 +1,29 @@ +from typing import Optional +import secrets + +from pydantic import BaseSettings + + +class AuthSettings(BaseSettings): + # Login settings + SECRET_KEY: str = secrets.token_hex(32) + ALGORITHM: str = "HS256" + ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 + REFRESH_TOKEN_EXPIRE_MINUTES: int = 70 + + # API Key to execute /process endpoint + API_KEY_SECRET_KEY: Optional[ + str + ] = "b82818e0ad4ff76615c5721ee21004b07d84cd9b87ba4d9cb42374da134b841a" + API_KEY_ALGORITHM: str = "HS256" + + # If AUTO_LOGIN = True + # > The application does not request login and logs in automatically as a super user. + AUTO_LOGIN: bool = True + FIRST_SUPERUSER: str = "langflow" + FIRST_SUPERUSER_PASSWORD: str = "langflow" + + class Config: + validate_assignment = True + extra = "ignore" + env_prefix = "LANGFLOW_" diff --git a/src/backend/langflow/services/settings/base.py b/src/backend/langflow/services/settings/base.py index 65a1a2909..4e1e2d35b 100644 --- a/src/backend/langflow/services/settings/base.py +++ b/src/backend/langflow/services/settings/base.py @@ -2,7 +2,6 @@ import contextlib import json import os from shutil import copy2 -import secrets from typing import Optional, List from pathlib import Path @@ -41,24 +40,6 @@ 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 - - # API Key to execute /process endpoint - API_KEY_SECRET_KEY: Optional[ - str - ] = "b82818e0ad4ff76615c5721ee21004b07d84cd9b87ba4d9cb42374da134b841a" - API_KEY_ALGORITHM: str = "HS256" - - # If AUTO_LOGIN = True - # > The application does not request login and logs in automatically as a super user. - AUTO_LOGIN: bool = True - FIRST_SUPERUSER: str = "langflow" - FIRST_SUPERUSER_PASSWORD: str = "langflow" - @validator("CONFIG_DIR", pre=True, allow_reuse=True) def set_langflow_dir(cls, value): if not value: diff --git a/src/backend/langflow/services/settings/manager.py b/src/backend/langflow/services/settings/manager.py index a357c4804..1a6c0feeb 100644 --- a/src/backend/langflow/services/settings/manager.py +++ b/src/backend/langflow/services/settings/manager.py @@ -1,4 +1,5 @@ from langflow.services.base import Service +from langflow.services.settings.auth import AuthSettings from langflow.services.settings.base import Settings from langflow.utils.logger import logger import os @@ -8,9 +9,10 @@ import yaml class SettingsManager(Service): name = "settings_manager" - def __init__(self, settings: Settings): + def __init__(self, settings: Settings, auth_settings: AuthSettings): super().__init__() self.settings = settings + self.auth_settings = auth_settings @classmethod def load_settings_from_yaml(cls, file_path: str) -> "SettingsManager": @@ -33,4 +35,5 @@ class SettingsManager(Service): ) settings = Settings(**settings_dict) - return cls(settings) + auth_settings = AuthSettings() + return cls(settings, auth_settings)