auth.py

#!/usr/bin/env python3
"""
Authentication module for News Dashboard
Handles user authentication, session management, and security
"""

import hashlib
import secrets
import json
import os
from datetime import datetime, timedelta
from typing import Dict, Optional, Tuple
import logging

logger = logging.getLogger(__name__)

class AuthManager:
    """Manages user authentication and sessions"""
    
    def __init__(self, users_file: str = "users.json", sessions_file: str = "sessions.json"):
        self.users_file = users_file
        self.sessions_file = sessions_file
        self.users = self._load_users()
        self.sessions = self._load_sessions()
        self.session_timeout = timedelta(hours=24)  # 24 hours session timeout
        
        # Create default admin user if no users exist
        if not self.users:
            self._create_default_admin()
    
    def _load_users(self) -> Dict[str, Dict]:
        """Load users from JSON file"""
        try:
            if os.path.exists(self.users_file):
                with open(self.users_file, 'r') as f:
                    return json.load(f)
        except Exception as e:
            logger.error(f"Error loading users: {e}")
        return {}
    
    def _save_users(self):
        """Save users to JSON file"""
        try:
            with open(self.users_file, 'w') as f:
                json.dump(self.users, f, indent=2)
        except Exception as e:
            logger.error(f"Error saving users: {e}")
    
    def _load_sessions(self) -> Dict[str, Dict]:
        """Load sessions from JSON file"""
        try:
            if os.path.exists(self.sessions_file):
                with open(self.sessions_file, 'r') as f:
                    return json.load(f)
        except Exception as e:
            logger.error(f"Error loading sessions: {e}")
        return {}
    
    def _save_sessions(self):
        """Save sessions to JSON file"""
        try:
            with open(self.sessions_file, 'w') as f:
                json.dump(self.sessions, f, indent=2)
        except Exception as e:
            logger.error(f"Error saving sessions: {e}")
    
    def _create_default_admin(self):
        """Create default admin user"""
        admin_password = "admin123"  # Default password - should be changed
        self.add_user("admin", admin_password, is_admin=True)
        logger.warning("Created default admin user with password 'admin123' - PLEASE CHANGE THIS!")
    
    def _hash_password(self, password: str) -> str:
        """Hash password using SHA-256 with salt"""
        salt = secrets.token_hex(16)
        password_hash = hashlib.sha256((password + salt).encode()).hexdigest()
        return f"{salt}:{password_hash}"
    
    def _verify_password(self, password: str, stored_hash: str) -> bool:
        """Verify password against stored hash"""
        try:
            salt, password_hash = stored_hash.split(':')
            return hashlib.sha256((password + salt).encode()).hexdigest() == password_hash
        except:
            return False
    
    def add_user(self, username: str, password: str, is_admin: bool = False) -> bool:
        """Add a new user"""
        if username in self.users:
            return False
        
        self.users[username] = {
            'password_hash': self._hash_password(password),
            'is_admin': is_admin,
            'created_at': datetime.now().isoformat(),
            'last_login': None
        }
        self._save_users()
        logger.info(f"Added user: {username}")
        return True
    
    def authenticate_user(self, username: str, password: str) -> Tuple[bool, str]:
        """Authenticate user and return (success, session_token)"""
        if username not in self.users:
            return False, ""
        
        user = self.users[username]
        if not self._verify_password(password, user['password_hash']):
            return False, ""
        
        # Update last login
        user['last_login'] = datetime.now().isoformat()
        self._save_users()
        
        # Create session
        session_token = secrets.token_urlsafe(32)
        self.sessions[session_token] = {
            'username': username,
            'created_at': datetime.now().isoformat(),
            'last_activity': datetime.now().isoformat()
        }
        self._save_sessions()
        
        logger.info(f"User {username} authenticated successfully")
        return True, session_token
    
    def validate_session(self, session_token: str) -> Tuple[bool, Optional[str]]:
        """Validate session token and return (valid, username)"""
        if not session_token or session_token not in self.sessions:
            return False, None
        
        session = self.sessions[session_token]
        last_activity = datetime.fromisoformat(session['last_activity'])
        
        # Check if session has expired
        if datetime.now() - last_activity > self.session_timeout:
            self.logout_user(session_token)
            return False, None
        
        # Update last activity
        session['last_activity'] = datetime.now().isoformat()
        self._save_sessions()
        
        return True, session['username']
    
    def logout_user(self, session_token: str) -> bool:
        """Logout user by removing session"""
        if session_token in self.sessions:
            del self.sessions[session_token]
            self._save_sessions()
            return True
        return False
    
    def is_admin(self, username: str) -> bool:
        """Check if user is admin"""
        return username in self.users and self.users[username].get('is_admin', False)
    
    def change_password(self, username: str, old_password: str, new_password: str) -> bool:
        """Change user password"""
        if username not in self.users:
            return False
        
        user = self.users[username]
        if not self._verify_password(old_password, user['password_hash']):
            return False
        
        user['password_hash'] = self._hash_password(new_password)
        self._save_users()
        logger.info(f"Password changed for user: {username}")
        return True
    
    def get_user_info(self, username: str) -> Optional[Dict]:
        """Get user information (without password hash)"""
        if username not in self.users:
            return None
        
        user = self.users[username].copy()
        del user['password_hash']  # Remove password hash from response
        return user
    
    def list_users(self) -> Dict[str, Dict]:
        """List all users (admin only)"""
        result = {}
        for username, user in self.users.items():
            result[username] = {
                'is_admin': user.get('is_admin', False),
                'created_at': user.get('created_at'),
                'last_login': user.get('last_login')
            }
        return result
    
    def delete_user(self, username: str) -> bool:
        """Delete user (admin only)"""
        if username not in self.users:
            return False
        
        # Remove all sessions for this user
        sessions_to_remove = []
        for token, session in self.sessions.items():
            if session['username'] == username:
                sessions_to_remove.append(token)
        
        for token in sessions_to_remove:
            del self.sessions[token]
        
        del self.users[username]
        self._save_users()
        self._save_sessions()
        logger.info(f"Deleted user: {username}")
        return True
    
    def cleanup_expired_sessions(self):
        """Remove expired sessions"""
        current_time = datetime.now()
        expired_sessions = []
        
        for token, session in self.sessions.items():
            last_activity = datetime.fromisoformat(session['last_activity'])
            if current_time - last_activity > self.session_timeout:
                expired_sessions.append(token)
        
        for token in expired_sessions:
            del self.sessions[token]
        
        if expired_sessions:
            self._save_sessions()
            logger.info(f"Cleaned up {len(expired_sessions)} expired sessions")


# Global auth manager instance
auth_manager = AuthManager()