mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-03 18:46:47 +09:00
147 lines
4.8 KiB
Python
147 lines
4.8 KiB
Python
from flask import Blueprint, request, jsonify
|
|
from flask_jwt_extended import jwt_required, get_jwt_identity
|
|
from app.middleware.rbac import admin_required
|
|
from app.services.ssl_service import SSLService, get_acme_contact, remember_acme_contact
|
|
|
|
ssl_bp = Blueprint('ssl', __name__)
|
|
|
|
|
|
@ssl_bp.route('/acme-contact', methods=['GET'])
|
|
@jwt_required()
|
|
def acme_contact():
|
|
"""The panel-wide ACME contact, for prefilling certificate forms.
|
|
|
|
Deliberately not admin-gated: issuing a certificate for an app you can
|
|
edit is not an admin action, so the surfaces that need this prefill are
|
|
reachable by non-admins too. The value is the address the operator chose
|
|
to receive expiry notices at, not a credential.
|
|
"""
|
|
return jsonify({'email': get_acme_contact() or ''}), 200
|
|
|
|
|
|
@ssl_bp.route('/certificates', methods=['GET'])
|
|
@admin_required
|
|
def list_certificates():
|
|
"""List all SSL certificates."""
|
|
certificates = SSLService.list_certificates()
|
|
return jsonify({'certificates': certificates}), 200
|
|
|
|
|
|
@ssl_bp.route('/certificates/<domain>', methods=['GET'])
|
|
@admin_required
|
|
def get_certificate(domain):
|
|
"""Get details for a specific certificate."""
|
|
info = SSLService.get_certificate_info(domain)
|
|
if info:
|
|
return jsonify({'certificate': info}), 200
|
|
return jsonify({'error': 'Certificate not found'}), 404
|
|
|
|
|
|
@ssl_bp.route('/certificates', methods=['POST'])
|
|
@admin_required
|
|
def obtain_certificate():
|
|
"""Obtain a new SSL certificate from Let's Encrypt."""
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return jsonify({'error': 'No data provided'}), 400
|
|
|
|
domains = data.get('domains')
|
|
# Falls back to the stored panel-wide contact, so a caller that has one
|
|
# configured no longer has to repeat it on every request.
|
|
email = get_acme_contact(data.get('email'))
|
|
|
|
if not domains or not email:
|
|
return jsonify({'error': 'domains and email are required'}), 400
|
|
|
|
result = SSLService.obtain_certificate(
|
|
domains=domains,
|
|
email=email,
|
|
webroot_path=data.get('webroot_path'),
|
|
use_nginx=data.get('use_nginx', True)
|
|
)
|
|
|
|
if result.get('success'):
|
|
remember_acme_contact(data.get('email'), user_id=get_jwt_identity())
|
|
|
|
return jsonify(result), 201 if result['success'] else 400
|
|
|
|
|
|
@ssl_bp.route('/certificates/<domain>/renew', methods=['POST'])
|
|
@admin_required
|
|
def renew_certificate(domain):
|
|
"""Renew a specific certificate."""
|
|
result = SSLService.renew_certificate(domain)
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@ssl_bp.route('/certificates/renew-all', methods=['POST'])
|
|
@admin_required
|
|
def renew_all_certificates():
|
|
"""Renew all certificates that need renewal."""
|
|
result = SSLService.renew_certificate()
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@ssl_bp.route('/certificates/<domain>', methods=['DELETE'])
|
|
@admin_required
|
|
def revoke_certificate(domain):
|
|
"""Revoke and delete a certificate."""
|
|
result = SSLService.revoke_certificate(domain)
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@ssl_bp.route('/certificates/<domain>/check', methods=['GET'])
|
|
@admin_required
|
|
def check_certificate_expiry(domain):
|
|
"""Check if a certificate is expiring soon."""
|
|
result = SSLService.check_expiry(domain)
|
|
return jsonify(result), 200
|
|
|
|
|
|
@ssl_bp.route('/auto-renewal', methods=['POST'])
|
|
@admin_required
|
|
def setup_auto_renewal():
|
|
"""Set up automatic certificate renewal."""
|
|
result = SSLService.setup_auto_renewal()
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@ssl_bp.route('/install-certbot', methods=['POST'])
|
|
@admin_required
|
|
def install_certbot():
|
|
"""Install certbot if not already installed."""
|
|
if SSLService.is_certbot_installed():
|
|
return jsonify({'success': True, 'message': 'Certbot is already installed'}), 200
|
|
|
|
result = SSLService.install_certbot()
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@ssl_bp.route('/status', methods=['GET'])
|
|
@admin_required
|
|
def get_ssl_status():
|
|
"""Get overall SSL status."""
|
|
is_installed = SSLService.is_certbot_installed()
|
|
certificates, list_errors = SSLService.list_certificates_report()
|
|
|
|
expiring_soon = []
|
|
check_errors = []
|
|
for cert in certificates:
|
|
check = SSLService.check_expiry(cert['name'])
|
|
if check.get('error'):
|
|
# A failed check must be visible as failed, not vanish from the
|
|
# expiring list and read as "not expiring".
|
|
check_errors.append({'name': cert['name'], 'error': check['error']})
|
|
elif check.get('expiring_soon'):
|
|
expiring_soon.append(cert['name'])
|
|
|
|
return jsonify({
|
|
'certbot_installed': is_installed,
|
|
'total_certificates': len(certificates),
|
|
'expiring_soon': expiring_soon,
|
|
'check_errors': check_errors,
|
|
'list_errors': list_errors,
|
|
'certificates': certificates
|
|
}), 200
|