Development
This commit is contained in:
parent
b63aed53fc
commit
6221ee3691
12 changed files with 511 additions and 245 deletions
|
|
@ -17,9 +17,9 @@ def actions_cardoptions_save():
|
|||
return redirect(_VIEW)
|
||||
|
||||
|
||||
@bp.route('/action/actions_cardpendingchanges_applyselected', methods=['POST'])
|
||||
@bp.route('/action/actions_cardpendingchanges_applynow', methods=['POST'])
|
||||
@require_level('administrator')
|
||||
def actions_cardpendingchanges_applyselected():
|
||||
def actions_cardpendingchanges_applynow():
|
||||
selected_uuids = request.form.getlist('selected_uuids')
|
||||
if not selected_uuids:
|
||||
flash('No items selected.', 'info')
|
||||
|
|
|
|||
|
|
@ -1,12 +1,14 @@
|
|||
import copy
|
||||
|
||||
from flask import Blueprint, request, redirect, flash
|
||||
from auth import require_level
|
||||
from config_utils import load_core, save_core, verify_core_hash, queued_msg
|
||||
from config_utils import load_core, save_core_with_snapshot, verify_core_hash
|
||||
import sanitize
|
||||
import validation as validate
|
||||
|
||||
bp = Blueprint('action_apply_banned_ips', __name__)
|
||||
|
||||
VIEW = '/view/view_banned_ips'
|
||||
VIEW = '/view/view_banned_ips'
|
||||
|
||||
|
||||
def _row_index():
|
||||
|
|
@ -24,7 +26,6 @@ def _hash_ok():
|
|||
|
||||
|
||||
def _parse_ip():
|
||||
"""Return validated IP string, or None after flashing an error."""
|
||||
raw = request.form.get('ip', '').strip()
|
||||
if not raw:
|
||||
flash('The configuration has not been saved because an IP address, CIDR, or wildcard pattern is required.', 'error')
|
||||
|
|
@ -43,24 +44,24 @@ def add_banned_ip():
|
|||
ip = _parse_ip()
|
||||
if ip is None:
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
core = load_core()
|
||||
core.setdefault('banned_ips', []).append({
|
||||
'description': description,
|
||||
'ip': ip,
|
||||
'enabled': True,
|
||||
})
|
||||
core = load_core()
|
||||
entry = {'description': description, 'ip': ip, 'enabled': True}
|
||||
core.setdefault('banned_ips', []).append(entry)
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='banned_ips', key=ip, operation='add',
|
||||
before=None, after=entry,
|
||||
description=f'Added banned IP: {ip}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -71,7 +72,6 @@ def toggle_banned_ip():
|
|||
if idx is None:
|
||||
flash('Invalid request.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -81,15 +81,21 @@ def toggle_banned_ip():
|
|||
flash('Entry not found.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
items[idx]['enabled'] = not items[idx].get('enabled', True)
|
||||
old_enabled = items[idx].get('enabled', True)
|
||||
items[idx]['enabled'] = not old_enabled
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
action = 'Enabled' if not old_enabled else 'Disabled'
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='banned_ips', key=items[idx]['ip'], operation='toggle',
|
||||
before={'enabled': old_enabled}, after={'enabled': not old_enabled},
|
||||
description=f'{action} banned IP: {items[idx]["ip"]}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -116,15 +122,20 @@ def edit_banned_ip():
|
|||
flash('Entry not found.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
before = copy.deepcopy(items[idx])
|
||||
items[idx].update({'description': description, 'ip': ip, 'enabled': enabled})
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='banned_ips', key=ip, operation='edit',
|
||||
before=before, after=copy.deepcopy(items[idx]),
|
||||
description=f'Edited banned IP: {ip}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -135,7 +146,6 @@ def delete_banned_ip():
|
|||
if idx is None:
|
||||
flash('Invalid request.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -151,7 +161,11 @@ def delete_banned_ip():
|
|||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='banned_ips', key=removed['ip'], operation='delete',
|
||||
before=removed, after=None,
|
||||
description=f'Deleted banned IP: {removed["ip"]}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
|
|
|||
|
|
@ -1,14 +1,15 @@
|
|||
import copy
|
||||
import ipaddress
|
||||
|
||||
from flask import Blueprint, request, redirect, flash
|
||||
from auth import require_level
|
||||
from config_utils import load_core, save_core, verify_core_hash, queued_msg
|
||||
from config_utils import load_core, save_core_with_snapshot, verify_core_hash
|
||||
import sanitize
|
||||
import validation as validate
|
||||
|
||||
bp = Blueprint('action_apply_dhcp_reservations', __name__)
|
||||
|
||||
VIEW = '/view/view_dhcp'
|
||||
VIEW = '/view/view_dhcp'
|
||||
|
||||
|
||||
def _row_index():
|
||||
|
|
@ -36,7 +37,6 @@ def _flat_index_to_vlan_res(vlans, flat_idx):
|
|||
|
||||
|
||||
def _parse_ip():
|
||||
"""Return validated IP string, or None after flashing an error."""
|
||||
raw = request.form.get('ip', '').strip()
|
||||
if not raw:
|
||||
flash('The configuration has not been saved because an IP address is required.', 'error')
|
||||
|
|
@ -49,8 +49,7 @@ def _parse_ip():
|
|||
|
||||
|
||||
def _check_ip_conflicts(ip, vlan):
|
||||
"""Return an error message if ip conflicts with pool range or server identities, else None."""
|
||||
dhcp = vlan.get('dhcp_information', {})
|
||||
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:
|
||||
|
|
@ -78,14 +77,12 @@ def add_dhcp_reservation():
|
|||
|
||||
if ip is None:
|
||||
return redirect(VIEW)
|
||||
|
||||
if not vlan_name:
|
||||
flash('The configuration has not been saved because a VLAN is required.', 'error')
|
||||
return redirect(VIEW)
|
||||
if not mac:
|
||||
flash('The configuration has not been saved because a MAC address is required.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -101,22 +98,27 @@ def add_dhcp_reservation():
|
|||
flash(f'The configuration has not been saved because {conflict}', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
vlan.setdefault('reservations', []).append({
|
||||
entry = {
|
||||
'description': description,
|
||||
'hostname': hostname,
|
||||
'mac': mac,
|
||||
'ip': ip,
|
||||
'radius_client': radius_client,
|
||||
'enabled': True,
|
||||
})
|
||||
}
|
||||
vlan.setdefault('reservations', []).append(entry)
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path=f'vlans.{vlan_name}.reservations', key=mac, operation='add',
|
||||
before=None, after=entry,
|
||||
description=f'Added DHCP reservation: {hostname or mac} ({ip})',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -127,7 +129,6 @@ def toggle_dhcp_reservation():
|
|||
if idx is None:
|
||||
flash('Invalid request.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -138,16 +139,23 @@ def toggle_dhcp_reservation():
|
|||
flash('Entry not found.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
res = vlans[vi]['reservations'][ri]
|
||||
res['enabled'] = not res.get('enabled', True)
|
||||
res = vlans[vi]['reservations'][ri]
|
||||
old_enabled = res.get('enabled', True)
|
||||
res['enabled'] = not old_enabled
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
vlan_name = vlans[vi]['name']
|
||||
action = 'Enabled' if not old_enabled else 'Disabled'
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path=f'vlans.{vlan_name}.reservations', key=res['mac'], operation='toggle',
|
||||
before={'enabled': old_enabled}, after={'enabled': not old_enabled},
|
||||
description=f'{action} DHCP reservation: {res.get("hostname") or res["mac"]}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -170,7 +178,6 @@ def edit_dhcp_reservation():
|
|||
if not mac:
|
||||
flash('The configuration has not been saved because a MAC address is required.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -186,7 +193,8 @@ def edit_dhcp_reservation():
|
|||
flash(f'The configuration has not been saved because {conflict}', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
res = vlans[vi]['reservations'][ri]
|
||||
res = vlans[vi]['reservations'][ri]
|
||||
before = copy.deepcopy(res)
|
||||
res.update({
|
||||
'description': description,
|
||||
'hostname': hostname,
|
||||
|
|
@ -200,9 +208,14 @@ def edit_dhcp_reservation():
|
|||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
vlan_name = vlans[vi]['name']
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path=f'vlans.{vlan_name}.reservations', key=mac, operation='edit',
|
||||
before=before, after=copy.deepcopy(res),
|
||||
description=f'Edited DHCP reservation: {hostname or mac} ({ip})',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -213,7 +226,6 @@ def delete_dhcp_reservation():
|
|||
if idx is None:
|
||||
flash('Invalid request.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -224,13 +236,18 @@ def delete_dhcp_reservation():
|
|||
flash('Entry not found.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
removed = vlans[vi]['reservations'].pop(ri)
|
||||
vlan_name = vlans[vi]['name']
|
||||
removed = vlans[vi]['reservations'].pop(ri)
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path=f'vlans.{vlan_name}.reservations', key=removed['mac'], operation='delete',
|
||||
before=removed, after=None,
|
||||
description=f'Deleted DHCP reservation: {removed.get("hostname") or removed["mac"]}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
|
|
|||
|
|
@ -1,14 +1,15 @@
|
|||
import copy
|
||||
import ipaddress
|
||||
|
||||
from flask import Blueprint, request, redirect, flash
|
||||
from auth import require_level
|
||||
from config_utils import load_core, save_core, verify_core_hash, queued_msg
|
||||
from config_utils import load_core, save_core_with_snapshot, verify_core_hash
|
||||
import sanitize
|
||||
import validation as validate
|
||||
|
||||
bp = Blueprint('action_apply_host_overrides', __name__)
|
||||
|
||||
VIEW = '/view/view_host_overrides'
|
||||
VIEW = '/view/view_host_overrides'
|
||||
|
||||
|
||||
def _vlan_networks(core):
|
||||
|
|
@ -18,14 +19,13 @@ def _vlan_networks(core):
|
|||
mask = v.get('subnet_mask', '')
|
||||
if subnet and mask:
|
||||
try:
|
||||
nets.append(ipaddress.IPv4Network(f"{subnet}/{mask}", strict=False))
|
||||
nets.append(ipaddress.IPv4Network(f'{subnet}/{mask}', strict=False))
|
||||
except ValueError:
|
||||
pass
|
||||
return nets
|
||||
|
||||
|
||||
def _ip_in_vlan(ip_str, core):
|
||||
"""Return True if ip_str falls within at least one configured VLAN subnet."""
|
||||
try:
|
||||
addr = ipaddress.IPv4Address(ip_str)
|
||||
except ValueError:
|
||||
|
|
@ -58,7 +58,6 @@ def add_host_override():
|
|||
if not host or not ip:
|
||||
flash('Hostname and IP address are required.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -67,20 +66,20 @@ def add_host_override():
|
|||
flash('IP address does not fall within any configured VLAN subnet.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
core.setdefault('host_overrides', []).append({
|
||||
'description': description,
|
||||
'host': host,
|
||||
'ip': ip,
|
||||
'enabled': True,
|
||||
})
|
||||
entry = {'description': description, 'host': host, 'ip': ip, 'enabled': True}
|
||||
core.setdefault('host_overrides', []).append(entry)
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='host_overrides', key=host, operation='add',
|
||||
before=None, after=entry,
|
||||
description=f'Added host override: {host} → {ip}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -91,7 +90,6 @@ def toggle_host_override():
|
|||
if idx is None:
|
||||
flash('Invalid request.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -101,15 +99,21 @@ def toggle_host_override():
|
|||
flash('Entry not found.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
items[idx]['enabled'] = not items[idx].get('enabled', True)
|
||||
old_enabled = items[idx].get('enabled', True)
|
||||
items[idx]['enabled'] = not old_enabled
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
action = 'Enabled' if not old_enabled else 'Disabled'
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='host_overrides', key=items[idx]['host'], operation='toggle',
|
||||
before={'enabled': old_enabled}, after={'enabled': not old_enabled},
|
||||
description=f'{action} host override: {items[idx]["host"]}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -129,11 +133,10 @@ def edit_host_override():
|
|||
if not host or not ip:
|
||||
flash('Hostname and IP address are required.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
core = load_core()
|
||||
core = load_core()
|
||||
if not _ip_in_vlan(ip, core):
|
||||
flash('IP address does not fall within any configured VLAN subnet.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
|
@ -143,15 +146,20 @@ def edit_host_override():
|
|||
flash('Entry not found.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
before = copy.deepcopy(items[idx])
|
||||
items[idx].update({'description': description, 'host': host, 'ip': ip, 'enabled': enabled})
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='host_overrides', key=host, operation='edit',
|
||||
before=before, after=copy.deepcopy(items[idx]),
|
||||
description=f'Edited host override: {host} → {ip}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -162,7 +170,6 @@ def delete_host_override():
|
|||
if idx is None:
|
||||
flash('Invalid request.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -178,7 +185,11 @@ def delete_host_override():
|
|||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='host_overrides', key=removed['host'], operation='delete',
|
||||
before=removed, after=None,
|
||||
description=f'Deleted host override: {removed["host"]}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
|
|
|||
|
|
@ -1,12 +1,14 @@
|
|||
import copy
|
||||
|
||||
from flask import Blueprint, request, redirect, flash
|
||||
from auth import require_level
|
||||
from config_utils import load_core, save_core, verify_core_hash, queued_msg
|
||||
from config_utils import load_core, save_core_with_snapshot, verify_core_hash
|
||||
import sanitize
|
||||
import validation as validate
|
||||
|
||||
bp = Blueprint('action_apply_inter_vlan', __name__)
|
||||
|
||||
VIEW = '/view/view_inter_vlan'
|
||||
VIEW = '/view/view_inter_vlan'
|
||||
|
||||
_VALID_PROTOS_STR = ', '.join(sorted(validate.VALID_PROTOCOLS))
|
||||
|
||||
|
|
@ -26,7 +28,6 @@ def _hash_ok():
|
|||
|
||||
|
||||
def _parse_entry():
|
||||
"""Parse and validate form fields. Returns (entry_dict, None) or (None, already_flashed)."""
|
||||
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()
|
||||
|
|
@ -71,13 +72,17 @@ def _parse_entry():
|
|||
}, None
|
||||
|
||||
|
||||
def _entry_key(entry):
|
||||
port = f':{entry["dst_port"]}' if entry.get('dst_port') else ''
|
||||
return f'{entry["protocol"]}:{entry["src_ip_or_subnet"]}→{entry["dst_ip_or_subnet"]}{port}'
|
||||
|
||||
|
||||
@bp.route('/action/add_inter_vlan', methods=['POST'])
|
||||
@require_level('administrator')
|
||||
def add_inter_vlan():
|
||||
entry, err = _parse_entry()
|
||||
if err:
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -88,9 +93,14 @@ def add_inter_vlan():
|
|||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
key = _entry_key(entry)
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='inter_vlan_exceptions', key=key, operation='add',
|
||||
before=None, after=entry,
|
||||
description=f'Added inter-VLAN rule: {key}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -101,7 +111,6 @@ def toggle_inter_vlan():
|
|||
if idx is None:
|
||||
flash('Invalid request.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -111,15 +120,22 @@ def toggle_inter_vlan():
|
|||
flash('Entry not found.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
items[idx]['enabled'] = not items[idx].get('enabled', True)
|
||||
old_enabled = items[idx].get('enabled', True)
|
||||
items[idx]['enabled'] = not old_enabled
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
key = _entry_key(items[idx])
|
||||
action = 'Enabled' if not old_enabled else 'Disabled'
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='inter_vlan_exceptions', key=key, operation='toggle',
|
||||
before={'enabled': old_enabled}, after={'enabled': not old_enabled},
|
||||
description=f'{action} inter-VLAN rule: {key}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -134,7 +150,6 @@ def edit_inter_vlan():
|
|||
entry, err = _parse_entry()
|
||||
if err:
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -144,16 +159,22 @@ def edit_inter_vlan():
|
|||
flash('Entry not found.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
items[idx] = entry
|
||||
before = copy.deepcopy(items[idx])
|
||||
items[idx] = entry
|
||||
items[idx]['enabled'] = request.form.get('enabled') == 'on'
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
key = _entry_key(entry)
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='inter_vlan_exceptions', key=key, operation='edit',
|
||||
before=before, after=copy.deepcopy(items[idx]),
|
||||
description=f'Edited inter-VLAN rule: {key}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -164,7 +185,6 @@ def delete_inter_vlan():
|
|||
if idx is None:
|
||||
flash('Invalid request.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -174,13 +194,18 @@ def delete_inter_vlan():
|
|||
flash('Entry not found.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
items.pop(idx)
|
||||
removed = items.pop(idx)
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
key = _entry_key(removed)
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='inter_vlan_exceptions', key=key, operation='delete',
|
||||
before=removed, after=None,
|
||||
description=f'Deleted inter-VLAN rule: {key}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
|
|
|||
|
|
@ -1,25 +1,30 @@
|
|||
import copy
|
||||
|
||||
from flask import Blueprint, request, redirect, flash
|
||||
from auth import require_level
|
||||
from config_utils import load_core, save_core, verify_core_hash, queued_msg
|
||||
from config_utils import load_core, save_core_with_snapshot, verify_core_hash
|
||||
import sanitize
|
||||
import validation as validate
|
||||
|
||||
bp = Blueprint('action_apply_mdns', __name__)
|
||||
|
||||
|
||||
|
||||
@bp.route('/action/apply_mdns', methods=['POST'])
|
||||
@require_level('administrator')
|
||||
def apply_mdns():
|
||||
mdns_enabled = 'mdns_enabled' in request.form
|
||||
mdns_enabled = 'mdns_enabled' in request.form
|
||||
|
||||
if not verify_core_hash(request.form.get('config_hash', '')):
|
||||
flash('Configuration was modified by another session. Please refresh and try again.', 'error')
|
||||
return redirect('/view/view_mdns')
|
||||
|
||||
core = load_core()
|
||||
mdns_reflect_vlans = sanitize.filterlist(request.form.getlist('mdns_reflect_vlans'),
|
||||
{v.get('name') for v in core.get('vlans', [])})
|
||||
mdns_reflect_vlans = sanitize.filterlist(
|
||||
request.form.getlist('mdns_reflect_vlans'),
|
||||
{v.get('name') for v in core.get('vlans', [])},
|
||||
)
|
||||
|
||||
before = copy.deepcopy(core.get('mdns_reflection', {}))
|
||||
core.setdefault('mdns_reflection', {}).update({
|
||||
'enabled': mdns_enabled,
|
||||
'reflect_vlans': mdns_reflect_vlans,
|
||||
|
|
@ -29,7 +34,11 @@ def apply_mdns():
|
|||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect('/view/view_mdns')
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='mdns_reflection', key='global', operation='edit',
|
||||
before=before or None, after=copy.deepcopy(core['mdns_reflection']),
|
||||
description='Updated mDNS reflection settings',
|
||||
), 'success')
|
||||
return redirect('/view/view_mdns')
|
||||
|
|
|
|||
|
|
@ -1,12 +1,14 @@
|
|||
import copy
|
||||
|
||||
from flask import Blueprint, request, redirect, flash
|
||||
from auth import require_level
|
||||
from config_utils import load_core, save_core, verify_core_hash, queued_msg
|
||||
from config_utils import load_core, save_core_with_snapshot, verify_core_hash
|
||||
import sanitize
|
||||
import validation as validate
|
||||
|
||||
bp = Blueprint('action_apply_port_forwarding', __name__)
|
||||
|
||||
VIEW = '/view/view_port_forwarding'
|
||||
VIEW = '/view/view_port_forwarding'
|
||||
|
||||
_VALID_PROTOS_STR = ', '.join(sorted(validate.VALID_PROTOCOLS))
|
||||
|
||||
|
|
@ -26,7 +28,6 @@ def _hash_ok():
|
|||
|
||||
|
||||
def _parse_entry():
|
||||
"""Parse and validate form fields. Returns (entry_dict, None) or (None, already_flashed)."""
|
||||
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()
|
||||
|
|
@ -78,7 +79,6 @@ def add_port_forward():
|
|||
entry, err = _parse_entry()
|
||||
if err:
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -89,9 +89,14 @@ def add_port_forward():
|
|||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
key = f'{entry["protocol"]}:{entry["dest_port"]}'
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='port_forwarding', key=key, operation='add',
|
||||
before=None, after=entry,
|
||||
description=f'Added port forward: {key} → {entry["nat_ip"]}:{entry["nat_port"]}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -102,7 +107,6 @@ def toggle_port_forward():
|
|||
if idx is None:
|
||||
flash('Invalid request.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -112,15 +116,22 @@ def toggle_port_forward():
|
|||
flash('Entry not found.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
items[idx]['enabled'] = not items[idx].get('enabled', True)
|
||||
old_enabled = items[idx].get('enabled', True)
|
||||
items[idx]['enabled'] = not old_enabled
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
key = f'{items[idx]["protocol"]}:{items[idx]["dest_port"]}'
|
||||
action = 'Enabled' if not old_enabled else 'Disabled'
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='port_forwarding', key=key, operation='toggle',
|
||||
before={'enabled': old_enabled}, after={'enabled': not old_enabled},
|
||||
description=f'{action} port forward: {key}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -135,7 +146,6 @@ def edit_port_forward():
|
|||
entry, err = _parse_entry()
|
||||
if err:
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -145,16 +155,22 @@ def edit_port_forward():
|
|||
flash('Entry not found.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
items[idx] = entry
|
||||
before = copy.deepcopy(items[idx])
|
||||
items[idx] = entry
|
||||
items[idx]['enabled'] = request.form.get('enabled') == 'on'
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
key = f'{entry["protocol"]}:{entry["dest_port"]}'
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='port_forwarding', key=key, operation='edit',
|
||||
before=before, after=copy.deepcopy(items[idx]),
|
||||
description=f'Edited port forward: {key} → {entry["nat_ip"]}:{entry["nat_port"]}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -165,7 +181,6 @@ def delete_port_forward():
|
|||
if idx is None:
|
||||
flash('Invalid request.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -181,7 +196,12 @@ def delete_port_forward():
|
|||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
key = f'{removed["protocol"]}:{removed["dest_port"]}'
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='port_forwarding', key=key, operation='delete',
|
||||
before=removed, after=None,
|
||||
description=f'Deleted port forward: {key}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
|
|
|||
|
|
@ -1,12 +1,17 @@
|
|||
import copy
|
||||
|
||||
from flask import Blueprint, request, redirect, flash
|
||||
from auth import require_level
|
||||
from config_utils import load_core, save_core, verify_core_hash, queued_msg
|
||||
from config_utils import load_core, save_core_with_snapshot, verify_core_hash
|
||||
import sanitize
|
||||
import validation as validate
|
||||
|
||||
bp = Blueprint('action_apply_vlans', __name__)
|
||||
|
||||
VIEW = '/view/view_vlans'
|
||||
VIEW = '/view/view_vlans'
|
||||
|
||||
_VLAN_FIELDS = ['name', 'is_vpn', 'subnet', 'subnet_mask', 'dnsmasq_log_queries',
|
||||
'radius_default', 'mdns_reflection', 'use_blocklists']
|
||||
|
||||
|
||||
def _row_index():
|
||||
|
|
@ -26,24 +31,24 @@ def _hash_ok():
|
|||
@bp.route('/action/add_vlan', methods=['POST'])
|
||||
@require_level('administrator')
|
||||
def add_vlan():
|
||||
name = sanitize.name(request.form.get('name', ''))
|
||||
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
|
||||
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_core().get('dns_blocking', {}).get('blocklists', [])})
|
||||
name = sanitize.name(request.form.get('name', ''))
|
||||
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
|
||||
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_core().get('dns_blocking', {}).get('blocklists', [])},
|
||||
)
|
||||
|
||||
if not name:
|
||||
flash('Name is required.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not subnet:
|
||||
flash('Subnet IP is required.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if subnet_mask is None:
|
||||
flash('Invalid subnet prefix (must be 1-30).', 'error')
|
||||
return redirect(VIEW)
|
||||
|
|
@ -62,20 +67,19 @@ def add_vlan():
|
|||
if any(validate.derive_vlan_id(v.get('subnet', ''), v.get('subnet_mask', 24)) == vlan_id for v in vlans):
|
||||
flash(f'VLAN {vlan_id} (derived from subnet) already exists.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
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(VIEW)
|
||||
|
||||
entry = {
|
||||
'name': name,
|
||||
'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,
|
||||
'name': name,
|
||||
'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,
|
||||
}
|
||||
if is_vpn:
|
||||
entry['peers'] = []
|
||||
|
|
@ -87,9 +91,13 @@ def add_vlan():
|
|||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='vlans', key=name, operation='add',
|
||||
before=None, after={k: entry[k] for k in _VLAN_FIELDS if k in entry},
|
||||
description=f'Added VLAN: {name} ({subnet}/{subnet_mask})',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -101,16 +109,16 @@ def edit_vlan():
|
|||
flash('Invalid request.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
name = sanitize.name(request.form.get('name', ''))
|
||||
subnet = sanitize.ip(request.form.get('subnet', ''))
|
||||
radius_default = 'radius_default' in request.form
|
||||
mdns_reflection = 'mdns_reflection' in request.form
|
||||
name = sanitize.name(request.form.get('name', ''))
|
||||
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_core().get('dns_blocking', {}).get('blocklists', [])})
|
||||
use_blocklists = sanitize.filterlist(
|
||||
request.form.getlist('use_blocklists'),
|
||||
{b.get('name') for b in load_core().get('dns_blocking', {}).get('blocklists', [])},
|
||||
)
|
||||
|
||||
# subnet_mask is only present when the column is visible (not all edit paths send it).
|
||||
# Validate if submitted; fall back to the stored value otherwise.
|
||||
subnet_mask_raw = request.form.get('subnet_mask')
|
||||
if subnet_mask_raw is not None:
|
||||
subnet_mask = sanitize.subnet_mask(subnet_mask_raw)
|
||||
|
|
@ -118,16 +126,14 @@ def edit_vlan():
|
|||
flash('Invalid subnet prefix (must be 1-30).', 'error')
|
||||
return redirect(VIEW)
|
||||
else:
|
||||
subnet_mask = None # resolved below after loading core
|
||||
subnet_mask = None
|
||||
|
||||
if not name:
|
||||
flash('Name is required.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not subnet:
|
||||
flash('Subnet IP is required.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -138,9 +144,7 @@ def edit_vlan():
|
|||
return redirect(VIEW)
|
||||
|
||||
existing = vlans[idx]
|
||||
# is_vpn is never changed via edit -- toggling it would invalidate peers/reservations.
|
||||
is_vpn = existing.get('is_vpn', False)
|
||||
# Use submitted subnet_mask, or fall back to whatever is already stored.
|
||||
final_mask = subnet_mask if subnet_mask is not None else existing.get('subnet_mask', 24)
|
||||
|
||||
vlan_id = validate.derive_vlan_id(subnet, final_mask)
|
||||
|
|
@ -164,6 +168,7 @@ def edit_vlan():
|
|||
flash('Only one VLAN can be the RADIUS default.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
before = {k: existing.get(k) for k in _VLAN_FIELDS}
|
||||
existing.update({
|
||||
'name': name,
|
||||
'is_vpn': is_vpn,
|
||||
|
|
@ -179,9 +184,13 @@ def edit_vlan():
|
|||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='vlans', key=name, operation='edit',
|
||||
before=before, after={k: existing.get(k) for k in _VLAN_FIELDS},
|
||||
description=f'Edited VLAN: {name}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
||||
|
||||
|
|
@ -192,7 +201,6 @@ def delete_vlan():
|
|||
if idx is None:
|
||||
flash('Invalid request.', 'error')
|
||||
return redirect(VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(VIEW)
|
||||
|
||||
|
|
@ -208,7 +216,12 @@ def delete_vlan():
|
|||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(VIEW)
|
||||
save_core(core)
|
||||
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path='vlans', key=removed['name'], operation='delete',
|
||||
before={k: removed.get(k) for k in _VLAN_FIELDS},
|
||||
after=None,
|
||||
description=f'Deleted VLAN: {removed["name"]}',
|
||||
), 'success')
|
||||
return redirect(VIEW)
|
||||
|
|
|
|||
|
|
@ -1,10 +1,11 @@
|
|||
import base64
|
||||
import copy
|
||||
import ipaddress
|
||||
import re
|
||||
|
||||
from flask import Blueprint, make_response, redirect, flash, request
|
||||
from auth import require_level
|
||||
from config_utils import load_core, save_core, verify_core_hash, queued_msg, CONFIGS_DIR, WEB_APP_DISPLAY_NAME
|
||||
from config_utils import load_core, save_core_with_snapshot, verify_core_hash, CONFIGS_DIR, WEB_APP_DISPLAY_NAME
|
||||
import sanitize
|
||||
import validation as validate
|
||||
|
||||
|
|
@ -24,7 +25,6 @@ def _wg_vlan_by_name(core, name):
|
|||
|
||||
|
||||
def _find_peer_by_flat_idx(core, flat_idx):
|
||||
"""Return (vlan, peer_list_index) by flat index across all VPN VLANs in order."""
|
||||
i = 0
|
||||
for vlan in core.get('vlans', []):
|
||||
if not vlan.get('is_vpn'):
|
||||
|
|
@ -38,7 +38,6 @@ def _find_peer_by_flat_idx(core, flat_idx):
|
|||
|
||||
|
||||
def _wg_iface(vlan, core):
|
||||
"""Return the WireGuard interface name (wg0, wg1, ...) for a VPN VLAN."""
|
||||
wg_vlans = [v for v in core.get('vlans', []) if v.get('is_vpn')]
|
||||
idx = next((i for i, v in enumerate(wg_vlans) if v is vlan), 0)
|
||||
return f'wg{idx}'
|
||||
|
|
@ -59,7 +58,6 @@ def _hash_ok():
|
|||
|
||||
|
||||
def _generate_wg_keypair():
|
||||
"""Generate an X25519 keypair compatible with WireGuard. Returns (private_b64, public_b64)."""
|
||||
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
|
||||
from cryptography.hazmat.primitives.serialization import (
|
||||
Encoding, PublicFormat, PrivateFormat, NoEncryption,
|
||||
|
|
@ -71,7 +69,6 @@ def _generate_wg_keypair():
|
|||
|
||||
|
||||
def _server_pubkey(iface):
|
||||
"""Read the server public key written by core.py --apply."""
|
||||
try:
|
||||
with open(f'{CONFIGS_DIR}/.{iface}.pub') as f:
|
||||
return f.read().strip()
|
||||
|
|
@ -80,7 +77,6 @@ def _server_pubkey(iface):
|
|||
|
||||
|
||||
def _build_client_conf(vlan, peer_name, peer_ip, private_key, server_pubkey):
|
||||
"""Build WireGuard client .conf content."""
|
||||
info = vlan.get('vpn_information', {})
|
||||
overrides = info.get('explicit_overrides', {})
|
||||
subnet = vlan.get('subnet', '')
|
||||
|
|
@ -89,22 +85,21 @@ def _build_client_conf(vlan, peer_name, peer_ip, private_key, server_pubkey):
|
|||
ident_ips = [s['ip'] for s in vlan.get('server_identities', []) if s.get('ip')]
|
||||
default = str(min((ipaddress.IPv4Address(ip) for ip in ident_ips),
|
||||
key=lambda x: x.packed[-1])) if ident_ips else str(next(network.hosts()))
|
||||
gateway = overrides.get('gateway') or default
|
||||
dns = overrides.get('dns_server') or gateway
|
||||
prefix = network.prefixlen
|
||||
mtu = overrides.get('mtu', '')
|
||||
endpoint = info.get('server_endpoint', '')
|
||||
gateway = overrides.get('gateway') or default
|
||||
dns = overrides.get('dns_server') or gateway
|
||||
prefix = network.prefixlen
|
||||
mtu = overrides.get('mtu', '')
|
||||
endpoint = info.get('server_endpoint', '')
|
||||
listen_port = info.get('listen_port', 51820)
|
||||
|
||||
split_tunnel = next(
|
||||
(p.get('split_tunnel', False) for p in vlan.get('peers', []) if p.get('name') == peer_name),
|
||||
False
|
||||
False,
|
||||
)
|
||||
allowed_ips = f'{subnet}/{prefix}' if split_tunnel else '0.0.0.0/0'
|
||||
|
||||
lines = [
|
||||
f'# Generated by {WEB_APP_DISPLAY_NAME}',
|
||||
'',
|
||||
f'# Generated by {WEB_APP_DISPLAY_NAME}', '',
|
||||
'[Interface]',
|
||||
f'PrivateKey = {private_key}',
|
||||
f'Address = {peer_ip}/{prefix}',
|
||||
|
|
@ -120,9 +115,8 @@ def _build_client_conf(vlan, peer_name, peer_ip, private_key, server_pubkey):
|
|||
|
||||
|
||||
def _conf_response(vlan, peer_name, peer_ip, private_key):
|
||||
"""Return a .conf file download response, or redirect with error if pubkey unavailable."""
|
||||
core = load_core()
|
||||
iface = _wg_iface(vlan, core)
|
||||
core = load_core()
|
||||
iface = _wg_iface(vlan, core)
|
||||
server_pub = _server_pubkey(iface)
|
||||
if not server_pub:
|
||||
flash('Peer saved. Run sudo python3 ~/routlin/core.py --apply to generate the server '
|
||||
|
|
@ -170,7 +164,7 @@ def apply_vpn():
|
|||
if not _hash_ok():
|
||||
return redirect(_VIEW)
|
||||
|
||||
core = load_core()
|
||||
core = load_core()
|
||||
vpn_vlan = _wg_vlan(core)
|
||||
if vpn_vlan is None:
|
||||
flash('No WireGuard VLAN found in configuration.', 'error')
|
||||
|
|
@ -181,7 +175,8 @@ def apply_vpn():
|
|||
flash(f'Listen port {listen_port} is already used by another VPN VLAN.', 'error')
|
||||
return redirect(_VIEW)
|
||||
|
||||
info = vpn_vlan.setdefault('vpn_information', {})
|
||||
before_info = copy.deepcopy(vpn_vlan.get('vpn_information', {}))
|
||||
info = vpn_vlan.setdefault('vpn_information', {})
|
||||
info['listen_port'] = listen_port
|
||||
info['server_endpoint'] = server_endpoint
|
||||
info['domain'] = domain
|
||||
|
|
@ -201,8 +196,14 @@ def apply_vpn():
|
|||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(_VIEW)
|
||||
save_core(core)
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
|
||||
vlan_name = vpn_vlan['name']
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path=f'vlans.{vlan_name}.vpn_information', key=vlan_name, operation='edit',
|
||||
before=before_info or None, after=copy.deepcopy(info),
|
||||
description=f'Updated VPN configuration for {vlan_name}',
|
||||
), 'success')
|
||||
return redirect(_VIEW)
|
||||
|
||||
|
||||
|
|
@ -225,11 +226,10 @@ def add_vpn_peer():
|
|||
if not peer_ip:
|
||||
flash(f'"{peer_ip_raw}" is not a valid IP address.', 'error')
|
||||
return redirect(_VIEW)
|
||||
|
||||
if not _hash_ok():
|
||||
return redirect(_VIEW)
|
||||
|
||||
core = load_core()
|
||||
core = load_core()
|
||||
vpn_vlan = _wg_vlan_by_name(core, peer_vlan_nm)
|
||||
if vpn_vlan is None:
|
||||
flash(f'VPN VLAN "{peer_vlan_nm}" not found.', 'error')
|
||||
|
|
@ -255,20 +255,27 @@ def add_vpn_peer():
|
|||
return redirect(_VIEW)
|
||||
|
||||
private_key, public_key = _generate_wg_keypair()
|
||||
peers.append({
|
||||
entry = {
|
||||
'name': peer_name,
|
||||
'ip': peer_ip,
|
||||
'public_key': public_key,
|
||||
'split_tunnel': split_tunnel,
|
||||
'enabled': enabled,
|
||||
})
|
||||
}
|
||||
peers.append(entry)
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(_VIEW)
|
||||
save_core(core)
|
||||
|
||||
save_core_with_snapshot(
|
||||
core,
|
||||
path=f'vlans.{peer_vlan_nm}.peers', key=peer_name, operation='add',
|
||||
before=None, after={k: v for k, v in entry.items() if k != 'public_key'},
|
||||
description=f'Added VPN peer: {peer_name} ({peer_ip})',
|
||||
queue=True,
|
||||
)
|
||||
return _conf_response(vpn_vlan, peer_name, peer_ip, private_key)
|
||||
|
||||
|
||||
|
|
@ -301,14 +308,21 @@ def edit_vpn_peer():
|
|||
flash(f'A peer named "{peer_name}" already exists.', 'error')
|
||||
return redirect(_VIEW)
|
||||
|
||||
before = copy.deepcopy({k: peers[peer_idx].get(k) for k in ('name', 'split_tunnel', 'enabled')})
|
||||
peers[peer_idx].update({'name': peer_name, 'split_tunnel': split_tunnel, 'enabled': enabled})
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(_VIEW)
|
||||
save_core(core)
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
|
||||
vlan_name = vlan['name']
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path=f'vlans.{vlan_name}.peers', key=peer_name, operation='edit',
|
||||
before=before, after={'name': peer_name, 'split_tunnel': split_tunnel, 'enabled': enabled},
|
||||
description=f'Edited VPN peer: {peer_name}',
|
||||
), 'success')
|
||||
return redirect(_VIEW)
|
||||
|
||||
|
||||
|
|
@ -328,15 +342,24 @@ def toggle_vpn_peer():
|
|||
flash('Peer not found.', 'error')
|
||||
return redirect(_VIEW)
|
||||
|
||||
peers = vlan.get('peers', [])
|
||||
peers[peer_idx]['enabled'] = not peers[peer_idx].get('enabled', True)
|
||||
peers = vlan.get('peers', [])
|
||||
old_enabled = peers[peer_idx].get('enabled', True)
|
||||
peers[peer_idx]['enabled'] = not old_enabled
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(_VIEW)
|
||||
save_core(core)
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
|
||||
peer_name = peers[peer_idx]['name']
|
||||
vlan_name = vlan['name']
|
||||
action = 'Enabled' if not old_enabled else 'Disabled'
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path=f'vlans.{vlan_name}.peers', key=peer_name, operation='toggle',
|
||||
before={'enabled': old_enabled}, after={'enabled': not old_enabled},
|
||||
description=f'{action} VPN peer: {peer_name}',
|
||||
), 'success')
|
||||
return redirect(_VIEW)
|
||||
|
||||
|
||||
|
|
@ -356,14 +379,22 @@ def delete_vpn_peer():
|
|||
flash('Peer not found.', 'error')
|
||||
return redirect(_VIEW)
|
||||
|
||||
vlan.get('peers', []).pop(peer_idx)
|
||||
peers = vlan.get('peers', [])
|
||||
removed = peers.pop(peer_idx)
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(_VIEW)
|
||||
save_core(core)
|
||||
flash(queued_msg('core apply'), 'success')
|
||||
|
||||
vlan_name = vlan['name']
|
||||
flash(save_core_with_snapshot(
|
||||
core,
|
||||
path=f'vlans.{vlan_name}.peers', key=removed['name'], operation='delete',
|
||||
before={k: removed.get(k) for k in ('name', 'ip', 'split_tunnel', 'enabled')},
|
||||
after=None,
|
||||
description=f'Deleted VPN peer: {removed["name"]}',
|
||||
), 'success')
|
||||
return redirect(_VIEW)
|
||||
|
||||
|
||||
|
|
@ -384,13 +415,21 @@ def regenerate_vpn_peer():
|
|||
return redirect(_VIEW)
|
||||
|
||||
private_key, public_key = _generate_wg_keypair()
|
||||
peer = vlan['peers'][peer_idx]
|
||||
peer = vlan['peers'][peer_idx]
|
||||
old_pub_key = peer.get('public_key', '')
|
||||
peer['public_key'] = public_key
|
||||
errors = validate.validate_config(core)
|
||||
if errors:
|
||||
for msg in errors:
|
||||
flash(msg, 'error')
|
||||
return redirect(_VIEW)
|
||||
save_core(core)
|
||||
|
||||
vlan_name = vlan['name']
|
||||
save_core_with_snapshot(
|
||||
core,
|
||||
path=f'vlans.{vlan_name}.peers', key=peer['name'], operation='regenerate',
|
||||
before={'public_key': old_pub_key}, after={'public_key': public_key},
|
||||
description=f'Regenerated keypair for VPN peer: {peer["name"]}',
|
||||
queue=True,
|
||||
)
|
||||
return _conf_response(vlan, peer['name'], peer['ip'], private_key)
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import json, subprocess, hashlib, os, uuid
|
||||
import copy, json, subprocess, hashlib, os, uuid
|
||||
from datetime import datetime, timezone
|
||||
from flask import session
|
||||
|
||||
|
|
@ -11,13 +11,14 @@ DASHBOARD_DONE = f'{CONFIGS_DIR}/.dashboard-done'
|
|||
DASHBOARD_LAST_RUN = f'{CONFIGS_DIR}/.dashboard-last-run'
|
||||
DASHBOARD_LOCK = f'{CONFIGS_DIR}/.dashboard-lock'
|
||||
DASHBOARD_PENDING = f'{CONFIGS_DIR}/.dashboard-pending'
|
||||
SNAPSHOTS_DIR = f'{CONFIGS_DIR}/.snapshots'
|
||||
HEALTH_FILE = f'{CONFIGS_DIR}/.health'
|
||||
PRODUCT_NAME = os.environ.get('PRODUCT_NAME', 'routlin')
|
||||
DASHB_TIMER_NAME = f'{PRODUCT_NAME}-dashboard-queue'
|
||||
DDNS_TIMER_NAME = f'{PRODUCT_NAME}-ddns-update'
|
||||
WEB_APP_DISPLAY_NAME = os.environ.get('WEB_APP_DISPLAY_NAME', f'{PRODUCT_NAME.capitalize()} Dashboard')
|
||||
DASHB_INTERVAL_SECS = 60
|
||||
QUEUE_MAX_LINES = 50
|
||||
PRODUCT_NAME = os.environ.get('PRODUCT_NAME', 'routlin')
|
||||
DASHB_TIMER_NAME = f'{PRODUCT_NAME}-dashboard-queue'
|
||||
DDNS_TIMER_NAME = f'{PRODUCT_NAME}-ddns-update'
|
||||
WEB_APP_DISPLAY_NAME = os.environ.get('WEB_APP_DISPLAY_NAME', f'{PRODUCT_NAME.capitalize()} Dashboard')
|
||||
DASHB_INTERVAL_SECS = 60
|
||||
QUEUE_MAX_LINES = 50
|
||||
|
||||
|
||||
def load_core():
|
||||
|
|
@ -70,7 +71,7 @@ def _read_pending(done_set):
|
|||
parts = line.split(None, 2)
|
||||
if len(parts) == 3:
|
||||
entry_uuid, entry_ts, rest = parts
|
||||
cmd_user = rest.rsplit(' (', 1)
|
||||
cmd_user = rest.rsplit(' (', 1)
|
||||
entry_cmd = cmd_user[0].strip('[]')
|
||||
entry_user = cmd_user[1].rstrip(')') if len(cmd_user) == 2 else ''
|
||||
if entry_uuid not in done_set:
|
||||
|
|
@ -117,7 +118,7 @@ def _apply_changes_immediately():
|
|||
|
||||
|
||||
def _read_dashboard_pending():
|
||||
"""Return list of (uuid, ts, cmd, user, description) from .dashboard-pending."""
|
||||
"""Return list of (uuid, ts, cmd, user) from .dashboard-pending."""
|
||||
items = []
|
||||
try:
|
||||
lines = open(DASHBOARD_PENDING).read().splitlines()
|
||||
|
|
@ -127,14 +128,13 @@ def _read_dashboard_pending():
|
|||
if not line.strip():
|
||||
continue
|
||||
try:
|
||||
main, _, desc = line.partition(' :: ')
|
||||
parts = main.split(None, 2)
|
||||
parts = line.split(None, 2)
|
||||
if len(parts) == 3:
|
||||
entry_uuid, entry_ts, rest = parts
|
||||
cmd_user = rest.rsplit(' (', 1)
|
||||
entry_cmd = cmd_user[0].strip('[]')
|
||||
entry_user = cmd_user[1].rstrip(')') if len(cmd_user) == 2 else ''
|
||||
items.append((entry_uuid, int(entry_ts), entry_cmd, entry_user, desc))
|
||||
items.append((entry_uuid, int(entry_ts), entry_cmd, entry_user))
|
||||
except Exception:
|
||||
pass
|
||||
return items
|
||||
|
|
@ -152,7 +152,7 @@ def flush_pending_to_queue():
|
|||
done_set = _load_done_set()
|
||||
existing_ids = {uu for uu, *_ in _read_pending(done_set)}
|
||||
with open(DASHBOARD_QUEUE, 'a') as f:
|
||||
for entry_uuid, entry_ts, entry_cmd, entry_user, _desc in items:
|
||||
for entry_uuid, entry_ts, entry_cmd, entry_user in items:
|
||||
if entry_uuid not in existing_ids:
|
||||
f.write(f'{entry_uuid} {entry_ts} [{entry_cmd}] ({entry_user})\n')
|
||||
open(DASHBOARD_PENDING, 'w').close()
|
||||
|
|
@ -177,7 +177,7 @@ def flush_selected_to_queue(selected_uuids):
|
|||
done_set = _load_done_set()
|
||||
existing_ids = {uu for uu, *_ in _read_pending(done_set)}
|
||||
with open(DASHBOARD_QUEUE, 'a') as f:
|
||||
for entry_uuid, entry_ts, entry_cmd, entry_user, _desc in items:
|
||||
for entry_uuid, entry_ts, entry_cmd, entry_user in items:
|
||||
if entry_uuid in selected_set and entry_uuid not in existing_ids:
|
||||
f.write(f'{entry_uuid} {entry_ts} [{entry_cmd}] ({entry_user})\n')
|
||||
_remove_pending_by_uuids(selected_set)
|
||||
|
|
@ -190,38 +190,40 @@ def delete_pending_by_uuids(selected_uuids):
|
|||
_remove_pending_by_uuids(set(selected_uuids))
|
||||
|
||||
|
||||
def _queue_pending_command(cmd, description=''):
|
||||
def _queue_pending_command(cmd):
|
||||
"""Append cmd to .dashboard-pending if not already present for this cmd+user."""
|
||||
existing = _read_dashboard_pending()
|
||||
current_user = session.get('email_address', 'unknown')
|
||||
for entry_uuid, entry_ts, entry_cmd, entry_user, _desc in existing:
|
||||
for entry_uuid, entry_ts, entry_cmd, entry_user in existing:
|
||||
if entry_cmd == cmd and entry_user == current_user:
|
||||
return entry_uuid, entry_ts
|
||||
entry_uuid = str(uuid.uuid4())
|
||||
now = datetime.now()
|
||||
entry_ts = int(now.timestamp())
|
||||
desc_suffix = f' :: {description}' if description else ''
|
||||
entry_ts = int(datetime.now().timestamp())
|
||||
with open(DASHBOARD_PENDING, 'a') as f:
|
||||
f.write(f'{entry_uuid} {entry_ts} [{cmd}] ({current_user}){desc_suffix}\n')
|
||||
f.write(f'{entry_uuid} {entry_ts} [{cmd}] ({current_user})\n')
|
||||
return entry_uuid, entry_ts
|
||||
|
||||
|
||||
def _queue_command(cmd, description=''):
|
||||
def _queue_pending_presigned(cmd, entry_uuid, entry_ts):
|
||||
"""Write a pre-generated entry to .dashboard-pending without dedup."""
|
||||
current_user = session.get('email_address', 'unknown')
|
||||
with open(DASHBOARD_PENDING, 'a') as f:
|
||||
f.write(f'{entry_uuid} {entry_ts} [{cmd}] ({current_user})\n')
|
||||
|
||||
|
||||
def _queue_command(cmd):
|
||||
if not _apply_changes_immediately():
|
||||
return _queue_pending_command(cmd, description)
|
||||
return _queue_pending_command(cmd)
|
||||
done_set = _load_done_set()
|
||||
pending = _read_pending(done_set)
|
||||
pending = _read_pending(done_set)
|
||||
current_user = session.get('email_address', 'unknown')
|
||||
for entry_uuid, entry_ts, entry_cmd, entry_user in pending:
|
||||
if entry_cmd == cmd and entry_user == current_user:
|
||||
return entry_uuid, entry_ts
|
||||
entry_uuid = str(uuid.uuid4())
|
||||
now = datetime.now()
|
||||
entry_ts = int(now.timestamp())
|
||||
dt_str = now.strftime('%Y-%m-%dT%H:%M:%S')
|
||||
user = session.get('email_address', 'unknown')
|
||||
entry_ts = int(datetime.now().timestamp())
|
||||
with open(DASHBOARD_QUEUE, 'a') as f:
|
||||
f.write(f'{entry_uuid} {entry_ts} {dt_str} [{cmd}] ({user})\n')
|
||||
f.write(f'{entry_uuid} {entry_ts} [{cmd}] ({current_user})\n')
|
||||
_trim_if_needed()
|
||||
return entry_uuid, entry_ts
|
||||
|
||||
|
|
@ -240,7 +242,7 @@ def _entry_ts_from_queue(entry_uuid):
|
|||
def _seconds_until_next_run():
|
||||
try:
|
||||
last_run = float(open(DASHBOARD_LAST_RUN).read().strip())
|
||||
elapsed = datetime.now(timezone.utc).timestamp() - last_run
|
||||
elapsed = datetime.now(timezone.utc).timestamp() - last_run
|
||||
return int(max(0, DASHB_INTERVAL_SECS - elapsed))
|
||||
except Exception:
|
||||
return None
|
||||
|
|
@ -260,18 +262,7 @@ def _lock_mtime():
|
|||
return None
|
||||
|
||||
|
||||
def queue_command(cmd, description=''):
|
||||
"""Queue a command without generating a flash message."""
|
||||
return _queue_command(cmd, description)
|
||||
|
||||
|
||||
def queued_msg(cmd=None, description='', action_label='Configuration saved'):
|
||||
"""Queue cmd if given, then return a timing message.
|
||||
Without cmd, just returns timing (for commands already queued by the caller).
|
||||
action_label replaces the 'Configuration saved' prefix for non-save actions."""
|
||||
entry_ts = None
|
||||
if cmd is not None:
|
||||
_entry_uuid, entry_ts = queue_command(cmd, description)
|
||||
def _build_timing_msg(entry_ts, cmd, action_label='Configuration saved'):
|
||||
if not _apply_changes_immediately():
|
||||
return f'{action_label}. Click Apply Now on the Configuration Changes card to apply.'
|
||||
if _is_locked():
|
||||
|
|
@ -284,15 +275,115 @@ def queued_msg(cmd=None, description='', action_label='Configuration saved'):
|
|||
return f'{action_label}. Changes will be applied {timing}.'
|
||||
if cmd is None:
|
||||
return f'{action_label}. The processing service is not running.'
|
||||
parts = cmd.split()
|
||||
parts = cmd.split()
|
||||
cli_cmd = f'sudo python3 {parts[0]}.py --{parts[1]}' if len(parts) == 2 else cmd
|
||||
install_cmd = f'sudo python3 install.py'
|
||||
install_cmd = 'sudo python3 install.py'
|
||||
from markupsafe import Markup
|
||||
return Markup(f'{action_label}. The command processing service is not installed. '
|
||||
f'Run <strong>{install_cmd}</strong> to enable it, '
|
||||
f'or <strong>{cli_cmd}</strong> to apply manually.')
|
||||
|
||||
|
||||
def queue_command(cmd, description=''):
|
||||
"""Queue a command without generating a flash message. description is ignored (kept for compat)."""
|
||||
return _queue_command(cmd)
|
||||
|
||||
|
||||
def queued_msg(cmd=None, description='', action_label='Configuration saved'):
|
||||
"""Queue cmd if given, then return a timing message. description is ignored."""
|
||||
entry_ts = None
|
||||
if cmd is not None:
|
||||
_entry_uuid, entry_ts = queue_command(cmd)
|
||||
return _build_timing_msg(entry_ts, cmd, action_label)
|
||||
|
||||
|
||||
# ── Snapshot system ───────────────────────────────────────────────────────────
|
||||
|
||||
def _pending_uuid_set():
|
||||
return {item[0] for item in _read_dashboard_pending()}
|
||||
|
||||
|
||||
def _find_snapshot_dependencies(path, key):
|
||||
"""Return UUIDs of still-pending snapshots that modified the same path+key."""
|
||||
try:
|
||||
pending = _pending_uuid_set()
|
||||
deps = []
|
||||
for fname in sorted(os.listdir(SNAPSHOTS_DIR)):
|
||||
if not fname.endswith('.json'):
|
||||
continue
|
||||
try:
|
||||
with open(os.path.join(SNAPSHOTS_DIR, fname)) as f:
|
||||
snap = json.load(f)
|
||||
if (snap.get('path') == path
|
||||
and snap.get('key') == str(key)
|
||||
and snap.get('uuid') in pending):
|
||||
deps.append(snap['uuid'])
|
||||
except Exception:
|
||||
pass
|
||||
return deps
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
def load_snapshot_for_uuid(entry_uuid):
|
||||
"""Return the snapshot dict for the given UUID, or None if not found."""
|
||||
try:
|
||||
for fname in os.listdir(SNAPSHOTS_DIR):
|
||||
if fname.endswith(f'-{entry_uuid}.json'):
|
||||
with open(os.path.join(SNAPSHOTS_DIR, fname)) as f:
|
||||
return json.load(f)
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def save_core_with_snapshot(new_core, path, key, operation, before, after,
|
||||
description='', cmd='core apply', queue=True):
|
||||
"""
|
||||
Write a .snapshots/{ts}-{uuid}.json file, save new_core to disk, and
|
||||
optionally create a pending queue entry. Returns a flash message string
|
||||
when queue=True, otherwise None.
|
||||
"""
|
||||
entry_uuid = str(uuid.uuid4())
|
||||
entry_ts = int(datetime.now().timestamp())
|
||||
current_user = session.get('email_address', 'unknown')
|
||||
|
||||
depends_on = _find_snapshot_dependencies(path, key)
|
||||
|
||||
os.makedirs(SNAPSHOTS_DIR, exist_ok=True)
|
||||
snapshot = {
|
||||
'uuid': entry_uuid,
|
||||
'ts': entry_ts,
|
||||
'cmd': cmd,
|
||||
'user': current_user,
|
||||
'operation': operation,
|
||||
'description': description,
|
||||
'path': path,
|
||||
'key': str(key),
|
||||
'before': before,
|
||||
'after': after,
|
||||
'depends_on': depends_on,
|
||||
}
|
||||
with open(os.path.join(SNAPSHOTS_DIR, f'{entry_ts}-{entry_uuid}.json'), 'w') as f:
|
||||
json.dump(snapshot, f, indent=2)
|
||||
|
||||
save_core(new_core)
|
||||
|
||||
if not queue:
|
||||
return None
|
||||
|
||||
if _apply_changes_immediately():
|
||||
with open(DASHBOARD_QUEUE, 'a') as f:
|
||||
f.write(f'{entry_uuid} {entry_ts} [{cmd}] ({current_user})\n')
|
||||
_trim_if_needed()
|
||||
else:
|
||||
_queue_pending_presigned(cmd, entry_uuid, entry_ts)
|
||||
|
||||
return _build_timing_msg(entry_ts, cmd)
|
||||
|
||||
|
||||
# ── Misc ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
def run_apply():
|
||||
try:
|
||||
subprocess.run(
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import json, re, subprocess, os, sys, html as html_mod
|
|||
import sanitize
|
||||
import validation as validate
|
||||
from datetime import datetime, timezone
|
||||
from config_utils import core_hash, get_pending_entries, get_dashboard_pending, _seconds_until_next_run, _format_timing, _is_locked, _lock_mtime, WEB_APP_DISPLAY_NAME, CONFIGS_DIR, DATA_DIR
|
||||
from config_utils import core_hash, get_pending_entries, get_dashboard_pending, load_snapshot_for_uuid, _seconds_until_next_run, _format_timing, _is_locked, _lock_mtime, WEB_APP_DISPLAY_NAME, CONFIGS_DIR, DATA_DIR
|
||||
|
||||
bp = Blueprint('view_page', __name__)
|
||||
|
||||
|
|
@ -594,16 +594,21 @@ def collect_tokens():
|
|||
pending_items = get_dashboard_pending()
|
||||
if pending_items:
|
||||
rows = ''
|
||||
for _uuid, ts, cmd, user, desc in pending_items:
|
||||
dt_str = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
|
||||
for _uuid, ts, cmd, user in pending_items:
|
||||
snap = load_snapshot_for_uuid(_uuid)
|
||||
dt_str = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
|
||||
snap_desc = e(snap['description']) if snap else ''
|
||||
before_html = _render_snap_val(snap.get('before') if snap else None)
|
||||
after_html = _render_snap_val(snap.get('after') if snap else None)
|
||||
snap_id = e(_uuid[:8]) if snap else ''
|
||||
rows += (f'<tr>'
|
||||
f'<td class="table-cell"><input type="checkbox" name="selected_uuids" value="{e(_uuid)}"/></td>'
|
||||
f'<td class="table-cell">{e(dt_str)}</td>'
|
||||
f'<td class="table-cell">{e(cmd)}</td>'
|
||||
f'<td class="table-cell">{e(desc)}</td>'
|
||||
f'<td class="table-cell"></td>'
|
||||
f'<td class="table-cell"></td>'
|
||||
f'<td class="table-cell"></td>'
|
||||
f'<td class="table-cell">{snap_desc}</td>'
|
||||
f'<td class="table-cell">{before_html}</td>'
|
||||
f'<td class="table-cell">{after_html}</td>'
|
||||
f'<td class="table-cell">{snap_id}</td>'
|
||||
f'<td class="table-cell">{e(user)}</td>'
|
||||
f'</tr>')
|
||||
select_all = (
|
||||
|
|
@ -769,6 +774,28 @@ def collect_tokens():
|
|||
def e(text):
|
||||
return html_mod.escape(str(text))
|
||||
|
||||
|
||||
def _render_snap_val(val):
|
||||
"""Return an HTML string for a snapshot before/after cell value."""
|
||||
if val is None:
|
||||
return ''
|
||||
if isinstance(val, dict) and len(val) == 1:
|
||||
k, v = next(iter(val.items()))
|
||||
text = f'{k}: {v}'
|
||||
elif isinstance(val, (dict, list)):
|
||||
text = json.dumps(val, separators=(',', ':'))
|
||||
else:
|
||||
text = str(val)
|
||||
trunc = (text[:23] + '…') if len(text) > 24 else text
|
||||
if trunc == text:
|
||||
return e(text)
|
||||
return (f'<details style="display:inline">'
|
||||
f'<summary style="cursor:pointer;list-style:none">{e(trunc)}</summary>'
|
||||
f'<pre style="margin:0.5rem 0;white-space:pre-wrap;font-size:0.85em">'
|
||||
f'{e(json.dumps(val, indent=2) if isinstance(val, (dict, list)) else text)}'
|
||||
f'</pre></details>')
|
||||
|
||||
|
||||
def apply_tokens(text, tokens):
|
||||
"""Substitute %TOKEN% placeholders. Values are NOT auto-escaped - callers
|
||||
that use results in HTML attribute or text context should call e() around
|
||||
|
|
|
|||
|
|
@ -608,7 +608,7 @@
|
|||
"items": [
|
||||
{
|
||||
"type": "form",
|
||||
"action": "/action/actions_cardpendingchanges_applyselected",
|
||||
"action": "/action/actions_cardpendingchanges_applynow",
|
||||
"method": "post",
|
||||
"items": [
|
||||
{
|
||||
|
|
@ -620,8 +620,8 @@
|
|||
"items": [
|
||||
{
|
||||
"type": "button_primary",
|
||||
"formaction": "/action/actions_cardpendingchanges_applyselected",
|
||||
"text": "Apply Selected"
|
||||
"formaction": "/action/actions_cardpendingchanges_applynow",
|
||||
"text": "Apply Now"
|
||||
},
|
||||
{
|
||||
"type": "button_secondary",
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue