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

__version__ = "0.3"

# Configuration parameters used to send notifications and change the logging
# behaviour
CONFIG = {
    "admin_email": "root",
    "smtp_server": "localhost",
    "smtp_port": 25,
    "smtp_user": "",
    "smtp_password": "",
    "log_level": logging.ERROR,

}

AVAILABLE_UPDATES = []

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

LOGGER = logging.getLogger("voixicron")
LOGGER.setLevel(CONFIG["log_level"])
CONSOLE_HANDLER = logging.StreamHandler()
LOGGER.addHandler(CONSOLE_HANDLER)

try:
    subprocess.call(["xbps-install", "-S"], stdout=DEVNULL, stderr=DEVNULL)
except subprocess.CalledProcessError:
    LOGGER.warn("Repository information could not be updated, voixicron will "
                "report outdated data!")

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"] = CONFIG["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(CONFIG["smtp_server"],
                                     CONFIG["smtp_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 CONFIG["smtp_user"]:
            try:
                SERVER_CONNECTION.login(CONFIG["smtp_user"],
                                        CONFIG["smtp_password"])
            except smtplib.SMTPAuthenticationError:
                LOGGER.error("Authentication on the mail server with user %s "
                             "failed.", CONFIG["smtp_user"])
            except smtplib.SMTPException:
                LOGGER.error("Error during authentication on the mail server.")
        try:
            SERVER_CONNECTION.sendmail("voixicron@%s" % HOSTNAME,
                                       CONFIG["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()