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

from pathlib import Path
import copy
import os

from flask import Blueprint, request, redirect, flash
from auth import require_level
from config_utils import load_config, record_group, diff_fields, verify_config_hash, queued_msg, queue_command
import sanitize
import mod_validation as validate

_PAGE = Path(__file__).parent.name

bp = Blueprint(_PAGE, __name__)

_EXCLUDE_PREFIXES = ('lo', 'wg', 'docker', 'br-', 'veth',
                     'tun', 'tap', 'ppp', 'virbr',
                     'podman', 'vnet', 'macvtap', 'fc-')


def _get_system_interfaces():
    try:
        return {
            n for n in os.listdir('/sys/class/net')
            if not n.startswith(_EXCLUDE_PREFIXES)
            and os.path.exists(f'/sys/class/net/{n}/device')
        }
    except Exception:
        return set()


def _valid_interface(name):
    return name in _get_system_interfaces()


@bp.route('/action/physicalinterfaces/physicalinterface_save', methods=['POST'])
@require_level('administrator')
def physicalinterface_save():
    wan = sanitize.interface_name(request.form.get('wan_interface', ''))
    lan = sanitize.interface_name(request.form.get('lan_interface', ''))

    if not wan or not lan:
        flash('Both WAN and LAN interfaces are required.', 'error')
        return redirect(f'/{_PAGE}')

    # WAN and LAN must be distinct physical interfaces
    err = validate.check_wan_lan_unique(wan, lan)
    if err:
        flash(err, 'error')
        return redirect(f'/{_PAGE}')

    if not verify_config_hash(request.form.get('config_hash', '')):
        flash('Configuration was modified by another session. Please refresh and try again.', 'error')
        return redirect(f'/{_PAGE}')

    available = _get_system_interfaces()
    # Interfaces must exist on this system (checked against physical-only interface list)
    for iface in (wan, lan):
        err = validate.check_interface_exists(iface, available)
        if err:
            flash(err, 'error')
            return redirect(f'/{_PAGE}')

    cfg    = load_config()
    before = copy.deepcopy(cfg.get('network_interfaces', {}))
    gen    = cfg.setdefault('network_interfaces', {})
    gen['wan_interface'] = wan
    gen['lan_interface'] = lan
    errors = validate.validate_config(cfg)
    if errors:
        for msg in errors:
            flash(msg, 'error')
        return redirect(f'/{_PAGE}')
    changes = diff_fields(before, cfg['network_interfaces'])
    flash(record_group(cfg, 'network_interfaces', None, None, changes, 'core apply'), 'success')
    return redirect(f'/{_PAGE}')


@bp.route('/action/physicalinterfaces/ifaceconfig_apply', methods=['POST'])
@require_level('administrator')
def ifaceconfig_apply():
    if not verify_config_hash(request.form.get('config_hash', '')):
        flash('Configuration was modified by another session. Please refresh and try again.', 'error')
        return redirect(f'/{_PAGE}')

    iface        = sanitize.interface_name(request.form.get('iface', ''))
    mtu          = request.form.get('mtu', '').strip()
    mac          = sanitize.mac(request.form.get('mac', ''))
    original_mtu = request.form.get('original_mtu', '').strip()
    original_mac = sanitize.mac(request.form.get('original_mac', ''))

    if not iface:
        flash('No interface specified.', 'error')
        return redirect(f'/{_PAGE}')

    if not _valid_interface(iface):
        flash(f"Interface '{iface}' does not exist on this system.", 'error')
        return redirect(f'/{_PAGE}')

    mtu_int = None
    if mtu:
        mtu_int = validate.int_range(mtu, 68, 9000)
        if mtu_int is None:
            flash('MTU must be an integer between 68 and 9000.', 'error')
            return redirect(f'/{_PAGE}')

    mac_raw = request.form.get('mac', '').strip()
    if mac_raw and not mac:
        flash('MAC address must be in the format aa:bb:cc:dd:ee:ff.', 'error')
        return redirect(f'/{_PAGE}')

    if not mtu_int and not mac:
        flash('No changes specified.', 'error')
        return redirect(f'/{_PAGE}')

    queued = False
    if mtu_int and str(mtu_int) != original_mtu:
        queue_command(f'mtu {iface} {mtu_int}')
        queued = True
    if mac and mac != original_mac:
        queue_command(f'mac {iface} {mac}')
        queued = True

    if not queued:
        flash('No changes detected.', 'info')
        return redirect(f'/{_PAGE}')

    flash(queued_msg(action_label='Changes queued'), 'success')
    return redirect(f'/{_PAGE}')