linuxrouter/docker/routlin-dash/app/pages/accountcreate/action.py

from pathlib import Path
from flask import Blueprint, request, session, redirect, flash
import json, os, bcrypt, secrets, smtplib
from datetime import datetime, timezone, timedelta
from email.message import EmailMessage
from auth import require_level
from config_utils import WEB_APP_DISPLAY_NAME, ACCOUNTS_FILE
import sanitize

_PAGE = Path(__file__).parent.name

bp = Blueprint(_PAGE, __name__)

CODE_TTL_MIN  = 15


def _load_accounts():
    try:
        with open(ACCOUNTS_FILE) as f:
            return json.load(f)
    except Exception:
        return {'accounts': []}


def _send_verification_email(to_address, code):
    host      = os.environ.get('SMTP_HOST', '')
    port      = int(os.environ.get('SMTP_PORT', 587))
    user      = os.environ.get('SMTP_USER', '')
    password  = os.environ.get('SMTP_PASSWORD', '')
    from_addr = os.environ.get('SMTP_FROM', user)

    if not host:
        raise RuntimeError('SMTP_HOST is not configured.')

    msg = EmailMessage()
    msg['Subject'] = f'{WEB_APP_DISPLAY_NAME} - Email Verification'
    msg['From']    = from_addr
    msg['To']      = to_address
    msg.set_content(
        f'Your verification code is: {code}\n\n'
        f'This code expires in {CODE_TTL_MIN} minutes.\n\n'
        f'If you did not request this, you can ignore this email.'
    )

    with smtplib.SMTP(host, port) as smtp:
        smtp.ehlo()
        if port != 465:
            smtp.starttls()
        if user and password:
            smtp.login(user, password)
        smtp.send_message(msg)


@bp.route('/action/accountcreate/form_create', methods=['POST'])
@require_level('nothing')
def form_create():
    # Abort if already logged in
    if session.get('access_level', 'nothing') != 'nothing':
        return redirect('/overview')

    email            = sanitize.email(request.form.get('email', ''))
    password         = request.form.get('password', '')
    password_confirm = request.form.get('password_confirm', '')
    tz               = sanitize.timezone(request.form.get('timezone', '').strip())

    if not email or not password or not password_confirm or not tz:
        flash('All fields are required.', 'error')
        return redirect(f'/{_PAGE}')

    if password != password_confirm:
        flash('Passwords do not match.', 'error')
        return redirect(f'/{_PAGE}')

    if len(password) < 8:
        flash('Password must be at least 8 characters.', 'error')
        return redirect(f'/{_PAGE}')

    accounts = _load_accounts().get('accounts', [])
    account  = next((a for a in accounts if a.get('email_address', '').lower() == email), None)

    if account is None:
        flash('Email address not recognised. Contact your manager.', 'error')
        return redirect(f'/{_PAGE}')

    if account.get('hashed_password'):
        flash('This account is already set up. Please log in instead.', 'error')
        return redirect(f'/{_PAGE}')

    salt    = bcrypt.gensalt()
    hashed  = bcrypt.hashpw(password.encode('utf-8'), salt)
    code    = f'{secrets.randbelow(1000000):06d}'
    expires = (datetime.now(tz=timezone.utc) + timedelta(minutes=CODE_TTL_MIN)).isoformat()

    try:
        _send_verification_email(account['email_address'], code)
    except Exception as exc:
        flash(f'Could not send verification email: {exc}', 'error')
        return redirect(f'/{_PAGE}')

    session['pending_create_account'] = {
        'email':           account['email_address'],
        'hashed_password': hashed.decode('utf-8'),
        'timezone':        tz,
        'code':            code,
        'expires':         expires,
    }

    return redirect('/accountverifyemail')