Development

This commit is contained in:
Matthew Grotke 2026-05-31 18:24:04 -04:00
parent 916d238602
commit 6c3abca58c
4 changed files with 208 additions and 303 deletions

View file

@ -28,18 +28,25 @@ def _hash_ok():
return True
@bp.route('/action/networklayout/addvlan_add', methods=['POST'])
@bp.route('/action/networklayout/vlans_addedit', methods=['POST'])
@require_level('administrator')
def addvlan_add():
name = sanitize.name(request.form.get('name', ''))
vlan_id = sanitize.vlan_id(request.form.get('vlan_id', ''))
is_vpn = 'is_vpn' in request.form
subnet = sanitize.ip(request.form.get('subnet', ''))
subnet_mask = sanitize.subnet_mask(request.form.get('subnet_mask', ''))
radius_default = 'radius_default' in request.form
mdns_reflection = 'mdns_reflection' in request.form
def vlans_addedit():
_ri = request.form.get('row_index', '').strip()
try:
edit_idx = int(_ri)
is_edit = True
except (ValueError, TypeError):
edit_idx = None
is_edit = False
name = sanitize.name(request.form.get('name', ''))
vlan_id = sanitize.vlan_id(request.form.get('vlan_id', ''))
subnet = sanitize.ip(request.form.get('subnet', ''))
subnet_mask = sanitize.subnet_mask(request.form.get('subnet_mask', ''))
radius_default = 'radius_default' in request.form
mdns_reflection = 'mdns_reflection' in request.form
dnsmasq_log_queries = 'dnsmasq_log_queries' in request.form
use_blocklists = sanitize.filterlist(
use_blocklists = sanitize.filterlist(
request.form.getlist('use_blocklists'),
{b.get('name') for b in load_config().get('dns_blocking', {}).get('blocklists', [])},
)
@ -56,9 +63,6 @@ def addvlan_add():
if subnet_mask is None:
flash('Invalid subnet prefix (must be 1-30).', 'error')
return redirect(f'/{_PAGE}')
if is_vpn and mdns_reflection:
flash('mDNS reflection is not supported on VPN VLANs.', 'error')
return redirect(f'/{_PAGE}')
_vlan_net = ipaddress.IPv4Network(f'{subnet}/{subnet_mask}', strict=False)
if ipaddress.IPv4Address(subnet) != _vlan_net.network_address:
@ -79,29 +83,29 @@ def addvlan_add():
new_identities = []
for raw in raw_identities:
ip_clean = sanitize.ip(str(raw.get('ip', '')))
if not ip_clean:
flash('Invalid IP address in identity.', 'error')
ip_clean = sanitize.ip(str(raw.get('ip', '')))
if not ip_clean:
flash('Invalid IP address in identity.', 'error')
return redirect(f'/{_PAGE}')
_addr = ipaddress.IPv4Address(ip_clean)
if _addr not in _vlan_net:
flash(f"Identity IP '{ip_clean}' is not in the VLAN subnet ({subnet}/{subnet_mask}).", 'error')
return redirect(f'/{_PAGE}')
if _addr == _vlan_net.network_address or _addr == _vlan_net.broadcast_address:
flash(f"Identity IP '{ip_clean}' cannot be the network or broadcast address.", 'error')
return redirect(f'/{_PAGE}')
ident = {'ip': ip_clean}
desc = str(raw.get('description', '')).strip()
if desc:
ident['description'] = desc
hostname_raw = str(raw.get('hostname', '')).strip()
if hostname_raw:
clean_hostname = sanitize.hostname(hostname_raw)
if clean_hostname is None:
flash(f"'{hostname_raw}' is not a valid hostname.", 'error')
return redirect(f'/{_PAGE}')
_addr = ipaddress.IPv4Address(ip_clean)
if _addr not in _vlan_net:
flash(f"Identity IP '{ip_clean}' is not in the VLAN subnet ({subnet}/{subnet_mask}).", 'error')
return redirect(f'/{_PAGE}')
if _addr == _vlan_net.network_address or _addr == _vlan_net.broadcast_address:
flash(f"Identity IP '{ip_clean}' cannot be the network or broadcast address.", 'error')
return redirect(f'/{_PAGE}')
ident = {'ip': ip_clean}
desc = str(raw.get('description', '')).strip()
if desc:
ident['description'] = desc
hostname_raw = str(raw.get('hostname', '')).strip()
if hostname_raw:
clean_hostname = sanitize.hostname(hostname_raw)
if clean_hostname is None:
flash(f"'{hostname_raw}' is not a valid hostname.", 'error')
return redirect(f'/{_PAGE}')
ident['hostname'] = clean_hostname
new_identities.append(ident)
ident['hostname'] = clean_hostname
new_identities.append(ident)
identity_ips = [ident['ip'] for ident in new_identities]
@ -109,7 +113,7 @@ def addvlan_add():
if gateway_raw and gateway_raw not in identity_ips:
flash(f"Gateway '{gateway_raw}' must match one of the server identity IPs.", 'error')
return redirect(f'/{_PAGE}')
inferred_gw = (min(identity_ips, key=lambda ip: int(ip.split('.')[-1])) if identity_ips else '')
inferred_gw = (min(identity_ips, key=lambda ip: int(ip.split('.')[-1])) if identity_ips else '')
new_stored_gw = gateway_raw if (gateway_raw and gateway_raw != inferred_gw) else ''
dns_override = 'dns_servers_override' in request.form
@ -149,7 +153,7 @@ def addvlan_add():
new_stored_ntp = ntp_ips if ntp_override else []
dhcp_domain_raw = request.form.get('dhcp_domain', '').strip()
dhcp_domain = sanitize.hostname(dhcp_domain_raw) if dhcp_domain_raw else 'local'
dhcp_domain = sanitize.hostname(dhcp_domain_raw) if dhcp_domain_raw else 'local'
if dhcp_domain_raw and dhcp_domain is None:
flash(f"'{dhcp_domain_raw}' is not a valid domain name.", 'error')
return redirect(f'/{_PAGE}')
@ -184,16 +188,6 @@ def addvlan_add():
return redirect(f'/{_PAGE}')
dhcp_lease_time = f'{_lt}{_unit_suffix}'
cfg = load_config()
vlans = cfg.setdefault('vlans', [])
if any(v.get('vlan_id') == vlan_id for v in vlans):
flash(f'VLAN ID {vlan_id} is already in use.', 'error')
return redirect(f'/{_PAGE}')
if radius_default and any(v.get('radius_default') for v in vlans):
flash('Only one VLAN can be the RADIUS default.', 'error')
return redirect(f'/{_PAGE}')
dhcp_info = {}
if dhcp_domain and dhcp_domain != 'local':
dhcp_info['domain'] = dhcp_domain
@ -213,262 +207,109 @@ def addvlan_add():
if dhcp_overrides:
dhcp_info['explicit_overrides'] = dhcp_overrides
entry = {
'name': name,
'vlan_id': vlan_id,
'is_vpn': is_vpn,
'subnet': subnet,
'subnet_mask': subnet_mask,
'dnsmasq_log_queries': dnsmasq_log_queries,
'use_blocklists': use_blocklists,
'radius_default': radius_default,
'mdns_reflection': mdns_reflection,
'server_identities': new_identities,
}
if dhcp_info:
entry['dhcp_information'] = dhcp_info
if is_vpn:
entry['peers'] = []
else:
entry['reservations'] = []
vlans.append(entry)
errors = validate.validate_config(cfg)
if errors:
for msg in errors:
flash(msg, 'error')
return redirect(f'/{_PAGE}')
cfg = load_config()
vlans = cfg.setdefault('vlans', [])
changes = diff_fields(None, entry)
flash(record_group(cfg, 'vlans', 'name', name, changes, 'core apply'), 'success')
return redirect(f'/{_PAGE}')
@bp.route('/action/networklayout/vlans_edit', methods=['POST'])
@require_level('administrator')
def vlans_edit():
idx = _row_index()
if idx is None:
flash('Invalid request.', 'error')
return redirect(f'/{_PAGE}')
name = sanitize.name(request.form.get('name', ''))
vlan_id = sanitize.vlan_id(request.form.get('vlan_id', ''))
subnet = sanitize.ip(request.form.get('subnet', ''))
radius_default = 'radius_default' in request.form
mdns_reflection = 'mdns_reflection' in request.form
dnsmasq_log_queries = 'dnsmasq_log_queries' in request.form
use_blocklists = sanitize.filterlist(
request.form.getlist('use_blocklists'),
{b.get('name') for b in load_config().get('dns_blocking', {}).get('blocklists', [])},
)
identity_ips_raw = [line.strip() for line in request.form.get('server_identity_ips', '').splitlines() if line.strip()]
identity_ips = []
for raw_ip in identity_ips_raw:
clean = sanitize.ip(raw_ip)
if not clean:
flash(f"'{raw_ip}' is not a valid IP address.", 'error')
if is_edit:
if edit_idx < 0 or edit_idx >= len(vlans):
flash('VLAN not found.', 'error')
return redirect(f'/{_PAGE}')
identity_ips.append(clean)
identity_descs = request.form.get('server_identity_descriptions', '').splitlines()
identity_hostnames = request.form.get('server_identity_hostnames', '').splitlines()
subnet_mask_raw = request.form.get('subnet_mask')
if subnet_mask_raw is not None:
subnet_mask = sanitize.subnet_mask(subnet_mask_raw)
if subnet_mask is None:
flash('Invalid subnet prefix (must be 1-30).', 'error')
existing = vlans[edit_idx]
is_vpn = existing.get('is_vpn', False)
if is_vpn and mdns_reflection:
flash('mDNS reflection is not supported on VPN VLANs.', 'error')
return redirect(f'/{_PAGE}')
else:
subnet_mask = None
if not name:
flash('Name is required.', 'error')
return redirect(f'/{_PAGE}')
if vlan_id is None:
flash('VLAN ID must be an integer between 1 and 4094.', 'error')
return redirect(f'/{_PAGE}')
if not subnet:
flash('Subnet IP is required.', 'error')
return redirect(f'/{_PAGE}')
if not _hash_ok():
return redirect(f'/{_PAGE}')
current_id = existing.get('vlan_id')
if current_id == 1 and vlan_id != 1:
flash('VLAN 1 is the physical interface and cannot change its ID.', 'error')
return redirect(f'/{_PAGE}')
cfg = load_config()
vlans = cfg.get('vlans', [])
if idx < 0 or idx >= len(vlans):
flash('VLAN not found.', 'error')
return redirect(f'/{_PAGE}')
if vlan_id != current_id and any(
v.get('vlan_id') == vlan_id for i, v in enumerate(vlans) if i != edit_idx
):
flash(f'VLAN ID {vlan_id} is already in use.', 'error')
return redirect(f'/{_PAGE}')
existing = vlans[idx]
is_vpn = existing.get('is_vpn', False)
final_mask = subnet_mask if subnet_mask is not None else existing.get('subnet_mask', 24)
if radius_default and any(i != edit_idx and v.get('radius_default') for i, v in enumerate(vlans)):
flash('Only one VLAN can be the RADIUS default.', 'error')
return redirect(f'/{_PAGE}')
if is_vpn and mdns_reflection:
flash('mDNS reflection is not supported on VPN VLANs.', 'error')
return redirect(f'/{_PAGE}')
if identity_ips:
_vlan_net = ipaddress.IPv4Network(f'{subnet}/{final_mask}', strict=False)
for _ip in identity_ips:
_addr = ipaddress.IPv4Address(_ip)
if _addr not in _vlan_net:
flash(f"Server identity IP '{_ip}' is not in the VLAN subnet ({subnet}/{final_mask}).", 'error')
return redirect(f'/{_PAGE}')
if _addr == _vlan_net.network_address or _addr == _vlan_net.broadcast_address:
flash(f"Server identity IP '{_ip}' cannot be the network or broadcast address.", 'error')
return redirect(f'/{_PAGE}')
current_id = existing.get('vlan_id')
if current_id == 1 and vlan_id != 1:
flash('VLAN 1 is the physical interface and cannot change its ID.', 'error')
return redirect(f'/{_PAGE}')
if vlan_id != current_id and any(v.get('vlan_id') == vlan_id for i, v in enumerate(vlans) if i != idx):
flash(f'VLAN ID {vlan_id} is already in use.', 'error')
return redirect(f'/{_PAGE}')
if radius_default and any(i != idx and v.get('radius_default') for i, v in enumerate(vlans)):
flash('Only one VLAN can be the RADIUS default.', 'error')
return redirect(f'/{_PAGE}')
old_identities = existing.get('server_identities', [])
new_identities = []
for i, ip in enumerate(identity_ips):
entry = dict(old_identities[i]) if i < len(old_identities) else {}
entry['ip'] = ip
desc = identity_descs[i].strip() if i < len(identity_descs) else ''
if desc:
entry['description'] = desc
before = copy.deepcopy(existing)
existing.update({
'name': name,
'vlan_id': vlan_id,
'subnet': subnet,
'subnet_mask': subnet_mask,
'dnsmasq_log_queries': dnsmasq_log_queries,
'radius_default': radius_default,
'mdns_reflection': mdns_reflection,
'use_blocklists': use_blocklists,
'server_identities': new_identities,
})
if dhcp_info:
existing['dhcp_information'] = dhcp_info
else:
entry.pop('description', None)
hostname_raw = identity_hostnames[i].strip() if i < len(identity_hostnames) else ''
if hostname_raw:
clean_hostname = sanitize.hostname(hostname_raw)
if clean_hostname is None:
flash(f"'{hostname_raw}' is not a valid hostname.", 'error')
return redirect(f'/{_PAGE}')
entry['hostname'] = clean_hostname
existing.pop('dhcp_information', None)
errors = validate.validate_config(cfg)
if errors:
for msg in errors:
flash(msg, 'error')
return redirect(f'/{_PAGE}')
changes = diff_fields(before, existing)
if not changes:
flash('No changes were made.', 'info')
return redirect(f'/{_PAGE}')
flash(record_group(cfg, 'vlans', 'name', name, changes, 'core apply'), 'success')
else:
is_vpn = 'is_vpn' in request.form
if is_vpn and mdns_reflection:
flash('mDNS reflection is not supported on VPN VLANs.', 'error')
return redirect(f'/{_PAGE}')
if any(v.get('vlan_id') == vlan_id for v in vlans):
flash(f'VLAN ID {vlan_id} is already in use.', 'error')
return redirect(f'/{_PAGE}')
if radius_default and any(v.get('radius_default') for v in vlans):
flash('Only one VLAN can be the RADIUS default.', 'error')
return redirect(f'/{_PAGE}')
entry = {
'name': name,
'vlan_id': vlan_id,
'is_vpn': is_vpn,
'subnet': subnet,
'subnet_mask': subnet_mask,
'dnsmasq_log_queries': dnsmasq_log_queries,
'use_blocklists': use_blocklists,
'radius_default': radius_default,
'mdns_reflection': mdns_reflection,
'server_identities': new_identities,
}
if dhcp_info:
entry['dhcp_information'] = dhcp_info
if is_vpn:
entry['peers'] = []
else:
entry.pop('hostname', None)
new_identities.append(entry)
entry['reservations'] = []
vlans.append(entry)
gateway_raw = sanitize.ip(request.form.get('gateway', ''))
if gateway_raw and gateway_raw not in identity_ips:
flash(f"Gateway '{gateway_raw}' must match one of the server identity IPs.", 'error')
return redirect(f'/{_PAGE}')
inferred_gw = (min(identity_ips, key=lambda ip: int(ip.split('.')[-1]))
if identity_ips else '')
new_stored_gw = gateway_raw if (gateway_raw and gateway_raw != inferred_gw) else ''
existing_gw = existing.get('dhcp_information', {}).get('explicit_overrides', {}).get('gateway', '')
dns_override = 'dns_servers_override' in request.form
dns_ips = []
for _line in request.form.get('dns_servers', '').splitlines():
_line = _line.strip()
if not _line:
continue
_clean = sanitize.ip(_line)
if not _clean:
flash(f"'{_line}' is not a valid DNS server IP.", 'error')
errors = validate.validate_config(cfg)
if errors:
for msg in errors:
flash(msg, 'error')
return redirect(f'/{_PAGE}')
dns_ips.append(_clean)
if dns_override and not dns_ips:
flash('At least one DNS server IP is required when override is enabled.', 'error')
return redirect(f'/{_PAGE}')
if dns_override and dns_ips:
_vlan_net = ipaddress.IPv4Network(f'{subnet}/{final_mask}', strict=False)
for _ip in dns_ips:
if ipaddress.IPv4Address(_ip) not in _vlan_net:
flash(f"DNS server '{_ip}' is not in the VLAN subnet ({subnet}/{final_mask}).", 'error')
return redirect(f'/{_PAGE}')
new_stored_dns = dns_ips if dns_override else []
_existing_dns = existing.get('dhcp_information', {}).get('explicit_overrides', {}).get('dns_servers', [])
existing_dns = _existing_dns if isinstance(_existing_dns, list) else ([_existing_dns] if _existing_dns else [])
ntp_override = 'ntp_server_override' in request.form
ntp_ips = []
for _line in request.form.get('ntp_servers', '').splitlines():
_line = _line.strip()
if not _line:
continue
_clean = sanitize.ip(_line)
if not _clean:
flash(f"'{_line}' is not a valid NTP server IP.", 'error')
return redirect(f'/{_PAGE}')
ntp_ips.append(_clean)
if ntp_override and not ntp_ips:
flash('At least one NTP server IP is required when override is enabled.', 'error')
return redirect(f'/{_PAGE}')
if ntp_override and ntp_ips:
_vlan_net = ipaddress.IPv4Network(f'{subnet}/{final_mask}', strict=False)
for _ip in ntp_ips:
if ipaddress.IPv4Address(_ip) not in _vlan_net:
flash(f"NTP server '{_ip}' is not in the VLAN subnet ({subnet}/{final_mask}).", 'error')
return redirect(f'/{_PAGE}')
new_stored_ntp = ntp_ips if ntp_override else []
_existing_ntp = existing.get('dhcp_information', {}).get('explicit_overrides', {}).get('ntp_servers', [])
existing_ntp = _existing_ntp if isinstance(_existing_ntp, list) else ([_existing_ntp] if _existing_ntp else [])
changes = diff_fields(None, entry)
flash(record_group(cfg, 'vlans', 'name', name, changes, 'core apply'), 'success')
_ids_unchanged = (
len(new_identities) == len(old_identities) and
all(
n.get('ip') == o.get('ip') and
n.get('description', '') == o.get('description', '') and
n.get('hostname', '') == o.get('hostname', '')
for n, o in zip(new_identities, old_identities)
)
)
if (name == existing.get('name', '')
and vlan_id == existing.get('vlan_id')
and subnet == existing.get('subnet', '')
and final_mask == existing.get('subnet_mask', 24)
and dnsmasq_log_queries == bool(existing.get('dnsmasq_log_queries', False))
and radius_default == bool(existing.get('radius_default', False))
and mdns_reflection == bool(existing.get('mdns_reflection', False))
and sorted(use_blocklists) == sorted(existing.get('use_blocklists', []))
and _ids_unchanged
and new_stored_gw == existing_gw
and new_stored_dns == existing_dns
and new_stored_ntp == existing_ntp):
flash('No changes were made.', 'info')
return redirect(f'/{_PAGE}')
before = copy.deepcopy(existing)
existing.update({
'name': name,
'vlan_id': vlan_id,
'is_vpn': is_vpn,
'subnet': subnet,
'subnet_mask': final_mask,
'dnsmasq_log_queries': dnsmasq_log_queries,
'radius_default': radius_default,
'mdns_reflection': mdns_reflection,
'use_blocklists': use_blocklists,
'server_identities': new_identities,
})
dhcp_overrides = existing.setdefault('dhcp_information', {}).setdefault('explicit_overrides', {})
if new_stored_gw:
dhcp_overrides['gateway'] = new_stored_gw
else:
dhcp_overrides.pop('gateway', None)
if new_stored_dns:
dhcp_overrides['dns_servers'] = new_stored_dns
else:
dhcp_overrides.pop('dns_servers', None)
if new_stored_ntp:
dhcp_overrides['ntp_servers'] = new_stored_ntp
else:
dhcp_overrides.pop('ntp_servers', None)
if not dhcp_overrides:
existing.get('dhcp_information', {}).pop('explicit_overrides', None)
errors = validate.validate_config(cfg)
if errors:
for msg in errors:
flash(msg, 'error')
return redirect(f'/{_PAGE}')
changes = diff_fields(before, existing)
flash(record_group(cfg, 'vlans', 'name', name, changes, 'core apply'), 'success')
return redirect(f'/{_PAGE}')