Development

docker/routlin-dash/app/pages/portwrangling/action.py

@@ -0,0 +1,189 @@
+from pathlib import Path
+import copy
+
+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__)
+
+_VALID_PROTOS_STR = ', '.join(sorted(validate.VALID_PROTOCOLS))
+
+
+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 _parse_entry():
+    description   = sanitize.text(request.form.get('description', ''))
+    protocol      = sanitize.filtervalue(request.form.get('protocol', ''), validate.VALID_PROTOCOLS)
+    dest_port_raw = request.form.get('dest_port', '').strip()
+    redirect_raw  = request.form.get('redirect_to', '').strip()
+
+    if not protocol:
+        flash(f'The configuration has not been saved because the protocol is invalid. '
+              f'Accepted values: {_VALID_PROTOS_STR}.', 'error')
+        return None, True
+
+    if not dest_port_raw:
+        flash('The configuration has not been saved because the destination port is required.', 'error')
+        return None, True
+    dest_port = validate.port(dest_port_raw)
+    if not dest_port:
+        flash(f'The configuration has not been saved because "{dest_port_raw}" is not a valid port number (1-65535).', 'error')
+        return None, True
+
+    if not redirect_raw:
+        flash('The configuration has not been saved because the redirect IP address is required.', 'error')
+        return None, True
+    redirect_to = validate.ip(redirect_raw)
+    if not redirect_to:
+        flash(f'The configuration has not been saved because "{redirect_raw}" is not a valid IP address.', 'error')
+        return None, True
+
+    return {
+        'description': description,
+        'protocol':    protocol,
+        'dest_port':   dest_port,
+        'redirect_to': redirect_to,
+        'enabled':     True,
+    }, None
+
+
+@bp.route('/action/portwrangling/addrule_add', methods=['POST'])
+@require_level('administrator')
+def addrule_add():
+    vlan_name = sanitize.name(request.form.get('vlan_name', ''))
+    entry, err = _parse_entry()
+    if err:
+        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 _hash_ok():
+        return redirect(f'/{_PAGE}')
+
+    cfg   = load_config()
+    vlan  = next((v for v in cfg.get('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}')
+
+    entry['vlan'] = vlan_name
+    cfg.setdefault('port_wrangling', []).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, 'port_wrangling', 'dest_port', entry['dest_port'], changes, 'core apply'), 'success')
+    return redirect(f'/{_PAGE}')
+
+
+@bp.route('/action/portwrangling/rules_toggle', methods=['POST'])
+@require_level('administrator')
+def rules_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()
+    items = cfg.get('port_wrangling', [])
+    if idx < 0 or idx >= len(items):
+        flash('Entry not found.', 'error')
+        return redirect(f'/{_PAGE}')
+
+    old_enabled        = items[idx].get('enabled', True)
+    before             = copy.deepcopy(items[idx])
+    items[idx]['enabled'] = not old_enabled
+    errors = validate.validate_config(cfg)
+    if errors:
+        for msg in errors:
+            flash(msg, 'error')
+        return redirect(f'/{_PAGE}')
+
+    changes = diff_fields(before, items[idx])
+    flash(record_group(cfg, 'port_wrangling', 'dest_port', items[idx].get('dest_port', ''), changes, 'core apply'), 'success')
+    return redirect(f'/{_PAGE}')
+
+
+@bp.route('/action/portwrangling/rules_edit', methods=['POST'])
+@require_level('administrator')
+def rules_edit():
+    idx = _row_index()
+    if idx is None:
+        flash('Invalid request.', 'error')
+        return redirect(f'/{_PAGE}')
+
+    entry, err = _parse_entry()
+    if err:
+        return redirect(f'/{_PAGE}')
+    if not _hash_ok():
+        return redirect(f'/{_PAGE}')
+
+    cfg   = load_config()
+    items = cfg.get('port_wrangling', [])
+    if idx < 0 or idx >= len(items):
+        flash('Entry not found.', 'error')
+        return redirect(f'/{_PAGE}')
+
+    before     = copy.deepcopy(items[idx])
+    entry['vlan']    = items[idx].get('vlan', '')
+    entry['enabled'] = request.form.get('enabled') == 'on'
+    items[idx] = entry
+    errors = validate.validate_config(cfg)
+    if errors:
+        for msg in errors:
+            flash(msg, 'error')
+        return redirect(f'/{_PAGE}')
+
+    changes = diff_fields(before, items[idx])
+    flash(record_group(cfg, 'port_wrangling', 'dest_port', items[idx].get('dest_port', ''), changes, 'core apply'), 'success')
+    return redirect(f'/{_PAGE}')
+
+
+@bp.route('/action/portwrangling/rules_delete', methods=['POST'])
+@require_level('administrator')
+def rules_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()
+    items = cfg.get('port_wrangling', [])
+    if idx < 0 or idx >= len(items):
+        flash('Entry not found.', 'error')
+        return redirect(f'/{_PAGE}')
+
+    removed = items.pop(idx)
+    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, 'port_wrangling', 'dest_port', removed.get('dest_port', ''), changes, 'core apply'), 'success')
+    return redirect(f'/{_PAGE}')