mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-10-30 00:27:11 +09:00
Juan Denis
55d8fb1e44
328 more routes used a JSON body as a dict straight from
request.get_json(), so a body of null, [] or a bare string crashed them
into a 500. Bare reads become json_object(); the
'get_json(silent=True) or {}' reads of action routes become
json_object(optional=True), which still treats a missing body as {}.
Routes that take arrays or pass the body through untouched are left
alone. The full backend suite shows no new failures.
Also carries three files' bare section citations (apps, backups,
notifications) and the extended citation guard, which needed them.
450 lines
14 KiB
Python
450 lines
14 KiB
Python
# Bucket: PER-APP. Per-app Python runtime routes gate on the shared
|
|
# app-access seam (can_access_app for reads, can_edit_app for mutations);
|
|
# host-level interpreter management stays admin-only.
|
|
from flask import Blueprint, jsonify
|
|
from flask_jwt_extended import jwt_required, get_jwt_identity
|
|
from app.middleware.rbac import admin_required
|
|
from app.models import User, Application
|
|
from app.services.python_service import PythonService
|
|
from app import db
|
|
from app.exceptions import not_found, permission_denied
|
|
from app.api._request import json_object
|
|
|
|
python_bp = Blueprint('python', __name__)
|
|
|
|
|
|
def get_app_or_404(app_id, current_user_id):
|
|
"""Get application and verify read access. Read endpoints rely on this gate;
|
|
mutating endpoints additionally carry @admin_required, so honoring per-resource
|
|
grants here (#33) safely opens only the read GETs to grantees. Uses can_access_app
|
|
(user.id), which also fixes the int-vs-str identity comparison."""
|
|
from app.services.resource_grant_service import ResourceGrantService
|
|
user = User.query.get(current_user_id)
|
|
app = Application.query_active().filter_by(id=app_id).first()
|
|
|
|
if not app:
|
|
raise not_found('service')
|
|
|
|
if not ResourceGrantService.can_access_app(user, app):
|
|
raise permission_denied()
|
|
|
|
if app.app_type not in ['flask', 'django']:
|
|
return None, ({'error': 'This service is not a Python service'}, 400)
|
|
|
|
return app, None
|
|
|
|
|
|
# Python version management
|
|
@python_bp.route('/versions', methods=['GET'])
|
|
@jwt_required()
|
|
def get_versions():
|
|
"""Get available Python versions."""
|
|
versions = PythonService.get_python_versions()
|
|
default = PythonService.get_default_python()
|
|
return jsonify({
|
|
'versions': versions,
|
|
'default': default
|
|
}), 200
|
|
|
|
|
|
# Application creation
|
|
@python_bp.route('/apps/flask', methods=['POST'])
|
|
@admin_required
|
|
def create_flask_app():
|
|
"""Create a new Flask application."""
|
|
data = json_object()
|
|
|
|
if not data:
|
|
return jsonify({'error': 'No data provided'}), 400
|
|
|
|
required_fields = ['name', 'path']
|
|
missing = [f for f in required_fields if f not in data]
|
|
if missing:
|
|
return jsonify({'error': f'Missing required fields: {", ".join(missing)}'}), 400
|
|
|
|
python_version = data.get('python_version', '3.11')
|
|
port = data.get('port', 8000)
|
|
|
|
# Create Flask app
|
|
result = PythonService.create_flask_app(
|
|
data['path'],
|
|
data['name'],
|
|
python_version
|
|
)
|
|
|
|
if not result['success']:
|
|
return jsonify(result), 400
|
|
|
|
# Create Gunicorn service
|
|
service_result = PythonService.create_gunicorn_service(
|
|
data['name'],
|
|
data['path'],
|
|
port=port,
|
|
workers=data.get('workers', 2)
|
|
)
|
|
|
|
if not service_result['success']:
|
|
return jsonify(service_result), 400
|
|
|
|
# Create application record
|
|
current_user_id = get_jwt_identity()
|
|
app = Application(
|
|
name=data['name'],
|
|
app_type='flask',
|
|
status='stopped',
|
|
python_version=python_version,
|
|
port=port,
|
|
root_path=data['path'],
|
|
user_id=current_user_id
|
|
)
|
|
db.session.add(app)
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'app_id': app.id,
|
|
'message': 'Flask service created'
|
|
}), 201
|
|
|
|
|
|
@python_bp.route('/apps/django', methods=['POST'])
|
|
@admin_required
|
|
def create_django_app():
|
|
"""Create a new Django application."""
|
|
data = json_object()
|
|
|
|
if not data:
|
|
return jsonify({'error': 'No data provided'}), 400
|
|
|
|
required_fields = ['name', 'path']
|
|
missing = [f for f in required_fields if f not in data]
|
|
if missing:
|
|
return jsonify({'error': f'Missing required fields: {", ".join(missing)}'}), 400
|
|
|
|
python_version = data.get('python_version', '3.11')
|
|
port = data.get('port', 8000)
|
|
|
|
# Create Django app
|
|
result = PythonService.create_django_app(
|
|
data['path'],
|
|
data['name'],
|
|
python_version
|
|
)
|
|
|
|
if not result['success']:
|
|
return jsonify(result), 400
|
|
|
|
# Create Gunicorn service
|
|
service_result = PythonService.create_gunicorn_service(
|
|
data['name'],
|
|
data['path'],
|
|
port=port,
|
|
workers=data.get('workers', 2)
|
|
)
|
|
|
|
if not service_result['success']:
|
|
return jsonify(service_result), 400
|
|
|
|
# Create application record
|
|
current_user_id = get_jwt_identity()
|
|
app = Application(
|
|
name=data['name'],
|
|
app_type='django',
|
|
status='stopped',
|
|
python_version=python_version,
|
|
port=port,
|
|
root_path=data['path'],
|
|
user_id=current_user_id
|
|
)
|
|
db.session.add(app)
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'app_id': app.id,
|
|
'project_name': result.get('project_name'),
|
|
'message': 'Django service created'
|
|
}), 201
|
|
|
|
|
|
# Virtual environment management
|
|
@python_bp.route('/apps/<int:app_id>/venv', methods=['POST'])
|
|
@admin_required
|
|
def create_venv(app_id):
|
|
"""Create or recreate virtual environment."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
result = PythonService.create_virtualenv(app.root_path, app.python_version)
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@python_bp.route('/apps/<int:app_id>/packages', methods=['GET'])
|
|
@jwt_required()
|
|
def get_packages(app_id):
|
|
"""Get installed packages."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
packages = PythonService.get_installed_packages(app.root_path)
|
|
return jsonify({'packages': packages}), 200
|
|
|
|
|
|
@python_bp.route('/apps/<int:app_id>/packages', methods=['POST'])
|
|
@admin_required
|
|
def install_packages(app_id):
|
|
"""Install packages."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
data = json_object()
|
|
packages = data.get('packages') if data else None
|
|
|
|
result = PythonService.install_requirements(app.root_path, packages)
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@python_bp.route('/apps/<int:app_id>/requirements', methods=['POST'])
|
|
@admin_required
|
|
def freeze_requirements(app_id):
|
|
"""Freeze requirements to requirements.txt."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
result = PythonService.freeze_requirements(app.root_path)
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
# Environment variables
|
|
@python_bp.route('/apps/<int:app_id>/env', methods=['GET'])
|
|
@jwt_required()
|
|
def get_env_vars(app_id):
|
|
"""Get environment variables."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
env_vars = PythonService.get_env_vars(app.root_path)
|
|
|
|
# Mask sensitive values
|
|
masked_vars = {}
|
|
sensitive_keys = ['SECRET', 'PASSWORD', 'KEY', 'TOKEN', 'PRIVATE']
|
|
for key, value in env_vars.items():
|
|
if any(s in key.upper() for s in sensitive_keys):
|
|
masked_vars[key] = '********'
|
|
else:
|
|
masked_vars[key] = value
|
|
|
|
return jsonify({'env_vars': masked_vars, 'count': len(env_vars)}), 200
|
|
|
|
|
|
@python_bp.route('/apps/<int:app_id>/env', methods=['PUT'])
|
|
@admin_required
|
|
def set_env_vars(app_id):
|
|
"""Set environment variables."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
data = json_object()
|
|
if not data or 'env_vars' not in data:
|
|
return jsonify({'error': 'env_vars is required'}), 400
|
|
|
|
result = PythonService.set_env_vars(app.root_path, data['env_vars'])
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@python_bp.route('/apps/<int:app_id>/env/<key>', methods=['DELETE'])
|
|
@admin_required
|
|
def delete_env_var(app_id, key):
|
|
"""Delete an environment variable."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
result = PythonService.delete_env_var(app.root_path, key)
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
# Process control
|
|
@python_bp.route('/apps/<int:app_id>/start', methods=['POST'])
|
|
@admin_required
|
|
def start_app(app_id):
|
|
"""Start the application."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
result = PythonService.start_app(app.name)
|
|
if result['success']:
|
|
app.status = 'running'
|
|
db.session.commit()
|
|
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@python_bp.route('/apps/<int:app_id>/stop', methods=['POST'])
|
|
@admin_required
|
|
def stop_app(app_id):
|
|
"""Stop the application."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
result = PythonService.stop_app(app.name)
|
|
if result['success']:
|
|
app.status = 'stopped'
|
|
db.session.commit()
|
|
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@python_bp.route('/apps/<int:app_id>/restart', methods=['POST'])
|
|
@admin_required
|
|
def restart_app(app_id):
|
|
"""Restart the application."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
result = PythonService.restart_app(app.name)
|
|
if result['success']:
|
|
app.status = 'running'
|
|
db.session.commit()
|
|
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@python_bp.route('/apps/<int:app_id>/status', methods=['GET'])
|
|
@jwt_required()
|
|
def get_status(app_id):
|
|
"""Get application status."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
status = PythonService.get_app_status(app.name)
|
|
return jsonify(status), 200
|
|
|
|
|
|
# Gunicorn configuration
|
|
@python_bp.route('/apps/<int:app_id>/gunicorn', methods=['GET'])
|
|
@jwt_required()
|
|
def get_gunicorn_config(app_id):
|
|
"""Get Gunicorn configuration."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
config = PythonService.get_gunicorn_config(app.root_path)
|
|
return jsonify(config), 200
|
|
|
|
|
|
@python_bp.route('/apps/<int:app_id>/gunicorn', methods=['PUT'])
|
|
@admin_required
|
|
def update_gunicorn_config(app_id):
|
|
"""Update Gunicorn configuration."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
data = json_object()
|
|
if not data or 'content' not in data:
|
|
return jsonify({'error': 'content is required'}), 400
|
|
|
|
result = PythonService.save_gunicorn_config(app.root_path, data['content'])
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
# Django-specific
|
|
@python_bp.route('/apps/<int:app_id>/migrate', methods=['POST'])
|
|
@admin_required
|
|
def run_migrations(app_id):
|
|
"""Run database migrations."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
result = PythonService.run_migrations(app.root_path, app.app_type)
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@python_bp.route('/apps/<int:app_id>/collectstatic', methods=['POST'])
|
|
@admin_required
|
|
def collect_static(app_id):
|
|
"""Collect static files (Django only)."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
if app.app_type != 'django':
|
|
return jsonify({'error': 'This only works for Django services'}), 400
|
|
|
|
result = PythonService.collect_static(app.root_path, app.app_type)
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
# Command execution
|
|
@python_bp.route('/apps/<int:app_id>/run', methods=['POST'])
|
|
@admin_required
|
|
def run_command(app_id):
|
|
"""Run a command in the app context."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
data = json_object()
|
|
if not data or 'command' not in data:
|
|
return jsonify({'error': 'command is required'}), 400
|
|
|
|
# Basic command validation - block dangerous commands
|
|
command = data['command']
|
|
blocked = ['rm -rf /', 'mkfs', 'dd if=', ':(){:|:&};:']
|
|
if any(b in command for b in blocked):
|
|
return jsonify({'error': 'Command not allowed'}), 403
|
|
|
|
result = PythonService.run_command(app.root_path, command)
|
|
return jsonify(result), 200
|
|
|
|
|
|
# Delete application
|
|
@python_bp.route('/apps/<int:app_id>', methods=['DELETE'])
|
|
@admin_required
|
|
def delete_app(app_id):
|
|
"""Delete a Python application."""
|
|
current_user_id = get_jwt_identity()
|
|
app, error = get_app_or_404(app_id, current_user_id)
|
|
if error:
|
|
return jsonify(error[0]), error[1]
|
|
|
|
data = json_object()
|
|
remove_files = data.get('remove_files', False)
|
|
|
|
result = PythonService.delete_app(app.name, app.root_path, remove_files)
|
|
|
|
if result['success']:
|
|
db.session.delete(app)
|
|
db.session.commit()
|
|
|
|
return jsonify(result), 200 if result['success'] else 400
|