mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-09 21:43:05 +09:00
Juan DenisandClaude Opus 5
031142111d
GHSA-r4q7-x795-vr4j was one forgotten check, but the audit behind it found the same shape across the surface: `@jwt_required()` authenticates, it does not authorize, and ServerKit has no default-deny to catch the difference. This closes the routes that audit left open — fleet-monitor alert acknowledge/resolve, recycle-bin restore, nginx test/diff, registrar sync/test, server-template drift checks, uptime tracking on/off, server ping and remote compose ps/logs, refresh-capabilities, template test-db-connection, deploy branches, buildpack detect — plus three inline gaps: a viewer could widen their own API key's scopes, the demo-deploy flag was doing double duty as an authorization one, and tagging a shared resource had no write check at all while attaching one did. `_ensure_group_mutable` only ever blocked *system* queue groups, so every queue_bus mutation now also calls `_ensure_group_accessible` (owner-or-admin) — renaming, draining or deleting another user's group was open to any authenticated caller. The rbac policy decorators authenticate through `auth_required()`, so they replace `@jwt_required()` rather than stack under it; stacking rejects API-key callers before the policy can authorize them (tests/test_auth_decorator_boundaries.py enforces this). Where a route then needed the caller it reads `get_current_user()` — `get_jwt_identity()` raises for an API-key request. The decorators also carry a `_sk_authz` marker so a static test can see the decision from the source. `migrations.get_migration_history` moves to `require_admin_user()` for the same reason its siblings did: the inline `User.ROLE_ADMIN` check it kept was already reading two names this module no longer imports, so the route raised NameError for every caller, admin included. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
from flask import Blueprint, request, jsonify
|
|
from flask_jwt_extended import jwt_required
|
|
from app.services.nginx_advanced_service import NginxAdvancedService
|
|
from app.middleware.rbac import get_current_user
|
|
|
|
nginx_advanced_bp = Blueprint('nginx_advanced', __name__)
|
|
|
|
|
|
@nginx_advanced_bp.route('/proxy/<domain>', methods=['GET'])
|
|
@jwt_required()
|
|
def get_proxy_rules(domain):
|
|
result = NginxAdvancedService.get_proxy_rules(domain)
|
|
if 'error' in result:
|
|
return jsonify(result), 404
|
|
return jsonify(result)
|
|
|
|
|
|
@nginx_advanced_bp.route('/proxy', methods=['POST'])
|
|
@jwt_required()
|
|
def create_proxy():
|
|
user = get_current_user()
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
data = request.get_json()
|
|
if not data or 'domain' not in data:
|
|
return jsonify({'error': 'domain required'}), 400
|
|
|
|
result = NginxAdvancedService.create_reverse_proxy(data)
|
|
return jsonify(result), 201
|
|
|
|
|
|
@nginx_advanced_bp.route('/test', methods=['POST'])
|
|
@jwt_required()
|
|
def test_config():
|
|
user = get_current_user()
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
result = NginxAdvancedService.test_config()
|
|
return jsonify(result)
|
|
|
|
|
|
@nginx_advanced_bp.route('/reload', methods=['POST'])
|
|
@jwt_required()
|
|
def reload():
|
|
user = get_current_user()
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
result = NginxAdvancedService.reload_nginx()
|
|
return jsonify(result)
|
|
|
|
|
|
@nginx_advanced_bp.route('/diff', methods=['POST'])
|
|
@jwt_required()
|
|
def preview_diff():
|
|
# Admin-only: the diff includes the CURRENT vhost file's contents, so this
|
|
# is a file read as much as a preview.
|
|
user = get_current_user()
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
data = request.get_json() or {}
|
|
domain = data.get('domain')
|
|
new_config = data.get('config', '')
|
|
if not domain:
|
|
return jsonify({'error': 'domain required'}), 400
|
|
result = NginxAdvancedService.preview_diff(domain, new_config)
|
|
if 'error' in result:
|
|
return jsonify(result), 400
|
|
return jsonify(result)
|
|
|
|
|
|
@nginx_advanced_bp.route('/logs/<domain>', methods=['GET'])
|
|
@jwt_required()
|
|
def get_logs(domain):
|
|
log_type = request.args.get('type', 'access')
|
|
lines = request.args.get('lines', 100, type=int)
|
|
result = NginxAdvancedService.get_vhost_logs(domain, log_type, lines)
|
|
return jsonify(result)
|
|
|
|
|
|
@nginx_advanced_bp.route('/lb-methods', methods=['GET'])
|
|
@jwt_required()
|
|
def get_lb_methods():
|
|
return jsonify({'methods': NginxAdvancedService.get_load_balancing_methods()})
|