shopdb-flask/shopdb/core/api/setup.py

"""First-run setup wizard support endpoints.

The wizard itself is frontend; these endpoints cover the pieces that need
server work: seeding common starter data and marking setup finished. Site,
plugin, and map configuration reuse the existing settings / plugins APIs.
"""

import hmac
import ipaddress
import os

from flask import Blueprint, request, current_app
from flask_jwt_extended import jwt_required
from werkzeug.security import generate_password_hash

from shopdb.extensions import db
from shopdb.core.models import Vendor, Setting, User, Role
from shopdb.utils.responses import success_response, error_response, ErrorCodes
from shopdb.utils.authz import require_role

setup_bp = Blueprint('setup', __name__)

# Common hardware vendors most sites will want on hand.
STARTER_VENDORS = [
    ('Dell Inc.', 'https://www.dell.com'),
    ('HP Inc.', 'https://www.hp.com'),
    ('Lenovo', 'https://www.lenovo.com'),
    ('Xerox', 'https://www.xerox.com'),
    ('Zebra Technologies', 'https://www.zebra.com'),
    ('Cisco', 'https://www.cisco.com'),
    ('Brother', 'https://www.brother.com'),
    ('Microsoft', 'https://www.microsoft.com'),
]


# Minimum length for the bootstrap administrator password. 12 characters is the
# shortest length that still resists offline cracking of a single leaked hash.
MIN_ADMIN_PASSWORD_LENGTH = 12


def _is_loopback(addr):
    """True when the request originated on the machine running the API."""
    if not addr:
        return False
    try:
        return ipaddress.ip_address(addr.split('%')[0]).is_loopback
    except ValueError:
        return False


def _bootstrap_allowed():
    """Proof of operator control for first-admin creation.

    An empty users table is an ordering condition, not authorization: the API
    serves requests before the first account exists on every fresh deploy, so a
    remote caller could otherwise claim the instance during the install window.
    Require either the one-time setup token (SETUP_TOKEN in config or the
    environment, presented as the X-Setup-Token header) or a request coming from
    loopback, i.e. the host the operator is installing on.
    """
    token = str(current_app.config.get('SETUP_TOKEN')
                or os.environ.get('SETUP_TOKEN') or '').strip()
    if token:
        supplied = (request.headers.get('X-Setup-Token') or '').strip()
        return bool(supplied) and hmac.compare_digest(supplied, token)
    return _is_loopback(request.remote_addr)


def _setup_already_complete():
    """setup_complete is written once the wizard finishes; treat it as final."""
    row = Setting.query.filter_by(key='setup_complete').first()
    return bool(row) and str(row.value).strip().lower() in ('true', '1', 'yes')


@setup_bp.route('/needs-admin', methods=['GET'])
def needs_admin():
    """True when no users exist yet - the app's very first run. Public so the
    login screen can offer to create the first admin."""
    return success_response({'needsadmin': User.query.count() == 0})


@setup_bp.route('/create-admin', methods=['POST'])
def create_admin():
    """Create the first admin. Only works while there are zero users, setup has
    not been marked complete, and the caller proves operator control (setup
    token or a loopback request)."""
    if User.query.count() > 0 or _setup_already_complete():
        return error_response(ErrorCodes.FORBIDDEN,
                              'An account already exists; sign in instead.', http_code=403)
    if not _bootstrap_allowed():
        return error_response(
            ErrorCodes.FORBIDDEN,
            'First-admin creation requires the setup token (X-Setup-Token) '
            'or a request from the server itself.',
            http_code=403)
    data = request.get_json() or {}
    username = (data.get('username') or '').strip()
    email = (data.get('email') or '').strip()
    password = data.get('password') or ''
    if not (username and email and password):
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'username, email and password are required')
    if len(password) < MIN_ADMIN_PASSWORD_LENGTH:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            f'password must be at least {MIN_ADMIN_PASSWORD_LENGTH} characters')

    admin_role = Role.query.filter_by(rolename='admin').first()
    if not admin_role:
        admin_role = Role(rolename='admin', description='Administrator')
        db.session.add(admin_role)

    user = User(username=username, email=email,
                passwordhash=generate_password_hash(password), isactive=True)
    user.roles.append(admin_role)
    db.session.add(user)
    db.session.commit()
    current_app.logger.warning('First administrator %r created from %s',
                               username, request.remote_addr)
    return success_response({'username': username}, message='Admin created', http_code=201)


@setup_bp.route('/seed-reference', methods=['POST'])
@jwt_required()
@require_role('admin')
def seed_reference():
    """Seed core reference data + permissions + default settings. Idempotent -
    reuses the same routines as the `flask seed ...` CLI commands."""
    # Invoke the seed CLI commands in a proper click context via the runner.
    runner = current_app.test_cli_runner()
    for command in ('reference-data', 'permissions', 'settings'):
        result = runner.invoke(args=['seed', command])
        if result.exit_code not in (0, None):
            return error_response(ErrorCodes.INTERNAL_ERROR,
                                  f'seed {command} failed: {result.output}', http_code=500)
    return success_response(message='Reference data, permissions and settings seeded.')


@setup_bp.route('/seed-starter', methods=['POST'])
@jwt_required()
@require_role('admin')
def seed_starter():
    """Add common vendors that are not already present. Idempotent."""
    existing = {v.vendor.strip().lower() for v in Vendor.query.all()}
    added = []
    for name, website in STARTER_VENDORS:
        if name.strip().lower() in existing:
            continue
        db.session.add(Vendor(vendor=name, website=website))
        added.append(name)
    db.session.commit()
    return success_response(
        {'added': added, 'addedcount': len(added)},
        message=f'Seeded {len(added)} vendor(s).'
    )


@setup_bp.route('/complete', methods=['POST'])
@jwt_required()
@require_role('admin')
def mark_complete():
    """Flag the first-run setup as finished."""
    row = Setting.query.filter_by(key='setup_complete').first()
    if row:
        row.value = 'true'
    else:
        db.session.add(Setting(key='setup_complete', value='true',
                               valuetype='boolean', category='site'))
    db.session.commit()
    return success_response({'complete': True}, message='Setup marked complete.')