scripts/voixicron.py

#! /usr/bin/env python3
"""
This script checks for available updates on Void Linux using xbps and notifies
the configured administrator account about them. It currently requires
python 3.x and no additional packages and is supposed to be run with cron or
another scheduled execution mechanism.
"""

import subprocess
import smtplib
import socket
import re
import logging
import sys

from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText


# Email server to be used to send the notification
ADMIN_EMAIL = "root"
# Mail server to be used to send the notification
MAIL_SERVER = "localhost"
# Port to use on the mail server
MAIL_SERVER_PORT = 25
# User name to log in to the mail server
MAIL_SERVER_USER = ""
# Password to log in to the mail server
MAIL_SERVER_PASSWORD = ""
# Log level for the stderr/stdout output
LOG_LEVEL = logging.ERROR

AVAILABLE_UPDATES = []

DEVNULL = open("/dev/null", "w")

LOGGER = logging.getLogger("voixicron")
LOGGER.setLevel(LOG_LEVEL)
CONSOLE_HANDLER = logging.StreamHandler()
LOGGER.addHandler(CONSOLE_HANDLER)

try:
    XBPS_INSTALL_RESULT = subprocess.check_output(["xbps-install", "-Sun"],
                                                  stderr=DEVNULL)
except subprocess.CalledProcessError as error:
    LOGGER.error("Could not get list of updated packages, xbps-install error:"
                 " %s", error.output)
    sys.exit(1)
else:
    XBPS_INSTALL_RESULT = XBPS_INSTALL_RESULT.decode("utf-8").split("\n")
    for line in XBPS_INSTALL_RESULT:
        if "update" in line:
            try:
                package = line.split(" ")[0]
                package_regex = "^([A-Za-z0-9-._]*)-([0-9].*_[0-9]*)$"
                match_object = re.match(package_regex, package)
                if match_object:
                    package_name = match_object.groups()[0]
                    new_package_version = match_object.groups()[1]
                else:
                    LOGGER.error("Not match for package %s", package)
                    continue
            except IndexError:
                continue
            else:
                try:
                    xbps_query_result = subprocess.check_output(
                        ["xbps-query", "--show", package_name], stderr=DEVNULL)
                except subprocess.CalledProcessError as error:
                    LOGGER.debug("Querying the installed version of package %s"
                                 " failed with error message: %s",
                                 package_name, error.output)
                else:
                    xbps_query_result = xbps_query_result.decode(
                        "utf-8").split("\n")
                    for line in xbps_query_result:
                        if "pkgver" in line:
                            installed_package = line.split(
                                ":")[1].strip()
                            match = re.match(package_regex,
                                             installed_package)
                            if match:
                                installed_package_version = match.groups()[1]
                                package_update = {
                                    "name": package_name,
                                    "new_version": new_package_version,
                                    "old_version": installed_package_version,
                                }
                            else:
                                LOGGER.error("Unable to detect installed "
                                             "version of package %s",
                                             package_name)
                                package_update = {
                                    "name": package_name,
                                    "new_version": new_package_version,
                                    "old_version": "unknown",
                                }
                            AVAILABLE_UPDATES.append(package_update)
                            break
if AVAILABLE_UPDATES:
    HOSTNAME = socket.getfqdn()
    MSG = MIMEMultipart()
    MAIL_TEXT = "The following package updates are available on host %s:\n\n"\
                % HOSTNAME
    MSG["Subject"] = "%s updates available on host %s"\
                     % (len(AVAILABLE_UPDATES), HOSTNAME)
    MSG["From"] = "voixicron@%s" % HOSTNAME
    MSG["To"] = ADMIN_EMAIL
    for update in AVAILABLE_UPDATES:
        MAIL_TEXT += "%s (%s -> %s)\n"\
                     % (update["name"],
                        update["old_version"],
                        update["new_version"])
    MAIL_TEXT += "\n\nYou can install the updates by issuing the command:" \
                 "\n\n\txbps-install -Su\n\nas root on %s\n\n--\nvoixicron"\
                 % HOSTNAME
    MSG_TEXT = MIMEText(MAIL_TEXT.encode("utf-8"), _charset="utf-8")
    MSG.attach(MSG_TEXT)
    SERVER_CONNECTION = smtplib.SMTP(MAIL_SERVER, MAIL_SERVER_PORT)
    try:
        SERVER_CONNECTION.connect()
    except (smtplib.SMTPConnectError, smtplib.SMTPServerDisconnected,
            socket.error):
        LOGGER.error("Failed to establish a connection to the SMTP server")
    else:
        try:
            SERVER_CONNECTION.starttls()
        except smtplib.SMTPException:
            pass
        if MAIL_SERVER_USER:
            try:
                SERVER_CONNECTION.login(MAIL_SERVER_USER, MAIL_SERVER_PASSWORD)
            except smtplib.SMTPAuthenticationError:
                LOGGER.error("Authentication on the mail server with user %s "
                             "failed.", MAIL_SERVER_USER)
            except smtplib.SMTPException:
                LOGGER.error("Error during authentication on the mail server.")
        try:
            SERVER_CONNECTION.sendmail("voixicron@%s" % HOSTNAME, ADMIN_EMAIL,
                                       MSG.as_string())
        except smtplib.SMTPRecipientsRefused:
            LOGGER.error("The mail server refused the recipient.")
        except smtplib.SMTPSenderRefused:
            LOGGER.error("The mail server refused the sender.")
        else:
            SERVER_CONNECTION.quit()
DEVNULL.close()