446 lines
18 KiB
Python
446 lines
18 KiB
Python
"""User records, account settings, password hashing, and request authentication."""
|
|
|
|
|
|
import bcrypt
|
|
from database_service import database_connection, dict_from_row
|
|
from datetime import datetime, timedelta, timezone
|
|
from email_service import get_email_service
|
|
from flask import request
|
|
import jwt
|
|
from runtime_config import (
|
|
DATABASE_URL,
|
|
JWT_SECRET,
|
|
SALT_ROUNDS,
|
|
logger,
|
|
)
|
|
import sqlite3
|
|
from typing import Dict, Optional
|
|
import uuid
|
|
|
|
__all__ = ['add_credit_transaction', 'add_credits', 'authenticate_user', 'create_user', 'generate_oauth_state', 'generate_token', 'generate_two_factor_token', 'get_bearer_token', 'get_full_user_by_id', 'get_full_user_by_username', 'get_new_user_credits', 'get_panel_settings', 'get_pricing', 'get_site_setting', 'get_user_by_apple_id', 'get_user_by_email', 'get_user_by_google_id', 'get_user_by_id', 'get_user_by_username', 'get_user_credits', 'hash_password', 'is_demo_enabled', 'require_admin', 'set_site_setting', 'update_user_credits', 'verify_oauth_state', 'verify_password', 'verify_token']
|
|
|
|
|
|
|
|
def get_site_setting(key: str, default: str = "") -> str:
|
|
"""Get a single setting value from site_settings table"""
|
|
try:
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT value FROM site_settings WHERE key = ?", (key,))
|
|
row = cursor.fetchone()
|
|
if row and row['value'] is not None:
|
|
return str(row['value'])
|
|
except Exception as e:
|
|
logger.error(f"Error fetching site setting {key}: {e}")
|
|
return default
|
|
|
|
def set_site_setting(key: str, value: str) -> bool:
|
|
"""Insert or update a setting in site_settings table"""
|
|
try:
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
if 'postgres' in DATABASE_URL:
|
|
cursor.execute(
|
|
"INSERT INTO site_settings (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = CURRENT_TIMESTAMP",
|
|
(key, str(value))
|
|
)
|
|
else:
|
|
cursor.execute("SELECT 1 FROM site_settings WHERE key = ?", (key,))
|
|
if cursor.fetchone():
|
|
cursor.execute(
|
|
"UPDATE site_settings SET value = ?, updated_at = CURRENT_TIMESTAMP WHERE key = ?",
|
|
(str(value), key)
|
|
)
|
|
else:
|
|
cursor.execute(
|
|
"INSERT INTO site_settings (key, value) VALUES (?, ?)",
|
|
(key, str(value))
|
|
)
|
|
conn.commit()
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Error setting site setting {key}: {e}")
|
|
return False
|
|
|
|
def is_demo_enabled() -> bool:
|
|
"""Check if demo server creation is globally enabled"""
|
|
val = get_site_setting('demo_servers_enabled', 'true')
|
|
return str(val).lower() in ('true', '1', 'yes', 'on')
|
|
|
|
def get_new_user_credits() -> int:
|
|
"""Get the initial credits granted to new users"""
|
|
try:
|
|
val = get_site_setting('new_user_credits', '0')
|
|
return max(0, int(float(val)))
|
|
except (ValueError, TypeError):
|
|
return 0
|
|
|
|
def get_panel_settings() -> Dict:
|
|
"""Get all panel settings"""
|
|
return {
|
|
'demo_servers_enabled': is_demo_enabled(),
|
|
'new_user_credits': get_new_user_credits()
|
|
}
|
|
|
|
def create_user(username: str, email: str, password: str = None, role: str = 'user', credits: Optional[int] = None,
|
|
display_name: str = None, first_name: str = None, last_name: str = None,
|
|
address: str = None, city: str = None, country: str = None,
|
|
postal_code: str = None, phone: str = None, google_id: str = None,
|
|
apple_id: str = None, email_verified: bool = False) -> Optional[Dict]:
|
|
"""Create a new user with extended profile and email verification"""
|
|
try:
|
|
user_id = f"user-{str(uuid.uuid4())[:8]}"
|
|
password_hash = hash_password(password) if password else None
|
|
email_verification_token = str(uuid.uuid4()) if not email_verified else None
|
|
|
|
if credits is None:
|
|
credits = get_new_user_credits()
|
|
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
INSERT INTO users (id, username, email, password_hash, google_id, apple_id, role, credits,
|
|
display_name, first_name, last_name, address, city,
|
|
country, postal_code, phone, email_verification_token,
|
|
email_verified)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
''', (user_id, username, email, password_hash, google_id, apple_id, role, credits,
|
|
display_name, first_name, last_name, address, city,
|
|
country, postal_code, phone, email_verification_token, email_verified))
|
|
conn.commit()
|
|
|
|
if credits > 0:
|
|
txn_id = f"txn-{str(uuid.uuid4())[:8]}"
|
|
cursor.execute('''
|
|
INSERT INTO credit_transactions (id, user_id, amount, type, description)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
''', (txn_id, user_id, credits, 'signup', 'Sign-up bonus credits'))
|
|
conn.commit()
|
|
|
|
# Create email verification token record if needed
|
|
if email_verification_token:
|
|
token_id = f"token-{str(uuid.uuid4())[:8]}"
|
|
expires_at = (datetime.now() + timedelta(hours=24)).isoformat()
|
|
|
|
cursor.execute('''
|
|
INSERT INTO email_verification_tokens (id, user_id, token, expires_at)
|
|
VALUES (?, ?, ?, ?)
|
|
''', (token_id, user_id, email_verification_token, expires_at))
|
|
conn.commit()
|
|
|
|
# Send verification email
|
|
email_svc = get_email_service()
|
|
if email_svc:
|
|
success = email_svc.send_verification_email(email, username, email_verification_token)
|
|
if success:
|
|
logger.info(f"✅ Verification email sent to {email}")
|
|
else:
|
|
logger.error(f"❌ Failed to send verification email to {email}")
|
|
else:
|
|
logger.warning("⚠️ Email service not available")
|
|
|
|
cursor.execute('''
|
|
SELECT id, username, email, role, credits, created_at, display_name,
|
|
first_name, last_name, address, city, country, postal_code, phone,
|
|
email_verified
|
|
FROM users WHERE id = ?
|
|
''', (user_id,))
|
|
user = dict_from_row(cursor.fetchone())
|
|
|
|
return user
|
|
except sqlite3.IntegrityError as e:
|
|
logger.error(f"User creation failed - integrity error: {e}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"User creation failed: {e}", exc_info=True)
|
|
return None
|
|
|
|
def get_user_by_google_id(google_id: str) -> Optional[Dict]:
|
|
"""Get user by Google ID"""
|
|
try:
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
SELECT id, username, email, role, credits, created_at, display_name,
|
|
first_name, last_name, address, city, country, postal_code, phone,
|
|
email_verified
|
|
FROM users WHERE google_id = ?
|
|
''', (google_id,))
|
|
row = cursor.fetchone()
|
|
return dict_from_row(row) if row else None
|
|
except Exception as e:
|
|
logger.error(f"Error getting user by Google ID: {e}")
|
|
return None
|
|
|
|
def get_user_by_apple_id(apple_id: str) -> Optional[Dict]:
|
|
"""Get user by Apple ID"""
|
|
try:
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
SELECT id, username, email, role, credits, created_at, display_name,
|
|
first_name, last_name, address, city, country, postal_code, phone,
|
|
email_verified
|
|
FROM users WHERE apple_id = ?
|
|
''', (apple_id,))
|
|
row = cursor.fetchone()
|
|
return dict_from_row(row) if row else None
|
|
except Exception as e:
|
|
logger.error(f"Error getting user by Apple ID: {e}")
|
|
return None
|
|
|
|
def get_user_by_email(email: str) -> Optional[Dict]:
|
|
"""Get user by email"""
|
|
try:
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
SELECT id, username, email, role, credits, created_at, display_name,
|
|
first_name, last_name, address, city, country, postal_code, phone,
|
|
email_verified, google_id, apple_id
|
|
FROM users WHERE email = ?
|
|
''', (email,))
|
|
row = cursor.fetchone()
|
|
return dict_from_row(row) if row else None
|
|
except Exception as e:
|
|
logger.error(f"Error getting user by email: {e}")
|
|
return None
|
|
|
|
def get_pricing() -> Dict[str, float]:
|
|
"""Get all pricing and multipliers from site_settings."""
|
|
try:
|
|
# Defaults
|
|
defaults = {
|
|
'price_cpu': 35.0,
|
|
'price_ram': 35.0,
|
|
'price_storage': 5.0,
|
|
'multiplier_vps': 1.0,
|
|
'multiplier_game': 0.8,
|
|
'multiplier_website': 1.5,
|
|
'multiplier_database': 1.3
|
|
}
|
|
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT key, value FROM site_settings WHERE key LIKE 'price_%' OR key LIKE 'multiplier_%'")
|
|
rows = cursor.fetchall()
|
|
|
|
pricing = {}
|
|
for row in rows:
|
|
key = row['key']
|
|
val_str = row['value']
|
|
try:
|
|
# Filter out 'None' or empty strings which might cause ValueError
|
|
if val_str and val_str.lower() != 'none':
|
|
pricing[key] = float(val_str)
|
|
except (ValueError, TypeError):
|
|
logger.warning(f"Invalid pricing value for {key}: {val_str}")
|
|
continue
|
|
|
|
# Ensure all defaults are present if not in DB or if value was invalid
|
|
for key, val in defaults.items():
|
|
if key not in pricing:
|
|
pricing[key] = val
|
|
|
|
return pricing
|
|
except Exception as e:
|
|
logger.error(f"Error getting pricing: {e}")
|
|
return {
|
|
'price_cpu': 35.0, 'price_ram': 35.0, 'price_storage': 5.0,
|
|
'multiplier_vps': 1.0, 'multiplier_game': 0.8,
|
|
'multiplier_website': 1.5, 'multiplier_database': 1.3
|
|
}
|
|
|
|
def get_user_by_id(user_id: str) -> Optional[Dict]:
|
|
"""Get user by ID"""
|
|
try:
|
|
# Ensure user_id is a string
|
|
user_id = str(user_id)
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
'SELECT id, username, email, role, credits, created_at FROM users WHERE id = ?',
|
|
(user_id,)
|
|
)
|
|
return dict_from_row(cursor.fetchone())
|
|
except Exception as e:
|
|
return None
|
|
|
|
def get_full_user_by_id(user_id: str) -> Optional[Dict]:
|
|
"""Get full user data by ID, including password and 2FA secret."""
|
|
try:
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))
|
|
return dict_from_row(cursor.fetchone())
|
|
except Exception as e:
|
|
logger.error(f"Error getting full user by ID: {e}")
|
|
return None
|
|
|
|
def get_user_by_username(username: str) -> Optional[Dict]:
|
|
"""Get user by username"""
|
|
try:
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
SELECT id, username, email, password_hash, role, credits, created_at, email_verified
|
|
FROM users WHERE username = ?
|
|
''', (username,))
|
|
return dict_from_row(cursor.fetchone())
|
|
except Exception as e:
|
|
logger.error(f"Error getting user by username: {e}")
|
|
return None
|
|
|
|
def get_full_user_by_username(username: str) -> Optional[Dict]:
|
|
"""Get full user data by username, including password and 2FA secret."""
|
|
try:
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('SELECT * FROM users WHERE username = ?', (username,))
|
|
return dict_from_row(cursor.fetchone())
|
|
except Exception as e:
|
|
logger.error(f"Error getting full user by username: {e}")
|
|
return None
|
|
|
|
def update_user_credits(user_id: str, new_credits: int) -> bool:
|
|
"""Update user credits"""
|
|
try:
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
'UPDATE users SET credits = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?',
|
|
(new_credits, user_id)
|
|
)
|
|
conn.commit()
|
|
return cursor.rowcount > 0
|
|
except Exception as e:
|
|
return False
|
|
|
|
def add_credit_transaction(user_id: str, amount: int, transaction_type: str, description: str = "") -> bool:
|
|
"""Add credit transaction"""
|
|
try:
|
|
transaction_id = f"txn-{str(uuid.uuid4())[:8]}"
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
INSERT INTO credit_transactions (id, user_id, amount, type, description)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
''', (transaction_id, user_id, amount, transaction_type, description))
|
|
conn.commit()
|
|
return True
|
|
except Exception as e:
|
|
return False
|
|
|
|
def generate_token(user_id: str) -> str:
|
|
payload = {
|
|
'userId': user_id,
|
|
'token_type': 'access',
|
|
'exp': datetime.now(timezone.utc) + timedelta(hours=24)
|
|
}
|
|
return jwt.encode(payload, JWT_SECRET, algorithm='HS256')
|
|
|
|
def generate_two_factor_token(user_id: str) -> str:
|
|
"""Create a short-lived token usable only to complete the 2FA step."""
|
|
payload = {
|
|
'userId': user_id,
|
|
'token_type': 'two_factor',
|
|
'exp': datetime.now(timezone.utc) + timedelta(minutes=5),
|
|
}
|
|
return jwt.encode(payload, JWT_SECRET, algorithm='HS256')
|
|
|
|
def verify_token(token: str) -> Optional[Dict]:
|
|
try:
|
|
return jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
|
|
except jwt.InvalidTokenError:
|
|
return None
|
|
|
|
|
|
def generate_oauth_state(provider: str, nonce: Optional[str] = None) -> str:
|
|
"""Sign short-lived OAuth state so callbacks can reject forged requests."""
|
|
payload = {
|
|
'purpose': 'oauth_state',
|
|
'provider': provider,
|
|
'exp': datetime.now(timezone.utc) + timedelta(minutes=10),
|
|
}
|
|
if nonce:
|
|
payload['nonce'] = nonce
|
|
return jwt.encode(payload, JWT_SECRET, algorithm='HS256')
|
|
|
|
|
|
def verify_oauth_state(state: Optional[str], provider: str) -> Optional[Dict]:
|
|
"""Validate a signed OAuth state value for the expected provider."""
|
|
if not state:
|
|
return None
|
|
try:
|
|
payload = jwt.decode(state, JWT_SECRET, algorithms=['HS256'])
|
|
except jwt.InvalidTokenError:
|
|
return None
|
|
if payload.get('purpose') != 'oauth_state' or payload.get('provider') != provider:
|
|
return None
|
|
return payload
|
|
|
|
def hash_password(password: str) -> str:
|
|
return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt(SALT_ROUNDS)).decode('utf-8')
|
|
|
|
def verify_password(password: str, hashed: str) -> bool:
|
|
return bcrypt.checkpw(password.encode('utf-8'), hashed.encode('utf-8'))
|
|
|
|
def authenticate_user(token: Optional[str]) -> Optional[Dict]:
|
|
if not token:
|
|
return None
|
|
payload = verify_token(token)
|
|
if not payload or payload.get('token_type') != 'access':
|
|
return None
|
|
user_id = payload.get('userId')
|
|
if not user_id:
|
|
return None
|
|
|
|
try:
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
SELECT id, username, email, role, credits, created_at, email_verified, two_factor_enabled
|
|
FROM users WHERE id = ?
|
|
''', (user_id,))
|
|
return dict_from_row(cursor.fetchone())
|
|
except Exception as e:
|
|
logger.error(f"Error in authenticate_user: {e}")
|
|
return None
|
|
|
|
def get_bearer_token() -> Optional[str]:
|
|
"""Return a well-formed bearer token from the current HTTP request."""
|
|
authorization = request.headers.get('Authorization', '')
|
|
scheme, separator, token = authorization.partition(' ')
|
|
if not separator or scheme.casefold() != 'bearer':
|
|
return None
|
|
token = token.strip()
|
|
return token or None
|
|
|
|
def require_admin() -> Optional[Dict]:
|
|
"""Return the authenticated admin user, or ``None`` for other callers."""
|
|
user = authenticate_user(get_bearer_token())
|
|
if user and user.get('role') == 'admin':
|
|
return user
|
|
return None
|
|
|
|
def add_credits(user_id: str, amount: int) -> bool:
|
|
"""Add credits to user account"""
|
|
try:
|
|
user = get_user_by_id(user_id)
|
|
if not user:
|
|
return False
|
|
|
|
new_credits = user.get('credits', 0) + amount
|
|
if update_user_credits(user_id, new_credits):
|
|
add_credit_transaction(user_id, amount, 'add', 'Credit top-up')
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"Error adding credits: {e}")
|
|
return False
|
|
|
|
def get_user_credits(user_id: str) -> int:
|
|
"""Get user's current credit balance"""
|
|
user = get_user_by_id(user_id)
|
|
return user.get('credits', 0) if user else 0
|