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.action_utils import apt_hold, run_apt_command, service_restart
from plinth.modules.apache.components import check_url
from plinth.modules.upgrades import (BACKPORTS_SOURCES_LIST, SOURCES_LIST,
                                     get_current_release, is_backports_current)

AUTO_CONF_FILE = '/etc/apt/apt.conf.d/20auto-upgrades'
LOG_FILE = '/var/log/unattended-upgrades/unattended-upgrades.log'
DPKG_LOG_FILE = '/var/log/unattended-upgrades/unattended-upgrades-dpkg.log'
RELEASE_FILE_URL = \
    'https://deb.debian.org/debian/dists/{}/Release'

APT_PREFERENCES_FREEDOMBOX = '''Explanation: This file is managed by FreedomBox, do not edit.
Explanation: Allow carefully selected updates to 'freedombox' from backports.
Package: freedombox
Pin: release a={}-backports
Pin-Priority: 500
'''

# Whenever these preferences needs to change, increment the version number
# upgrades app. This ensures that setup is run again and the new contents are
# overwritten on the old file.
APT_PREFERENCES_APPS = '''Explanation: This file is managed by FreedomBox, do not edit.
Explanation: matrix-synapse 0.99.5 introduces room version 4. Older version
Explanation: 0.99.2 in buster won't be able join newly created rooms.
Package: matrix-synapse
Pin: release a=buster-backports
Pin-Priority: 500

Explanation: matrix-synapse >= 1.2 requires python3-service-identity >= 18.1
Package: python3-service-identity
Pin: release a=buster-backports
Pin-Priority: 500

Explanation: matrix-synapse >= 1.5 requires python3-typing-extensions >= 3.7.4
Package: python3-typing-extensions
Pin: release a=buster-backports
Pin-Priority: 500

Explanation: matrix-synapse >= 1.11 requires python3-signedjson >= 1.1.0
Package: python3-signedjson
Pin: release a=buster-backports
Pin-Priority: 500

Explanation: matrix-synapse >= 1.12 requires python3-twisted >= 18.9.0-8~
Package: python3-twisted
Pin: release a=buster-backports
Pin-Priority: 500

Explanation: python3-twisted requires matching version of python3-twisted-bin
Package: python3-twisted-bin
Pin: release a=buster-backports
Pin-Priority: 500

Explanation: matrix-synapse >= 1.16 requires python3-attr >= 19.1.0~
Package: python3-attr
Pin: release a=buster-backports
Pin-Priority: 500

Explanation: matrix-synapse >= 1.19 requires python3-canonicaljson >= 1.2.0
Package: python3-canonicaljson
Pin: release a=buster-backports
Pin-Priority: 500
'''

dist_upgrade_flag = pathlib.Path(
    '/var/lib/freedombox/dist-upgrade-in-progress')


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', help='Setup apt preferences')

    activate_backports = subparsers.add_parser(
        'activate-backports', help='Activate backports if possible')
    activate_backports.add_argument('--develop', required=False, default=False,
                                    action='store_true',
                                    help='Development mode')

    dist_upgrade = subparsers.add_parser(
        'dist-upgrade', help='Perform dist upgrade if possible')
    dist_upgrade.add_argument('--develop', required=False, default=False,
                              action='store_true', help='Development mode')
    dist_upgrade.add_argument('--test', required=False, default=False,
                              action='store_true',
                              help='Test dist-upgrade from stable to testing')

    subparsers.required = True
    return parser.parse_args()


def _run():
    """Run unattended-upgrades"""
    subprocess.run(['dpkg', '--configure', '-a'])
    run_apt_command(['--fix-broken', 'install'])
    subprocess.Popen(['systemctl', 'start', 'freedombox-manual-upgrade'],
                     stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                     stderr=subprocess.DEVNULL, close_fds=True,
                     start_new_session=True)


def subcommand_run(_):
    """Run unattended-upgrades"""
    try:
        _run()
    except FileNotFoundError:
        print('Error: systemctl 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 _check_auto():
    """Check if automatic upgrades are enabled."""
    arguments = [
        'apt-config', 'shell', 'UpdateInterval',
        'APT::Periodic::Update-Package-Lists'
    ]
    output = subprocess.check_output(arguments).decode()
    update_interval = 0
    match = re.match(r"UpdateInterval='(.*)'", output)
    if match:
        update_interval = int(match.group(1))

    return bool(update_interval)


def subcommand_check_auto(_):
    """Check if automatic upgrades are enabled"""
    try:
        print(_check_auto())
    except subprocess.CalledProcessError as error:
        print('Error: {0}'.format(error), file=sys.stderr)
        sys.exit(1)


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:
        print('==> ' + os.path.basename(LOG_FILE))
        with open(LOG_FILE, 'r') as file_handle:
            print(file_handle.read())
    except IOError:
        pass

    try:
        print('==> ' + os.path.basename(DPKG_LOG_FILE))
        with open(DPKG_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, dist, backports=False):
    """Return whether the release for dist[-backports] is available."""
    wrapper = None
    if protocol == 'tor+http':
        wrapper = 'torsocks'

    if backports:
        dist += '-backports'

    result = check_url(RELEASE_FILE_URL.format(dist), wrapper=wrapper)
    return result == 'passed'


def _add_backports_sources(sources_list, protocol, dist):
    """Add 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 {dist}-backports main
deb-src {protocol}://deb.debian.org/debian {dist}-backports main
'''
    sources = sources.format(protocol=protocol, dist=dist)
    with open(sources_list, 'w') as file_handle:
        file_handle.write(sources)


def _check_and_backports_sources(develop=False):
    """Add 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)

    if is_backports_current():
        print('Repositories list up-to-date. Skipping update.')
        return

    try:
        with open('/etc/dpkg/origins/default', 'r') as default_origin:
            matches = [
                re.match(r'Vendor:\s+Debian', line, flags=re.IGNORECASE)
                for line in default_origin.readlines()
            ]
    except FileNotFoundError:
        print('Could not open /etc/dpkg/origins/default')
        return

    if not any(matches):
        print('System is running a derivative of Debian. Skip enabling '
              'backports.')
        return

    release, dist = get_current_release()
    if release == 'unstable' or (release == 'testing' and not develop):
        print(f'System release is {release}. Skip enabling backports.')
        return

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

    if not _is_release_file_available(protocol, dist, backports=True):
        print(f'Release file for {dist}-backports is not available yet.')
        return

    print(f'{dist}-backports is now available. Adding to sources.')
    _add_backports_sources(BACKPORTS_SOURCES_LIST, protocol, dist)
    # In case of dist upgrade, rewrite the preferences file.
    _add_apt_preferences()


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

    # Don't try to remove 50freedombox3.pref as this file is shipped with the
    # Debian package and is removed using maintainer scripts.

    _, dist = get_current_release()
    if dist == 'sid':
        print(f'System distribution is {dist}. Skip setting apt preferences '
              'for backports.')
    else:
        print(f'Setting apt preferences for {dist}-backports.')
        with open(base_path / '50freedombox4.pref', 'w') as file_handle:
            file_handle.write(APT_PREFERENCES_FREEDOMBOX.format(dist))
        with open(base_path / '51freedombox-apps.pref', 'w') as file_handle:
            file_handle.write(APT_PREFERENCES_APPS)


def _check_and_dist_upgrade(develop=False, test_upgrade=False):
    """Check for new stable release. If there is one, and updates are
    enabled, perform dist-upgrade.

    If develop is True, check for possible upgrade from stable to testing.
    If test_upgrade is True, also perform the upgrade to testing.
    """
    if dist_upgrade_flag.exists():
        print('Continuing previously interrupted dist-upgrade.')
        _perform_dist_upgrade()
        return

    release, dist = get_current_release()
    if release in ['unstable', 'testing']:
        print(f'System release is {release}. Skip checking for new stable '
              'release.')
        return

    check_dists = ['stable']
    if develop:
        check_dists.append('testing')

    codename = None
    for check_dist in check_dists:
        url = RELEASE_FILE_URL.format(check_dist)
        command = ['curl', '--silent', '--location', '--fail', url]
        protocol = _get_protocol()
        if protocol == 'tor+http':
            command.insert(0, 'torsocks')
            print('Package download over Tor is enabled.')

        try:
            output = subprocess.check_output(command).decode()
        except (subprocess.CalledProcessError, FileNotFoundError):
            print(f'Error while checking for new {check_dist} release')
        else:
            for line in output.split('\n'):
                if line.startswith('Codename:'):
                    codename = line.split()[1]

    if not codename:
        print('"Codename:" not found in release file.')
        return

    if codename == dist:
        print(f'{dist} is already the latest release.')
        return

    if not _check_auto():
        print('Automatic updates are not enabled.')
        return

    if check_dist == 'testing' and not test_upgrade:
        print(f'Skipping dist-upgrade to {check_dist} since --test is not '
              'set.')
        return

    output = subprocess.check_output(['df', '--output=avail,pcent', '/'])
    output = output.decode().split('\n')[1].split()
    free_space, free_percent = int(output[0]), int(output[1][:-1])
    if free_space < 1000000 or free_percent < 10:
        print('Not enough free space in /.')
        return

    print(f'Upgrading from {dist} to {codename}...')
    with open(SOURCES_LIST, 'r') as sources_list:
        lines = sources_list.readlines()

    with open(SOURCES_LIST, 'w') as sources_list:
        for line in lines:
            # E.g. replace 'buster' with 'bullseye'.
            new_line = line.replace(dist, codename)
            if check_dist == 'testing':
                # E.g. replace 'stable' with 'bullseye'.
                new_line = new_line.replace('stable', codename)

            # Security suite name renamed starting with bullseye
            if 'security' in new_line:
                new_line = new_line.replace('/updates', '-security')

            sources_list.write(new_line)

    print('Dist upgrade in progress. Setting flag.')
    dist_upgrade_flag.touch(mode=0o660)
    _perform_dist_upgrade()


def _perform_dist_upgrade():
    """Perform upgrade to next release of Debian."""
    # Hold freedombox package during entire dist upgrade.
    print('Holding freedombox package...')
    with apt_hold():
        print('Updating Apt cache...')
        run_apt_command(['update'])

        print('Upgrading base-files and unattended-upgrades...')
        run_apt_command(['install', 'base-files'])
        run_apt_command(['install', 'unattended-upgrades'])

        print('Running unattended-upgrade...')
        subprocess.run(['unattended-upgrade', '--verbose'])

        # Remove obsolete packages that may prevent other packages from
        # upgrading.
        print('Removing libgcc1...')
        run_apt_command(['remove', 'libgcc1'])

        # Hold packages known to have conffile prompts. FreedomBox service
        # will handle their upgrade later.
        print('Holding firewalld and radicale packages...')
        with apt_hold(['firewalld', 'radicale']):
            print('Running apt full-upgrade...')
            run_apt_command(['full-upgrade'])

        print('Running apt autoremove...')
        run_apt_command(['autoremove'])

    print('Dist upgrade complete. Removing flag.')
    if dist_upgrade_flag.exists():
        dist_upgrade_flag.unlink()

    # FreedomBox Service may have tried to restart several times
    # during the upgrade, but failed. It will stop trying after 5
    # retries. Restart it once more to ensure it is running.
    print('Restarting FreedomBox service...')
    service_restart('plinth')


def subcommand_setup(_):
    """Setup apt preferences."""
    _add_apt_preferences()


def subcommand_activate_backports(arguments):
    """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(arguments.develop)


def subcommand_dist_upgrade(arguments):
    """Perform major distribution upgrade.

    Check if a new stable release is available, and perform dist-upgrade if
    updates are enabled.

    """
    _check_and_dist_upgrade(arguments.develop, arguments.test)


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()