FM / DistUpgrade

/usr/lib/python3/dist-packages/DistUpgrade/

/usr/lib/python3/dist-packages/DistUpgrade

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
apt_btrfs_snapshot.py	9915	0644	edit dl rm
apt_clone.py	33530	0644	edit dl rm
dist-upgrade.py	129	0644	edit dl rm
distro.py	23763	0644	edit dl rm
DistUpgradeApport.py	5058	0644	edit dl rm
DistUpgradeCache.py	54225	0644	edit dl rm
DistUpgradeConfigParser.py	3837	0644	edit dl rm
DistUpgradeController.py	101282	0644	edit dl rm
DistUpgradeFetcher.py	6009	0644	edit dl rm
DistUpgradeFetcherCore.py	12745	0644	edit dl rm
DistUpgradeFetcherKDE.py	9950	0644	edit dl rm
DistUpgradeFetcherSelf.py	1955	0644	edit dl rm
DistUpgradeGettext.py	3043	0644	edit dl rm
DistUpgradeMain.py	9359	0644	edit dl rm
DistUpgradePatcher.py	4018	0644	edit dl rm
DistUpgradeQuirks.py	70360	0644	edit dl rm
DistUpgradeVersion.py	21	0644	edit dl rm
DistUpgradeView.py	17344	0644	edit dl rm
DistUpgradeViewGtk3.py	34670	0644	edit dl rm
DistUpgradeViewKDE.py	41846	0644	edit dl rm
DistUpgradeViewNonInteractive.py	13862	0644	edit dl rm
DistUpgradeViewText.py	12829	0644	edit dl rm
GtkProgress.py	4020	0644	edit dl rm
MetaRelease.py	17543	0644	edit dl rm
QUrlOpener.py	3321	0644	edit dl rm
ReleaseNotesViewer.py	7658	0644	edit dl rm
ReleaseNotesViewerWebkit.py	2983	0644	edit dl rm
SimpleGtk3builderApp.py	2056	0644	edit dl rm
SimpleGtkbuilderApp.py	2037	0644	edit dl rm
telemetry.py	3478	0644	edit dl rm
utils.py	18541	0644	edit dl rm
xorg_fix_proprietary.py	4083	0644	edit dl rm
__init__.py	0	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/DistUpgrade/DistUpgradeFetcherCore.py (12745B)

# DistUpgradeFetcherCore.py
# -*- Mode: Python; indent-tabs-mode: nil; tab-width: 4; coding: utf-8 -*-
#
#  Copyright (c) 2006 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

from string import Template
import os
import apt_pkg
import logging
import tarfile
import tempfile
import shutil
import socket
import sys
import subprocess
from gettext import gettext as _
from aptsources.sourceslist import SourcesList
from urllib.request import urlopen
from urllib.error import HTTPError

from .utils import get_dist, url_downloadable, country_mirror
from .DistUpgradeViewText import readline

class DistUpgradeFetcherCore(object):
    " base class (without GUI) for the upgrade fetcher "

DEFAULT_MIRROR = "http://archive.ubuntu.com/ubuntu"
    DEFAULT_COMPONENT = "main"
    DEBUG = "DEBUG_UPDATE_MANAGER" in os.environ

def __init__(self, new_dist, progress):
        self.new_dist = new_dist
        self.current_dist_name = get_dist()
        self._progress = progress
        # options to pass to the release upgrader when it is run
        self.run_options = []

def _debug(self, msg):
        " helper to show debug information "
        if self.DEBUG:
            sys.stderr.write(msg + "\n")

def showReleaseNotes(self):
        if '--frontend=DistUpgradeViewNonInteractive' in self.run_options:
            return True
        if self.new_dist.releaseNotesURI is not None:
            uri = self._expandUri(self.new_dist.releaseNotesURI)
            timeout = socket.getdefaulttimeout()
            try:
                socket.setdefaulttimeout(5)
                release_notes = urlopen(uri)
                notes = release_notes.read().decode("UTF-8", "replace")
            except HTTPError:
                self.error(_("Could not find the release announcement"),
                           _("The server may be overloaded."))
                return False
            except IOError:
                self.error(_("Could not download the release announcement"),
                           _("Please check your internet connection."))
                return False
            socket.setdefaulttimeout(timeout)
        print()
        print(notes)
        print(_("Continue [yN] "), end="")
        res = readline()
        if res.strip().lower().startswith(_("y")):
            return True
        return False

def error(self, summary, message):
        """ minimal implementation for error display, should be overwriten
            by subclasses that want to more fancy method
        """
        print(summary)
        print(message)
        return False

def authenticate(self):
        if self.new_dist.upgradeToolSig:
            f = self.tmpdir + "/" + os.path.basename(self.new_dist.upgradeTool)
            sig = self.tmpdir + "/" + os.path.basename(
                self.new_dist.upgradeToolSig)
            print(_("authenticate '%(file)s' against '%(signature)s' ") % {
                'file': os.path.basename(f),
                'signature': os.path.basename(sig)})
            if self.gpgauthenticate(f, sig):
                return True
        return False

def gpgauthenticate(self, file, signature,
                        keyring=None):
        """ authenticated a file against a given signature, if no keyring
            is given use the apt default keyring
        """
        gpg = ["apt-key"]

if not keyring:
            keyring = '/usr/share/keyrings/ubuntu-archive-keyring.gpg'

gpg += ["--keyring", keyring]

gpg += ["verify", signature, file]
        ret = subprocess.call(gpg, stderr=subprocess.PIPE)
        return ret == 0

def extractDistUpgrader(self):
        # extract the tarball
        fname = os.path.join(self.tmpdir, os.path.basename(self.uri))
        print(_("extracting '%s'") % os.path.basename(fname))
        if not os.path.exists(fname):
            return False
        try:
            tar = tarfile.open(self.tmpdir + "/" +
                               os.path.basename(self.uri), "r")
            for tarinfo in tar:
                tar.extract(tarinfo)
            tar.close()
        except tarfile.ReadError as e:
            logging.error("failed to open tarfile (%s)" % e)
            return False
        return True

def verifyDistUprader(self):
        # FIXME: check an internal dependency file to make sure
        #        that the script will run correctly

# see if we have a script file that we can run
        self.script = script = "%s/%s" % (self.tmpdir, self.new_dist.name)
        if not os.path.exists(script):
            return self.error(_("Could not run the upgrade tool"),
                              _("Could not run the upgrade tool") + ".  " +
                              _("This is most likely a bug in the upgrade "
                                "tool. Please report it as a bug using the "
                                "command 'ubuntu-bug "
                                "ubuntu-release-upgrader-core'."))
        return True

def mirror_from_sources_list(self, uri, default_uri):
        """
        try to figure what the mirror is from current sources.list

do this by looing for matching DEFAULT_COMPONENT, current dist
        in sources.list and then doing a http HEAD/ftp size request
        to see if the uri is available on this server
        """
        self._debug("mirror_from_sources_list: %s" % self.current_dist_name)
        sources = SourcesList(withMatcher=False)
        seen = set()
        for e in sources.list:
            if e.disabled or e.invalid or not e.type == "deb":
                continue
            # check if we probed this mirror already
            if e.uri in seen:
                continue
            # we are using the main mirror already, so we are fine
            if (e.uri.startswith(default_uri) and
                    e.dist == self.current_dist_name and
                    self.DEFAULT_COMPONENT in e.comps):
                return uri
            elif (e.dist == self.current_dist_name and "main" in e.comps):
                mirror_uri = e.uri + uri[len(default_uri):]
                if url_downloadable(mirror_uri, self._debug):
                    return mirror_uri
                seen.add(e.uri)
        self._debug("no mirror found")
        return ""

def _expandUri(self, uri):
        """
        expand the uri so that it uses a mirror if the url starts
        with a well known string (like archive.ubuntu.com)
        """
        # try to guess the mirror from the sources.list
        if uri.startswith(self.DEFAULT_MIRROR):
            self._debug("trying to find suitable mirror")
            new_uri = self.mirror_from_sources_list(uri, self.DEFAULT_MIRROR)
            if new_uri:
                return new_uri
        # if that fails, use old method
        uri_template = Template(uri)
        m = country_mirror()
        new_uri = uri_template.safe_substitute(countrymirror=m)
        # be paranoid and check if the given uri is really downloadable
        try:
            if not url_downloadable(new_uri, self._debug):
                raise Exception("failed to download %s" % new_uri)
        except Exception as e:
            self._debug("url '%s' could not be downloaded" % e)
            # else fallback to main server
            new_uri = uri_template.safe_substitute(countrymirror='')
        return new_uri

def fetchDistUpgrader(self):
        " download the tarball with the upgrade script "
        tmpdir = tempfile.mkdtemp(prefix="ubuntu-release-upgrader-")
        self.tmpdir = tmpdir
        os.chdir(tmpdir)
        logging.debug("using tmpdir: '%s'" % tmpdir)
        # turn debugging on here (if required)
        if self.DEBUG > 0:
            apt_pkg.config.set("Debug::Acquire::http", "1")
            apt_pkg.config.set("Debug::Acquire::ftp", "1")
        #os.listdir(tmpdir)
        fetcher = apt_pkg.Acquire(self._progress)
        if self.new_dist.upgradeToolSig is not None:
            uri = self._expandUri(self.new_dist.upgradeToolSig)
            af1 = apt_pkg.AcquireFile(fetcher,
                                      uri,
                                      descr=_("Upgrade tool signature"))
            # reference it here to shut pyflakes up
            af1
        if self.new_dist.upgradeTool is not None:
            self.uri = self._expandUri(self.new_dist.upgradeTool)
            af2 = apt_pkg.AcquireFile(fetcher,
                                      self.uri,
                                      descr=_("Upgrade tool"))
            # reference it here to shut pyflakes up
            af2
            result = fetcher.run()
            if result != fetcher.RESULT_CONTINUE:
                logging.warning("fetch result != continue (%s)" % result)
                return False
            # check that both files are really there and non-null
            for f in [os.path.basename(self.new_dist.upgradeToolSig),
                      os.path.basename(self.new_dist.upgradeTool)]:
                if not (os.path.exists(f) and os.path.getsize(f) > 0):
                    logging.warning("file '%s' missing" % f)
                    return False
            return True
        return False

def runDistUpgrader(self):
        args = [self.script] + self.run_options
        if os.getuid() != 0:
            os.execv("/usr/bin/sudo", ["sudo", "-E"] + args)
        else:
            os.execv(self.script, args)

def cleanup(self):
        # cleanup
        os.chdir("..")
        # del tmpdir
        shutil.rmtree(self.tmpdir)

def run(self):
        # see if we have release notes
        if not self.showReleaseNotes():
            return
        if not self.fetchDistUpgrader():
            self.error(_("Failed to fetch"),
                       _("Fetching the upgrade failed. There may be a network "
                         "problem. "))
            return
        if not self.authenticate():
            self.error(_("Authentication failed"),
                       _("Authenticating the upgrade failed. There may be a "
                         "problem with the network or with the server. "))
            self.cleanup()
            return
        if not self.extractDistUpgrader():
            self.error(_("Failed to extract"),
                       _("Extracting the upgrade failed. There may be a "
                         "problem with the network or with the server. "))

return
        if not self.verifyDistUprader():
            self.error(_("Verification failed"),
                       _("Verifying the upgrade failed.  There may be a "
                         "problem with the network or with the server. "))
            self.cleanup()
            return
        try:
            # check if we can execute, if we run it via sudo we will
            # not know otherwise, pkexec will not raise a exception
            if not os.access(self.script, os.X_OK):
                ex = OSError("Can not execute '%s'" % self.script)
                ex.errno = 13
                raise ex
            self.runDistUpgrader()
        except OSError as e:
            if e.errno == 13:
                self.error(_("Can not run the upgrade"),
                           _("This usually is caused by a system where /tmp "
                             "is mounted noexec. Please remount without "
                             "noexec and run the upgrade again."))
                return False
            else:
                self.error(_("Can not run the upgrade"),
                           _("The error message is '%s'.") % e.strerror)
        return True

if __name__ == "__main__":
    d = DistUpgradeFetcherCore(None, None)
#    print(d.authenticate('/tmp/Release','/tmp/Release.gpg'))
    print("got mirror: '%s'" %
          d.mirror_from_sources_list(
              "http://archive.ubuntu.com/ubuntu/dists/intrepid-proposed/main/"
              "dist-upgrader-all/0.93.34/intrepid.tar.gz",
              "http://archive.ubuntu.com/ubuntu"))

cmd: