mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-10-27 23:28:03 +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>
94 lines
2.2 KiB
Python
94 lines
2.2 KiB
Python
"""
|
|
Uptime Tracking API
|
|
|
|
Endpoints for server uptime monitoring and statistics.
|
|
"""
|
|
|
|
from flask import Blueprint, request, jsonify
|
|
from flask_jwt_extended import jwt_required
|
|
|
|
from app.services.uptime_service import UptimeService
|
|
from app.middleware.rbac import admin_required
|
|
|
|
uptime_bp = Blueprint('uptime', __name__)
|
|
|
|
|
|
@uptime_bp.route('/current', methods=['GET'])
|
|
@jwt_required()
|
|
def get_current_uptime():
|
|
"""Get current server uptime."""
|
|
uptime = UptimeService.get_current_uptime()
|
|
return jsonify(uptime)
|
|
|
|
|
|
@uptime_bp.route('/stats', methods=['GET'])
|
|
@jwt_required()
|
|
def get_uptime_stats():
|
|
"""Get comprehensive uptime statistics."""
|
|
stats = UptimeService.get_uptime_stats()
|
|
return jsonify(stats)
|
|
|
|
|
|
@uptime_bp.route('/graph', methods=['GET'])
|
|
@jwt_required()
|
|
def get_uptime_graph():
|
|
"""
|
|
Get uptime graph data.
|
|
|
|
Query params:
|
|
period: '24h', '7d', '30d', '90d' (default: '24h')
|
|
"""
|
|
period = request.args.get('period', '24h')
|
|
if period not in ['24h', '7d', '30d', '90d']:
|
|
period = '24h'
|
|
|
|
data = UptimeService.get_uptime_graph_data(period)
|
|
return jsonify(data)
|
|
|
|
|
|
@uptime_bp.route('/history', methods=['GET'])
|
|
@jwt_required()
|
|
def get_uptime_history():
|
|
"""
|
|
Get uptime check history.
|
|
|
|
Query params:
|
|
hours: Number of hours to look back (default: 24)
|
|
"""
|
|
hours = request.args.get('hours', 24, type=int)
|
|
hours = min(max(hours, 1), 24 * 90) # Limit to 90 days
|
|
|
|
history = UptimeService.get_uptime_history(hours)
|
|
return jsonify({
|
|
'hours': hours,
|
|
'checks': history,
|
|
'total': len(history)
|
|
})
|
|
|
|
|
|
@uptime_bp.route('/tracking/start', methods=['POST'])
|
|
@admin_required
|
|
def start_tracking():
|
|
"""Start uptime tracking."""
|
|
result = UptimeService.start_tracking()
|
|
if result.get('success'):
|
|
return jsonify(result)
|
|
return jsonify(result), 400
|
|
|
|
|
|
@uptime_bp.route('/tracking/stop', methods=['POST'])
|
|
@admin_required
|
|
def stop_tracking():
|
|
"""Stop uptime tracking."""
|
|
result = UptimeService.stop_tracking()
|
|
return jsonify(result)
|
|
|
|
|
|
@uptime_bp.route('/tracking/status', methods=['GET'])
|
|
@jwt_required()
|
|
def get_tracking_status():
|
|
"""Get uptime tracking status."""
|
|
return jsonify({
|
|
'is_tracking': UptimeService.is_tracking()
|
|
})
|