mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-03 02:26:43 +09:00
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>
49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
from flask import Blueprint, jsonify, current_app
|
|
from flask_jwt_extended import jwt_required
|
|
|
|
from app.services.migration_service import MigrationService
|
|
from app.middleware.rbac import require_admin_user
|
|
|
|
migrations_bp = Blueprint('migrations', __name__)
|
|
|
|
|
|
@migrations_bp.route('/status', methods=['GET'])
|
|
def get_migration_status():
|
|
"""Check if migrations are pending. No auth required (called before login)."""
|
|
status = MigrationService.get_status()
|
|
return jsonify(status), 200
|
|
|
|
|
|
@migrations_bp.route('/backup', methods=['POST'])
|
|
@jwt_required()
|
|
def create_backup():
|
|
"""Create a database backup before applying migrations. Admin only."""
|
|
require_admin_user()
|
|
|
|
result = MigrationService.create_backup(current_app)
|
|
if result['success']:
|
|
return jsonify(result), 200
|
|
return jsonify(result), 500
|
|
|
|
|
|
@migrations_bp.route('/apply', methods=['POST'])
|
|
@jwt_required()
|
|
def apply_migrations():
|
|
"""Apply all pending migrations. Admin only."""
|
|
require_admin_user()
|
|
|
|
result = MigrationService.apply_migrations(current_app)
|
|
if result['success']:
|
|
return jsonify(result), 200
|
|
return jsonify(result), 500
|
|
|
|
|
|
@migrations_bp.route('/history', methods=['GET'])
|
|
@jwt_required()
|
|
def get_migration_history():
|
|
"""Return all migration revisions. Admin only."""
|
|
require_admin_user()
|
|
|
|
history = MigrationService.get_migration_history(current_app)
|
|
return jsonify({'revisions': history}), 200
|