gh-160446: replace instead of writing through in tarfile/zipfile

Lib/_archive.py

@@ -0,0 +1,33 @@
+"""Internal helpers shared by tarfile and zipfile."""
+
+import os
+import sys
+
+
+if sys.platform in ("win32", "emscripten"):
+    # These follow a dangling symlink even with O_EXCL, so remove what is
+    # there first.
+    def create_or_replace(path):
+        """Open path for writing as a new file, replacing what is there.
+
+        An existing symlink or hard link is replaced, instead of written
+        through to the file it refers to.
+        """
+        try:
+            os.unlink(path)
+        except FileNotFoundError:
+            pass
+        return open(path, "xb")
+else:
+    # O_EXCL refuses any existing file, including a dangling symlink.
+    def create_or_replace(path):
+        """Open path for writing as a new file, replacing what is there.
+
+        An existing symlink or hard link is replaced, instead of written
+        through to the file it refers to.
+        """
+        try:
+            return open(path, "xb")
+        except FileExistsError:
+            os.unlink(path)
+            return open(path, "xb")

Lib/tarfile.py

@@ -44,6 +44,7 @@
 import struct
 import copy
 import re
+import _archive
 
 try:
     import pwd
@@ -2766,7 +2767,7 @@ def makefile(self, tarinfo, targetpath):
         source = self.fileobj
         source.seek(tarinfo.offset_data)
         bufsize = self.copybufsize
-        with bltn_open(targetpath, "wb") as target:
+        with _archive.create_or_replace(targetpath) as target:
             if tarinfo.sparse is not None:
                 for offset, size in tarinfo.sparse:
                     target.seek(offset)

Lib/test/archiver_tests.py

@@ -53,8 +53,9 @@ def test_overwrite_dir_as_file(self):
         target = os.path.join(self.testdir, 'test')
         os.mkdir(target)
         with self.open(self.ar_with_file) as ar:
-            with self.assertRaises(PermissionError if sys.platform == 'win32'
-                                   else IsADirectoryError):
+            # Removing the directory fails with EISDIR on Linux and EPERM on
+            # other platforms.
+            with self.assertRaises((IsADirectoryError, PermissionError)):
                 self.extractall(ar)
         self.assertTrue(os.path.isdir(target))
 
@@ -81,30 +82,42 @@ def test_overwrite_file_as_implicit_dir(self):
 
     @os_helper.skip_unless_symlink
     def test_overwrite_file_symlink_as_file(self):
-        # XXX: It is potential security vulnerability.
         target = os.path.join(self.testdir, 'test')
         target2 = os.path.join(self.testdir, 'test2')
         self.create_file(target2, b'content')
         os.symlink('test2', target)
         with self.open(self.ar_with_file) as ar:
             self.extractall(ar)
-        self.assertTrue(os.path.islink(target))
-        self.assertTrue(os.path.isfile(target2))
-        with open(target2, 'rb') as f:
+        self.assertFalse(os.path.islink(target))
+        with open(target, 'rb') as f:
             self.assertEqual(f.read(), b'newcontent')
+        with open(target2, 'rb') as f:
+            self.assertEqual(f.read(), b'content')
 
     @os_helper.skip_unless_symlink
     def test_overwrite_broken_file_symlink_as_file(self):
-        # XXX: It is potential security vulnerability.
         target = os.path.join(self.testdir, 'test')
         target2 = os.path.join(self.testdir, 'test2')
         os.symlink('test2', target)
         with self.open(self.ar_with_file) as ar:
             self.extractall(ar)
-        self.assertTrue(os.path.islink(target))
-        self.assertTrue(os.path.isfile(target2))
-        with open(target2, 'rb') as f:
+        self.assertFalse(os.path.islink(target))
+        with open(target, 'rb') as f:
             self.assertEqual(f.read(), b'newcontent')
+        self.assertFalse(os.path.lexists(target2))
+
+    @os_helper.skip_unless_hardlink
+    def test_overwrite_file_hardlink_as_file(self):
+        target = os.path.join(self.testdir, 'test')
+        target2 = os.path.join(self.testdir, 'test2')
+        self.create_file(target2, b'content')
+        os.link(target2, target)
+        with self.open(self.ar_with_file) as ar:
+            self.extractall(ar)
+        with open(target, 'rb') as f:
+            self.assertEqual(f.read(), b'newcontent')
+        with open(target2, 'rb') as f:
+            self.assertEqual(f.read(), b'content')
 
     @os_helper.skip_unless_symlink
     def test_overwrite_dir_symlink_as_dir(self):

Lib/zipfile/__init__.py

@@ -3,6 +3,7 @@
 
 XXX references to utf-8 need further investigation.
 """
+import _archive
 import binascii
 import io
 import os
@@ -2518,7 +2519,7 @@ def _extract_member(self, member, targetpath, pwd):
             return targetpath
 
         with self.open(member, pwd=pwd) as source, \
-             open(targetpath, "wb") as target:
+             _archive.create_or_replace(targetpath) as target:
             shutil.copyfileobj(source, target)
 
         return targetpath

Misc/NEWS.d/next/Library/2026-10-22-21-00-00.gh-issue-160446.tFlnk1.rst

@@ -0,0 +1,4 @@
+:mod:`tarfile` and :mod:`zipfile` now replace an existing file when
+extracting a regular file of the same name, instead of writing to it.
+Previously, an existing symlink or hardlink at that path was written through,
+which could modify a file outside the extraction directory.