Merge branch 'master' of https://github.com/GameMaker2k/PyArchiveFile

Author: Kazuki Suzuki Przyborowski

pyarchivefile.py

@@ -125,6 +125,38 @@
 except Exception:
     PATH_TYPES = (basestring,)
 
+def _ensure_text(s, encoding="utf-8", errors="replace", allow_none=False):
+    """
+    Normalize any input to text_type (unicode on Py2, str on Py3).
+
+    - bytes/bytearray/memoryview -> decode
+    - os.PathLike -> fspath then normalize
+    - None -> "" (unless allow_none=True, then return None)
+    - everything else -> text_type(s)
+    """
+    if s is None:
+        return None if allow_none else text_type("")
+
+    if isinstance(s, text_type):
+        return s
+
+    if isinstance(s, (bytes_type, bytearray, memoryview)):
+        return bytes(s).decode(encoding, errors)
+
+    # Handle pathlib.Path & other path-like objects
+    try:
+        import os
+        if hasattr(os, "fspath"):
+            fs = os.fspath(s)
+            if isinstance(fs, text_type):
+                return fs
+            if isinstance(fs, (bytes_type, bytearray, memoryview)):
+                return bytes(fs).decode(encoding, errors)
+    except Exception:
+        pass
+
+    return text_type(s)
+
 def to_text(s, encoding="utf-8", errors="ignore"):
     if s is None:
         return u""
@@ -578,7 +610,7 @@ def _get(section_dict, key, default=None):
 __version_date__ = str(__version_date_info__[0]) + "." + str(
     __version_date_info__[1]).zfill(2) + "." + str(__version_date_info__[2]).zfill(2)
 __revision__ = __version_info__[3]
-__revision_id__ = "$Id$"
+__revision_id__ = "$Id: 2855200b33e29201ad0c5e5fc6da5347b404c16c $"
 if(__version_info__[4] is not None):
     __version_date_plusrc__ = __version_date__ + \
         "-" + str(__version_date_info__[4])
@@ -884,36 +916,64 @@ def VerbosePrintOutReturn(dbgtxt, outtype="log", dbgenable=True, dgblevel=20, **
     return dbgtxt
 
 
+def _split_posix(name):
+    """
+    Return a list of path parts without collapsing '..'.
+    - Normalize backslashes to '/'
+    - Strip leading './' (repeated)
+    - Remove '' and '.' parts; keep '..' for traversal detection
+    """
+    if not name:
+        return []
+    n = name.replace(u"\\", u"/")
+    while n.startswith(u"./"):
+        n = n[2:]
+    return [p for p in n.split(u"/") if p not in (u"", u".")]
+
+def _is_abs_like(name):
+    """Detect absolute-like paths across platforms (/, \\, drive letters, UNC)."""
+    if not name:
+        return False
+    n = name.replace(u"\\", u"/")
+
+    # POSIX absolute
+    if n.startswith(u"/"):
+        return True
 
-def _split_posix(path_text):
-    """Split POSIX paths regardless of OS; return list of components."""
-    # Normalize leading './'
-    if path_text.startswith(u'./'):
-        path_text = path_text[2:]
-    # Strip redundant slashes
-    path_text = re.sub(u'/+', u'/', path_text)
-    # Drop trailing '/' so 'dir/' -> ['dir']
-    if path_text.endswith(u'/'):
-        path_text = path_text[:-1]
-    return path_text.split(u'/') if path_text else []
+    # Windows UNC (\\server\share\...) -> after replace: startswith '//'
+    if n.startswith(u"//"):
+        return True
 
-def _is_abs_like(s):
-    """Absolute targets (POSIX or Windows-drive style)."""
-    return s.startswith(u'/') or s.startswith(u'\\') or re.match(u'^[A-Za-z]:[/\\\\]', s)
+    # Windows drive: 'C:/', 'C:\', or bare 'C:' (treat as absolute-like conservatively)
+    if len(n) >= 2 and n[1] == u":":
+        if len(n) == 2:
+            return True
+        if n[2:3] in (u"/", u"\\"):
+            return True
+    return False
 
-def _resolves_outside(base_rel, target_rel):
+def _resolves_outside(parent, target):
     """
-    Given a base directory (relative, POSIX) and a target (relative),
-    return True if base/target resolves outside of base.
-    We anchor under '/' so normpath is root-anchored and portable.
+    Does a symlink from 'parent' to 'target' escape parent?
+    - Absolute-like target => escape.
+    - Compare normalized '/<parent>/<target>' against '/<parent>'.
+    - 'parent' is POSIX-style ('' means archive root).
     """
-    base_clean = u'/'.join(_split_posix(base_rel))
-    target_clean = u'/'.join(_split_posix(target_rel))
-    base_abs = u'/' + base_clean if base_clean else u'/'
-    combined = pp.normpath(pp.join(base_abs, target_clean))
-    if combined == base_abs or combined.startswith(base_abs + u'/'):
-        return False
-    return True
+    parent = _ensure_text(parent or u"")
+    target = _ensure_text(target or u"")
+
+    # Absolute target is unsafe by definition
+    if _is_abs_like(target):
+        return True
+
+    import posixpath as pp
+    root = u"/"
+    base = pp.normpath(pp.join(root, parent))   # '/dir/sub' or '/'
+    cand = pp.normpath(pp.join(base, target))   # resolved target under '/'
+
+    # ensure trailing slash on base for the prefix test
+    base_slash = base if base.endswith(u"/") else (base + u"/")
+    return not (cand == base or cand.startswith(base_slash))
 
 def _to_bytes(data, encoding="utf-8", errors="strict"):
     """
@@ -1017,9 +1077,6 @@ def _to_text(s, encoding="utf-8", errors="replace", normalize=None, prefer_surro
 
     return out
 
-def ensure_text(s, **kw):
-    return _to_text(s, **kw)
-
 def _quote_path_for_wire(path_text):
     # Percent-encode as UTF-8; return ASCII bytes text
     try:
@@ -1385,7 +1442,7 @@ def _guess_filename(url, filename):
         return filename
     path = urlparse(url).path or ''
     base = os.path.basename(path)
-    return base or 'OutFile.'+__file_format_extension__
+    return base or 'ArchiveFile'+__file_format_extension__
 
 # ---- progress + rate limiting helpers ----
 try:
@@ -1680,79 +1737,6 @@ def _pace_rate(last_ts, sent_bytes_since_ts, rate_limit_bps, add_bytes):
         sent_bytes_since_ts = 0
     return (sleep_s, last_ts, sent_bytes_since_ts)
 
-
-def _split_posix(name):
-    """
-    Return a list of path parts without collapsing '..'.
-    - Normalize backslashes to '/'
-    - Strip leading './' and redundant slashes
-    - Keep '..' parts for traversal detection
-    """
-    if not name:
-        return []
-    n = name.replace(u"\\", u"/")
-    # drop leading ./ repeatedly
-    while n.startswith(u"./"):
-        n = n[2:]
-    # split and filter empty and '.'
-    parts = [p for p in n.split(u"/") if p not in (u"", u".")]
-    return parts
-
-def _is_abs_like(name):
-    """Detect absolute-like paths across platforms (/, \, drive letters)."""
-    if not name:
-        return False
-    n = name.replace(u"\\", u"/")
-    if n.startswith(u"/"):
-        return True
-    # Windows drive: C:/ or C:\  (allow lowercase too)
-    if len(n) >= 3 and n[1] == u":" and n[2] in (u"/", u"\\"):
-        return True
-    return False
-
-def _resolves_outside(parent, target):
-    """
-    Does a symlink from 'parent' to 'target' escape parent?
-    - Treat absolute target as escaping.
-    - For relative target, join parent + target, normpath, then check if it starts with parent.
-    - Parent is POSIX-style path ('' means root of archive).
-    """
-    parent = _ensure_text(parent or u"")
-    target = _ensure_text(target or u"")
-    # absolute target is unsafe by definition
-    if _is_abs_like(target):
-        return True
-
-    # Build a virtual root '/' so we can compare safely
-    # e.g., parent='dir/sub', target='../../etc' -> '/dir/sub/../../etc' -> '/etc' (escapes)
-    import posixpath
-    root = u"/"
-    base = posixpath.normpath(posixpath.join(root, parent))  # '/dir/sub'
-    candidate = posixpath.normpath(posixpath.join(base, target))  # resolved path under '/'
-
-    # Ensure base always ends with a slash for prefix test
-    base_slash = base if base.endswith(u"/") else (base + u"/")
-    # candidate must be base itself or inside base
-    if candidate == base or candidate.startswith(base_slash):
-        return False
-    return True
-
-def _symlink_type(ftype):
-    """
-    Return True if ftype denotes a symlink.
-    Accepts: 2 (int), '2', 'symlink', 'link' (case-insensitive).
-    """
-    if ftype is None:
-        return False
-    # numeric or numeric string
-    try:
-        if int(ftype) == 2:
-            return True
-    except Exception:
-        pass
-    s = ensure_text(ftype).strip().lower()
-    return s in (u"2", u"symlink", u"link", u"symbolic_link", u"symbolic-link")
-
 def DetectTarBombFoxFileArray(listarrayfiles,
                               top_file_ratio_threshold=0.6,
                               min_members_for_ratio=4,
@@ -1768,7 +1752,7 @@ def DetectTarBombFoxFileArray(listarrayfiles,
         has_symlinks (bool)
     """
     if to_text is None:
-        to_text = ensure_text
+        to_text = _ensure_text
 
     files = listarrayfiles or {}
     members = files.get('ffilelist') or []
@@ -8977,6 +8961,7 @@ def UncompressFileAlt(fp, formatspecs=__file_format_multi_dict__, filestart=0,
 
 def UncompressFile(infile, formatspecs=__file_format_multi_dict__, mode="rb",
                    filestart=0, use_mmap=False):
+
     """
     Opens a path, detects compression by header, and returns a FileLikeAdapter.
     If uncompressed and use_mmap=True, returns an mmap-backed reader.
@@ -9721,6 +9706,14 @@ def CompressOpenFile(outfile, compressionenable=True, compressionlevel=None,
     if outfile is None:
         return False
 
+    # If caller already gave us a FileLikeAdapter => honor it and return it.
+    if isinstance(fp, FileLikeAdapter):
+        try:
+            fp.write_through = True
+        except Exception:
+            pass
+        return fp
+
     fbasename, fextname = os.path.splitext(outfile)
     compressionlevel = 9 if compressionlevel is None else int(compressionlevel)
     mode = "w" if PY2 else "wb"
@@ -9893,22 +9886,22 @@ def PackArchiveFileFromSevenZipFile(infile, outfile, fmttype="auto", compression
         return AppendFilesWithContentFromSevenZipToOutFile(infile, outfile, fmttype, compression, compresswholefile, compressionlevel, compressionuselist, extradata, jsondata, checksumtype, formatspecs, verbose, returnfp)
 
 
-def PackArchiveFileFromInFile(infile, outfile, fmttype="auto", compression="auto", compresswholefile=True, compressionlevel=None, compressionuselist=compressionlistalt, checksumtype=["crc32", "crc32", "crc32"], extradata=[], formatspecs=__file_format_dict__, verbose=False, returnfp=False):
-    checkcompressfile = CheckCompressionSubType(infile, formatspecs, filestart, True)
+def PackArchiveFileFromInFile(infile, outfile, fmttype="auto", compression="auto", compresswholefile=True, compressionlevel=None, compressionuselist=compressionlistalt, checksumtype=["crc32", "crc32", "crc32"], extradata=[], jsondata={}, formatspecs=__file_format_dict__, verbose=False, returnfp=False):
+    checkcompressfile = CheckCompressionSubType(infile, formatspecs, 0, True)
     if(IsNestedDict(formatspecs) and checkcompressfile in formatspecs):
         formatspecs = formatspecs[checkcompressfile]
     if(verbose):
         logging.basicConfig(format="%(message)s", stream=sys.stdout, level=logging.DEBUG)
     if(checkcompressfile == "tarfile" and TarFileCheck(infile)):
-        return PackArchiveFileFromTarFile(infile, outfile, fmttype, compression, compresswholefile, compressionlevel, compressionuselist, checksumtype, extradata, formatspecs, verbose, returnfp)
+        return PackArchiveFileFromTarFile(infile, outfile, fmttype, compression, compresswholefile, compressionlevel, compressionuselist, checksumtype, extradata, jsondata, formatspecs, verbose, returnfp)
     elif(checkcompressfile == "zipfile" and zipfile.is_zipfile(infile)):
-        return PackArchiveFileFromZipFile(infile, outfile, fmttype, compression, compresswholefile, compressionlevel, compressionuselist, checksumtype, extradata, formatspecs, verbose, returnfp)
+        return PackArchiveFileFromZipFile(infile, outfile, fmttype, compression, compresswholefile, compressionlevel, compressionuselist, checksumtype, extradata, jsondata, formatspecs, verbose, returnfp)
     elif(rarfile_support and checkcompressfile == "rarfile" and (rarfile.is_rarfile(infile) or rarfile.is_rarfile_sfx(infile))):
-        return PackArchiveFileFromRarFile(infile, outfile, fmttype, compression, compresswholefile, compressionlevel, compressionuselist, checksumtype, extradata, formatspecs, verbose, returnfp)
+        return PackArchiveFileFromRarFile(infile, outfile, fmttype, compression, compresswholefile, compressionlevel, compressionuselist, checksumtype, extradata, jsondata, formatspecs, verbose, returnfp)
     elif(py7zr_support and checkcompressfile == "7zipfile" and py7zr.is_7zfile(infile)):
-        return PackArchiveFileFromSevenZipFile(infile, outfile, fmttype, compression, compresswholefile, compressionlevel, compressionuselist, checksumtype, extradata, formatspecs, verbose, returnfp)
+        return PackArchiveFileFromSevenZipFile(infile, outfile, fmttype, compression, compresswholefile, compressionlevel, compressionuselist, checksumtype, extradata, jsondata, formatspecs, verbose, returnfp)
     elif(IsSingleDict(formatspecs) and checkcompressfile == formatspecs['format_magic']):
-        return RePackArchiveFile(infile, outfile, fmttype, compression, compresswholefile, compressionlevel, False, 0, 0, checksumtype, False, extradata, formatspecs, verbose, returnfp)
+        return RePackArchiveFile(infile, outfile, fmttype, compression, compresswholefile, compressionlevel, False, 0, 0, checksumtype, False, extradata, jsondata, formatspecs, verbose, returnfp)
     else:
         return False
     return False