BetterUptime automatic update of IPs for iptable

betteruptime.py

import requests, hashlib, os, ipaddress, subprocess
from typing import List

BETTERUPTIME_URL = 'https://uptime.betterstack.com/ips.txt'


def run(cmd: List[str], check: bool = True, capture: bool = False) -> subprocess.CompletedProcess:
    return subprocess.run(
        cmd,
        check=check,
        text=True,
        stdout=subprocess.PIPE if capture else None,
        stderr=subprocess.PIPE if capture else None,
    )


def process(content):
    ips_v4 = []
    ips_v6 = []
    for line in content.splitlines():
        line = line.strip()
        if not line or line.startswith('#'):
            continue

        if line.find(':') > -1:
            ips_v6.append(line)
        else:
            ips_v4.append(line)

    # Create temporary ipsets
    run(['ipset', 'create', 'better_uptime_v4_tmp', 'hash:net', 'family', 'inet', '-exist'], check=True)
    run(['ipset', 'create', 'better_uptime_v6_tmp', 'hash:net', 'family', 'inet6', '-exist'], check=True)

    for ip in ips_v4:
        run(['ipset', 'add', 'better_uptime_v4_tmp', ip, '-exist'], check=True)

    for ip in ips_v6:
        run(['ipset', 'add', 'better_uptime_v6_tmp', ip, '-exist'], check=True)

    run(['ipset', 'swap', 'better_uptime_v4', 'better_uptime_v4_tmp'], check=True)
    run(['ipset', 'swap', 'better_uptime_v6', 'better_uptime_v6_tmp'], check=True)

    run(['ipset', 'destroy', 'better_uptime_v4_tmp'], check=True)
    run(['ipset', 'destroy', 'better_uptime_v6_tmp'], check=True)


def main():
    # First, we load the Better Uptime IP list
    response = requests.get(BETTERUPTIME_URL)
    try:
        response.raise_for_status()
    except requests.HTTPError as e:
        print(f"Failed to fetch Better Uptime IP list: {e}")
        exit(1)

    # Now, we hash the file contents
    content = response.text
    sha256_hash = hashlib.sha256(content.encode('utf-8')).hexdigest()

    # And we compare to local BetterUptime file hash located at /tmp/betteruptime-ips.txt.sha256
    try:
        with open('/tmp/betteruptime-ips.txt.sha256', 'r') as f:
            local_hash = f.read().strip()
    except FileNotFoundError:
        local_hash = ''
    
    if sha256_hash == local_hash:
        print("Better Uptime IP list is up to date.")
        return

    # If we reach here, the file has changed
    with open('/tmp/betteruptime-ips.txt.sha256', 'w') as f:
        f.write(sha256_hash)

    process(content)
    print("Better Uptime IP list successfully updated.")


if __name__ == '__main__':
    if os.geteuid() != 0:
        raise SystemExit("This script must be run as root (ipset requires privileges).")

    main()