roof
777ffa4fb2
Unit Tests / test (push) Successful in 15m23s
_check_pic_ngo_available was hardcoding https://ddns.pic.ngo, ignoring DDNS_URL. Now imports DDNS_API_BASE from setup_manager so both the availability check and DDNS registration use the same configured URL. API container now receives DDNS_URL and DDNS_TOTP_SECRET from env. Default DDNS_URL points to http://ddns.pic.ngo:8080/api/v1 (the FastAPI service runs on port 8080 without TLS termination in front). Also returns 503 (not 500) when the DDNS service is unreachable. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
128 lines
4.4 KiB
Python
128 lines
4.4 KiB
Python
import logging
|
|
import re
|
|
import urllib.request
|
|
import urllib.error
|
|
import json as _json
|
|
from flask import Blueprint, request, jsonify
|
|
from setup_manager import DDNS_API_BASE
|
|
|
|
logger = logging.getLogger('picell')
|
|
|
|
setup_bp = Blueprint('setup', __name__, url_prefix='/api/setup')
|
|
|
|
_DOMAIN_RE = re.compile(r'^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z]{2,})+$', re.I)
|
|
|
|
|
|
def _get_setup_manager():
|
|
from app import setup_manager
|
|
return setup_manager
|
|
|
|
|
|
@setup_bp.route('/status', methods=['GET'])
|
|
def get_setup_status():
|
|
"""Return wizard status and available options."""
|
|
sm = _get_setup_manager()
|
|
if sm.is_setup_complete():
|
|
return jsonify({'error': 'Setup already complete'}), 410
|
|
return jsonify(sm.get_setup_status())
|
|
|
|
|
|
@setup_bp.route('/validate', methods=['POST'])
|
|
def validate_setup_step():
|
|
"""Validate a single wizard step.
|
|
|
|
Supported steps: ``cell_name``, ``password``,
|
|
``pic_ngo_available``, ``cloudflare_token``, ``duckdns_token``.
|
|
"""
|
|
sm = _get_setup_manager()
|
|
if sm.is_setup_complete():
|
|
return jsonify({'error': 'Setup already complete'}), 410
|
|
|
|
body = request.get_json(silent=True) or {}
|
|
step = body.get('step', '')
|
|
data = body.get('data', {})
|
|
|
|
if step == 'cell_name':
|
|
errors = sm.validate_cell_name(data.get('cell_name', ''))
|
|
return jsonify({'valid': len(errors) == 0, 'errors': errors})
|
|
|
|
if step == 'password':
|
|
errors = sm.validate_password(data.get('password', ''))
|
|
return jsonify({'valid': len(errors) == 0, 'errors': errors})
|
|
|
|
if step == 'pic_ngo_available':
|
|
name = data.get('cell_name', '').strip()
|
|
errors = sm.validate_cell_name(name)
|
|
if errors:
|
|
return jsonify({'available': False, 'errors': errors})
|
|
try:
|
|
available = _check_pic_ngo_available(name)
|
|
return jsonify({'available': available})
|
|
except Exception:
|
|
return jsonify({'available': False, 'error': 'DDNS service unreachable'}), 503
|
|
|
|
if step == 'cloudflare_token':
|
|
token = data.get('token', '').strip()
|
|
if not token:
|
|
return jsonify({'valid': False, 'error': 'Token is required.'})
|
|
valid = _verify_cloudflare_token(token)
|
|
return jsonify({'valid': valid})
|
|
|
|
if step == 'duckdns_token':
|
|
subdomain = data.get('subdomain', '').strip()
|
|
token = data.get('token', '').strip()
|
|
if not token or not subdomain:
|
|
return jsonify({'valid': False, 'error': 'Subdomain and token are required.'})
|
|
valid = _verify_duckdns_token(subdomain, token)
|
|
return jsonify({'valid': valid})
|
|
|
|
return jsonify({'valid': False, 'errors': [f"Unknown step: {step!r}"]}), 400
|
|
|
|
|
|
@setup_bp.route('/complete', methods=['POST'])
|
|
def complete_setup():
|
|
"""Complete the first-run wizard and create the admin account."""
|
|
sm = _get_setup_manager()
|
|
if sm.is_setup_complete():
|
|
return jsonify({'error': 'Setup already complete'}), 410
|
|
|
|
payload = request.get_json(silent=True) or {}
|
|
result = sm.complete_setup(payload)
|
|
status_code = 200 if result.get('success') else 400
|
|
return jsonify(result), status_code
|
|
|
|
|
|
# ── external validation helpers ───────────────────────────────────────────────
|
|
|
|
def _check_pic_ngo_available(name: str) -> bool:
|
|
try:
|
|
url = f'{DDNS_API_BASE}/api/v1/check/{name}'
|
|
with urllib.request.urlopen(url, timeout=8) as resp:
|
|
body = _json.loads(resp.read())
|
|
return bool(body.get('available'))
|
|
except Exception as exc:
|
|
logger.warning(f'DDNS availability check failed for {name!r}: {exc}')
|
|
raise
|
|
|
|
|
|
def _verify_cloudflare_token(token: str) -> bool:
|
|
try:
|
|
req = urllib.request.Request(
|
|
'https://api.cloudflare.com/client/v4/user/tokens/verify',
|
|
headers={'Authorization': f'Bearer {token}'},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=8) as resp:
|
|
body = _json.loads(resp.read())
|
|
return bool(body.get('success'))
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _verify_duckdns_token(subdomain: str, token: str) -> bool:
|
|
try:
|
|
url = f'https://www.duckdns.org/update?domains={subdomain}&token={token}&ip='
|
|
with urllib.request.urlopen(url, timeout=8) as resp:
|
|
return resp.read().strip() == b'OK'
|
|
except Exception:
|
|
return False
|