# archival.py - revision archival for mercurial # # Copyright 2006 Vadim Gelfer # # This software may be used and distributed according to the terms of the # GNU General Public License version 2 or any later version. from __future__ import absolute_import import gzip import os import struct import tarfile import time import zipfile import zlib from .i18n import _ from .node import nullrev from .pycompat import open from . import ( error, formatter, match as matchmod, pycompat, scmutil, util, vfs as vfsmod, ) stringio = util.stringio # from unzip source code: _UNX_IFREG = 0x8000 _UNX_IFLNK = 0xA000 def tidyprefix(dest, kind, prefix): '''choose prefix to use for names in archive. make sure prefix is safe for consumers.''' if prefix: prefix = util.normpath(prefix) else: if not isinstance(dest, bytes): raise ValueError(b'dest must be string if no prefix') prefix = os.path.basename(dest) lower = prefix.lower() for sfx in exts.get(kind, []): if lower.endswith(sfx): prefix = prefix[: -len(sfx)] break lpfx = os.path.normpath(util.localpath(prefix)) prefix = util.pconvert(lpfx) if not prefix.endswith(b'/'): prefix += b'/' # Drop the leading '.' path component if present, so Windows can read the # zip files (issue4634) if prefix.startswith(b'./'): prefix = prefix[2:] if prefix.startswith(b'../') or os.path.isabs(lpfx) or b'/../' in prefix: raise error.Abort(_(b'archive prefix contains illegal components')) return prefix exts = { b'tar': [b'.tar'], b'tbz2': [b'.tbz2', b'.tar.bz2'], b'tgz': [b'.tgz', b'.tar.gz'], b'zip': [b'.zip'], b'txz': [b'.txz', b'.tar.xz'], } def guesskind(dest): for kind, extensions in pycompat.iteritems(exts): if any(dest.endswith(ext) for ext in extensions): return kind return None def _rootctx(repo): # repo[0] may be hidden for rev in repo: return repo[rev] return repo[nullrev] # {tags} on ctx includes local tags and 'tip', with no current way to limit # that to global tags. Therefore, use {latesttag} as a substitute when # the distance is 0, since that will be the list of global tags on ctx. _defaultmetatemplate = br''' repo: {root} node: {ifcontains(rev, revset("wdir()"), "{p1node}{dirty}", "{node}")} branch: {branch|utf8} {ifeq(latesttagdistance, 0, join(latesttag % "tag: {tag}", "\n"), separate("\n", join(latesttag % "latesttag: {tag}", "\n"), "latesttagdistance: {latesttagdistance}", "changessincelatesttag: {changessincelatesttag}"))} '''[ 1: ] # drop leading '\n' def buildmetadata(ctx): '''build content of .hg_archival.txt''' repo = ctx.repo() opts = { b'template': repo.ui.config( b'experimental', b'archivemetatemplate', _defaultmetatemplate ) } out = util.stringio() fm = formatter.formatter(repo.ui, out, b'archive', opts) fm.startitem() fm.context(ctx=ctx) fm.data(root=_rootctx(repo).hex()) if ctx.rev() is None: dirty = b'' if ctx.dirty(missing=True): dirty = b'+' fm.data(dirty=dirty) fm.end() return out.getvalue() class tarit(object): '''write archive to tar file or stream. can write uncompressed, or compress with gzip or bzip2.''' class GzipFileWithTime(gzip.GzipFile): def __init__(self, *args, **kw): timestamp = None if 'timestamp' in kw: timestamp = kw.pop('timestamp') if timestamp is None: self.timestamp = time.time() else: self.timestamp = timestamp gzip.GzipFile.__init__(self, *args, **kw) def _write_gzip_header(self): self.fileobj.write(b'\037\213') # magic header self.fileobj.write(b'\010') # compression method fname = self.name if fname and fname.endswith(b'.gz'): fname = fname[:-3] flags = 0 if fname: flags = gzip.FNAME # pytype: disable=module-attr self.fileobj.write(pycompat.bytechr(flags)) gzip.write32u( # pytype: disable=module-attr self.fileobj, int(self.timestamp) ) self.fileobj.write(b'\002') self.fileobj.write(b'\377') if fname: self.fileobj.write(fname + b'\000') def __init__(self, dest, mtime, kind=b''): self.mtime = mtime self.fileobj = None def taropen(mode, name=b'', fileobj=None): if kind == b'gz': mode = mode[0:1] if not fileobj: fileobj = open(name, mode + b'b') gzfileobj = self.GzipFileWithTime( name, pycompat.sysstr(mode + b'b'), zlib.Z_BEST_COMPRESSION, fileobj, timestamp=mtime, ) self.fileobj = gzfileobj return tarfile.TarFile.taropen( # pytype: disable=attribute-error name, pycompat.sysstr(mode), gzfileobj ) else: return tarfile.open(name, pycompat.sysstr(mode + kind), fileobj) if isinstance(dest, bytes): self.z = taropen(b'w:', name=dest) else: self.z = taropen(b'w|', fileobj=dest) def addfile(self, name, mode, islink, data): name = pycompat.fsdecode(name) i = tarfile.TarInfo(name) i.mtime = self.mtime i.size = len(data) if islink: i.type = tarfile.SYMTYPE i.mode = 0o777 i.linkname = pycompat.fsdecode(data) data = None i.size = 0 else: i.mode = mode data = stringio(data) self.z.addfile(i, data) def done(self): self.z.close() if self.fileobj: self.fileobj.close() class zipit(object): '''write archive to zip file or stream. can write uncompressed, or compressed with deflate.''' def __init__(self, dest, mtime, compress=True): if isinstance(dest, bytes): dest = pycompat.fsdecode(dest) self.z = zipfile.ZipFile( dest, 'w', compress and zipfile.ZIP_DEFLATED or zipfile.ZIP_STORED ) # Python's zipfile module emits deprecation warnings if we try # to store files with a date before 1980. epoch = 315532800 # calendar.timegm((1980, 1, 1, 0, 0, 0, 1, 1, 0)) if mtime < epoch: mtime = epoch self.mtime = mtime self.date_time = time.gmtime(mtime)[:6] def addfile(self, name, mode, islink, data): i = zipfile.ZipInfo(pycompat.fsdecode(name), self.date_time) i.compress_type = self.z.compression # pytype: disable=attribute-error # unzip will not honor unix file modes unless file creator is # set to unix (id 3). i.create_system = 3 ftype = _UNX_IFREG if islink: mode = 0o777 ftype = _UNX_IFLNK i.external_attr = (mode | ftype) << 16 # add "extended-timestamp" extra block, because zip archives # without this will be extracted with unexpected timestamp, # if TZ is not configured as GMT i.extra += struct.pack( b'