FreedomBox/actions/upgrades

#!/usr/bin/python3
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Configures or runs unattended-upgrades
"""

import argparse
import os
import pathlib
import re
import subprocess
import sys

from plinth.modules.apache.components import check_url

AUTO_CONF_FILE = '/etc/apt/apt.conf.d/20auto-upgrades'
LOG_FILE = '/var/log/unattended-upgrades/unattended-upgrades.log'
BUSTER_BACKPORTS_RELEASE_FILE_URL = \
    'https://deb.debian.org/debian/dists/buster-backports/Release'


def parse_arguments():
    """Return parsed command line arguments as dictionary"""
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest='subcommand', help='Sub command')

    subparsers.add_parser('run', help='Upgrade packages on the system')
    subparsers.add_parser('check-auto',
                          help='Check if automatic upgrades are enabled')
    subparsers.add_parser('enable-auto', help='Enable automatic upgrades')
    subparsers.add_parser('disable-auto', help='Disable automatic upgrades.')
    subparsers.add_parser('get-log', help='Print the automatic upgrades log')

    subparsers.add_parser('setup-repositories',
                          help='Setup software repositories for FreedomBox')

    subparsers.required = True
    return parser.parse_args()


def subcommand_run(_):
    """Run unattended-upgrades"""
    try:
        subprocess.Popen(['unattended-upgrades', '-v'],
                         stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                         stderr=subprocess.DEVNULL, close_fds=True,
                         start_new_session=True)
    except FileNotFoundError:
        print('Error: unattended-upgrades is not available.', file=sys.stderr)
        sys.exit(2)
    except Exception as error:
        print('Error: {0}'.format(error), file=sys.stderr)
        sys.exit(3)


def subcommand_check_auto(_):
    """Check if automatic upgrades are enabled"""
    arguments = [
        'apt-config', 'shell', 'UpdateInterval',
        'APT::Periodic::Update-Package-Lists'
    ]
    try:
        output = subprocess.check_output(arguments).decode()
    except subprocess.CalledProcessError as error:
        print('Error: {0}'.format(error), file=sys.stderr)
        sys.exit(1)

    update_interval = 0
    match = re.match(r"UpdateInterval='(.*)'", output)
    if match:
        update_interval = int(match.group(1))

    print(bool(update_interval))


def subcommand_enable_auto(_):
    """Enable automatic upgrades"""
    with open(AUTO_CONF_FILE, 'w') as conffile:
        conffile.write('APT::Periodic::Update-Package-Lists "1";\n')
        conffile.write('APT::Periodic::Unattended-Upgrade "1";\n')


def subcommand_disable_auto(_):
    """Disable automatic upgrades"""
    with open(AUTO_CONF_FILE, 'w') as conffile:
        conffile.write('APT::Periodic::Update-Package-Lists "0";\n')
        conffile.write('APT::Periodic::Unattended-Upgrade "0";\n')


def subcommand_get_log(_):
    """Print the automatic upgrades log."""
    try:
        with open(LOG_FILE, 'r') as file_handle:
            print(file_handle.read())
    except IOError:
        pass


def _get_protocol():
    """Return the protocol to use for newly added repository sources."""
    try:
        from plinth.modules.tor import utils
        if utils.is_apt_transport_tor_enabled():
            return 'tor+http'
    except Exception:
        pass

    return 'http'


def _is_release_file_available(protocol):
    """Return whether the release for backports is available."""
    wrapper = None
    if protocol == 'tor+http':
        wrapper = 'torsocks'

    result = check_url(BUSTER_BACKPORTS_RELEASE_FILE_URL, wrapper=wrapper)
    return result == 'passed'


def _add_buster_backports_sources(sources_list, protocol):
    """Add buster backports sources to freedombox repositories list."""
    sources = '''# This file is managed by FreedomBox, do not edit.
# Allow carefully selected updates to 'freedombox' from backports.

deb {protocol}://deb.debian.org/debian buster-backports main
deb-src {protocol}://deb.debian.org/debian buster-backports main
'''
    sources = sources.format(protocol=protocol)
    with open(sources_list, 'w') as file_handle:
        file_handle.write(sources)


def _check_and_backports_sources():
    """Add buster backports sources after checking if it is available."""
    old_sources_list = '/etc/apt/sources.list.d/freedombox.list'
    if os.path.exists(old_sources_list):
        os.remove(old_sources_list)

    sources_list = '/etc/apt/sources.list.d/freedombox2.list'
    if os.path.exists(sources_list):
        print('Repositories list up-to-date. Skipping update.')
        return

    protocol = _get_protocol()
    if protocol == 'tor+http':
        print('Package download over Tor is enabled.')

    if not _is_release_file_available(protocol):
        print('Release file for Buster backports is not available yet.')
        return

    print('Buster backports is now available. Adding to sources.')
    _add_buster_backports_sources(sources_list, protocol)


def _add_apt_preferences():
    """Setup APT preferences to upgrade selected packages from backports."""
    for file_name in ['50freedombox.pref', '50freedombox2.pref']:
        full_path = pathlib.Path('/etc/apt/preferences.d') / file_name
        if full_path.exists():
            full_path.unlink()


def subcommand_setup_repositories(_):
    """Setup software repositories needed for FreedomBox.

    Repositories list for now only contains the backports. If the file exists,
    assume that it contains backports.

    """
    _check_and_backports_sources()
    _add_apt_preferences()


def main():
    """Parse arguments and perform all duties"""
    arguments = parse_arguments()

    subcommand = arguments.subcommand.replace('-', '_')
    subcommand_method = globals()['subcommand_' + subcommand]
    subcommand_method(arguments)


if __name__ == '__main__':
    main()