Search
Search
Search
Search
Information
Information
Light
Dark
Open actions menu
Basic upload method
Bypass upload method
Tips!
If you encounter an error (by firewall) while uploading using both methods,
try changing extension of the file before uploading it and rename it right after.
This uploader supports multiple file upload.
Submit
~
lib
python3
dist-packages
DistUpgrade
File Content:
DistUpgradeFetcherCore.py
# 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"))
Edit
Download
Unzip
Chmod
Delete
