butter-portal/app/routes.py

import io
import re
import socket
import subprocess

from app import app
from flask import render_template, flash, redirect, url_for, send_file, send_from_directory
from app.forms import LoginForm, SettingsForm, Step1Form, Step2Form, Step3Form, Step4Form
from flask_login import login_user, current_user, logout_user, login_required
import sqlalchemy as sa
from app import db
from app.models import User, Setting
import os
import json
from flask_babel import _
import base64
import random
from wordfreq import top_n_list
import secrets
import string
import glob
import time
import qrcode
from flask_babel import lazy_gettext as _l
from install_madmail import run_madmail_installer

CHANGES_REQUIRING_RESTART = ['wifi_password', 'ssid', 'enable_access_point', 'enable_chat', 'enable_delta_chat', 'butterbox_hostname', 'ssh_access_settings', 'root_account_settings', 'root_password']
RASPAP_INSTALLED = os.path.exists("/var/www/html/raspap")

def resolve_butterbox_ip() -> str:
    """Best-effort lookup of the address clients should use to reach this box.

    ``BUTTERBOX_DEFAULT_IP`` is only correct while the box serves its own
    access point. When it's joined to another network its address differs, so
    try to discover the real one and only fall back to the configured default.
    """
    hostname = get_setting('butterbox_hostname') or app.config['BUTTERBOX_HOSTNAME']

    # 1. Resolve the box's own mDNS/.local name. This is the same name the
    #    DeltaChat link uses for the mail host, so avahi/nss-mdns is present.
    try:
        ip = socket.gethostbyname(f"{hostname}.local")
        if ip and not ip.startswith("127."):
            return ip
    except OSError:
        pass

    # 2. Fall back to the address of the primary outbound interface. The
    #    connect() picks a route without sending any packets.
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
            s.connect(("8.8.8.8", 80))
            ip = s.getsockname()[0]
            if ip and not ip.startswith("127."):
                return ip
    except OSError:
        pass

    # 3. Give up and use the configured access-point default.
    return app.config['BUTTERBOX_DEFAULT_IP']


def gen_username() -> str:
    words = top_n_list("en", 5000)
    prefix = random.randint(1000, 9999)
    word1 = re.sub(r'[^a-zA-Z ]', '', random.choice(words))
    word2 = re.sub(r'[^a-zA-Z ]', '', random.choice(words))
    print(word1, word2)
    return f"{word1}{word2}{prefix}"

def gen_password() -> str:
    characters = string.ascii_letters + string.digits
    password = ''.join(secrets.choice(characters) for i in range(20))
    return password
def get_file_suffix(path: str) -> str:
    return os.path.splitext(path)[1].strip('.') # in case of multiple exts, like .tar.gz, it only splits on the last one

def get_file_icon_url(path: str) -> str:
    if os.path.isdir(path):
        return url_for("static", filename=f"images/extension-icons/directory.svg")
    suffix = get_file_suffix(path)
    if suffix:
        if suffix in ['apk', 'deb', 'dmg', 'exe', 'jpg', 'mp3', 'pbf', 'pdf', 'png',
                      'ppt', 'pptx']:
            return url_for("static", filename=f"images/extension-icons/ext-{suffix}.svg")
    return url_for("static", filename=f"images/extension-icons/ext-unknown.svg")

def get_last_modified(path: str) -> str:
    mod_time = time.ctime(os.path.getmtime(path))
    #print(mod_time)
    #dt = datetime.strptime(mod_time, "%a %b %d %H:%M:%S %Y")
    #formatted_time = dt.strftime("%Y-%m-%d %H:%M")
    return mod_time

def get_files_in_path(path: str):
    file_list = []
    if os.path.exists(path):
        list_of_files = glob.glob(path + "/*", recursive=True)
        file_list = [
            {"name": x.replace(path, "").strip("/"), "path": x, "is_file": os.path.isfile(x),
             "is_dir": os.path.isdir(x),
             "filetype": get_file_suffix(x), "last_modified": get_last_modified(x),
             "icon_url": get_file_icon_url(x),
             "relative_path": x.replace(app.config["BUTTERBOX_USB_PATH"], "")} for x
            in list_of_files]
    return file_list

def get_setting(name) -> str:
    setting = db.session.scalar(sa.select(Setting).where(Setting.key == name))
    return str(setting.value)

def set_setting(name: str, value: str):
    setting = db.session.scalar(sa.select(Setting).where(Setting.key == name))
    setting.value = value
    db.session.add(setting)

def dump_settings(filename: str) -> None:
    settings = db.session.execute(sa.select(Setting)).scalars().all()
    settings_dict = {s.key: s.value for s in settings}
    #settings_dict.pop('admin_password')
    with open(filename, "w") as f:
        json.dump(settings_dict, f, indent=4)
@app.route('/')
@app.route('/index')
def index():
    if get_setting("first_setup") == "true":
        return redirect(url_for('first_setup'))
    enable_chat = get_setting("enable_chat")
    enable_file_viewer = get_setting("enable_file_viewer")
    enable_deltachat = get_setting("enable_deltachat")
    service_array = []
    usb_inserted = False # actual test of whether USB is inserted
    if os.path.exists(app.config["BUTTERBOX_USB_PATH"]):
        usb_inserted = True
    usb_has_maps = False # actual test of whether USB has maps folder
    usb_has_appstore = False # actual test of whether USB has an appstore
    #enable_wifi_sharing = get_setting("enable_wifi_sharing")
    #if enable_wifi_sharing == 'true':
    #    service_array.append({"name": _l("Share Access"), "image": url_for("static", filename="images/share-icon.svg"), "url": url_for("share")})
    if enable_deltachat == 'true':
        service_array.append({"name": _l("Secure Messenger"), "image": url_for("static", filename="images/SecureMessenger.svg"), "url": url_for("messaging") })
    if enable_chat == 'true':
        service_array.append({"name": _l("Local Chat"), "image": url_for("static", filename="images/LocalChat.svg"), "url": f"{app.config["CONVENE_INSTALL_PATH"]}/#/room/join/%23public%3abutterbox.local"})
    if enable_file_viewer == 'true' and usb_has_appstore:
        service_array.append({"name": _l("Apps"), "image": url_for("static", filename="images/Apps.svg")})
    if enable_file_viewer == 'true' and usb_has_maps:
        service_array.append({"name": _l("Maps"), "image": url_for("static", filename="images/Maps.svg")})
    if enable_file_viewer == 'true':
        name = _l("Files")
        if not usb_inserted:
            name = _l("Insert USB to browse files")
        service_array.append({
                             "name": name,
                             "image": url_for("static", filename="images/Files.svg"),
                             "url": url_for("files", path=""),})
    return render_template('index.html', title='Home', get_setting=get_setting, services=service_array)

@app.route('/files/', defaults={'path': ''})
@app.route('/files/<path:path>')
def files(path):
    base_path = app.config["BUTTERBOX_USB_PATH"]
    if not os.path.exists(base_path):
        return redirect(url_for('index'))
    current_path = os.path.join(base_path, path)
    render_files = []
    if os.path.exists(current_path):
        render_files = get_files_in_path(current_path)
        filenames = [x['name'] for x in render_files]
        if 'index.html' in filenames:
            index_path = render_files[filenames.index('index.html')]['path']
            return redirect(url_for('serve_file', filepath=index_path))
    else:
        return redirect(url_for('files', path=""))
    return render_template('usb-file-viewer.html', title='File Viewer', current_path=current_path, render_files=render_files, get_setting=get_setting)

@app.route('/serve_file/<path:filepath>')
def serve_file(filepath):
    if not filepath.startswith(app.config["BUTTERBOX_USB_PATH"]):
        redirect(url_for('files', path=""))
    return send_file("/" + filepath)

@app.route('/login', methods=['GET', 'POST'])
def login():
    if current_user.is_authenticated:
        return redirect(url_for('admin'))
    form = LoginForm()
    form.remember_me.data = False
    if form.validate_on_submit():
        user = db.session.scalar(sa.select(User).where(User.username == form.username.data))
        if user is None or not user.check_password(form.password.data):
            flash(_('Invalid username or password'))
            return redirect(url_for('login'))
        login_user(user)
        return redirect(url_for('admin'))
    return render_template('login.html', title='Sign in', form=form, get_setting=get_setting)


@app.route('/first_setup')
def first_setup():
    if get_setting("first_setup") == "true":
        return render_template('first_setup_main_page.html', get_setting=get_setting)
    return redirect(url_for('admin'))

@app.route('/admin_setup')
def admin_setup():
    if get_setting("first_setup") == "true":
        return render_template('admin_setup.html', get_setting=get_setting)
    return redirect(url_for('admin'))

@app.route('/step1', methods=['GET', 'POST'])
def step1():
    form = Step1Form()
    step1_settings = ['enable_chat', 'enable_deltachat', 'enable_file_viewer']
    if not form.is_submitted():
        for s in step1_settings:
            getattr(form, s).data = (get_setting(s) == "true")
    if form.validate_on_submit():
        if form.submit.data:
            for s in step1_settings:
                setting_value = getattr(form, s).data
                setting_value = str(setting_value).lower()
                set_setting(s, setting_value)
            db.session.commit()
            return redirect(url_for('step2'))
    if get_setting("first_setup") == "true":
        return render_template('step1.html', form=form, get_setting=get_setting)
    return redirect(url_for('admin'))


@app.route('/step2', methods=['GET', 'POST'])
def step2():
    form = Step2Form()
    step2_settings = ['butterbox_hostname', 'butterbox_name']
    if not form.is_submitted():
        for s in step2_settings:
            getattr(form, s).data = get_setting(s)
    if form.validate_on_submit():
        if form.submit.data:
            for s in step2_settings:
                setting_value = getattr(form, s).data
                if s == 'butterbox_hostname':
                    setting_value = setting_value.lower().replace(" ", "")
                set_setting(s, setting_value)
            new_logo = form.butterbox_logo.data
            if new_logo.filename:
                logo_stream = form.butterbox_logo.data.stream
                b64_logo = base64.b64encode(logo_stream.read()).decode('utf-8')
                file_mimetype = form.butterbox_logo.data.mimetype
                new_value = f"data:{file_mimetype};base64,{b64_logo}"
                existing_value = get_setting('butterbox_logo')
                if new_value != existing_value:
                    set_setting('butterbox_logo', new_value)
            db.session.commit()
            linux_date_arg = str(form.butterbox_date.data).replace(" ", "T") + "Z"
            output = subprocess.run(["sudo", "/usr/bin/date", "-s", linux_date_arg], capture_output=True, text=True)
            if output.returncode != 0:
                flash(f"Could not set date. Please set date manually: {output}.", category="error")
            return redirect(url_for('step3'))

    if get_setting("first_setup") == "true":
        return render_template('step2.html', form=form, get_setting=get_setting)
    return redirect(url_for('admin'))
@app.route('/step3', methods=['GET', 'POST'])
def step3():
    form = Step3Form()
    step3_bool_settings = ['enable_wifi_sharing', 'enable_access_point']
    step3_settings = ['ssid', 'wifi_password']
    if not form.is_submitted():
        for s in step3_bool_settings:
            getattr(form, s).data = (get_setting(s) == "true")
        for s in step3_settings:
            getattr(form, s).data = get_setting(s)
    if form.validate_on_submit():
        if form.submit.data:
            for s in (step3_settings + step3_bool_settings):
                setting_value = getattr(form, s).data
                if s in step3_bool_settings:
                    setting_value = str(setting_value).lower()
                set_setting(s, setting_value)
            db.session.commit()
            return redirect(url_for('step4'))
    if get_setting("first_setup") == "true":
        return render_template('step3.html', raspap_installed=RASPAP_INSTALLED, form=form, get_setting=get_setting)
    return redirect(url_for('admin'))

@app.route('/step4', methods=['GET', 'POST'])
def step4():
    form = Step4Form()
    step4_settings = ['root_account_settings', 'ssh_access_settings', 'root_password', 'admin_password']
    if not form.is_submitted():
        for s in step4_settings:
            getattr(form, s).data = get_setting(s)
    if form.validate_on_submit():
        if form.submit.data:
            new_admin_password = form.admin_password.data
            if new_admin_password:
                admin_user = db.session.scalar(sa.select(User).where(User.username == 'admin'))
                if not admin_user.check_password(new_admin_password):
                    admin_user.set_password(new_admin_password)
                    db.session.add(admin_user)
            step4_settings.remove('admin_password')
            for s in step4_settings:
                setting_value = getattr(form, s).data
                set_setting(s, setting_value)
            set_setting('first_setup', "false")
            dump_settings("settings.txt")
            if get_setting("enable_deltachat") == "true":
                run_madmail_installer()
            db.session.commit()
            return redirect(url_for('setup_complete'))
    if get_setting("first_setup") == "true":
        return render_template('step4.html', form=form, get_setting=get_setting)
    return redirect(url_for('admin'))

@app.route('/setup_complete')
def setup_complete():
    if get_setting("first_setup"):
        return render_template('setup_complete.html', get_setting=get_setting)
    return redirect(url_for('admin'))


@app.route('/logout')
def logout():
    logout_user()
    return redirect(url_for('admin_setup'))
@app.route('/admin', methods=['GET'])
def admin():
    if get_setting("first_setup") == "true":
        return redirect(url_for('first_setup'))
    return redirect(url_for('admin_settings'))

@app.route('/admin_settings', methods=['GET', 'POST'])
@login_required
def admin_settings():
    form = SettingsForm()
    populate_settings = ['butterbox_name', 'wifi_password', 'ssid', 'root_account_settings', 'ssh_access_settings', 'root_password', 'admin_password']
    bool_settings = ['enable_access_point','enable_file_viewer', 'enable_chat', 'enable_deltachat', 'enable_wifi_sharing']
    populate_settings.extend(bool_settings)
    if not form.is_submitted():
        for s in populate_settings:
            if s in bool_settings:
                getattr(form, s).data = (get_setting(s) == "true")
            else:
                getattr(form, s).data = get_setting(s)
    non_admin_settings_changed = False
    if not form.validate_on_submit():
        print(form.errors)
    if form.validate_on_submit():
        if form.submit.data:
            deltachat_newly_enabled = False
            for s in populate_settings:
                new_value = getattr(form, s).data
                if s in bool_settings:
                    new_value = str(new_value).lower() # all settings are str fow now
                if s == 'butterbox_hostname':
                    new_value = new_value.lower().replace(" ", "")
                existing_value = get_setting(s)
                if new_value != existing_value:
                    print(f"New value was changed for {s}. Existing value was {existing_value}, new value was {new_value}")
                    set_setting(s, new_value)
                    non_admin_settings_changed = True
                    if s in CHANGES_REQUIRING_RESTART:
                        app.config['SETTINGS_CHANGED'] = True
                    if s == 'enable_deltachat' and new_value == "true":
                        deltachat_newly_enabled = True
            if deltachat_newly_enabled:
                run_madmail_installer()

            new_logo = form.butterbox_logo.data
            if new_logo.filename:
                logo_stream = form.butterbox_logo.data.stream
                b64_logo = base64.b64encode(logo_stream.read()).decode('utf-8')
                file_mimetype = form.butterbox_logo.data.mimetype
                new_value = f"data:{file_mimetype};base64,{b64_logo}"
                existing_value = get_setting('butterbox_logo')
                if new_value != existing_value:
                    set_setting('butterbox_logo', new_value)
                    non_admin_settings_changed = True
            new_admin_password = form.admin_password.data
            if new_admin_password:
                admin_user = db.session.scalar(sa.select(User).where(User.username == 'admin'))
                if not admin_user.check_password(new_admin_password):
                    admin_user.set_password(new_admin_password)
                    db.session.add(admin_user)
                    non_admin_settings_changed = True
            if app.config['SETTINGS_CHANGED']:
                flash(_("⚠️ Some settings may not fully take effect until the Butter Box restarts. Click 'Apply Changes' to restart."))
            else:
                if non_admin_settings_changed:
                    flash(
                        _("Settings successfully changed."))
            db.session.commit()

        if form.apply_changes.data:
            set_setting('apply_changes', "true")
            dump_settings("settings.txt")
            flash(_("⚠️ Changes applied! If needed, the system will restart. This may take up to two minutes."))
    return render_template('admin.html', raspap_installed=RASPAP_INSTALLED, get_setting=get_setting, form=form)


@app.route('/messaging', methods=['GET', 'POST'])
def messaging():
    return render_template('messaging.html', get_setting=get_setting)

@app.route("/deltachat_credentials", methods=["POST"])
def deltachat_credentials():
    ip = resolve_butterbox_ip()
    username = gen_username()
    password = gen_password()
    hostname = f"{get_setting('butterbox_hostname')}.local"
    dclink = f"dclogin:{username}@{hostname}/?p={password}&v=1&ih={ip}&ip=993&sh={ip}&sp=465&ic=3&ss=default"
    img = qrcode.make(dclink)
    file_object = io.BytesIO()
    img.save(file_object, 'PNG')
    base64img = "data:image/png;base64," + base64.b64encode(file_object.getvalue()).decode('ascii')
    return render_template('deltachat_creds.html',dclink=dclink, base64img=base64img, get_setting=get_setting)

@app.route('/share')
def share():
    display_wifi_password = False
    wifi_password = get_setting("wifi_password")
    if wifi_password:
        display_wifi_password = True
    wifi_ssid = get_setting("ssid")
    wifi_encryption_type = "WPA2"
    img = qrcode.make(f"WIFI:T:{wifi_encryption_type};S:{wifi_ssid};P:{wifi_password};;")
    img.save("app/static/images/wifi_qr_code.png")
    return render_template('share.html', get_setting=get_setting, display_wifi_password=display_wifi_password)