← Back
Editing: debian_support.py
""" Facilities to deal with Debian-specific metadata """ # Copyright (C) 2005 Florian Weimer <fw@deneb.enyo.de> # Copyright (C) 2010 John Wright <jsw@debian.org> # # 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA import os import os.path import re try: # pylint: disable=unused-import from typing import ( Any, AnyStr, BinaryIO, Dict, Iterable, Iterator, Generator, List, Match, NoReturn, Optional, Pattern, Text, TextIO, Tuple, Union, ) except ImportError: # Lack of typing is not important at runtime pass from debian.deprecation import function_deprecated_by try: import apt_pkg try: apt_pkg.init() _have_apt_pkg = True except apt_pkg.Error: # If dpkg (e.g., tupledata) is missing, we can import apt_pkg but .init() # will raise an exception _have_apt_pkg = False except ImportError: _have_apt_pkg = False # Use the built-in _sha extension instead of hashlib to avoid a dependency on # OpenSSL, which is incompatible with the GPL. try: import _sha1 # type: ignore new_sha1 = _sha1.sha1 except ImportError: def new_sha1(*args): # pylint: disable=unused-argument # type: (bytes) -> str raise NotImplementedError( "Built-in sha1 implementation not found; cannot use hashlib" " implementation because it depends on OpenSSL, which" " may not be linked with this library due to license" " incompatibilities") # Use the built-in _sha256 extension instead of hashlib to avoid a dependency on # OpenSSL, which is incompatible with the GPL. try: import _sha256 # type: ignore new_sha256 = _sha256.sha256 except ImportError: def new_sha256(*args): # pylint: disable=unused-argument # type: (bytes) -> str raise NotImplementedError( "Built-in sha1 implementation not found; cannot use hashlib" " implementation because it depends on OpenSSL, which" " may not be linked with this library due to license" " incompatibilities") # Re-exports import debian._arch_table DpkgArchTable = debian._arch_table.DpkgArchTable class ParseError(Exception): """An exception which is used to signal a parse failure. Attributes: filename - name of the file lineno - line number in the file msg - error message """ def __init__(self, filename, # type: str lineno, # type: int msg # type: str ): # type: (...) -> None assert isinstance(lineno, int) self.filename = filename self.lineno = lineno self.msg = msg super(ParseError, self).__init__(self) def __str__(self): # type: () -> str return self.msg def __repr__(self): # type: () -> str return "ParseError(%r, %d, %r)" % (self.filename, self.lineno, self.msg) def print_out(self, file): # type: (TextIO) -> None """Writes a machine-parsable error message to file.""" file.write("%s:%d: %s\n" % (self.filename, self.lineno, self.msg)) file.flush() printOut = function_deprecated_by(print_out) class BaseVersion(object): """Base class for classes representing Debian versions It doesn't implement any comparison, but it does check for valid versions according to Section 5.6.12 in the Debian Policy Manual. Since splitting the version into epoch, upstream_version, and debian_revision components is pretty much free with the validation, it sets those fields as properties of the object, and sets the raw version to the full_version property. A missing epoch or debian_revision results in the respective property set to None. Setting any of the properties results in the full_version being recomputed and the rest of the properties set from that. It also implements __str__, just returning the raw version given to the initializer. """ re_valid_version = re.compile( r"^((?P<epoch>\d+):)?" "(?P<upstream_version>[A-Za-z0-9.+:~-]+?)" "(-(?P<debian_revision>[A-Za-z0-9+.~]+))?$") magic_attrs = ( 'full_version', 'epoch', 'upstream_version', 'debian_revision', 'debian_version') def __init__(self, version): # type: (Optional[Union[str, BaseVersion]]) -> None if isinstance(version, BaseVersion): version = str(version) self.full_version = version def _set_full_version(self, version): # type: (str) -> None m = self.re_valid_version.match(version) if not m: raise ValueError("Invalid version string %r" % version) # If there no epoch ("1:..."), then the upstream version can not # contain a :. if m.group("epoch") is None and ":" in m.group("upstream_version"): raise ValueError("Invalid version string %r" % version) # pylint: disable=attribute-defined-outside-init self.__full_version = version # pylint: disable = unused-private-member self.__epoch = m.group("epoch") self.__upstream_version = m.group("upstream_version") self.__debian_revision = m.group("debian_revision") def __setattr__(self, attr, value): # type: (str, Optional[Text]) -> None if attr not in self.magic_attrs: super(BaseVersion, self).__setattr__(attr, value) return # For compatibility with the old changelog.Version class if attr == "debian_version": attr = "debian_revision" if attr == "full_version": self._set_full_version(str(value)) else: if value is not None: value = str(value) private = "_BaseVersion__%s" % attr old_value = getattr(self, private) setattr(self, private, value) try: self._update_full_version() except ValueError: # Don't leave it in an invalid state setattr(self, private, old_value) self._update_full_version() raise ValueError("Setting %s to %r results in invalid version" % (attr, value)) def __getattr__(self, attr): # type: (str) -> Optional[str] if attr not in self.magic_attrs: return super(BaseVersion, self).__getattribute__(attr) # type: ignore # For compatibility with the old changelog.Version class if attr == "debian_version": attr = "debian_revision" private = "_BaseVersion__%s" % attr return getattr(self, private) # type: ignore def _update_full_version(self): # type: () -> None version = "" if self.__epoch is not None: version += self.__epoch + ":" version += self.__upstream_version if self.__debian_revision: version += "-" + self.__debian_revision self.full_version = version def __str__(self): # type: () -> str return self.full_version if self.full_version is not None else "" def __repr__(self): # type: () -> str return "%s('%s')" % (self.__class__.__name__, self) def _compare(self, other): # type: (Any) -> int raise NotImplementedError # TODO: Once we support only Python >= 2.7, we can simplify this using # @functools.total_ordering. def __lt__(self, other): # type: (Any) -> bool return self._compare(other) < 0 def __le__(self, other): # type: (Any) -> bool return self._compare(other) <= 0 def __eq__(self, other): # type: (Any) -> bool return self._compare(other) == 0 def __ne__(self, other): # type: (Any) -> bool return self._compare(other) != 0 def __ge__(self, other): # type: (Any) -> bool return self._compare(other) >= 0 def __gt__(self, other): # type: (Any) -> bool return self._compare(other) > 0 def __hash__(self): # type: () -> int return hash(str(self)) class AptPkgVersion(BaseVersion): """Represents a Debian package version, using apt_pkg.VersionCompare""" def __init__(self, version): # type: (Optional[Union[str, BaseVersion]]) -> None if not _have_apt_pkg: raise NotImplementedError("apt_pkg not available; install the " "python-apt package") super(AptPkgVersion, self).__init__(version) def _compare(self, other): # type: (Any) -> int return apt_pkg.version_compare(str(self), str(other)) # NativeVersion based on the DpkgVersion class by Raphael Hertzog in # svn://svn.debian.org/qa/trunk/pts/www/bin/common.py r2361 class NativeVersion(BaseVersion): """Represents a Debian package version, with native Python comparison""" re_all_digits_or_not = re.compile(r"\d+|\D+") re_digits = re.compile(r"\d+") re_digit = re.compile(r"\d") re_alpha = re.compile("[A-Za-z]") def _compare(self, other): # type: (Any) -> int # Convert other into an instance of BaseVersion if it's not already. # (All we need is epoch, upstream_version, and debian_revision # attributes, which BaseVersion gives us.) Requires other's string # representation to be the raw version. # If other is not defined, then the current version is bigger if other is None: return 1 if not isinstance(other, BaseVersion): try: other = BaseVersion(str(other)) except ValueError as e: raise ValueError("Couldn't convert %r to BaseVersion: %s" % (other, e)) lepoch = int(self.epoch or "0") repoch = int(other.epoch or "0") if lepoch < repoch: return -1 if lepoch > repoch: return 1 res = self._version_cmp_part(self.upstream_version or "0", other.upstream_version or "0") if res != 0: return res return self._version_cmp_part(self.debian_revision or "0", other.debian_revision or "0") @classmethod def _order(cls, x): # type: (str) -> int """Return an integer value for character x""" if x == '~': return -1 if cls.re_digit.match(x): return int(x) + 1 if cls.re_alpha.match(x): return ord(x) return ord(x) + 256 @classmethod def _version_cmp_string(cls, va, vb): # type: (str, str) -> int la = [cls._order(x) for x in va] lb = [cls._order(x) for x in vb] while la or lb: a = 0 b = 0 if la: a = la.pop(0) if lb: b = lb.pop(0) if a < b: return -1 if a > b: return 1 return 0 @classmethod def _version_cmp_part(cls, va, vb): # type: (str, str) -> int la = cls.re_all_digits_or_not.findall(va) lb = cls.re_all_digits_or_not.findall(vb) while la or lb: a = "0" b = "0" if la: a = la.pop(0) if lb: b = lb.pop(0) if cls.re_digits.match(a) and cls.re_digits.match(b): aval = int(a) bval = int(b) if aval < bval: return -1 if aval > bval: return 1 else: res = cls._version_cmp_string(a, b) if res != 0: return res return 0 if _have_apt_pkg: class Version(AptPkgVersion): pass else: class Version(NativeVersion): # type: ignore pass def version_compare(a, b): # type: (Any, Any) -> int va = Version(a) vb = Version(b) if va < vb: return -1 if va > vb: return 1 return 0 class PackageFile: """A Debian package file. Objects of this class can be used to read Debian's Source and Packages files.""" re_field = re.compile(r'^([A-Za-z][A-Za-z0-9-_]+):(?:\s*(.*?))?\s*$') re_continuation = re.compile(r'^\s+(?:\.|(\S.*?)\s*)$') def __init__(self, name, # type: str file_obj=None, # type: Optional[Union[TextIO, BinaryIO]] encoding="utf-8", # type: str ): # type: (...) -> None """Creates a new package file object. name - the name of the file the data comes from file_obj - an alternate data source; the default is to open the file with the indicated name. """ if file_obj is None: file_obj = open(name, 'rb') # pylint: disable=consider-using-with self.name = name self.file = file_obj self.lineno = 0 self.encoding = encoding def __iter__(self): # type: () -> Generator[List[Tuple[str, str]], None, None] line = self._aux_read_line() self.lineno += 1 pkg = [] # type: List[Tuple[str, str]] while line: if line.strip(' \t') == '\n': if not pkg: self.raise_syntax_error('expected package record') yield pkg pkg = [] line = self._aux_read_line() self.lineno += 1 continue match = self.re_field.match(line) # type: Optional[Match[str]] if not match: self.raise_syntax_error("expected package field") (name, contents) = match.groups() contents = contents or '' while True: line = self._aux_read_line() self.lineno += 1 match = self.re_continuation.match(line) if match: (ncontents,) = match.groups() if ncontents is None: ncontents = "" contents = "%s\n%s" % (contents, ncontents) else: break pkg.append((name, contents)) if pkg: yield pkg def _aux_read_line(self): # type: () -> str # Not always readline returns a byte object, also str # can be returned (i.e: StringIO) line = self.file.readline() if isinstance(line, bytes): return line.decode(self.encoding) return line def raise_syntax_error(self, msg, lineno=None): # type: (str, Optional[int]) -> NoReturn if lineno is None: lineno = self.lineno raise ParseError(self.name, lineno, msg) raiseSyntaxError = function_deprecated_by(raise_syntax_error) class PseudoEnum: """A base class for types which resemble enumeration types.""" def __init__(self, name, # type: str order, # type: Any ): self._name = name self._order = order def __repr__(self): # type: () -> str return '%s(%r)' % (self.__class__.__name__, self._name) def __str__(self): # type: () -> str return self._name # TODO: Once we support only Python >= 2.7, we can simplify this using # @functools.total_ordering. def __lt__(self, other): # type: (Any) -> Any return self._order < other._order def __le__(self, other): # type: (Any) -> Any return self._order <= other._order def __eq__(self, other): # type: (Any) -> Any return self._order == other._order def __ne__(self, other): # type: (Any) -> Any return self._order != other._order def __ge__(self, other): # type: (Any) -> Any return self._order >= other._order def __gt__(self, other): # type: (Any) -> Any return self._order > other._order def __hash__(self): # type: () -> int return hash(self._order) class Release(PseudoEnum): """ Debian release defined with respect to its name, order of release and version. The latter can be empty in case of 'sid'. See https://www.debian.org/releases/ """ releases = {} # type: Dict[str, Release] def __init__(self, name, # type: str order, # type: Any version="" # type: str ): super(Release, self).__init__(name, order) self.version = version def list_releases(): # type: () -> Dict[str, Release] """ Returns dict of Debian releases """ releases = {} rels = (("buzz", "1.1"), ("rex", "1.2"), ("bo", "1.3"), ("hamm", "2.0"), ("slink", "2.1"), ("potato", "2.2"), ("woody", "3.0"), ("sarge", "3.1"), ("etch", "4.0"), ("lenny", "5.0"), ("squeeze", "6.0"), ("wheezy", "7"), ("jessie", "8"), ("stretch", "9"), ("buster", "10"), ("bullseye", "11"), ("bookworm", "12"), ("trixie", "13"), ("forky", "14"), ("sid", "")) for idx, rel in enumerate(rels): name, version = rel releases[name] = Release(name, idx, version) Release.releases = releases return releases listReleases = function_deprecated_by(list_releases) _release_list = list_releases() def intern_release(name, releases=None): # type: (str, Optional[Any]) -> Any if releases is None: releases = _release_list return releases.get(name) internRelease = function_deprecated_by(intern_release) del listReleases del list_releases def read_lines_sha256(lines): # type: (Union[List[bytes], List[str]]) -> str m = new_sha256() for l in lines: if isinstance(l, bytes): m.update(l) else: m.update(l.encode("UTF-8")) return m.hexdigest() # type: ignore def read_lines_sha1(lines): # type: (Union[List[bytes], List[str]]) -> str m = new_sha1() for l in lines: if isinstance(l, bytes): m.update(l) else: m.update(l.encode("UTF-8")) return m.hexdigest() # type: ignore readLinesSHA1 = function_deprecated_by(read_lines_sha1) _patch_re_raw = r'^(\d+)(?:,(\d+))?([acd])$' _patch_re = re.compile(_patch_re_raw) # type: Pattern[str] _patch_re_b = re.compile(_patch_re_raw.encode('UTF-8')) # type: Pattern[bytes] def patches_from_ed_script( source, # type: Iterable[AnyStr] re_cmd=None, # type: Optional[Pattern[AnyStr]] ): # type: (...) -> Iterator[Tuple[int, int, List[AnyStr]]] """Converts source to a stream of patches. Patches are triples of line indexes: - number of the first line to be replaced - one plus the number of the last line to be replaced - list of line replacements This is enough to model arbitrary additions, deletions and replacements. """ i = iter(source) patch_re = re_cmd for line in i: if not patch_re: patch_re = _patch_re_b if isinstance(line, bytes) else _patch_re match = patch_re.match(line) if match is None: raise ValueError("invalid patch command: %r" % line) (first_, last_, cmd) = match.groups() first = int(first_) last = None if last_ is None else int(last_) # using ord() makes this work for str and bytes objects if ord(cmd) == 100: # cmd == d first = first - 1 if last is None: last = first + 1 yield (first, last, []) continue if ord(cmd) == 97: # cmd == a if last is not None: raise ValueError("invalid patch argument: %r" % line) last = first else: # cmd == c first = first - 1 if last is None: last = first + 1 lines = [] for c in i: if c in ('', b''): raise ValueError("end of stream in command: %r" % line) if c in ('.\n', '.', b'.\n', b'.'): break lines.append(c) yield (first, last, lines) patchesFromEdScript = function_deprecated_by(patches_from_ed_script) def patch_lines( lines, # type: List[AnyStr] patches, # type: Iterable[Tuple[int, int, List[AnyStr]]] ): # type: (...) -> None """Applies patches to lines. Updates lines in place.""" for (first, last, args) in patches: lines[first:last] = args patchLines = function_deprecated_by(patch_lines) def replace_file(lines, local, encoding="UTF-8"): # type: (List[str], str, str) -> None local_new = local + '.new' try: with open(local_new, 'w+', encoding=encoding) as new_file: for l in lines: new_file.write(l) os.replace(local_new, local) finally: if os.path.exists(local_new): os.unlink(local_new) replaceFile = function_deprecated_by(replace_file) def download_gunzip_lines(remote): # type: (str) -> List[str] """Downloads a file from a remote location and gunzips it. Returns the lines in the file.""" # pylint: disable=import-outside-toplevel import gzip from urllib.request import urlopen with urlopen(remote) as zfd: with gzip.open(zfd, mode="rt") as gfd: return gfd.readlines() # type: ignore downloadGunzipLines = function_deprecated_by(download_gunzip_lines) def download_file(remote, local): # type: (str, str) -> List[str] """Copies a gzipped remote file to the local system. remote - URL, without the .gz suffix local - name of the local file """ lines = download_gunzip_lines(remote + '.gz') replace_file(lines, local) return lines downloadFile = function_deprecated_by(download_file) def update_file(remote, local, verbose=False): # type: (str, str, bool) -> List[str] """Updates the local file by downloading a remote patch. Returns a list of lines in the local file. """ try: with open(local, 'r', encoding="UTF-8") as local_file: lines = local_file.readlines() except IOError: if verbose: print("update_file: no local copy, downloading full file") return download_file(remote, local) patches_to_apply = [] # type: List[str] patch_hashes = {} # type: Dict[str, str] # pylint: disable=import-outside-toplevel from urllib.request import urlopen index_name = remote + '.diff/Index' re_whitespace = re.compile(r'\s+') try: with urlopen(index_name) as index_url: index_fields = list(PackageFile(index_name, index_url)) except ParseError: # FIXME: urllib does not raise a proper exception, so we parse # the error message. if verbose: print("update_file: could not interpret patch index file") return download_file(remote, local) except IOError: if verbose: print("update_file: could not download patch index file") return download_file(remote, local) if 'SHA256-Current' in [k for fields in index_fields for k,v in fields]: if verbose: print("update_file: using sha256") prefix = 'SHA256' local_hash = read_lines_sha256(lines) read_lines = read_lines_sha256 else: if verbose: print("update_file: using sha1") prefix = 'SHA1' local_hash = read_lines_sha1(lines) read_lines = read_lines_sha1 for fields in index_fields: for (field, value) in fields: if field == prefix+'-Current': (remote_hash, _) = re_whitespace.split(value) if local_hash == remote_hash: if verbose: print("update_file: local file is up-to-date") return lines continue if field == prefix+'-History': for entry in value.splitlines(): if entry == '': continue (hist_hash, _, patch_name) = re_whitespace.split(entry) # After the first patch, we have to apply all # remaining patches. if patches_to_apply or hist_hash == local_hash: patches_to_apply.append(patch_name) continue if field == prefix+'-Patches': for entry in value.splitlines(): if entry == '': continue (patch_hash, _, patch_name) = re_whitespace.split(entry) patch_hashes[patch_name] = patch_hash continue if verbose: print("update_file: field %r ignored" % field) if not patches_to_apply: if verbose: print("update_file: could not find historic entry", local_hash) return download_file(remote, local) for patch_name in patches_to_apply: if verbose: print("update_file: downloading patch %r" % patch_name) patch_contents = download_gunzip_lines( remote + '.diff/' + patch_name + '.gz') if read_lines(patch_contents) != patch_hashes[patch_name]: raise ValueError("patch %r was garbled" % patch_name) patch_contents_unicode = list(patch_contents) patch_lines(lines, patches_from_ed_script(patch_contents_unicode)) new_hash = read_lines(lines) if new_hash != remote_hash: raise ValueError("patch failed, got %s instead of %s" % (new_hash, remote_hash)) replace_file(lines, local) return lines updateFile = function_deprecated_by(update_file) def merge_as_sets(*args): # type: ignore """Create an order set (represented as a list) of the objects in the sequences passed as arguments.""" s = {} for x in args: for y in x: s[y] = True return sorted(s) mergeAsSets = function_deprecated_by(merge_as_sets)
Save File
Cancel