2025-11-04 21:02:56 +01:00
|
|
|
"""
|
|
|
|
|
Authentication Manager Module
|
|
|
|
|
Handles all authentication-related operations including:
|
|
|
|
|
- Loading/saving auth configuration
|
|
|
|
|
- Password hashing and verification
|
|
|
|
|
- JWT token generation and validation
|
|
|
|
|
- Auth status checking
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
import os
|
|
|
|
|
import json
|
|
|
|
|
import hashlib
|
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
|
from pathlib import Path
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
import jwt
|
|
|
|
|
JWT_AVAILABLE = True
|
|
|
|
|
except ImportError:
|
|
|
|
|
JWT_AVAILABLE = False
|
|
|
|
|
print("Warning: PyJWT not available. Authentication features will be limited.")
|
|
|
|
|
|
|
|
|
|
# Configuration
|
|
|
|
|
CONFIG_DIR = Path.home() / ".config" / "proxmenux-monitor"
|
|
|
|
|
AUTH_CONFIG_FILE = CONFIG_DIR / "auth.json"
|
|
|
|
|
JWT_SECRET = "proxmenux-monitor-secret-key-change-in-production"
|
|
|
|
|
JWT_ALGORITHM = "HS256"
|
|
|
|
|
TOKEN_EXPIRATION_HOURS = 24
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def ensure_config_dir():
|
|
|
|
|
"""Ensure the configuration directory exists"""
|
|
|
|
|
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def load_auth_config():
|
|
|
|
|
"""
|
|
|
|
|
Load authentication configuration from file
|
|
|
|
|
Returns dict with structure:
|
|
|
|
|
{
|
|
|
|
|
"enabled": bool,
|
|
|
|
|
"username": str,
|
|
|
|
|
"password_hash": str,
|
|
|
|
|
"declined": bool, # True if user explicitly declined auth
|
|
|
|
|
"configured": bool # True if auth has been set up (enabled or declined)
|
|
|
|
|
}
|
|
|
|
|
"""
|
|
|
|
|
if not AUTH_CONFIG_FILE.exists():
|
|
|
|
|
return {
|
|
|
|
|
"enabled": False,
|
|
|
|
|
"username": None,
|
|
|
|
|
"password_hash": None,
|
|
|
|
|
"declined": False,
|
|
|
|
|
"configured": False
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
with open(AUTH_CONFIG_FILE, 'r') as f:
|
|
|
|
|
config = json.load(f)
|
|
|
|
|
# Ensure all required fields exist
|
|
|
|
|
config.setdefault("declined", False)
|
|
|
|
|
config.setdefault("configured", config.get("enabled", False) or config.get("declined", False))
|
|
|
|
|
return config
|
|
|
|
|
except Exception as e:
|
|
|
|
|
print(f"Error loading auth config: {e}")
|
|
|
|
|
return {
|
|
|
|
|
"enabled": False,
|
|
|
|
|
"username": None,
|
|
|
|
|
"password_hash": None,
|
|
|
|
|
"declined": False,
|
|
|
|
|
"configured": False
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def save_auth_config(config):
|
|
|
|
|
"""Save authentication configuration to file"""
|
|
|
|
|
ensure_config_dir()
|
|
|
|
|
try:
|
|
|
|
|
with open(AUTH_CONFIG_FILE, 'w') as f:
|
|
|
|
|
json.dump(config, f, indent=2)
|
|
|
|
|
return True
|
|
|
|
|
except Exception as e:
|
|
|
|
|
print(f"Error saving auth config: {e}")
|
|
|
|
|
return False
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def hash_password(password):
|
|
|
|
|
"""Hash a password using SHA-256"""
|
|
|
|
|
return hashlib.sha256(password.encode()).hexdigest()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def verify_password(password, password_hash):
|
|
|
|
|
"""Verify a password against its hash"""
|
|
|
|
|
return hash_password(password) == password_hash
|
|
|
|
|
|
|
|
|
|
|
2025-11-07 20:05:29 +01:00
|
|
|
def generate_token(username):
|
2025-11-04 21:02:56 +01:00
|
|
|
"""Generate a JWT token for the given username"""
|
|
|
|
|
if not JWT_AVAILABLE:
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
payload = {
|
|
|
|
|
'username': username,
|
2025-11-07 20:05:29 +01:00
|
|
|
'exp': datetime.utcnow() + timedelta(hours=TOKEN_EXPIRATION_HOURS),
|
|
|
|
|
'iat': datetime.utcnow()
|
2025-11-04 21:02:56 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
|
|
|
|
return token
|
|
|
|
|
except Exception as e:
|
|
|
|
|
print(f"Error generating token: {e}")
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def verify_token(token):
|
|
|
|
|
"""
|
|
|
|
|
Verify a JWT token
|
|
|
|
|
Returns username if valid, None otherwise
|
|
|
|
|
"""
|
|
|
|
|
if not JWT_AVAILABLE or not token:
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
|
|
|
|
|
return payload.get('username')
|
|
|
|
|
except jwt.ExpiredSignatureError:
|
|
|
|
|
print("Token has expired")
|
|
|
|
|
return None
|
|
|
|
|
except jwt.InvalidTokenError as e:
|
|
|
|
|
print(f"Invalid token: {e}")
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def get_auth_status():
|
|
|
|
|
"""
|
|
|
|
|
Get current authentication status
|
|
|
|
|
Returns dict with:
|
|
|
|
|
{
|
2025-11-04 21:42:38 +01:00
|
|
|
"auth_enabled": bool,
|
|
|
|
|
"auth_configured": bool,
|
2025-11-04 21:02:56 +01:00
|
|
|
"declined": bool,
|
2025-11-04 21:42:38 +01:00
|
|
|
"username": str or None,
|
|
|
|
|
"authenticated": bool
|
2025-11-04 21:02:56 +01:00
|
|
|
}
|
|
|
|
|
"""
|
|
|
|
|
config = load_auth_config()
|
|
|
|
|
return {
|
2025-11-04 21:42:38 +01:00
|
|
|
"auth_enabled": config.get("enabled", False),
|
2025-11-07 20:05:29 +01:00
|
|
|
"auth_configured": config.get("configured", False), # Frontend expects this field name
|
2025-11-04 21:02:56 +01:00
|
|
|
"declined": config.get("declined", False),
|
2025-11-04 21:42:38 +01:00
|
|
|
"username": config.get("username") if config.get("enabled") else None,
|
|
|
|
|
"authenticated": False # Will be set to True by the route handler if token is valid
|
2025-11-04 21:02:56 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def setup_auth(username, password):
|
|
|
|
|
"""
|
|
|
|
|
Set up authentication with username and password
|
|
|
|
|
Returns (success: bool, message: str)
|
|
|
|
|
"""
|
|
|
|
|
if not username or not password:
|
|
|
|
|
return False, "Username and password are required"
|
|
|
|
|
|
|
|
|
|
if len(password) < 6:
|
|
|
|
|
return False, "Password must be at least 6 characters"
|
|
|
|
|
|
|
|
|
|
config = {
|
|
|
|
|
"enabled": True,
|
|
|
|
|
"username": username,
|
|
|
|
|
"password_hash": hash_password(password),
|
|
|
|
|
"declined": False,
|
|
|
|
|
"configured": True
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if save_auth_config(config):
|
|
|
|
|
return True, "Authentication configured successfully"
|
|
|
|
|
else:
|
|
|
|
|
return False, "Failed to save authentication configuration"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def decline_auth():
|
|
|
|
|
"""
|
|
|
|
|
Mark authentication as declined by user
|
|
|
|
|
Returns (success: bool, message: str)
|
|
|
|
|
"""
|
|
|
|
|
config = load_auth_config()
|
|
|
|
|
config["enabled"] = False
|
|
|
|
|
config["declined"] = True
|
|
|
|
|
config["configured"] = True
|
|
|
|
|
config["username"] = None
|
|
|
|
|
config["password_hash"] = None
|
|
|
|
|
|
|
|
|
|
if save_auth_config(config):
|
|
|
|
|
return True, "Authentication declined"
|
|
|
|
|
else:
|
|
|
|
|
return False, "Failed to save configuration"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def disable_auth():
|
|
|
|
|
"""
|
|
|
|
|
Disable authentication (different from decline - can be re-enabled)
|
|
|
|
|
Returns (success: bool, message: str)
|
|
|
|
|
"""
|
2025-11-07 20:05:29 +01:00
|
|
|
config = load_auth_config()
|
|
|
|
|
config["enabled"] = False
|
|
|
|
|
config["username"] = None
|
|
|
|
|
config["password_hash"] = None
|
|
|
|
|
config["declined"] = False
|
|
|
|
|
config["configured"] = False
|
2025-11-04 21:02:56 +01:00
|
|
|
|
|
|
|
|
if save_auth_config(config):
|
2025-11-07 20:05:29 +01:00
|
|
|
return True, "Authentication disabled"
|
2025-11-04 21:02:56 +01:00
|
|
|
else:
|
|
|
|
|
return False, "Failed to save configuration"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def enable_auth():
|
|
|
|
|
"""
|
|
|
|
|
Enable authentication (must already be configured)
|
|
|
|
|
Returns (success: bool, message: str)
|
|
|
|
|
"""
|
|
|
|
|
config = load_auth_config()
|
|
|
|
|
|
|
|
|
|
if not config.get("username") or not config.get("password_hash"):
|
|
|
|
|
return False, "Authentication not configured. Please set up username and password first."
|
|
|
|
|
|
|
|
|
|
config["enabled"] = True
|
|
|
|
|
config["declined"] = False
|
|
|
|
|
|
|
|
|
|
if save_auth_config(config):
|
|
|
|
|
return True, "Authentication enabled"
|
|
|
|
|
else:
|
|
|
|
|
return False, "Failed to save configuration"
|
|
|
|
|
|
|
|
|
|
|
2025-11-07 20:05:29 +01:00
|
|
|
def change_password(old_password, new_password):
|
2025-11-04 21:02:56 +01:00
|
|
|
"""
|
|
|
|
|
Change the authentication password
|
|
|
|
|
Returns (success: bool, message: str)
|
|
|
|
|
"""
|
|
|
|
|
config = load_auth_config()
|
|
|
|
|
|
|
|
|
|
if not config.get("enabled"):
|
|
|
|
|
return False, "Authentication is not enabled"
|
|
|
|
|
|
2025-11-07 20:05:29 +01:00
|
|
|
if not verify_password(old_password, config.get("password_hash", "")):
|
2025-11-04 21:02:56 +01:00
|
|
|
return False, "Current password is incorrect"
|
|
|
|
|
|
|
|
|
|
if len(new_password) < 6:
|
|
|
|
|
return False, "New password must be at least 6 characters"
|
|
|
|
|
|
|
|
|
|
config["password_hash"] = hash_password(new_password)
|
|
|
|
|
|
|
|
|
|
if save_auth_config(config):
|
|
|
|
|
return True, "Password changed successfully"
|
|
|
|
|
else:
|
|
|
|
|
return False, "Failed to save new password"
|
|
|
|
|
|
|
|
|
|
|
2025-11-07 20:05:29 +01:00
|
|
|
def authenticate(username, password):
|
2025-11-04 21:02:56 +01:00
|
|
|
"""
|
|
|
|
|
Authenticate a user with username and password
|
|
|
|
|
Returns (success: bool, token: str or None, message: str)
|
|
|
|
|
"""
|
|
|
|
|
config = load_auth_config()
|
|
|
|
|
|
|
|
|
|
if not config.get("enabled"):
|
|
|
|
|
return False, None, "Authentication is not enabled"
|
|
|
|
|
|
|
|
|
|
if username != config.get("username"):
|
|
|
|
|
return False, None, "Invalid username or password"
|
|
|
|
|
|
|
|
|
|
if not verify_password(password, config.get("password_hash", "")):
|
|
|
|
|
return False, None, "Invalid username or password"
|
|
|
|
|
|
2025-11-07 20:05:29 +01:00
|
|
|
token = generate_token(username)
|
2025-11-04 21:02:56 +01:00
|
|
|
if token:
|
|
|
|
|
return True, token, "Authentication successful"
|
|
|
|
|
else:
|
|
|
|
|
return False, None, "Failed to generate authentication token"
|