mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-10 05:53:05 +09:00
Juan DenisandClaude Opus 4.8
5f28f3e700
- DdnsHost model: a hostname bound to a DNS zone plus a secret token; the token is masked in API responses except when first minted. - DdnsService.update_ip reuses DNSZoneService so a configured provider (e.g. Cloudflare) is synced automatically; chooses A vs AAAA by IP family; no-op when the IP is unchanged. - API: admin CRUD for hosts + a public, token-authenticated /ddns/update endpoint that a router or cron calls when its IP changes (IP from ?ip= or the request source). Wrong token -> 401, bad IP -> 400. - 8 tests covering the service flow and both endpoints. UI panel to follow. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
from flask import Blueprint, request, jsonify
|
|
from flask_jwt_extended import jwt_required, get_jwt_identity
|
|
from app.models.user import User
|
|
from app.services.ddns_service import DdnsService
|
|
|
|
ddns_bp = Blueprint('ddns', __name__)
|
|
|
|
|
|
def _admin():
|
|
user = User.query.get(get_jwt_identity())
|
|
return user if user and user.is_admin else None
|
|
|
|
|
|
# --- Management (authenticated, admin) ---
|
|
|
|
@ddns_bp.route('/hosts', methods=['GET'])
|
|
@jwt_required()
|
|
def list_hosts():
|
|
hosts = DdnsService.list_hosts()
|
|
return jsonify({'hosts': [h.to_dict() for h in hosts]})
|
|
|
|
|
|
@ddns_bp.route('/hosts', methods=['POST'])
|
|
@jwt_required()
|
|
def create_host():
|
|
if not _admin():
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
try:
|
|
host = DdnsService.create_host(request.get_json() or {})
|
|
# Return the token once, on creation — it is masked everywhere else.
|
|
return jsonify(host.to_dict(include_token=True)), 201
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
|
|
@ddns_bp.route('/hosts/<int:host_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
def delete_host(host_id):
|
|
if not _admin():
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
if not DdnsService.delete_host(host_id):
|
|
return jsonify({'error': 'Host not found'}), 404
|
|
return jsonify({'message': 'Host deleted'})
|
|
|
|
|
|
@ddns_bp.route('/hosts/<int:host_id>/regenerate-token', methods=['POST'])
|
|
@jwt_required()
|
|
def regenerate_token(host_id):
|
|
if not _admin():
|
|
return jsonify({'error': 'Admin access required'}), 403
|
|
host = DdnsService.regenerate_token(host_id)
|
|
if not host:
|
|
return jsonify({'error': 'Host not found'}), 404
|
|
return jsonify(host.to_dict(include_token=True))
|
|
|
|
|
|
# --- Update endpoint (public, token-authenticated) ---
|
|
|
|
def _client_ip():
|
|
forwarded = request.headers.get('X-Forwarded-For', '')
|
|
if forwarded:
|
|
return forwarded.split(',')[0].strip()
|
|
return request.remote_addr
|
|
|
|
|
|
@ddns_bp.route('/update', methods=['GET', 'POST'])
|
|
def update():
|
|
"""Called by a router or cron when its public IP changes. There is no JWT
|
|
here — the per-host token is the credential. The IP can be passed explicitly
|
|
(?ip=) or inferred from the request source address."""
|
|
body = request.get_json(silent=True) or {}
|
|
token = request.args.get('token') or request.headers.get('X-DDNS-Token') or body.get('token')
|
|
if not token:
|
|
return jsonify({'error': 'token required'}), 401
|
|
|
|
ip = request.args.get('ip') or body.get('ip') or _client_ip()
|
|
try:
|
|
status, host = DdnsService.update_ip(token, ip)
|
|
except ValueError as e:
|
|
message = str(e)
|
|
code = 401 if 'token' in message.lower() else 400
|
|
return jsonify({'error': message}), code
|
|
|
|
return jsonify({'status': status, 'hostname': host.hostname, 'ip': ip})
|