dvwn/nl2sql-api

1# Path: frontend/components/auth.py2# Authentication component for handling user login and registration.3import hashlib4import json5import os6import streamlit as st7import pandas as pd8from typing import Tuple9from pathlib import Path10from dotenv import load_dotenv11 12load_dotenv()13 14AUTH_DB_PATH = Path("backend/src/database/auth_db.json")15 16# Hashing password17def hash_password(password):18    """ Securely hashes a password """19    return hashlib.sha256(password.encode()).hexdigest()20 21# Load user JSON-databases22def load_userDB():23    """ Load authenticated credentials from local JSON file """24    if not AUTH_DB_PATH.exists():25        AUTH_DB_PATH.parent.mkdir(parents=True, exist_ok=True)26        admin_user = os.getenv("DEFAULT_ADMIN_PASS")27        if not admin_user:28            raise ValueError("SECURITY HALT: DEFAULT_ADMIN_PASS environment is missing. Cannot initialize database securely.")29        30        default_db = {31            "admin": {32                "password": hash_password(admin_user),33                "history": []34            }35        }36 37        with open(AUTH_DB_PATH, "w") as f:38            json.dump(default_db, f, indent=4)39        return default_db40    41    with open(AUTH_DB_PATH, "r") as f:42        return json.load(f)43 44# Sace user data into the JSON-databases    45def save_userDB(db_data: dict):46    with open(AUTH_DB_PATH, "w") as f:47        json.dump(db_data, f, indent=4)48 49# User registration/create new account50def register_user(username: str, password: str) -> Tuple[bool, str]:51    """ Registers a new user record safely into the system """52    if not username or not password:53        return False, "Fields cannot be empty."54    55    db = load_userDB()56    if username in db:57        return False, "Username already exists. Please choose a different one."58    59    db[username] = {60        "password": hash_password(password),61        "history": []62    }63    save_userDB(db)64    return True, "Account registered successfully!"65 66# User login67def verify_login(username: str, password: str) -> bool:68    """ Verifies credentials against the authentication store """69    db = load_userDB()70    if username not in db:71        return False72    return db[username]["password"] == hash_password(password)73 74# Save user's conversation75def save_chat_history(username: str, session_messages: list):76    """ Saves user's conversation logs to history """77    if not username or username == "guest" or not session_messages:78        return79    80    db = load_userDB()81    if username in db:82        if "sessions" not in db[username]:83            db[username]["sessions"] = []84 85        # Sanitize messages to make df JSON-serializable86        sanitized_msg = []87        for msg in session_messages:88            clean_msg = {}89            for key, value in msg.items():90                if isinstance(value, pd.DataFrame):91                    clean_msg[key] = value.to_dict(orient="records")92                else:93                    clean_msg[key] = value94            sanitized_msg.append(clean_msg)95 96        first_prompt = next((m["content"] for m in session_messages if m["role"] == "user"), "New Query")97        truncated_title = first_prompt[:25]98 99        # Check whether update an existing session / append new history100        existing_sessions = db[username]["sessions"]101        session_exists = False102 103        for ses in existing_sessions:104            if ses.get("title") == truncated_title:105                ses["messages"] = sanitized_msg106                session_exists = True107                break108        109        if not session_exists:110            existing_sessions.append({111                "title": truncated_title,112                "messages": sanitized_msg113            })114 115        db[username]["sessions"] = existing_sessions116        save_userDB(db)117        st.session_state.chat_sessions = existing_sessions118 119# Retrieves user's history120def get_history(username: str) -> list:121    """ Retrieves saved conversation history for verified users """122    db = load_userDB()123    if username in db:124        return db[username].get("sessions", [])125    return []126 127@st.dialog("Account Access Panel")128def render_auth_dialog():129    """130    Streamlit UI native dialog: Handle custom login & signup cleanly without template injection131    """132    st.subheader("Sign In or Register")133    tab1, tab2 = st.tabs([":material/lock_person: Login", ":material/person_book: Create Account"])134 135    with tab1:136        login_user = st.text_input("Username", key="login_user_input")137        login_pass = st.text_input("Password", type="password", key="login_pass_input")138        if st.button("Authenticate", type="primary", use_container_width=True):139            if verify_login(login_user, login_pass):140                st.session_state.auth_stat = "logged_in"141                st.session_state.username = login_user142                st.session_state.messages = get_history(login_user)143                st.success(f"Welcome back, {login_user}!")144                st.rerun()145            else:146                st.error("Invalid username or password credentials.")147 148    with tab2:149        reg_user = st.text_input("Username", key="reg_user_input")150        reg_pass = st.text_input("Password", type="password", key="reg_user_pass")151        reg_pass_confirm = st.text_input("Re-enter Password", type="password", key="reg_user_pass_confirm") 152        if st.button("Sign Up Now", use_container_width=True):153            if reg_pass != reg_pass_confirm:154                st.error("Password does not match!")155            else:156                success, msg = register_user(reg_user, reg_pass)157                if success:158                    st.success(msg)159                    # Login user automatically160                    st.session_state.auth_stat = "logged_in"161                    st.session_state.username = reg_user162                    st.session_state.messages = []163                    st.rerun()164                else:165                    st.error(msg)