pic/api/routes/setup.py

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)
    if result.get('success'):
        try:
            from app import config_manager, service_bus, EventType
            identity = config_manager.configs.get('_identity', {})
            service_bus.publish_event(EventType.IDENTITY_CHANGED, 'setup', {
                'cell_name': identity.get('cell_name'),
                'domain': identity.get('domain'),
                'domain_name': identity.get('domain_name'),
                'domain_mode': identity.get('domain_mode'),
                'effective_domain': config_manager.get_effective_domain(),
            })
        except Exception as exc:
            logger.warning(f'Failed to publish IDENTITY_CHANGED after setup: {exc}')
    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