TA-ProxMenu/inc/notifications.py

#!/usr/bin/env python3
"""TAPM notification policy UI and scheduled Proxmox VE version checks."""
import argparse
import fcntl
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import uuid

HERE = Path(__file__).resolve().parent
UNIT_DIR = Path('/etc/systemd/system')
TEMPLATE_DIR = Path('/etc/pve/notification-templates/default')
STATE_DIR = Path('/var/lib/ta-proxmenu/notifications')
SERVICE = 'tapm-version-notifications'
LABELS = {
    'all': 'All available package updates',
    'version': 'Proxmox VE version updates only',
    'disabled': 'Disable update notifications',
}
NAME = re.compile(r'[A-Za-z0-9][A-Za-z0-9_.-]{0,62}\Z')
VERSION = re.compile(r'[0-9][A-Za-z0-9.+:~\-]{0,100}\Z')


class NotificationError(Exception):
    pass


def command(args, payload=None, timeout=30):
    try:
        result = subprocess.run(args, input=payload, capture_output=True,
                                text=True, timeout=timeout)
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise NotificationError('The requested notification operation could not be completed.') from exc
    if result.returncode:
        raise NotificationError('The requested notification operation failed. Check Proxmox configuration and connectivity.')
    return result.stdout


def api(method, path, **params):
    args = ['pvesh', method, path, '--output-format', 'json']
    for key, value in params.items():
        args += ['--' + key.replace('_', '-'), str(value)]
    raw = command(args)
    try:
        return json.loads(raw) if raw.strip() else None
    except ValueError as exc:
        raise NotificationError('Proxmox returned an unreadable response.') from exc


def bridge(action, **params):
    raw = command(['perl', str(HERE / 'notifications-bridge.pl')],
                  json.dumps(dict(action=action, **params)), timeout=75)
    try:
        return json.loads(raw)
    except ValueError as exc:
        raise NotificationError('Notification configuration could not be read.') from exc


def local_node():
    name = command(['hostname', '-s']).strip()
    if not NAME.fullmatch(name):
        raise NotificationError('The host name is not valid.')
    return name


def effective_mode(snapshot):
    policy = snapshot.get('policy', {})
    native = snapshot.get('notify', {}).get('package-updates', 'auto')
    if policy.get('mode') == 'version' and native == 'never':
        return 'version'
    return {'always': 'all', 'never': 'disabled'}.get(native, 'auto')


def current_status():
    snapshot = bridge('read')
    mode = effective_mode(snapshot)
    print('Update notification policy: ' + LABELS.get(mode, 'Proxmox default (subscription-dependent)'))
    print('Scope: this Proxmox datacenter (all nodes when clustered).')
    if mode == 'version':
        owner = snapshot['policy'].get('owner', 'unknown')
        print('Hourly version checker host: ' + owner)
        if owner == local_node():
            try:
                command(['systemctl', 'is-enabled', SERVICE + '.timer'])
                command(['systemctl', 'is-active', SERVICE + '.timer'])
                print('Version checker timer: enabled and active')
            except NotificationError:
                print('Version checker timer: needs attention; reapply the version-only policy.')
        else:
            print('Timer status must be checked on that host; checks pause while it is offline.')
    return snapshot


def atomic_write(path, content):
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(path.name + '.' + uuid.uuid4().hex + '.tmp')
    try:
        with temporary.open('x') as stream:
            stream.write(content)
        os.replace(temporary, path)
    finally:
        if temporary.exists():
            temporary.unlink()


def prepare_checker():
    # Separate unit: never disable or replace Proxmox's daily update service.
    service = f'''[Unit]
Description=TAPM Proxmox VE version notifications
After=pve-cluster.service network-online.target
Wants=network-online.target

[Service]
Type=oneshot
ExecStart=/usr/bin/python3 {HERE}/notifications.py run
TimeoutStartSec=15min
UMask=0077
'''
    timer = '''[Unit]
Description=Check for Proxmox VE version updates hourly

[Timer]
OnCalendar=hourly
RandomizedDelaySec=5min
Persistent=true

[Install]
WantedBy=timers.target
'''
    atomic_write(UNIT_DIR / (SERVICE + '.service'), service)
    atomic_write(UNIT_DIR / (SERVICE + '.timer'), timer)
    templates = {
        'subject.txt': 'Proxmox VE version update available on {{hostname}}\n',
        'body.txt': 'Host: {{hostname}}\nInstalled Proxmox VE version: {{installed}}\nAvailable Proxmox VE version: {{available}}\n\nA newer version is available from this host\'s configured repositories.\n',
        'body.html': '<p>Host: {{hostname}}</p><p>Installed Proxmox VE version: {{installed}}<br>Available Proxmox VE version: {{available}}</p><p>A newer version is available from this host&#39;s configured repositories.</p>\n',
    }
    for suffix, content in templates.items():
        atomic_write(TEMPLATE_DIR / ('tapm-pve-version-' + suffix + '.hbs'), content)
    command(['systemctl', 'daemon-reload'])
    command(['systemctl', 'enable', '--now', SERVICE + '.timer'])
    command(['systemctl', 'is-active', SERVICE + '.timer'])


def apply_policy(mode):
    expected = current_status()
    owner = local_node()
    print('\nProposed policy: ' + LABELS[mode])
    print('This changes update alerts across this datacenter. Other notification types retain their settings.')
    print('Normal Proxmox repository checks remain enabled. No packages will be upgraded.')
    if mode == 'version':
        print(f'An hourly checker on {owner} will inspect online nodes using their existing repository indexes.')
        print('Checks pause while this host is offline; reapply here to repair, or on another host to move the checker.')
        print('Only new Proxmox VE versions trigger alerts, including patch releases. Kernel-only updates do not.')
        print('Existing update notification routing and destinations are used. No post-upgrade alerts are sent.')
    elif mode == 'all':
        print('The native Proxmox daily update job sends alerts through existing notification routing.')
    if input('Apply this policy (type yes to continue)? ').strip().lower() != 'yes':
        print('No changes made.')
        return
    if mode == 'version':
        prepare_checker()
    policy = dict(mode=mode, owner=owner, generation=uuid.uuid4().hex)
    bridge('apply', expected=expected, policy=policy)
    print('Update notification policy applied: ' + LABELS[mode])
    if mode != 'version' and (UNIT_DIR / (SERVICE + '.timer')).exists():
        try:
            command(['systemctl', 'disable', '--now', SERVICE + '.timer'])
        except NotificationError:
            print('The local timer could not be stopped. It will remain inactive in this notification mode.')
    print('Previous policy saved in /etc/pve/priv/tapm-update-policy.previous.json.')


def candidate_update(rows):
    if not isinstance(rows, list):
        raise NotificationError('The update list could not be read.')
    for row in rows:
        if not isinstance(row, dict) or row.get('Package') != 'pve-manager':
            continue
        installed, available = row.get('OldVersion', ''), row.get('Version', '')
        if not isinstance(installed, str) or not isinstance(available, str) or not VERSION.fullmatch(installed) or not VERSION.fullmatch(available):
            raise NotificationError('The Proxmox VE versions could not be read.')
        # Use Debian version ordering, including epochs, revisions and tildes.
        comparison = subprocess.run(['dpkg', '--compare-versions', available, 'gt', installed],
                                    capture_output=True, timeout=10)
        if comparison.returncode == 0:
            return installed, available
        if comparison.returncode != 1:
            raise NotificationError('The Proxmox VE versions could not be compared.')
    return None


def check_versions():
    snapshot = bridge('read')
    if effective_mode(snapshot) != 'version' or snapshot['policy'].get('owner') != local_node():
        return 0
    nodes = api('get', '/nodes')
    if not isinstance(nodes, list) or not nodes:
        raise NotificationError('The Proxmox node list could not be read.')
    failures = 0
    for node in nodes:
        name = node.get('node', '') if isinstance(node, dict) else ''
        if not NAME.fullmatch(name):
            failures += 1
            continue
        if node.get('status') != 'online':
            print(f'{name}: offline; version check deferred.')
            failures += 1
            continue
        try:
            versions = candidate_update(api('get', f'/nodes/{name}/apt/update'))
            if not versions:
                continue
            installed, available = versions
            result = bridge('send', generation=snapshot['policy']['generation'],
                            node=name, installed=installed, available=available)
            if result.get('sent'):
                print(f'{name}: Proxmox VE {installed} -> {available}; notification submitted to configured routing.')
        except (NotificationError, OSError, subprocess.TimeoutExpired):
            print(f'{name}: version check or notification failed; will retry on the next check.')
            failures += 1
    return 1 if failures else 0


def delivery_status():
    current_status()
    targets = api('get', '/cluster/notifications/targets')
    matchers = api('get', '/cluster/notifications/matchers')
    print('\nNotification destinations:')
    for target in targets:
        state = 'disabled' if target.get('disable') else 'enabled'
        print(f"  {target['name']} ({target['type']}, {state})")
    if not targets:
        print('  None configured.')
    print('\nRouting rules:')
    for matcher in matchers:
        state = 'disabled' if matcher.get('disable') else 'enabled'
        print(f"  {matcher['name']} ({state})")
        for key in ('target', 'match-field', 'match-severity', 'match-calendar', 'mode', 'invert-match'):
            if key in matcher:
                print(f"    {key}: {matcher[key]}")
    print('\nConfigure recipients, delivery credentials and routing in the Proxmox Web UI: Datacenter > Notifications.')
    print('Version-only alerts use the existing system-update routing rules.')


def test_delivery():
    targets = [t for t in api('get', '/cluster/notifications/targets') if not t.get('disable')]
    if not targets:
        print('No enabled notification destinations. Configure one in Datacenter > Notifications first.')
        return
    for index, target in enumerate(targets, 1):
        print(f"  {index}. {target['name']} ({target['type']})")
    choice = input('Destination number (Enter to cancel): ').strip()
    if not choice:
        return
    if not choice.isdigit() or not 1 <= int(choice) <= len(targets):
        raise NotificationError('Choose a destination from the list.')
    target = targets[int(choice) - 1]['name']
    if not NAME.fullmatch(target):
        raise NotificationError('The destination name is not valid.')
    if input(f'Send a test notification to {target} (type yes to continue)? ').strip().lower() != 'yes':
        return
    api('create', f'/cluster/notifications/targets/{target}/test')
    print('Test notification submitted. Confirm receipt at the destination; this tests delivery, not routing rules.')


def main():
    parser = argparse.ArgumentParser(description='Proxmox notification settings')
    parser.add_argument('action', choices=['status', 'delivery', 'test', 'apply', 'run'])
    parser.add_argument('mode', nargs='?', choices=list(LABELS))
    args = parser.parse_args()
    if os.geteuid() != 0:
        print('Run this action as root on a Proxmox host.', file=sys.stderr)
        return 1
    try:
        if args.action == 'status':
            current_status()
        elif args.action == 'delivery':
            delivery_status()
        elif args.action == 'test':
            test_delivery()
        elif args.action == 'apply':
            if args.mode is None:
                raise NotificationError('Select an update notification policy.')
            apply_policy(args.mode)
        else:
            STATE_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
            with (STATE_DIR / 'check.lock').open('a') as lock:
                try:
                    fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
                except BlockingIOError:
                    return 0
                return check_versions()
    except (EOFError, KeyboardInterrupt):
        print('\nCancelled.')
        return 1
    except (NotificationError, OSError, ValueError, KeyError, TypeError) as exc:
        print(str(exc) if isinstance(exc, NotificationError) else 'Notification operation failed; verify configuration and permissions.', file=sys.stderr)
        return 1
    return 0


if __name__ == '__main__':
    sys.exit(main())