import json import socket import unittest import pytest from dummyserver.testcase import HTTPDummyProxyTestCase, IPv6HTTPDummyProxyTestCase from dummyserver.server import ( DEFAULT_CA, DEFAULT_CA_BAD, get_unreachable_address) from .. import TARPIT_HOST, requires_network from urllib3._collections import HTTPHeaderDict from urllib3.poolmanager import proxy_from_url, ProxyManager from urllib3.exceptions import ( MaxRetryError, SSLError, ProxyError, ConnectTimeoutError) from urllib3.connectionpool import connection_from_url, VerifiedHTTPSConnection class TestHTTPProxyManager(HTTPDummyProxyTestCase): def setUp(self): self.http_url = 'http://%s:%d' % (self.http_host, self.http_port) self.http_url_alt = 'http://%s:%d' % (self.http_host_alt, self.http_port) self.https_url = 'https://%s:%d' % (self.https_host, self.https_port) self.https_url_alt = 'https://%s:%d' % (self.https_host_alt, self.https_port) self.proxy_url = 'http://%s:%d' % (self.proxy_host, self.proxy_port) def test_basic_proxy(self): http = proxy_from_url(self.proxy_url) self.addCleanup(http.clear) r = http.request('GET', '%s/' % self.http_url) self.assertEqual(r.status, 200) r = http.request('GET', '%s/' % self.https_url) self.assertEqual(r.status, 200) def test_nagle_proxy(self): """ Test that proxy connections do not have TCP_NODELAY turned on """ http = proxy_from_url(self.proxy_url) self.addCleanup(http.clear) hc2 = http.connection_from_host(self.http_host, self.http_port) conn = hc2._get_conn() self.addCleanup(conn.close) hc2._make_request(conn, 'GET', '/') tcp_nodelay_setting = conn.sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) self.assertEqual(tcp_nodelay_setting, 0, ("Expected TCP_NODELAY for proxies to be set " "to zero, instead was %s" % tcp_nodelay_setting)) def test_proxy_conn_fail(self): host, port = get_unreachable_address() http = proxy_from_url('http://%s:%s/' % (host, port), retries=1, timeout=0.05) self.addCleanup(http.clear) self.assertRaises(MaxRetryError, http.request, 'GET', '%s/' % self.https_url) self.assertRaises(MaxRetryError, http.request, 'GET', '%s/' % self.http_url) try: http.request('GET', '%s/' % self.http_url) self.fail("Failed to raise retry error.") except MaxRetryError as e: self.assertEqual(type(e.reason), ProxyError) def test_oldapi(self): http = ProxyManager(connection_from_url(self.proxy_url)) self.addCleanup(http.clear) r = http.request('GET', '%s/' % self.http_url) self.assertEqual(r.status, 200) r = http.request('GET', '%s/' % self.https_url) self.assertEqual(r.status, 200) def test_proxy_verified(self): http = proxy_from_url(self.proxy_url, cert_reqs='REQUIRED', ca_certs=DEFAULT_CA_BAD) self.addCleanup(http.clear) https_pool = http._new_pool('https', self.https_host, self.https_port) try: https_pool.request('GET', '/', retries=0) self.fail("Didn't raise SSL error with wrong CA") except MaxRetryError as e: self.assertIsInstance(e.reason, SSLError) self.assertTrue('certificate verify failed' in str(e.reason), "Expected 'certificate verify failed'," "instead got: %r" % e.reason) http = proxy_from_url(self.proxy_url, cert_reqs='REQUIRED', ca_certs=DEFAULT_CA) https_pool = http._new_pool('https', self.https_host, self.https_port) conn = https_pool._new_conn() self.assertEqual(conn.__class__, VerifiedHTTPSConnection) https_pool.request('GET', '/') # Should succeed without exceptions. http = proxy_from_url(self.proxy_url, cert_reqs='REQUIRED', ca_certs=DEFAULT_CA) https_fail_pool = http._new_pool('https', '127.0.0.1', self.https_port) try: https_fail_pool.request('GET', '/', retries=0) self.fail("Didn't raise SSL invalid common name") except MaxRetryError as e: self.assertIsInstance(e.reason, SSLError) self.assertTrue("doesn't match" in str(e.reason)) def test_redirect(self): http = proxy_from_url(self.proxy_url) self.addCleanup(http.clear) r = http.request('GET', '%s/redirect' % self.http_url, fields={'target': '%s/' % self.http_url}, redirect=False) self.assertEqual(r.status, 303) r = http.request('GET', '%s/redirect' % self.http_url, fields={'target': '%s/' % self.http_url}) self.assertEqual(r.status, 200) self.assertEqual(r.data, b'Dummy server!') def test_cross_host_redirect(self): http = proxy_from_url(self.proxy_url) self.addCleanup(http.clear) cross_host_location = '%s/echo?a=b' % self.http_url_alt try: http.request('GET', '%s/redirect' % self.http_url, fields={'target': cross_host_location}, timeout=1, retries=0) self.fail("We don't want to follow redirects here.") except MaxRetryError: pass r = http.request('GET', '%s/redirect' % self.http_url, fields={'target': '%s/echo?a=b' % self.http_url_alt}, timeout=1, retries=1) self.assertNotEqual(r._pool.host, self.http_host_alt) def test_cross_protocol_redirect(self): http = proxy_from_url(self.proxy_url) self.addCleanup(http.clear) cross_protocol_location = '%s/echo?a=b' % self.https_url try: http.request('GET', '%s/redirect' % self.http_url, fields={'target': cross_protocol_location}, timeout=1, retries=0) self.fail("We don't want to follow redirects here.") except MaxRetryError: pass r = http.request('GET', '%s/redirect' % self.http_url, fields={'target': '%s/echo?a=b' % self.https_url}, timeout=1, retries=1) self.assertEqual(r._pool.host, self.https_host) def test_headers(self): http = proxy_from_url(self.proxy_url, headers={'Foo': 'bar'}, proxy_headers={'Hickory': 'dickory'}) self.addCleanup(http.clear) r = http.request_encode_url('GET', '%s/headers' % self.http_url) returned_headers = json.loads(r.data.decode()) self.assertEqual(returned_headers.get('Foo'), 'bar') self.assertEqual(returned_headers.get('Hickory'), 'dickory') self.assertEqual(returned_headers.get('Host'), '%s:%s' % (self.http_host, self.http_port)) r = http.request_encode_url('GET', '%s/headers' % self.http_url_alt) returned_headers = json.loads(r.data.decode()) self.assertEqual(returned_headers.get('Foo'), 'bar') self.assertEqual(returned_headers.get('Hickory'), 'dickory') self.assertEqual(returned_headers.get('Host'), '%s:%s' % (self.http_host_alt, self.http_port)) r = http.request_encode_url('GET', '%s/headers' % self.https_url) returned_headers = json.loads(r.data.decode()) self.assertEqual(returned_headers.get('Foo'), 'bar') self.assertEqual(returned_headers.get('Hickory'), None) self.assertEqual(returned_headers.get('Host'), '%s:%s' % (self.https_host, self.https_port)) r = http.request_encode_url('GET', '%s/headers' % self.https_url_alt) returned_headers = json.loads(r.data.decode()) self.assertEqual(returned_headers.get('Foo'), 'bar') self.assertEqual(returned_headers.get('Hickory'), None) self.assertEqual(returned_headers.get('Host'), '%s:%s' % (self.https_host_alt, self.https_port)) r = http.request_encode_body('POST', '%s/headers' % self.http_url) returned_headers = json.loads(r.data.decode()) self.assertEqual(returned_headers.get('Foo'), 'bar') self.assertEqual(returned_headers.get('Hickory'), 'dickory') self.assertEqual(returned_headers.get('Host'), '%s:%s' % (self.http_host, self.http_port)) r = http.request_encode_url('GET', '%s/headers' % self.http_url, headers={'Baz': 'quux'}) returned_headers = json.loads(r.data.decode()) self.assertEqual(returned_headers.get('Foo'), None) self.assertEqual(returned_headers.get('Baz'), 'quux') self.assertEqual(returned_headers.get('Hickory'), 'dickory') self.assertEqual(returned_headers.get('Host'), '%s:%s' % (self.http_host, self.http_port)) r = http.request_encode_url('GET', '%s/headers' % self.https_url, headers={'Baz': 'quux'}) returned_headers = json.loads(r.data.decode()) self.assertEqual(returned_headers.get('Foo'), None) self.assertEqual(returned_headers.get('Baz'), 'quux') self.assertEqual(returned_headers.get('Hickory'), None) self.assertEqual(returned_headers.get('Host'), '%s:%s' % (self.https_host, self.https_port)) r = http.request_encode_body('GET', '%s/headers' % self.http_url, headers={'Baz': 'quux'}) returned_headers = json.loads(r.data.decode()) self.assertEqual(returned_headers.get('Foo'), None) self.assertEqual(returned_headers.get('Baz'), 'quux') self.assertEqual(returned_headers.get('Hickory'), 'dickory') self.assertEqual(returned_headers.get('Host'), '%s:%s' % (self.http_host, self.http_port)) r = http.request_encode_body('GET', '%s/headers' % self.https_url, headers={'Baz': 'quux'}) returned_headers = json.loads(r.data.decode()) self.assertEqual(returned_headers.get('Foo'), None) self.assertEqual(returned_headers.get('Baz'), 'quux') self.assertEqual(returned_headers.get('Hickory'), None) self.assertEqual(returned_headers.get('Host'), '%s:%s' % (self.https_host, self.https_port)) def test_headerdict(self): default_headers = HTTPHeaderDict(a='b') proxy_headers = HTTPHeaderDict() proxy_headers.add('foo', 'bar') http = proxy_from_url( self.proxy_url, headers=default_headers, proxy_headers=proxy_headers) self.addCleanup(http.clear) request_headers = HTTPHeaderDict(baz='quux') r = http.request('GET', '%s/headers' % self.http_url, headers=request_headers) returned_headers = json.loads(r.data.decode()) self.assertEqual(returned_headers.get('Foo'), 'bar') self.assertEqual(returned_headers.get('Baz'), 'quux') def test_proxy_pooling(self): http = proxy_from_url(self.proxy_url) self.addCleanup(http.clear) for x in range(2): http.urlopen('GET', self.http_url) self.assertEqual(len(http.pools), 1) for x in range(2): http.urlopen('GET', self.http_url_alt) self.assertEqual(len(http.pools), 1) for x in range(2): http.urlopen('GET', self.https_url) self.assertEqual(len(http.pools), 2) for x in range(2): http.urlopen('GET', self.https_url_alt) self.assertEqual(len(http.pools), 3) def test_proxy_pooling_ext(self): http = proxy_from_url(self.proxy_url) self.addCleanup(http.clear) hc1 = http.connection_from_url(self.http_url) hc2 = http.connection_from_host(self.http_host, self.http_port) hc3 = http.connection_from_url(self.http_url_alt) hc4 = http.connection_from_host(self.http_host_alt, self.http_port) self.assertEqual(hc1, hc2) self.assertEqual(hc2, hc3) self.assertEqual(hc3, hc4) sc1 = http.connection_from_url(self.https_url) sc2 = http.connection_from_host(self.https_host, self.https_port, scheme='https') sc3 = http.connection_from_url(self.https_url_alt) sc4 = http.connection_from_host(self.https_host_alt, self.https_port, scheme='https') self.assertEqual(sc1, sc2) self.assertNotEqual(sc2, sc3) self.assertEqual(sc3, sc4) @pytest.mark.timeout(0.5) @requires_network def test_https_proxy_timeout(self): https = proxy_from_url('https://{host}'.format(host=TARPIT_HOST)) self.addCleanup(https.clear) try: https.request('GET', self.http_url, timeout=0.001) self.fail("Failed to raise retry error.") except MaxRetryError as e: self.assertEqual(type(e.reason), ConnectTimeoutError) @pytest.mark.timeout(0.5) @requires_network def test_https_proxy_pool_timeout(self): https = proxy_from_url('https://{host}'.format(host=TARPIT_HOST), timeout=0.001) self.addCleanup(https.clear) try: https.request('GET', self.http_url) self.fail("Failed to raise retry error.") except MaxRetryError as e: self.assertEqual(type(e.reason), ConnectTimeoutError) def test_scheme_host_case_insensitive(self): """Assert that upper-case schemes and hosts are normalized.""" http = proxy_from_url(self.proxy_url.upper()) self.addCleanup(http.clear) r = http.request('GET', '%s/' % self.http_url.upper()) self.assertEqual(r.status, 200) r = http.request('GET', '%s/' % self.https_url.upper()) self.assertEqual(r.status, 200) class TestIPv6HTTPProxyManager(IPv6HTTPDummyProxyTestCase): def setUp(self): self.http_url = 'http://%s:%d' % (self.http_host, self.http_port) self.http_url_alt = 'http://%s:%d' % (self.http_host_alt, self.http_port) self.https_url = 'https://%s:%d' % (self.https_host, self.https_port) self.https_url_alt = 'https://%s:%d' % (self.https_host_alt, self.https_port) self.proxy_url = 'http://[%s]:%d' % (self.proxy_host, self.proxy_port) def test_basic_ipv6_proxy(self): http = proxy_from_url(self.proxy_url) self.addCleanup(http.clear) r = http.request('GET', '%s/' % self.http_url) self.assertEqual(r.status, 200) r = http.request('GET', '%s/' % self.https_url) self.assertEqual(r.status, 200) if __name__ == '__main__': unittest.main()