Files
ServerKit/backend/app/api/wordpress.py
T
Juan Denis 06d63cbc52 Add standalone WordPress and SSL management UI
Implements backend endpoints and service logic for managing standalone WordPress via Docker, including install, uninstall, start, stop, and restart. Adds Nginx integration for WordPress at /wordpress. Introduces a new frontend page for SSL certificate management with Certbot install, obtain, renew, and revoke actions. Updates the WordPress frontend page to support one-click install, status, and lifecycle management. Updates documentation and install instructions for improved clarity.
2026-01-25 22:48:12 -05:00

529 lines
16 KiB
Python

from flask import Blueprint, request, jsonify
from flask_jwt_extended import jwt_required, get_jwt_identity
from app.models import User, Application
from app.services.wordpress_service import WordPressService
from app import db
wordpress_bp = Blueprint('wordpress', __name__)
# ==================== STANDALONE WORDPRESS (DOCKER) ENDPOINTS ====================
@wordpress_bp.route('/standalone/status', methods=['GET'])
@jwt_required()
def get_standalone_status():
"""Get WordPress standalone installation status."""
result = WordPressService.get_wordpress_standalone_status()
return jsonify(result), 200
@wordpress_bp.route('/standalone/requirements', methods=['GET'])
@jwt_required()
def get_standalone_requirements():
"""Get resource requirements for WordPress installation."""
result = WordPressService.get_wordpress_resource_requirements()
return jsonify(result), 200
@wordpress_bp.route('/standalone/install', methods=['POST'])
@jwt_required()
def install_standalone():
"""Install WordPress via Docker Compose."""
data = request.get_json() or {}
result = WordPressService.install_wordpress_standalone(
admin_email=data.get('adminEmail')
)
if result.get('success'):
return jsonify(result), 201
return jsonify(result), 400
@wordpress_bp.route('/standalone/uninstall', methods=['POST'])
@jwt_required()
def uninstall_standalone():
"""Uninstall standalone WordPress and optionally remove data."""
data = request.get_json() or {}
result = WordPressService.uninstall_wordpress_standalone(
remove_data=data.get('removeData', False)
)
if result.get('success'):
return jsonify(result), 200
return jsonify(result), 400
@wordpress_bp.route('/standalone/start', methods=['POST'])
@jwt_required()
def start_standalone():
"""Start WordPress containers."""
result = WordPressService.start_wordpress_standalone()
if result.get('success'):
return jsonify(result), 200
return jsonify(result), 400
@wordpress_bp.route('/standalone/stop', methods=['POST'])
@jwt_required()
def stop_standalone():
"""Stop WordPress containers."""
result = WordPressService.stop_wordpress_standalone()
if result.get('success'):
return jsonify(result), 200
return jsonify(result), 400
@wordpress_bp.route('/standalone/restart', methods=['POST'])
@jwt_required()
def restart_standalone():
"""Restart WordPress containers."""
result = WordPressService.restart_wordpress_standalone()
if result.get('success'):
return jsonify(result), 200
return jsonify(result), 400
# ==================== LEGACY WP-CLI ENDPOINTS ====================
def admin_required(fn):
"""Decorator to require admin role."""
from functools import wraps
@wraps(fn)
def wrapper(*args, **kwargs):
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
if not user or user.role != 'admin':
return jsonify({'error': 'Admin access required'}), 403
return fn(*args, **kwargs)
return wrapper
@wordpress_bp.route('/install', methods=['POST'])
@jwt_required()
@admin_required
def install_wordpress():
"""Install a new WordPress site."""
data = request.get_json()
if not data:
return jsonify({'error': 'No data provided'}), 400
required_fields = ['path', 'site_url', 'admin_email', 'db_name', 'db_user', 'db_password']
missing = [f for f in required_fields if f not in data]
if missing:
return jsonify({'error': f'Missing required fields: {", ".join(missing)}'}), 400
result = WordPressService.install_wordpress(data['path'], data)
if result['success']:
# Create application record
current_user_id = get_jwt_identity()
app = Application(
name=data.get('site_title', 'WordPress Site'),
app_type='wordpress',
status='running',
php_version=data.get('php_version', '8.2'),
root_path=data['path'],
user_id=current_user_id
)
db.session.add(app)
db.session.commit()
result['app_id'] = app.id
return jsonify(result), 201 if result['success'] else 400
@wordpress_bp.route('/sites/<int:app_id>/info', methods=['GET'])
@jwt_required()
def get_wordpress_info(app_id):
"""Get WordPress site info."""
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
app = Application.query.get(app_id)
if not app:
return jsonify({'error': 'Application not found'}), 404
if user.role != 'admin' and app.user_id != current_user_id:
return jsonify({'error': 'Access denied'}), 403
if app.app_type != 'wordpress':
return jsonify({'error': 'Application is not a WordPress site'}), 400
info = WordPressService.get_wordpress_info(app.root_path)
if not info:
return jsonify({'error': 'Could not get WordPress info'}), 400
return jsonify({'info': info}), 200
@wordpress_bp.route('/sites/<int:app_id>/update', methods=['POST'])
@jwt_required()
@admin_required
def update_wordpress(app_id):
"""Update WordPress core."""
app = Application.query.get(app_id)
if not app:
return jsonify({'error': 'Application not found'}), 404
if app.app_type != 'wordpress':
return jsonify({'error': 'Application is not a WordPress site'}), 400
result = WordPressService.update_wordpress(app.root_path)
return jsonify(result), 200 if result['success'] else 400
# Plugin management
@wordpress_bp.route('/sites/<int:app_id>/plugins', methods=['GET'])
@jwt_required()
def get_plugins(app_id):
"""Get installed plugins."""
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
app = Application.query.get(app_id)
if not app:
return jsonify({'error': 'Application not found'}), 404
if user.role != 'admin' and app.user_id != current_user_id:
return jsonify({'error': 'Access denied'}), 403
plugins = WordPressService.get_plugins(app.root_path)
return jsonify({'plugins': plugins}), 200
@wordpress_bp.route('/sites/<int:app_id>/plugins', methods=['POST'])
@jwt_required()
@admin_required
def install_plugin(app_id):
"""Install a plugin."""
app = Application.query.get(app_id)
data = request.get_json()
if not app:
return jsonify({'error': 'Application not found'}), 404
if not data or 'plugin' not in data:
return jsonify({'error': 'plugin is required'}), 400
result = WordPressService.install_plugin(
app.root_path,
data['plugin'],
data.get('activate', True)
)
return jsonify(result), 200 if result['success'] else 400
@wordpress_bp.route('/sites/<int:app_id>/plugins/<plugin>', methods=['DELETE'])
@jwt_required()
@admin_required
def uninstall_plugin(app_id, plugin):
"""Uninstall a plugin."""
app = Application.query.get(app_id)
if not app:
return jsonify({'error': 'Application not found'}), 404
result = WordPressService.uninstall_plugin(app.root_path, plugin)
return jsonify(result), 200 if result['success'] else 400
@wordpress_bp.route('/sites/<int:app_id>/plugins/<plugin>/activate', methods=['POST'])
@jwt_required()
@admin_required
def activate_plugin(app_id, plugin):
"""Activate a plugin."""
app = Application.query.get(app_id)
if not app:
return jsonify({'error': 'Application not found'}), 404
result = WordPressService.activate_plugin(app.root_path, plugin)
return jsonify(result), 200 if result['success'] else 400
@wordpress_bp.route('/sites/<int:app_id>/plugins/<plugin>/deactivate', methods=['POST'])
@jwt_required()
@admin_required
def deactivate_plugin(app_id, plugin):
"""Deactivate a plugin."""
app = Application.query.get(app_id)
if not app:
return jsonify({'error': 'Application not found'}), 404
result = WordPressService.deactivate_plugin(app.root_path, plugin)
return jsonify(result), 200 if result['success'] else 400
@wordpress_bp.route('/sites/<int:app_id>/plugins/update', methods=['POST'])
@jwt_required()
@admin_required
def update_plugins(app_id):
"""Update plugins."""
app = Application.query.get(app_id)
data = request.get_json()
if not app:
return jsonify({'error': 'Application not found'}), 404
plugins = data.get('plugins') if data else None
result = WordPressService.update_plugins(app.root_path, plugins)
return jsonify(result), 200 if result['success'] else 400
# Theme management
@wordpress_bp.route('/sites/<int:app_id>/themes', methods=['GET'])
@jwt_required()
def get_themes(app_id):
"""Get installed themes."""
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
app = Application.query.get(app_id)
if not app:
return jsonify({'error': 'Application not found'}), 404
if user.role != 'admin' and app.user_id != current_user_id:
return jsonify({'error': 'Access denied'}), 403
themes = WordPressService.get_themes(app.root_path)
return jsonify({'themes': themes}), 200
@wordpress_bp.route('/sites/<int:app_id>/themes', methods=['POST'])
@jwt_required()
@admin_required
def install_theme(app_id):
"""Install a theme."""
app = Application.query.get(app_id)
data = request.get_json()
if not app:
return jsonify({'error': 'Application not found'}), 404
if not data or 'theme' not in data:
return jsonify({'error': 'theme is required'}), 400
result = WordPressService.install_theme(
app.root_path,
data['theme'],
data.get('activate', False)
)
return jsonify(result), 200 if result['success'] else 400
@wordpress_bp.route('/sites/<int:app_id>/themes/<theme>/activate', methods=['POST'])
@jwt_required()
@admin_required
def activate_theme(app_id, theme):
"""Activate a theme."""
app = Application.query.get(app_id)
if not app:
return jsonify({'error': 'Application not found'}), 404
result = WordPressService.activate_theme(app.root_path, theme)
return jsonify(result), 200 if result['success'] else 400
# Backup management
@wordpress_bp.route('/sites/<int:app_id>/backup', methods=['POST'])
@jwt_required()
@admin_required
def create_backup(app_id):
"""Create a backup."""
app = Application.query.get(app_id)
data = request.get_json() or {}
if not app:
return jsonify({'error': 'Application not found'}), 404
result = WordPressService.backup_wordpress(
app.root_path,
data.get('include_db', True)
)
return jsonify(result), 200 if result['success'] else 400
@wordpress_bp.route('/sites/<int:app_id>/backups', methods=['GET'])
@jwt_required()
def list_backups(app_id):
"""List backups for a site."""
current_user_id = get_jwt_identity()
user = User.query.get(current_user_id)
app = Application.query.get(app_id)
if not app:
return jsonify({'error': 'Application not found'}), 404
if user.role != 'admin' and app.user_id != current_user_id:
return jsonify({'error': 'Access denied'}), 403
import os
site_name = os.path.basename(app.root_path)
backups = WordPressService.list_backups(site_name)
return jsonify({'backups': backups}), 200
@wordpress_bp.route('/sites/<int:app_id>/restore', methods=['POST'])
@jwt_required()
@admin_required
def restore_backup(app_id):
"""Restore a backup."""
app = Application.query.get(app_id)
data = request.get_json()
if not app:
return jsonify({'error': 'Application not found'}), 404
if not data or 'backup_name' not in data:
return jsonify({'error': 'backup_name is required'}), 400
result = WordPressService.restore_backup(data['backup_name'], app.root_path)
return jsonify(result), 200 if result['success'] else 400
@wordpress_bp.route('/backups/<backup_name>', methods=['DELETE'])
@jwt_required()
@admin_required
def delete_backup(backup_name):
"""Delete a backup."""
result = WordPressService.delete_backup(backup_name)
return jsonify(result), 200 if result['success'] else 400
# Security and maintenance
@wordpress_bp.route('/sites/<int:app_id>/harden', methods=['POST'])
@jwt_required()
@admin_required
def harden_wordpress(app_id):
"""Apply security hardening."""
app = Application.query.get(app_id)
if not app:
return jsonify({'error': 'Application not found'}), 404
result = WordPressService.harden_wordpress(app.root_path)
return jsonify(result), 200 if result['success'] else 400
@wordpress_bp.route('/sites/<int:app_id>/search-replace', methods=['POST'])
@jwt_required()
@admin_required
def search_replace(app_id):
"""Search and replace in database."""
app = Application.query.get(app_id)
data = request.get_json()
if not app:
return jsonify({'error': 'Application not found'}), 404
if not data or 'search' not in data or 'replace' not in data:
return jsonify({'error': 'search and replace are required'}), 400
result = WordPressService.search_replace(
app.root_path,
data['search'],
data['replace'],
data.get('dry_run', True)
)
return jsonify(result), 200 if result['success'] else 400
@wordpress_bp.route('/sites/<int:app_id>/optimize', methods=['POST'])
@jwt_required()
@admin_required
def optimize_database(app_id):
"""Optimize WordPress database."""
app = Application.query.get(app_id)
if not app:
return jsonify({'error': 'Application not found'}), 404
result = WordPressService.optimize_database(app.root_path)
return jsonify(result), 200 if result['success'] else 400
@wordpress_bp.route('/sites/<int:app_id>/flush-cache', methods=['POST'])
@jwt_required()
@admin_required
def flush_cache(app_id):
"""Flush WordPress cache."""
app = Application.query.get(app_id)
if not app:
return jsonify({'error': 'Application not found'}), 404
result = WordPressService.flush_cache(app.root_path)
return jsonify(result), 200 if result['success'] else 400
# User management
@wordpress_bp.route('/sites/<int:app_id>/users', methods=['POST'])
@jwt_required()
@admin_required
def create_wp_user(app_id):
"""Create a WordPress user."""
app = Application.query.get(app_id)
data = request.get_json()
if not app:
return jsonify({'error': 'Application not found'}), 404
if not data or 'username' not in data or 'email' not in data:
return jsonify({'error': 'username and email are required'}), 400
result = WordPressService.create_user(
app.root_path,
data['username'],
data['email'],
data.get('role', 'subscriber'),
data.get('password')
)
return jsonify(result), 201 if result['success'] else 400
@wordpress_bp.route('/sites/<int:app_id>/users/<user>/reset-password', methods=['POST'])
@jwt_required()
@admin_required
def reset_wp_password(app_id, user):
"""Reset a WordPress user's password."""
app = Application.query.get(app_id)
data = request.get_json() or {}
if not app:
return jsonify({'error': 'Application not found'}), 404
result = WordPressService.reset_password(app.root_path, user, data.get('password'))
return jsonify(result), 200 if result['success'] else 400
# WP-CLI status
@wordpress_bp.route('/wp-cli/status', methods=['GET'])
@jwt_required()
@admin_required
def wp_cli_status():
"""Check WP-CLI installation status."""
installed = WordPressService.is_wp_cli_installed()
return jsonify({
'installed': installed,
'path': WordPressService.WP_CLI_PATH if installed else None
}), 200
@wordpress_bp.route('/wp-cli/install', methods=['POST'])
@jwt_required()
@admin_required
def install_wp_cli():
"""Install WP-CLI."""
result = WordPressService.install_wp_cli()
return jsonify(result), 200 if result['success'] else 400