[Python-checkins] r70582 - in python/branches/py3k: Lib/test/test_fileio.py Modules/_fileio.c
kristjan.jonsson
python-checkins at python.org
Tue Mar 24 16:27:43 CET 2009
Author: kristjan.jonsson
Date: Tue Mar 24 16:27:42 2009
New Revision: 70582
Log:
http://bugs.python.org/issue5544
Guard _fileio.c against other malicious os.close(f.fileno()) attempts.
Add tests to test_fileio.py to verify behaviour.
Modified:
python/branches/py3k/Lib/test/test_fileio.py
python/branches/py3k/Modules/_fileio.c
Modified: python/branches/py3k/Lib/test/test_fileio.py
==============================================================================
--- python/branches/py3k/Lib/test/test_fileio.py (original)
+++ python/branches/py3k/Lib/test/test_fileio.py Tue Mar 24 16:27:42 2009
@@ -6,6 +6,7 @@
import unittest
from array import array
from weakref import proxy
+from functools import wraps
from test.support import (TESTFN, findfile, check_warnings, run_unittest,
make_bad_fd)
@@ -114,20 +115,106 @@
else:
self.fail("Should have raised IOError")
- def testErrnoOnClose(self):
- # Test that the IOError's `errno` attribute is correctly set when
- # close() fails. Here we first close the file descriptor ourselves so
- # that close() fails with EBADF ('Bad file descriptor').
- f = self.f
- os.close(f.fileno())
- self.f = None
+ #A set of functions testing that we get expected behaviour if someone has
+ #manually closed the internal file descriptor. First, a decorator:
+ def ClosedFD(func):
+ @wraps(func)
+ def wrapper(self):
+ #forcibly close the fd before invoking the problem function
+ f = self.f
+ os.close(f.fileno())
+ try:
+ func(self, f)
+ finally:
+ try:
+ self.f.close()
+ except IOError:
+ pass
+ return wrapper
+
+ def ClosedFDRaises(func):
+ @wraps(func)
+ def wrapper(self):
+ #forcibly close the fd before invoking the problem function
+ f = self.f
+ os.close(f.fileno())
+ try:
+ func(self, f)
+ except IOError as e:
+ self.assertEqual(e.errno, errno.EBADF)
+ else:
+ self.fail("Should have raised IOError")
+ finally:
+ try:
+ self.f.close()
+ except IOError:
+ pass
+ return wrapper
+
+ @ClosedFDRaises
+ def testErrnoOnClose(self, f):
+ f.close()
+
+ @ClosedFDRaises
+ def testErrnoOnClosedWrite(self, f):
+ f.write('a')
+
+ @ClosedFDRaises
+ def testErrnoOnClosedSeek(self, f):
+ f.seek(0)
+
+ @ClosedFDRaises
+ def testErrnoOnClosedTell(self, f):
+ f.tell()
+
+ @ClosedFDRaises
+ def testErrnoOnClosedTruncate(self, f):
+ f.truncate(0)
+
+ @ClosedFD
+ def testErrnoOnClosedSeekable(self, f):
+ f.seekable()
+
+ @ClosedFD
+ def testErrnoOnClosedReadable(self, f):
+ f.readable()
+
+ @ClosedFD
+ def testErrnoOnClosedWritable(self, f):
+ f.writable()
+
+ @ClosedFD
+ def testErrnoOnClosedFileno(self, f):
+ f.fileno()
+
+ @ClosedFD
+ def testErrnoOnClosedIsatty(self, f):
+ self.assertEqual(f.isatty(), False)
+
+ def ReopenForRead(self):
try:
- f.close()
- except IOError as e:
- self.assertEqual(e.errno, errno.EBADF)
- else:
- self.fail("Should have raised IOError")
+ self.f.close()
+ except IOError:
+ pass
+ self.f = _FileIO(TESTFN, 'r')
+ os.close(self.f.fileno())
+ return self.f
+ @ClosedFDRaises
+ def testErrnoOnClosedRead(self, f):
+ f = self.ReopenForRead()
+ f.read(1)
+
+ @ClosedFDRaises
+ def testErrnoOnClosedReadall(self, f):
+ f = self.ReopenForRead()
+ f.readall()
+
+ @ClosedFDRaises
+ def testErrnoOnClosedReadinto(self, f):
+ f = self.ReopenForRead()
+ a = array('b', b'x'*10)
+ f.readinto(a)
class OtherFileTests(unittest.TestCase):
Modified: python/branches/py3k/Modules/_fileio.c
==============================================================================
--- python/branches/py3k/Modules/_fileio.c (original)
+++ python/branches/py3k/Modules/_fileio.c Tue Mar 24 16:27:42 2009
@@ -475,10 +475,13 @@
if (!PyArg_ParseTuple(args, "w*", &pbuf))
return NULL;
- Py_BEGIN_ALLOW_THREADS
- errno = 0;
- n = read(self->fd, pbuf.buf, pbuf.len);
- Py_END_ALLOW_THREADS
+ if (_PyVerify_fd(self->fd)) {
+ Py_BEGIN_ALLOW_THREADS
+ errno = 0;
+ n = read(self->fd, pbuf.buf, pbuf.len);
+ Py_END_ALLOW_THREADS
+ } else
+ n = -1;
PyBuffer_Release(&pbuf);
if (n < 0) {
if (errno == EAGAIN)
@@ -522,6 +525,9 @@
Py_ssize_t total = 0;
int n;
+ if (!_PyVerify_fd(self->fd))
+ return PyErr_SetFromErrno(PyExc_IOError);
+
result = PyBytes_FromStringAndSize(NULL, SMALLCHUNK);
if (result == NULL)
return NULL;
@@ -596,10 +602,13 @@
return NULL;
ptr = PyBytes_AS_STRING(bytes);
- Py_BEGIN_ALLOW_THREADS
- errno = 0;
- n = read(self->fd, ptr, size);
- Py_END_ALLOW_THREADS
+ if (_PyVerify_fd(self->fd)) {
+ Py_BEGIN_ALLOW_THREADS
+ errno = 0;
+ n = read(self->fd, ptr, size);
+ Py_END_ALLOW_THREADS
+ } else
+ n = -1;
if (n < 0) {
if (errno == EAGAIN)
@@ -632,10 +641,13 @@
if (!PyArg_ParseTuple(args, "s*", &pbuf))
return NULL;
- Py_BEGIN_ALLOW_THREADS
- errno = 0;
- n = write(self->fd, pbuf.buf, pbuf.len);
- Py_END_ALLOW_THREADS
+ if (_PyVerify_fd(self->fd)) {
+ Py_BEGIN_ALLOW_THREADS
+ errno = 0;
+ n = write(self->fd, pbuf.buf, pbuf.len);
+ Py_END_ALLOW_THREADS
+ } else
+ n = -1;
PyBuffer_Release(&pbuf);
@@ -688,13 +700,16 @@
return NULL;
}
- Py_BEGIN_ALLOW_THREADS
+ if (_PyVerify_fd(fd)) {
+ Py_BEGIN_ALLOW_THREADS
#if defined(MS_WIN64) || defined(MS_WINDOWS)
- res = _lseeki64(fd, pos, whence);
+ res = _lseeki64(fd, pos, whence);
#else
- res = lseek(fd, pos, whence);
+ res = lseek(fd, pos, whence);
#endif
- Py_END_ALLOW_THREADS
+ Py_END_ALLOW_THREADS
+ } else
+ res = -1;
if (res < 0)
return PyErr_SetFromErrno(PyExc_IOError);
@@ -757,13 +772,15 @@
/* Move to the position to be truncated. */
posobj = portable_lseek(fd, posobj, 0);
}
+ if (posobj == NULL)
+ return NULL;
#if defined(HAVE_LARGEFILE_SUPPORT)
pos = PyLong_AsLongLong(posobj);
#else
pos = PyLong_AsLong(posobj);
#endif
- if (PyErr_Occurred())
+ if (pos == -1 && PyErr_Occurred())
return NULL;
#ifdef MS_WINDOWS
More information about the Python-checkins
mailing list