linuxrouter/docker/routlin-dash/app/config_utils.py

import copy, json, subprocess, hashlib, os, uuid
from datetime import datetime, timezone
from flask import session

CONFIGS_DIR        = '/routlin_location'
DATA_DIR           = '/data'
ACCOUNTS_FILE      = f'{DATA_DIR}/authorized_accounts.json'
CONFIG_FILE        = f'{CONFIGS_DIR}/config.json'
DASHBOARD_QUEUE    = f'{CONFIGS_DIR}/.dashboard-queue'
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


def load_config():
    try:
        with open(CONFIG_FILE) as f:
            return json.load(f)
    except Exception:
        return {}


def save_config(data):
    with open(CONFIG_FILE, 'w') as f:
        json.dump(data, f, indent=2)


def config_hash():
    try:
        with open(CONFIG_FILE, 'rb') as f:
            return hashlib.md5(f.read()).hexdigest()
    except Exception:
        return ''


def verify_config_hash(submitted):
    if not submitted:
        return True
    return submitted == config_hash()


def _load_done_set():
    try:
        done = set()
        for line in open(DASHBOARD_DONE).read().splitlines():
            parts = line.split()
            if parts:
                done.add(parts[0])
        return done
    except Exception:
        return set()


def _read_pending(done_set):
    pending = []
    try:
        lines = open(DASHBOARD_QUEUE).read().splitlines()
    except Exception:
        return pending
    for line in lines:
        try:
            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 ''
                if entry_uuid not in done_set:
                    pending.append((entry_uuid, int(entry_ts), entry_cmd, entry_user))
        except Exception:
            pass
    return pending


def get_pending_entries():
    return _read_pending(_load_done_set())


def _format_timing(secs):
    if secs is None:
        return None
    if secs <= 5:
        return 'momentarily'
    if secs < 60:
        return f'in about {secs} seconds'
    mins = round(secs / 60)
    return f'in about {mins} minute{"s" if mins != 1 else ""}'


def _trim_if_needed():
    try:
        lines = [l for l in open(DASHBOARD_QUEUE).read().splitlines() if l]
        if len(lines) <= QUEUE_MAX_LINES:
            return
        done_set = _load_done_set()
        pending = [l for l in lines if l.split()[0] not in done_set]
        with open(DASHBOARD_QUEUE, 'w') as f:
            f.write('\n'.join(pending) + ('\n' if pending else ''))
        open(DASHBOARD_DONE, 'w').close()
    except Exception:
        pass


def _apply_changes_immediately():
    try:
        return session.get('apply_changes_immediately', False)
    except Exception:
        return False


def _read_dashboard_pending():
    """Return list of (uuid, ts, cmd, user) from .dashboard-pending."""
    items = []
    try:
        lines = open(DASHBOARD_PENDING).read().splitlines()
    except Exception:
        return items
    for line in lines:
        if not line.strip():
            continue
        try:
            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))
        except Exception:
            pass
    return items


def get_dashboard_pending():
    return _read_dashboard_pending()


def get_dashboard_done():
    """Return list of (uuid, applied_ts) from .dashboard-done, newest first."""
    items = []
    try:
        lines = open(DASHBOARD_DONE).read().splitlines()
    except Exception:
        return items
    for line in lines:
        if not line.strip():
            continue
        try:
            parts = line.split(None, 1)
            if len(parts) >= 2:
                items.append((parts[0], int(parts[1])))
            elif len(parts) == 1:
                items.append((parts[0], None))
        except Exception:
            pass
    items.reverse()
    return items


def flush_pending_to_queue():
    """Move all entries from .dashboard-pending to .dashboard-queue and clear pending."""
    items = _read_dashboard_pending()
    if not items:
        return
    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 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()
    _trim_if_needed()




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 in existing:
        if entry_cmd == cmd and entry_user == current_user:
            return entry_uuid, entry_ts
    entry_uuid = str(uuid.uuid4())
    entry_ts   = int(datetime.now().timestamp())
    with open(DASHBOARD_PENDING, 'a') as f:
        f.write(f'{entry_uuid} {entry_ts} [{cmd}] ({current_user})\n')
    return entry_uuid, entry_ts


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)
    done_set = _load_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())
    entry_ts   = int(datetime.now().timestamp())
    with open(DASHBOARD_QUEUE, 'a') as f:
        f.write(f'{entry_uuid} {entry_ts} [{cmd}] ({current_user})\n')
    _trim_if_needed()
    return entry_uuid, entry_ts


def _entry_ts_from_queue(entry_uuid):
    try:
        for line in open(DASHBOARD_QUEUE).read().splitlines():
            parts = line.split(None, 2)
            if len(parts) >= 2 and parts[0] == entry_uuid:
                return int(parts[1])
    except Exception:
        pass
    return None


def _seconds_until_next_run():
    try:
        last_run = float(open(DASHBOARD_LAST_RUN).read().strip())
        elapsed  = datetime.now(timezone.utc).timestamp() - last_run
        return int(max(0, DASHB_INTERVAL_SECS - elapsed))
    except Exception:
        return None


def _is_locked():
    try:
        return os.path.getsize(DASHBOARD_LOCK) > 0
    except Exception:
        return False


def _lock_mtime():
    try:
        return os.path.getmtime(DASHBOARD_LOCK)
    except Exception:
        return None


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():
        mtime = _lock_mtime()
        if entry_ts is not None and mtime and entry_ts < mtime:
            return f'{action_label}. Changes are being applied now.'
        return f'{action_label}. Changes will be applied on the next run.'
    timing = _format_timing(_seconds_until_next_run())
    if timing:
        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()
    cli_cmd     = f'sudo python3 {parts[0]}.py --{parts[1]}' if len(parts) == 2 else cmd
    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 _items_match(item, ref):
    """Return True if item and ref refer to the same entity by a common identifier field."""
    if not isinstance(item, dict) or not isinstance(ref, dict):
        return item == ref
    for field in ('ip', 'name', 'mac_address', 'host', 'id', 'address'):
        if field in ref and field in item:
            return item[field] == ref[field]
    return item == ref


def revert_snapshot_to_config(entry_uuid):
    """Apply the inverse of a snapshot to config.json and queue a new pending change.

    Returns (flash_message, success_bool).
    """
    snap = load_snapshot_for_uuid(entry_uuid)
    if not snap:
        return f'Snapshot not found for {entry_uuid[:8]}.', False

    path      = snap['path']
    key       = snap['key']
    before    = snap['before']   # original state to restore
    after     = snap['after']    # applied state to undo
    operation = snap['operation']

    if operation == 'revert':
        return 'This change is already a revert; cannot revert again.', False

    core = load_config()

    if key == 'global':
        if before is None:
            core.pop(path, None)
        else:
            core[path] = before
    else:
        items = core.setdefault(path, [])
        if operation == 'add':
            core[path] = [x for x in items if not _items_match(x, after)]
        elif operation == 'delete':
            if before:
                core[path].append(before)
        else:
            if before and after:
                for i, item in enumerate(items):
                    if _items_match(item, after):
                        items[i] = before
                        break

    msg = save_config_with_snapshot(
        core, path=path, key=key, operation='revert',
        before=after, after=before,
        description=f"Reverted: {snap.get('description', '')}",
        cmd=snap.get('cmd', 'core apply'),
    )
    return msg or 'Reverted.', True


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_config_with_snapshot(new_config, path, key, operation, before, after,
                             description='', cmd='core apply', queue=True):
    """
    Write a .snapshots/{ts}-{uuid}.json file, save new_config to disk, and
    optionally create a pending queue entry.  Returns a flash message string.

    queue=False: skips queueing and records the change directly in
    .dashboard-done so it appears in Change History without a pending step.
    """
    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_config(new_config)

    if not queue:
        with open(DASHBOARD_DONE, 'a') as f:
            f.write(f'{entry_uuid} {entry_ts}\n')
        return 'Saved.'

    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(
            ['python3', f'{CONFIGS_DIR}/core.py', '--apply'],
            capture_output=True, timeout=30
        )
    except Exception:
        pass


def run_update_blocklists():
    try:
        subprocess.run(
            ['python3', f'{CONFIGS_DIR}/core.py', '--update-blocklists'],
            capture_output=True, timeout=120
        )
    except Exception:
        pass