mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-02 18:17:11 +09:00
Juan DenisandClaude Opus 5
fe02ce8eeb
Plan 76 milestone A, drift/bug row 2 — scheduled as a fix, not a refactor,
because it is a behaviour difference shipping today.
rbac.get_current_user() is the only lookup that knows about both auth
mechanisms: it returns g.api_key_user for an X-API-Key request and falls back
to the JWT. 73 call sites answered "who is calling" with
`User.query.get(get_jwt_identity())` instead, behind 19 near-identical private
helpers (12 named get_current_user(), shadowing the real door).
The audit that scheduled this row predicted those return None for API-key
callers. Measured, it is worse: auth_required() deliberately skips
verify_jwt_in_request() once the key middleware has authenticated, so there is
no JWT context and get_jwt_identity() *raises*. Nine handlers reached through
an API-key-capable policy decorator therefore returned 500 to every API-key
caller:
bandwidth.get_app_bandwidth @viewer_required
servers.set_server_workspace @developer_required
servers.create_server @developer_required
sso.get_sso_config / update_provider_config / update_general_settings
source_connections.update_{github,gitlab,bitbucket}_admin_config
source_connections.complete_github_app_manifest
databases.execute_{mysql,pg,sqlite,docker}_query (see below)
Three service helpers had the same lookup inside a blanket `except Exception`,
which converted the raise into a silent user_id=None — the write landed and
the audit trail lost the actor (shared_resource_service, cf_ops_change_service,
secrets_webhooks).
The four databases.execute_*_query sites were a second admin check *inside* an
already-@admin_required route: dead code that always passed for JWT callers and
crashed for API-key ones. They now call require_admin_user() rather than
re-deriving the answer.
Also folds in a sixth _require_admin() found in the cloudflare-ops extension
source (29 gates, the same inverted contract as jobs.py), and fixes
serverkit-gui's blueprint at its sibling-repo source rather than in this repo's
live copy.
Ratchet (second-wave rule 1 — a migration without one regrows):
tests/identity_door_census.py counts both bypass shapes by AST, direct and the
two-line `uid = get_jwt_identity(); User.query.get(uid)` variant. The ceiling is
87, all of them on @jwt_required()-only routes where reading the JWT is correct
today; the ratchet's third assertion holds the API-key-capable population at
exactly zero, so migrating one of those routes to a policy decorator without
fixing its lookup now fails. Verified non-vacuous: an injected bypass fails both
assertions and passes again on revert.
The first wave's controller-boundary baseline is regenerated — this removes 74
persistence crossings (593 -> 519), a debt reduction with no new entries.
Gate: tests/test_api_key_identity.py (7, each red before the fix),
test_identity_door_ratchet.py (10), test_api_controller_boundaries.py (4), plus
the 65 test files covering every touched module: 1024 passed, 10 skipped.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
98 lines
3.1 KiB
Python
98 lines
3.1 KiB
Python
from flask import Blueprint, request, jsonify
|
|
from flask_jwt_extended import jwt_required
|
|
from app.services.advanced_ssl_service import AdvancedSSLService
|
|
from app.services.audit_service import AuditService
|
|
from app.models.audit_log import AuditLog
|
|
from app.middleware.rbac import get_current_user
|
|
|
|
advanced_ssl_bp = Blueprint('advanced_ssl', __name__)
|
|
|
|
|
|
@advanced_ssl_bp.route('/profiles', methods=['GET'])
|
|
@jwt_required()
|
|
def get_profiles():
|
|
return jsonify({'profiles': AdvancedSSLService.get_ssl_profiles()})
|
|
|
|
|
|
@advanced_ssl_bp.route('/wildcard', methods=['POST'])
|
|
@jwt_required()
|
|
def issue_wildcard():
|
|
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')
|
|
provider = data.get('dns_provider')
|
|
creds = data.get('credentials', {})
|
|
|
|
if not domain or not provider:
|
|
return jsonify({'error': 'domain and dns_provider required'}), 400
|
|
|
|
result = AdvancedSSLService.issue_wildcard_cert(domain, provider, creds)
|
|
AuditService.log(
|
|
action=AuditLog.ACTION_RESOURCE_CREATE, user_id=user.id,
|
|
target_type='ssl_wildcard', target_id=0,
|
|
details={'domain': domain, 'success': result.get('success')}
|
|
)
|
|
status = 200 if result.get('success') else 400
|
|
return jsonify(result), status
|
|
|
|
|
|
@advanced_ssl_bp.route('/san', methods=['POST'])
|
|
@jwt_required()
|
|
def issue_san():
|
|
user = get_current_user()
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
|
|
data = request.get_json() or {}
|
|
domains = data.get('domains', [])
|
|
if not domains:
|
|
return jsonify({'error': 'domains list required'}), 400
|
|
|
|
try:
|
|
result = AdvancedSSLService.issue_san_cert(domains)
|
|
status = 200 if result.get('success') else 400
|
|
return jsonify(result), status
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
|
|
@advanced_ssl_bp.route('/upload', methods=['POST'])
|
|
@jwt_required()
|
|
def upload_cert():
|
|
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')
|
|
cert = data.get('certificate')
|
|
key = data.get('private_key')
|
|
chain = data.get('chain')
|
|
|
|
if not domain or not cert or not key:
|
|
return jsonify({'error': 'domain, certificate, and private_key required'}), 400
|
|
|
|
try:
|
|
result = AdvancedSSLService.upload_custom_cert(domain, cert, key, chain)
|
|
return jsonify(result), 201
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
|
|
@advanced_ssl_bp.route('/health/<domain>', methods=['GET'])
|
|
@jwt_required()
|
|
def cert_health(domain):
|
|
result = AdvancedSSLService.get_cert_health(domain)
|
|
return jsonify(result)
|
|
|
|
|
|
@advanced_ssl_bp.route('/expiry-alerts', methods=['GET'])
|
|
@jwt_required()
|
|
def expiry_alerts():
|
|
days = request.args.get('days', 30, type=int)
|
|
alerts = AdvancedSSLService.get_expiry_alerts(days)
|
|
return jsonify({'alerts': alerts, 'threshold_days': days})
|