"""Account API routes for AnchorHost.""" from database_service import database_connection, dict_from_row from datetime import datetime, timedelta from email_service import get_email_service from flask import Blueprint, jsonify, request from runtime_config import ( logger, ) from server_lifecycle import delete_server from user_service import authenticate_user, get_bearer_token, get_user_by_username, hash_password, verify_password import uuid __all__ = ['change_user_email', 'change_user_password', 'delete_user_account', 'export_user_data', 'get_user_profile', 'resend_verification', 'update_user_profile'] bp = Blueprint("account", __name__) @bp.route('/api/user/profile', methods=['GET']) def get_user_profile(): """Get user profile data""" try: token = get_bearer_token() user = authenticate_user(token) if not user: return jsonify({'success': False, 'error': 'Unauthorized'}), 401 with database_connection() as conn: cursor = conn.cursor() # Get full user data including profile fields cursor.execute(''' SELECT id, username, email, role, credits, created_at, updated_at, display_name, first_name, last_name, address, city, country, postal_code, phone, email_verified, email_verification_token, password_changed_at, two_factor_enabled, last_login, login_count, google_id, apple_id FROM users WHERE id = ? ''', (user['id'],)) user_data = dict_from_row(cursor.fetchone()) if user_data: # Remove sensitive fields if 'email_verification_token' in user_data: user_data.pop('email_verification_token') if 'two_factor_secret' in user_data: user_data.pop('two_factor_secret', None) return jsonify({ 'success': True, 'user': user_data }), 200 else: return jsonify({'success': False, 'error': 'User not found'}), 404 except Exception as e: logger.error(f"Error getting user profile: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @bp.route('/api/user/profile', methods=['PUT']) def update_user_profile(): """Update user profile""" try: token = get_bearer_token() user = authenticate_user(token) if not user: return jsonify({'success': False, 'error': 'Unauthorized'}), 401 data = request.get_json() with database_connection() as conn: cursor = conn.cursor() # Build update query dynamically based on provided fields update_fields = [] params = [] allowed_fields = ['display_name', 'first_name', 'last_name', 'address', 'city', 'country', 'postal_code', 'phone'] for field in allowed_fields: if field in data: update_fields.append(f"{field} = ?") params.append(data[field]) if not update_fields: return jsonify({'success': False, 'error': 'No fields to update'}), 400 # Add updated_at update_fields.append("updated_at = CURRENT_TIMESTAMP") params.append(user['id']) query = f"UPDATE users SET {', '.join(update_fields)} WHERE id = ?" cursor.execute(query, params) conn.commit() logger.info(f"✓ Profile updated for user: {user['username']}") return jsonify({'success': True, 'message': 'Profile updated successfully'}), 200 except Exception as e: logger.error(f"Error updating profile: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @bp.route('/api/user/email', methods=['PUT']) def change_user_email(): """Change user email address""" try: token = get_bearer_token() user = authenticate_user(token) if not user: return jsonify({'success': False, 'error': 'Unauthorized'}), 401 data = request.get_json() new_email = data.get('email') if not new_email: return jsonify({'success': False, 'error': 'Email required'}), 400 # Validate email format import re if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', new_email): return jsonify({'success': False, 'error': 'Invalid email format'}), 400 with database_connection() as conn: cursor = conn.cursor() # Check if email is already in use cursor.execute('SELECT id FROM users WHERE email = ? AND id != ?', (new_email, user['id'])) if cursor.fetchone(): return jsonify({'success': False, 'error': 'Email already in use'}), 400 # Generate new verification token verification_token = str(uuid.uuid4()) # Update user email and set as unverified cursor.execute(''' UPDATE users SET email = ?, email_verified = FALSE, email_verification_token = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? ''', (new_email, verification_token, user['id'])) # Create verification token record 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'], verification_token, expires_at)) conn.commit() # Send verification email to new address email_svc = get_email_service() if email_svc: email_svc.send_verification_email(new_email, user['username'], verification_token) logger.info(f"✓ Email changed for user: {user['username']} -> {new_email}") return jsonify({ 'success': True, 'message': 'Email changed. Please check your new email for verification link.' }), 200 except Exception as e: logger.error(f"Error changing email: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @bp.route('/api/user/password', methods=['PUT']) def change_user_password(): """Change user password""" try: token = get_bearer_token() user = authenticate_user(token) if not user: return jsonify({'success': False, 'error': 'Unauthorized'}), 401 data = request.get_json() current_password = data.get('currentPassword') new_password = data.get('newPassword') if not current_password or not new_password: return jsonify({'success': False, 'error': 'Current and new passwords required'}), 400 # Validate new password length if len(new_password) < 6: return jsonify({'success': False, 'error': 'Password must be at least 6 characters'}), 400 # Verify current password user_full = get_user_by_username(user['username']) if not verify_password(current_password, user_full['password_hash']): return jsonify({'success': False, 'error': 'Current password is incorrect'}), 401 # Hash new password new_password_hash = hash_password(new_password) with database_connection() as conn: cursor = conn.cursor() cursor.execute(''' UPDATE users SET password_hash = ?, password_changed_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ? ''', (new_password_hash, user['id'])) conn.commit() logger.info(f"✓ Password changed for user: {user['username']}") return jsonify({'success': True, 'message': 'Password changed successfully'}), 200 except Exception as e: logger.error(f"Error changing password: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @bp.route('/api/user/export-data', methods=['GET']) def export_user_data(): """Export all user data""" try: token = get_bearer_token() user = authenticate_user(token) if not user: return jsonify({'success': False, 'error': 'Unauthorized'}), 401 with database_connection() as conn: cursor = conn.cursor() # Get user data cursor.execute('SELECT * FROM users WHERE id = ?', (user['id'],)) user_data = dict_from_row(cursor.fetchone()) # Remove sensitive fields if user_data: user_data.pop('password_hash', None) user_data.pop('email_verification_token', None) user_data.pop('two_factor_secret', None) # Get servers cursor.execute('SELECT * FROM servers WHERE owner_id = ?', (user['id'],)) servers = [dict_from_row(row) for row in cursor.fetchall()] # Get domains cursor.execute('SELECT * FROM user_domains WHERE user_id = ?', (user['id'],)) domains = [dict_from_row(row) for row in cursor.fetchall()] # Get tickets cursor.execute('SELECT * FROM tickets WHERE user_id = ?', (user['id'],)) tickets = [dict_from_row(row) for row in cursor.fetchall()] # Get credit transactions cursor.execute('SELECT * FROM credit_transactions WHERE user_id = ?', (user['id'],)) transactions = [dict_from_row(row) for row in cursor.fetchall()] export_data = { 'user': user_data, 'servers': servers, 'domains': domains, 'tickets': tickets, 'credit_transactions': transactions, 'exported_at': datetime.now().isoformat() } logger.info(f"✓ Data exported for user: {user['username']}") return jsonify({'success': True, 'data': export_data}), 200 except Exception as e: logger.error(f"Error exporting data: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @bp.route('/api/user/account', methods=['DELETE']) def delete_user_account(): """Delete user account and all associated data""" try: token = get_bearer_token() user = authenticate_user(token) if not user: return jsonify({'success': False, 'error': 'Unauthorized'}), 401 data = request.get_json() confirmation = data.get('confirmation') if confirmation != 'DELETE MY ACCOUNT': return jsonify({'success': False, 'error': 'Invalid confirmation text'}), 400 with database_connection() as conn: cursor = conn.cursor() # Delete all user's servers (cascade will handle related data) cursor.execute('SELECT id FROM servers WHERE owner_id = ?', (user['id'],)) server_ids = [row['id'] for row in cursor.fetchall()] # Delete servers from Kubernetes for server_id in server_ids: try: delete_server(server_id, user['id']) except Exception as e: logger.error(f"Error deleting server {server_id}: {e}") # Delete user (cascade will handle most related data) cursor.execute('DELETE FROM users WHERE id = ?', (user['id'],)) conn.commit() logger.info(f"✓ Account deleted for user: {user['username']}") return jsonify({'success': True, 'message': 'Account deleted successfully'}), 200 except Exception as e: logger.error(f"Error deleting account: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @bp.route('/api/user/resend-verification', methods=['POST']) def resend_verification(): """Resend verification email""" try: token = get_bearer_token() user = authenticate_user(token) if not user: return jsonify({'success': False, 'error': 'Unauthorized'}), 401 # Check if already verified if user.get('email_verified'): return jsonify({'success': False, 'error': 'Email already verified'}), 400 # Generate new token new_token = str(uuid.uuid4()) with database_connection() as conn: cursor = conn.cursor() # Delete old tokens cursor.execute('DELETE FROM email_verification_tokens WHERE user_id = ?', (user['id'],)) # Create new 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'], new_token, expires_at)) # Update user cursor.execute(''' UPDATE users SET email_verification_token = ? WHERE id = ? ''', (new_token, user['id'])) conn.commit() # Send email email_svc = get_email_service() if email_svc: success = email_svc.send_verification_email(user['email'], user['username'], new_token) if success: return jsonify({'success': True, 'message': 'Verification email sent'}), 200 else: return jsonify({'success': False, 'error': 'Failed to send email'}), 500 else: return jsonify({'success': False, 'error': 'Email service unavailable'}), 503 except Exception as e: logger.error(f"Error resending verification: {e}") return jsonify({'success': False, 'error': str(e)}), 500