Issue #15989: Fix several occurrences of integer overflow

when result of PyInt_AsLong() or PyLong_AsLong() narrowed
to int without checks.

This is a backport of changesets 13e2e44db99d and 525407d89277.

6 files changed, 82 insertions, 4 deletions

diff --git a/Lib/ctypes/test/test_structures.py b/Lib/ctypes/test/test_structures.py
index 1bde101..2d7c816 100644
--- a/Lib/ctypes/test/test_structures.py
+++ b/Lib/ctypes/test/test_structures.py
@@ -1,6 +1,7 @@
 import unittest
 from ctypes import *
 from struct import calcsize
+import _testcapi
 
 class SubclassesTest(unittest.TestCase):
     def test_subclass(self):
@@ -199,6 +200,14 @@ class StructureTestCase(unittest.TestCase):
              "_pack_": -1}
         self.assertRaises(ValueError, type(Structure), "X", (Structure,), d)
 
+        # Issue 15989
+        d = {"_fields_": [("a", c_byte)],
+             "_pack_": _testcapi.INT_MAX + 1}
+        self.assertRaises(ValueError, type(Structure), "X", (Structure,), d)
+        d = {"_fields_": [("a", c_byte)],
+             "_pack_": _testcapi.UINT_MAX + 2}
+        self.assertRaises(ValueError, type(Structure), "X", (Structure,), d)
+
     def test_initializers(self):
         class Person(Structure):
             _fields_ = [("name", c_char*6),
diff --git a/Lib/test/string_tests.py b/Lib/test/string_tests.py
index a161512..f73d0ee 100644
--- a/Lib/test/string_tests.py
+++ b/Lib/test/string_tests.py
@@ -5,6 +5,7 @@ Common tests shared by test_str, test_unicode, test_userstring and test_string.
 import unittest, string, sys, struct
 from test import test_support
 from UserList import UserList
+import _testcapi
 
 class Sequence:
     def __init__(self, seq='wxyz'): self.seq = seq
@@ -1113,6 +1114,20 @@ class MixinStrUnicodeUserStringTest:
         self.checkraises(TypeError, '%10.*f', '__mod__', ('foo', 42.))
         self.checkraises(ValueError, '%10', '__mod__', (42,))
 
+        if _testcapi.PY_SSIZE_T_MAX < sys.maxint:
+            self.checkraises(OverflowError, '%*s', '__mod__',
+                            (_testcapi.PY_SSIZE_T_MAX + 1, ''))
+        if _testcapi.INT_MAX < sys.maxint:
+            self.checkraises(OverflowError, '%.*f', '__mod__',
+                            (_testcapi.INT_MAX + 1, 1. / 7))
+        # Issue 15989
+        if 1 << (_testcapi.PY_SSIZE_T_MAX.bit_length() + 1) <= sys.maxint:
+            self.checkraises(OverflowError, '%*s', '__mod__',
+                            (1 << (_testcapi.PY_SSIZE_T_MAX.bit_length() + 1), ''))
+        if _testcapi.UINT_MAX < sys.maxint:
+            self.checkraises(OverflowError, '%.*f', '__mod__',
+                            (_testcapi.UINT_MAX + 1, 1. / 7))
+
         class X(object): pass
         self.checkraises(TypeError, 'abc', '__mod__', X())
 
diff --git a/Lib/test/test_fcntl.py b/Lib/test/test_fcntl.py
index df09391..7a35f61 100644
--- a/Lib/test/test_fcntl.py
+++ b/Lib/test/test_fcntl.py
@@ -6,6 +6,7 @@ OS/2+EMX doesn't support the file locking operations.
 import os
 import struct
 import sys
+import _testcapi
 import unittest
 from test.test_support import (verbose, TESTFN, unlink, run_unittest,
     import_module)
@@ -81,6 +82,26 @@ class TestFcntl(unittest.TestCase):
             rv = fcntl.fcntl(self.f, fcntl.F_SETLKW, lockdata)
         self.f.close()
 
+    def test_fcntl_bad_file(self):
+        class F:
+            def __init__(self, fn):
+                self.fn = fn
+            def fileno(self):
+                return self.fn
+        self.assertRaises(ValueError, fcntl.fcntl, -1, fcntl.F_SETFL, os.O_NONBLOCK)
+        self.assertRaises(ValueError, fcntl.fcntl, F(-1), fcntl.F_SETFL, os.O_NONBLOCK)
+        self.assertRaises(TypeError, fcntl.fcntl, 'spam', fcntl.F_SETFL, os.O_NONBLOCK)
+        self.assertRaises(TypeError, fcntl.fcntl, F('spam'), fcntl.F_SETFL, os.O_NONBLOCK)
+        # Issue 15989
+        self.assertRaises(ValueError, fcntl.fcntl, _testcapi.INT_MAX + 1,
+                                                   fcntl.F_SETFL, os.O_NONBLOCK)
+        self.assertRaises(ValueError, fcntl.fcntl, F(_testcapi.INT_MAX + 1),
+                                                   fcntl.F_SETFL, os.O_NONBLOCK)
+        self.assertRaises(ValueError, fcntl.fcntl, _testcapi.INT_MIN - 1,
+                                                   fcntl.F_SETFL, os.O_NONBLOCK)
+        self.assertRaises(ValueError, fcntl.fcntl, F(_testcapi.INT_MIN - 1),
+                                                   fcntl.F_SETFL, os.O_NONBLOCK)
+
     def test_fcntl_64_bit(self):
         # Issue #1309352: fcntl shouldn't fail when the third arg fits in a
         # C 'long' but not in a C 'int'.
diff --git a/Lib/test/test_fileio.py b/Lib/test/test_fileio.py
index 71ec44c..fb83255 100644
--- a/Lib/test/test_fileio.py
+++ b/Lib/test/test_fileio.py
@@ -10,6 +10,7 @@ from array import array
 from weakref import proxy
 from functools import wraps
 from UserList import UserList
+import _testcapi
 
 from test.test_support import TESTFN, check_warnings, run_unittest, make_bad_fd
 from test.test_support import py3k_bytes as bytes
@@ -343,6 +344,9 @@ class OtherFileTests(unittest.TestCase):
         if sys.platform == 'win32':
             import msvcrt
             self.assertRaises(IOError, msvcrt.get_osfhandle, make_bad_fd())
+        # Issue 15989
+        self.assertRaises(TypeError, _FileIO, _testcapi.INT_MAX + 1)
+        self.assertRaises(TypeError, _FileIO, _testcapi.INT_MIN - 1)
 
     def testBadModeArgument(self):
         # verify that we get a sensible error message for bad mode argument
diff --git a/Lib/test/test_poll.py b/Lib/test/test_poll.py
index d33af91..55294f8 100644
--- a/Lib/test/test_poll.py
+++ b/Lib/test/test_poll.py
@@ -1,6 +1,7 @@
 # Test case for the os.poll() function
 
 import os, select, random, unittest
+import _testcapi
 from test.test_support import TESTFN, run_unittest
 
 try:
@@ -150,6 +151,15 @@ class PollTests(unittest.TestCase):
         if x != 5:
             self.fail('Overflow must have occurred')
 
+        pollster = select.poll()
+        # Issue 15989
+        self.assertRaises(OverflowError, pollster.register, 0,
+                          _testcapi.SHRT_MAX + 1)
+        self.assertRaises(OverflowError, pollster.register, 0,
+                          _testcapi.USHRT_MAX + 1)
+        self.assertRaises(OverflowError, pollster.poll, _testcapi.INT_MAX + 1)
+        self.assertRaises(OverflowError, pollster.poll, _testcapi.UINT_MAX + 1)
+
 def test_main():
     run_unittest(PollTests)
 
diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
index fec62ef..05d044e 100644
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -6,6 +6,7 @@ from test import test_support
 import errno
 import socket
 import select
+import _testcapi
 import time
 import traceback
 import Queue
@@ -700,11 +701,17 @@ class GeneralModuleTests(unittest.TestCase):
     def test_sendall_interrupted_with_timeout(self):
         self.check_sendall_interrupted(True)
 
-    def testListenBacklog0(self):
+    def test_listen_backlog(self):
+        for backlog in 0, -1:
+            srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            srv.bind((HOST, 0))
+            srv.listen(backlog)
+            srv.close()
+
+        # Issue 15989
         srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         srv.bind((HOST, 0))
-        # backlog = 0
-        srv.listen(0)
+        self.assertRaises(OverflowError, srv.listen, _testcapi.INT_MAX + 1)
         srv.close()
 
     @unittest.skipUnless(SUPPORTS_IPV6, 'IPv6 required for this test.')
@@ -808,6 +815,11 @@ class BasicTCPTest(SocketConnectedTest):
 
     def _testShutdown(self):
         self.serv_conn.send(MSG)
+        # Issue 15989
+        self.assertRaises(OverflowError, self.serv_conn.shutdown,
+                          _testcapi.INT_MAX + 1)
+        self.assertRaises(OverflowError, self.serv_conn.shutdown,
+                          2 + (_testcapi.UINT_MAX + 1))
         self.serv_conn.shutdown(2)
 
 @unittest.skipUnless(thread, 'Threading required for this test.')
@@ -883,7 +895,10 @@ class NonBlockingTCPTests(ThreadedTCPSocketTest):
 
     def testSetBlocking(self):
         # Testing whether set blocking works
-        self.serv.setblocking(0)
+        self.serv.setblocking(True)
+        self.assertIsNone(self.serv.gettimeout())
+        self.serv.setblocking(False)
+        self.assertEqual(self.serv.gettimeout(), 0.0)
         start = time.time()
         try:
             self.serv.accept()
@@ -891,6 +906,10 @@ class NonBlockingTCPTests(ThreadedTCPSocketTest):
             pass
         end = time.time()
         self.assertTrue((end - start) < 1.0, "Error setting non-blocking mode.")
+        # Issue 15989
+        if _testcapi.UINT_MAX < _testcapi.ULONG_MAX:
+            self.serv.setblocking(_testcapi.UINT_MAX + 1)
+            self.assertIsNone(self.serv.gettimeout())
 
     def _testSetBlocking(self):
         pass