Support .netrc by trust_env

CHANGES/2581.feature

@@ -0,0 +1 @@
+Support `.netrc` by `trust_env`

CONTRIBUTORS.txt

@@ -199,6 +199,7 @@ W. Trevor King
 Will McGugan
 Willem de Groot
 Wilson Ong
+Wei Lin
 Yannick Koechlin
 Yannick Péroux
 Yegor Roganov

aiohttp/helpers.py

@@ -7,6 +7,7 @@
 import datetime
 import functools
 import inspect
+import netrc
 import os
 import re
 import time
@@ -111,12 +112,40 @@ def strip_auth_from_url(url):
         return url.with_user(None), auth
 
 
+def netrc_from_env():
+    netrc_obj = None
+    netrc_path = os.environ.get('NETRC')
+    try:
+        if netrc_path is not None:
+            netrc_path = Path(netrc_path)
+        else:
+            home_dir = Path.home()
+            if os.name == 'nt':  # pragma: no cover
+                netrc_path = home_dir.joinpath('_netrc')
+            else:
+                netrc_path = home_dir.joinpath('.netrc')
+
+        if netrc_path and netrc_path.is_file():
+            try:
+                netrc_obj = netrc.netrc(str(netrc_path))
+            except (netrc.NetrcParseError, IOError) as e:
+                client_logger.warning(".netrc file parses fail: %s", e)
+
+        if netrc_obj is None:
+            client_logger.warning("could't find .netrc file")
+    except RuntimeError as e:  # pragma: no cover
+        """ handle error raised by pathlib """
+        client_logger.warning("could't find .netrc file: %s", e)
+    return netrc_obj
+
+
 ProxyInfo = namedtuple('ProxyInfo', 'proxy proxy_auth')
 
 
 def proxies_from_env():
     proxy_urls = {k: URL(v) for k, v in getproxies().items()
                   if k in ('http', 'https')}
+    netrc_obj = netrc_from_env()
     stripped = {k: strip_auth_from_url(v) for k, v in proxy_urls.items()}
     ret = {}
     for proto, val in stripped.items():
@@ -125,6 +154,12 @@ def proxies_from_env():
             client_logger.warning(
                 "HTTPS proxies %s are not supported, ignoring", proxy)
             continue
+        if netrc_obj and auth is None:
+            auth_from_netrc = netrc_obj.authenticators(proxy.host)
+            if auth_from_netrc is not None:
+                *logins, password = auth_from_netrc
+                auth = BasicAuth(logins[0] if logins[0] else logins[1],
+                                 password)
         ret[proto] = ProxyInfo(proxy, auth)
     return ret
 

tests/test_proxy_functional.py

@@ -1,5 +1,6 @@
 import asyncio
 import os
+import pathlib
 from unittest import mock
 
 import pytest
@@ -478,10 +479,22 @@ def _make_ssl_transport_dummy(self, rawsock, protocol, sslcontext,
         _make_ssl_transport_dummy)
 
 
+original_is_file = pathlib.Path.is_file
+
+
+def mock_is_file(self):
+    """ make real netrc file invisible in home dir """
+    if self.name in ['_netrc', '.netrc'] and self.parent == self.home():
+        return False
+    else:
+        return original_is_file(self)
+
+
 async def test_proxy_from_env_http(proxy_test_server, get_request, mocker):
     url = 'http://aiohttp.io/path'
     proxy = await proxy_test_server()
     mocker.patch.dict(os.environ, {'http_proxy': str(proxy.url)})
+    mocker.patch('pathlib.Path.is_file', mock_is_file)
 
     await get_request(url=url, trust_env=True)
 
@@ -511,11 +524,79 @@ async def test_proxy_from_env_http_with_auth(proxy_test_server,
     assert proxy.request.headers['Proxy-Authorization'] == auth.encode()
 
 
+async def test_proxy_from_env_http_with_auth_from_netrc(
+        proxy_test_server, get_request, tmpdir, mocker):
+    url = 'http://aiohttp.io/path'
+    proxy = await proxy_test_server()
+    auth = aiohttp.BasicAuth('user', 'pass')
+    netrc_file = tmpdir.join('test_netrc')
+    netrc_file_data = 'machine 127.0.0.1 login %s password %s' % (
+        auth.login, auth.password)
+    with open(str(netrc_file), 'w') as f:
+        f.write(netrc_file_data)
+    mocker.patch.dict(os.environ, {'http_proxy': str(proxy.url),
+                                   'NETRC': str(netrc_file)})
+
+    await get_request(url=url, trust_env=True)
+
+    assert len(proxy.requests_list) == 1
+    assert proxy.request.method == 'GET'
+    assert proxy.request.host == 'aiohttp.io'
+    assert proxy.request.path_qs == 'http://aiohttp.io/path'
+    assert proxy.request.headers['Proxy-Authorization'] == auth.encode()
+
+
+async def test_proxy_from_env_http_without_auth_from_netrc(
+        proxy_test_server, get_request, tmpdir, mocker):
+    url = 'http://aiohttp.io/path'
+    proxy = await proxy_test_server()
+    auth = aiohttp.BasicAuth('user', 'pass')
+    netrc_file = tmpdir.join('test_netrc')
+    netrc_file_data = 'machine 127.0.0.2 login %s password %s' % (
+        auth.login, auth.password)
+    with open(str(netrc_file), 'w') as f:
+        f.write(netrc_file_data)
+    mocker.patch.dict(os.environ, {'http_proxy': str(proxy.url),
+                                   'NETRC': str(netrc_file)})
+
+    await get_request(url=url, trust_env=True)
+
+    assert len(proxy.requests_list) == 1
+    assert proxy.request.method == 'GET'
+    assert proxy.request.host == 'aiohttp.io'
+    assert proxy.request.path_qs == 'http://aiohttp.io/path'
+    assert 'Proxy-Authorization' not in proxy.request.headers
+
+
+async def test_proxy_from_env_http_without_auth_from_wrong_netrc(
+        proxy_test_server, get_request, tmpdir, mocker):
+    url = 'http://aiohttp.io/path'
+    proxy = await proxy_test_server()
+    auth = aiohttp.BasicAuth('user', 'pass')
+    netrc_file = tmpdir.join('test_netrc')
+    invalid_data = 'machine 127.0.0.1 %s pass %s' % (
+        auth.login, auth.password)
+    with open(str(netrc_file), 'w') as f:
+        f.write(invalid_data)
+
+    mocker.patch.dict(os.environ, {'http_proxy': str(proxy.url),
+                                   'NETRC': str(netrc_file)})
+
+    await get_request(url=url, trust_env=True)
+
+    assert len(proxy.requests_list) == 1
+    assert proxy.request.method == 'GET'
+    assert proxy.request.host == 'aiohttp.io'
+    assert proxy.request.path_qs == 'http://aiohttp.io/path'
+    assert 'Proxy-Authorization' not in proxy.request.headers
+
+
 @pytest.mark.xfail
 async def xtest_proxy_from_env_https(proxy_test_server, get_request, mocker):
     url = 'https://aiohttp.io/path'
     proxy = await proxy_test_server()
     mocker.patch.dict(os.environ, {'https_proxy': str(proxy.url)})
+    mock.patch('pathlib.Path.is_file', mock_is_file)
 
     await get_request(url=url, trust_env=True)