""" Authentication Web Interface for OpenManus Mobile number + password based authentication forms """ import asyncio import sqlite3 from typing import Optional, Tuple import gradio as gr from app.auth import UserSignupRequest, UserLoginRequest from app.auth_service import AuthService from app.logger import logger class AuthInterface: """Authentication interface with Gradio""" def __init__(self, db_path: str = "openmanus.db"): self.db_path = db_path self.auth_service = None self.current_session = None self.init_database() def init_database(self): """Initialize database with schema""" try: conn = sqlite3.connect(self.db_path) # Create users table with mobile auth conn.execute( """ CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, mobile_number TEXT UNIQUE NOT NULL, full_name TEXT NOT NULL, password_hash TEXT NOT NULL, avatar_url TEXT, preferences TEXT, is_active BOOLEAN DEFAULT TRUE, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ) """ ) # Create sessions table conn.execute( """ CREATE TABLE IF NOT EXISTS sessions ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, title TEXT, metadata TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, expires_at DATETIME, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) """ ) conn.commit() conn.close() logger.info("Database initialized successfully") except Exception as e: logger.error(f"Database initialization error: {str(e)}") def get_db_connection(self): """Get database connection""" return sqlite3.connect(self.db_path) async def handle_signup( self, full_name: str, mobile_number: str, password: str, confirm_password: str ) -> Tuple[str, bool, dict]: """Handle user signup""" try: # Validate input if not all([full_name, mobile_number, password, confirm_password]): return "All fields are required", False, gr.update(visible=True) # Create signup request signup_data = UserSignupRequest( full_name=full_name, mobile_number=mobile_number, password=password, confirm_password=confirm_password, ) # Process signup db_conn = self.get_db_connection() auth_service = AuthService(db_conn) result = await auth_service.register_user(signup_data) db_conn.close() if result.success: self.current_session = { "session_id": result.session_id, "user_id": result.user_id, "full_name": result.full_name, } return ( f"Welcome {result.full_name}! Account created successfully.", True, gr.update(visible=False), ) else: return result.message, False, gr.update(visible=True) except ValueError as e: return str(e), False, gr.update(visible=True) except Exception as e: logger.error(f"Signup error: {str(e)}") return "An error occurred during signup", False, gr.update(visible=True) async def handle_login( self, mobile_number: str, password: str ) -> Tuple[str, bool, dict]: """Handle user login""" try: # Validate input if not all([mobile_number, password]): return ( "Mobile number and password are required", False, gr.update(visible=True), ) # Create login request login_data = UserLoginRequest( mobile_number=mobile_number, password=password ) # Process login db_conn = self.get_db_connection() auth_service = AuthService(db_conn) result = await auth_service.login_user(login_data) db_conn.close() if result.success: self.current_session = { "session_id": result.session_id, "user_id": result.user_id, "full_name": result.full_name, } return ( f"Welcome back, {result.full_name}!", True, gr.update(visible=False), ) else: return result.message, False, gr.update(visible=True) except ValueError as e: return str(e), False, gr.update(visible=True) except Exception as e: logger.error(f"Login error: {str(e)}") return "An error occurred during login", False, gr.update(visible=True) def handle_logout(self) -> Tuple[str, bool, dict]: """Handle user logout""" if self.current_session: # In a real app, you'd delete the session from database self.current_session = None return "Logged out successfully", False, gr.update(visible=True) def create_interface(self) -> gr.Interface: """Create the authentication interface""" with gr.Blocks( title="OpenManus Authentication", theme=gr.themes.Soft() ) as auth_interface: gr.Markdown( """ # 🔐 OpenManus Authentication ### Secure Mobile Number + Password Login System """ ) # Session status session_status = gr.Textbox( value="Not logged in", label="Status", interactive=False ) # Auth forms container with gr.Column(visible=True) as auth_forms: with gr.Tabs(): # Login Tab with gr.TabItem("🔑 Login"): gr.Markdown("### Login with your mobile number and password") login_mobile = gr.Textbox( label="📱 Mobile Number", placeholder="Enter your mobile number (e.g., +1234567890)", lines=1, ) login_password = gr.Textbox( label="🔒 Password", type="password", placeholder="Enter your password", lines=1, ) login_btn = gr.Button("🔑 Login", variant="primary", size="lg") login_result = gr.Textbox(label="Result", interactive=False) # Signup Tab with gr.TabItem("📝 Sign Up"): gr.Markdown("### Create your new account") signup_fullname = gr.Textbox( label="👤 Full Name", placeholder="Enter your full name", lines=1, ) signup_mobile = gr.Textbox( label="📱 Mobile Number", placeholder="Enter your mobile number (e.g., +1234567890)", lines=1, ) signup_password = gr.Textbox( label="🔒 Password", type="password", placeholder="Create a strong password (min 8 chars, include uppercase, lowercase, digit)", lines=1, ) signup_confirm_password = gr.Textbox( label="🔒 Confirm Password", type="password", placeholder="Confirm your password", lines=1, ) signup_btn = gr.Button( "📝 Create Account", variant="primary", size="lg" ) signup_result = gr.Textbox(label="Result", interactive=False) # Logged in section with gr.Column(visible=False) as logged_in_section: gr.Markdown("### ✅ You are logged in!") user_info = gr.Markdown("Welcome!") logout_btn = gr.Button("🚪 Logout", variant="secondary") logout_result = gr.Textbox(label="Result", interactive=False) # Password requirements info with gr.Accordion("📋 Password Requirements", open=False): gr.Markdown( """ **Password must contain:** - At least 8 characters - At least 1 uppercase letter (A-Z) - At least 1 lowercase letter (a-z) - At least 1 digit (0-9) - Maximum 128 characters **Mobile Number Format:** - 10-15 digits - Can include country code - Examples: +1234567890, 1234567890, +91987654321 """ ) # Event handlers def sync_signup(*args): """Synchronous wrapper for signup""" return asyncio.run(self.handle_signup(*args)) def sync_login(*args): """Synchronous wrapper for login""" return asyncio.run(self.handle_login(*args)) def update_ui_after_auth(result_text, success, auth_forms_update): """Update UI after authentication""" if success: return ( result_text, # session_status auth_forms_update, # auth_forms visibility gr.update(visible=True), # logged_in_section visibility f"### 👋 {self.current_session['full_name'] if self.current_session else 'User'}", # user_info ) else: return ( "Not logged in", # session_status auth_forms_update, # auth_forms visibility gr.update(visible=False), # logged_in_section visibility "Welcome!", # user_info ) def update_ui_after_logout(result_text, success, auth_forms_update): """Update UI after logout""" return ( "Not logged in", # session_status auth_forms_update, # auth_forms visibility gr.update(visible=False), # logged_in_section visibility "Welcome!", # user_info ) # Login button click login_btn.click( fn=sync_login, inputs=[login_mobile, login_password], outputs=[login_result, gr.State(), gr.State()], ).then( fn=update_ui_after_auth, inputs=[login_result, gr.State(), gr.State()], outputs=[session_status, auth_forms, logged_in_section, user_info], ) # Signup button click signup_btn.click( fn=sync_signup, inputs=[ signup_fullname, signup_mobile, signup_password, signup_confirm_password, ], outputs=[signup_result, gr.State(), gr.State()], ).then( fn=update_ui_after_auth, inputs=[signup_result, gr.State(), gr.State()], outputs=[session_status, auth_forms, logged_in_section, user_info], ) # Logout button click logout_btn.click( fn=self.handle_logout, outputs=[logout_result, gr.State(), gr.State()] ).then( fn=update_ui_after_logout, inputs=[logout_result, gr.State(), gr.State()], outputs=[session_status, auth_forms, logged_in_section, user_info], ) return auth_interface # Standalone authentication app def create_auth_app(db_path: str = "openmanus.db") -> gr.Interface: """Create standalone authentication app""" auth_interface = AuthInterface(db_path) return auth_interface.create_interface() if __name__ == "__main__": # Run standalone auth interface for testing auth_app = create_auth_app() auth_app.launch(server_name="0.0.0.0", server_port=7860, share=False, debug=True)