Test older Python versions on older Ubuntu versions because of the breaking changes to OpenSSL.

.github/workflows/ci.yml

@@ -146,16 +146,31 @@ jobs:
     strategy:
       matrix:
         python-version: [2.7, pypy-2.7, pypy-3.6, 3.6, 3.7, 3.8, 3.9]
-        os: [ubuntu-latest, macos-latest]
+        # ubuntu-latest is at least 20.04. But this breaks the SSL
+        # tests because Ubuntu increased the default OpenSSL
+        # strictness.
+        # The standard library has only been updated with fixes for
+        # this in 3.8+.
+        os: [ubuntu-18.04, macos-latest, ubuntu-latest]
         exclude:
           - os: macos-latest
             python-version: pypy-2.7
           - os: macos-latest
             python-version: pypy-3.6
           - os: macos-latest
-            python-version: 3.5
-          - os: macos-latest
             python-version: 3.6
+          - os: ubuntu-latest
+            python-version: 2.7
+          - os: ubuntu-latest
+            python-version: pypy-2.7
+          - os: ubuntu-latest
+            python-version: pypy-3.6
+          - os: ubuntu-latest
+            python-version: 3.6
+          - os: ubuntu-latest
+            python-version: 3.7
+          - os: ubuntu-latest
+            python-version: 3.8
     steps:
       - name: checkout
         uses: actions/checkout@v2
@@ -375,7 +390,7 @@ jobs:
     strategy:
       matrix:
         python-version: [2.7, 3.9]
-        os: [ubuntu-latest]
+        os: [ubuntu-18.04]
     steps:
       - name: checkout
         uses: actions/checkout@v2

src/gevent/testing/patched_tests_setup.py

@@ -1336,6 +1336,16 @@ if PY39:
         'test_subprocess.POSIXProcessTestTest.test_send_signal_race',
     ]
 
+    # These were added for fixes sometime between 3.9.1 and 3.9.5
+    if sys.version_info[:3] < (3, 9, 5):
+        disabled_tests += [
+            'test_ftplib.TestFTPClass.test_makepasv_issue43285_security_disabled',
+            'test_ftplib.TestFTPClass.test_makepasv_issue43285_security_enabled_default',
+            'test_httplib.BasicTest.test_dir_with_added_behavior_on_status',
+            'test_httplib.TunnelTests.test_tunnel_connect_single_send_connection_setup',
+            'test_ssl.TestSSLDebug.test_msg_callback_deadlock_bpo43577',
+        ]
+
 if TRAVIS:
     disabled_tests += [
         # These tests frequently break when we try to use newer Travis CI images,

src/greentest/2.7/lock_tests.py

@@ -295,7 +295,7 @@ class EventTests(BaseTestCase):
         self.assertEqual(results1, [False] * N)
         for r, dt in results2:
             self.assertFalse(r)
-            self.assertTrue(dt >= 0.19, dt) # XXX: gevent: modified for libuv from 0.2 sometimes get 0.199865
+            self.assertTrue(dt >= 0.2, dt)
         # The event is set
         results1 = []
         results2 = []
@@ -440,7 +440,7 @@ class ConditionTests(BaseTestCase):
         Bunch(f, N).wait_for_finished()
         self.assertEqual(len(results), 5)
         for dt in results:
-            self.assertTrue(dt >= 0.19, dt) # XXX: gevent: gevent: modified from 0.2. sometimes get 0.199865
+            self.assertTrue(dt >= 0.2, dt)
 
 
 class BaseSemaphoreTests(BaseTestCase):

src/greentest/2.7/subprocessdata/sigchild_ignore.py

-import signal, subprocess, sys
+import signal, subprocess, sys, time
 # On Linux this causes os.waitpid to fail with OSError as the OS has already
 # reaped our child process.  The wait() passing the OSError on to the caller
 # and causing us to exit with an error is what we are testing against.
 signal.signal(signal.SIGCHLD, signal.SIG_IGN)
 subprocess.Popen([sys.executable, '-c', 'print("albatross")']).wait()
+# Also ensure poll() handles an errno.ECHILD appropriately.
+p = subprocess.Popen([sys.executable, '-c', 'print("albatross")'])
+num_polls = 0
+while p.poll() is None:
+    # Waiting for the process to finish.
+    time.sleep(0.01)  # Avoid being a CPU busy loop.
+    num_polls += 1
+    if num_polls > 3000:
+        raise RuntimeError('poll should have returned 0 within 30 seconds')

src/greentest/2.7/test_httplib.py

@@ -702,6 +702,31 @@ class BasicTest(TestCase):
         with self.assertRaisesRegexp(socket.error, "Invalid response"):
             conn._tunnel()
 
+    def test_putrequest_override_domain_validation(self):
+        """
+        It should be possible to override the default validation
+        behavior in putrequest (bpo-38216).
+        """
+        class UnsafeHTTPConnection(httplib.HTTPConnection):
+            def _validate_path(self, url):
+                pass
+
+        conn = UnsafeHTTPConnection('example.com')
+        conn.sock = FakeSocket('')
+        conn.putrequest('GET', '/\x00')
+
+    def test_putrequest_override_host_validation(self):
+        class UnsafeHTTPConnection(httplib.HTTPConnection):
+            def _validate_host(self, url):
+                pass
+
+        conn = UnsafeHTTPConnection('example.com\r\n')
+        conn.sock = FakeSocket('')
+        # set skip_host so a ValueError is not raised upon adding the
+        # invalid URL as the value of the "Host:" header
+        conn.putrequest('GET', '/', skip_host=1)
+
+
 class OfflineTest(TestCase):
     def test_responses(self):
         self.assertEqual(httplib.responses[httplib.NOT_FOUND], "Not Found")

src/greentest/2.7/test_timeout.py

@@ -106,6 +106,7 @@ class TimeoutTestCase(unittest.TestCase):
     def tearDown(self):
         self.sock.close()
 
+    @unittest.skipIf(True, 'need to replace these hosts; see bpo-35518')
     def testConnectTimeout(self):
         # Choose a private address that is unlikely to exist to prevent
         # failures due to the connect succeeding before the timeout.

src/greentest/2.7/test_urllib.py

@@ -257,6 +257,31 @@ class urlopen_HttpTests(unittest.TestCase, FakeHTTPMixin):
         finally:
             self.unfakehttp()
 
+    def test_url_with_control_char_rejected(self):
+        for char_no in range(0, 0x21) + range(0x7f, 0x100):
+            char = chr(char_no)
+            schemeless_url = "//localhost:7777/test%s/" % char
+            self.fakehttp(b"HTTP/1.1 200 OK\r\n\r\nHello.")
+            try:
+                # urllib quotes the URL so there is no injection.
+                resp = urllib.urlopen("http:" + schemeless_url)
+                self.assertNotIn(char, resp.geturl())
+            finally:
+                self.unfakehttp()
+
+    def test_url_with_newline_header_injection_rejected(self):
+        self.fakehttp(b"HTTP/1.1 200 OK\r\n\r\nHello.")
+        host = "localhost:7777?a=1 HTTP/1.1\r\nX-injected: header\r\nTEST: 123"
+        schemeless_url = "//" + host + ":8080/test/?test=a"
+        try:
+            # urllib quotes the URL so there is no injection.
+            resp = urllib.urlopen("http:" + schemeless_url)
+            self.assertNotIn(' ', resp.geturl())
+            self.assertNotIn('\r', resp.geturl())
+            self.assertNotIn('\n', resp.geturl())
+        finally:
+            self.unfakehttp()
+
     def test_read_bogus(self):
         # urlopen() should raise IOError for many error codes.
         self.fakehttp('''HTTP/1.1 401 Authentication Required
@@ -1023,6 +1048,17 @@ class URLopener_Tests(unittest.TestCase):
             "spam://c:|windows%/:=&?~#+!$,;'@()*[]|/path/"),
             "//c:|windows%/:=&?~#+!$,;'@()*[]|/path/")
 
+    def test_local_file_open(self):
+        # bpo-35907, CVE-2019-9948: urllib must reject local_file:// scheme
+        class DummyURLopener(urllib.URLopener):
+            def open_local_file(self, url):
+                return url
+        for url in ('local_file://example', 'local-file://example'):
+            self.assertRaises(IOError, urllib.urlopen, url)
+            self.assertRaises(IOError, urllib.URLopener().open, url)
+            self.assertRaises(IOError, urllib.URLopener().retrieve, url)
+            self.assertRaises(IOError, DummyURLopener().open, url)
+            self.assertRaises(IOError, DummyURLopener().retrieve, url)
 
 # Just commented them out.
 # Can't really tell why keep failing in windows and sparc.

src/greentest/2.7/test_urllib2.py

@@ -15,6 +15,9 @@ try:
 except ImportError:
     ssl = None
 
+from test.test_urllib import FakeHTTPMixin
+
+
 # XXX
 # Request
 # CacheFTPHandler (hard to write)
@@ -1262,7 +1265,7 @@ class HandlerTests(unittest.TestCase):
         self.assertEqual(len(http_handler.requests), 1)
         self.assertFalse(http_handler.requests[0].has_header(auth_header))
 
-class MiscTests(unittest.TestCase):
+class MiscTests(unittest.TestCase, FakeHTTPMixin):
 
     def test_build_opener(self):
         class MyHTTPHandler(urllib2.HTTPHandler): pass
@@ -1317,6 +1320,70 @@ class MiscTests(unittest.TestCase):
             "Unsupported digest authentication algorithm 'invalid'"
         )
 
+    @unittest.skipUnless(ssl, "ssl module required")
+    def test_url_path_with_control_char_rejected(self):
+        for char_no in range(0, 0x21) + range(0x7f, 0x100):
+            char = chr(char_no)
+            schemeless_url = "//localhost:7777/test%s/" % char
+            self.fakehttp(b"HTTP/1.1 200 OK\r\n\r\nHello.")
+            try:
+                # We explicitly test urllib.request.urlopen() instead of the top
+                # level 'def urlopen()' function defined in this... (quite ugly)
+                # test suite.  They use different url opening codepaths.  Plain
+                # urlopen uses FancyURLOpener which goes via a codepath that
+                # calls urllib.parse.quote() on the URL which makes all of the
+                # above attempts at injection within the url _path_ safe.
+                escaped_char_repr = repr(char).replace('\\', r'\\')
+                InvalidURL = httplib.InvalidURL
+                with self.assertRaisesRegexp(
+                    InvalidURL, "contain control.*" + escaped_char_repr):
+                    urllib2.urlopen("http:" + schemeless_url)
+                with self.assertRaisesRegexp(
+                    InvalidURL, "contain control.*" + escaped_char_repr):
+                    urllib2.urlopen("https:" + schemeless_url)
+            finally:
+                self.unfakehttp()
+
+    @unittest.skipUnless(ssl, "ssl module required")
+    def test_url_path_with_newline_header_injection_rejected(self):
+        self.fakehttp(b"HTTP/1.1 200 OK\r\n\r\nHello.")
+        host = "localhost:7777?a=1 HTTP/1.1\r\nX-injected: header\r\nTEST: 123"
+        schemeless_url = "//" + host + ":8080/test/?test=a"
+        try:
+            # We explicitly test urllib2.urlopen() instead of the top
+            # level 'def urlopen()' function defined in this... (quite ugly)
+            # test suite.  They use different url opening codepaths.  Plain
+            # urlopen uses FancyURLOpener which goes via a codepath that
+            # calls urllib.parse.quote() on the URL which makes all of the
+            # above attempts at injection within the url _path_ safe.
+            InvalidURL = httplib.InvalidURL
+            with self.assertRaisesRegexp(InvalidURL,
+                    r"contain control.*\\r.*(found at least . .)"):
+                urllib2.urlopen("http:{}".format(schemeless_url))
+            with self.assertRaisesRegexp(InvalidURL,
+                    r"contain control.*\\n"):
+                urllib2.urlopen("https:{}".format(schemeless_url))
+        finally:
+            self.unfakehttp()
+
+    @unittest.skipUnless(ssl, "ssl module required")
+    def test_url_host_with_control_char_rejected(self):
+        for char_no in list(range(0, 0x21)) + [0x7f]:
+            char = chr(char_no)
+            schemeless_url = "//localhost{}/test/".format(char)
+            self.fakehttp(b"HTTP/1.1 200 OK\r\n\r\nHello.")
+            try:
+                escaped_char_repr = repr(char).replace('\\', r'\\')
+                InvalidURL = httplib.InvalidURL
+                with self.assertRaisesRegexp(InvalidURL,
+                    "contain control.*{}".format(escaped_char_repr)):
+                        urllib2.urlopen("http:{}".format(schemeless_url))
+                with self.assertRaisesRegexp(InvalidURL,
+                    "contain control.*{}".format(escaped_char_repr)):
+                        urllib2.urlopen("https:{}".format(schemeless_url))
+            finally:
+                self.unfakehttp()
+
 
 class RequestTests(unittest.TestCase):
 
@@ -1410,6 +1477,7 @@ class RequestTests(unittest.TestCase):
             self.fail("err.info() failed")
         self.assertEqual(err.info(), "Content-Length:42")
 
+
 def test_main(verbose=None):
     # gevent: disabled doctests. they hang on OSX on Travis.
     # from test import test_urllib2

src/greentest/2.7/test_urllib2net.py

@@ -85,7 +85,7 @@ class CloseSocketTest(unittest.TestCase):
         # underlying socket
 
         # delve deep into response to fetch socket._socketobject
-        response = _urlopen_with_retry("http://www.example.com/")
+        response = _urlopen_with_retry(test_support.TEST_HTTP_URL)
         abused_fileobject = response.fp
         # self.assertIs(abused_fileobject.__class__, socket._fileobject) # XXX: gevent: disable
         httpresponse = abused_fileobject._sock
@@ -169,7 +169,7 @@ class OtherNetworkTests(unittest.TestCase):
                     "http://www.pythontest.net/index.html#frag")
 
     def test_fileno(self):
-        req = urllib2.Request("http://www.example.com")
+        req = urllib2.Request(test_support.TEST_HTTP_URL)
         opener = urllib2.build_opener()
         res = opener.open(req)
         try:
@@ -180,7 +180,7 @@ class OtherNetworkTests(unittest.TestCase):
             res.close()
 
     def test_custom_headers(self):
-        url = "http://www.example.com"
+        url = test_support.TEST_HTTP_URL
         with test_support.transient_internet(url):
             opener = urllib2.build_opener()
             request = urllib2.Request(url)
@@ -258,14 +258,14 @@ class OtherNetworkTests(unittest.TestCase):
 class TimeoutTest(unittest.TestCase):
     def test_http_basic(self):
         self.assertIsNone(socket.getdefaulttimeout())
-        url = "http://www.example.com"
+        url = test_support.TEST_HTTP_URL
         with test_support.transient_internet(url, timeout=None):
             u = _urlopen_with_retry(url)
             self.assertIsNone(u.fp._sock.fp._sock.gettimeout())
 
     def test_http_default_timeout(self):
         self.assertIsNone(socket.getdefaulttimeout())
-        url = "http://www.example.com"
+        url = test_support.TEST_HTTP_URL
         with test_support.transient_internet(url):
             socket.setdefaulttimeout(60)
             try:
@@ -276,7 +276,7 @@ class TimeoutTest(unittest.TestCase):
 
     def test_http_no_timeout(self):
         self.assertIsNone(socket.getdefaulttimeout())
-        url = "http://www.example.com"
+        url = test_support.TEST_HTTP_URL
         with test_support.transient_internet(url):
             socket.setdefaulttimeout(60)
             try:
@@ -286,7 +286,7 @@ class TimeoutTest(unittest.TestCase):
             self.assertIsNone(u.fp._sock.fp._sock.gettimeout())
 
     def test_http_timeout(self):
-        url = "http://www.example.com"
+        url = test_support.TEST_HTTP_URL
         with test_support.transient_internet(url):
             u = _urlopen_with_retry(url, timeout=120)
             self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 120)

src/greentest/2.7/test_wsgiref.py

@@ -13,7 +13,7 @@ import os
 import re
 import sys
 
-from test import test_support
+from test import support
 
 class MockServer(WSGIServer):
     """Non-socket HTTP server"""
@@ -377,32 +377,62 @@ class TestHandler(ErrorHandler):
 
 
 class HandlerTests(TestCase):
-
-    def checkEnvironAttrs(self, handler):
-        env = handler.environ
-        for attr in [
-            'version','multithread','multiprocess','run_once','file_wrapper'
-        ]:
-            if attr=='file_wrapper' and handler.wsgi_file_wrapper is None:
-                continue
-            self.assertEqual(getattr(handler,'wsgi_'+attr),env['wsgi.'+attr])
-
-    def checkOSEnviron(self,handler):
-        empty = {}; setup_testing_defaults(empty)
-        env = handler.environ
-        from os import environ
-        for k,v in environ.items():
-            if k not in empty:
-                self.assertEqual(env[k],v)
-        for k,v in empty.items():
-            self.assertIn(k, env)
+    # testEnviron() can produce long error message
+    maxDiff = 80 * 50
 
     def testEnviron(self):
-        h = TestHandler(X="Y")
-        h.setup_environ()
-        self.checkEnvironAttrs(h)
-        self.checkOSEnviron(h)
-        self.assertEqual(h.environ["X"],"Y")
+        os_environ = {
+            # very basic environment
+            'HOME': '/my/home',
+            'PATH': '/my/path',
+            'LANG': 'fr_FR.UTF-8',
+
+            # set some WSGI variables
+            'SCRIPT_NAME': 'test_script_name',
+            'SERVER_NAME': 'test_server_name',
+        }
+
+        with support.swap_attr(TestHandler, 'os_environ', os_environ):
+            # override X and HOME variables
+            handler = TestHandler(X="Y", HOME="/override/home")
+            handler.setup_environ()
+
+        # Check that wsgi_xxx attributes are copied to wsgi.xxx variables
+        # of handler.environ
+        for attr in ('version', 'multithread', 'multiprocess', 'run_once',
+                     'file_wrapper'):
+            self.assertEqual(getattr(handler, 'wsgi_' + attr),
+                             handler.environ['wsgi.' + attr])
+
+        # Test handler.environ as a dict
+        expected = {}
+        setup_testing_defaults(expected)
+        # Handler inherits os_environ variables which are not overriden
+        # by SimpleHandler.add_cgi_vars() (SimpleHandler.base_env)
+        for key, value in os_environ.items():
+            if key not in expected:
+                expected[key] = value
+        expected.update({
+            # X doesn't exist in os_environ
+            "X": "Y",
+            # HOME is overriden by TestHandler
+            'HOME': "/override/home",
+
+            # overriden by setup_testing_defaults()
+            "SCRIPT_NAME": "",
+            "SERVER_NAME": "127.0.0.1",
+
+            # set by BaseHandler.setup_environ()
+            'wsgi.input': handler.get_stdin(),
+            'wsgi.errors': handler.get_stderr(),
+            'wsgi.version': (1, 0),
+            'wsgi.run_once': False,
+            'wsgi.url_scheme': 'http',
+            'wsgi.multithread': True,
+            'wsgi.multiprocess': True,
+            'wsgi.file_wrapper': util.FileWrapper,
+        })
+        self.assertDictEqual(handler.environ, expected)
 
     def testCGIEnviron(self):
         h = BaseCGIHandler(None,None,None,{})
@@ -565,7 +595,7 @@ class HandlerTests(TestCase):
 
 
 def test_main():
-    test_support.run_unittest(__name__)
+    support.run_unittest(__name__)
 
 if __name__ == "__main__":
     test_main()

src/greentest/2.7/version

-2.7.16
+2.7.18