Update src/components/auth.py

src/components/auth.py

@@ -1,5 +1,5 @@
 """
-Common issues and fixes for authentication errors in Hugging Face Spaces deployment
+FastAPI compatible authentication module
 """
 
 import jwt
@@ -8,119 +8,116 @@ import json
 import os
 from datetime import datetime, timedelta
 from typing import Optional, Dict, Any
-from functools import wraps
-
-# Import Flask components only when available
-try:
-    from flask import request, jsonify, session, redirect, url_for
-    FLASK_AVAILABLE = True
-except ImportError:
-    FLASK_AVAILABLE = False
-    request = None
-    jsonify = None
-    session = None
-    redirect = None
-    url_for = None
+from pathlib import Path
 
 class AuthManager:
     def __init__(self, secret_key: str = None):
         self.secret_key = secret_key or os.environ.get('SECRET_KEY', 'dev-secret-key-change-in-production')
         
-        # HUGGING FACE FIX: Use absolute paths or ensure directory exists
-        self.base_dir = os.path.abspath(os.path.dirname(__file__))
-        self.data_dir = os.path.join(self.base_dir, 'data')
-        self.users_file = os.path.join(self.data_dir, 'users.json')
-        self.session_file = os.path.join(self.data_dir, 'active_sessions.json')
+        # Use absolute paths and handle Hugging Face Spaces
+        self.base_dir = Path(__file__).parent.parent  # Go up from src/components
+        self.data_dir = self.base_dir / 'data'
+        
+        # Fallback to current directory if base_dir doesn't work
+        if not self.data_dir.exists():
+            self.data_dir = Path.cwd() / 'data'
+        
+        self.users_file = self.data_dir / 'users.json'
+        self.session_file = self.data_dir / 'active_sessions.json'
+        
+        # In-memory fallback for read-only environments
+        self.users_memory = {}
+        self.sessions_memory = {}
+        self.use_memory = False
         
-        self.active_sessions = {}
         self.ensure_users_file()
+        self.ensure_admin_user()  # Ensure admin exists on startup
         
-        # DEBUG: Print initialization info
-        print(f"DEBUG: AuthManager initialized")
-        print(f"DEBUG: Data directory: {self.data_dir}")
-        print(f"DEBUG: Users file: {self.users_file}")
-        print(f"DEBUG: Users file exists: {os.path.exists(self.users_file)}")
+        print(f"✅ AuthManager initialized")
+        print(f"📁 Data directory: {self.data_dir}")
+        print(f"📄 Users file: {self.users_file}")
+        print(f"💾 Using memory storage: {self.use_memory}")
     
     def ensure_users_file(self):
         """Ensure users file exists with better error handling"""
         try:
-            os.makedirs(self.data_dir, exist_ok=True)
-            print(f"DEBUG: Data directory created/exists: {self.data_dir}")
+            self.data_dir.mkdir(parents=True, exist_ok=True)
             
-            if not os.path.exists(self.users_file):
+            if not self.users_file.exists():
                 with open(self.users_file, 'w') as f:
                     json.dump({}, f)
-                print(f"DEBUG: Created empty users file: {self.users_file}")
-            else:
-                print(f"DEBUG: Users file already exists: {self.users_file}")
-                
+                print(f"✅ Created users file: {self.users_file}")
+            
+            # Test write permissions
+            test_data = self.load_users()
+            self.save_users(test_data)
+            print(f"✅ Write permissions confirmed")
+            
         except Exception as e:
-            print(f"ERROR: Failed to create users file: {e}")
-            # Create fallback in-memory storage
-            self.users_file = None
-            print("DEBUG: Using in-memory user storage as fallback")
+            print(f"⚠️ File system error: {e}")
+            print(f"🔄 Switching to in-memory storage")
+            self.use_memory = True
+            self.users_memory = {}
+            self.sessions_memory = {}
     
-    def hash_password(self, password: str) -> str:
-        """Hash password with bcrypt"""
+    def ensure_admin_user(self):
+        """Ensure admin user exists on startup"""
         try:
-            return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
+            result = self.create_default_admin()
+            if result.get('success'):
+                print(f"✅ Admin user ready: {result.get('message', 'Available')}")
         except Exception as e:
-            print(f"ERROR: Password hashing failed: {e}")
-            raise
+            print(f"⚠️ Admin user creation failed: {e}")
+    
+    def hash_password(self, password: str) -> str:
+        """Hash password with bcrypt"""
+        return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
     
     def verify_password(self, password: str, hashed: str) -> bool:
         """Verify password against hash"""
         try:
             return bcrypt.checkpw(password.encode('utf-8'), hashed.encode('utf-8'))
         except Exception as e:
-            print(f"ERROR: Password verification failed: {e}")
+            print(f"❌ Password verification error: {e}")
             return False
     
     def load_users(self) -> Dict[str, Any]:
-        """Load users from file with better error handling"""
-        if not self.users_file:
-            print("DEBUG: No users file, returning empty dict")
-            return {}
-            
+        """Load users from file or memory"""
+        if self.use_memory:
+            return self.users_memory.copy()
+        
         try:
-            with open(self.users_file, 'r') as f:
-                users = json.load(f)
-                print(f"DEBUG: Loaded {len(users)} users from file")
-                return users
-        except FileNotFoundError:
-            print("DEBUG: Users file not found, returning empty dict")
+            if self.users_file.exists():
+                with open(self.users_file, 'r') as f:
+                    users = json.load(f)
+                    return users
             return {}
         except Exception as e:
-            print(f"ERROR: Failed to load users: {e}")
+            print(f"❌ Error loading users: {e}")
             return {}
     
     def save_users(self, users: Dict[str, Any]):
-        """Save users to file with better error handling"""
-        if not self.users_file:
-            print("DEBUG: No users file configured, cannot save")
+        """Save users to file or memory"""
+        if self.use_memory:
+            self.users_memory = users.copy()
             return
-            
+        
         try:
             with open(self.users_file, 'w') as f:
                 json.dump(users, f, indent=2)
-            print(f"DEBUG: Saved {len(users)} users to file")
         except Exception as e:
-            print(f"ERROR: Failed to save users: {e}")
+            print(f"❌ Error saving users: {e}")
     
     def create_user(self, username: str, email: str, password: str) -> Dict[str, Any]:
-        """Create a new user with debug info"""
-        print(f"DEBUG: Creating user '{username}' with email '{email}'")
-        
+        """Create a new user"""
         users = self.load_users()
         
         if username in users:
-            print(f"DEBUG: Username '{username}' already exists")
             return {'success': False, 'error': 'Username already exists'}
         
         # Check if email already exists
         for user_data in users.values():
             if user_data.get('email') == email:
-                print(f"DEBUG: Email '{email}' already registered")
                 return {'success': False, 'error': 'Email already registered'}
         
         # Create user
@@ -134,34 +131,30 @@ class AuthManager:
         }
         
         self.save_users(users)
-        print(f"DEBUG: User '{username}' created successfully with ID '{user_id}'")
         return {'success': True, 'user_id': user_id}
     
     def authenticate_user(self, username: str, password: str) -> Dict[str, Any]:
-        """Authenticate user credentials with detailed debug info"""
-        print(f"DEBUG: Authenticating user '{username}'")
+        """Authenticate user credentials with detailed logging"""
+        print(f"🔐 Authentication attempt for: {username}")
         
         users = self.load_users()
-        print(f"DEBUG: Total users in database: {len(users)}")
-        print(f"DEBUG: Available usernames: {list(users.keys())}")
+        print(f"📊 Total users in database: {len(users)}")
         
         if username not in users:
-            print(f"DEBUG: Username '{username}' not found in database")
+            print(f"❌ Username '{username}' not found")
+            print(f"📋 Available usernames: {list(users.keys())}")
             return {'success': False, 'error': 'Invalid username or password'}
         
         user = users[username]
-        print(f"DEBUG: Found user '{username}', checking password...")
         
         if not self.verify_password(password, user['password_hash']):
-            print(f"DEBUG: Password verification failed for '{username}'")
+            print(f"❌ Invalid password for '{username}'")
             return {'success': False, 'error': 'Invalid username or password'}
         
         if not user.get('is_active', True):
-            print(f"DEBUG: User '{username}' is not active")
+            print(f"❌ User '{username}' is not active")
             return {'success': False, 'error': 'Account is disabled'}
         
-        print(f"DEBUG: User '{username}' authenticated successfully")
-        
         # Generate JWT token
         try:
             token = jwt.encode({
@@ -170,11 +163,10 @@ class AuthManager:
                 'exp': datetime.utcnow() + timedelta(hours=8)
             }, self.secret_key, algorithm='HS256')
             
-            print(f"DEBUG: JWT token generated for '{username}'")
-            
             # Track active session
             self.add_active_session(user['user_id'], token)
             
+            print(f"✅ Authentication successful for '{username}'")
             return {
                 'success': True,
                 'token': token,
@@ -182,42 +174,34 @@ class AuthManager:
                 'username': username
             }
         except Exception as e:
-            print(f"ERROR: JWT token generation failed: {e}")
+            print(f"❌ JWT generation failed: {e}")
             return {'success': False, 'error': 'Authentication failed'}
     
     def verify_token(self, token: str) -> Optional[Dict[str, Any]]:
-        """Verify JWT token with debug info"""
-        print(f"DEBUG: Verifying token...")
-        
+        """Verify JWT token"""
         try:
             payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
             user_id = payload.get('user_id')
-            print(f"DEBUG: Token decoded successfully for user_id: {user_id}")
             
             # Check if session is still active
             if not self.is_session_active(user_id, token):
-                print(f"DEBUG: Session not active for user_id: {user_id}")
                 return None
             
             # Update session activity
             self.update_session_activity(user_id)
-            print(f"DEBUG: Token verified successfully for user_id: {user_id}")
-            
             return payload
         except jwt.ExpiredSignatureError:
-            print("DEBUG: Token expired")
+            print("🕐 Token expired")
             return None
-        except jwt.InvalidTokenError as e:
-            print(f"DEBUG: Invalid token: {e}")
+        except jwt.InvalidTokenError:
+            print("❌ Invalid token")
             return None
         except Exception as e:
-            print(f"ERROR: Token verification failed: {e}")
+            print(f"❌ Token verification error: {e}")
             return None
     
     def create_default_admin(self) -> Dict[str, Any]:
         """Create default admin user if it doesn't exist"""
-        print("DEBUG: Creating default admin user...")
-        
         users = self.load_users()
         
         admin_username = "admin"
@@ -225,59 +209,54 @@ class AuthManager:
         
         # Check if admin already exists
         if admin_username in users:
-            print("DEBUG: Admin user already exists")
             return {'success': True, 'message': 'Admin user already exists'}
         
-        # Check if user_id already exists
-        for user_data in users.values():
-            if user_data.get('user_id') == admin_user_id:
-                print("DEBUG: Admin user ID already exists")
-                return {'success': True, 'message': 'Admin user ID already exists'}
-        
         # Create admin user
-        users[admin_username] = {
-            'user_id': admin_user_id,
-            'email': '[email protected]',
-            'password_hash': self.hash_password('admin123'),
-            'created_at': datetime.now().isoformat(),
-            'is_active': True,
-            'is_admin': True
-        }
-        
-        self.save_users(users)
-        print("DEBUG: Default admin user created successfully")
-        
-        return {
-            'success': True, 
-            'message': 'Default admin user created',
-            'username': admin_username,
-            'password': 'admin123',
-            'note': 'Please change the default password after first login'
-        }
+        try:
+            users[admin_username] = {
+                'user_id': admin_user_id,
+                'email': 'admin@researchmate.local',
+                'password_hash': self.hash_password('admin123'),
+                'created_at': datetime.now().isoformat(),
+                'is_active': True,
+                'is_admin': True
+            }
+            
+            self.save_users(users)
+            return {
+                'success': True, 
+                'message': 'Default admin user created',
+                'username': admin_username,
+                'password': 'admin123'
+            }
+        except Exception as e:
+            print(f"❌ Failed to create admin user: {e}")
+            return {'success': False, 'error': str(e)}
     
     def load_active_sessions(self) -> Dict[str, Any]:
-        """Load active sessions from file"""
-        if not self.session_file:
-            return {}
-            
+        """Load active sessions"""
+        if self.use_memory:
+            return self.sessions_memory.copy()
+        
         try:
-            if os.path.exists(self.session_file):
+            if self.session_file.exists():
                 with open(self.session_file, 'r') as f:
                     return json.load(f)
         except Exception as e:
-            print(f"ERROR: Failed to load active sessions: {e}")
+            print(f"❌ Error loading sessions: {e}")
         return {}
     
     def save_active_sessions(self, sessions: Dict[str, Any]):
-        """Save active sessions to file"""
-        if not self.session_file:
+        """Save active sessions"""
+        if self.use_memory:
+            self.sessions_memory = sessions.copy()
             return
-            
+        
         try:
             with open(self.session_file, 'w') as f:
                 json.dump(sessions, f, indent=2)
         except Exception as e:
-            print(f"ERROR: Failed to save active sessions: {e}")
+            print(f"❌ Error saving sessions: {e}")
     
     def add_active_session(self, user_id: str, token: str):
         """Add an active session"""
@@ -288,132 +267,86 @@ class AuthManager:
             'last_activity': datetime.now().isoformat()
         }
         self.save_active_sessions(sessions)
-        print(f"DEBUG: Added active session for user_id: {user_id}")
-    
-    def remove_active_session(self, user_id: str):
-        """Remove an active session"""
-        sessions = self.load_active_sessions()
-        if user_id in sessions:
-            del sessions[user_id]
-            self.save_active_sessions(sessions)
-            print(f"DEBUG: Removed active session for user_id: {user_id}")
     
     def is_session_active(self, user_id: str, token: str) -> bool:
         """Check if a session is active"""
         sessions = self.load_active_sessions()
         if user_id not in sessions:
-            print(f"DEBUG: No session found for user_id: {user_id}")
             return False
         
         session = sessions[user_id]
         if session.get('token') != token:
-            print(f"DEBUG: Token mismatch for user_id: {user_id}")
             return False
         
         # Check if session is expired
         try:
             created_at = datetime.fromisoformat(session['created_at'])
             if datetime.now() - created_at > timedelta(hours=8):
-                print(f"DEBUG: Session expired for user_id: {user_id}")
                 self.remove_active_session(user_id)
                 return False
-        except Exception as e:
-            print(f"ERROR: Failed to check session expiry: {e}")
+        except:
             return False
         
         return True
     
+    def remove_active_session(self, user_id: str):
+        """Remove an active session"""
+        sessions = self.load_active_sessions()
+        if user_id in sessions:
+            del sessions[user_id]
+            self.save_active_sessions(sessions)
+    
     def update_session_activity(self, user_id: str):
         """Update last activity time for a session"""
         sessions = self.load_active_sessions()
         if user_id in sessions:
             sessions[user_id]['last_activity'] = datetime.now().isoformat()
             self.save_active_sessions(sessions)
-
-# Global auth manager
-auth_manager = AuthManager()
-
-def require_auth(f):
-    """Decorator to require authentication"""
-    @wraps(f)
-    def decorated_function(*args, **kwargs):
-        if not FLASK_AVAILABLE:
-            return f(*args, **kwargs)
-            
-        user = auth_manager.get_current_user(request)
-        if not user:
-            print("DEBUG: Authentication required but no user found")
-            if request.is_json:
-                return jsonify({'success': False, 'error': 'Authentication required'}), 401
-            else:
-                return redirect(url_for('login'))
-        return f(*args, **kwargs)
-    return decorated_function
-
-def get_current_user() -> Optional[Dict[str, Any]]:
-    """Get current authenticated user"""
-    if not FLASK_AVAILABLE:
-        return None
-    return auth_manager.get_current_user(request)
-
-# ADDITIONAL DEBUGGING FUNCTIONS
-
-def debug_auth_status():
-    """Debug function to check authentication status"""
-    print("=== AUTH DEBUG STATUS ===")
-    users = auth_manager.load_users()
-    print(f"Total users: {len(users)}")
-    for username, user_data in users.items():
-        print(f"  - {username}: {user_data.get('user_id', 'No ID')}")
     
-    sessions = auth_manager.load_active_sessions()
-    print(f"Active sessions: {len(sessions)}")
-    for user_id, session_data in sessions.items():
-        print(f"  - {user_id}: {session_data.get('created_at', 'No timestamp')}")
+    def logout_user(self, user_id: str):
+        """Logout user and invalidate session"""
+        self.remove_active_session(user_id)
+        return {'success': True, 'message': 'Logged out successfully'}
     
-    print(f"Users file exists: {os.path.exists(auth_manager.users_file) if auth_manager.users_file else 'No file path'}")
-    print(f"Session file exists: {os.path.exists(auth_manager.session_file) if auth_manager.session_file else 'No file path'}")
-    print("========================")
-
-# FLASK ROUTE EXAMPLE WITH DEBUG
-def create_debug_login_route(app):
-    """Example login route with debugging"""
-    @app.route('/api/auth/login', methods=['POST'])
-    def login():
-        print("DEBUG: Login endpoint called")
+    def cleanup_expired_sessions(self):
+        """Clean up expired sessions"""
+        sessions = self.load_active_sessions()
+        current_time = datetime.now()
         
-        try:
-            data = request.get_json()
-            print(f"DEBUG: Request data: {data}")
-            
-            if not data:
-                print("DEBUG: No JSON data received")
-                return jsonify({'success': False, 'error': 'No data provided'}), 400
-            
-            username = data.get('username')
-            password = data.get('password')
-            
-            print(f"DEBUG: Login attempt for username: {username}")
-            
-            if not username or not password:
-                print("DEBUG: Missing username or password")
-                return jsonify({'success': False, 'error': 'Username and password required'}), 400
-            
-            # Ensure admin user exists
-            admin_result = auth_manager.create_default_admin()
-            print(f"DEBUG: Admin creation result: {admin_result}")
-            
-            # Authenticate user
-            result = auth_manager.authenticate_user(username, password)
-            print(f"DEBUG: Authentication result: {result}")
-            
-            if result['success']:
-                print("DEBUG: Authentication successful")
-                return jsonify(result), 200
-            else:
-                print("DEBUG: Authentication failed")
-                return jsonify(result), 401
-                
-        except Exception as e:
-            print(f"ERROR: Login route exception: {e}")
-            return jsonify({'success': False, 'error': 'Internal server error'}), 500
+        expired_sessions = []
+        for user_id, session in sessions.items():
+            try:
+                created_at = datetime.fromisoformat(session['created_at'])
+                if current_time - created_at > timedelta(hours=8):
+                    expired_sessions.append(user_id)
+            except:
+                expired_sessions.append(user_id)
+        
+        for user_id in expired_sessions:
+            del sessions[user_id]
+        
+        if expired_sessions:
+            self.save_active_sessions(sessions)
+        
+        return len(expired_sessions)
+    
+    def debug_status(self):
+        """Debug authentication status"""
+        users = self.load_users()
+        sessions = self.load_active_sessions()
+        
+        print("=== AUTH DEBUG STATUS ===")
+        print(f"Storage mode: {'Memory' if self.use_memory else 'File'}")
+        print(f"Users file exists: {self.users_file.exists() if not self.use_memory else 'N/A'}")
+        print(f"Total users: {len(users)}")
+        print(f"Active sessions: {len(sessions)}")
+        
+        if users:
+            print("Users:")
+            for username, user_data in users.items():
+                print(f"  - {username}: {user_data.get('user_id', 'No ID')}")
+        
+        print("========================")
+
+# Global auth manager instance
+auth_manager = AuthManager()