File size: 8,267 Bytes
439e1dd |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 |
#!/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()
|