bpo-33871: Fix os.sendfile(), os.writev(), os.readv(), etc. (GH-7931)

* Fix integer overflow in os.readv(), os.writev(), os.preadv()
  and os.pwritev() and in os.sendfile() with headers or trailers
  arguments (on BSD-based OSes and MacOS).

* Fix sending the part of the file in os.sendfile() on MacOS.
  Using the trailers argument could cause sending more bytes from
  the input file than was specified.

Thanks Ned Deily for testing on 32-bit MacOS.
This commit is contained in:
Serhiy Storchaka 2018-07-31 10:24:54 +03:00 committed by GitHub
parent f1d36d8efa
commit 9d5727326a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 114 additions and 32 deletions

View file

@ -20,6 +20,9 @@ import warnings
_DUMMY_SYMLINK = os.path.join(tempfile.gettempdir(),
support.TESTFN + '-dummy-symlink')
requires_32b = unittest.skipUnless(sys.maxsize < 2**32,
'test is only meaningful on 32-bit builds')
class PosixTester(unittest.TestCase):
def setUp(self):
@ -284,6 +287,7 @@ class PosixTester(unittest.TestCase):
finally:
os.close(fd)
@unittest.skipUnless(hasattr(posix, 'preadv'), "test needs posix.preadv()")
@unittest.skipUnless(hasattr(posix, 'RWF_HIPRI'), "test needs posix.RWF_HIPRI")
def test_preadv_flags(self):
fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT)
@ -295,6 +299,19 @@ class PosixTester(unittest.TestCase):
finally:
os.close(fd)
@unittest.skipUnless(hasattr(posix, 'preadv'), "test needs posix.preadv()")
@requires_32b
def test_preadv_overflow_32bits(self):
fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT)
try:
buf = [bytearray(2**16)] * 2**15
with self.assertRaises(OSError) as cm:
os.preadv(fd, buf, 0)
self.assertEqual(cm.exception.errno, errno.EINVAL)
self.assertEqual(bytes(buf[0]), b'\0'* 2**16)
finally:
os.close(fd)
@unittest.skipUnless(hasattr(posix, 'pwrite'), "test needs posix.pwrite()")
def test_pwrite(self):
fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT)
@ -320,6 +337,7 @@ class PosixTester(unittest.TestCase):
finally:
os.close(fd)
@unittest.skipUnless(hasattr(posix, 'pwritev'), "test needs posix.pwritev()")
@unittest.skipUnless(hasattr(posix, 'os.RWF_SYNC'), "test needs os.RWF_SYNC")
def test_pwritev_flags(self):
fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT)
@ -334,6 +352,17 @@ class PosixTester(unittest.TestCase):
finally:
os.close(fd)
@unittest.skipUnless(hasattr(posix, 'pwritev'), "test needs posix.pwritev()")
@requires_32b
def test_pwritev_overflow_32bits(self):
fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT)
try:
with self.assertRaises(OSError) as cm:
os.pwritev(fd, [b"x" * 2**16] * 2**15, 0)
self.assertEqual(cm.exception.errno, errno.EINVAL)
finally:
os.close(fd)
@unittest.skipUnless(hasattr(posix, 'posix_fallocate'),
"test needs posix.posix_fallocate()")
def test_posix_fallocate(self):
@ -435,6 +464,17 @@ class PosixTester(unittest.TestCase):
finally:
os.close(fd)
@unittest.skipUnless(hasattr(posix, 'writev'), "test needs posix.writev()")
@requires_32b
def test_writev_overflow_32bits(self):
fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT)
try:
with self.assertRaises(OSError) as cm:
os.writev(fd, [b"x" * 2**16] * 2**15)
self.assertEqual(cm.exception.errno, errno.EINVAL)
finally:
os.close(fd)
@unittest.skipUnless(hasattr(posix, 'readv'), "test needs posix.readv()")
def test_readv(self):
fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT)
@ -457,6 +497,19 @@ class PosixTester(unittest.TestCase):
finally:
os.close(fd)
@unittest.skipUnless(hasattr(posix, 'readv'), "test needs posix.readv()")
@requires_32b
def test_readv_overflow_32bits(self):
fd = os.open(support.TESTFN, os.O_RDWR | os.O_CREAT)
try:
buf = [bytearray(2**16)] * 2**15
with self.assertRaises(OSError) as cm:
os.readv(fd, buf)
self.assertEqual(cm.exception.errno, errno.EINVAL)
self.assertEqual(bytes(buf[0]), b'\0'* 2**16)
finally:
os.close(fd)
@unittest.skipUnless(hasattr(posix, 'dup'),
'test needs posix.dup()')
def test_dup(self):