FreedomBox/actions/packages

#!/usr/bin/python3
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Wrapper to handle package installation with apt-get.
"""

import argparse
import json
import logging
import os
import subprocess
import sys
from collections import defaultdict
from importlib import import_module

import apt.cache
import apt_inst
import apt_pkg

from plinth import cfg
from plinth.action_utils import (apt_hold_freedombox, is_package_manager_busy,
                                 run_apt_command)

logger = logging.getLogger(__name__)


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('update', help='update the package lists')

    subparser = subparsers.add_parser('install', help='install packages')
    subparser.add_argument(
        '--skip-recommends', action='store_true',
        help='whether to skip installing recommended packages')
    subparser.add_argument(
        '--force-configuration', choices=['new', 'old'],
        help='force old/new configuration files during install')
    subparser.add_argument(
        '--reinstall', action='store_true',
        help='force re-installation of package even if it is current')
    subparser.add_argument(
        '--force-missing-configuration', action='store_true',
        help='force installation of missing configuration files')
    subparser.add_argument(
        'module', help='name of module for which package is being installed')
    subparser.add_argument('packages', nargs='+',
                           help='list of packages to install')
    subparsers.add_parser('is-package-manager-busy',
                          help='Return whether package manager is busy')
    subparser = subparsers.add_parser(
        'filter-conffile-packages',
        help='Filter out packages that do not have pending conffile prompts')
    subparser.add_argument('--packages', required=True,
                           help='List of packages to filter', nargs='+')

    subparsers.required = True
    return parser.parse_args()


def subcommand_update(arguments):
    """Update apt package lists."""
    sys.exit(run_apt_command(['update']))


def subcommand_install(arguments):
    """Install packages using apt-get."""
    try:
        _assert_managed_packages(arguments.module, arguments.packages)
    except Exception as exception:
        print('Access check failed:', exception, file=sys.stderr)
        sys.exit(99)

    extra_arguments = []
    if arguments.skip_recommends:
        extra_arguments.append('--no-install-recommends')

    if arguments.force_configuration == 'old':
        extra_arguments += [
            '-o', 'Dpkg::Options::=--force-confdef', '-o',
            'Dpkg::Options::=--force-confold'
        ]
    elif arguments.force_configuration == 'new':
        extra_arguments += ['-o', 'Dpkg::Options::=--force-confnew']

    if arguments.reinstall:
        extra_arguments.append('--reinstall')

    if arguments.force_missing_configuration:
        extra_arguments += ['-o', 'Dpkg::Options::=--force-confmiss']

    subprocess.run(['dpkg', '--configure', '-a'])
    with apt_hold_freedombox():
        run_apt_command(['--fix-broken', 'install'])
        returncode = run_apt_command(['install'] + extra_arguments +
                                     arguments.packages)

    sys.exit(returncode)


def _assert_managed_packages(module, packages):
    """Check that list of packages are in fact managed by module."""
    cfg.read()
    module_file = os.path.join(cfg.config_dir, 'modules-enabled', module)

    with open(module_file, 'r') as file_handle:
        module_path = file_handle.read().strip()

    module = import_module(module_path)
    for package in packages:
        assert package in module.managed_packages


def subcommand_is_package_manager_busy(_):
    """Check whether package manager is busy.

    An exit code of zero indicates that package manager is busy."""
    if not is_package_manager_busy():
        sys.exit(-1)


def subcommand_filter_conffile_packages(arguments):
    """Return filtered list of packages which have pending conffile prompts.

    When considering which file needs a configuration file prompt, mimic the
    behavior of unattended-upgrades package. This is because when
    unattended-upgrades gives up on the package due to conffile prompts, that
    is when this logic needs to step in and perform the upgrades.

    The logic is (roughly):

      - Read /var/lib/dpkg/status file to read hashes as provided by currently
        installed version of a package.

      - Read each configuration file for the package from disk and compute
        hashes.

      - If the hashes match, package has no configuration file that got
        modified. There will be no conffile prompt.

      - If hashes don't match then check if the version of the package being
        upgraded to has the same hash as the old version of the package or in
        the new version or the package that configuration file has been
        removed. If the conditions satisfy, then there will be no conffile
        prompt. Otherwise, package will have conffile prompt.

      - A further condition for showing conffile prompt is when new package
        brings in additional configuration files not known before and some of
        which are already present on the disk and mismatch with incoming files.

    """
    apt_pkg.init()  # Read configuration that will be used later.
    packages = set(arguments.packages)

    status_hashes, current_versions = _get_conffile_hashes_from_status_file(
        packages)

    mismatched_hashes = _filter_matching_package_hashes(status_hashes)

    downloaded_files = _download_packages(packages)

    new_package_hashes, new_versions = \
        _get_conffile_hashes_from_downloaded_files(
            packages, downloaded_files, status_hashes, mismatched_hashes)

    packages_info = {}
    for package in packages:
        modified_conffiles = _get_modified_conffiles(
            status_hashes[package], mismatched_hashes[package],
            new_package_hashes[package])
        if not modified_conffiles:
            continue

        package_info = {
            'current_version': current_versions[package],
            'new_version': new_versions[package],
            'modified_conffiles': modified_conffiles
        }
        packages_info[package] = package_info

    print(json.dumps(packages_info))


def _get_modified_conffiles(status_hashes, mismatched_hashes,
                            new_package_hashes):
    """Return list of conffiles that will cause prompts for a package."""
    modified_conffiles = []
    for conffile, hash_value in mismatched_hashes.items():
        if conffile not in new_package_hashes:
            # Modified configuration file not present new package
            continue

        if status_hashes[conffile] == new_package_hashes[conffile]:
            # Configuration file is same in both versions of package. Conffile
            # prompt is not triggered even if the file is modified on disk.
            continue

        modified_conffiles.append(conffile)

    for conffile, hash_value in new_package_hashes.items():
        if conffile in status_hashes:
            # File is not new, it was read and matched against new/old hashes
            continue

        if not os.path.exists(conffile):
            # New configuration file brought by new package doesn't not already
            # exist on disk.
            continue

        if _get_conffile_hash(conffile) != hash_value:
            # New configuration file brought by new package doesn't match file
            # on the disk.
            #
            # If existing file is a directory, unattended-upgrades allows it,
            # we still treat it as a conffile prompt. This should be okay.
            modified_conffiles.append(conffile)

    return modified_conffiles


def _get_conffile_hashes_from_status_file(packages):
    """For each of the packages, return a dict of conffile hashes.

    Work on all packages at the same time to avoid parsing the status file
    multiple times.

    """
    package_hashes = defaultdict(dict)
    package_versions = defaultdict(lambda: None)

    status_file = apt_pkg.config.find('Dir::State::status')
    with apt_pkg.TagFile(status_file) as tag_file:
        for section in tag_file:
            package = section.get('Package')
            if package not in packages:
                continue

            hashes = _parse_conffiles_value(section.get('Conffiles', ''))
            package_hashes[package] = hashes
            package_versions[package] = section.get('Version')

    return package_hashes, package_versions


def _parse_conffiles_value(value):
    """Parse and return the list of conffiles as found in dpkg status file."""
    conffiles = {}
    for line in value.splitlines():
        parts = line.strip().split()
        if len(parts) > 2 and parts[2] == 'obsolete':
            continue

        md5sum = parts[1]
        if md5sum == 'newconffile':  # (LP: #936870)
            continue

        file_path = parts[0]
        conffiles[file_path] = md5sum

    return conffiles


def _filter_matching_package_hashes(package_hashes):
    """Return hashes of only conffiles that don't match for each package."""
    mismatched_hashes = defaultdict(dict)
    for package, hashes in package_hashes.items():
        system_hashes = {}
        for conffile, md5sum in hashes.items():
            system_md5sum = _get_conffile_hash(conffile)
            if md5sum != system_md5sum:
                system_hashes[conffile] = system_md5sum

        if system_hashes:
            mismatched_hashes[package] = system_hashes

    return mismatched_hashes


def _get_conffile_hash(conffile):
    """Return hash of a conffile in the system."""
    try:
        with open(conffile, 'rb') as file_handle:
            return apt_pkg.md5sum(file_handle)
    except (FileNotFoundError, OSError):
        return None


def _download_packages(packages):
    """Download the package for upgrade."""
    sources_list = apt_pkg.SourceList()
    sources_list.read_main_list()

    apt_pkg_cache = apt_pkg.Cache(None)  # None prevents progress messages
    apt_cache = apt.cache.Cache()
    dep_cache = apt_pkg.DepCache(apt_pkg_cache)
    for package_name in packages:
        package = apt_cache[package_name]
        if package.is_upgradable:
            dep_cache.mark_install(apt_pkg_cache[package_name], True,
                                   not package.is_auto_installed)

    package_manager = apt_pkg.PackageManager(dep_cache)
    records = apt_pkg.PackageRecords(apt_pkg_cache)
    fetcher = apt_pkg.Acquire()
    package_manager.get_archives(fetcher, sources_list, records)
    run_result = fetcher.run()
    if run_result != apt_pkg.Acquire.RESULT_CONTINUE:
        logger.error('Downloading packages failed.')
        sys.exit(1)

    downloaded_files = []
    for item in fetcher.items:
        if not item.complete or item.status == item.STAT_ERROR or \
           item.status == item.STAT_AUTH_ERROR:
            continue

        if not item.is_trusted:
            continue

        if not os.path.exists(item.destfile):
            continue

        if not item.destfile.endswith('.deb'):
            continue

        downloaded_files.append(item.destfile)

    return downloaded_files


def _get_conffile_hashes_from_downloaded_files(packages, downloaded_files,
                                               status_hashes,
                                               mismatched_hashes):
    """Retrieve the conffile hashes from downloaded .deb files."""
    new_hashes = defaultdict(dict)
    new_versions = defaultdict(lambda: None)
    for downloaded_file in downloaded_files:
        try:
            package_name, hashes, new_version = \
                _get_conffile_hashes_from_downloaded_file(
                    packages, downloaded_file, status_hashes,
                    mismatched_hashes)
        except (LookupError, apt_pkg.Error, ValueError):
            continue

        new_hashes[package_name] = hashes
        new_versions[package_name] = new_version

    return new_hashes, new_versions


def _get_conffile_hashes_from_downloaded_file(packages, downloaded_file,
                                              status_hashes,
                                              mismatched_hashes):
    """Retrieve the conffile hashes from a single downloaded .deb file."""
    deb_file = apt_inst.DebFile(downloaded_file)

    control = deb_file.control.extractdata('control')

    section = apt_pkg.TagSection(control)
    package_name = section['Package']
    if package_name not in packages:
        raise ValueError

    new_version = section['Version']

    conffiles = deb_file.control.extractdata('conffiles')
    conffiles = conffiles.decode().strip().split()

    status_hashes = status_hashes.get(package_name, {})
    mismatched_hashes = mismatched_hashes.get(package_name, {})

    hashes = {}
    for conffile in conffiles:
        if conffile in status_hashes and conffile not in mismatched_hashes:
            # File already in old package and there is no change on disk.
            # Optimization to make sure we read as fewer files as possible.
            continue

        conffile_data = deb_file.data.extractdata(conffile.lstrip('/'))
        md5sum = apt_pkg.md5sum(conffile_data)
        hashes[conffile] = md5sum

    return package_name, hashes, new_version


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