mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-10-29 16:17:25 +09:00
Juan Denis
99a19c2fb7
The first pass left routes that hand the body straight to a service (create_zone(data), MonitorService.create(data), ...). Those services read it as a dict, so a body of null, [] or a bare string still crashed them into a 500. Agent long-poll, external webhook receivers and the routes that already type-check their body are left alone.
79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
"""Domain-registrar connections — the portfolio + expiry surface.
|
|
|
|
Connections are admin-managed (like DNS providers). Reading the portfolio is
|
|
available to any authenticated user so the Domains page and Connections cards
|
|
can show "what we own and when it lapses".
|
|
"""
|
|
|
|
from flask import Blueprint, jsonify
|
|
from flask_jwt_extended import jwt_required
|
|
|
|
from app.services.registrar_service import RegistrarService
|
|
from app.middleware.rbac import admin_required, get_current_user
|
|
from app.api._request import json_object
|
|
|
|
registrars_bp = Blueprint('registrars', __name__)
|
|
|
|
|
|
@registrars_bp.route('/connections', methods=['GET'])
|
|
@jwt_required()
|
|
def list_connections():
|
|
conns = RegistrarService.list_connections()
|
|
return jsonify({'connections': [c.to_dict() for c in conns]})
|
|
|
|
|
|
@registrars_bp.route('/connections', methods=['POST'])
|
|
@jwt_required()
|
|
def add_connection():
|
|
user = get_current_user()
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
data = json_object()
|
|
try:
|
|
conn = RegistrarService.add_connection(data, user.id)
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
# Validate the credentials immediately; drop the row if they don't work so
|
|
# we never persist a dead connection.
|
|
test = RegistrarService.test_connection(conn)
|
|
if not test.get('success'):
|
|
RegistrarService.delete_connection(conn.id)
|
|
return jsonify({'error': test.get('error', 'Connection test failed')}), 400
|
|
return jsonify({'connection': conn.to_dict(), 'test': test}), 201
|
|
|
|
|
|
@registrars_bp.route('/connections/<int:cid>', methods=['DELETE'])
|
|
@jwt_required()
|
|
def delete_connection(cid):
|
|
user = get_current_user()
|
|
if not user or not user.is_admin:
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
if not RegistrarService.delete_connection(cid):
|
|
return jsonify({'error': 'Not found'}), 404
|
|
return jsonify({'message': 'Registrar disconnected'})
|
|
|
|
|
|
@registrars_bp.route('/connections/<int:cid>/test', methods=['POST'])
|
|
@admin_required
|
|
def test_connection(cid):
|
|
conn = RegistrarService.get_connection(cid)
|
|
if not conn:
|
|
return jsonify({'error': 'Not found'}), 404
|
|
result = RegistrarService.test_connection(conn)
|
|
return jsonify(result), 200 if result.get('success') else 400
|
|
|
|
|
|
@registrars_bp.route('/domains', methods=['GET'])
|
|
@jwt_required()
|
|
def list_domains():
|
|
"""Aggregated domain portfolio across all connected registrars."""
|
|
return jsonify({'domains': RegistrarService.list_all_domains()})
|
|
|
|
|
|
@registrars_bp.route('/sync', methods=['POST'])
|
|
@admin_required
|
|
def sync():
|
|
"""Force-refresh the portfolio (and stamp last_synced_at)."""
|
|
return jsonify({'domains': RegistrarService.sync_now()})
|