~ubuntu-core-dev/update-manager/main

# MetaRelease.py

# -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*-

#

#  Copyright (c) 2004,2005 Canonical

#

#  Author: Michael Vogt <michael.vogt@ubuntu.com>

#

#  This program is free software; you can redistribute it and/or

#  modify it under the terms of the GNU General Public License as

#  published by the Free Software Foundation; either version 2 of the

#  License, or (at your option) any later version.

#

#  This program is distributed in the hope that it will be useful,

#  but WITHOUT ANY WARRANTY; without even the implied warranty of

#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

#  GNU General Public License for more details.

#

#  You should have received a copy of the GNU General Public License

#  along with this program; if not, write to the Free Software

#  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307

#  USA

import apt

import apt_pkg

import distro_info

import configparser

from http.client import BadStatusLine

import logging

import email.utils

import os

import socket

import sys

import time

import threading

from urllib.parse import quote

from urllib.request import Request, urlopen

from urllib.error import HTTPError, URLError

from .utils import (

    get_lang,

    get_dist,

    get_dist_version,

    get_ubuntu_flavor,

    get_ubuntu_flavor_name,

)

class MetaReleaseParseError(Exception):

    pass

class Dist(object):

    def __init__(self, name, version, date, supported):

        self.name = name

        self.version = version

        self.date = date

        self.supported = supported

        self.releaseNotesURI = None

        self.releaseNotesHtmlUri = None

        self.upgradeTool = None

        self.upgradeToolSig = None

        # the server may report that the upgrade is broken currently

        self.upgrade_broken = None

class MetaReleaseCore(object):

    """

    A MetaReleaseCore object abstracts the list of released

    distributions.

    """

    DEBUG = "DEBUG_UPDATE_MANAGER" in os.environ

    # some constants

    CONF = "/etc/update-manager/release-upgrades"

    CONF_METARELEASE = "/etc/update-manager/meta-release"

    def __init__(

        self,

        useDevelopmentRelease=False,

        useProposed=False,

        debug=False,

        forceLTS=False,

        forceDownload=False,

        cache=None,

    ):

        if debug:

            self.DEBUG = True

        self._debug(

            "MetaRelease.__init__() useDevel=%s useProposed=%s"

            % (useDevelopmentRelease, useProposed)

        )

        # force download instead of sending if-modified-since

        self.forceDownload = forceDownload

        self.useDevelopmentRelease = useDevelopmentRelease

        # information about the available dists

        self.downloaded = threading.Event()

        self.upgradable_to = None

        self.new_dist = None

        if cache is None:

            cache = apt.Cache()

        self.flavor = get_ubuntu_flavor(cache=cache)

        self.flavor_name = get_ubuntu_flavor_name(cache=cache)

        self.current_dist_name = get_dist()

        self.current_dist_version = get_dist_version()

        self.no_longer_supported = None

        self.prompt = None

        # default (if the conf file is missing)

        base_uri = "https://changelogs.ubuntu.com/"

        self.METARELEASE_URI = base_uri + "meta-release"

        self.METARELEASE_URI_LTS = base_uri + "meta-release-lts"

        self.METARELEASE_URI_UNSTABLE_POSTFIX = "-development"

        self.METARELEASE_URI_PROPOSED_POSTFIX = "-proposed"

        # check the meta-release config first

        parser = configparser.ConfigParser()

        if os.path.exists(self.CONF_METARELEASE):

            try:

                parser.read(self.CONF_METARELEASE)

            except configparser.Error as e:

                sys.stderr.write(

                    "ERROR: failed to read '%s':\n%s"

                    % (self.CONF_METARELEASE, e)

                )

                return

            # make changing the metarelease file and the location

            # for the files easy

            if parser.has_section("METARELEASE"):

                sec = "METARELEASE"

                for k in [

                    "URI",

                    "URI_LTS",

                    "URI_UNSTABLE_POSTFIX",

                    "URI_PROPOSED_POSTFIX",

                ]:

                    if parser.has_option(sec, k):

                        self._debug(

                            "%s: %s "

                            % (self.CONF_METARELEASE, parser.get(sec, k))

                        )

                        setattr(self, "%s_%s" % (sec, k), parser.get(sec, k))

        # check the config file first to figure if we want lts upgrades only

        parser = configparser.ConfigParser()

        if os.path.exists(self.CONF):

            try:

                parser.read(self.CONF)

            except configparser.Error as e:

                sys.stderr.write(

                    "ERROR: failed to read '%s':\n%s" % (self.CONF, e)

                )

                return

            # now check which specific url to use

            if parser.has_option("DEFAULT", "Prompt"):

                prompt = parser.get("DEFAULT", "Prompt").lower()

                if prompt == "never" or prompt == "no":

                    self.prompt = "never"

                    # nothing to do for this object

                    # FIXME: what about no longer supported?

                    self.downloaded.set()

                    return

                elif prompt == "lts":

                    self.prompt = "lts"

                    # the Prompt=lts setting only makes sense when running on

                    # a LTS, otherwise it would result in users not receiving

                    # any distro upgrades

                    di = distro_info.UbuntuDistroInfo()

                    if di.is_lts(self.current_dist_name):

                        self.METARELEASE_URI = self.METARELEASE_URI_LTS

                    else:

                        self._debug("Prompt=lts for non-LTS, ignoring")

                else:

                    self.prompt = "normal"

        # needed for the _tryUpgradeSelf() code in DistUpgradeController

        if forceLTS:

            self.METARELEASE_URI = self.METARELEASE_URI_LTS

        # devel and proposed "just" change the postfix

        if useDevelopmentRelease:

            self.METARELEASE_URI += self.METARELEASE_URI_UNSTABLE_POSTFIX

        elif useProposed:

            self.METARELEASE_URI += self.METARELEASE_URI_PROPOSED_POSTFIX

        self._debug("metarelease-uri: %s" % self.METARELEASE_URI)

        self.metarelease_information = None

        if not self._buildMetaReleaseFile():

            self._debug("_buildMetaReleaseFile failed")

            return

        # we start the download thread here and we have a timeout

        threading.Thread(target=self.download).start()

        # threading.Thread(target=self.check).start()

    def _buildMetaReleaseFile(self):

        # build the metarelease_file name

        self.METARELEASE_FILE = os.path.join(

            "/var/lib/update-manager/", os.path.basename(self.METARELEASE_URI)

        )

        # check if we can write to the global location, if not,

        # write to homedir

        try:

            open(self.METARELEASE_FILE, "a").close()

        except IOError:

            cache_dir = os.getenv(

                "XDG_CACHE_HOME", os.path.expanduser("~/.cache")

            )

            # Take special care when creating this directory; ~/.cache needs

            # to be created with mode 0700, but the other directories do

            # not.

            cache_parent_dir = os.path.split(cache_dir)[0]

            if not os.path.exists(cache_parent_dir):

                try:

                    os.makedirs(cache_parent_dir)

                except OSError as e:

                    sys.stderr.write("mkdir() failed: '%s'" % e)

                    return False

            if not os.path.exists(cache_dir):

                try:

                    os.mkdir(cache_dir, 0o700)

                except OSError as e:

                    sys.stderr.write("mkdir() failed: '%s'" % e)

                    return False

            path = os.path.join(cache_dir, "update-manager-core")

            if not os.path.exists(path):

                try:

                    os.mkdir(path)

                except OSError as e:

                    sys.stderr.write("mkdir() failed: '%s'" % e)

                    return False

            self.METARELEASE_FILE = os.path.join(

                path, os.path.basename(self.METARELEASE_URI)

            )

        # if it is empty, remove it to avoid I-M-S hits on empty file

        try:

            if os.path.getsize(self.METARELEASE_FILE) == 0:

                os.unlink(self.METARELEASE_FILE)

        except Exception:

            pass

        return True

    def dist_no_longer_supported(self, dist):

        """virtual function that is called when the distro is no longer

        supported

        """

        self.no_longer_supported = dist

    def new_dist_available(self, dist):

        """virtual function that is called when a new distro release

        is available

        """

        self.new_dist = dist

    def parse(self):

        self._debug("MetaRelease.parse()")

        current_dist_name = self.current_dist_name

        self._debug("current dist name: '%s'" % current_dist_name)

        current_dist = None

        dists = []

        # parse the metarelease_information file

        index_tag = apt_pkg.TagFile(self.metarelease_information)

        try:

            while index_tag.step():

                for required_key in ("Dist", "Version", "Supported", "Date"):

                    if required_key not in index_tag.section:

                        raise MetaReleaseParseError(

                            "Required key '%s' missing" % required_key

                        )

                name = index_tag.section["Dist"]

                self._debug("found distro name: '%s'" % name)

                rawdate = index_tag.section["Date"]

                parseddate = list(email.utils.parsedate(rawdate))

                parseddate[8] = 0  # assume no DST

                date = time.mktime(tuple(parseddate))

                supported = int(index_tag.section["Supported"])

                version = index_tag.section["Version"]

                # add the information to a new date object

                dist = Dist(name, version, date, supported)

                if "ReleaseNotes" in index_tag.section:

                    dist.releaseNotesURI = index_tag.section["ReleaseNotes"]

                    lang = get_lang()

                    if lang:

                        dist.releaseNotesURI += "?lang=%s" % lang

                if "ReleaseNotesHtml" in index_tag.section:

                    dist.releaseNotesHtmlUri = index_tag.section[

                        "ReleaseNotesHtml"

                    ]

                    query = self._get_release_notes_uri_query_string(dist)

                    if query:

                        dist.releaseNotesHtmlUri += query

                if "UpgradeTool" in index_tag.section:

                    dist.upgradeTool = index_tag.section["UpgradeTool"]

                if "UpgradeToolSignature" in index_tag.section:

                    dist.upgradeToolSig = index_tag.section[

                        "UpgradeToolSignature"

                    ]

                if "UpgradeBroken" in index_tag.section:

                    dist.upgrade_broken = index_tag.section["UpgradeBroken"]

                dists.append(dist)

                if name == current_dist_name:

                    current_dist = dist

        except apt_pkg.Error:

            raise MetaReleaseParseError(

                "Unable to parse %s" % self.METARELEASE_URI

            )

        self.metarelease_information.close()

        self.metarelease_information = None

        # first check if the current runing distro is in the meta-release

        # information. if not, we assume that we run on something not

        # supported and silently return

        if current_dist is None:

            self._debug("current dist not found in meta-release file\n")

            return False

        # then see what we can upgrade to

        upgradable_to = ""

        for dist in dists:

            if dist.date > current_dist.date:

                # Only offer to upgrade to an unsupported release if running

                # with useDevelopmentRelease, this way one can upgrade from an

                # LTS release to the next supported non-LTS release e.g. from

                # 14.04 to 15.04.

                if not dist.supported and not self.useDevelopmentRelease:

                    continue

                upgradable_to = dist

                self._debug("new dist: %s" % upgradable_to)

                break

        # only warn if unsupported and a new dist is available (because

        # the development version is also unsupported)

        if upgradable_to != "" and not current_dist.supported:

            self.upgradable_to = upgradable_to

            self.dist_no_longer_supported(current_dist)

        if upgradable_to != "":

            self.upgradable_to = upgradable_to

            self.new_dist_available(upgradable_to)

        # parsing done and sucessfully

        return True

    # the network thread that tries to fetch the meta-index file

    # can't touch the gui, runs as a thread

    def download(self):

        self._debug("MetaRelease.download()")

        lastmodified = 0

        req = Request(self.METARELEASE_URI)

        # make sure that we always get the latest file (#107716)

        req.add_header("Cache-Control", "No-Cache")

        req.add_header("Pragma", "no-cache")

        if os.access(self.METARELEASE_FILE, os.W_OK):

            try:

                lastmodified = os.stat(self.METARELEASE_FILE).st_mtime

            except OSError:

                pass

        if lastmodified > 0 and not self.forceDownload:

            req.add_header(

                "If-Modified-Since", time.asctime(time.gmtime(lastmodified))

            )

        try:

            # open

            uri = urlopen(req, timeout=20)

            # sometime there is a root owned meta-relase file

            # there, try to remove it so that we get it

            # with proper permissions

            if os.path.exists(self.METARELEASE_FILE) and not os.access(

                self.METARELEASE_FILE, os.W_OK

            ):

                try:

                    os.unlink(self.METARELEASE_FILE)

                except OSError as e:

                    print(

                        "Can't unlink '%s' (%s)" % (self.METARELEASE_FILE, e)

                    )

            # we may get exception here on e.g. disk full

            try:

                f = open(self.METARELEASE_FILE, "w+")

                for line in uri.readlines():

                    f.write(line.decode("UTF-8"))

                f.flush()

                f.seek(0, 0)

                self.metarelease_information = f

            except IOError:

                pass

            uri.close()

        # http error

        except HTTPError as e:

            # mvo: only reuse local info on "not-modified"

            if e.code == 304 and os.path.exists(self.METARELEASE_FILE):

                self._debug("reading file '%s'" % self.METARELEASE_FILE)

                self.metarelease_information = open(self.METARELEASE_FILE, "r")

            else:

                self._debug("result of meta-release download: '%s'" % e)

        # generic network error

        except (URLError, BadStatusLine, socket.timeout) as e:

            self._debug("result of meta-release download: '%s'" % e)

            print(

                "Failed to connect to %s. Check your Internet connection "

                "or proxy settings" % self.METARELEASE_URI

            )

        # now check the information we have

        if self.metarelease_information is not None:

            self._debug("have self.metarelease_information")

            try:

                self.parse()

            except Exception:

                logging.exception(

                    "parse failed for '%s'" % self.METARELEASE_FILE

                )

                # no use keeping a broken file around

                os.remove(self.METARELEASE_FILE)

            # we don't want to keep a meta-release file around when it

            # has a "Broken" flag, this ensures we are not bitten by

            # I-M-S/cache issues

            if self.new_dist and self.new_dist.upgrade_broken:

                os.remove(self.METARELEASE_FILE)

        else:

            self._debug("NO self.metarelease_information")

        self.downloaded.set()

    @property

    def downloading(self):

        return not self.downloaded.is_set()

    def _get_release_notes_uri_query_string(self, dist):

        q = "?"

        # get the lang

        lang = get_lang()

        if lang:

            q += "lang=%s&" % lang

        # get the os

        q += "os=%s&" % self.flavor

        # get the version to upgrade to

        q += "ver=%s" % dist.version

        # the archive didn't respond well to ? being %3F

        return quote(q, "/?")

    def _debug(self, msg):

        if self.DEBUG:

            sys.stderr.write(msg + "\n")

if __name__ == "__main__":

    meta = MetaReleaseCore(False, False)