Aluode/PerceptionLabPortable
0
1#-------------------------------------------------------------------2# tarfile.py3#-------------------------------------------------------------------4# Copyright (C) 2002 Lars Gustaebel <lars@gustaebel.de>5# All rights reserved.6#7# Permission is hereby granted, free of charge, to any person8# obtaining a copy of this software and associated documentation9# files (the "Software"), to deal in the Software without10# restriction, including without limitation the rights to use,11# copy, modify, merge, publish, distribute, sublicense, and/or sell12# copies of the Software, and to permit persons to whom the13# Software is furnished to do so, subject to the following14# conditions:15#16# The above copyright notice and this permission notice shall be17# included in all copies or substantial portions of the Software.18#19# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,20# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES21# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND22# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT23# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,24# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING25# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR26# OTHER DEALINGS IN THE SOFTWARE.27#28"""Read from and write to tar format archives.29"""30 31version = "0.9.0"32__author__ = "Lars Gust\u00e4bel (lars@gustaebel.de)"33__credits__ = "Gustavo Niemeyer, Niels Gust\u00e4bel, Richard Townsend."34 35#---------36# Imports37#---------38from builtins import open as bltn_open39import sys40import os41import io42import shutil43import stat44import time45import struct46import copy47import re48 49from .compat.py38 import removesuffix50 51try:52 import pwd53except ImportError:54 pwd = None55try:56 import grp57except ImportError:58 grp = None59 60# os.symlink on Windows prior to 6.0 raises NotImplementedError61# OSError (winerror=1314) will be raised if the caller does not hold the62# SeCreateSymbolicLinkPrivilege privilege63symlink_exception = (AttributeError, NotImplementedError, OSError)64 65# from tarfile import *66__all__ = ["TarFile", "TarInfo", "is_tarfile", "TarError", "ReadError",67 "CompressionError", "StreamError", "ExtractError", "HeaderError",68 "ENCODING", "USTAR_FORMAT", "GNU_FORMAT", "PAX_FORMAT",69 "DEFAULT_FORMAT", "open","fully_trusted_filter", "data_filter",70 "tar_filter", "FilterError", "AbsoluteLinkError",71 "OutsideDestinationError", "SpecialFileError", "AbsolutePathError",72 "LinkOutsideDestinationError"]73 74 75#---------------------------------------------------------76# tar constants77#---------------------------------------------------------78NUL = b"\0" # the null character79BLOCKSIZE = 512 # length of processing blocks80RECORDSIZE = BLOCKSIZE * 20 # length of records81GNU_MAGIC = b"ustar \0" # magic gnu tar string82POSIX_MAGIC = b"ustar\x0000" # magic posix tar string83 84LENGTH_NAME = 100 # maximum length of a filename85LENGTH_LINK = 100 # maximum length of a linkname86LENGTH_PREFIX = 155 # maximum length of the prefix field87 88REGTYPE = b"0" # regular file89AREGTYPE = b"\0" # regular file90LNKTYPE = b"1" # link (inside tarfile)91SYMTYPE = b"2" # symbolic link92CHRTYPE = b"3" # character special device93BLKTYPE = b"4" # block special device94DIRTYPE = b"5" # directory95FIFOTYPE = b"6" # fifo special device96CONTTYPE = b"7" # contiguous file97 98GNUTYPE_LONGNAME = b"L" # GNU tar longname99GNUTYPE_LONGLINK = b"K" # GNU tar longlink100GNUTYPE_SPARSE = b"S" # GNU tar sparse file101 102XHDTYPE = b"x" # POSIX.1-2001 extended header103XGLTYPE = b"g" # POSIX.1-2001 global header104SOLARIS_XHDTYPE = b"X" # Solaris extended header105 106USTAR_FORMAT = 0 # POSIX.1-1988 (ustar) format107GNU_FORMAT = 1 # GNU tar format108PAX_FORMAT = 2 # POSIX.1-2001 (pax) format109DEFAULT_FORMAT = PAX_FORMAT110 111#---------------------------------------------------------112# tarfile constants113#---------------------------------------------------------114# File types that tarfile supports:115SUPPORTED_TYPES = (REGTYPE, AREGTYPE, LNKTYPE,116 SYMTYPE, DIRTYPE, FIFOTYPE,117 CONTTYPE, CHRTYPE, BLKTYPE,118 GNUTYPE_LONGNAME, GNUTYPE_LONGLINK,119 GNUTYPE_SPARSE)120 121# File types that will be treated as a regular file.122REGULAR_TYPES = (REGTYPE, AREGTYPE,123 CONTTYPE, GNUTYPE_SPARSE)124 125# File types that are part of the GNU tar format.126GNU_TYPES = (GNUTYPE_LONGNAME, GNUTYPE_LONGLINK,127 GNUTYPE_SPARSE)128 129# Fields from a pax header that override a TarInfo attribute.130PAX_FIELDS = ("path", "linkpath", "size", "mtime",131 "uid", "gid", "uname", "gname")132 133# Fields from a pax header that are affected by hdrcharset.134PAX_NAME_FIELDS = {"path", "linkpath", "uname", "gname"}135 136# Fields in a pax header that are numbers, all other fields137# are treated as strings.138PAX_NUMBER_FIELDS = {139 "atime": float,140 "ctime": float,141 "mtime": float,142 "uid": int,143 "gid": int,144 "size": int145}146 147#---------------------------------------------------------148# initialization149#---------------------------------------------------------150if os.name == "nt":151 ENCODING = "utf-8"152else:153 ENCODING = sys.getfilesystemencoding()154 155#---------------------------------------------------------156# Some useful functions157#---------------------------------------------------------158 159def stn(s, length, encoding, errors):160 """Convert a string to a null-terminated bytes object.161 """162 if s is None:163 raise ValueError("metadata cannot contain None")164 s = s.encode(encoding, errors)165 return s[:length] + (length - len(s)) * NUL166 167def nts(s, encoding, errors):168 """Convert a null-terminated bytes object to a string.169 """170 p = s.find(b"\0")171 if p != -1:172 s = s[:p]173 return s.decode(encoding, errors)174 175def nti(s):176 """Convert a number field to a python number.177 """178 # There are two possible encodings for a number field, see179 # itn() below.180 if s[0] in (0o200, 0o377):181 n = 0182 for i in range(len(s) - 1):183 n <<= 8184 n += s[i + 1]185 if s[0] == 0o377:186 n = -(256 ** (len(s) - 1) - n)187 else:188 try:189 s = nts(s, "ascii", "strict")190 n = int(s.strip() or "0", 8)191 except ValueError:192 raise InvalidHeaderError("invalid header")193 return n194 195def itn(n, digits=8, format=DEFAULT_FORMAT):196 """Convert a python number to a number field.197 """198 # POSIX 1003.1-1988 requires numbers to be encoded as a string of199 # octal digits followed by a null-byte, this allows values up to200 # (8**(digits-1))-1. GNU tar allows storing numbers greater than201 # that if necessary. A leading 0o200 or 0o377 byte indicate this202 # particular encoding, the following digits-1 bytes are a big-endian203 # base-256 representation. This allows values up to (256**(digits-1))-1.204 # A 0o200 byte indicates a positive number, a 0o377 byte a negative205 # number.206 original_n = n207 n = int(n)208 if 0 <= n < 8 ** (digits - 1):209 s = bytes("%0*o" % (digits - 1, n), "ascii") + NUL210 elif format == GNU_FORMAT and -256 ** (digits - 1) <= n < 256 ** (digits - 1):211 if n >= 0:212 s = bytearray([0o200])213 else:214 s = bytearray([0o377])215 n = 256 ** digits + n216 217 for i in range(digits - 1):218 s.insert(1, n & 0o377)219 n >>= 8220 else:221 raise ValueError("overflow in number field")222 223 return s224 225def calc_chksums(buf):226 """Calculate the checksum for a member's header by summing up all227 characters except for the chksum field which is treated as if228 it was filled with spaces. According to the GNU tar sources,229 some tars (Sun and NeXT) calculate chksum with signed char,230 which will be different if there are chars in the buffer with231 the high bit set. So we calculate two checksums, unsigned and232 signed.233 """234 unsigned_chksum = 256 + sum(struct.unpack_from("148B8x356B", buf))235 signed_chksum = 256 + sum(struct.unpack_from("148b8x356b", buf))236 return unsigned_chksum, signed_chksum237 238def copyfileobj(src, dst, length=None, exception=OSError, bufsize=None):239 """Copy length bytes from fileobj src to fileobj dst.240 If length is None, copy the entire content.241 """242 bufsize = bufsize or 16 * 1024243 if length == 0:244 return245 if length is None:246 shutil.copyfileobj(src, dst, bufsize)247 return248 249 blocks, remainder = divmod(length, bufsize)250 for b in range(blocks):251 buf = src.read(bufsize)252 if len(buf) < bufsize:253 raise exception("unexpected end of data")254 dst.write(buf)255 256 if remainder != 0:257 buf = src.read(remainder)258 if len(buf) < remainder:259 raise exception("unexpected end of data")260 dst.write(buf)261 return262 263def _safe_print(s):264 encoding = getattr(sys.stdout, 'encoding', None)265 if encoding is not None:266 s = s.encode(encoding, 'backslashreplace').decode(encoding)267 print(s, end=' ')268 269 270class TarError(Exception):271 """Base exception."""272 pass273class ExtractError(TarError):274 """General exception for extract errors."""275 pass276class ReadError(TarError):277 """Exception for unreadable tar archives."""278 pass279class CompressionError(TarError):280 """Exception for unavailable compression methods."""281 pass282class StreamError(TarError):283 """Exception for unsupported operations on stream-like TarFiles."""284 pass285class HeaderError(TarError):286 """Base exception for header errors."""287 pass288class EmptyHeaderError(HeaderError):289 """Exception for empty headers."""290 pass291class TruncatedHeaderError(HeaderError):292 """Exception for truncated headers."""293 pass294class EOFHeaderError(HeaderError):295 """Exception for end of file headers."""296 pass297class InvalidHeaderError(HeaderError):298 """Exception for invalid headers."""299 pass300class SubsequentHeaderError(HeaderError):301 """Exception for missing and invalid extended headers."""302 pass303 304#---------------------------305# internal stream interface306#---------------------------307class _LowLevelFile:308 """Low-level file object. Supports reading and writing.309 It is used instead of a regular file object for streaming310 access.311 """312 313 def __init__(self, name, mode):314 mode = {315 "r": os.O_RDONLY,316 "w": os.O_WRONLY | os.O_CREAT | os.O_TRUNC,317 }[mode]318 if hasattr(os, "O_BINARY"):319 mode |= os.O_BINARY320 self.fd = os.open(name, mode, 0o666)321 322 def close(self):323 os.close(self.fd)324 325 def read(self, size):326 return os.read(self.fd, size)327 328 def write(self, s):329 os.write(self.fd, s)330 331class _Stream:332 """Class that serves as an adapter between TarFile and333 a stream-like object. The stream-like object only334 needs to have a read() or write() method that works with bytes,335 and the method is accessed blockwise.336 Use of gzip or bzip2 compression is possible.337 A stream-like object could be for example: sys.stdin.buffer,338 sys.stdout.buffer, a socket, a tape device etc.339 340 _Stream is intended to be used only internally.341 """342 343 def __init__(self, name, mode, comptype, fileobj, bufsize,344 compresslevel):345 """Construct a _Stream object.346 """347 self._extfileobj = True348 if fileobj is None:349 fileobj = _LowLevelFile(name, mode)350 self._extfileobj = False351 352 if comptype == '*':353 # Enable transparent compression detection for the354 # stream interface355 fileobj = _StreamProxy(fileobj)356 comptype = fileobj.getcomptype()357 358 self.name = name or ""359 self.mode = mode360 self.comptype = comptype361 self.fileobj = fileobj362 self.bufsize = bufsize363 self.buf = b""364 self.pos = 0365 self.closed = False366 367 try:368 if comptype == "gz":369 try:370 import zlib371 except ImportError:372 raise CompressionError("zlib module is not available") from None373 self.zlib = zlib374 self.crc = zlib.crc32(b"")375 if mode == "r":376 self.exception = zlib.error377 self._init_read_gz()378 else:379 self._init_write_gz(compresslevel)380 381 elif comptype == "bz2":382 try:383 import bz2384 except ImportError:385 raise CompressionError("bz2 module is not available") from None386 if mode == "r":387 self.dbuf = b""388 self.cmp = bz2.BZ2Decompressor()389 self.exception = OSError390 else:391 self.cmp = bz2.BZ2Compressor(compresslevel)392 393 elif comptype == "xz":394 try:395 import lzma396 except ImportError:397 raise CompressionError("lzma module is not available") from None398 if mode == "r":399 self.dbuf = b""400 self.cmp = lzma.LZMADecompressor()401 self.exception = lzma.LZMAError402 else:403 self.cmp = lzma.LZMACompressor()404 405 elif comptype != "tar":406 raise CompressionError("unknown compression type %r" % comptype)407 408 except:409 if not self._extfileobj:410 self.fileobj.close()411 self.closed = True412 raise413 414 def __del__(self):415 if hasattr(self, "closed") and not self.closed:416 self.close()417 418 def _init_write_gz(self, compresslevel):419 """Initialize for writing with gzip compression.420 """421 self.cmp = self.zlib.compressobj(compresslevel,422 self.zlib.DEFLATED,423 -self.zlib.MAX_WBITS,424 self.zlib.DEF_MEM_LEVEL,425 0)426 timestamp = struct.pack("<L", int(time.time()))427 self.__write(b"\037\213\010\010" + timestamp + b"\002\377")428 if self.name.endswith(".gz"):429 self.name = self.name[:-3]430 # Honor "directory components removed" from RFC1952431 self.name = os.path.basename(self.name)432 # RFC1952 says we must use ISO-8859-1 for the FNAME field.433 self.__write(self.name.encode("iso-8859-1", "replace") + NUL)434 435 def write(self, s):436 """Write string s to the stream.437 """438 if self.comptype == "gz":439 self.crc = self.zlib.crc32(s, self.crc)440 self.pos += len(s)441 if self.comptype != "tar":442 s = self.cmp.compress(s)443 self.__write(s)444 445 def __write(self, s):446 """Write string s to the stream if a whole new block447 is ready to be written.448 """449 self.buf += s450 while len(self.buf) > self.bufsize:451 self.fileobj.write(self.buf[:self.bufsize])452 self.buf = self.buf[self.bufsize:]453 454 def close(self):455 """Close the _Stream object. No operation should be456 done on it afterwards.457 """458 if self.closed:459 return460 461 self.closed = True462 try:463 if self.mode == "w" and self.comptype != "tar":464 self.buf += self.cmp.flush()465 466 if self.mode == "w" and self.buf:467 self.fileobj.write(self.buf)468 self.buf = b""469 if self.comptype == "gz":470 self.fileobj.write(struct.pack("<L", self.crc))471 self.fileobj.write(struct.pack("<L", self.pos & 0xffffFFFF))472 finally:473 if not self._extfileobj:474 self.fileobj.close()475 476 def _init_read_gz(self):477 """Initialize for reading a gzip compressed fileobj.478 """479 self.cmp = self.zlib.decompressobj(-self.zlib.MAX_WBITS)480 self.dbuf = b""481 482 # taken from gzip.GzipFile with some alterations483 if self.__read(2) != b"\037\213":484 raise ReadError("not a gzip file")485 if self.__read(1) != b"\010":486 raise CompressionError("unsupported compression method")487 488 flag = ord(self.__read(1))489 self.__read(6)490 491 if flag & 4:492 xlen = ord(self.__read(1)) + 256 * ord(self.__read(1))493 self.read(xlen)494 if flag & 8:495 while True:496 s = self.__read(1)497 if not s or s == NUL:498 break499 if flag & 16:500 while True:501 s = self.__read(1)502 if not s or s == NUL:503 break504 if flag & 2:505 self.__read(2)506 507 def tell(self):508 """Return the stream's file pointer position.509 """510 return self.pos511 512 def seek(self, pos=0):513 """Set the stream's file pointer to pos. Negative seeking514 is forbidden.515 """516 if pos - self.pos >= 0:517 blocks, remainder = divmod(pos - self.pos, self.bufsize)518 for i in range(blocks):519 self.read(self.bufsize)520 self.read(remainder)521 else:522 raise StreamError("seeking backwards is not allowed")523 return self.pos524 525 def read(self, size):526 """Return the next size number of bytes from the stream."""527 assert size is not None528 buf = self._read(size)529 self.pos += len(buf)530 return buf531 532 def _read(self, size):533 """Return size bytes from the stream.534 """535 if self.comptype == "tar":536 return self.__read(size)537 538 c = len(self.dbuf)539 t = [self.dbuf]540 while c < size:541 # Skip underlying buffer to avoid unaligned double buffering.542 if self.buf:543 buf = self.buf544 self.buf = b""545 else:546 buf = self.fileobj.read(self.bufsize)547 if not buf:548 break549 try:550 buf = self.cmp.decompress(buf)551 except self.exception as e:552 raise ReadError("invalid compressed data") from e553 t.append(buf)554 c += len(buf)555 t = b"".join(t)556 self.dbuf = t[size:]557 return t[:size]558 559 def __read(self, size):560 """Return size bytes from stream. If internal buffer is empty,561 read another block from the stream.562 """563 c = len(self.buf)564 t = [self.buf]565 while c < size:566 buf = self.fileobj.read(self.bufsize)567 if not buf:568 break569 t.append(buf)570 c += len(buf)571 t = b"".join(t)572 self.buf = t[size:]573 return t[:size]574# class _Stream575 576class _StreamProxy(object):577 """Small proxy class that enables transparent compression578 detection for the Stream interface (mode 'r|*').579 """580 581 def __init__(self, fileobj):582 self.fileobj = fileobj583 self.buf = self.fileobj.read(BLOCKSIZE)584 585 def read(self, size):586 self.read = self.fileobj.read587 return self.buf588 589 def getcomptype(self):590 if self.buf.startswith(b"\x1f\x8b\x08"):591 return "gz"592 elif self.buf[0:3] == b"BZh" and self.buf[4:10] == b"1AY&SY":593 return "bz2"594 elif self.buf.startswith((b"\x5d\x00\x00\x80", b"\xfd7zXZ")):595 return "xz"596 else:597 return "tar"598 599 def close(self):600 self.fileobj.close()601# class StreamProxy602 603#------------------------604# Extraction file object605#------------------------606class _FileInFile(object):607 """A thin wrapper around an existing file object that608 provides a part of its data as an individual file609 object.610 """611 612 def __init__(self, fileobj, offset, size, name, blockinfo=None):613 self.fileobj = fileobj614 self.offset = offset615 self.size = size616 self.position = 0617 self.name = name618 self.closed = False619 620 if blockinfo is None:621 blockinfo = [(0, size)]622 623 # Construct a map with data and zero blocks.624 self.map_index = 0625 self.map = []626 lastpos = 0627 realpos = self.offset628 for offset, size in blockinfo:629 if offset > lastpos:630 self.map.append((False, lastpos, offset, None))631 self.map.append((True, offset, offset + size, realpos))632 realpos += size633 lastpos = offset + size634 if lastpos < self.size:635 self.map.append((False, lastpos, self.size, None))636 637 def flush(self):638 pass639 640 @property641 def mode(self):642 return 'rb'643 644 def readable(self):645 return True646 647 def writable(self):648 return False649 650 def seekable(self):651 return self.fileobj.seekable()652 653 def tell(self):654 """Return the current file position.655 """656 return self.position657 658 def seek(self, position, whence=io.SEEK_SET):659 """Seek to a position in the file.660 """661 if whence == io.SEEK_SET:662 self.position = min(max(position, 0), self.size)663 elif whence == io.SEEK_CUR:664 if position < 0:665 self.position = max(self.position + position, 0)666 else:667 self.position = min(self.position + position, self.size)668 elif whence == io.SEEK_END:669 self.position = max(min(self.size + position, self.size), 0)670 else:671 raise ValueError("Invalid argument")672 return self.position673 674 def read(self, size=None):675 """Read data from the file.676 """677 if size is None:678 size = self.size - self.position679 else:680 size = min(size, self.size - self.position)681 682 buf = b""683 while size > 0:684 while True:685 data, start, stop, offset = self.map[self.map_index]686 if start <= self.position < stop:687 break688 else:689 self.map_index += 1690 if self.map_index == len(self.map):691 self.map_index = 0692 length = min(size, stop - self.position)693 if data:694 self.fileobj.seek(offset + (self.position - start))695 b = self.fileobj.read(length)696 if len(b) != length:697 raise ReadError("unexpected end of data")698 buf += b699 else:700 buf += NUL * length701 size -= length702 self.position += length703 return buf704 705 def readinto(self, b):706 buf = self.read(len(b))707 b[:len(buf)] = buf708 return len(buf)709 710 def close(self):711 self.closed = True712#class _FileInFile713 714class ExFileObject(io.BufferedReader):715 716 def __init__(self, tarfile, tarinfo):717 fileobj = _FileInFile(tarfile.fileobj, tarinfo.offset_data,718 tarinfo.size, tarinfo.name, tarinfo.sparse)719 super().__init__(fileobj)720#class ExFileObject721 722 723#-----------------------------724# extraction filters (PEP 706)725#-----------------------------726 727class FilterError(TarError):728 pass729 730class AbsolutePathError(FilterError):731 def __init__(self, tarinfo):732 self.tarinfo = tarinfo733 super().__init__(f'member {tarinfo.name!r} has an absolute path')734 735class OutsideDestinationError(FilterError):736 def __init__(self, tarinfo, path):737 self.tarinfo = tarinfo738 self._path = path739 super().__init__(f'{tarinfo.name!r} would be extracted to {path!r}, '740 + 'which is outside the destination')741 742class SpecialFileError(FilterError):743 def __init__(self, tarinfo):744 self.tarinfo = tarinfo745 super().__init__(f'{tarinfo.name!r} is a special file')746 747class AbsoluteLinkError(FilterError):748 def __init__(self, tarinfo):749 self.tarinfo = tarinfo750 super().__init__(f'{tarinfo.name!r} is a link to an absolute path')751 752class LinkOutsideDestinationError(FilterError):753 def __init__(self, tarinfo, path):754 self.tarinfo = tarinfo755 self._path = path756 super().__init__(f'{tarinfo.name!r} would link to {path!r}, '757 + 'which is outside the destination')758 759def _get_filtered_attrs(member, dest_path, for_data=True):760 new_attrs = {}761 name = member.name762 dest_path = os.path.realpath(dest_path)763 # Strip leading / (tar's directory separator) from filenames.764 # Include os.sep (target OS directory separator) as well.765 if name.startswith(('/', os.sep)):766 name = new_attrs['name'] = member.path.lstrip('/' + os.sep)767 if os.path.isabs(name):768 # Path is absolute even after stripping.769 # For example, 'C:/foo' on Windows.770 raise AbsolutePathError(member)771 # Ensure we stay in the destination772 target_path = os.path.realpath(os.path.join(dest_path, name))773 if os.path.commonpath([target_path, dest_path]) != dest_path:774 raise OutsideDestinationError(member, target_path)775 # Limit permissions (no high bits, and go-w)776 mode = member.mode777 if mode is not None:778 # Strip high bits & group/other write bits779 mode = mode & 0o755780 if for_data:781 # For data, handle permissions & file types782 if member.isreg() or member.islnk():783 if not mode & 0o100:784 # Clear executable bits if not executable by user785 mode &= ~0o111786 # Ensure owner can read & write787 mode |= 0o600788 elif member.isdir() or member.issym():789 # Ignore mode for directories & symlinks790 mode = None791 else:792 # Reject special files793 raise SpecialFileError(member)794 if mode != member.mode:795 new_attrs['mode'] = mode796 if for_data:797 # Ignore ownership for 'data'798 if member.uid is not None:799 new_attrs['uid'] = None800 if member.gid is not None:801 new_attrs['gid'] = None802 if member.uname is not None:803 new_attrs['uname'] = None804 if member.gname is not None:805 new_attrs['gname'] = None806 # Check link destination for 'data'807 if member.islnk() or member.issym():808 if os.path.isabs(member.linkname):809 raise AbsoluteLinkError(member)810 if member.issym():811 target_path = os.path.join(dest_path,812 os.path.dirname(name),813 member.linkname)814 else:815 target_path = os.path.join(dest_path,816 member.linkname)817 target_path = os.path.realpath(target_path)818 if os.path.commonpath([target_path, dest_path]) != dest_path:819 raise LinkOutsideDestinationError(member, target_path)820 return new_attrs821 822def fully_trusted_filter(member, dest_path):823 return member824 825def tar_filter(member, dest_path):826 new_attrs = _get_filtered_attrs(member, dest_path, False)827 if new_attrs:828 return member.replace(**new_attrs, deep=False)829 return member830 831def data_filter(member, dest_path):832 new_attrs = _get_filtered_attrs(member, dest_path, True)833 if new_attrs:834 return member.replace(**new_attrs, deep=False)835 return member836 837_NAMED_FILTERS = {838 "fully_trusted": fully_trusted_filter,839 "tar": tar_filter,840 "data": data_filter,841}842 843#------------------844# Exported Classes845#------------------846 847# Sentinel for replace() defaults, meaning "don't change the attribute"848_KEEP = object()849 850class TarInfo(object):851 """Informational class which holds the details about an852 archive member given by a tar header block.853 TarInfo objects are returned by TarFile.getmember(),854 TarFile.getmembers() and TarFile.gettarinfo() and are855 usually created internally.856 """857 858 __slots__ = dict(859 name = 'Name of the archive member.',860 mode = 'Permission bits.',861 uid = 'User ID of the user who originally stored this member.',862 gid = 'Group ID of the user who originally stored this member.',863 size = 'Size in bytes.',864 mtime = 'Time of last modification.',865 chksum = 'Header checksum.',866 type = ('File type. type is usually one of these constants: '867 'REGTYPE, AREGTYPE, LNKTYPE, SYMTYPE, DIRTYPE, FIFOTYPE, '868 'CONTTYPE, CHRTYPE, BLKTYPE, GNUTYPE_SPARSE.'),869 linkname = ('Name of the target file name, which is only present '870 'in TarInfo objects of type LNKTYPE and SYMTYPE.'),871 uname = 'User name.',872 gname = 'Group name.',873 devmajor = 'Device major number.',874 devminor = 'Device minor number.',875 offset = 'The tar header starts here.',876 offset_data = "The file's data starts here.",877 pax_headers = ('A dictionary containing key-value pairs of an '878 'associated pax extended header.'),879 sparse = 'Sparse member information.',880 _tarfile = None,881 _sparse_structs = None,882 _link_target = None,883 )884 885 def __init__(self, name=""):886 """Construct a TarInfo object. name is the optional name887 of the member.888 """889 self.name = name # member name890 self.mode = 0o644 # file permissions891 self.uid = 0 # user id892 self.gid = 0 # group id893 self.size = 0 # file size894 self.mtime = 0 # modification time895 self.chksum = 0 # header checksum896 self.type = REGTYPE # member type897 self.linkname = "" # link name898 self.uname = "" # user name899 self.gname = "" # group name900 self.devmajor = 0 # device major number901 self.devminor = 0 # device minor number902 903 self.offset = 0 # the tar header starts here904 self.offset_data = 0 # the file's data starts here905 906 self.sparse = None # sparse member information907 self.pax_headers = {} # pax header information908 909 @property910 def tarfile(self):911 import warnings912 warnings.warn(913 'The undocumented "tarfile" attribute of TarInfo objects '914 + 'is deprecated and will be removed in Python 3.16',915 DeprecationWarning, stacklevel=2)916 return self._tarfile917 918 @tarfile.setter919 def tarfile(self, tarfile):920 import warnings921 warnings.warn(922 'The undocumented "tarfile" attribute of TarInfo objects '923 + 'is deprecated and will be removed in Python 3.16',924 DeprecationWarning, stacklevel=2)925 self._tarfile = tarfile926 927 @property928 def path(self):929 'In pax headers, "name" is called "path".'930 return self.name931 932 @path.setter933 def path(self, name):934 self.name = name935 936 @property937 def linkpath(self):938 'In pax headers, "linkname" is called "linkpath".'939 return self.linkname940 941 @linkpath.setter942 def linkpath(self, linkname):943 self.linkname = linkname944 945 def __repr__(self):946 return "<%s %r at %#x>" % (self.__class__.__name__,self.name,id(self))947 948 def replace(self, *,949 name=_KEEP, mtime=_KEEP, mode=_KEEP, linkname=_KEEP,950 uid=_KEEP, gid=_KEEP, uname=_KEEP, gname=_KEEP,951 deep=True, _KEEP=_KEEP):952 """Return a deep copy of self with the given attributes replaced.953 """954 if deep:955 result = copy.deepcopy(self)956 else:957 result = copy.copy(self)958 if name is not _KEEP:959 result.name = name960 if mtime is not _KEEP:961 result.mtime = mtime962 if mode is not _KEEP:963 result.mode = mode964 if linkname is not _KEEP:965 result.linkname = linkname966 if uid is not _KEEP:967 result.uid = uid968 if gid is not _KEEP:969 result.gid = gid970 if uname is not _KEEP:971 result.uname = uname972 if gname is not _KEEP:973 result.gname = gname974 return result975 976 def get_info(self):977 """Return the TarInfo's attributes as a dictionary.978 """979 if self.mode is None:980 mode = None981 else:982 mode = self.mode & 0o7777983 info = {984 "name": self.name,985 "mode": mode,986 "uid": self.uid,987 "gid": self.gid,988 "size": self.size,989 "mtime": self.mtime,990 "chksum": self.chksum,991 "type": self.type,992 "linkname": self.linkname,993 "uname": self.uname,994 "gname": self.gname,995 "devmajor": self.devmajor,996 "devminor": self.devminor997 }998 999 if info["type"] == DIRTYPE and not info["name"].endswith("/"):1000 info["name"] += "/"1001 1002 return info1003 1004 def tobuf(self, format=DEFAULT_FORMAT, encoding=ENCODING, errors="surrogateescape"):1005 """Return a tar header as a string of 512 byte blocks.1006 """1007 info = self.get_info()1008 for name, value in info.items():1009 if value is None:1010 raise ValueError("%s may not be None" % name)1011 1012 if format == USTAR_FORMAT:1013 return self.create_ustar_header(info, encoding, errors)1014 elif format == GNU_FORMAT:1015 return self.create_gnu_header(info, encoding, errors)1016 elif format == PAX_FORMAT:1017 return self.create_pax_header(info, encoding)1018 else:1019 raise ValueError("invalid format")1020 1021 def create_ustar_header(self, info, encoding, errors):1022 """Return the object as a ustar header block.1023 """1024 info["magic"] = POSIX_MAGIC1025 1026 if len(info["linkname"].encode(encoding, errors)) > LENGTH_LINK:1027 raise ValueError("linkname is too long")1028 1029 if len(info["name"].encode(encoding, errors)) > LENGTH_NAME:1030 info["prefix"], info["name"] = self._posix_split_name(info["name"], encoding, errors)1031 1032 return self._create_header(info, USTAR_FORMAT, encoding, errors)1033 1034 def create_gnu_header(self, info, encoding, errors):1035 """Return the object as a GNU header block sequence.1036 """1037 info["magic"] = GNU_MAGIC1038 1039 buf = b""1040 if len(info["linkname"].encode(encoding, errors)) > LENGTH_LINK:1041 buf += self._create_gnu_long_header(info["linkname"], GNUTYPE_LONGLINK, encoding, errors)1042 1043 if len(info["name"].encode(encoding, errors)) > LENGTH_NAME:1044 buf += self._create_gnu_long_header(info["name"], GNUTYPE_LONGNAME, encoding, errors)1045 1046 return buf + self._create_header(info, GNU_FORMAT, encoding, errors)1047 1048 def create_pax_header(self, info, encoding):1049 """Return the object as a ustar header block. If it cannot be1050 represented this way, prepend a pax extended header sequence1051 with supplement information.1052 """1053 info["magic"] = POSIX_MAGIC1054 pax_headers = self.pax_headers.copy()1055 1056 # Test string fields for values that exceed the field length or cannot1057 # be represented in ASCII encoding.1058 for name, hname, length in (1059 ("name", "path", LENGTH_NAME), ("linkname", "linkpath", LENGTH_LINK),1060 ("uname", "uname", 32), ("gname", "gname", 32)):1061 1062 if hname in pax_headers:1063 # The pax header has priority.1064 continue1065 1066 # Try to encode the string as ASCII.1067 try:1068 info[name].encode("ascii", "strict")1069 except UnicodeEncodeError:1070 pax_headers[hname] = info[name]1071 continue1072 1073 if len(info[name]) > length:1074 pax_headers[hname] = info[name]1075 1076 # Test number fields for values that exceed the field limit or values1077 # that like to be stored as float.1078 for name, digits in (("uid", 8), ("gid", 8), ("size", 12), ("mtime", 12)):1079 needs_pax = False1080 1081 val = info[name]1082 val_is_float = isinstance(val, float)1083 val_int = round(val) if val_is_float else val1084 if not 0 <= val_int < 8 ** (digits - 1):1085 # Avoid overflow.1086 info[name] = 01087 needs_pax = True1088 elif val_is_float:1089 # Put rounded value in ustar header, and full1090 # precision value in pax header.1091 info[name] = val_int1092 needs_pax = True1093 1094 # The existing pax header has priority.1095 if needs_pax and name not in pax_headers:1096 pax_headers[name] = str(val)1097 1098 # Create a pax extended header if necessary.1099 if pax_headers:1100 buf = self._create_pax_generic_header(pax_headers, XHDTYPE, encoding)1101 else:1102 buf = b""1103 1104 return buf + self._create_header(info, USTAR_FORMAT, "ascii", "replace")1105 1106 @classmethod1107 def create_pax_global_header(cls, pax_headers):1108 """Return the object as a pax global header block sequence.1109 """1110 return cls._create_pax_generic_header(pax_headers, XGLTYPE, "utf-8")1111 1112 def _posix_split_name(self, name, encoding, errors):1113 """Split a name longer than 100 chars into a prefix1114 and a name part.1115 """1116 components = name.split("/")1117 for i in range(1, len(components)):1118 prefix = "/".join(components[:i])1119 name = "/".join(components[i:])1120 if len(prefix.encode(encoding, errors)) <= LENGTH_PREFIX and \1121 len(name.encode(encoding, errors)) <= LENGTH_NAME:1122 break1123 else:1124 raise ValueError("name is too long")1125 1126 return prefix, name1127 1128 @staticmethod1129 def _create_header(info, format, encoding, errors):1130 """Return a header block. info is a dictionary with file1131 information, format must be one of the *_FORMAT constants.1132 """1133 has_device_fields = info.get("type") in (CHRTYPE, BLKTYPE)1134 if has_device_fields:1135 devmajor = itn(info.get("devmajor", 0), 8, format)1136 devminor = itn(info.get("devminor", 0), 8, format)1137 else:1138 devmajor = stn("", 8, encoding, errors)1139 devminor = stn("", 8, encoding, errors)1140 1141 # None values in metadata should cause ValueError.1142 # itn()/stn() do this for all fields except type.1143 filetype = info.get("type", REGTYPE)1144 if filetype is None:1145 raise ValueError("TarInfo.type must not be None")1146 1147 parts = [1148 stn(info.get("name", ""), 100, encoding, errors),1149 itn(info.get("mode", 0) & 0o7777, 8, format),1150 itn(info.get("uid", 0), 8, format),1151 itn(info.get("gid", 0), 8, format),1152 itn(info.get("size", 0), 12, format),1153 itn(info.get("mtime", 0), 12, format),1154 b" ", # checksum field1155 filetype,1156 stn(info.get("linkname", ""), 100, encoding, errors),1157 info.get("magic", POSIX_MAGIC),1158 stn(info.get("uname", ""), 32, encoding, errors),1159 stn(info.get("gname", ""), 32, encoding, errors),1160 devmajor,1161 devminor,1162 stn(info.get("prefix", ""), 155, encoding, errors)1163 ]1164 1165 buf = struct.pack("%ds" % BLOCKSIZE, b"".join(parts))1166 chksum = calc_chksums(buf[-BLOCKSIZE:])[0]1167 buf = buf[:-364] + bytes("%06o\0" % chksum, "ascii") + buf[-357:]1168 return buf1169 1170 @staticmethod1171 def _create_payload(payload):1172 """Return the string payload filled with zero bytes1173 up to the next 512 byte border.1174 """1175 blocks, remainder = divmod(len(payload), BLOCKSIZE)1176 if remainder > 0:1177 payload += (BLOCKSIZE - remainder) * NUL1178 return payload1179 1180 @classmethod1181 def _create_gnu_long_header(cls, name, type, encoding, errors):1182 """Return a GNUTYPE_LONGNAME or GNUTYPE_LONGLINK sequence1183 for name.1184 """1185 name = name.encode(encoding, errors) + NUL1186 1187 info = {}1188 info["name"] = "././@LongLink"1189 info["type"] = type1190 info["size"] = len(name)1191 info["magic"] = GNU_MAGIC1192 1193 # create extended header + name blocks.1194 return cls._create_header(info, USTAR_FORMAT, encoding, errors) + \1195 cls._create_payload(name)1196 1197 @classmethod1198 def _create_pax_generic_header(cls, pax_headers, type, encoding):1199 """Return a POSIX.1-2008 extended or global header sequence1200 that contains a list of keyword, value pairs. The values