mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-09 05:24:49 +09:00
Juan DenisandClaude Opus 4.8
363a09dc7e
Rate-limit buckets, login lockout and audit-log source IPs hand-parsed the leftmost X-Forwarded-For token — a client-controlled value — letting an attacker rotate it to dodge per-IP limits or forge audit trails. - config: add TRUST_PROXY_HEADERS / TRUSTED_PROXY_HOPS (+ _env_bool/_env_int) - create_app: wrap in gated Werkzeug ProxyFix (rightmost trusted hops) before the limiter/handlers so every remote_addr consumer sees the real client - new app/utils/client_ip.py::get_client_ip() — the single trusted seam - replace all six leftmost-XFF parsers (rate_limit, audit middleware, api_key_auth, api_analytics, audit_service, ddns) with get_client_ip() - deploy: root .env.example ships TRUST_PROXY_HEADERS=true (behind nginx); backend/.env.example documents it off for dev - tests: test_trusted_client_ip.py (7) incl. real flask-limiter bucket isolation Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
85 lines
2.9 KiB
Python
85 lines
2.9 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():
|
|
# DDNS legitimately wants the real client IP — the same trusted seam every
|
|
# other consumer uses (ProxyFix-corrected remote_addr, see utils.client_ip).
|
|
from app.utils.client_ip import get_client_ip
|
|
return get_client_ip()
|
|
|
|
|
|
@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})
|