scripts/voixicron.py

#! /usr/bin/env python3
# -*- coding: utf8 -*-
"""
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
import json
import argparse

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

__version__ = "0.3.1"

# Configuration parameters used to send notifications and change the logging
# behaviour. The option configuration file may contain any of the keys used
# below in a json object.
CONFIG = {
    "admin_email": "root",
    "smtp_server": "localhost",
    "smtp_port": 25,
    "smtp_user": "",
    "smtp_password": "",
    "log_level": logging.INFO
}

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

PARSER = argparse.ArgumentParser(description="checks for available updates on "
                                             "Void Linux systems.")
PARSER.add_argument("-c", "--config", help="path to json configuration file")
ARGS = PARSER.parse_args()

if ARGS.config:
    try:
        with open(ARGS.config, "r") as conf_file:
            try:
                PARSED_JSON = json.load(conf_file)
            except ValueError:
                LOGGER.warn("The configuration is not a valid json document, "
                            "it will be ignored.")
            else:
                CONFIG.update(PARSED_JSON)
                LOGGER.setLevel(CONFIG["log_level"])
    except IOError as error:
        LOGGER.warn("Configuration file could not be opened, error: %s",
                    error.strerror)

AVAILABLE_UPDATES = []

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

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=subprocess.STDOUT)
except subprocess.CalledProcessError as error:
    LOGGER.error("Could not get list of available updates, "
                 "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)
    try:
        SERVER_CONNECTION = smtplib.SMTP(CONFIG["smtp_server"],
                                         CONFIG["smtp_port"])
    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()