Nym Mixnet & dVPN: A Node Operator's Guide: Automatic Upgrades

README.md

This gist belongs to the blog post Nym Mixnet & dVPN: A Node Operator's Guide.

nym-update.py

#!/usr/bin/env python3
import os
import json
import requests
from datetime import datetime, timedelta, timezone
import click
from loguru import logger

# Usage Notes
# curl -LsSf https://astral.sh/uv/install.sh | sh
# cd /root/nym-update
# uv venv py312nu --python 3.12 --seed
# source /root/nym-update/py312nu/bin/activate
# uv pip install -r ./requirements.in
# /root/nym-update/py312nu/bin/python /root/nym-update/nym-update.py --test-alert
# /root/nym-update/py312nu/bin/python /root/nym-update/nym-update.py --debug
#
# cp nym-update.service /etc/systemd/system/nym-update.service
# cp nym-update.timer   /etc/systemd/system/nym-update.timer
#
# sudo systemctl daemon-reload
# sudo systemctl enable nym-update.timer
# sudo systemctl start nym-update.timer
# sudo journalctl -u nym-update.service -f -n 100

# Configuration Constants
INSTALL_DIR = "/root"
RELEASE_INFO_FILE = INSTALL_DIR + "/release_info.json"  # Adjust path as needed
GITHUB_RELEASES_URL = "https://api.github.com/repos/nymtech/nym/releases"
GITHUB_REF_URL_TEMPLATE = "https://api.github.com/repos/nymtech/nym/git/ref/tags/{tag}"
ASSET_NAME = "nym-node"
CHECK_INTERVAL_DAYS = 2

def pushover_alert(which='nym-binaries-v2025.3-ruta'):
    """
    Sends a Pushover alert with basic HTML formatting.
    
    You may want to store your actual 'token' and 'user' in a secure place,
    rather than hard-coding them here.
    """
    message = ('<font color="#fc0303">' + which + '</font> ' + '<a href="https://github.com/nymtech/nym/releases/tag/' + which + '">' + which + '</a>\n\n' + 'https://github.com/nymtech/nym/releases/tag/' + which)
    data = {
        'token': '...',   # Replace with your actual Pushover token
        'user':  '...',   # Replace with your actual Pushover user key
        'sound': 'none',
        'title': 'New Nym Binaries: ' + which,
        'html':  1,
        'message': message,
    }
    try:
        r = requests.post("https://api.pushover.net/1/messages.json", data=data)
        r.raise_for_status()
        logger.info("Pushover alert sent successfully: {}", r.text)
    except Exception as e:
        logger.error("Failed to send Pushover alert: {}", e)


def fetch_latest_release():
    """
    Fetch the latest release from GitHub whose 'name' contains 'Nym Binaries'.
    Return the full JSON release object.
    """
    logger.debug("Fetching releases from GitHub: {}", GITHUB_RELEASES_URL)
    try:
        response = requests.get(GITHUB_RELEASES_URL, timeout=15)
        response.raise_for_status()
    except Exception as e:
        logger.error("Error fetching releases: {}", e)
        raise

    releases = response.json()
    # Filter for the first release whose name includes "Nym Binaries"
    for release in releases:
        if "Nym Binaries" in release.get("name", ""):
            logger.debug("Found release matching 'Nym Binaries': {}", release["tag_name"])
            return release

    logger.error("No release found with a name that includes 'Nym Binaries'.")
    return None


def fetch_commit_id(tag: str) -> str:
    """
    Given a tag (e.g., 'nym-binaries-v2025.3-ruta'), fetch the commit ID via the
    GitHub Git Reference API endpoint.
    """
    url = GITHUB_REF_URL_TEMPLATE.format(tag=tag)
    logger.debug("Fetching commit ID from: {}", url)
    try:
        response = requests.get(url, timeout=15)
        response.raise_for_status()
    except Exception as e:
        logger.error("Error fetching commit info: {}", e)
        raise

    data = response.json()
    commit_id = data.get("object", {}).get("sha")
    if not commit_id:
        logger.error("Could not extract commit ID for tag {}", tag)
        raise ValueError("Commit ID not found.")
    logger.debug("Commit ID for {} is: {}", tag, commit_id)
    return commit_id


def get_asset_download_url(release: dict, asset_name: str) -> str:
    """
    Given a release object and the asset name (e.g., 'nym-node'),
    return the browser_download_url for that asset.
    """
    assets = release.get("assets", [])
    for asset in assets:
        if asset.get("name") == asset_name:
            logger.debug("Found asset {} with URL: {}", asset_name, asset["browser_download_url"])
            return asset["browser_download_url"]
    logger.error("Asset '{}' not found in the release assets.", asset_name)
    return ""


def load_saved_release_info() -> dict:
    """
    Load previously stored release information (tag, commit, timestamp).
    Return None if no file or an error occurs.
    """
    if not os.path.exists(RELEASE_INFO_FILE):
        logger.debug("No saved release info file at {}.", RELEASE_INFO_FILE)
        return None
    try:
        with open(RELEASE_INFO_FILE, "r") as f:
            data = json.load(f)
            logger.debug("Loaded release info: {}", data)
            return data
    except Exception as e:
        logger.error("Error reading release info file: {}", e)
        return None


def save_release_info(tag: str, commit: str, timestamp: datetime) -> None:
    """
    Save current release information (tag, commit, timestamp).
    """
    data = {
        "tag": tag,
        "commit": commit,
        "timestamp": timestamp.isoformat(),
        "downloaded": False,
    }
    try:
        with open(RELEASE_INFO_FILE, "w") as f:
            json.dump(data, f)
        logger.debug("Saved release info: {}", data)
    except Exception as e:
        logger.error("Error saving release info: {}", e)


def download_asset(url: str, destination: str) -> None:
    """
    Download the asset from 'url' and save to 'destination'.
    """
    logger.info("Starting download from: {}", url)
    try:
        with requests.get(url, stream=True, timeout=30) as r:
            r.raise_for_status()
            with open(destination, "wb") as f:
                for chunk in r.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)
        logger.info("Downloaded asset to {}", destination)
    except Exception as e:
        logger.error("Error downloading asset: {}", e)
        raise


def compute_install_path(tag: str) -> str:
    """
    Convert the tag, e.g., 'nym-binaries-v2025.3-ruta', into
    '/root/nym-node-v2025.3-ruta'
    """
    suffix = tag.replace("nym-binaries-", "")  # e.g., 'v2025.3-ruta'
    filename = f"nym-node-{suffix}"           # e.g., 'nym-node-v2025.3-ruta'
    return os.path.join(INSTALL_DIR, filename)


@click.command()
@click.option('-d', '--debug', is_flag=True, help='Enable debug logging.')
@click.option('--test-alert', is_flag=True, help='Test the pushover alert and exit.')
def main(debug, test_alert):
    """
    CLI tool to check for updates to the 'nym-node' binary on GitHub.
    
    It records a new release's tag, commit ID, and timestamp in a local file,
    waits 2 days, and if no newer releases appear in that window, downloads the file
    and places it in /root/ with a name derived from the tag.
    
    Example usage:
      python3 nym_update.py
      python3 nym_update.py --debug
      python3 nym_update.py --test-alert
    """
    # Configure logging
    if debug:
        logger.remove()
        logger.add(lambda msg: click.echo(msg, err=True), level="DEBUG")
    else:
        logger.remove()
        logger.add(lambda msg: click.echo(msg, err=True), level="INFO")

    # If we're just testing the alert, do that and exit
    if test_alert:
        logger.info("Testing Pushover alert...")
        pushover_alert("test-release-alert")
        return

    # 1. Fetch the latest "Nym Binaries" release
    try:
        release = fetch_latest_release()
        if not release:
            logger.error("No release found. Exiting.")
            return
        tag = release["tag_name"]
        commit = fetch_commit_id(tag)
    except Exception as e:
        logger.error("Failed to fetch release or commit info: {}", e)
        return

    # 2. Load what we previously saved
    saved_info = load_saved_release_info()
    # now = datetime.utcnow()
    now = datetime.now(timezone.utc)

    # 3. If we have saved info, check if it's the same release
    if saved_info:
        saved_tag = saved_info.get("tag")
        saved_timestamp_str = saved_info.get("timestamp")
        try:
            saved_timestamp = datetime.fromisoformat(saved_timestamp_str)
        except Exception as e:
            logger.error("Error parsing saved timestamp: {}", e)
            saved_timestamp = now

        if saved_info.get("tag") == tag and saved_info.get("downloaded"):
            logger.info("Release {} was already downloaded. No further action required.", tag)
            return

        if tag != saved_tag:
            # There's a new tag, so save it and wait for the next run
            logger.info("New release detected: {} (old was {})", tag, saved_tag)
            save_release_info(tag, commit, now)
            return
        else:
            # Same tag as before -> check if enough days have passed
            elapsed = now - saved_timestamp
            if elapsed < timedelta(days=CHECK_INTERVAL_DAYS):
                logger.info(
                    "Release {} is still within the {}-day wait window. "
                    "Elapsed: {}. Waiting for more updates or final confirmation.",
                    tag, CHECK_INTERVAL_DAYS, elapsed
                )
                return
            else:
                logger.info("Release {} has been stable for {} days. Proceeding with download.", tag, CHECK_INTERVAL_DAYS)
    else:
        # No previously stored info - store info now and wait for next run
        logger.info("No release info stored. Saving current release info and waiting for stabilization.")
        save_release_info(tag, commit, now)
        return

    # 4. Download the asset if we have waited the required days
    download_url = get_asset_download_url(release, ASSET_NAME)
    if not download_url:
        logger.error("No download URL found for asset '{}'. Aborting.", ASSET_NAME)
        return

    install_path = compute_install_path(tag)
    logger.info("Downloading '{}' to '{}'.", ASSET_NAME, install_path)
    try:
        download_asset(download_url, install_path)
    except Exception as e:
        logger.error("Download failed: {}", e)
        return

    # 5. Send an alert
    pushover_alert(tag)
    logger.info("Update complete. New binary installed at '{}'.", install_path)

    # Mark release as downloaded so we don't re-download it next time
    saved_info = load_saved_release_info()
    if saved_info:
        saved_info["downloaded"] = True
    try:
        with open(RELEASE_INFO_FILE, "w") as f:
            json.dump(saved_info, f)
        logger.debug("Marked release as downloaded: {}", saved_info)
    except Exception as e:
        logger.error("Error marking release as downloaded: {}", e)


if __name__ == "__main__":
    main()

nym-update.service

[Unit]
Description=Check and Update Nym Binary

[Service]
Type=oneshot
ExecStart=/root/nym-update/py312nu/bin/python /root/nym-update/nym-update.py

nym-update.timer

[Unit]
Description=Run Nym Update Check Daily at 4:00 AM UTC

[Timer]
OnCalendar=*-*-* 04:00:00 UTC
Persistent=true

[Install]
WantedBy=timers.target

requirements.in

click
loguru
requests