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

from pathlib import Path
import copy
import ipaddress

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
import sanitize
import validation as validate

_PAGE = Path(__file__).parent.name

bp = Blueprint(_PAGE, __name__)

def _row_index():
    try:
        return int(request.form.get('row_index', ''))
    except (ValueError, TypeError):
        return None


def _hash_ok():
    if not verify_config_hash(request.form.get('config_hash', '')):
        flash('Configuration was modified by another session. Please refresh and try again.', 'error')
        return False
    return True


def _flat_index_to_vlan_res(vlans, flat_idx):
    pos = 0
    for vi, vlan in enumerate(vlans):
        for ri in range(len(vlan.get('reservations', []))):
            if pos == flat_idx:
                return vi, ri
            pos += 1
    return None, None


def _parse_ip():
    raw = request.form.get('ip', '').strip()
    if not raw:
        return 'dynamic'
    ip = validate.ip(raw)
    if not ip:
        flash(f'The configuration has not been saved because "{raw}" is not a valid IP address.', 'error')
        return None
    return ip


def _check_ip_conflicts(ip, vlan):
    if ip == 'dynamic':
        return None
    dhcp       = vlan.get('dhcp_information', {})
    pool_start = dhcp.get('dynamic_pool_start')
    pool_end   = dhcp.get('dynamic_pool_end')
    if pool_start and pool_end:
        try:
            if (ipaddress.IPv4Address(pool_start) <= ipaddress.IPv4Address(ip)
                    <= ipaddress.IPv4Address(pool_end)):
                return f'{ip} falls within the dynamic pool range ({pool_start}-{pool_end}).'
        except Exception:
            pass
    identity_ips = {s['ip'] for s in vlan.get('server_identities', []) if s.get('ip')}
    if ip in identity_ips:
        return f'{ip} is already assigned as a server identity IP.'
    return None


@bp.route('/action/dhcp/addreservation_add', methods=['POST'])
@require_level('administrator')
def addreservation_add():
    vlan_name     = sanitize.name(request.form.get('vlan_name', ''))
    description   = sanitize.text(request.form.get('description', ''))
    hostname      = validate.domainname(request.form.get('hostname', ''))
    mac           = sanitize.mac(request.form.get('mac', ''))
    ip            = _parse_ip()
    radius_client = 'radius_client' in request.form

    if ip is None:
        return redirect(f'/{_PAGE}')
    if not vlan_name:
        flash('The configuration has not been saved because a VLAN is required.', 'error')
        return redirect(f'/{_PAGE}')
    if not mac:
        flash('The configuration has not been saved because a MAC address is required.', 'error')
        return redirect(f'/{_PAGE}')
    if not _hash_ok():
        return redirect(f'/{_PAGE}')

    cfg  = load_config()
    vlans = cfg.get('vlans', [])
    vlan  = next((v for v in vlans if v.get('name') == vlan_name), None)
    if vlan is None:
        flash(f'The configuration has not been saved because VLAN "{vlan_name}" was not found.', 'error')
        return redirect(f'/{_PAGE}')

    conflict = _check_ip_conflicts(ip, vlan)
    if conflict:
        flash(f'The configuration has not been saved because {conflict}', 'error')
        return redirect(f'/{_PAGE}')

    entry = {
        'description':   description,
        'hostname':      hostname,
        'mac':           mac,
        'ip':            ip,
        'radius_client': radius_client,
        'enabled':       True,
    }
    vlan.setdefault('reservations', []).append(entry)
    errors = validate.validate_config(cfg)
    if errors:
        for msg in errors:
            flash(msg, 'error')
        return redirect(f'/{_PAGE}')

    changes = diff_fields(None, entry)
    flash(record_group(cfg, f'vlans[name={vlan_name}].reservations', 'mac', mac, changes, 'core apply'), 'success')
    return redirect(f'/{_PAGE}')


@bp.route('/action/dhcp/reservations_toggle', methods=['POST'])
@require_level('administrator')
def reservations_toggle():
    idx = _row_index()
    if idx is None:
        flash('Invalid request.', 'error')
        return redirect(f'/{_PAGE}')
    if not _hash_ok():
        return redirect(f'/{_PAGE}')

    cfg  = load_config()
    vlans = cfg.get('vlans', [])
    vi, ri = _flat_index_to_vlan_res(vlans, idx)
    if vi is None:
        flash('Entry not found.', 'error')
        return redirect(f'/{_PAGE}')

    res         = vlans[vi]['reservations'][ri]
    old_enabled = res.get('enabled', True)
    before      = copy.deepcopy(res)
    res['enabled'] = not old_enabled
    errors = validate.validate_config(cfg)
    if errors:
        for msg in errors:
            flash(msg, 'error')
        return redirect(f'/{_PAGE}')

    vlan_name = vlans[vi]['name']
    changes = diff_fields(before, res)
    flash(record_group(cfg, f'vlans[name={vlan_name}].reservations', 'mac', res['mac'], changes, 'core apply'), 'success')
    return redirect(f'/{_PAGE}')


@bp.route('/action/dhcp/reservations_edit', methods=['POST'])
@require_level('administrator')
def reservations_edit():
    idx = _row_index()
    if idx is None:
        flash('Invalid request.', 'error')
        return redirect(f'/{_PAGE}')

    description   = sanitize.text(request.form.get('description', ''))
    hostname      = validate.domainname(request.form.get('hostname', ''))
    mac           = sanitize.mac(request.form.get('mac', ''))
    ip            = _parse_ip()
    radius_client = 'radius_client' in request.form

    if ip is None:
        return redirect(f'/{_PAGE}')
    if not mac:
        flash('The configuration has not been saved because a MAC address is required.', 'error')
        return redirect(f'/{_PAGE}')
    if not _hash_ok():
        return redirect(f'/{_PAGE}')

    cfg  = load_config()
    vlans = cfg.get('vlans', [])
    vi, ri = _flat_index_to_vlan_res(vlans, idx)
    if vi is None:
        flash('Entry not found.', 'error')
        return redirect(f'/{_PAGE}')

    conflict = _check_ip_conflicts(ip, vlans[vi])
    if conflict:
        flash(f'The configuration has not been saved because {conflict}', 'error')
        return redirect(f'/{_PAGE}')

    res    = vlans[vi]['reservations'][ri]
    before = copy.deepcopy(res)
    res.update({
        'description':   description,
        'hostname':      hostname,
        'mac':           mac,
        'ip':            ip,
        'radius_client': radius_client,
        'enabled':       'enabled' in request.form,
    })
    errors = validate.validate_config(cfg)
    if errors:
        for msg in errors:
            flash(msg, 'error')
        return redirect(f'/{_PAGE}')

    vlan_name = vlans[vi]['name']
    changes = diff_fields(before, res)
    flash(record_group(cfg, f'vlans[name={vlan_name}].reservations', 'mac', mac, changes, 'core apply'), 'success')
    return redirect(f'/{_PAGE}')


@bp.route('/action/dhcp/reservations_delete', methods=['POST'])
@require_level('administrator')
def reservations_delete():
    idx = _row_index()
    if idx is None:
        flash('Invalid request.', 'error')
        return redirect(f'/{_PAGE}')
    if not _hash_ok():
        return redirect(f'/{_PAGE}')

    cfg  = load_config()
    vlans = cfg.get('vlans', [])
    vi, ri = _flat_index_to_vlan_res(vlans, idx)
    if vi is None:
        flash('Entry not found.', 'error')
        return redirect(f'/{_PAGE}')

    vlan_name = vlans[vi]['name']
    removed   = vlans[vi]['reservations'].pop(ri)
    errors = validate.validate_config(cfg)
    if errors:
        for msg in errors:
            flash(msg, 'error')
        return redirect(f'/{_PAGE}')

    changes = diff_fields(removed, None)
    flash(record_group(cfg, f'vlans[name={vlan_name}].reservations', 'mac', removed['mac'], changes, 'core apply'), 'success')
    return redirect(f'/{_PAGE}')