Skip to content

Commit b7f27ae

Browse files
kyrofakyrofa
committed
Support authentication
`vcs_base.load_url()` currently doesn't support authentication. Add support for both basic and token-based authentication by parsing netrc-formatted files. Use `appdirs` to support vcstool-specific authentication files for both the user and the system (user takes precedence). Signed-off-by: Kyle Fazzari <[email protected]>
1 parent bb9a810 commit b7f27ae

File tree

3 files changed

+370
-2
lines changed

3 files changed

+370
-2
lines changed

‎.travis.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ matrix:
1212
install:
1313
# newer versions of PyYAML dropped support for Python 3.4
1414
- if [ $TRAVIS_PYTHON_VERSION == "3.4" ]; then pip install PyYAML==5.2; fi
15-
- pip install coverage flake8 flake8-docstrings flake8-import-order pytest PyYAML
15+
- pip install appdirs coverage flake8 flake8-docstrings flake8-import-order pytest PyYAML mock
1616
script:
1717
- PYTHONPATH=`pwd` pytest -s -v test
1818
notifications:

‎test/test_base.py‎

Lines changed: 252 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,252 @@
1+
import os
2+
import shutil
3+
import tempfile
4+
import textwrap
5+
import unittest
6+
7+
try:
8+
from urllib.error import HTTPError
9+
except ImportError:
10+
from urllib2 import HTTPError
11+
12+
13+
try:
14+
from unittest import mock
15+
except ImportError:
16+
import mock
17+
18+
from vcstool.clients import vcs_base
19+
20+
21+
class TestBase(unittest.TestCase):
22+
23+
def setUp(self):
24+
self.default_auth_dir = tempfile.mkdtemp()
25+
self.addCleanup(shutil.rmtree, self.default_auth_dir)
26+
self.user_auth_dir = tempfile.mkdtemp()
27+
self.addCleanup(shutil.rmtree, self.user_auth_dir)
28+
self.system_auth_dir = tempfile.mkdtemp()
29+
self.addCleanup(shutil.rmtree, self.system_auth_dir)
30+
31+
self._previous_home = os.getenv("HOME")
32+
os.environ["HOME"] = self.default_auth_dir
33+
34+
patcher = mock.patch(
35+
'vcstool.clients.vcs_base.appdirs.user_config_dir',
36+
return_value=self.user_auth_dir)
37+
patcher.start()
38+
self.addCleanup(patcher.stop)
39+
40+
patcher = mock.patch(
41+
'vcstool.clients.vcs_base.appdirs.site_config_dir',
42+
return_value=self.system_auth_dir)
43+
patcher.start()
44+
self.addCleanup(patcher.stop)
45+
46+
def tearDown(self):
47+
if self._previous_home:
48+
os.environ["HOME"] = self._previous_home
49+
else:
50+
del os.environ["HOME"]
51+
52+
@mock.patch('vcstool.clients.vcs_base.urlopen', autospec=True)
53+
@mock.patch(
54+
'vcstool.clients.vcs_base._authenticated_urlopen', autospec=True)
55+
def test_load_url_calls_urlopen(
56+
self, authenticated_urlopen_mock, urlopen_mock):
57+
urlopen_read_mock = urlopen_mock.return_value.read
58+
59+
vcs_base.load_url('example.com', timeout=123)
60+
61+
urlopen_mock.assert_called_once_with('example.com', timeout=123)
62+
urlopen_read_mock.assert_called_once_with()
63+
self.assertFalse(authenticated_urlopen_mock.mock_calls)
64+
65+
@mock.patch('vcstool.clients.vcs_base.urlopen', autospec=True)
66+
@mock.patch(
67+
'vcstool.clients.vcs_base._authenticated_urlopen', autospec=True)
68+
def test_load_url_calls_authenticated_urlopen(
69+
self, authenticated_urlopen_mock, urlopen_mock):
70+
for code in (401, 404):
71+
urlopen_mock.side_effect = [
72+
HTTPError(None, code, None, None, None)]
73+
urlopen_read_mock = urlopen_mock.return_value.read
74+
75+
vcs_base.load_url('example.com', timeout=123)
76+
77+
urlopen_mock.assert_called_once_with('example.com', timeout=123)
78+
self.assertFalse(urlopen_read_mock.mock_calls)
79+
80+
authenticated_urlopen_mock.assert_called_once_with(
81+
'example.com', timeout=123)
82+
83+
authenticated_urlopen_mock.reset_mock()
84+
urlopen_mock.reset_mock()
85+
86+
def test_netrc_open_no_such_file(self):
87+
try:
88+
self.assertEqual(vcs_base._authenticated_urlopen(
89+
'https://example.com'), None)
90+
except Exception:
91+
self.fail(
92+
'The lack of a .netrc file should not result in an exception')
93+
94+
def test_netrc_file_precedence(self):
95+
machine = 'example.com'
96+
97+
default_auth_file_path = os.path.join(self.default_auth_dir, '.netrc')
98+
user_auth_file_path = os.path.join(
99+
self.user_auth_dir, vcs_base._AUTHENTICATION_CONFIGURATION_FILE)
100+
system_auth_file_path = os.path.join(
101+
self.system_auth_dir, vcs_base._AUTHENTICATION_CONFIGURATION_FILE)
102+
103+
for path in (
104+
default_auth_file_path, user_auth_file_path,
105+
system_auth_file_path):
106+
_create_netrc_file(path, textwrap.dedent('''\
107+
machine %s
108+
password %s
109+
''' % (machine, path)))
110+
111+
credentials = vcs_base._credentials_for_machine(machine)
112+
self.assertIsNotNone(credentials)
113+
self.assertEqual(len(credentials), 3)
114+
self.assertEqual(credentials[2], default_auth_file_path)
115+
116+
# Remove default auth file and assert that the user auth file is used
117+
os.unlink(default_auth_file_path)
118+
credentials = vcs_base._credentials_for_machine(machine)
119+
self.assertIsNotNone(credentials)
120+
self.assertEqual(len(credentials), 3)
121+
self.assertEqual(credentials[2], user_auth_file_path)
122+
123+
# Remove user auth file and assert that the system auth file is used
124+
os.unlink(user_auth_file_path)
125+
credentials = vcs_base._credentials_for_machine(machine)
126+
self.assertIsNotNone(credentials)
127+
self.assertEqual(len(credentials), 3)
128+
self.assertEqual(credentials[2], system_auth_file_path)
129+
130+
# Remove system auth file and assert that no creds are found
131+
os.unlink(system_auth_file_path)
132+
self.assertIsNone(vcs_base._credentials_for_machine(machine))
133+
134+
def test_netrc_file_skip_errors(self):
135+
machine = 'example.com'
136+
137+
default_auth_file_path = os.path.join(self.default_auth_dir, '.netrc')
138+
user_auth_file_path = os.path.join(
139+
self.user_auth_dir, vcs_base._AUTHENTICATION_CONFIGURATION_FILE)
140+
141+
_create_netrc_file(default_auth_file_path, 'skip-me-invalid')
142+
143+
_create_netrc_file(user_auth_file_path, textwrap.dedent('''\
144+
machine %s
145+
password %s
146+
''' % (machine, user_auth_file_path)))
147+
148+
credentials = vcs_base._credentials_for_machine(machine)
149+
self.assertIsNotNone(credentials)
150+
self.assertEqual(len(credentials), 3)
151+
self.assertEqual(credentials[2], user_auth_file_path)
152+
153+
def test_auth_parts(self):
154+
user_auth_file_path = os.path.join(
155+
self.user_auth_dir, vcs_base._AUTHENTICATION_CONFIGURATION_FILE)
156+
user_auth_file_part_path = os.path.join(
157+
self.user_auth_dir,
158+
vcs_base._AUTHENTICATION_CONFIGURATION_PARTS_DIR, 'test.conf')
159+
os.makedirs(os.path.dirname(user_auth_file_part_path))
160+
161+
auth_machine = 'auth.example.com'
162+
parts_machine = 'parts.example.com'
163+
164+
for path in (user_auth_file_path, user_auth_file_part_path):
165+
_create_netrc_file(path, textwrap.dedent('''\
166+
machine %s
167+
password %s
168+
''' % (auth_machine, path)))
169+
with open(user_auth_file_part_path, 'a') as f:
170+
f.write('machine %s\n' % parts_machine)
171+
f.write('password %s\n' % path)
172+
173+
credentials = vcs_base._credentials_for_machine(auth_machine)
174+
self.assertIsNotNone(credentials)
175+
self.assertEqual(len(credentials), 3)
176+
self.assertEqual(credentials[2], user_auth_file_path)
177+
178+
credentials = vcs_base._credentials_for_machine(parts_machine)
179+
self.assertIsNotNone(credentials)
180+
self.assertEqual(len(credentials), 3)
181+
self.assertEqual(credentials[2], user_auth_file_part_path)
182+
183+
@mock.patch('vcstool.clients.vcs_base.urlopen', autospec=True)
184+
@mock.patch('vcstool.clients.vcs_base.build_opener', autospec=True)
185+
def test_authenticated_urlopen_basic_auth(
186+
self, build_opener_mock, urlopen_mock):
187+
open_mock = build_opener_mock.return_value.open
188+
189+
machine = 'example.com'
190+
_create_netrc_file(
191+
os.path.join(self.default_auth_dir, '.netrc'),
192+
textwrap.dedent('''\
193+
machine %s
194+
login username
195+
password password
196+
''' % machine))
197+
198+
url = 'https://%s/foo/bar' % machine
199+
vcs_base._authenticated_urlopen(url)
200+
201+
self.assertFalse(urlopen_mock.mock_calls)
202+
203+
class _HTTPBasicAuthHandlerMatcher(object):
204+
def __init__(self, test):
205+
self.test = test
206+
207+
def __eq__(self, other):
208+
manager = other.passwd
209+
self.test.assertEqual(
210+
manager.find_user_password(None, 'example.com'),
211+
('username', 'password'))
212+
return True
213+
214+
build_opener_mock.assert_called_once_with(
215+
_HTTPBasicAuthHandlerMatcher(self))
216+
open_mock.assert_called_once_with(url, timeout=None)
217+
218+
@mock.patch('vcstool.clients.vcs_base.urlopen', autospec=True)
219+
@mock.patch('vcstool.clients.vcs_base.build_opener', autospec=True)
220+
def test_authenticated_urlopen_token_auth(
221+
self, build_opener_mock, urlopen_mock):
222+
machine = 'example.com'
223+
_create_netrc_file(
224+
os.path.join(self.default_auth_dir, '.netrc'),
225+
textwrap.dedent('''\
226+
machine %s
227+
password password
228+
''' % machine))
229+
230+
url = 'https://%s/foo/bar' % machine
231+
vcs_base._authenticated_urlopen(url)
232+
233+
self.assertFalse(build_opener_mock.mock_calls)
234+
235+
class _RequestMatcher(object):
236+
def __init__(self, test):
237+
self.test = test
238+
239+
def __eq__(self, other):
240+
self.test.assertEqual(other.get_full_url(), url)
241+
self.test.assertEqual(
242+
other.get_header('Private-token'), 'password')
243+
return True
244+
245+
urlopen_mock.assert_called_once_with(
246+
_RequestMatcher(self), timeout=None)
247+
248+
249+
def _create_netrc_file(path, contents):
250+
with open(path, 'w') as f:
251+
f.write(contents)
252+
os.chmod(path, 0o600)

0 commit comments

Comments
 (0)