FreedomBox/plinth/modules/upgrades/__init__.py

# SPDX-License-Identifier: AGPL-3.0-or-later
"""FreedomBox app for upgrades."""

import logging
import os
import subprocess

from aptsources import sourceslist
from django.utils.translation import gettext_lazy as _
from django.utils.translation import gettext_noop

import plinth
from plinth import action_utils
from plinth import app as app_module
from plinth import cfg, glib, kvstore, menu, package
from plinth.config import DropinConfigs
from plinth.daemon import RelatedDaemon
from plinth.diagnostic_check import DiagnosticCheck, Result
from plinth.modules.backups.components import BackupRestore
from plinth.package import Packages

from . import distupgrade, manifest, privileged

first_boot_steps = [
    {
        'id': 'backports_wizard',
        'url': 'upgrades:backports-firstboot',
        'order': 5,
    },
]

_description = [
    _('Check for and apply the latest software and security updates.'),
    _('Updates are run at 06:00 everyday according to local time zone. Set '
      'your time zone in Date & Time app. Apps are restarted after update '
      'causing them to be unavailable briefly. If system reboot is deemed '
      'necessary, it is done automatically at 02:00 causing all apps to be '
      'unavailable briefly.')
]

BACKPORTS_REQUESTED_KEY = 'upgrades_backports_requested'

DIST_UPGRADE_ENABLED_KEY = 'upgrades_dist_upgrade_enabled'

PKG_HOLD_DIAG_CHECK_ID = 'upgrades-package-holds'

logger = logging.getLogger(__name__)


class UpgradesApp(app_module.App):
    """FreedomBox app for software upgrades."""

    app_id = 'upgrades'

    _version = 18

    can_be_disabled = False

    def __init__(self) -> None:
        """Create components for the app."""
        super().__init__()

        info = app_module.Info(app_id=self.app_id, version=self._version,
                               is_essential=True, name=_('Software Update'),
                               icon='fa-refresh', description=_description,
                               manual_page='Upgrades', tags=manifest.tags)
        self.add(info)

        menu_item = menu.Menu('menu-upgrades', info.name, info.icon, info.tags,
                              'upgrades:index',
                              parent_url_name='system:system', order=50)
        self.add(menu_item)

        packages = Packages('packages-upgrades',
                            ['unattended-upgrades', 'needrestart'])
        self.add(packages)

        dropin_configs = DropinConfigs('dropin-configs-upgrades', [
            '/etc/apt/apt.conf.d/20freedombox',
            '/etc/apt/apt.conf.d/20freedombox-allow-release-info-change',
            '/etc/apt/apt.conf.d/60unattended-upgrades',
            '/etc/needrestart/conf.d/freedombox.conf',
        ])
        self.add(dropin_configs)

        daemon = RelatedDaemon('related-daemon-upgrades',
                               'freedombox-dist-upgrade')
        self.add(daemon)

        backup_restore = BackupRestore('backup-restore-upgrades',
                                       **manifest.backup)
        self.add(backup_restore)

    def post_init(self):
        """Perform post initialization operations."""
        self._show_new_release_notification()

        # Check every day if backports becomes available, then configure it if
        # selected by user.
        glib.schedule(24 * 3600, setup_repositories)

        # Check every day if new stable release becomes available, then perform
        # dist-upgrade if updates are enabled
        glib.schedule(24 * 3600, check_dist_upgrade)

    def _show_new_release_notification(self):
        """When upgraded to new release, show a notification."""
        from plinth.notification import Notification
        try:
            note = Notification.get('upgrades-new-release')
            if note.data['version'] == plinth.__version__:
                # User already has notification for update to this version. It
                # may be dismissed or not yet dismissed
                return

            # User currently has a notification for an older version, update.
            dismiss = False
        except KeyError:
            # Don't show notification for the first version user runs, create
            # but don't show it.
            dismiss = True

        data = {
            'version': plinth.__version__,
            'app_name': 'translate:' + gettext_noop('Software Update'),
            'app_icon': 'fa-refresh'
        }
        title = gettext_noop('FreedomBox Updated')
        note = Notification.update_or_create(
            id='upgrades-new-release', app_id='upgrades', severity='info',
            title=title, body_template='upgrades-new-release.html', data=data,
            group='admin')
        note.dismiss(should_dismiss=dismiss)

    def _show_first_manual_update_notification(self):
        """After first setup, show notification to manually run updates."""
        from plinth.notification import Notification
        title = gettext_noop('Run software update manually')
        message = gettext_noop(
            'Automatic software update runs daily by default. For the first '
            'time, manually run it now.')
        data = {
            'app_name': 'translate:' + gettext_noop('Software Update'),
            'app_icon': 'fa-refresh'
        }
        actions = [{
            'type': 'link',
            'class': 'primary',
            'text': gettext_noop('Go to {app_name}'),
            'url': 'upgrades:index'
        }, {
            'type': 'dismiss'
        }]
        Notification.update_or_create(id='upgrades-first-manual-update',
                                      app_id='upgrades', severity='info',
                                      title=title, message=message,
                                      actions=actions, data=data,
                                      group='admin', dismissed=False)

    def setup(self, old_version):
        """Install and configure the app."""
        super().setup(old_version)

        # Enable automatic upgrades but only on first install
        if not old_version and not cfg.develop:
            privileged.enable_auto()

        # Request user to run manual update as a one time activity
        if not old_version:
            self._show_first_manual_update_notification()

        # Update apt preferences whenever on first install and on version
        # increment.
        privileged.setup()

        # When upgrading from a version without first boot wizard for
        # backports, assume backports have been requested.
        if old_version and old_version < 7:
            set_backports_requested(can_activate_backports())

        # Enable dist upgrade for new installs, and once when upgrading
        # from version without flag.
        if not old_version or old_version < 8:
            set_dist_upgrade_enabled(can_enable_dist_upgrade())

        # Try to setup apt repositories, if needed, if possible, on first
        # install and on version increment.
        setup_repositories(None)

    def diagnose(self) -> list[DiagnosticCheck]:
        """Run diagnostics and return the results."""
        results = super().diagnose()
        results.append(_diagnose_held_packages())
        return results

    def repair(self, failed_checks: list) -> bool:
        """Handle repair for custom diagnostic."""
        remaining_checks = []
        for check in failed_checks:
            if check.check_id == PKG_HOLD_DIAG_CHECK_ID:
                privileged.release_held_packages()
            else:
                remaining_checks.append(check)

        return super().repair(remaining_checks)


def setup_repositories(_):
    """Setup apt repositories for backports."""
    if is_backports_requested():
        privileged.activate_backports(cfg.develop)


def check_dist_upgrade(_):
    """Check for upgrade to new stable release."""
    if is_dist_upgrade_enabled():
        status = distupgrade.get_status()
        if status['next_action'] in ('continue', 'ready'):
            privileged.start_dist_upgrade()
        else:
            logger.info('Not ready for distribution upgrade - %s', status)


def is_backports_requested():
    """Return whether user has chosen to activate backports."""
    return kvstore.get_default(BACKPORTS_REQUESTED_KEY, False)


def set_backports_requested(requested):
    """Set whether user has chosen to activate backports."""
    kvstore.set(BACKPORTS_REQUESTED_KEY, requested)
    logger.info('Backports requested - %s', requested)


def is_dist_upgrade_enabled():
    """Return whether user has enabled dist upgrade."""
    return kvstore.get_default(DIST_UPGRADE_ENABLED_KEY, False)


def set_dist_upgrade_enabled(enabled=True):
    """Set whether user has enabled dist upgrade."""
    kvstore.set(DIST_UPGRADE_ENABLED_KEY, enabled)
    logger.info('Distribution upgrade configured - %s', enabled)


def is_backports_enabled():
    """Return whether backports are enabled in the system configuration."""
    return os.path.exists(privileged.BACKPORTS_SOURCES_LIST)


def get_current_release():
    """Return current release and codename as a tuple."""
    output = subprocess.check_output(
        ['lsb_release', '--release', '--codename',
         '--short']).decode().strip()
    lines = output.split('\n')
    return lines[0], lines[1]


def is_backports_current():
    """Return whether backports are enabled for the current release."""
    if not is_backports_enabled():
        return False

    _, dist = get_current_release()
    dist += '-backports'
    sources = sourceslist.SourcesList()
    for source in sources:
        if source.dist == dist:
            return True

    return False


def can_activate_backports():
    """Return whether backports can be activated."""
    if cfg.develop:
        return True

    # Release will be 'n/a' in latest unstable and testing distributions.
    release, _ = get_current_release()
    return release not in ['unstable', 'testing', 'n/a']


def can_enable_dist_upgrade():
    """Return whether dist upgrade can be enabled."""
    release, _ = get_current_release()
    return release not in ['unstable', 'testing', 'n/a']


def _diagnose_held_packages():
    """Check if any packages have holds."""
    check = DiagnosticCheck(PKG_HOLD_DIAG_CHECK_ID,
                            gettext_noop('Check for package holds'),
                            Result.NOT_DONE)
    if (package.is_package_manager_busy()
            or action_utils.service_is_running('freedombox-dist-upgrade')):
        check.result = Result.SKIPPED
        return check

    output = subprocess.check_output(['apt-mark', 'showhold']).decode().strip()
    held_packages = output.split()
    check.result = Result.FAILED if held_packages else Result.PASSED
    return check