mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-12 15:03:55 +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>
70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
"""Recycle Bin — list, restore and purge soft-deleted records.
|
|
|
|
Type-agnostic on purpose: every route works off a `kind` registered in
|
|
recycle_bin_service, so adding a soft-deletable model needs no change here.
|
|
"""
|
|
from flask import Blueprint, jsonify, request
|
|
from flask_jwt_extended import jwt_required, get_jwt_identity
|
|
|
|
from app.middleware.rbac import admin_required
|
|
from app.services import recycle_bin_service
|
|
|
|
recycle_bin_bp = Blueprint('recycle_bin', __name__)
|
|
|
|
|
|
@recycle_bin_bp.route('', methods=['GET'])
|
|
@jwt_required()
|
|
def list_deleted():
|
|
"""Everything currently in the bin, newest deletion first."""
|
|
kind = request.args.get('kind', '').strip() or None
|
|
if kind and kind not in recycle_bin_service.registered_kinds():
|
|
return jsonify({'error': f'Unknown kind: {kind}'}), 400
|
|
try:
|
|
limit = min(int(request.args.get('limit', 200)), 500)
|
|
except (TypeError, ValueError):
|
|
limit = 200
|
|
return jsonify({
|
|
'items': recycle_bin_service.list_deleted(kind=kind, limit=limit),
|
|
'kinds': recycle_bin_service.registered_kinds(),
|
|
'retention_days': recycle_bin_service.DEFAULT_RETENTION_DAYS,
|
|
})
|
|
|
|
|
|
@recycle_bin_bp.route('/<kind>/<int:record_id>/restore', methods=['POST'])
|
|
@admin_required
|
|
def restore(kind, record_id):
|
|
try:
|
|
item, err = recycle_bin_service.restore(kind, record_id)
|
|
except KeyError:
|
|
return jsonify({'error': f'Unknown kind: {kind}'}), 400
|
|
if item is None:
|
|
return jsonify({'error': err or 'not found'}), 404
|
|
# A non-null err with an item means the row came back but a side effect
|
|
# (re-writing a vhost, say) failed — a warning, not a failure.
|
|
return jsonify({'item': item, 'warning': err}), 200
|
|
|
|
|
|
@recycle_bin_bp.route('/<kind>/<int:record_id>', methods=['DELETE'])
|
|
@admin_required
|
|
def purge(kind, record_id):
|
|
"""Delete for real. Admin-only: this is the one irreversible action."""
|
|
try:
|
|
ok, err = recycle_bin_service.purge(kind, record_id)
|
|
except KeyError:
|
|
return jsonify({'error': f'Unknown kind: {kind}'}), 400
|
|
if not ok:
|
|
return jsonify({'error': err}), 404 if err == 'not found' else 400
|
|
return jsonify({'success': True})
|
|
|
|
|
|
@recycle_bin_bp.route('/purge-expired', methods=['POST'])
|
|
@admin_required
|
|
def purge_expired():
|
|
body = request.get_json(silent=True) or {}
|
|
days = body.get('retention_days', recycle_bin_service.DEFAULT_RETENTION_DAYS)
|
|
try:
|
|
days = max(1, int(days))
|
|
except (TypeError, ValueError):
|
|
days = recycle_bin_service.DEFAULT_RETENTION_DAYS
|
|
return jsonify({'purged': recycle_bin_service.purge_expired(days)})
|