kuma-strapper/backend/services/push_scripts.py

"""Bash script templates for push monitor cronjobs."""

from typing import Optional


def get_push_url(kuma_url: str, push_token: str) -> str:
    """Build the full push URL for Uptime Kuma."""
    base = kuma_url.rstrip("/")
    return f"{base}/api/push/{push_token}"


SCRIPT_TEMPLATES = {
    "heartbeat": '''#!/bin/bash
# Kuma Push Monitor - Heartbeat
# Simple "I'm alive" ping for remote/unreachable hosts
PUSH_URL="{push_url}"
HOSTNAME=$(hostname)

curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=up&msg=${{HOSTNAME}}%20alive"
''',

    "disk": '''#!/bin/bash
# Kuma Push Monitor - Disk Usage
# Fails if any mount exceeds threshold
PUSH_URL="{push_url}"
THRESHOLD=90

# Find worst disk usage (excluding tmpfs, devtmpfs, etc.)
WORST=$(df -h -x tmpfs -x devtmpfs -x squashfs 2>/dev/null | awk 'NR>1 && $5+0 > 0 {{gsub(/%/,"",$5); print $5, $6}}' | sort -rn | head -1)
USAGE=$(echo "$WORST" | awk '{{print $1}}')
MOUNT=$(echo "$WORST" | awk '{{print $2}}')

if [ -z "$USAGE" ]; then
    curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=up&msg=No%20disks%20found"
elif [ "$USAGE" -gt "$THRESHOLD" ]; then
    curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=down&msg=Disk%20$MOUNT%20at%20${{USAGE}}%25"
else
    curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=up&msg=Disk%20OK%20(worst:%20${{USAGE}}%25%20on%20$MOUNT)"
fi
''',

    "memory": '''#!/bin/bash
# Kuma Push Monitor - Memory Usage
# Fails if memory usage exceeds threshold
PUSH_URL="{push_url}"
THRESHOLD=90

USAGE=$(free | awk '/Mem:/ {{printf "%.0f", $3/$2 * 100}}')

if [ "$USAGE" -gt "$THRESHOLD" ]; then
    curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=down&msg=Memory%20at%20${{USAGE}}%25"
else
    curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=up&msg=Memory%20OK%20(${{USAGE}}%25)"
fi
''',

    "cpu": '''#!/bin/bash
# Kuma Push Monitor - CPU Load
# Fails if 5-minute load average exceeds threshold (as % of cores)
PUSH_URL="{push_url}"
THRESHOLD=95

CORES=$(nproc)
LOAD=$(awk '{{print $2}}' /proc/loadavg)  # 5-min average
PERCENT=$(awk "BEGIN {{printf \\"%.0f\\", ($LOAD / $CORES) * 100}}")

if [ "$PERCENT" -gt "$THRESHOLD" ]; then
    curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=down&msg=CPU%20load%20${{PERCENT}}%25%20(${{LOAD}}/${{CORES}})"
else
    curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=up&msg=CPU%20OK%20(${{PERCENT}}%25)"
fi
''',

    "updates": '''#!/bin/bash
# Kuma Push Monitor - Security Updates
# Fails if security updates are pending
PUSH_URL="{push_url}"

# Try apt (Debian/Ubuntu)
if command -v apt-get &>/dev/null; then
    apt-get update -qq 2>/dev/null
    SECURITY=$(apt-get -s upgrade 2>/dev/null | grep -c "^Inst.*security" || echo "0")
    if [ "$SECURITY" -gt 0 ]; then
        curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=down&msg=${{SECURITY}}%20security%20updates%20pending"
    else
        curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=up&msg=No%20security%20updates"
    fi
# Try dnf (RHEL/Fedora)
elif command -v dnf &>/dev/null; then
    SECURITY=$(dnf check-update --security 2>/dev/null | grep -c "^[a-zA-Z]" || echo "0")
    if [ "$SECURITY" -gt 0 ]; then
        curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=down&msg=${{SECURITY}}%20security%20updates%20pending"
    else
        curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=up&msg=No%20security%20updates"
    fi
# Try yum (older RHEL/CentOS)
elif command -v yum &>/dev/null; then
    SECURITY=$(yum check-update --security 2>/dev/null | grep -c "^[a-zA-Z]" || echo "0")
    if [ "$SECURITY" -gt 0 ]; then
        curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=down&msg=${{SECURITY}}%20security%20updates%20pending"
    else
        curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=up&msg=No%20security%20updates"
    fi
else
    curl -fsS -m 10 -o /dev/null "$PUSH_URL?status=up&msg=Unknown%20package%20manager"
fi
''',
}


def generate_script(push_metric: str, push_url: str) -> Optional[str]:
    """Generate a push monitoring script for the given metric type.

    Args:
        push_metric: One of 'heartbeat', 'disk', 'memory', 'cpu', 'updates'
        push_url: The full Uptime Kuma push URL including token

    Returns:
        The bash script content, or None if invalid metric type
    """
    template = SCRIPT_TEMPLATES.get(push_metric)
    if not template:
        return None
    return template.format(push_url=push_url)


def get_script_filename(push_metric: str, monitor_id: int) -> str:
    """Generate the script filename for a push monitor.

    Args:
        push_metric: The metric type
        monitor_id: The Uptime Kuma monitor ID

    Returns:
        Filename like 'kuma-push-disk-123.sh'
    """
    return f"kuma-push-{push_metric}-{monitor_id}.sh"


def get_script_path(push_metric: str, monitor_id: int) -> str:
    """Generate the full path for a push monitor script.

    Args:
        push_metric: The metric type
        monitor_id: The Uptime Kuma monitor ID

    Returns:
        Full path like '/usr/local/bin/kuma-push-disk-123.sh'
    """
    return f"/usr/local/bin/{get_script_filename(push_metric, monitor_id)}"


def get_cronjob_entry(push_metric: str, monitor_id: int, interval_minutes: int = 5) -> str:
    """Generate the crontab entry for a push monitor script.

    Args:
        push_metric: The metric type
        monitor_id: The Uptime Kuma monitor ID
        interval_minutes: How often to run (default 5 minutes)

    Returns:
        Crontab entry string
    """
    script_path = get_script_path(push_metric, monitor_id)
    return f"*/{interval_minutes} * * * * {script_path} >/dev/null 2>&1"