[Jython-checkins] jython: Updates httplib, urllib, urllib2 to latest. Fixes #2481
jim.baker
jython-checkins at python.org
Tue Sep 6 01:14:01 EDT 2016
https://hg.python.org/jython/rev/91083509a11c
changeset: 7962:91083509a11c
user: Jim Baker <jim.baker at rackspace.com>
date: Mon Sep 05 23:13:56 2016 -0600
summary:
Updates httplib, urllib, urllib2 to latest. Fixes #2481
In 2.7.1, Jython supports CPython 2.7's recent updates for always
verifying SSL certs. However, httplib, urllib, urllib2 have support
for selectively turning this off, so we needed to refresh this from
CPython and repatch.
As part of this update there's also a fix for the HTTPoxy vulnerability
(https://bugs.python.org/issue27568), although this would only have
been possuble if Jython were used as a CGI script (very unlikely),
instead of being used in a web container (ModJy, etc).
files:
CPythonLib.includes | 1 +
Lib/test/test_httplib.py | 890 ++++++++
Lib/test/test_urllib.py | 301 ++-
Lib/urllib.py | 1630 ---------------
lib-python/2.7/rfc822.py | 9 +-
lib-python/2.7/test/test_httplib.py | 381 +++-
lib-python/2.7/test/test_rfc822.py | 6 +
lib-python/2.7/test/test_urllib.py | 296 ++-
lib-python/2.7/test/test_urllib2.py | 59 +-
lib-python/2.7/urllib.py | 203 +-
lib-python/2.7/urllib2.py | 79 +-
11 files changed, 1974 insertions(+), 1881 deletions(-)
diff --git a/CPythonLib.includes b/CPythonLib.includes
--- a/CPythonLib.includes
+++ b/CPythonLib.includes
@@ -167,6 +167,7 @@
traceback.py
tty.py
tzparse.py
+urllib.py
urllib2.py
urlparse.py
user.py
diff --git a/Lib/test/test_httplib.py b/Lib/test/test_httplib.py
new file mode 100644
--- /dev/null
+++ b/Lib/test/test_httplib.py
@@ -0,0 +1,890 @@
+import httplib
+import itertools
+import array
+import StringIO
+import socket
+import errno
+import os
+import tempfile
+
+import unittest
+TestCase = unittest.TestCase
+
+from test import test_support
+
+here = os.path.dirname(__file__)
+# Self-signed cert file for 'localhost'
+CERT_localhost = os.path.join(here, 'keycert.pem')
+# Self-signed cert file for 'fakehostname'
+CERT_fakehostname = os.path.join(here, 'keycert2.pem')
+# Self-signed cert file for self-signed.pythontest.net
+CERT_selfsigned_pythontestdotnet = os.path.join(here, 'selfsigned_pythontestdotnet.pem')
+
+HOST = test_support.HOST
+
+class FakeSocket:
+ def __init__(self, text, fileclass=StringIO.StringIO, host=None, port=None):
+ self.text = text
+ self.fileclass = fileclass
+ self.data = ''
+ self.file_closed = False
+ self.host = host
+ self.port = port
+
+ def sendall(self, data):
+ self.data += ''.join(data)
+
+ def makefile(self, mode, bufsize=None):
+ if mode != 'r' and mode != 'rb':
+ raise httplib.UnimplementedFileMode()
+ # keep the file around so we can check how much was read from it
+ self.file = self.fileclass(self.text)
+ self.file.close = self.file_close #nerf close ()
+ return self.file
+
+ def file_close(self):
+ self.file_closed = True
+
+ def close(self):
+ pass
+
+class EPipeSocket(FakeSocket):
+
+ def __init__(self, text, pipe_trigger):
+ # When sendall() is called with pipe_trigger, raise EPIPE.
+ FakeSocket.__init__(self, text)
+ self.pipe_trigger = pipe_trigger
+
+ def sendall(self, data):
+ if self.pipe_trigger in data:
+ raise socket.error(errno.EPIPE, "gotcha")
+ self.data += data
+
+ def close(self):
+ pass
+
+class NoEOFStringIO(StringIO.StringIO):
+ """Like StringIO, but raises AssertionError on EOF.
+
+ This is used below to test that httplib doesn't try to read
+ more from the underlying file than it should.
+ """
+ def read(self, n=-1):
+ data = StringIO.StringIO.read(self, n)
+ if data == '':
+ raise AssertionError('caller tried to read past EOF')
+ return data
+
+ def readline(self, length=None):
+ data = StringIO.StringIO.readline(self, length)
+ if data == '':
+ raise AssertionError('caller tried to read past EOF')
+ return data
+
+
+class HeaderTests(TestCase):
+ def test_auto_headers(self):
+ # Some headers are added automatically, but should not be added by
+ # .request() if they are explicitly set.
+
+ class HeaderCountingBuffer(list):
+ def __init__(self):
+ self.count = {}
+ def append(self, item):
+ kv = item.split(':')
+ if len(kv) > 1:
+ # item is a 'Key: Value' header string
+ lcKey = kv[0].lower()
+ self.count.setdefault(lcKey, 0)
+ self.count[lcKey] += 1
+ list.append(self, item)
+
+ for explicit_header in True, False:
+ for header in 'Content-length', 'Host', 'Accept-encoding':
+ conn = httplib.HTTPConnection('example.com')
+ conn.sock = FakeSocket('blahblahblah')
+ conn._buffer = HeaderCountingBuffer()
+
+ body = 'spamspamspam'
+ headers = {}
+ if explicit_header:
+ headers[header] = str(len(body))
+ conn.request('POST', '/', body, headers)
+ self.assertEqual(conn._buffer.count[header.lower()], 1)
+
+ def test_content_length_0(self):
+
+ class ContentLengthChecker(list):
+ def __init__(self):
+ list.__init__(self)
+ self.content_length = None
+ def append(self, item):
+ kv = item.split(':', 1)
+ if len(kv) > 1 and kv[0].lower() == 'content-length':
+ self.content_length = kv[1].strip()
+ list.append(self, item)
+
+ # Here, we're testing that methods expecting a body get a
+ # content-length set to zero if the body is empty (either None or '')
+ bodies = (None, '')
+ methods_with_body = ('PUT', 'POST', 'PATCH')
+ for method, body in itertools.product(methods_with_body, bodies):
+ conn = httplib.HTTPConnection('example.com')
+ conn.sock = FakeSocket(None)
+ conn._buffer = ContentLengthChecker()
+ conn.request(method, '/', body)
+ self.assertEqual(
+ conn._buffer.content_length, '0',
+ 'Header Content-Length incorrect on {}'.format(method)
+ )
+
+ # For these methods, we make sure that content-length is not set when
+ # the body is None because it might cause unexpected behaviour on the
+ # server.
+ methods_without_body = (
+ 'GET', 'CONNECT', 'DELETE', 'HEAD', 'OPTIONS', 'TRACE',
+ )
+ for method in methods_without_body:
+ conn = httplib.HTTPConnection('example.com')
+ conn.sock = FakeSocket(None)
+ conn._buffer = ContentLengthChecker()
+ conn.request(method, '/', None)
+ self.assertEqual(
+ conn._buffer.content_length, None,
+ 'Header Content-Length set for empty body on {}'.format(method)
+ )
+
+ # If the body is set to '', that's considered to be "present but
+ # empty" rather than "missing", so content length would be set, even
+ # for methods that don't expect a body.
+ for method in methods_without_body:
+ conn = httplib.HTTPConnection('example.com')
+ conn.sock = FakeSocket(None)
+ conn._buffer = ContentLengthChecker()
+ conn.request(method, '/', '')
+ self.assertEqual(
+ conn._buffer.content_length, '0',
+ 'Header Content-Length incorrect on {}'.format(method)
+ )
+
+ # If the body is set, make sure Content-Length is set.
+ for method in itertools.chain(methods_without_body, methods_with_body):
+ conn = httplib.HTTPConnection('example.com')
+ conn.sock = FakeSocket(None)
+ conn._buffer = ContentLengthChecker()
+ conn.request(method, '/', ' ')
+ self.assertEqual(
+ conn._buffer.content_length, '1',
+ 'Header Content-Length incorrect on {}'.format(method)
+ )
+
+ def test_putheader(self):
+ conn = httplib.HTTPConnection('example.com')
+ conn.sock = FakeSocket(None)
+ conn.putrequest('GET','/')
+ conn.putheader('Content-length',42)
+ self.assertIn('Content-length: 42', conn._buffer)
+
+ conn.putheader('Foo', ' bar ')
+ self.assertIn(b'Foo: bar ', conn._buffer)
+ conn.putheader('Bar', '\tbaz\t')
+ self.assertIn(b'Bar: \tbaz\t', conn._buffer)
+ conn.putheader('Authorization', 'Bearer mytoken')
+ self.assertIn(b'Authorization: Bearer mytoken', conn._buffer)
+ conn.putheader('IterHeader', 'IterA', 'IterB')
+ self.assertIn(b'IterHeader: IterA\r\n\tIterB', conn._buffer)
+ conn.putheader('LatinHeader', b'\xFF')
+ self.assertIn(b'LatinHeader: \xFF', conn._buffer)
+ conn.putheader('Utf8Header', b'\xc3\x80')
+ self.assertIn(b'Utf8Header: \xc3\x80', conn._buffer)
+ conn.putheader('C1-Control', b'next\x85line')
+ self.assertIn(b'C1-Control: next\x85line', conn._buffer)
+ conn.putheader('Embedded-Fold-Space', 'is\r\n allowed')
+ self.assertIn(b'Embedded-Fold-Space: is\r\n allowed', conn._buffer)
+ conn.putheader('Embedded-Fold-Tab', 'is\r\n\tallowed')
+ self.assertIn(b'Embedded-Fold-Tab: is\r\n\tallowed', conn._buffer)
+ conn.putheader('Key Space', 'value')
+ self.assertIn(b'Key Space: value', conn._buffer)
+ conn.putheader('KeySpace ', 'value')
+ self.assertIn(b'KeySpace : value', conn._buffer)
+ conn.putheader(b'Nonbreak\xa0Space', 'value')
+ self.assertIn(b'Nonbreak\xa0Space: value', conn._buffer)
+ conn.putheader(b'\xa0NonbreakSpace', 'value')
+ self.assertIn(b'\xa0NonbreakSpace: value', conn._buffer)
+
+ def test_ipv6host_header(self):
+ # Default host header on IPv6 transaction should be wrapped by [] if
+ # it is an IPv6 address
+ expected = 'GET /foo HTTP/1.1\r\nHost: [2001::]:81\r\n' \
+ 'Accept-Encoding: identity\r\n\r\n'
+ conn = httplib.HTTPConnection('[2001::]:81')
+ sock = FakeSocket('')
+ conn.sock = sock
+ conn.request('GET', '/foo')
+ self.assertTrue(sock.data.startswith(expected))
+
+ expected = 'GET /foo HTTP/1.1\r\nHost: [2001:102A::]\r\n' \
+ 'Accept-Encoding: identity\r\n\r\n'
+ conn = httplib.HTTPConnection('[2001:102A::]')
+ sock = FakeSocket('')
+ conn.sock = sock
+ conn.request('GET', '/foo')
+ self.assertTrue(sock.data.startswith(expected))
+
+ def test_malformed_headers_coped_with(self):
+ # Issue 19996
+ body = "HTTP/1.1 200 OK\r\nFirst: val\r\n: nval\r\nSecond: val\r\n\r\n"
+ sock = FakeSocket(body)
+ resp = httplib.HTTPResponse(sock)
+ resp.begin()
+
+ self.assertEqual(resp.getheader('First'), 'val')
+ self.assertEqual(resp.getheader('Second'), 'val')
+
+ def test_invalid_headers(self):
+ conn = httplib.HTTPConnection('example.com')
+ conn.sock = FakeSocket('')
+ conn.putrequest('GET', '/')
+
+ # http://tools.ietf.org/html/rfc7230#section-3.2.4, whitespace is no
+ # longer allowed in header names
+ cases = (
+ (b'Invalid\r\nName', b'ValidValue'),
+ (b'Invalid\rName', b'ValidValue'),
+ (b'Invalid\nName', b'ValidValue'),
+ (b'\r\nInvalidName', b'ValidValue'),
+ (b'\rInvalidName', b'ValidValue'),
+ (b'\nInvalidName', b'ValidValue'),
+ (b' InvalidName', b'ValidValue'),
+ (b'\tInvalidName', b'ValidValue'),
+ (b'Invalid:Name', b'ValidValue'),
+ (b':InvalidName', b'ValidValue'),
+ (b'ValidName', b'Invalid\r\nValue'),
+ (b'ValidName', b'Invalid\rValue'),
+ (b'ValidName', b'Invalid\nValue'),
+ (b'ValidName', b'InvalidValue\r\n'),
+ (b'ValidName', b'InvalidValue\r'),
+ (b'ValidName', b'InvalidValue\n'),
+ )
+ for name, value in cases:
+ with self.assertRaisesRegexp(ValueError, 'Invalid header'):
+ conn.putheader(name, value)
+
+
+class BasicTest(TestCase):
+ def test_status_lines(self):
+ # Test HTTP status lines
+
+ body = "HTTP/1.1 200 Ok\r\n\r\nText"
+ sock = FakeSocket(body)
+ resp = httplib.HTTPResponse(sock)
+ resp.begin()
+ self.assertEqual(resp.read(0), '') # Issue #20007
+ self.assertFalse(resp.isclosed())
+ self.assertEqual(resp.read(), 'Text')
+ self.assertTrue(resp.isclosed())
+
+ body = "HTTP/1.1 400.100 Not Ok\r\n\r\nText"
+ sock = FakeSocket(body)
+ resp = httplib.HTTPResponse(sock)
+ self.assertRaises(httplib.BadStatusLine, resp.begin)
+
+ def test_bad_status_repr(self):
+ exc = httplib.BadStatusLine('')
+ self.assertEqual(repr(exc), '''BadStatusLine("\'\'",)''')
+
+ def test_partial_reads(self):
+ # if we have a length, the system knows when to close itself
+ # same behaviour than when we read the whole thing with read()
+ body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText"
+ sock = FakeSocket(body)
+ resp = httplib.HTTPResponse(sock)
+ resp.begin()
+ self.assertEqual(resp.read(2), 'Te')
+ self.assertFalse(resp.isclosed())
+ self.assertEqual(resp.read(2), 'xt')
+ self.assertTrue(resp.isclosed())
+
+ def test_partial_reads_no_content_length(self):
+ # when no length is present, the socket should be gracefully closed when
+ # all data was read
+ body = "HTTP/1.1 200 Ok\r\n\r\nText"
+ sock = FakeSocket(body)
+ resp = httplib.HTTPResponse(sock)
+ resp.begin()
+ self.assertEqual(resp.read(2), 'Te')
+ self.assertFalse(resp.isclosed())
+ self.assertEqual(resp.read(2), 'xt')
+ self.assertEqual(resp.read(1), '')
+ self.assertTrue(resp.isclosed())
+
+ def test_partial_reads_incomplete_body(self):
+ # if the server shuts down the connection before the whole
+ # content-length is delivered, the socket is gracefully closed
+ body = "HTTP/1.1 200 Ok\r\nContent-Length: 10\r\n\r\nText"
+ sock = FakeSocket(body)
+ resp = httplib.HTTPResponse(sock)
+ resp.begin()
+ self.assertEqual(resp.read(2), 'Te')
+ self.assertFalse(resp.isclosed())
+ self.assertEqual(resp.read(2), 'xt')
+ self.assertEqual(resp.read(1), '')
+ self.assertTrue(resp.isclosed())
+
+ def test_host_port(self):
+ # Check invalid host_port
+
+ # Note that httplib does not accept user:password@ in the host-port.
+ for hp in ("www.python.org:abc", "user:password at www.python.org"):
+ self.assertRaises(httplib.InvalidURL, httplib.HTTP, hp)
+
+ for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000", "fe80::207:e9ff:fe9b",
+ 8000),
+ ("www.python.org:80", "www.python.org", 80),
+ ("www.python.org", "www.python.org", 80),
+ ("www.python.org:", "www.python.org", 80),
+ ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 80)):
+ http = httplib.HTTP(hp)
+ c = http._conn
+ if h != c.host:
+ self.fail("Host incorrectly parsed: %s != %s" % (h, c.host))
+ if p != c.port:
+ self.fail("Port incorrectly parsed: %s != %s" % (p, c.host))
+
+ def test_response_headers(self):
+ # test response with multiple message headers with the same field name.
+ text = ('HTTP/1.1 200 OK\r\n'
+ 'Set-Cookie: Customer="WILE_E_COYOTE";'
+ ' Version="1"; Path="/acme"\r\n'
+ 'Set-Cookie: Part_Number="Rocket_Launcher_0001"; Version="1";'
+ ' Path="/acme"\r\n'
+ '\r\n'
+ 'No body\r\n')
+ hdr = ('Customer="WILE_E_COYOTE"; Version="1"; Path="/acme"'
+ ', '
+ 'Part_Number="Rocket_Launcher_0001"; Version="1"; Path="/acme"')
+ s = FakeSocket(text)
+ r = httplib.HTTPResponse(s)
+ r.begin()
+ cookies = r.getheader("Set-Cookie")
+ if cookies != hdr:
+ self.fail("multiple headers not combined properly")
+
+ def test_read_head(self):
+ # Test that the library doesn't attempt to read any data
+ # from a HEAD request. (Tickles SF bug #622042.)
+ sock = FakeSocket(
+ 'HTTP/1.1 200 OK\r\n'
+ 'Content-Length: 14432\r\n'
+ '\r\n',
+ NoEOFStringIO)
+ resp = httplib.HTTPResponse(sock, method="HEAD")
+ resp.begin()
+ if resp.read() != "":
+ self.fail("Did not expect response from HEAD request")
+
+ def test_too_many_headers(self):
+ headers = '\r\n'.join('Header%d: foo' % i for i in xrange(200)) + '\r\n'
+ text = ('HTTP/1.1 200 OK\r\n' + headers)
+ s = FakeSocket(text)
+ r = httplib.HTTPResponse(s)
+ self.assertRaises(httplib.HTTPException, r.begin)
+
+ def test_send_file(self):
+ expected = 'GET /foo HTTP/1.1\r\nHost: example.com\r\n' \
+ 'Accept-Encoding: identity\r\nContent-Length:'
+
+ body = open(__file__, 'rb')
+ conn = httplib.HTTPConnection('example.com')
+ sock = FakeSocket(body)
+ conn.sock = sock
+ conn.request('GET', '/foo', body)
+ self.assertTrue(sock.data.startswith(expected))
+ self.assertIn('def test_send_file', sock.data)
+
+ def test_send_tempfile(self):
+ expected = ('GET /foo HTTP/1.1\r\nHost: example.com\r\n'
+ 'Accept-Encoding: identity\r\nContent-Length: 9\r\n\r\n'
+ 'fake\ndata')
+
+ with tempfile.TemporaryFile() as body:
+ body.write('fake\ndata')
+ body.seek(0)
+
+ conn = httplib.HTTPConnection('example.com')
+ sock = FakeSocket(body)
+ conn.sock = sock
+ conn.request('GET', '/foo', body)
+ self.assertEqual(sock.data, expected)
+
+ def test_send(self):
+ expected = 'this is a test this is only a test'
+ conn = httplib.HTTPConnection('example.com')
+ sock = FakeSocket(None)
+ conn.sock = sock
+ conn.send(expected)
+ self.assertEqual(expected, sock.data)
+ sock.data = ''
+ conn.send(array.array('c', expected))
+ self.assertEqual(expected, sock.data)
+ sock.data = ''
+ conn.send(StringIO.StringIO(expected))
+ self.assertEqual(expected, sock.data)
+
+ def test_chunked(self):
+ chunked_start = (
+ 'HTTP/1.1 200 OK\r\n'
+ 'Transfer-Encoding: chunked\r\n\r\n'
+ 'a\r\n'
+ 'hello worl\r\n'
+ '1\r\n'
+ 'd\r\n'
+ )
+ sock = FakeSocket(chunked_start + '0\r\n')
+ resp = httplib.HTTPResponse(sock, method="GET")
+ resp.begin()
+ self.assertEqual(resp.read(), 'hello world')
+ resp.close()
+
+ for x in ('', 'foo\r\n'):
+ sock = FakeSocket(chunked_start + x)
+ resp = httplib.HTTPResponse(sock, method="GET")
+ resp.begin()
+ try:
+ resp.read()
+ except httplib.IncompleteRead, i:
+ self.assertEqual(i.partial, 'hello world')
+ self.assertEqual(repr(i),'IncompleteRead(11 bytes read)')
+ self.assertEqual(str(i),'IncompleteRead(11 bytes read)')
+ else:
+ self.fail('IncompleteRead expected')
+ finally:
+ resp.close()
+
+ def test_chunked_head(self):
+ chunked_start = (
+ 'HTTP/1.1 200 OK\r\n'
+ 'Transfer-Encoding: chunked\r\n\r\n'
+ 'a\r\n'
+ 'hello world\r\n'
+ '1\r\n'
+ 'd\r\n'
+ )
+ sock = FakeSocket(chunked_start + '0\r\n')
+ resp = httplib.HTTPResponse(sock, method="HEAD")
+ resp.begin()
+ self.assertEqual(resp.read(), '')
+ self.assertEqual(resp.status, 200)
+ self.assertEqual(resp.reason, 'OK')
+ self.assertTrue(resp.isclosed())
+
+ def test_negative_content_length(self):
+ sock = FakeSocket('HTTP/1.1 200 OK\r\n'
+ 'Content-Length: -1\r\n\r\nHello\r\n')
+ resp = httplib.HTTPResponse(sock, method="GET")
+ resp.begin()
+ self.assertEqual(resp.read(), 'Hello\r\n')
+ self.assertTrue(resp.isclosed())
+
+ def test_incomplete_read(self):
+ sock = FakeSocket('HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nHello\r\n')
+ resp = httplib.HTTPResponse(sock, method="GET")
+ resp.begin()
+ try:
+ resp.read()
+ except httplib.IncompleteRead as i:
+ self.assertEqual(i.partial, 'Hello\r\n')
+ self.assertEqual(repr(i),
+ "IncompleteRead(7 bytes read, 3 more expected)")
+ self.assertEqual(str(i),
+ "IncompleteRead(7 bytes read, 3 more expected)")
+ self.assertTrue(resp.isclosed())
+ else:
+ self.fail('IncompleteRead expected')
+
+ def test_epipe(self):
+ sock = EPipeSocket(
+ "HTTP/1.0 401 Authorization Required\r\n"
+ "Content-type: text/html\r\n"
+ "WWW-Authenticate: Basic realm=\"example\"\r\n",
+ b"Content-Length")
+ conn = httplib.HTTPConnection("example.com")
+ conn.sock = sock
+ self.assertRaises(socket.error,
+ lambda: conn.request("PUT", "/url", "body"))
+ resp = conn.getresponse()
+ self.assertEqual(401, resp.status)
+ self.assertEqual("Basic realm=\"example\"",
+ resp.getheader("www-authenticate"))
+
+ def test_filenoattr(self):
+ # Just test the fileno attribute in the HTTPResponse Object.
+ body = "HTTP/1.1 200 Ok\r\n\r\nText"
+ sock = FakeSocket(body)
+ resp = httplib.HTTPResponse(sock)
+ self.assertTrue(hasattr(resp,'fileno'),
+ 'HTTPResponse should expose a fileno attribute')
+
+ # Test lines overflowing the max line size (_MAXLINE in http.client)
+
+ def test_overflowing_status_line(self):
+ self.skipTest("disabled for HTTP 0.9 support")
+ body = "HTTP/1.1 200 Ok" + "k" * 65536 + "\r\n"
+ resp = httplib.HTTPResponse(FakeSocket(body))
+ self.assertRaises((httplib.LineTooLong, httplib.BadStatusLine), resp.begin)
+
+ def test_overflowing_header_line(self):
+ body = (
+ 'HTTP/1.1 200 OK\r\n'
+ 'X-Foo: bar' + 'r' * 65536 + '\r\n\r\n'
+ )
+ resp = httplib.HTTPResponse(FakeSocket(body))
+ self.assertRaises(httplib.LineTooLong, resp.begin)
+
+ def test_overflowing_chunked_line(self):
+ body = (
+ 'HTTP/1.1 200 OK\r\n'
+ 'Transfer-Encoding: chunked\r\n\r\n'
+ + '0' * 65536 + 'a\r\n'
+ 'hello world\r\n'
+ '0\r\n'
+ )
+ resp = httplib.HTTPResponse(FakeSocket(body))
+ resp.begin()
+ self.assertRaises(httplib.LineTooLong, resp.read)
+
+ def test_early_eof(self):
+ # Test httpresponse with no \r\n termination,
+ body = "HTTP/1.1 200 Ok"
+ sock = FakeSocket(body)
+ resp = httplib.HTTPResponse(sock)
+ resp.begin()
+ self.assertEqual(resp.read(), '')
+ self.assertTrue(resp.isclosed())
+
+ def test_error_leak(self):
+ # Test that the socket is not leaked if getresponse() fails
+ conn = httplib.HTTPConnection('example.com')
+ response = []
+ class Response(httplib.HTTPResponse):
+ def __init__(self, *pos, **kw):
+ response.append(self) # Avoid garbage collector closing the socket
+ httplib.HTTPResponse.__init__(self, *pos, **kw)
+ conn.response_class = Response
+ conn.sock = FakeSocket('') # Emulate server dropping connection
+ conn.request('GET', '/')
+ self.assertRaises(httplib.BadStatusLine, conn.getresponse)
+ self.assertTrue(response)
+ #self.assertTrue(response[0].closed)
+ self.assertTrue(conn.sock.file_closed)
+
+ def test_proxy_tunnel_without_status_line(self):
+ # Issue 17849: If a proxy tunnel is created that does not return
+ # a status code, fail.
+ body = 'hello world'
+ conn = httplib.HTTPConnection('example.com', strict=False)
+ conn.set_tunnel('foo')
+ conn.sock = FakeSocket(body)
+ with self.assertRaisesRegexp(socket.error, "Invalid response"):
+ conn._tunnel()
+
+class OfflineTest(TestCase):
+ def test_responses(self):
+ self.assertEqual(httplib.responses[httplib.NOT_FOUND], "Not Found")
+
+
+class TestServerMixin:
+ """A limited socket server mixin.
+
+ This is used by test cases for testing http connection end points.
+ """
+ def setUp(self):
+ self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.port = test_support.bind_port(self.serv)
+ self.source_port = test_support.find_unused_port()
+ self.serv.listen(5)
+ self.conn = None
+
+ def tearDown(self):
+ if self.conn:
+ self.conn.close()
+ self.conn = None
+ self.serv.close()
+ self.serv = None
+
+class SourceAddressTest(TestServerMixin, TestCase):
+ def testHTTPConnectionSourceAddress(self):
+ self.conn = httplib.HTTPConnection(HOST, self.port,
+ source_address=('', self.source_port))
+ self.conn.connect()
+ self.assertEqual(self.conn.sock.getsockname()[1], self.source_port)
+
+ @unittest.skipIf(not hasattr(httplib, 'HTTPSConnection'),
+ 'httplib.HTTPSConnection not defined')
+ def testHTTPSConnectionSourceAddress(self):
+ self.conn = httplib.HTTPSConnection(HOST, self.port,
+ source_address=('', self.source_port))
+ # We don't test anything here other the constructor not barfing as
+ # this code doesn't deal with setting up an active running SSL server
+ # for an ssl_wrapped connect() to actually return from.
+
+
+class HTTPTest(TestServerMixin, TestCase):
+ def testHTTPConnection(self):
+ self.conn = httplib.HTTP(host=HOST, port=self.port, strict=None)
+ self.conn.connect()
+ self.assertEqual(self.conn._conn.host, HOST)
+ self.assertEqual(self.conn._conn.port, self.port)
+
+ def testHTTPWithConnectHostPort(self):
+ testhost = 'unreachable.test.domain'
+ testport = '80'
+ self.conn = httplib.HTTP(host=testhost, port=testport)
+ self.conn.connect(host=HOST, port=self.port)
+ self.assertNotEqual(self.conn._conn.host, testhost)
+ self.assertNotEqual(self.conn._conn.port, testport)
+ self.assertEqual(self.conn._conn.host, HOST)
+ self.assertEqual(self.conn._conn.port, self.port)
+
+
+class TimeoutTest(TestCase):
+ PORT = None
+
+ def setUp(self):
+ self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ TimeoutTest.PORT = test_support.bind_port(self.serv)
+ self.serv.listen(5)
+
+ def tearDown(self):
+ self.serv.close()
+ self.serv = None
+
+ def testTimeoutAttribute(self):
+ '''This will prove that the timeout gets through
+ HTTPConnection and into the socket.
+ '''
+ # default -- use global socket timeout
+ self.assertIsNone(socket.getdefaulttimeout())
+ socket.setdefaulttimeout(30)
+ try:
+ httpConn = httplib.HTTPConnection(HOST, TimeoutTest.PORT)
+ httpConn.connect()
+ finally:
+ socket.setdefaulttimeout(None)
+ self.assertEqual(httpConn.sock.gettimeout(), 30)
+ httpConn.close()
+
+ # no timeout -- do not use global socket default
+ self.assertIsNone(socket.getdefaulttimeout())
+ socket.setdefaulttimeout(30)
+ try:
+ httpConn = httplib.HTTPConnection(HOST, TimeoutTest.PORT,
+ timeout=None)
+ httpConn.connect()
+ finally:
+ socket.setdefaulttimeout(None)
+ self.assertEqual(httpConn.sock.gettimeout(), None)
+ httpConn.close()
+
+ # a value
+ httpConn = httplib.HTTPConnection(HOST, TimeoutTest.PORT, timeout=30)
+ httpConn.connect()
+ self.assertEqual(httpConn.sock.gettimeout(), 30)
+ httpConn.close()
+
+
+class HTTPSTest(TestCase):
+
+ def setUp(self):
+ if not hasattr(httplib, 'HTTPSConnection'):
+ self.skipTest('ssl support required')
+
+ def make_server(self, certfile):
+ from test.ssl_servers import make_https_server
+ return make_https_server(self, certfile=certfile)
+
+ def test_attributes(self):
+ # simple test to check it's storing the timeout
+ h = httplib.HTTPSConnection(HOST, TimeoutTest.PORT, timeout=30)
+ self.assertEqual(h.timeout, 30)
+
+ def test_networked(self):
+ # Default settings: requires a valid cert from a trusted CA
+ import ssl
+ test_support.requires('network')
+ with test_support.transient_internet('self-signed.pythontest.net'):
+ h = httplib.HTTPSConnection('self-signed.pythontest.net', 443)
+ with self.assertRaises(ssl.SSLError) as exc_info:
+ h.request('GET', '/')
+ if test_support.is_jython:
+ return # FIXME: SSLError.reason not yet available for Jython
+ self.assertEqual(exc_info.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
+
+ def test_networked_noverification(self):
+ # Switch off cert verification
+ import ssl
+ test_support.requires('network')
+ with test_support.transient_internet('self-signed.pythontest.net'):
+ context = ssl._create_stdlib_context()
+ h = httplib.HTTPSConnection('self-signed.pythontest.net', 443,
+ context=context)
+ h.request('GET', '/')
+ resp = h.getresponse()
+ self.assertIn('nginx', resp.getheader('server'))
+
+ @test_support.system_must_validate_cert
+ def test_networked_trusted_by_default_cert(self):
+ # Default settings: requires a valid cert from a trusted CA
+ test_support.requires('network')
+ with test_support.transient_internet('www.python.org'):
+ h = httplib.HTTPSConnection('www.python.org', 443)
+ h.request('GET', '/')
+ resp = h.getresponse()
+ content_type = resp.getheader('content-type')
+ self.assertIn('text/html', content_type)
+
+ def test_networked_good_cert(self):
+ # We feed the server's cert as a validating cert
+ import ssl
+ test_support.requires('network')
+ with test_support.transient_internet('self-signed.pythontest.net'):
+ context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+ context.verify_mode = ssl.CERT_REQUIRED
+ context.load_verify_locations(CERT_selfsigned_pythontestdotnet)
+ h = httplib.HTTPSConnection('self-signed.pythontest.net', 443, context=context)
+ h.request('GET', '/')
+ resp = h.getresponse()
+ server_string = resp.getheader('server')
+ self.assertIn('nginx', server_string)
+
+ def test_networked_bad_cert(self):
+ # We feed a "CA" cert that is unrelated to the server's cert
+ import ssl
+ test_support.requires('network')
+ with test_support.transient_internet('self-signed.pythontest.net'):
+ context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+ context.verify_mode = ssl.CERT_REQUIRED
+ context.load_verify_locations(CERT_localhost)
+ h = httplib.HTTPSConnection('self-signed.pythontest.net', 443, context=context)
+ with self.assertRaises(ssl.SSLError) as exc_info:
+ h.request('GET', '/')
+ if test_support.is_jython:
+ return # FIXME: SSLError.reason not yet available for Jython
+ self.assertEqual(exc_info.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
+
+ def test_local_unknown_cert(self):
+ # The custom cert isn't known to the default trust bundle
+ import ssl
+ server = self.make_server(CERT_localhost)
+ h = httplib.HTTPSConnection('localhost', server.port)
+ with self.assertRaises(ssl.SSLError) as exc_info:
+ h.request('GET', '/')
+ if test_support.is_jython:
+ return # FIXME: SSLError.reason not yet available for Jython
+ self.assertEqual(exc_info.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
+
+ @unittest.skipIf(test_support.is_jython,
+ 'FIXME: Failing test on Jython (causes other exceptions if not skipped)')
+ def test_local_good_hostname(self):
+ # The (valid) cert validates the HTTP hostname
+ import ssl
+ server = self.make_server(CERT_localhost)
+ context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+ context.verify_mode = ssl.CERT_REQUIRED
+ # FIXME this appears to be a problem in Jython certifying
+ # localhost certs
+ context.load_verify_locations(CERT_localhost)
+ h = httplib.HTTPSConnection('localhost', server.port, context=context)
+ h.request('GET', '/nonexistent')
+ resp = h.getresponse()
+ self.assertEqual(resp.status, 404)
+
+ @unittest.skipIf(test_support.is_jython,
+ 'FIXME: Jython does not raise proper SSL error')
+ def test_local_bad_hostname(self):
+ # The (valid) cert doesn't validate the HTTP hostname
+ import ssl
+ server = self.make_server(CERT_fakehostname)
+ context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+ context.verify_mode = ssl.CERT_REQUIRED
+ context.check_hostname = True
+ context.load_verify_locations(CERT_fakehostname)
+ h = httplib.HTTPSConnection('localhost', server.port, context=context)
+ with self.assertRaises(ssl.CertificateError):
+ h.request('GET', '/')
+ h.close()
+ # With context.check_hostname=False, the mismatching is ignored
+ context.check_hostname = False
+ h = httplib.HTTPSConnection('localhost', server.port, context=context)
+ h.request('GET', '/nonexistent')
+ resp = h.getresponse()
+ self.assertEqual(resp.status, 404)
+
+ def test_host_port(self):
+ # Check invalid host_port
+
+ for hp in ("www.python.org:abc", "user:password at www.python.org"):
+ self.assertRaises(httplib.InvalidURL, httplib.HTTPSConnection, hp)
+
+ for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000",
+ "fe80::207:e9ff:fe9b", 8000),
+ ("www.python.org:443", "www.python.org", 443),
+ ("www.python.org:", "www.python.org", 443),
+ ("www.python.org", "www.python.org", 443),
+ ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 443),
+ ("[fe80::207:e9ff:fe9b]:", "fe80::207:e9ff:fe9b",
+ 443)):
+ c = httplib.HTTPSConnection(hp)
+ self.assertEqual(h, c.host)
+ self.assertEqual(p, c.port)
+
+
+class TunnelTests(TestCase):
+ def test_connect(self):
+ response_text = (
+ 'HTTP/1.0 200 OK\r\n\r\n' # Reply to CONNECT
+ 'HTTP/1.1 200 OK\r\n' # Reply to HEAD
+ 'Content-Length: 42\r\n\r\n'
+ )
+
+ def create_connection(address, timeout=None, source_address=None):
+ return FakeSocket(response_text, host=address[0], port=address[1])
+
+ conn = httplib.HTTPConnection('proxy.com')
+ conn._create_connection = create_connection
+
+ # Once connected, we should not be able to tunnel anymore
+ conn.connect()
+ self.assertRaises(RuntimeError, conn.set_tunnel, 'destination.com')
+
+ # But if close the connection, we are good.
+ conn.close()
+ conn.set_tunnel('destination.com')
+ conn.request('HEAD', '/', '')
+
+ self.assertEqual(conn.sock.host, 'proxy.com')
+ self.assertEqual(conn.sock.port, 80)
+ self.assertIn('CONNECT destination.com', conn.sock.data)
+ # issue22095
+ self.assertNotIn('Host: destination.com:None', conn.sock.data)
+ self.assertIn('Host: destination.com', conn.sock.data)
+
+ self.assertNotIn('Host: proxy.com', conn.sock.data)
+
+ conn.close()
+
+ conn.request('PUT', '/', '')
+ self.assertEqual(conn.sock.host, 'proxy.com')
+ self.assertEqual(conn.sock.port, 80)
+ self.assertTrue('CONNECT destination.com' in conn.sock.data)
+ self.assertTrue('Host: destination.com' in conn.sock.data)
+
+
+ at test_support.reap_threads
+def test_main(verbose=None):
+ test_support.run_unittest(HeaderTests, OfflineTest, BasicTest, TimeoutTest,
+ HTTPTest, HTTPSTest, SourceAddressTest,
+ TunnelTests)
+
+if __name__ == '__main__':
+ test_main()
diff --git a/Lib/test/test_urllib.py b/Lib/test/test_urllib.py
--- a/Lib/test/test_urllib.py
+++ b/Lib/test/test_urllib.py
@@ -1,13 +1,14 @@
"""Regresssion tests for urllib"""
+import collections
import urllib
import httplib
+import io
import unittest
import os
import sys
import mimetools
import tempfile
-import StringIO
from test import test_support
from base64 import b64encode
@@ -21,37 +22,43 @@
return "%" + hex_repr
+def fakehttp(fakedata):
+ class FakeSocket(io.BytesIO):
+
+ def sendall(self, data):
+ FakeHTTPConnection.buf = data
+
+ def makefile(self, *args, **kwds):
+ return self
+
+ def read(self, amt=None):
+ if self.closed:
+ return b""
+ return io.BytesIO.read(self, amt)
+
+ def readline(self, length=None):
+ if self.closed:
+ return b""
+ return io.BytesIO.readline(self, length)
+
+ class FakeHTTPConnection(httplib.HTTPConnection):
+
+ # buffer to store data for verification in urlopen tests.
+ buf = ""
+
+ def connect(self):
+ self.sock = FakeSocket(self.fakedata)
+ self.__class__.fakesock = self.sock
+ FakeHTTPConnection.fakedata = fakedata
+
+ return FakeHTTPConnection
+
+
class FakeHTTPMixin(object):
def fakehttp(self, fakedata):
- class FakeSocket(StringIO.StringIO):
-
- def sendall(self, data):
- FakeHTTPConnection.buf = data
-
- def makefile(self, *args, **kwds):
- return self
-
- def read(self, amt=None):
- if self.closed:
- return ""
- return StringIO.StringIO.read(self, amt)
-
- def readline(self, length=None):
- if self.closed:
- return ""
- return StringIO.StringIO.readline(self, length)
-
- class FakeHTTPConnection(httplib.HTTPConnection):
-
- # buffer to store data for verification in urlopen tests.
- buf = ""
-
- def connect(self):
- self.sock = FakeSocket(fakedata)
-
assert httplib.HTTP._connection_class == httplib.HTTPConnection
- httplib.HTTP._connection_class = FakeHTTPConnection
+ httplib.HTTP._connection_class = fakehttp(fakedata)
def unfakehttp(self):
httplib.HTTP._connection_class = httplib.HTTPConnection
@@ -107,9 +114,8 @@
def test_fileno(self):
file_num = self.returned_obj.fileno()
- if not test_support.is_jython:
- self.assert_(isinstance(file_num, int),
- "fileno() did not return an int")
+ if not test_support.is_jython: # does not apply to jython - fileno is an obj
+ self.assertIsInstance(file_num, int, "fileno() did not return an int")
self.assertEqual(os.read(file_num, len(self.text)), self.text,
"Reading on the file descriptor returned by fileno() "
"did not return the expected text")
@@ -160,8 +166,71 @@
# getproxies_environment use lowered case truncated (no '_proxy') keys
self.assertEqual('localhost', proxies['no'])
# List of no_proxies with space.
- self.env.set('NO_PROXY', 'localhost, anotherdomain.com, newdomain.com')
+ self.env.set('NO_PROXY', 'localhost, anotherdomain.com, newdomain.com:1234')
self.assertTrue(urllib.proxy_bypass_environment('anotherdomain.com'))
+ self.assertTrue(urllib.proxy_bypass_environment('anotherdomain.com:8888'))
+ self.assertTrue(urllib.proxy_bypass_environment('newdomain.com:1234'))
+
+ def test_proxy_cgi_ignore(self):
+ try:
+ self.env.set('HTTP_PROXY', 'http://somewhere:3128')
+ proxies = urllib.getproxies_environment()
+ self.assertEqual('http://somewhere:3128', proxies['http'])
+ self.env.set('REQUEST_METHOD', 'GET')
+ proxies = urllib.getproxies_environment()
+ self.assertNotIn('http', proxies)
+ finally:
+ self.env.unset('REQUEST_METHOD')
+ self.env.unset('HTTP_PROXY')
+
+ def test_proxy_bypass_environment_host_match(self):
+ bypass = urllib.proxy_bypass_environment
+ self.env.set('NO_PROXY',
+ 'localhost, anotherdomain.com, newdomain.com:1234')
+ self.assertTrue(bypass('localhost'))
+ self.assertTrue(bypass('LocalHost')) # MixedCase
+ self.assertTrue(bypass('LOCALHOST')) # UPPERCASE
+ self.assertTrue(bypass('newdomain.com:1234'))
+ self.assertTrue(bypass('anotherdomain.com:8888'))
+ self.assertTrue(bypass('www.newdomain.com:1234'))
+ self.assertFalse(bypass('prelocalhost'))
+ self.assertFalse(bypass('newdomain.com')) # no port
+ self.assertFalse(bypass('newdomain.com:1235')) # wrong port
+
+class ProxyTests_withOrderedEnv(unittest.TestCase):
+
+ def setUp(self):
+ # We need to test conditions, where variable order _is_ significant
+ self._saved_env = os.environ
+ # Monkey patch os.environ, start with empty fake environment
+ os.environ = collections.OrderedDict()
+
+ def tearDown(self):
+ os.environ = self._saved_env
+
+ def test_getproxies_environment_prefer_lowercase(self):
+ # Test lowercase preference with removal
+ os.environ['no_proxy'] = ''
+ os.environ['No_Proxy'] = 'localhost'
+ self.assertFalse(urllib.proxy_bypass_environment('localhost'))
+ self.assertFalse(urllib.proxy_bypass_environment('arbitrary'))
+ os.environ['http_proxy'] = ''
+ os.environ['HTTP_PROXY'] = 'http://somewhere:3128'
+ proxies = urllib.getproxies_environment()
+ self.assertEqual({}, proxies)
+ # Test lowercase preference of proxy bypass and correct matching including ports
+ os.environ['no_proxy'] = 'localhost, noproxy.com, my.proxy:1234'
+ os.environ['No_Proxy'] = 'xyz.com'
+ self.assertTrue(urllib.proxy_bypass_environment('localhost'))
+ self.assertTrue(urllib.proxy_bypass_environment('noproxy.com:5678'))
+ self.assertTrue(urllib.proxy_bypass_environment('my.proxy:1234'))
+ self.assertFalse(urllib.proxy_bypass_environment('my.proxy'))
+ self.assertFalse(urllib.proxy_bypass_environment('arbitrary'))
+ # Test lowercase preference with replacement
+ os.environ['http_proxy'] = 'http://somewhere:3128'
+ os.environ['Http_Proxy'] = 'http://somewhereelse:3128'
+ proxies = urllib.getproxies_environment()
+ self.assertEqual('http://somewhere:3128', proxies['http'])
class urlopen_HttpTests(unittest.TestCase, FakeHTTPMixin):
@@ -211,10 +280,26 @@
Content-Type: text/html; charset=iso-8859-1
""")
try:
- self.assertRaises(IOError, urllib.urlopen, "http://python.org/")
+ msg = "Redirection to url 'file:"
+ with self.assertRaisesRegexp(IOError, msg):
+ urllib.urlopen("http://python.org/")
finally:
self.unfakehttp()
+ def test_redirect_limit_independent(self):
+ # Ticket #12923: make sure independent requests each use their
+ # own retry limit.
+ for i in range(urllib.FancyURLopener().maxtries):
+ self.fakehttp(b'''HTTP/1.1 302 Found
+Location: file://guidocomputer.athome.com:/python/license
+Connection: close
+''')
+ try:
+ self.assertRaises(IOError, urllib.urlopen,
+ "http://something")
+ finally:
+ self.unfakehttp()
+
def test_empty_socket(self):
# urlopen() raises IOError if the underlying socket does not send any
# data. (#1680230)
@@ -229,13 +314,13 @@
'file://localhost/a/missing/file.py')
fd, tmp_file = tempfile.mkstemp()
tmp_fileurl = 'file://localhost/' + tmp_file.replace(os.path.sep, '/')
+ self.assertTrue(os.path.exists(tmp_file))
try:
- self.assertTrue(os.path.exists(tmp_file))
fp = urllib.urlopen(tmp_fileurl)
+ fp.close()
finally:
os.close(fd)
- fp.close()
- os.unlink(tmp_file)
+ os.unlink(tmp_file)
self.assertFalse(os.path.exists(tmp_file))
self.assertRaises(IOError, urllib.urlopen, tmp_fileurl)
@@ -775,21 +860,131 @@
class Utility_Tests(unittest.TestCase):
"""Testcase to test the various utility functions in the urllib."""
+ # In Python 3 this test class is moved to test_urlparse.
+
+ def test_splittype(self):
+ splittype = urllib.splittype
+ self.assertEqual(splittype('type:opaquestring'), ('type', 'opaquestring'))
+ self.assertEqual(splittype('opaquestring'), (None, 'opaquestring'))
+ self.assertEqual(splittype(':opaquestring'), (None, ':opaquestring'))
+ self.assertEqual(splittype('type:'), ('type', ''))
+ self.assertEqual(splittype('type:opaque:string'), ('type', 'opaque:string'))
+
+ def test_splithost(self):
+ splithost = urllib.splithost
+ self.assertEqual(splithost('//www.example.org:80/foo/bar/baz.html'),
+ ('www.example.org:80', '/foo/bar/baz.html'))
+ self.assertEqual(splithost('//www.example.org:80'),
+ ('www.example.org:80', ''))
+ self.assertEqual(splithost('/foo/bar/baz.html'),
+ (None, '/foo/bar/baz.html'))
+
+ def test_splituser(self):
+ splituser = urllib.splituser
+ self.assertEqual(splituser('User:Pass at www.python.org:080'),
+ ('User:Pass', 'www.python.org:080'))
+ self.assertEqual(splituser('@www.python.org:080'),
+ ('', 'www.python.org:080'))
+ self.assertEqual(splituser('www.python.org:080'),
+ (None, 'www.python.org:080'))
+ self.assertEqual(splituser('User:Pass@'),
+ ('User:Pass', ''))
+ self.assertEqual(splituser('User at example.com:Pass at www.python.org:080'),
+ ('User at example.com:Pass', 'www.python.org:080'))
def test_splitpasswd(self):
- """Some of the password examples are not sensible, but it is added to
- confirming to RFC2617 and addressing issue4675.
- """
- self.assertEqual(('user', 'ab'),urllib.splitpasswd('user:ab'))
- self.assertEqual(('user', 'a\nb'),urllib.splitpasswd('user:a\nb'))
- self.assertEqual(('user', 'a\tb'),urllib.splitpasswd('user:a\tb'))
- self.assertEqual(('user', 'a\rb'),urllib.splitpasswd('user:a\rb'))
- self.assertEqual(('user', 'a\fb'),urllib.splitpasswd('user:a\fb'))
- self.assertEqual(('user', 'a\vb'),urllib.splitpasswd('user:a\vb'))
- self.assertEqual(('user', 'a:b'),urllib.splitpasswd('user:a:b'))
- self.assertEqual(('user', 'a b'),urllib.splitpasswd('user:a b'))
- self.assertEqual(('user 2', 'ab'),urllib.splitpasswd('user 2:ab'))
- self.assertEqual(('user+1', 'a+b'),urllib.splitpasswd('user+1:a+b'))
+ # Some of the password examples are not sensible, but it is added to
+ # confirming to RFC2617 and addressing issue4675.
+ splitpasswd = urllib.splitpasswd
+ self.assertEqual(splitpasswd('user:ab'), ('user', 'ab'))
+ self.assertEqual(splitpasswd('user:a\nb'), ('user', 'a\nb'))
+ self.assertEqual(splitpasswd('user:a\tb'), ('user', 'a\tb'))
+ self.assertEqual(splitpasswd('user:a\rb'), ('user', 'a\rb'))
+ self.assertEqual(splitpasswd('user:a\fb'), ('user', 'a\fb'))
+ self.assertEqual(splitpasswd('user:a\vb'), ('user', 'a\vb'))
+ self.assertEqual(splitpasswd('user:a:b'), ('user', 'a:b'))
+ self.assertEqual(splitpasswd('user:a b'), ('user', 'a b'))
+ self.assertEqual(splitpasswd('user 2:ab'), ('user 2', 'ab'))
+ self.assertEqual(splitpasswd('user+1:a+b'), ('user+1', 'a+b'))
+ self.assertEqual(splitpasswd('user:'), ('user', ''))
+ self.assertEqual(splitpasswd('user'), ('user', None))
+ self.assertEqual(splitpasswd(':ab'), ('', 'ab'))
+
+ def test_splitport(self):
+ splitport = urllib.splitport
+ self.assertEqual(splitport('parrot:88'), ('parrot', '88'))
+ self.assertEqual(splitport('parrot'), ('parrot', None))
+ self.assertEqual(splitport('parrot:'), ('parrot', None))
+ self.assertEqual(splitport('127.0.0.1'), ('127.0.0.1', None))
+ self.assertEqual(splitport('parrot:cheese'), ('parrot:cheese', None))
+ self.assertEqual(splitport('[::1]:88'), ('[::1]', '88'))
+ self.assertEqual(splitport('[::1]'), ('[::1]', None))
+ self.assertEqual(splitport(':88'), ('', '88'))
+
+ def test_splitnport(self):
+ splitnport = urllib.splitnport
+ self.assertEqual(splitnport('parrot:88'), ('parrot', 88))
+ self.assertEqual(splitnport('parrot'), ('parrot', -1))
+ self.assertEqual(splitnport('parrot', 55), ('parrot', 55))
+ self.assertEqual(splitnport('parrot:'), ('parrot', -1))
+ self.assertEqual(splitnport('parrot:', 55), ('parrot', 55))
+ self.assertEqual(splitnport('127.0.0.1'), ('127.0.0.1', -1))
+ self.assertEqual(splitnport('127.0.0.1', 55), ('127.0.0.1', 55))
+ self.assertEqual(splitnport('parrot:cheese'), ('parrot', None))
+ self.assertEqual(splitnport('parrot:cheese', 55), ('parrot', None))
+
+ def test_splitquery(self):
+ # Normal cases are exercised by other tests; ensure that we also
+ # catch cases with no port specified (testcase ensuring coverage)
+ splitquery = urllib.splitquery
+ self.assertEqual(splitquery('http://python.org/fake?foo=bar'),
+ ('http://python.org/fake', 'foo=bar'))
+ self.assertEqual(splitquery('http://python.org/fake?foo=bar?'),
+ ('http://python.org/fake?foo=bar', ''))
+ self.assertEqual(splitquery('http://python.org/fake'),
+ ('http://python.org/fake', None))
+ self.assertEqual(splitquery('?foo=bar'), ('', 'foo=bar'))
+
+ def test_splittag(self):
+ splittag = urllib.splittag
+ self.assertEqual(splittag('http://example.com?foo=bar#baz'),
+ ('http://example.com?foo=bar', 'baz'))
+ self.assertEqual(splittag('http://example.com?foo=bar#'),
+ ('http://example.com?foo=bar', ''))
+ self.assertEqual(splittag('#baz'), ('', 'baz'))
+ self.assertEqual(splittag('http://example.com?foo=bar'),
+ ('http://example.com?foo=bar', None))
+ self.assertEqual(splittag('http://example.com?foo=bar#baz#boo'),
+ ('http://example.com?foo=bar#baz', 'boo'))
+
+ def test_splitattr(self):
+ splitattr = urllib.splitattr
+ self.assertEqual(splitattr('/path;attr1=value1;attr2=value2'),
+ ('/path', ['attr1=value1', 'attr2=value2']))
+ self.assertEqual(splitattr('/path;'), ('/path', ['']))
+ self.assertEqual(splitattr(';attr1=value1;attr2=value2'),
+ ('', ['attr1=value1', 'attr2=value2']))
+ self.assertEqual(splitattr('/path'), ('/path', []))
+
+ def test_splitvalue(self):
+ # Normal cases are exercised by other tests; test pathological cases
+ # with no key/value pairs. (testcase ensuring coverage)
+ splitvalue = urllib.splitvalue
+ self.assertEqual(splitvalue('foo=bar'), ('foo', 'bar'))
+ self.assertEqual(splitvalue('foo='), ('foo', ''))
+ self.assertEqual(splitvalue('=bar'), ('', 'bar'))
+ self.assertEqual(splitvalue('foobar'), ('foobar', None))
+ self.assertEqual(splitvalue('foo=bar=baz'), ('foo', 'bar=baz'))
+
+ def test_toBytes(self):
+ result = urllib.toBytes(u'http://www.python.org')
+ self.assertEqual(result, 'http://www.python.org')
+ self.assertRaises(UnicodeError, urllib.toBytes,
+ test_support.u(r'http://www.python.org/medi\u00e6val'))
+
+ def test_unwrap(self):
+ url = urllib.unwrap('<URL:type://host/path>')
+ self.assertEqual(url, 'type://host/path')
class URLopener_Tests(unittest.TestCase):
@@ -814,7 +1009,7 @@
# Everywhere else they work ok, but on those machines, sometimes
# fail in one of the tests, sometimes in other. I have a linux, and
# the tests go ok.
-# If anybody has one of the problematic enviroments, please help!
+# If anybody has one of the problematic environments, please help!
# . Facundo
#
# def server(evt):
@@ -860,7 +1055,7 @@
# def testTimeoutNone(self):
# # global default timeout is ignored
# import socket
-# self.assertTrue(socket.getdefaulttimeout() is None)
+# self.assertIsNone(socket.getdefaulttimeout())
# socket.setdefaulttimeout(30)
# try:
# ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9093, [])
@@ -872,7 +1067,7 @@
# def testTimeoutDefault(self):
# # global default timeout is used
# import socket
-# self.assertTrue(socket.getdefaulttimeout() is None)
+# self.assertIsNone(socket.getdefaulttimeout())
# socket.setdefaulttimeout(30)
# try:
# ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9093, [])
@@ -906,6 +1101,8 @@
Pathname_Tests,
Utility_Tests,
URLopener_Tests,
+ ProxyTests,
+ ProxyTests_withOrderedEnv,
#FTPWrapperTests,
)
diff --git a/Lib/urllib.py b/Lib/urllib.py
deleted file mode 100644
--- a/Lib/urllib.py
+++ /dev/null
@@ -1,1630 +0,0 @@
-"""Open an arbitrary URL.
-
-See the following document for more info on URLs:
-"Names and Addresses, URIs, URLs, URNs, URCs", at
-http://www.w3.org/pub/WWW/Addressing/Overview.html
-
-See also the HTTP spec (from which the error codes are derived):
-"HTTP - Hypertext Transfer Protocol", at
-http://www.w3.org/pub/WWW/Protocols/
-
-Related standards and specs:
-- RFC1808: the "relative URL" spec. (authoritative status)
-- RFC1738 - the "URL standard". (authoritative status)
-- RFC1630 - the "URI spec". (informational status)
-
-The object returned by URLopener().open(file) will differ per
-protocol. All you know is that is has methods read(), readline(),
-readlines(), fileno(), close() and info(). The read*(), fileno()
-and close() methods work like those of open files.
-The info() method returns a mimetools.Message object which can be
-used to query various info about the object, if available.
-(mimetools.Message objects are queried with the getheader() method.)
-"""
-
-import string
-import socket
-import os
-import time
-import sys
-import base64
-import re
-
-from urlparse import urljoin as basejoin
-
-__all__ = ["urlopen", "URLopener", "FancyURLopener", "urlretrieve",
- "urlcleanup", "quote", "quote_plus", "unquote", "unquote_plus",
- "urlencode", "url2pathname", "pathname2url", "splittag",
- "localhost", "thishost", "ftperrors", "basejoin", "unwrap",
- "splittype", "splithost", "splituser", "splitpasswd", "splitport",
- "splitnport", "splitquery", "splitattr", "splitvalue",
- "getproxies"]
-
-__version__ = '1.17' # XXX This version is not always updated :-(
-
-MAXFTPCACHE = 10 # Trim the ftp cache beyond this size
-
-# Helper for non-unix systems
-if (os._name if sys.platform.startswith('java') else os.name) == 'nt':
- from nturl2path import url2pathname, pathname2url
-elif os.name == 'riscos':
- from rourl2path import url2pathname, pathname2url
-else:
- def url2pathname(pathname):
- """OS-specific conversion from a relative URL of the 'file' scheme
- to a file system path; not recommended for general use."""
- return unquote(pathname)
-
- def pathname2url(pathname):
- """OS-specific conversion from a file system path to a relative URL
- of the 'file' scheme; not recommended for general use."""
- return quote(pathname)
-
-# This really consists of two pieces:
-# (1) a class which handles opening of all sorts of URLs
-# (plus assorted utilities etc.)
-# (2) a set of functions for parsing URLs
-# XXX Should these be separated out into different modules?
-
-
-# Shortcut for basic usage
-_urlopener = None
-def urlopen(url, data=None, proxies=None, context=None):
- """Create a file-like object for the specified URL to read from."""
- from warnings import warnpy3k
- warnpy3k("urllib.urlopen() has been removed in Python 3.0 in "
- "favor of urllib2.urlopen()", stacklevel=2)
-
- global _urlopener
- if proxies is not None or context is not None:
- opener = FancyURLopener(proxies=proxies, context=context)
- elif not _urlopener:
- opener = FancyURLopener()
- _urlopener = opener
- else:
- opener = _urlopener
- if data is None:
- return opener.open(url)
- else:
- return opener.open(url, data)
-def urlretrieve(url, filename=None, reporthook=None, data=None, context=None):
- global _urlopener
- if context is not None:
- opener = FancyURLopener(context=context)
- elif not _urlopener:
- _urlopener = opener = FancyURLopener()
- else:
- opener = _urlopener
- return opener.retrieve(url, filename, reporthook, data)
-def urlcleanup():
- if _urlopener:
- _urlopener.cleanup()
- _safe_quoters.clear()
- ftpcache.clear()
-
-# check for SSL
-try:
- import ssl
-except:
- _have_ssl = False
-else:
- _have_ssl = True
-
-# exception raised when downloaded size does not match content-length
-class ContentTooShortError(IOError):
- def __init__(self, message, content):
- IOError.__init__(self, message)
- self.content = content
-
-ftpcache = {}
-class URLopener:
- """Class to open URLs.
- This is a class rather than just a subroutine because we may need
- more than one set of global protocol-specific options.
- Note -- this is a base class for those who don't want the
- automatic handling of errors type 302 (relocated) and 401
- (authorization needed)."""
-
- __tempfiles = None
-
- version = "Python-urllib/%s" % __version__
-
- # Constructor
- def __init__(self, proxies=None, context=None, **x509):
- if proxies is None:
- proxies = getproxies()
- assert hasattr(proxies, 'has_key'), "proxies must be a mapping"
- self.proxies = proxies
- self.key_file = x509.get('key_file')
- self.cert_file = x509.get('cert_file')
- self.context = context
- self.addheaders = [('User-Agent', self.version)]
- self.__tempfiles = []
- self.__unlink = os.unlink # See cleanup()
- self.tempcache = None
- # Undocumented feature: if you assign {} to tempcache,
- # it is used to cache files retrieved with
- # self.retrieve(). This is not enabled by default
- # since it does not work for changing documents (and I
- # haven't got the logic to check expiration headers
- # yet).
- self.ftpcache = ftpcache
- # Undocumented feature: you can use a different
- # ftp cache by assigning to the .ftpcache member;
- # in case you want logically independent URL openers
- # XXX This is not threadsafe. Bah.
-
- def __del__(self):
- self.close()
-
- def close(self):
- self.cleanup()
-
- def cleanup(self):
- # This code sometimes runs when the rest of this module
- # has already been deleted, so it can't use any globals
- # or import anything.
- if self.__tempfiles:
- for file in self.__tempfiles:
- try:
- self.__unlink(file)
- except OSError:
- pass
- del self.__tempfiles[:]
- if self.tempcache:
- self.tempcache.clear()
-
- def addheader(self, *args):
- """Add a header to be used by the HTTP interface only
- e.g. u.addheader('Accept', 'sound/basic')"""
- self.addheaders.append(args)
-
- # External interface
- def open(self, fullurl, data=None):
- """Use URLopener().open(file) instead of open(file, 'r')."""
- fullurl = unwrap(toBytes(fullurl))
- # percent encode url, fixing lame server errors for e.g, like space
- # within url paths.
- fullurl = quote(fullurl, safe="%/:=&?~#+!$,;'@()*[]|")
- if self.tempcache and fullurl in self.tempcache:
- filename, headers = self.tempcache[fullurl]
- fp = open(filename, 'rb')
- return addinfourl(fp, headers, fullurl)
- urltype, url = splittype(fullurl)
- if not urltype:
- urltype = 'file'
- if urltype in self.proxies:
- proxy = self.proxies[urltype]
- urltype, proxyhost = splittype(proxy)
- host, selector = splithost(proxyhost)
- url = (host, fullurl) # Signal special case to open_*()
- else:
- proxy = None
- name = 'open_' + urltype
- self.type = urltype
- name = name.replace('-', '_')
- if not hasattr(self, name):
- if proxy:
- return self.open_unknown_proxy(proxy, fullurl, data)
- else:
- return self.open_unknown(fullurl, data)
- try:
- if data is None:
- return getattr(self, name)(url)
- else:
- return getattr(self, name)(url, data)
- except socket.error, msg:
- raise IOError, ('socket error', msg), sys.exc_info()[2]
-
- def open_unknown(self, fullurl, data=None):
- """Overridable interface to open unknown URL type."""
- type, url = splittype(fullurl)
- raise IOError, ('url error', 'unknown url type', type)
-
- def open_unknown_proxy(self, proxy, fullurl, data=None):
- """Overridable interface to open unknown URL type."""
- type, url = splittype(fullurl)
- raise IOError, ('url error', 'invalid proxy for %s' % type, proxy)
-
- # External interface
- def retrieve(self, url, filename=None, reporthook=None, data=None):
- """retrieve(url) returns (filename, headers) for a local object
- or (tempfilename, headers) for a remote object."""
- url = unwrap(toBytes(url))
- if self.tempcache and url in self.tempcache:
- return self.tempcache[url]
- type, url1 = splittype(url)
- if filename is None and (not type or type == 'file'):
- try:
- fp = self.open_local_file(url1)
- hdrs = fp.info()
- fp.close()
- return url2pathname(splithost(url1)[1]), hdrs
- except IOError:
- pass
- fp = self.open(url, data)
- try:
- headers = fp.info()
- if filename:
- tfp = open(filename, 'wb')
- else:
- import tempfile
- garbage, path = splittype(url)
- garbage, path = splithost(path or "")
- path, garbage = splitquery(path or "")
- path, garbage = splitattr(path or "")
- suffix = os.path.splitext(path)[1]
- (fd, filename) = tempfile.mkstemp(suffix)
- self.__tempfiles.append(filename)
- tfp = os.fdopen(fd, 'wb')
- try:
- result = filename, headers
- if self.tempcache is not None:
- self.tempcache[url] = result
- bs = 1024*8
- size = -1
- read = 0
- blocknum = 0
- if "content-length" in headers:
- size = int(headers["Content-Length"])
- if reporthook:
- reporthook(blocknum, bs, size)
- while 1:
- block = fp.read(bs)
- if block == "":
- break
- read += len(block)
- tfp.write(block)
- blocknum += 1
- if reporthook:
- reporthook(blocknum, bs, size)
- finally:
- tfp.close()
- finally:
- fp.close()
-
- # raise exception if actual size does not match content-length header
- if size >= 0 and read < size:
- raise ContentTooShortError("retrieval incomplete: got only %i out "
- "of %i bytes" % (read, size), result)
-
- return result
-
- # Each method named open_<type> knows how to open that type of URL
-
- def open_http(self, url, data=None):
- """Use HTTP protocol."""
- import httplib
- user_passwd = None
- proxy_passwd= None
- if isinstance(url, str):
- host, selector = splithost(url)
- if host:
- user_passwd, host = splituser(host)
- host = unquote(host)
- realhost = host
- else:
- host, selector = url
- # check whether the proxy contains authorization information
- proxy_passwd, host = splituser(host)
- # now we proceed with the url we want to obtain
- urltype, rest = splittype(selector)
- url = rest
- user_passwd = None
- if urltype.lower() != 'http':
- realhost = None
- else:
- realhost, rest = splithost(rest)
- if realhost:
- user_passwd, realhost = splituser(realhost)
- if user_passwd:
- selector = "%s://%s%s" % (urltype, realhost, rest)
- if proxy_bypass(realhost):
- host = realhost
-
- #print "proxy via http:", host, selector
- if not host: raise IOError, ('http error', 'no host given')
-
- if proxy_passwd:
- proxy_passwd = unquote(proxy_passwd)
- proxy_auth = base64.b64encode(proxy_passwd).strip()
- else:
- proxy_auth = None
-
- if user_passwd:
- user_passwd = unquote(user_passwd)
- auth = base64.b64encode(user_passwd).strip()
- else:
- auth = None
- h = httplib.HTTP(host)
- if data is not None:
- h.putrequest('POST', selector)
- h.putheader('Content-Type', 'application/x-www-form-urlencoded')
- h.putheader('Content-Length', '%d' % len(data))
- else:
- h.putrequest('GET', selector)
- if proxy_auth: h.putheader('Proxy-Authorization', 'Basic %s' % proxy_auth)
- if auth: h.putheader('Authorization', 'Basic %s' % auth)
- if realhost: h.putheader('Host', realhost)
- for args in self.addheaders: h.putheader(*args)
- h.endheaders(data)
- errcode, errmsg, headers = h.getreply()
- fp = h.getfile()
- if errcode == -1:
- if fp: fp.close()
- # something went wrong with the HTTP status line
- raise IOError, ('http protocol error', 0,
- 'got a bad status line', None)
- # According to RFC 2616, "2xx" code indicates that the client's
- # request was successfully received, understood, and accepted.
- if (200 <= errcode < 300):
- return addinfourl(fp, headers, "http:" + url, errcode)
- else:
- if data is None:
- return self.http_error(url, fp, errcode, errmsg, headers)
- else:
- return self.http_error(url, fp, errcode, errmsg, headers, data)
-
- def http_error(self, url, fp, errcode, errmsg, headers, data=None):
- """Handle http errors.
- Derived class can override this, or provide specific handlers
- named http_error_DDD where DDD is the 3-digit error code."""
- # First check if there's a specific handler for this error
- name = 'http_error_%d' % errcode
- if hasattr(self, name):
- method = getattr(self, name)
- if data is None:
- result = method(url, fp, errcode, errmsg, headers)
- else:
- result = method(url, fp, errcode, errmsg, headers, data)
- if result: return result
- return self.http_error_default(url, fp, errcode, errmsg, headers)
-
- def http_error_default(self, url, fp, errcode, errmsg, headers):
- """Default error handler: close the connection and raise IOError."""
- fp.close()
- raise IOError, ('http error', errcode, errmsg, headers)
-
- if _have_ssl:
- def open_https(self, url, data=None):
- """Use HTTPS protocol."""
-
- import httplib
- user_passwd = None
- proxy_passwd = None
- if isinstance(url, str):
- host, selector = splithost(url)
- if host:
- user_passwd, host = splituser(host)
- host = unquote(host)
- realhost = host
- else:
- host, selector = url
- # here, we determine, whether the proxy contains authorization information
- proxy_passwd, host = splituser(host)
- urltype, rest = splittype(selector)
- url = rest
- user_passwd = None
- if urltype.lower() != 'https':
- realhost = None
- else:
- realhost, rest = splithost(rest)
- if realhost:
- user_passwd, realhost = splituser(realhost)
- if user_passwd:
- selector = "%s://%s%s" % (urltype, realhost, rest)
- #print "proxy via https:", host, selector
- if not host: raise IOError, ('https error', 'no host given')
- if proxy_passwd:
- proxy_passwd = unquote(proxy_passwd)
- proxy_auth = base64.b64encode(proxy_passwd).strip()
- else:
- proxy_auth = None
- if user_passwd:
- user_passwd = unquote(user_passwd)
- auth = base64.b64encode(user_passwd).strip()
- else:
- auth = None
- h = httplib.HTTPS(host, 0,
- key_file=self.key_file,
- cert_file=self.cert_file,
- context=self.context)
- if data is not None:
- h.putrequest('POST', selector)
- h.putheader('Content-Type',
- 'application/x-www-form-urlencoded')
- h.putheader('Content-Length', '%d' % len(data))
- else:
- h.putrequest('GET', selector)
- if proxy_auth: h.putheader('Proxy-Authorization', 'Basic %s' % proxy_auth)
- if auth: h.putheader('Authorization', 'Basic %s' % auth)
- if realhost: h.putheader('Host', realhost)
- for args in self.addheaders: h.putheader(*args)
- h.endheaders(data)
- errcode, errmsg, headers = h.getreply()
- fp = h.getfile()
- if errcode == -1:
- if fp: fp.close()
- # something went wrong with the HTTP status line
- raise IOError, ('http protocol error', 0,
- 'got a bad status line', None)
- # According to RFC 2616, "2xx" code indicates that the client's
- # request was successfully received, understood, and accepted.
- if (200 <= errcode < 300):
- return addinfourl(fp, headers, "https:" + url, errcode)
- else:
- if data is None:
- return self.http_error(url, fp, errcode, errmsg, headers)
- else:
- return self.http_error(url, fp, errcode, errmsg, headers,
- data)
-
- def open_file(self, url):
- """Use local file or FTP depending on form of URL."""
- if not isinstance(url, str):
- raise IOError, ('file error', 'proxy support for file protocol currently not implemented')
- if url[:2] == '//' and url[2:3] != '/' and url[2:12].lower() != 'localhost/':
- return self.open_ftp(url)
- else:
- return self.open_local_file(url)
-
- def open_local_file(self, url):
- """Use local file."""
- import mimetypes, mimetools, email.utils
- try:
- from cStringIO import StringIO
- except ImportError:
- from StringIO import StringIO
- host, file = splithost(url)
- localname = url2pathname(file)
- try:
- stats = os.stat(localname)
- except OSError, e:
- raise IOError(e.errno, e.strerror, e.filename)
- size = stats.st_size
- modified = email.utils.formatdate(stats.st_mtime, usegmt=True)
- mtype = mimetypes.guess_type(url)[0]
- headers = mimetools.Message(StringIO(
- 'Content-Type: %s\nContent-Length: %d\nLast-modified: %s\n' %
- (mtype or 'text/plain', size, modified)))
- if not host:
- urlfile = file
- if file[:1] == '/':
- urlfile = 'file://' + file
- elif file[:2] == './':
- raise ValueError("local file url may start with / or file:. Unknown url of type: %s" % url)
- return addinfourl(open(localname, 'rb'),
- headers, urlfile)
- host, port = splitport(host)
- if not port \
- and socket.gethostbyname(host) in (localhost(), thishost()):
- urlfile = file
- if file[:1] == '/':
- urlfile = 'file://' + file
- return addinfourl(open(localname, 'rb'),
- headers, urlfile)
- raise IOError, ('local file error', 'not on local host')
-
- def open_ftp(self, url):
- """Use FTP protocol."""
- if not isinstance(url, str):
- raise IOError, ('ftp error', 'proxy support for ftp protocol currently not implemented')
- import mimetypes, mimetools
- try:
- from cStringIO import StringIO
- except ImportError:
- from StringIO import StringIO
- host, path = splithost(url)
- if not host: raise IOError, ('ftp error', 'no host given')
- host, port = splitport(host)
- user, host = splituser(host)
- if user: user, passwd = splitpasswd(user)
- else: passwd = None
- host = unquote(host)
- user = user or ''
- passwd = passwd or ''
- host = socket.gethostbyname(host)
- if not port:
- import ftplib
- port = ftplib.FTP_PORT
- else:
- port = int(port)
- path, attrs = splitattr(path)
- path = unquote(path)
- dirs = path.split('/')
- dirs, file = dirs[:-1], dirs[-1]
- if dirs and not dirs[0]: dirs = dirs[1:]
- if dirs and not dirs[0]: dirs[0] = '/'
- key = user, host, port, '/'.join(dirs)
- # XXX thread unsafe!
- if len(self.ftpcache) > MAXFTPCACHE:
- # Prune the cache, rather arbitrarily
- for k in self.ftpcache.keys():
- if k != key:
- v = self.ftpcache[k]
- del self.ftpcache[k]
- v.close()
- try:
- if not key in self.ftpcache:
- self.ftpcache[key] = \
- ftpwrapper(user, passwd, host, port, dirs)
- if not file: type = 'D'
- else: type = 'I'
- for attr in attrs:
- attr, value = splitvalue(attr)
- if attr.lower() == 'type' and \
- value in ('a', 'A', 'i', 'I', 'd', 'D'):
- type = value.upper()
- (fp, retrlen) = self.ftpcache[key].retrfile(file, type)
- mtype = mimetypes.guess_type("ftp:" + url)[0]
- headers = ""
- if mtype:
- headers += "Content-Type: %s\n" % mtype
- if retrlen is not None and retrlen >= 0:
- headers += "Content-Length: %d\n" % retrlen
- headers = mimetools.Message(StringIO(headers))
- return addinfourl(fp, headers, "ftp:" + url)
- except ftperrors(), msg:
- raise IOError, ('ftp error', msg), sys.exc_info()[2]
-
- def open_data(self, url, data=None):
- """Use "data" URL."""
- if not isinstance(url, str):
- raise IOError, ('data error', 'proxy support for data protocol currently not implemented')
- # ignore POSTed data
- #
- # syntax of data URLs:
- # dataurl := "data:" [ mediatype ] [ ";base64" ] "," data
- # mediatype := [ type "/" subtype ] *( ";" parameter )
- # data := *urlchar
- # parameter := attribute "=" value
- import mimetools
- try:
- from cStringIO import StringIO
- except ImportError:
- from StringIO import StringIO
- try:
- [type, data] = url.split(',', 1)
- except ValueError:
- raise IOError, ('data error', 'bad data URL')
- if not type:
- type = 'text/plain;charset=US-ASCII'
- semi = type.rfind(';')
- if semi >= 0 and '=' not in type[semi:]:
- encoding = type[semi+1:]
- type = type[:semi]
- else:
- encoding = ''
- msg = []
- msg.append('Date: %s'%time.strftime('%a, %d %b %Y %H:%M:%S GMT',
- time.gmtime(time.time())))
- msg.append('Content-type: %s' % type)
- if encoding == 'base64':
- data = base64.decodestring(data)
- else:
- data = unquote(data)
- msg.append('Content-Length: %d' % len(data))
- msg.append('')
- msg.append(data)
- msg = '\n'.join(msg)
- f = StringIO(msg)
- headers = mimetools.Message(f, 0)
- #f.fileno = None # needed for addinfourl
- return addinfourl(f, headers, url)
-
-
-class FancyURLopener(URLopener):
- """Derived class with handlers for errors we can handle (perhaps)."""
-
- def __init__(self, *args, **kwargs):
- URLopener.__init__(self, *args, **kwargs)
- self.auth_cache = {}
- self.tries = 0
- self.maxtries = 10
-
- def http_error_default(self, url, fp, errcode, errmsg, headers):
- """Default error handling -- don't raise an exception."""
- return addinfourl(fp, headers, "http:" + url, errcode)
-
- def http_error_302(self, url, fp, errcode, errmsg, headers, data=None):
- """Error 302 -- relocated (temporarily)."""
- self.tries += 1
- if self.maxtries and self.tries >= self.maxtries:
- if hasattr(self, "http_error_500"):
- meth = self.http_error_500
- else:
- meth = self.http_error_default
- self.tries = 0
- return meth(url, fp, 500,
- "Internal Server Error: Redirect Recursion", headers)
- result = self.redirect_internal(url, fp, errcode, errmsg, headers,
- data)
- self.tries = 0
- return result
-
- def redirect_internal(self, url, fp, errcode, errmsg, headers, data):
- if 'location' in headers:
- newurl = headers['location']
- elif 'uri' in headers:
- newurl = headers['uri']
- else:
- return
- fp.close()
- # In case the server sent a relative URL, join with original:
- newurl = basejoin(self.type + ":" + url, newurl)
-
- # For security reasons we do not allow redirects to protocols
- # other than HTTP, HTTPS or FTP.
- newurl_lower = newurl.lower()
- if not (newurl_lower.startswith('http://') or
- newurl_lower.startswith('https://') or
- newurl_lower.startswith('ftp://')):
- raise IOError('redirect error', errcode,
- errmsg + " - Redirection to url '%s' is not allowed" %
- newurl,
- headers)
-
- return self.open(newurl)
-
- def http_error_301(self, url, fp, errcode, errmsg, headers, data=None):
- """Error 301 -- also relocated (permanently)."""
- return self.http_error_302(url, fp, errcode, errmsg, headers, data)
-
- def http_error_303(self, url, fp, errcode, errmsg, headers, data=None):
- """Error 303 -- also relocated (essentially identical to 302)."""
- return self.http_error_302(url, fp, errcode, errmsg, headers, data)
-
- def http_error_307(self, url, fp, errcode, errmsg, headers, data=None):
- """Error 307 -- relocated, but turn POST into error."""
- if data is None:
- return self.http_error_302(url, fp, errcode, errmsg, headers, data)
- else:
- return self.http_error_default(url, fp, errcode, errmsg, headers)
-
- def http_error_401(self, url, fp, errcode, errmsg, headers, data=None):
- """Error 401 -- authentication required.
- This function supports Basic authentication only."""
- if not 'www-authenticate' in headers:
- URLopener.http_error_default(self, url, fp,
- errcode, errmsg, headers)
- stuff = headers['www-authenticate']
- import re
- match = re.match('[ \t]*([^ \t]+)[ \t]+realm="([^"]*)"', stuff)
- if not match:
- URLopener.http_error_default(self, url, fp,
- errcode, errmsg, headers)
- scheme, realm = match.groups()
- if scheme.lower() != 'basic':
- URLopener.http_error_default(self, url, fp,
- errcode, errmsg, headers)
- name = 'retry_' + self.type + '_basic_auth'
- if data is None:
- return getattr(self,name)(url, realm)
- else:
- return getattr(self,name)(url, realm, data)
-
- def http_error_407(self, url, fp, errcode, errmsg, headers, data=None):
- """Error 407 -- proxy authentication required.
- This function supports Basic authentication only."""
- if not 'proxy-authenticate' in headers:
- URLopener.http_error_default(self, url, fp,
- errcode, errmsg, headers)
- stuff = headers['proxy-authenticate']
- import re
- match = re.match('[ \t]*([^ \t]+)[ \t]+realm="([^"]*)"', stuff)
- if not match:
- URLopener.http_error_default(self, url, fp,
- errcode, errmsg, headers)
- scheme, realm = match.groups()
- if scheme.lower() != 'basic':
- URLopener.http_error_default(self, url, fp,
- errcode, errmsg, headers)
- name = 'retry_proxy_' + self.type + '_basic_auth'
- if data is None:
- return getattr(self,name)(url, realm)
- else:
- return getattr(self,name)(url, realm, data)
-
- def retry_proxy_http_basic_auth(self, url, realm, data=None):
- host, selector = splithost(url)
- newurl = 'http://' + host + selector
- proxy = self.proxies['http']
- urltype, proxyhost = splittype(proxy)
- proxyhost, proxyselector = splithost(proxyhost)
- i = proxyhost.find('@') + 1
- proxyhost = proxyhost[i:]
- user, passwd = self.get_user_passwd(proxyhost, realm, i)
- if not (user or passwd): return None
- proxyhost = quote(user, safe='') + ':' + quote(passwd, safe='') + '@' + proxyhost
- self.proxies['http'] = 'http://' + proxyhost + proxyselector
- if data is None:
- return self.open(newurl)
- else:
- return self.open(newurl, data)
-
- def retry_proxy_https_basic_auth(self, url, realm, data=None):
- host, selector = splithost(url)
- newurl = 'https://' + host + selector
- proxy = self.proxies['https']
- urltype, proxyhost = splittype(proxy)
- proxyhost, proxyselector = splithost(proxyhost)
- i = proxyhost.find('@') + 1
- proxyhost = proxyhost[i:]
- user, passwd = self.get_user_passwd(proxyhost, realm, i)
- if not (user or passwd): return None
- proxyhost = quote(user, safe='') + ':' + quote(passwd, safe='') + '@' + proxyhost
- self.proxies['https'] = 'https://' + proxyhost + proxyselector
- if data is None:
- return self.open(newurl)
- else:
- return self.open(newurl, data)
-
- def retry_http_basic_auth(self, url, realm, data=None):
- host, selector = splithost(url)
- i = host.find('@') + 1
- host = host[i:]
- user, passwd = self.get_user_passwd(host, realm, i)
- if not (user or passwd): return None
- host = quote(user, safe='') + ':' + quote(passwd, safe='') + '@' + host
- newurl = 'http://' + host + selector
- if data is None:
- return self.open(newurl)
- else:
- return self.open(newurl, data)
-
- def retry_https_basic_auth(self, url, realm, data=None):
- host, selector = splithost(url)
- i = host.find('@') + 1
- host = host[i:]
- user, passwd = self.get_user_passwd(host, realm, i)
- if not (user or passwd): return None
- host = quote(user, safe='') + ':' + quote(passwd, safe='') + '@' + host
- newurl = 'https://' + host + selector
- if data is None:
- return self.open(newurl)
- else:
- return self.open(newurl, data)
-
- def get_user_passwd(self, host, realm, clear_cache=0):
- key = realm + '@' + host.lower()
- if key in self.auth_cache:
- if clear_cache:
- del self.auth_cache[key]
- else:
- return self.auth_cache[key]
- user, passwd = self.prompt_user_passwd(host, realm)
- if user or passwd: self.auth_cache[key] = (user, passwd)
- return user, passwd
-
- def prompt_user_passwd(self, host, realm):
- """Override this in a GUI environment!"""
- import getpass
- try:
- user = raw_input("Enter username for %s at %s: " % (realm,
- host))
- passwd = getpass.getpass("Enter password for %s in %s at %s: " %
- (user, realm, host))
- return user, passwd
- except KeyboardInterrupt:
- print
- return None, None
-
-
-# Utility functions
-
-_localhost = None
-def localhost():
- """Return the IP address of the magic hostname 'localhost'."""
- global _localhost
- if _localhost is None:
- _localhost = socket.gethostbyname('localhost')
- return _localhost
-
-_thishost = None
-def thishost():
- """Return the IP address of the current host."""
- global _thishost
- if _thishost is None:
- try:
- _thishost = socket.gethostbyname(socket.gethostname())
- except socket.gaierror:
- _thishost = socket.gethostbyname('localhost')
- return _thishost
-
-_ftperrors = None
-def ftperrors():
- """Return the set of errors raised by the FTP class."""
- global _ftperrors
- if _ftperrors is None:
- import ftplib
- _ftperrors = ftplib.all_errors
- return _ftperrors
-
-_noheaders = None
-def noheaders():
- """Return an empty mimetools.Message object."""
- global _noheaders
- if _noheaders is None:
- import mimetools
- try:
- from cStringIO import StringIO
- except ImportError:
- from StringIO import StringIO
- _noheaders = mimetools.Message(StringIO(), 0)
- _noheaders.fp.close() # Recycle file descriptor
- return _noheaders
-
-
-# Utility classes
-
-class ftpwrapper:
- """Class used by open_ftp() for cache of open FTP connections."""
-
- def __init__(self, user, passwd, host, port, dirs,
- timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
- persistent=True):
- self.user = user
- self.passwd = passwd
- self.host = host
- self.port = port
- self.dirs = dirs
- self.timeout = timeout
- self.refcount = 0
- self.keepalive = persistent
- try:
- self.init()
- except:
- self.close()
- raise
-
- def init(self):
- import ftplib
- self.busy = 0
- self.ftp = ftplib.FTP()
- self.ftp.connect(self.host, self.port, self.timeout)
- self.ftp.login(self.user, self.passwd)
- _target = '/'.join(self.dirs)
- self.ftp.cwd(_target)
-
- def retrfile(self, file, type):
- import ftplib
- self.endtransfer()
- if type in ('d', 'D'): cmd = 'TYPE A'; isdir = 1
- else: cmd = 'TYPE ' + type; isdir = 0
- try:
- self.ftp.voidcmd(cmd)
- except ftplib.all_errors:
- self.init()
- self.ftp.voidcmd(cmd)
- conn = None
- if file and not isdir:
- # Try to retrieve as a file
- try:
- cmd = 'RETR ' + file
- conn, retrlen = self.ftp.ntransfercmd(cmd)
- except ftplib.error_perm, reason:
- if str(reason)[:3] != '550':
- raise IOError, ('ftp error', reason), sys.exc_info()[2]
- if not conn:
- # Set transfer mode to ASCII!
- self.ftp.voidcmd('TYPE A')
- # Try a directory listing. Verify that directory exists.
- if file:
- pwd = self.ftp.pwd()
- try:
- try:
- self.ftp.cwd(file)
- except ftplib.error_perm, reason:
- raise IOError, ('ftp error', reason), sys.exc_info()[2]
- finally:
- self.ftp.cwd(pwd)
- cmd = 'LIST ' + file
- else:
- cmd = 'LIST'
- conn, retrlen = self.ftp.ntransfercmd(cmd)
- self.busy = 1
- ftpobj = addclosehook(conn.makefile('rb'), self.file_close)
- self.refcount += 1
- conn.close()
- # Pass back both a suitably decorated object and a retrieval length
- return (ftpobj, retrlen)
-
- def endtransfer(self):
- if not self.busy:
- return
- self.busy = 0
- try:
- self.ftp.voidresp()
- except ftperrors():
- pass
-
- def close(self):
- self.keepalive = False
- if self.refcount <= 0:
- self.real_close()
-
- def file_close(self):
- self.endtransfer()
- self.refcount -= 1
- if self.refcount <= 0 and not self.keepalive:
- self.real_close()
-
- def real_close(self):
- self.endtransfer()
- try:
- self.ftp.close()
- except ftperrors():
- pass
-
-class addbase:
- """Base class for addinfo and addclosehook."""
-
- def __init__(self, fp):
- self.fp = fp
- self.read = self.fp.read
- self.readline = self.fp.readline
- if hasattr(self.fp, "readlines"): self.readlines = self.fp.readlines
- if hasattr(self.fp, "fileno"):
- self.fileno = self.fp.fileno
- else:
- self.fileno = lambda: None
- if hasattr(self.fp, "__iter__"):
- self.__iter__ = self.fp.__iter__
- if hasattr(self.fp, "next"):
- self.next = self.fp.next
-
- def __repr__(self):
- return '<%s at %r whose fp = %r>' % (self.__class__.__name__,
- id(self), self.fp)
-
- def close(self):
- self.read = None
- self.readline = None
- self.readlines = None
- self.fileno = None
- if self.fp: self.fp.close()
- self.fp = None
-
-class addclosehook(addbase):
- """Class to add a close hook to an open file."""
-
- def __init__(self, fp, closehook, *hookargs):
- addbase.__init__(self, fp)
- self.closehook = closehook
- self.hookargs = hookargs
-
- def close(self):
- try:
- closehook = self.closehook
- hookargs = self.hookargs
- if closehook:
- self.closehook = None
- self.hookargs = None
- closehook(*hookargs)
- finally:
- addbase.close(self)
-
-
-class addinfo(addbase):
- """class to add an info() method to an open file."""
-
- def __init__(self, fp, headers):
- addbase.__init__(self, fp)
- self.headers = headers
-
- def info(self):
- return self.headers
-
-class addinfourl(addbase):
- """class to add info() and geturl() methods to an open file."""
-
- def __init__(self, fp, headers, url, code=None):
- addbase.__init__(self, fp)
- self.headers = headers
- self.url = url
- self.code = code
-
- def info(self):
- return self.headers
-
- def getcode(self):
- return self.code
-
- def geturl(self):
- return self.url
-
-
-# Utilities to parse URLs (most of these return None for missing parts):
-# unwrap('<URL:type://host/path>') --> 'type://host/path'
-# splittype('type:opaquestring') --> 'type', 'opaquestring'
-# splithost('//host[:port]/path') --> 'host[:port]', '/path'
-# splituser('user[:passwd]@host[:port]') --> 'user[:passwd]', 'host[:port]'
-# splitpasswd('user:passwd') -> 'user', 'passwd'
-# splitport('host:port') --> 'host', 'port'
-# splitquery('/path?query') --> '/path', 'query'
-# splittag('/path#tag') --> '/path', 'tag'
-# splitattr('/path;attr1=value1;attr2=value2;...') ->
-# '/path', ['attr1=value1', 'attr2=value2', ...]
-# splitvalue('attr=value') --> 'attr', 'value'
-# unquote('abc%20def') -> 'abc def'
-# quote('abc def') -> 'abc%20def')
-
-try:
- unicode
-except NameError:
- def _is_unicode(x):
- return 0
-else:
- def _is_unicode(x):
- return isinstance(x, unicode)
-
-def toBytes(url):
- """toBytes(u"URL") --> 'URL'."""
- # Most URL schemes require ASCII. If that changes, the conversion
- # can be relaxed
- if _is_unicode(url):
- try:
- url = url.encode("ASCII")
- except UnicodeError:
- raise UnicodeError("URL " + repr(url) +
- " contains non-ASCII characters")
- return url
-
-def unwrap(url):
- """unwrap('<URL:type://host/path>') --> 'type://host/path'."""
- url = url.strip()
- if url[:1] == '<' and url[-1:] == '>':
- url = url[1:-1].strip()
- if url[:4] == 'URL:': url = url[4:].strip()
- return url
-
-_typeprog = None
-def splittype(url):
- """splittype('type:opaquestring') --> 'type', 'opaquestring'."""
- global _typeprog
- if _typeprog is None:
- import re
- _typeprog = re.compile('^([^/:]+):')
-
- match = _typeprog.match(url)
- if match:
- scheme = match.group(1)
- return scheme.lower(), url[len(scheme) + 1:]
- return None, url
-
-_hostprog = None
-def splithost(url):
- """splithost('//host[:port]/path') --> 'host[:port]', '/path'."""
- global _hostprog
- if _hostprog is None:
- import re
- _hostprog = re.compile('^//([^/?]*)(.*)$')
-
- match = _hostprog.match(url)
- if match:
- host_port = match.group(1)
- path = match.group(2)
- if path and not path.startswith('/'):
- path = '/' + path
- return host_port, path
- return None, url
-
-_userprog = None
-def splituser(host):
- """splituser('user[:passwd]@host[:port]') --> 'user[:passwd]', 'host[:port]'."""
- global _userprog
- if _userprog is None:
- import re
- _userprog = re.compile('^(.*)@(.*)$')
-
- match = _userprog.match(host)
- if match: return match.group(1, 2)
- return None, host
-
-_passwdprog = None
-def splitpasswd(user):
- """splitpasswd('user:passwd') -> 'user', 'passwd'."""
- global _passwdprog
- if _passwdprog is None:
- import re
- _passwdprog = re.compile('^([^:]*):(.*)$',re.S)
-
- match = _passwdprog.match(user)
- if match: return match.group(1, 2)
- return user, None
-
-# splittag('/path#tag') --> '/path', 'tag'
-_portprog = None
-def splitport(host):
- """splitport('host:port') --> 'host', 'port'."""
- global _portprog
- if _portprog is None:
- import re
- _portprog = re.compile('^(.*):([0-9]*)$')
-
- match = _portprog.match(host)
- if match:
- host, port = match.groups()
- if port:
- return host, port
- return host, None
-
-_nportprog = None
-def splitnport(host, defport=-1):
- """Split host and port, returning numeric port.
- Return given default port if no ':' found; defaults to -1.
- Return numerical port if a valid number are found after ':'.
- Return None if ':' but not a valid number."""
- global _nportprog
- if _nportprog is None:
- import re
- _nportprog = re.compile('^(.*):(.*)$')
-
- match = _nportprog.match(host)
- if match:
- host, port = match.group(1, 2)
- if port:
- try:
- nport = int(port)
- except ValueError:
- nport = None
- return host, nport
- return host, defport
-
-_queryprog = None
-def splitquery(url):
- """splitquery('/path?query') --> '/path', 'query'."""
- global _queryprog
- if _queryprog is None:
- import re
- _queryprog = re.compile('^(.*)\?([^?]*)$')
-
- match = _queryprog.match(url)
- if match: return match.group(1, 2)
- return url, None
-
-_tagprog = None
-def splittag(url):
- """splittag('/path#tag') --> '/path', 'tag'."""
- global _tagprog
- if _tagprog is None:
- import re
- _tagprog = re.compile('^(.*)#([^#]*)$')
-
- match = _tagprog.match(url)
- if match: return match.group(1, 2)
- return url, None
-
-def splitattr(url):
- """splitattr('/path;attr1=value1;attr2=value2;...') ->
- '/path', ['attr1=value1', 'attr2=value2', ...]."""
- words = url.split(';')
- return words[0], words[1:]
-
-_valueprog = None
-def splitvalue(attr):
- """splitvalue('attr=value') --> 'attr', 'value'."""
- global _valueprog
- if _valueprog is None:
- import re
- _valueprog = re.compile('^([^=]*)=(.*)$')
-
- match = _valueprog.match(attr)
- if match: return match.group(1, 2)
- return attr, None
-
-# urlparse contains a duplicate of this method to avoid a circular import. If
-# you update this method, also update the copy in urlparse. This code
-# duplication does not exist in Python3.
-
-_hexdig = '0123456789ABCDEFabcdef'
-_hextochr = dict((a + b, chr(int(a + b, 16)))
- for a in _hexdig for b in _hexdig)
-_asciire = re.compile('([\x00-\x7f]+)')
-
-def unquote(s):
- """unquote('abc%20def') -> 'abc def'."""
- res = s.split('%')
- # fastpath
- if len(res) == 1:
- return s
- buf = [res[0]]
- is_unicode = isinstance(s, unicode)
- for item in res[1:]:
- try:
- if is_unicode:
- buf.append(unichr(int(item[:2], 16)))
- buf.append(item[2:])
- else:
- buf.append(_hextochr[item[:2]])
- buf.append(item[2:])
- except KeyError:
- buf.append('%')
- buf.append(item)
- return ''.join(buf)
-
-def unquote_plus(s):
- """unquote('%7e/abc+def') -> '~/abc def'"""
- s = s.replace('+', ' ')
- return unquote(s)
-
-always_safe = ('ABCDEFGHIJKLMNOPQRSTUVWXYZ'
- 'abcdefghijklmnopqrstuvwxyz'
- '0123456789' '_.-')
-_safe_map = {}
-for i, c in zip(xrange(256), str(bytearray(xrange(256)))):
- _safe_map[c] = c if (i < 128 and c in always_safe) else '%{:02X}'.format(i)
-_safe_quoters = {}
-
-def quote(s, safe='/'):
- """quote('abc def') -> 'abc%20def'
-
- Each part of a URL, e.g. the path info, the query, etc., has a
- different set of reserved characters that must be quoted.
-
- RFC 2396 Uniform Resource Identifiers (URI): Generic Syntax lists
- the following reserved characters.
-
- reserved = ";" | "/" | "?" | ":" | "@" | "&" | "=" | "+" |
- "$" | ","
-
- Each of these characters is reserved in some component of a URL,
- but not necessarily in all of them.
-
- By default, the quote function is intended for quoting the path
- section of a URL. Thus, it will not encode '/'. This character
- is reserved, but in typical usage the quote function is being
- called on a path where the existing slash characters are used as
- reserved characters.
- """
- # fastpath
- if not s:
- if s is None:
- raise TypeError('None object cannot be quoted')
- return s
- cachekey = (safe, always_safe)
- try:
- (quoter, safe) = _safe_quoters[cachekey]
- except KeyError:
- safe_map = _safe_map.copy()
- safe_map.update([(c, c) for c in safe])
- quoter = safe_map.__getitem__
- safe = always_safe + safe
- _safe_quoters[cachekey] = (quoter, safe)
- if not s.rstrip(safe):
- return s
- return ''.join(map(quoter, s))
-
-def quote_plus(s, safe=''):
- """Quote the query fragment of a URL; replacing ' ' with '+'"""
- if ' ' in s:
- s = quote(s, safe + ' ')
- return s.replace(' ', '+')
- return quote(s, safe)
-
-def urlencode(query, doseq=0):
- """Encode a sequence of two-element tuples or dictionary into a URL query string.
-
- If any values in the query arg are sequences and doseq is true, each
- sequence element is converted to a separate parameter.
-
- If the query arg is a sequence of two-element tuples, the order of the
- parameters in the output will match the order of parameters in the
- input.
- """
-
- if hasattr(query,"items"):
- # mapping objects
- query = query.items()
- else:
- # it's a bother at times that strings and string-like objects are
- # sequences...
- try:
- # non-sequence items should not work with len()
- # non-empty strings will fail this
- if len(query) and not isinstance(query[0], tuple):
- raise TypeError
- # zero-length sequences of all types will get here and succeed,
- # but that's a minor nit - since the original implementation
- # allowed empty dicts that type of behavior probably should be
- # preserved for consistency
- except TypeError:
- ty,va,tb = sys.exc_info()
- raise TypeError, "not a valid non-string sequence or mapping object", tb
-
- l = []
- if not doseq:
- # preserve old behavior
- for k, v in query:
- k = quote_plus(str(k))
- v = quote_plus(str(v))
- l.append(k + '=' + v)
- else:
- for k, v in query:
- k = quote_plus(str(k))
- if isinstance(v, str):
- v = quote_plus(v)
- l.append(k + '=' + v)
- elif _is_unicode(v):
- # is there a reasonable way to convert to ASCII?
- # encode generates a string, but "replace" or "ignore"
- # lose information and "strict" can raise UnicodeError
- v = quote_plus(v.encode("ASCII","replace"))
- l.append(k + '=' + v)
- else:
- try:
- # is this a sufficient test for sequence-ness?
- len(v)
- except TypeError:
- # not a sequence
- v = quote_plus(str(v))
- l.append(k + '=' + v)
- else:
- # loop over the sequence
- for elt in v:
- l.append(k + '=' + quote_plus(str(elt)))
- return '&'.join(l)
-
-# Proxy handling
-def getproxies_environment():
- """Return a dictionary of scheme -> proxy server URL mappings.
-
- Scan the environment for variables named <scheme>_proxy;
- this seems to be the standard convention. If you need a
- different way, you can pass a proxies dictionary to the
- [Fancy]URLopener constructor.
-
- """
- proxies = {}
- for name, value in os.environ.items():
- name = name.lower()
- if value and name[-6:] == '_proxy':
- proxies[name[:-6]] = value
- return proxies
-
-def proxy_bypass_environment(host):
- """Test if proxies should not be used for a particular host.
-
- Checks the environment for a variable named no_proxy, which should
- be a list of DNS suffixes separated by commas, or '*' for all hosts.
- """
- no_proxy = os.environ.get('no_proxy', '') or os.environ.get('NO_PROXY', '')
- # '*' is special case for always bypass
- if no_proxy == '*':
- return 1
- # strip port off host
- hostonly, port = splitport(host)
- # check if the host ends with any of the DNS suffixes
- no_proxy_list = [proxy.strip() for proxy in no_proxy.split(',')]
- for name in no_proxy_list:
- if name and (hostonly.endswith(name) or host.endswith(name)):
- return 1
- # otherwise, don't bypass
- return 0
-
-
-if sys.platform == 'darwin':
- from _scproxy import _get_proxy_settings, _get_proxies
-
- def proxy_bypass_macosx_sysconf(host):
- """
- Return True iff this host shouldn't be accessed using a proxy
-
- This function uses the MacOSX framework SystemConfiguration
- to fetch the proxy information.
- """
- import re
- import socket
- from fnmatch import fnmatch
-
- hostonly, port = splitport(host)
-
- def ip2num(ipAddr):
- parts = ipAddr.split('.')
- parts = map(int, parts)
- if len(parts) != 4:
- parts = (parts + [0, 0, 0, 0])[:4]
- return (parts[0] << 24) | (parts[1] << 16) | (parts[2] << 8) | parts[3]
-
- proxy_settings = _get_proxy_settings()
-
- # Check for simple host names:
- if '.' not in host:
- if proxy_settings['exclude_simple']:
- return True
-
- hostIP = None
-
- for value in proxy_settings.get('exceptions', ()):
- # Items in the list are strings like these: *.local, 169.254/16
- if not value: continue
-
- m = re.match(r"(\d+(?:\.\d+)*)(/\d+)?", value)
- if m is not None:
- if hostIP is None:
- try:
- hostIP = socket.gethostbyname(hostonly)
- hostIP = ip2num(hostIP)
- except socket.error:
- continue
-
- base = ip2num(m.group(1))
- mask = m.group(2)
- if mask is None:
- mask = 8 * (m.group(1).count('.') + 1)
-
- else:
- mask = int(mask[1:])
- mask = 32 - mask
-
- if (hostIP >> mask) == (base >> mask):
- return True
-
- elif fnmatch(host, value):
- return True
-
- return False
-
- def getproxies_macosx_sysconf():
- """Return a dictionary of scheme -> proxy server URL mappings.
-
- This function uses the MacOSX framework SystemConfiguration
- to fetch the proxy information.
- """
- return _get_proxies()
-
- def proxy_bypass(host):
- if getproxies_environment():
- return proxy_bypass_environment(host)
- else:
- return proxy_bypass_macosx_sysconf(host)
-
- def getproxies():
- return getproxies_environment() or getproxies_macosx_sysconf()
-
-elif os.name == 'nt':
- def getproxies_registry():
- """Return a dictionary of scheme -> proxy server URL mappings.
-
- Win32 uses the registry to store proxies.
-
- """
- proxies = {}
- try:
- import _winreg
- except ImportError:
- # Std module, so should be around - but you never know!
- return proxies
- try:
- internetSettings = _winreg.OpenKey(_winreg.HKEY_CURRENT_USER,
- r'Software\Microsoft\Windows\CurrentVersion\Internet Settings')
- proxyEnable = _winreg.QueryValueEx(internetSettings,
- 'ProxyEnable')[0]
- if proxyEnable:
- # Returned as Unicode but problems if not converted to ASCII
- proxyServer = str(_winreg.QueryValueEx(internetSettings,
- 'ProxyServer')[0])
- if '=' in proxyServer:
- # Per-protocol settings
- for p in proxyServer.split(';'):
- protocol, address = p.split('=', 1)
- # See if address has a type:// prefix
- import re
- if not re.match('^([^/:]+)://', address):
- address = '%s://%s' % (protocol, address)
- proxies[protocol] = address
- else:
- # Use one setting for all protocols
- if proxyServer[:5] == 'http:':
- proxies['http'] = proxyServer
- else:
- proxies['http'] = 'http://%s' % proxyServer
- proxies['https'] = 'https://%s' % proxyServer
- proxies['ftp'] = 'ftp://%s' % proxyServer
- internetSettings.Close()
- except (WindowsError, ValueError, TypeError):
- # Either registry key not found etc, or the value in an
- # unexpected format.
- # proxies already set up to be empty so nothing to do
- pass
- return proxies
-
- def getproxies():
- """Return a dictionary of scheme -> proxy server URL mappings.
-
- Returns settings gathered from the environment, if specified,
- or the registry.
-
- """
- return getproxies_environment() or getproxies_registry()
-
- def proxy_bypass_registry(host):
- try:
- import _winreg
- import re
- except ImportError:
- # Std modules, so should be around - but you never know!
- return 0
- try:
- internetSettings = _winreg.OpenKey(_winreg.HKEY_CURRENT_USER,
- r'Software\Microsoft\Windows\CurrentVersion\Internet Settings')
- proxyEnable = _winreg.QueryValueEx(internetSettings,
- 'ProxyEnable')[0]
- proxyOverride = str(_winreg.QueryValueEx(internetSettings,
- 'ProxyOverride')[0])
- # ^^^^ Returned as Unicode but problems if not converted to ASCII
- except WindowsError:
- return 0
- if not proxyEnable or not proxyOverride:
- return 0
- # try to make a host list from name and IP address.
- rawHost, port = splitport(host)
- host = [rawHost]
- try:
- addr = socket.gethostbyname(rawHost)
- if addr != rawHost:
- host.append(addr)
- except socket.error:
- pass
- try:
- fqdn = socket.getfqdn(rawHost)
- if fqdn != rawHost:
- host.append(fqdn)
- except socket.error:
- pass
- # make a check value list from the registry entry: replace the
- # '<local>' string by the localhost entry and the corresponding
- # canonical entry.
- proxyOverride = proxyOverride.split(';')
- # now check if we match one of the registry values.
- for test in proxyOverride:
- if test == '<local>':
- if '.' not in rawHost:
- return 1
- test = test.replace(".", r"\.") # mask dots
- test = test.replace("*", r".*") # change glob sequence
- test = test.replace("?", r".") # change glob char
- for val in host:
- # print "%s <--> %s" %( test, val )
- if re.match(test, val, re.I):
- return 1
- return 0
-
- def proxy_bypass(host):
- """Return a dictionary of scheme -> proxy server URL mappings.
-
- Returns settings gathered from the environment, if specified,
- or the registry.
-
- """
- if getproxies_environment():
- return proxy_bypass_environment(host)
- else:
- return proxy_bypass_registry(host)
-
-else:
- # By default use environment variables
- getproxies = getproxies_environment
- proxy_bypass = proxy_bypass_environment
-
-# Test and time quote() and unquote()
-def test1():
- s = ''
- for i in range(256): s = s + chr(i)
- s = s*4
- t0 = time.time()
- qs = quote(s)
- uqs = unquote(qs)
- t1 = time.time()
- if uqs != s:
- print 'Wrong!'
- print repr(s)
- print repr(qs)
- print repr(uqs)
- print round(t1 - t0, 3), 'sec'
-
-
-def reporthook(blocknum, blocksize, totalsize):
- # Report during remote transfers
- print "Block number: %d, Block size: %d, Total size: %d" % (
- blocknum, blocksize, totalsize)
diff --git a/lib-python/2.7/rfc822.py b/lib-python/2.7/rfc822.py
--- a/lib-python/2.7/rfc822.py
+++ b/lib-python/2.7/rfc822.py
@@ -179,6 +179,11 @@
lst.append(line)
self.dict[headerseen] = line[len(headerseen)+1:].strip()
continue
+ elif headerseen is not None:
+ # An empty header name. These aren't allowed in HTTP, but it's
+ # probably a benign mistake. Don't add the header, just keep
+ # going.
+ continue
else:
# It's not a header line; throw it back and stop here.
if not self.dict:
@@ -202,7 +207,7 @@
data in RFC 2822-like formats with special header formats.
"""
i = line.find(':')
- if i > 0:
+ if i > -1:
return line[:i].lower()
return None
@@ -956,7 +961,7 @@
According to RFC 1123, day and month names must always be in
English. If not for that, this code could use strftime(). It
- can't because strftime() honors the locale and could generated
+ can't because strftime() honors the locale and could generate
non-English names.
"""
if timeval is None:
diff --git a/lib-python/2.7/test/test_httplib.py b/lib-python/2.7/test/test_httplib.py
--- a/lib-python/2.7/test/test_httplib.py
+++ b/lib-python/2.7/test/test_httplib.py
@@ -1,15 +1,25 @@
import httplib
+import itertools
import array
-import httplib
import StringIO
import socket
import errno
+import os
+import tempfile
import unittest
TestCase = unittest.TestCase
from test import test_support
+here = os.path.dirname(__file__)
+# Self-signed cert file for 'localhost'
+CERT_localhost = os.path.join(here, 'keycert.pem')
+# Self-signed cert file for 'fakehostname'
+CERT_fakehostname = os.path.join(here, 'keycert2.pem')
+# Self-signed cert file for self-signed.pythontest.net
+CERT_selfsigned_pythontestdotnet = os.path.join(here, 'selfsigned_pythontestdotnet.pem')
+
HOST = test_support.HOST
class FakeSocket:
@@ -17,6 +27,7 @@
self.text = text
self.fileclass = fileclass
self.data = ''
+ self.file_closed = False
self.host = host
self.port = port
@@ -26,7 +37,13 @@
def makefile(self, mode, bufsize=None):
if mode != 'r' and mode != 'rb':
raise httplib.UnimplementedFileMode()
- return self.fileclass(self.text)
+ # keep the file around so we can check how much was read from it
+ self.file = self.fileclass(self.text)
+ self.file.close = self.file_close #nerf close ()
+ return self.file
+
+ def file_close(self):
+ self.file_closed = True
def close(self):
pass
@@ -107,21 +124,59 @@
self.content_length = kv[1].strip()
list.append(self, item)
- # POST with empty body
- conn = httplib.HTTPConnection('example.com')
- conn.sock = FakeSocket(None)
- conn._buffer = ContentLengthChecker()
- conn.request('POST', '/', '')
- self.assertEqual(conn._buffer.content_length, '0',
- 'Header Content-Length not set')
+ # Here, we're testing that methods expecting a body get a
+ # content-length set to zero if the body is empty (either None or '')
+ bodies = (None, '')
+ methods_with_body = ('PUT', 'POST', 'PATCH')
+ for method, body in itertools.product(methods_with_body, bodies):
+ conn = httplib.HTTPConnection('example.com')
+ conn.sock = FakeSocket(None)
+ conn._buffer = ContentLengthChecker()
+ conn.request(method, '/', body)
+ self.assertEqual(
+ conn._buffer.content_length, '0',
+ 'Header Content-Length incorrect on {}'.format(method)
+ )
- # PUT request with empty body
- conn = httplib.HTTPConnection('example.com')
- conn.sock = FakeSocket(None)
- conn._buffer = ContentLengthChecker()
- conn.request('PUT', '/', '')
- self.assertEqual(conn._buffer.content_length, '0',
- 'Header Content-Length not set')
+ # For these methods, we make sure that content-length is not set when
+ # the body is None because it might cause unexpected behaviour on the
+ # server.
+ methods_without_body = (
+ 'GET', 'CONNECT', 'DELETE', 'HEAD', 'OPTIONS', 'TRACE',
+ )
+ for method in methods_without_body:
+ conn = httplib.HTTPConnection('example.com')
+ conn.sock = FakeSocket(None)
+ conn._buffer = ContentLengthChecker()
+ conn.request(method, '/', None)
+ self.assertEqual(
+ conn._buffer.content_length, None,
+ 'Header Content-Length set for empty body on {}'.format(method)
+ )
+
+ # If the body is set to '', that's considered to be "present but
+ # empty" rather than "missing", so content length would be set, even
+ # for methods that don't expect a body.
+ for method in methods_without_body:
+ conn = httplib.HTTPConnection('example.com')
+ conn.sock = FakeSocket(None)
+ conn._buffer = ContentLengthChecker()
+ conn.request(method, '/', '')
+ self.assertEqual(
+ conn._buffer.content_length, '0',
+ 'Header Content-Length incorrect on {}'.format(method)
+ )
+
+ # If the body is set, make sure Content-Length is set.
+ for method in itertools.chain(methods_without_body, methods_with_body):
+ conn = httplib.HTTPConnection('example.com')
+ conn.sock = FakeSocket(None)
+ conn._buffer = ContentLengthChecker()
+ conn.request(method, '/', ' ')
+ self.assertEqual(
+ conn._buffer.content_length, '1',
+ 'Header Content-Length incorrect on {}'.format(method)
+ )
def test_putheader(self):
conn = httplib.HTTPConnection('example.com')
@@ -130,9 +185,36 @@
conn.putheader('Content-length',42)
self.assertIn('Content-length: 42', conn._buffer)
+ conn.putheader('Foo', ' bar ')
+ self.assertIn(b'Foo: bar ', conn._buffer)
+ conn.putheader('Bar', '\tbaz\t')
+ self.assertIn(b'Bar: \tbaz\t', conn._buffer)
+ conn.putheader('Authorization', 'Bearer mytoken')
+ self.assertIn(b'Authorization: Bearer mytoken', conn._buffer)
+ conn.putheader('IterHeader', 'IterA', 'IterB')
+ self.assertIn(b'IterHeader: IterA\r\n\tIterB', conn._buffer)
+ conn.putheader('LatinHeader', b'\xFF')
+ self.assertIn(b'LatinHeader: \xFF', conn._buffer)
+ conn.putheader('Utf8Header', b'\xc3\x80')
+ self.assertIn(b'Utf8Header: \xc3\x80', conn._buffer)
+ conn.putheader('C1-Control', b'next\x85line')
+ self.assertIn(b'C1-Control: next\x85line', conn._buffer)
+ conn.putheader('Embedded-Fold-Space', 'is\r\n allowed')
+ self.assertIn(b'Embedded-Fold-Space: is\r\n allowed', conn._buffer)
+ conn.putheader('Embedded-Fold-Tab', 'is\r\n\tallowed')
+ self.assertIn(b'Embedded-Fold-Tab: is\r\n\tallowed', conn._buffer)
+ conn.putheader('Key Space', 'value')
+ self.assertIn(b'Key Space: value', conn._buffer)
+ conn.putheader('KeySpace ', 'value')
+ self.assertIn(b'KeySpace : value', conn._buffer)
+ conn.putheader(b'Nonbreak\xa0Space', 'value')
+ self.assertIn(b'Nonbreak\xa0Space: value', conn._buffer)
+ conn.putheader(b'\xa0NonbreakSpace', 'value')
+ self.assertIn(b'\xa0NonbreakSpace: value', conn._buffer)
+
def test_ipv6host_header(self):
- # Default host header on IPv6 transaction should wrapped by [] if
- # its actual IPv6 address
+ # Default host header on IPv6 transaction should be wrapped by [] if
+ # it is an IPv6 address
expected = 'GET /foo HTTP/1.1\r\nHost: [2001::]:81\r\n' \
'Accept-Encoding: identity\r\n\r\n'
conn = httplib.HTTPConnection('[2001::]:81')
@@ -149,6 +231,45 @@
conn.request('GET', '/foo')
self.assertTrue(sock.data.startswith(expected))
+ def test_malformed_headers_coped_with(self):
+ # Issue 19996
+ body = "HTTP/1.1 200 OK\r\nFirst: val\r\n: nval\r\nSecond: val\r\n\r\n"
+ sock = FakeSocket(body)
+ resp = httplib.HTTPResponse(sock)
+ resp.begin()
+
+ self.assertEqual(resp.getheader('First'), 'val')
+ self.assertEqual(resp.getheader('Second'), 'val')
+
+ def test_invalid_headers(self):
+ conn = httplib.HTTPConnection('example.com')
+ conn.sock = FakeSocket('')
+ conn.putrequest('GET', '/')
+
+ # http://tools.ietf.org/html/rfc7230#section-3.2.4, whitespace is no
+ # longer allowed in header names
+ cases = (
+ (b'Invalid\r\nName', b'ValidValue'),
+ (b'Invalid\rName', b'ValidValue'),
+ (b'Invalid\nName', b'ValidValue'),
+ (b'\r\nInvalidName', b'ValidValue'),
+ (b'\rInvalidName', b'ValidValue'),
+ (b'\nInvalidName', b'ValidValue'),
+ (b' InvalidName', b'ValidValue'),
+ (b'\tInvalidName', b'ValidValue'),
+ (b'Invalid:Name', b'ValidValue'),
+ (b':InvalidName', b'ValidValue'),
+ (b'ValidName', b'Invalid\r\nValue'),
+ (b'ValidName', b'Invalid\rValue'),
+ (b'ValidName', b'Invalid\nValue'),
+ (b'ValidName', b'InvalidValue\r\n'),
+ (b'ValidName', b'InvalidValue\r'),
+ (b'ValidName', b'InvalidValue\n'),
+ )
+ for name, value in cases:
+ with self.assertRaisesRegexp(ValueError, 'Invalid header'):
+ conn.putheader(name, value)
+
class BasicTest(TestCase):
def test_status_lines(self):
@@ -279,6 +400,22 @@
conn.sock = sock
conn.request('GET', '/foo', body)
self.assertTrue(sock.data.startswith(expected))
+ self.assertIn('def test_send_file', sock.data)
+
+ def test_send_tempfile(self):
+ expected = ('GET /foo HTTP/1.1\r\nHost: example.com\r\n'
+ 'Accept-Encoding: identity\r\nContent-Length: 9\r\n\r\n'
+ 'fake\ndata')
+
+ with tempfile.TemporaryFile() as body:
+ body.write('fake\ndata')
+ body.seek(0)
+
+ conn = httplib.HTTPConnection('example.com')
+ sock = FakeSocket(body)
+ conn.sock = sock
+ conn.request('GET', '/foo', body)
+ self.assertEqual(sock.data, expected)
def test_send(self):
expected = 'this is a test this is only a test'
@@ -425,12 +562,42 @@
self.assertEqual(resp.read(), '')
self.assertTrue(resp.isclosed())
+ def test_error_leak(self):
+ # Test that the socket is not leaked if getresponse() fails
+ conn = httplib.HTTPConnection('example.com')
+ response = []
+ class Response(httplib.HTTPResponse):
+ def __init__(self, *pos, **kw):
+ response.append(self) # Avoid garbage collector closing the socket
+ httplib.HTTPResponse.__init__(self, *pos, **kw)
+ conn.response_class = Response
+ conn.sock = FakeSocket('') # Emulate server dropping connection
+ conn.request('GET', '/')
+ self.assertRaises(httplib.BadStatusLine, conn.getresponse)
+ self.assertTrue(response)
+ #self.assertTrue(response[0].closed)
+ self.assertTrue(conn.sock.file_closed)
+
+ def test_proxy_tunnel_without_status_line(self):
+ # Issue 17849: If a proxy tunnel is created that does not return
+ # a status code, fail.
+ body = 'hello world'
+ conn = httplib.HTTPConnection('example.com', strict=False)
+ conn.set_tunnel('foo')
+ conn.sock = FakeSocket(body)
+ with self.assertRaisesRegexp(socket.error, "Invalid response"):
+ conn._tunnel()
+
class OfflineTest(TestCase):
def test_responses(self):
self.assertEqual(httplib.responses[httplib.NOT_FOUND], "Not Found")
-class SourceAddressTest(TestCase):
+class TestServerMixin:
+ """A limited socket server mixin.
+
+ This is used by test cases for testing http connection end points.
+ """
def setUp(self):
self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.port = test_support.bind_port(self.serv)
@@ -445,6 +612,7 @@
self.serv.close()
self.serv = None
+class SourceAddressTest(TestServerMixin, TestCase):
def testHTTPConnectionSourceAddress(self):
self.conn = httplib.HTTPConnection(HOST, self.port,
source_address=('', self.source_port))
@@ -461,6 +629,24 @@
# for an ssl_wrapped connect() to actually return from.
+class HTTPTest(TestServerMixin, TestCase):
+ def testHTTPConnection(self):
+ self.conn = httplib.HTTP(host=HOST, port=self.port, strict=None)
+ self.conn.connect()
+ self.assertEqual(self.conn._conn.host, HOST)
+ self.assertEqual(self.conn._conn.port, self.port)
+
+ def testHTTPWithConnectHostPort(self):
+ testhost = 'unreachable.test.domain'
+ testport = '80'
+ self.conn = httplib.HTTP(host=testhost, port=testport)
+ self.conn.connect(host=HOST, port=self.port)
+ self.assertNotEqual(self.conn._conn.host, testhost)
+ self.assertNotEqual(self.conn._conn.port, testport)
+ self.assertEqual(self.conn._conn.host, HOST)
+ self.assertEqual(self.conn._conn.port, self.port)
+
+
class TimeoutTest(TestCase):
PORT = None
@@ -507,35 +693,138 @@
httpConn.close()
-class HTTPSTimeoutTest(TestCase):
-# XXX Here should be tests for HTTPS, there isn't any right now!
+class HTTPSTest(TestCase):
+
+ def setUp(self):
+ if not hasattr(httplib, 'HTTPSConnection'):
+ self.skipTest('ssl support required')
+
+ def make_server(self, certfile):
+ from test.ssl_servers import make_https_server
+ return make_https_server(self, certfile=certfile)
def test_attributes(self):
- # simple test to check it's storing it
- if hasattr(httplib, 'HTTPSConnection'):
- h = httplib.HTTPSConnection(HOST, TimeoutTest.PORT, timeout=30)
- self.assertEqual(h.timeout, 30)
+ # simple test to check it's storing the timeout
+ h = httplib.HTTPSConnection(HOST, TimeoutTest.PORT, timeout=30)
+ self.assertEqual(h.timeout, 30)
- @unittest.skipIf(not hasattr(httplib, 'HTTPS'), 'httplib.HTTPS not available')
+ def test_networked(self):
+ # Default settings: requires a valid cert from a trusted CA
+ import ssl
+ test_support.requires('network')
+ with test_support.transient_internet('self-signed.pythontest.net'):
+ h = httplib.HTTPSConnection('self-signed.pythontest.net', 443)
+ with self.assertRaises(ssl.SSLError) as exc_info:
+ h.request('GET', '/')
+ self.assertEqual(exc_info.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
+
+ def test_networked_noverification(self):
+ # Switch off cert verification
+ import ssl
+ test_support.requires('network')
+ with test_support.transient_internet('self-signed.pythontest.net'):
+ context = ssl._create_stdlib_context()
+ h = httplib.HTTPSConnection('self-signed.pythontest.net', 443,
+ context=context)
+ h.request('GET', '/')
+ resp = h.getresponse()
+ self.assertIn('nginx', resp.getheader('server'))
+
+ @test_support.system_must_validate_cert
+ def test_networked_trusted_by_default_cert(self):
+ # Default settings: requires a valid cert from a trusted CA
+ test_support.requires('network')
+ with test_support.transient_internet('www.python.org'):
+ h = httplib.HTTPSConnection('www.python.org', 443)
+ h.request('GET', '/')
+ resp = h.getresponse()
+ content_type = resp.getheader('content-type')
+ self.assertIn('text/html', content_type)
+
+ def test_networked_good_cert(self):
+ # We feed the server's cert as a validating cert
+ import ssl
+ test_support.requires('network')
+ with test_support.transient_internet('self-signed.pythontest.net'):
+ context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+ context.verify_mode = ssl.CERT_REQUIRED
+ context.load_verify_locations(CERT_selfsigned_pythontestdotnet)
+ h = httplib.HTTPSConnection('self-signed.pythontest.net', 443, context=context)
+ h.request('GET', '/')
+ resp = h.getresponse()
+ server_string = resp.getheader('server')
+ self.assertIn('nginx', server_string)
+
+ def test_networked_bad_cert(self):
+ # We feed a "CA" cert that is unrelated to the server's cert
+ import ssl
+ test_support.requires('network')
+ with test_support.transient_internet('self-signed.pythontest.net'):
+ context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+ context.verify_mode = ssl.CERT_REQUIRED
+ context.load_verify_locations(CERT_localhost)
+ h = httplib.HTTPSConnection('self-signed.pythontest.net', 443, context=context)
+ with self.assertRaises(ssl.SSLError) as exc_info:
+ h.request('GET', '/')
+ self.assertEqual(exc_info.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
+
+ def test_local_unknown_cert(self):
+ # The custom cert isn't known to the default trust bundle
+ import ssl
+ server = self.make_server(CERT_localhost)
+ h = httplib.HTTPSConnection('localhost', server.port)
+ with self.assertRaises(ssl.SSLError) as exc_info:
+ h.request('GET', '/')
+ self.assertEqual(exc_info.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
+
+ def test_local_good_hostname(self):
+ # The (valid) cert validates the HTTP hostname
+ import ssl
+ server = self.make_server(CERT_localhost)
+ context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+ context.verify_mode = ssl.CERT_REQUIRED
+ context.load_verify_locations(CERT_localhost)
+ h = httplib.HTTPSConnection('localhost', server.port, context=context)
+ h.request('GET', '/nonexistent')
+ resp = h.getresponse()
+ self.assertEqual(resp.status, 404)
+
+ def test_local_bad_hostname(self):
+ # The (valid) cert doesn't validate the HTTP hostname
+ import ssl
+ server = self.make_server(CERT_fakehostname)
+ context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+ context.verify_mode = ssl.CERT_REQUIRED
+ context.check_hostname = True
+ context.load_verify_locations(CERT_fakehostname)
+ h = httplib.HTTPSConnection('localhost', server.port, context=context)
+ with self.assertRaises(ssl.CertificateError):
+ h.request('GET', '/')
+ h.close()
+ # With context.check_hostname=False, the mismatching is ignored
+ context.check_hostname = False
+ h = httplib.HTTPSConnection('localhost', server.port, context=context)
+ h.request('GET', '/nonexistent')
+ resp = h.getresponse()
+ self.assertEqual(resp.status, 404)
+
def test_host_port(self):
# Check invalid host_port
- # Note that httplib does not accept user:password@ in the host-port.
for hp in ("www.python.org:abc", "user:password at www.python.org"):
- self.assertRaises(httplib.InvalidURL, httplib.HTTP, hp)
+ self.assertRaises(httplib.InvalidURL, httplib.HTTPSConnection, hp)
- for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000", "fe80::207:e9ff:fe9b",
- 8000),
- ("pypi.python.org:443", "pypi.python.org", 443),
- ("pypi.python.org", "pypi.python.org", 443),
- ("pypi.python.org:", "pypi.python.org", 443),
- ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 443)):
- http = httplib.HTTPS(hp)
- c = http._conn
- if h != c.host:
- self.fail("Host incorrectly parsed: %s != %s" % (h, c.host))
- if p != c.port:
- self.fail("Port incorrectly parsed: %s != %s" % (p, c.host))
+ for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000",
+ "fe80::207:e9ff:fe9b", 8000),
+ ("www.python.org:443", "www.python.org", 443),
+ ("www.python.org:", "www.python.org", 443),
+ ("www.python.org", "www.python.org", 443),
+ ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 443),
+ ("[fe80::207:e9ff:fe9b]:", "fe80::207:e9ff:fe9b",
+ 443)):
+ c = httplib.HTTPSConnection(hp)
+ self.assertEqual(h, c.host)
+ self.assertEqual(p, c.port)
class TunnelTests(TestCase):
@@ -563,10 +852,12 @@
self.assertEqual(conn.sock.host, 'proxy.com')
self.assertEqual(conn.sock.port, 80)
- self.assertTrue('CONNECT destination.com' in conn.sock.data)
- self.assertTrue('Host: destination.com' in conn.sock.data)
+ self.assertIn('CONNECT destination.com', conn.sock.data)
+ # issue22095
+ self.assertNotIn('Host: destination.com:None', conn.sock.data)
+ self.assertIn('Host: destination.com', conn.sock.data)
- self.assertTrue('Host: proxy.com' not in conn.sock.data)
+ self.assertNotIn('Host: proxy.com', conn.sock.data)
conn.close()
@@ -577,9 +868,11 @@
self.assertTrue('Host: destination.com' in conn.sock.data)
+ at test_support.reap_threads
def test_main(verbose=None):
test_support.run_unittest(HeaderTests, OfflineTest, BasicTest, TimeoutTest,
- HTTPSTimeoutTest, SourceAddressTest, TunnelTests)
+ HTTPTest, HTTPSTest, SourceAddressTest,
+ TunnelTests)
if __name__ == '__main__':
test_main()
diff --git a/lib-python/2.7/test/test_rfc822.py b/lib-python/2.7/test/test_rfc822.py
--- a/lib-python/2.7/test/test_rfc822.py
+++ b/lib-python/2.7/test/test_rfc822.py
@@ -248,6 +248,12 @@
eq(rfc822.quote('foo\\wacky"name'), 'foo\\\\wacky\\"name')
eq(rfc822.unquote('"foo\\\\wacky\\"name"'), 'foo\\wacky"name')
+ def test_invalid_headers(self):
+ eq = self.assertEqual
+ msg = self.create_message("First: val\n: otherval\nSecond: val2\n")
+ eq(msg.getheader('First'), 'val')
+ eq(msg.getheader('Second'), 'val2')
+
def test_main():
test_support.run_unittest(MessageTestCase)
diff --git a/lib-python/2.7/test/test_urllib.py b/lib-python/2.7/test/test_urllib.py
--- a/lib-python/2.7/test/test_urllib.py
+++ b/lib-python/2.7/test/test_urllib.py
@@ -1,13 +1,14 @@
"""Regresssion tests for urllib"""
+import collections
import urllib
import httplib
+import io
import unittest
import os
import sys
import mimetools
import tempfile
-import StringIO
from test import test_support
from base64 import b64encode
@@ -21,37 +22,43 @@
return "%" + hex_repr
+def fakehttp(fakedata):
+ class FakeSocket(io.BytesIO):
+
+ def sendall(self, data):
+ FakeHTTPConnection.buf = data
+
+ def makefile(self, *args, **kwds):
+ return self
+
+ def read(self, amt=None):
+ if self.closed:
+ return b""
+ return io.BytesIO.read(self, amt)
+
+ def readline(self, length=None):
+ if self.closed:
+ return b""
+ return io.BytesIO.readline(self, length)
+
+ class FakeHTTPConnection(httplib.HTTPConnection):
+
+ # buffer to store data for verification in urlopen tests.
+ buf = ""
+
+ def connect(self):
+ self.sock = FakeSocket(self.fakedata)
+ self.__class__.fakesock = self.sock
+ FakeHTTPConnection.fakedata = fakedata
+
+ return FakeHTTPConnection
+
+
class FakeHTTPMixin(object):
def fakehttp(self, fakedata):
- class FakeSocket(StringIO.StringIO):
-
- def sendall(self, data):
- FakeHTTPConnection.buf = data
-
- def makefile(self, *args, **kwds):
- return self
-
- def read(self, amt=None):
- if self.closed:
- return ""
- return StringIO.StringIO.read(self, amt)
-
- def readline(self, length=None):
- if self.closed:
- return ""
- return StringIO.StringIO.readline(self, length)
-
- class FakeHTTPConnection(httplib.HTTPConnection):
-
- # buffer to store data for verification in urlopen tests.
- buf = ""
-
- def connect(self):
- self.sock = FakeSocket(fakedata)
-
assert httplib.HTTP._connection_class == httplib.HTTPConnection
- httplib.HTTP._connection_class = FakeHTTPConnection
+ httplib.HTTP._connection_class = fakehttp(fakedata)
def unfakehttp(self):
httplib.HTTP._connection_class = httplib.HTTPConnection
@@ -158,8 +165,71 @@
# getproxies_environment use lowered case truncated (no '_proxy') keys
self.assertEqual('localhost', proxies['no'])
# List of no_proxies with space.
- self.env.set('NO_PROXY', 'localhost, anotherdomain.com, newdomain.com')
+ self.env.set('NO_PROXY', 'localhost, anotherdomain.com, newdomain.com:1234')
self.assertTrue(urllib.proxy_bypass_environment('anotherdomain.com'))
+ self.assertTrue(urllib.proxy_bypass_environment('anotherdomain.com:8888'))
+ self.assertTrue(urllib.proxy_bypass_environment('newdomain.com:1234'))
+
+ def test_proxy_cgi_ignore(self):
+ try:
+ self.env.set('HTTP_PROXY', 'http://somewhere:3128')
+ proxies = urllib.getproxies_environment()
+ self.assertEqual('http://somewhere:3128', proxies['http'])
+ self.env.set('REQUEST_METHOD', 'GET')
+ proxies = urllib.getproxies_environment()
+ self.assertNotIn('http', proxies)
+ finally:
+ self.env.unset('REQUEST_METHOD')
+ self.env.unset('HTTP_PROXY')
+
+ def test_proxy_bypass_environment_host_match(self):
+ bypass = urllib.proxy_bypass_environment
+ self.env.set('NO_PROXY',
+ 'localhost, anotherdomain.com, newdomain.com:1234')
+ self.assertTrue(bypass('localhost'))
+ self.assertTrue(bypass('LocalHost')) # MixedCase
+ self.assertTrue(bypass('LOCALHOST')) # UPPERCASE
+ self.assertTrue(bypass('newdomain.com:1234'))
+ self.assertTrue(bypass('anotherdomain.com:8888'))
+ self.assertTrue(bypass('www.newdomain.com:1234'))
+ self.assertFalse(bypass('prelocalhost'))
+ self.assertFalse(bypass('newdomain.com')) # no port
+ self.assertFalse(bypass('newdomain.com:1235')) # wrong port
+
+class ProxyTests_withOrderedEnv(unittest.TestCase):
+
+ def setUp(self):
+ # We need to test conditions, where variable order _is_ significant
+ self._saved_env = os.environ
+ # Monkey patch os.environ, start with empty fake environment
+ os.environ = collections.OrderedDict()
+
+ def tearDown(self):
+ os.environ = self._saved_env
+
+ def test_getproxies_environment_prefer_lowercase(self):
+ # Test lowercase preference with removal
+ os.environ['no_proxy'] = ''
+ os.environ['No_Proxy'] = 'localhost'
+ self.assertFalse(urllib.proxy_bypass_environment('localhost'))
+ self.assertFalse(urllib.proxy_bypass_environment('arbitrary'))
+ os.environ['http_proxy'] = ''
+ os.environ['HTTP_PROXY'] = 'http://somewhere:3128'
+ proxies = urllib.getproxies_environment()
+ self.assertEqual({}, proxies)
+ # Test lowercase preference of proxy bypass and correct matching including ports
+ os.environ['no_proxy'] = 'localhost, noproxy.com, my.proxy:1234'
+ os.environ['No_Proxy'] = 'xyz.com'
+ self.assertTrue(urllib.proxy_bypass_environment('localhost'))
+ self.assertTrue(urllib.proxy_bypass_environment('noproxy.com:5678'))
+ self.assertTrue(urllib.proxy_bypass_environment('my.proxy:1234'))
+ self.assertFalse(urllib.proxy_bypass_environment('my.proxy'))
+ self.assertFalse(urllib.proxy_bypass_environment('arbitrary'))
+ # Test lowercase preference with replacement
+ os.environ['http_proxy'] = 'http://somewhere:3128'
+ os.environ['Http_Proxy'] = 'http://somewhereelse:3128'
+ proxies = urllib.getproxies_environment()
+ self.assertEqual('http://somewhere:3128', proxies['http'])
class urlopen_HttpTests(unittest.TestCase, FakeHTTPMixin):
@@ -209,10 +279,26 @@
Content-Type: text/html; charset=iso-8859-1
""")
try:
- self.assertRaises(IOError, urllib.urlopen, "http://python.org/")
+ msg = "Redirection to url 'file:"
+ with self.assertRaisesRegexp(IOError, msg):
+ urllib.urlopen("http://python.org/")
finally:
self.unfakehttp()
+ def test_redirect_limit_independent(self):
+ # Ticket #12923: make sure independent requests each use their
+ # own retry limit.
+ for i in range(urllib.FancyURLopener().maxtries):
+ self.fakehttp(b'''HTTP/1.1 302 Found
+Location: file://guidocomputer.athome.com:/python/license
+Connection: close
+''')
+ try:
+ self.assertRaises(IOError, urllib.urlopen,
+ "http://something")
+ finally:
+ self.unfakehttp()
+
def test_empty_socket(self):
# urlopen() raises IOError if the underlying socket does not send any
# data. (#1680230)
@@ -227,13 +313,13 @@
'file://localhost/a/missing/file.py')
fd, tmp_file = tempfile.mkstemp()
tmp_fileurl = 'file://localhost/' + tmp_file.replace(os.path.sep, '/')
+ self.assertTrue(os.path.exists(tmp_file))
try:
- self.assertTrue(os.path.exists(tmp_file))
fp = urllib.urlopen(tmp_fileurl)
+ fp.close()
finally:
os.close(fd)
- fp.close()
- os.unlink(tmp_file)
+ os.unlink(tmp_file)
self.assertFalse(os.path.exists(tmp_file))
self.assertRaises(IOError, urllib.urlopen, tmp_fileurl)
@@ -773,21 +859,131 @@
class Utility_Tests(unittest.TestCase):
"""Testcase to test the various utility functions in the urllib."""
+ # In Python 3 this test class is moved to test_urlparse.
+
+ def test_splittype(self):
+ splittype = urllib.splittype
+ self.assertEqual(splittype('type:opaquestring'), ('type', 'opaquestring'))
+ self.assertEqual(splittype('opaquestring'), (None, 'opaquestring'))
+ self.assertEqual(splittype(':opaquestring'), (None, ':opaquestring'))
+ self.assertEqual(splittype('type:'), ('type', ''))
+ self.assertEqual(splittype('type:opaque:string'), ('type', 'opaque:string'))
+
+ def test_splithost(self):
+ splithost = urllib.splithost
+ self.assertEqual(splithost('//www.example.org:80/foo/bar/baz.html'),
+ ('www.example.org:80', '/foo/bar/baz.html'))
+ self.assertEqual(splithost('//www.example.org:80'),
+ ('www.example.org:80', ''))
+ self.assertEqual(splithost('/foo/bar/baz.html'),
+ (None, '/foo/bar/baz.html'))
+
+ def test_splituser(self):
+ splituser = urllib.splituser
+ self.assertEqual(splituser('User:Pass at www.python.org:080'),
+ ('User:Pass', 'www.python.org:080'))
+ self.assertEqual(splituser('@www.python.org:080'),
+ ('', 'www.python.org:080'))
+ self.assertEqual(splituser('www.python.org:080'),
+ (None, 'www.python.org:080'))
+ self.assertEqual(splituser('User:Pass@'),
+ ('User:Pass', ''))
+ self.assertEqual(splituser('User at example.com:Pass at www.python.org:080'),
+ ('User at example.com:Pass', 'www.python.org:080'))
def test_splitpasswd(self):
- """Some of the password examples are not sensible, but it is added to
- confirming to RFC2617 and addressing issue4675.
- """
- self.assertEqual(('user', 'ab'),urllib.splitpasswd('user:ab'))
- self.assertEqual(('user', 'a\nb'),urllib.splitpasswd('user:a\nb'))
- self.assertEqual(('user', 'a\tb'),urllib.splitpasswd('user:a\tb'))
- self.assertEqual(('user', 'a\rb'),urllib.splitpasswd('user:a\rb'))
- self.assertEqual(('user', 'a\fb'),urllib.splitpasswd('user:a\fb'))
- self.assertEqual(('user', 'a\vb'),urllib.splitpasswd('user:a\vb'))
- self.assertEqual(('user', 'a:b'),urllib.splitpasswd('user:a:b'))
- self.assertEqual(('user', 'a b'),urllib.splitpasswd('user:a b'))
- self.assertEqual(('user 2', 'ab'),urllib.splitpasswd('user 2:ab'))
- self.assertEqual(('user+1', 'a+b'),urllib.splitpasswd('user+1:a+b'))
+ # Some of the password examples are not sensible, but it is added to
+ # confirming to RFC2617 and addressing issue4675.
+ splitpasswd = urllib.splitpasswd
+ self.assertEqual(splitpasswd('user:ab'), ('user', 'ab'))
+ self.assertEqual(splitpasswd('user:a\nb'), ('user', 'a\nb'))
+ self.assertEqual(splitpasswd('user:a\tb'), ('user', 'a\tb'))
+ self.assertEqual(splitpasswd('user:a\rb'), ('user', 'a\rb'))
+ self.assertEqual(splitpasswd('user:a\fb'), ('user', 'a\fb'))
+ self.assertEqual(splitpasswd('user:a\vb'), ('user', 'a\vb'))
+ self.assertEqual(splitpasswd('user:a:b'), ('user', 'a:b'))
+ self.assertEqual(splitpasswd('user:a b'), ('user', 'a b'))
+ self.assertEqual(splitpasswd('user 2:ab'), ('user 2', 'ab'))
+ self.assertEqual(splitpasswd('user+1:a+b'), ('user+1', 'a+b'))
+ self.assertEqual(splitpasswd('user:'), ('user', ''))
+ self.assertEqual(splitpasswd('user'), ('user', None))
+ self.assertEqual(splitpasswd(':ab'), ('', 'ab'))
+
+ def test_splitport(self):
+ splitport = urllib.splitport
+ self.assertEqual(splitport('parrot:88'), ('parrot', '88'))
+ self.assertEqual(splitport('parrot'), ('parrot', None))
+ self.assertEqual(splitport('parrot:'), ('parrot', None))
+ self.assertEqual(splitport('127.0.0.1'), ('127.0.0.1', None))
+ self.assertEqual(splitport('parrot:cheese'), ('parrot:cheese', None))
+ self.assertEqual(splitport('[::1]:88'), ('[::1]', '88'))
+ self.assertEqual(splitport('[::1]'), ('[::1]', None))
+ self.assertEqual(splitport(':88'), ('', '88'))
+
+ def test_splitnport(self):
+ splitnport = urllib.splitnport
+ self.assertEqual(splitnport('parrot:88'), ('parrot', 88))
+ self.assertEqual(splitnport('parrot'), ('parrot', -1))
+ self.assertEqual(splitnport('parrot', 55), ('parrot', 55))
+ self.assertEqual(splitnport('parrot:'), ('parrot', -1))
+ self.assertEqual(splitnport('parrot:', 55), ('parrot', 55))
+ self.assertEqual(splitnport('127.0.0.1'), ('127.0.0.1', -1))
+ self.assertEqual(splitnport('127.0.0.1', 55), ('127.0.0.1', 55))
+ self.assertEqual(splitnport('parrot:cheese'), ('parrot', None))
+ self.assertEqual(splitnport('parrot:cheese', 55), ('parrot', None))
+
+ def test_splitquery(self):
+ # Normal cases are exercised by other tests; ensure that we also
+ # catch cases with no port specified (testcase ensuring coverage)
+ splitquery = urllib.splitquery
+ self.assertEqual(splitquery('http://python.org/fake?foo=bar'),
+ ('http://python.org/fake', 'foo=bar'))
+ self.assertEqual(splitquery('http://python.org/fake?foo=bar?'),
+ ('http://python.org/fake?foo=bar', ''))
+ self.assertEqual(splitquery('http://python.org/fake'),
+ ('http://python.org/fake', None))
+ self.assertEqual(splitquery('?foo=bar'), ('', 'foo=bar'))
+
+ def test_splittag(self):
+ splittag = urllib.splittag
+ self.assertEqual(splittag('http://example.com?foo=bar#baz'),
+ ('http://example.com?foo=bar', 'baz'))
+ self.assertEqual(splittag('http://example.com?foo=bar#'),
+ ('http://example.com?foo=bar', ''))
+ self.assertEqual(splittag('#baz'), ('', 'baz'))
+ self.assertEqual(splittag('http://example.com?foo=bar'),
+ ('http://example.com?foo=bar', None))
+ self.assertEqual(splittag('http://example.com?foo=bar#baz#boo'),
+ ('http://example.com?foo=bar#baz', 'boo'))
+
+ def test_splitattr(self):
+ splitattr = urllib.splitattr
+ self.assertEqual(splitattr('/path;attr1=value1;attr2=value2'),
+ ('/path', ['attr1=value1', 'attr2=value2']))
+ self.assertEqual(splitattr('/path;'), ('/path', ['']))
+ self.assertEqual(splitattr(';attr1=value1;attr2=value2'),
+ ('', ['attr1=value1', 'attr2=value2']))
+ self.assertEqual(splitattr('/path'), ('/path', []))
+
+ def test_splitvalue(self):
+ # Normal cases are exercised by other tests; test pathological cases
+ # with no key/value pairs. (testcase ensuring coverage)
+ splitvalue = urllib.splitvalue
+ self.assertEqual(splitvalue('foo=bar'), ('foo', 'bar'))
+ self.assertEqual(splitvalue('foo='), ('foo', ''))
+ self.assertEqual(splitvalue('=bar'), ('', 'bar'))
+ self.assertEqual(splitvalue('foobar'), ('foobar', None))
+ self.assertEqual(splitvalue('foo=bar=baz'), ('foo', 'bar=baz'))
+
+ def test_toBytes(self):
+ result = urllib.toBytes(u'http://www.python.org')
+ self.assertEqual(result, 'http://www.python.org')
+ self.assertRaises(UnicodeError, urllib.toBytes,
+ test_support.u(r'http://www.python.org/medi\u00e6val'))
+
+ def test_unwrap(self):
+ url = urllib.unwrap('<URL:type://host/path>')
+ self.assertEqual(url, 'type://host/path')
class URLopener_Tests(unittest.TestCase):
@@ -812,7 +1008,7 @@
# Everywhere else they work ok, but on those machines, sometimes
# fail in one of the tests, sometimes in other. I have a linux, and
# the tests go ok.
-# If anybody has one of the problematic enviroments, please help!
+# If anybody has one of the problematic environments, please help!
# . Facundo
#
# def server(evt):
@@ -858,7 +1054,7 @@
# def testTimeoutNone(self):
# # global default timeout is ignored
# import socket
-# self.assertTrue(socket.getdefaulttimeout() is None)
+# self.assertIsNone(socket.getdefaulttimeout())
# socket.setdefaulttimeout(30)
# try:
# ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9093, [])
@@ -870,7 +1066,7 @@
# def testTimeoutDefault(self):
# # global default timeout is used
# import socket
-# self.assertTrue(socket.getdefaulttimeout() is None)
+# self.assertIsNone(socket.getdefaulttimeout())
# socket.setdefaulttimeout(30)
# try:
# ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9093, [])
@@ -904,6 +1100,8 @@
Pathname_Tests,
Utility_Tests,
URLopener_Tests,
+ ProxyTests,
+ ProxyTests_withOrderedEnv,
#FTPWrapperTests,
)
diff --git a/lib-python/2.7/test/test_urllib2.py b/lib-python/2.7/test/test_urllib2.py
--- a/lib-python/2.7/test/test_urllib2.py
+++ b/lib-python/2.7/test/test_urllib2.py
@@ -1,12 +1,19 @@
import unittest
from test import test_support
+from test import test_urllib
import os
import socket
import StringIO
import urllib2
-from urllib2 import Request, OpenerDirector
+from urllib2 import Request, OpenerDirector, AbstractDigestAuthHandler
+import httplib
+
+try:
+ import ssl
+except ImportError:
+ ssl = None
# XXX
# Request
@@ -20,7 +27,7 @@
self.assertRaises(ValueError, urllib2.urlopen, 'bogus url')
# XXX Name hacking to get this to work on Windows.
- fname = os.path.abspath(urllib2.__file__).replace('\\', '/')
+ fname = os.path.abspath(urllib2.__file__).replace(os.sep, '/')
# And more hacking to get it to work on MacOS. This assumes
# urllib.pathname2url works, unfortunately...
@@ -47,6 +54,14 @@
for string, list in tests:
self.assertEqual(urllib2.parse_http_list(string), list)
+ @unittest.skipUnless(ssl, "ssl module required")
+ def test_cafile_and_context(self):
+ context = ssl.create_default_context()
+ with self.assertRaises(ValueError):
+ urllib2.urlopen(
+ "https://localhost", cafile="/nonexistent/path", context=context
+ )
+
def test_request_headers_dict():
"""
@@ -405,7 +420,7 @@
self._count = 0
self.requests = []
def http_open(self, req):
- import mimetools, httplib, copy
+ import mimetools, copy
from StringIO import StringIO
self.requests.append(copy.deepcopy(req))
if self._count == 0:
@@ -591,8 +606,8 @@
self.assertIsInstance(args[0], Request)
# response from opener.open is None, because there's no
# handler that defines http_open to handle it
- self.assertTrue(args[1] is None or
- isinstance(args[1], MockResponse))
+ if args[1] is not None:
+ self.assertIsInstance(args[1], MockResponse)
def sanepathname2url(path):
@@ -924,7 +939,8 @@
MockHeaders({"location": to_url}))
except urllib2.HTTPError:
# 307 in response to POST requires user OK
- self.assertTrue(code == 307 and data is not None)
+ self.assertEqual(code, 307)
+ self.assertIsNotNone(data)
self.assertEqual(o.req.get_full_url(), to_url)
try:
self.assertEqual(o.req.get_method(), "GET")
@@ -1022,6 +1038,22 @@
fp = o.open('http://www.example.com')
self.assertEqual(fp.geturl(), redirected_url.strip())
+ def test_redirect_no_path(self):
+ # Issue 14132: Relative redirect strips original path
+ real_class = httplib.HTTPConnection
+ response1 = b"HTTP/1.1 302 Found\r\nLocation: ?query\r\n\r\n"
+ httplib.HTTPConnection = test_urllib.fakehttp(response1)
+ self.addCleanup(setattr, httplib, "HTTPConnection", real_class)
+ urls = iter(("/path", "/path?query"))
+ def request(conn, method, url, *pos, **kw):
+ self.assertEqual(url, next(urls))
+ real_class.request(conn, method, url, *pos, **kw)
+ # Change response for subsequent connection
+ conn.__class__.fakedata = b"HTTP/1.1 200 OK\r\n\r\nHello!"
+ httplib.HTTPConnection.request = request
+ fp = urllib2.urlopen("http://python.org/path")
+ self.assertEqual(fp.geturl(), "http://python.org/path?query")
+
def test_proxy(self):
o = OpenerDirector()
ph = urllib2.ProxyHandler(dict(http="proxy.example.com:3128"))
@@ -1276,6 +1308,16 @@
else:
self.assertTrue(False)
+ def test_unsupported_algorithm(self):
+ handler = AbstractDigestAuthHandler()
+ with self.assertRaises(ValueError) as exc:
+ handler.get_algorithm_impls('invalid')
+ self.assertEqual(
+ str(exc.exception),
+ "Unsupported digest authentication algorithm 'invalid'"
+ )
+
+
class RequestTests(unittest.TestCase):
def setUp(self):
@@ -1336,6 +1378,11 @@
req = Request(url)
self.assertEqual(req.get_full_url(), url)
+ def test_private_attributes(self):
+ self.assertFalse(hasattr(self.get, '_Request__r_xxx'))
+ # Issue #6500: infinite recursion
+ self.assertFalse(hasattr(self.get, '_Request__r_method'))
+
def test_HTTPError_interface(self):
"""
Issue 13211 reveals that HTTPError didn't implement the URLError
diff --git a/lib-python/2.7/urllib.py b/lib-python/2.7/urllib.py
--- a/lib-python/2.7/urllib.py
+++ b/lib-python/2.7/urllib.py
@@ -28,6 +28,7 @@
import time
import sys
import base64
+import re
from urlparse import urljoin as basejoin
@@ -68,15 +69,15 @@
# Shortcut for basic usage
_urlopener = None
-def urlopen(url, data=None, proxies=None):
+def urlopen(url, data=None, proxies=None, context=None):
"""Create a file-like object for the specified URL to read from."""
from warnings import warnpy3k
warnpy3k("urllib.urlopen() has been removed in Python 3.0 in "
"favor of urllib2.urlopen()", stacklevel=2)
global _urlopener
- if proxies is not None:
- opener = FancyURLopener(proxies=proxies)
+ if proxies is not None or context is not None:
+ opener = FancyURLopener(proxies=proxies, context=context)
elif not _urlopener:
opener = FancyURLopener()
_urlopener = opener
@@ -86,11 +87,15 @@
return opener.open(url)
else:
return opener.open(url, data)
-def urlretrieve(url, filename=None, reporthook=None, data=None):
+def urlretrieve(url, filename=None, reporthook=None, data=None, context=None):
global _urlopener
- if not _urlopener:
- _urlopener = FancyURLopener()
- return _urlopener.retrieve(url, filename, reporthook, data)
+ if context is not None:
+ opener = FancyURLopener(context=context)
+ elif not _urlopener:
+ _urlopener = opener = FancyURLopener()
+ else:
+ opener = _urlopener
+ return opener.retrieve(url, filename, reporthook, data)
def urlcleanup():
if _urlopener:
_urlopener.cleanup()
@@ -125,13 +130,14 @@
version = "Python-urllib/%s" % __version__
# Constructor
- def __init__(self, proxies=None, **x509):
+ def __init__(self, proxies=None, context=None, **x509):
if proxies is None:
proxies = getproxies()
assert hasattr(proxies, 'has_key'), "proxies must be a mapping"
self.proxies = proxies
self.key_file = x509.get('key_file')
self.cert_file = x509.get('cert_file')
+ self.context = context
self.addheaders = [('User-Agent', self.version)]
self.__tempfiles = []
self.__unlink = os.unlink # See cleanup()
@@ -421,7 +427,8 @@
auth = None
h = httplib.HTTPS(host, 0,
key_file=self.key_file,
- cert_file=self.cert_file)
+ cert_file=self.cert_file,
+ context=self.context)
if data is not None:
h.putrequest('POST', selector)
h.putheader('Content-Type',
@@ -622,18 +629,20 @@
def http_error_302(self, url, fp, errcode, errmsg, headers, data=None):
"""Error 302 -- relocated (temporarily)."""
self.tries += 1
- if self.maxtries and self.tries >= self.maxtries:
- if hasattr(self, "http_error_500"):
- meth = self.http_error_500
- else:
- meth = self.http_error_default
+ try:
+ if self.maxtries and self.tries >= self.maxtries:
+ if hasattr(self, "http_error_500"):
+ meth = self.http_error_500
+ else:
+ meth = self.http_error_default
+ return meth(url, fp, 500,
+ "Internal Server Error: Redirect Recursion",
+ headers)
+ result = self.redirect_internal(url, fp, errcode, errmsg,
+ headers, data)
+ return result
+ finally:
self.tries = 0
- return meth(url, fp, 500,
- "Internal Server Error: Redirect Recursion", headers)
- result = self.redirect_internal(url, fp, errcode, errmsg, headers,
- data)
- self.tries = 0
- return result
def redirect_internal(self, url, fp, errcode, errmsg, headers, data):
if 'location' in headers:
@@ -818,7 +827,10 @@
"""Return the IP address of the current host."""
global _thishost
if _thishost is None:
- _thishost = socket.gethostbyname(socket.gethostname())
+ try:
+ _thishost = socket.gethostbyname(socket.gethostname())
+ except socket.gaierror:
+ _thishost = socket.gethostbyname('localhost')
return _thishost
_ftperrors = None
@@ -861,7 +873,11 @@
self.timeout = timeout
self.refcount = 0
self.keepalive = persistent
- self.init()
+ try:
+ self.init()
+ except:
+ self.close()
+ raise
def init(self):
import ftplib
@@ -869,8 +885,8 @@
self.ftp = ftplib.FTP()
self.ftp.connect(self.host, self.port, self.timeout)
self.ftp.login(self.user, self.passwd)
- for dir in self.dirs:
- self.ftp.cwd(dir)
+ _target = '/'.join(self.dirs)
+ self.ftp.cwd(_target)
def retrfile(self, file, type):
import ftplib
@@ -916,13 +932,7 @@
return (ftpobj, retrlen)
def endtransfer(self):
- if not self.busy:
- return
self.busy = 0
- try:
- self.ftp.voidresp()
- except ftperrors():
- pass
def close(self):
self.keepalive = False
@@ -980,11 +990,16 @@
self.hookargs = hookargs
def close(self):
- if self.closehook:
- self.closehook(*self.hookargs)
- self.closehook = None
- self.hookargs = None
- addbase.close(self)
+ try:
+ closehook = self.closehook
+ hookargs = self.hookargs
+ if closehook:
+ self.closehook = None
+ self.hookargs = None
+ closehook(*hookargs)
+ finally:
+ addbase.close(self)
+
class addinfo(addbase):
"""class to add an info() method to an open file."""
@@ -1121,10 +1136,13 @@
global _portprog
if _portprog is None:
import re
- _portprog = re.compile('^(.*):([0-9]+)$')
+ _portprog = re.compile('^(.*):([0-9]*)$')
match = _portprog.match(host)
- if match: return match.group(1, 2)
+ if match:
+ host, port = match.groups()
+ if port:
+ return host, port
return host, None
_nportprog = None
@@ -1141,12 +1159,12 @@
match = _nportprog.match(host)
if match:
host, port = match.group(1, 2)
- try:
- if not port: raise ValueError, "no digits"
- nport = int(port)
- except ValueError:
- nport = None
- return host, nport
+ if port:
+ try:
+ nport = int(port)
+ except ValueError:
+ nport = None
+ return host, nport
return host, defport
_queryprog = None
@@ -1198,22 +1216,35 @@
_hexdig = '0123456789ABCDEFabcdef'
_hextochr = dict((a + b, chr(int(a + b, 16)))
for a in _hexdig for b in _hexdig)
+_asciire = re.compile('([\x00-\x7f]+)')
def unquote(s):
"""unquote('abc%20def') -> 'abc def'."""
- res = s.split('%')
+ if _is_unicode(s):
+ if '%' not in s:
+ return s
+ bits = _asciire.split(s)
+ res = [bits[0]]
+ append = res.append
+ for i in range(1, len(bits), 2):
+ append(unquote(str(bits[i])).decode('latin1'))
+ append(bits[i + 1])
+ return ''.join(res)
+
+ bits = s.split('%')
# fastpath
- if len(res) == 1:
+ if len(bits) == 1:
return s
- s = res[0]
- for item in res[1:]:
+ res = [bits[0]]
+ append = res.append
+ for item in bits[1:]:
try:
- s += _hextochr[item[:2]] + item[2:]
+ append(_hextochr[item[:2]])
+ append(item[2:])
except KeyError:
- s += '%' + item
- except UnicodeDecodeError:
- s += unichr(int(item[:2], 16)) + item[2:]
- return s
+ append('%')
+ append(item)
+ return ''.join(res)
def unquote_plus(s):
"""unquote('%7e/abc+def') -> '~/abc def'"""
@@ -1342,25 +1373,51 @@
"""Return a dictionary of scheme -> proxy server URL mappings.
Scan the environment for variables named <scheme>_proxy;
- this seems to be the standard convention. If you need a
- different way, you can pass a proxies dictionary to the
+ this seems to be the standard convention. In order to prefer lowercase
+ variables, we process the environment in two passes, first matches any
+ and second matches only lower case proxies.
+
+ If you need a different way, you can pass a proxies dictionary to the
[Fancy]URLopener constructor.
-
"""
+ # Get all variables
proxies = {}
for name, value in os.environ.items():
name = name.lower()
if value and name[-6:] == '_proxy':
proxies[name[:-6]] = value
+
+ # CVE-2016-1000110 - If we are running as CGI script, forget HTTP_PROXY
+ # (non-all-lowercase) as it may be set from the web server by a "Proxy:"
+ # header from the client
+ # If "proxy" is lowercase, it will still be used thanks to the next block
+ if 'REQUEST_METHOD' in os.environ:
+ proxies.pop('http', None)
+
+ # Get lowercase variables
+ for name, value in os.environ.items():
+ if name[-6:] == '_proxy':
+ name = name.lower()
+ if value:
+ proxies[name[:-6]] = value
+ else:
+ proxies.pop(name[:-6], None)
+
return proxies
-def proxy_bypass_environment(host):
+def proxy_bypass_environment(host, proxies=None):
"""Test if proxies should not be used for a particular host.
- Checks the environment for a variable named no_proxy, which should
- be a list of DNS suffixes separated by commas, or '*' for all hosts.
+ Checks the proxies dict for the value of no_proxy, which should be a
+ list of comma separated DNS suffixes, or '*' for all hosts.
"""
- no_proxy = os.environ.get('no_proxy', '') or os.environ.get('NO_PROXY', '')
+ if proxies is None:
+ proxies = getproxies_environment()
+ # don't bypass, if no_proxy isn't specified
+ try:
+ no_proxy = proxies['no']
+ except KeyError:
+ return 0
# '*' is special case for always bypass
if no_proxy == '*':
return 1
@@ -1369,8 +1426,12 @@
# check if the host ends with any of the DNS suffixes
no_proxy_list = [proxy.strip() for proxy in no_proxy.split(',')]
for name in no_proxy_list:
- if name and (hostonly.endswith(name) or host.endswith(name)):
- return 1
+ if name:
+ name = re.escape(name)
+ pattern = r'(.+\.)?%s$' % name
+ if (re.match(pattern, hostonly, re.I)
+ or re.match(pattern, host, re.I)):
+ return 1
# otherwise, don't bypass
return 0
@@ -1446,8 +1507,14 @@
return _get_proxies()
def proxy_bypass(host):
- if getproxies_environment():
- return proxy_bypass_environment(host)
+ """Return True, if a host should be bypassed.
+
+ Checks proxy settings gathered from the environment, if specified, or
+ from the MacOSX framework SystemConfiguration.
+ """
+ proxies = getproxies_environment()
+ if proxies:
+ return proxy_bypass_environment(host, proxies)
else:
return proxy_bypass_macosx_sysconf(host)
@@ -1563,14 +1630,14 @@
return 0
def proxy_bypass(host):
- """Return a dictionary of scheme -> proxy server URL mappings.
+ """Return True, if the host should be bypassed.
- Returns settings gathered from the environment, if specified,
+ Checks proxy settings gathered from the environment, if specified,
or the registry.
-
"""
- if getproxies_environment():
- return proxy_bypass_environment(host)
+ proxies = getproxies_environment()
+ if proxies:
+ return proxy_bypass_environment(host, proxies)
else:
return proxy_bypass_registry(host)
diff --git a/lib-python/2.7/urllib2.py b/lib-python/2.7/urllib2.py
--- a/lib-python/2.7/urllib2.py
+++ b/lib-python/2.7/urllib2.py
@@ -109,6 +109,14 @@
except ImportError:
from StringIO import StringIO
+# check for SSL
+try:
+ import ssl
+except ImportError:
+ _have_ssl = False
+else:
+ _have_ssl = True
+
from urllib import (unwrap, unquote, splittype, splithost, quote,
addinfourl, splitport, splittag, toBytes,
splitattr, ftpwrapper, splituser, splitpasswd, splitvalue)
@@ -120,11 +128,30 @@
__version__ = sys.version[:3]
_opener = None
-def urlopen(url, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
+def urlopen(url, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
+ cafile=None, capath=None, cadefault=False, context=None):
global _opener
- if _opener is None:
- _opener = build_opener()
- return _opener.open(url, data, timeout)
+ if cafile or capath or cadefault:
+ if context is not None:
+ raise ValueError(
+ "You can't pass both context and any of cafile, capath, and "
+ "cadefault"
+ )
+ if not _have_ssl:
+ raise ValueError('SSL support not available')
+ context = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH,
+ cafile=cafile,
+ capath=capath)
+ https_handler = HTTPSHandler(context=context)
+ opener = build_opener(https_handler)
+ elif context:
+ https_handler = HTTPSHandler(context=context)
+ opener = build_opener(https_handler)
+ elif _opener is None:
+ _opener = opener = build_opener()
+ else:
+ opener = _opener
+ return opener.open(url, data, timeout)
def install_opener(opener):
global _opener
@@ -221,11 +248,9 @@
# methods getting called in a non-standard order. this may be
# too complicated and/or unnecessary.
# XXX should the __r_XXX attributes be public?
- if attr[:12] == '_Request__r_':
- name = attr[12:]
- if hasattr(Request, 'get_' + name):
- getattr(self, 'get_' + name)()
- return getattr(self, attr)
+ if attr in ('_Request__r_type', '_Request__r_host'):
+ getattr(self, 'get_' + attr[12:])()
+ return self.__dict__[attr]
raise AttributeError, attr
def get_method(self):
@@ -584,7 +609,7 @@
# fix a possible malformed URL
urlparts = urlparse.urlparse(newurl)
- if not urlparts.path:
+ if not urlparts.path and urlparts.netloc:
urlparts = list(urlparts)
urlparts[2] = "/"
newurl = urlparse.urlunparse(urlparts)
@@ -843,10 +868,7 @@
password_mgr = HTTPPasswordMgr()
self.passwd = password_mgr
self.add_password = self.passwd.add_password
- self.retried = 0
- def reset_retry_count(self):
- self.retried = 0
def http_error_auth_reqed(self, authreq, host, req, headers):
# host may be an authority (without userinfo) or a URL with an
@@ -854,13 +876,6 @@
# XXX could be multiple headers
authreq = headers.get(authreq, None)
- if self.retried > 5:
- # retry sending the username:password 5 times before failing.
- raise HTTPError(req.get_full_url(), 401, "basic auth failed",
- headers, None)
- else:
- self.retried += 1
-
if authreq:
mo = AbstractBasicAuthHandler.rx.search(authreq)
if mo:
@@ -869,17 +884,14 @@
warnings.warn("Basic Auth Realm was unquoted",
UserWarning, 2)
if scheme.lower() == 'basic':
- response = self.retry_http_basic_auth(host, req, realm)
- if response and response.code != 401:
- self.retried = 0
- return response
+ return self.retry_http_basic_auth(host, req, realm)
def retry_http_basic_auth(self, host, req, realm):
user, pw = self.passwd.find_user_password(realm, host)
if pw is not None:
raw = "%s:%s" % (user, pw)
auth = 'Basic %s' % base64.b64encode(raw).strip()
- if req.headers.get(self.auth_header, None) == auth:
+ if req.get_header(self.auth_header, None) == auth:
return None
req.add_unredirected_header(self.auth_header, auth)
return self.parent.open(req, timeout=req.timeout)
@@ -895,7 +907,6 @@
url = req.get_full_url()
response = self.http_error_auth_reqed('www-authenticate',
url, req, headers)
- self.reset_retry_count()
return response
@@ -911,7 +922,6 @@
authority = req.get_host()
response = self.http_error_auth_reqed('proxy-authenticate',
authority, req, headers)
- self.reset_retry_count()
return response
@@ -1061,6 +1071,9 @@
elif algorithm == 'SHA':
H = lambda x: hashlib.sha1(x).hexdigest()
# XXX MD5-sess
+ else:
+ raise ValueError("Unsupported digest authentication "
+ "algorithm %r" % algorithm.lower())
KD = lambda s, d: H("%s:%s" % (s, d))
return H, KD
@@ -1136,7 +1149,7 @@
return request
- def do_open(self, http_class, req):
+ def do_open(self, http_class, req, **http_conn_args):
"""Return an addinfourl object for the request, using http_class.
http_class must implement the HTTPConnection API from httplib.
@@ -1150,7 +1163,8 @@
if not host:
raise URLError('no host given')
- h = http_class(host, timeout=req.timeout) # will parse host:port
+ # will parse host:port
+ h = http_class(host, timeout=req.timeout, **http_conn_args)
h.set_debuglevel(self._debuglevel)
headers = dict(req.unredirected_hdrs)
@@ -1218,8 +1232,13 @@
if hasattr(httplib, 'HTTPS'):
class HTTPSHandler(AbstractHTTPHandler):
+ def __init__(self, debuglevel=0, context=None):
+ AbstractHTTPHandler.__init__(self, debuglevel)
+ self._context = context
+
def https_open(self, req):
- return self.do_open(httplib.HTTPSConnection, req)
+ return self.do_open(httplib.HTTPSConnection, req,
+ context=self._context)
https_request = AbstractHTTPHandler.do_request_
--
Repository URL: https://hg.python.org/jython
More information about the Jython-checkins
mailing list