from pathlib import Path from flask import Blueprint, request, session, redirect, flash import os, bcrypt, secrets, smtplib import time from email.message import EmailMessage import auth import config_utils import sanitize _PAGE = Path(__file__).parent.name bp = Blueprint(_PAGE, __name__) CODE_TTL_SECS = 15 * 60 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'{config_utils.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 15 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) def _tz_to_offset_seconds(tz_str): try: from zoneinfo import ZoneInfo from datetime import datetime return int(datetime.now(ZoneInfo(tz_str)).utcoffset().total_seconds()) except Exception: import settings as _s return _s.get_host_utc_offset() @bp.route('/action/accountcreate/form_create', methods=['POST']) @auth.require_level('nothing') def form_create(): 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}') account = config_utils.get_account_by_email(email) 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).decode('utf-8') code = f'{secrets.randbelow(1000000):06d}' expires_ts = int(time.time()) + CODE_TTL_SECS tz_offset = _tz_to_offset_seconds(tz) 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}') try: con = config_utils.open_accounts_db() con.execute( '''INSERT OR REPLACE INTO pending_verifications (email, hashed_password, tz_offset_seconds, code, expires_ts) VALUES (?,?,?,?,?)''', (account['email_address'].lower(), hashed, tz_offset, code, expires_ts) ) con.commit() con.close() except Exception as exc: flash(f'Could not store verification: {exc}', 'error') return redirect(f'/{_PAGE}') session['pending_verify_email'] = account['email_address'] return redirect('/accountverifyemail')