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 mod_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) src_raw = request.form.get('src_ip_or_subnet', '').strip() dst_raw = request.form.get('dst_ip_or_subnet', '').strip() dest_port_start_raw = request.form.get('dest_port_start', '').strip() dest_port_end_raw = request.form.get('dest_port_end', '').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 src_raw: flash('The configuration has not been saved because a source IP or subnet is required.', 'error') return None, True src = validate.ip_or_cidr(src_raw) if not src: flash(f'The configuration has not been saved because "{src_raw}" is not a valid IP address or subnet.', 'error') return None, True if not dst_raw: flash('The configuration has not been saved because a destination IP or subnet is required.', 'error') return None, True dst = validate.ip_or_cidr(dst_raw) if not dst: flash(f'The configuration has not been saved because "{dst_raw}" is not a valid IP address or subnet.', 'error') return None, True dest_port_start = '' if dest_port_start_raw: dest_port_start = validate.port(dest_port_start_raw) if not dest_port_start: flash(f'The configuration has not been saved because "{dest_port_start_raw}" is not a valid port number (1-65535).', 'error') return None, True dest_port_end = '' if dest_port_end_raw: dest_port_end = validate.port(dest_port_end_raw) if not dest_port_end: flash(f'The configuration has not been saved because "{dest_port_end_raw}" is not a valid port number (1-65535).', 'error') return None, True if dest_port_start and dest_port_end and int(dest_port_start) > int(dest_port_end): flash('Port range min must not be greater than max.', 'error') return None, True return { 'description': description, 'protocol': protocol, 'src_ip_or_subnet': src, 'dst_ip_or_subnet': dst, 'dest_port_start': dest_port_start, 'dest_port_end': dest_port_end, 'enabled': True, }, None @bp.route('/action/intervlan/addexception_add', methods=['POST']) @require_level('administrator') def addexception_add(): entry, err = _parse_entry() if err: return redirect(f'/{_PAGE}') if not _hash_ok(): return redirect(f'/{_PAGE}') cfg = load_config() cfg.setdefault('inter_vlan_exceptions', []).append(entry) errors = validate.validate_config(cfg) if errors: for msg in errors: flash(msg, 'error') return redirect(f'/{_PAGE}') src = entry.get('src_ip_or_subnet', '') changes = diff_fields(None, entry) flash(record_group(cfg, 'inter_vlan_exceptions', 'src_ip_or_subnet', src, changes, 'core apply'), 'success') return redirect(f'/{_PAGE}') @bp.route('/action/intervlan/table_toggle', methods=['POST']) @require_level('administrator') def table_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('inter_vlan_exceptions', []) 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}') src = items[idx].get('src_ip_or_subnet', '') changes = diff_fields(before, items[idx]) flash(record_group(cfg, 'inter_vlan_exceptions', 'src_ip_or_subnet', src, changes, 'core apply'), 'success') return redirect(f'/{_PAGE}') @bp.route('/action/intervlan/table_edit', methods=['POST']) @require_level('administrator') def table_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('inter_vlan_exceptions', []) if idx < 0 or idx >= len(items): flash('Entry not found.', 'error') return redirect(f'/{_PAGE}') before = copy.deepcopy(items[idx]) items[idx] = entry items[idx]['enabled'] = request.form.get('enabled') == 'on' errors = validate.validate_config(cfg) if errors: for msg in errors: flash(msg, 'error') return redirect(f'/{_PAGE}') src = items[idx].get('src_ip_or_subnet', '') changes = diff_fields(before, items[idx]) flash(record_group(cfg, 'inter_vlan_exceptions', 'src_ip_or_subnet', src, changes, 'core apply'), 'success') return redirect(f'/{_PAGE}') @bp.route('/action/intervlan/table_delete', methods=['POST']) @require_level('administrator') def table_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('inter_vlan_exceptions', []) 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}') src = removed.get('src_ip_or_subnet', '') changes = diff_fields(removed, None) flash(record_group(cfg, 'inter_vlan_exceptions', 'src_ip_or_subnet', src, changes, 'core apply'), 'success') return redirect(f'/{_PAGE}')