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:
MetaRelease.py
# 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 from __future__ import absolute_import, print_function import apt import apt_pkg import distro_info try: import configparser except ImportError: import ConfigParser as configparser try: from http.client import BadStatusLine except ImportError: from httplib import BadStatusLine import logging import email.utils import os import socket import sys import time import threading try: from urllib.parse import quote from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError except ImportError: from urllib2 import HTTPError, Request, URLError, urlopen, quote 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)
Edit
Download
Unzip
Chmod
Delete
