diff --git a/.changes/next-release/enhancement-CodeArtifact-66239.json b/.changes/next-release/enhancement-CodeArtifact-66239.json new file mode 100644 index 000000000000..4916ea4d412f --- /dev/null +++ b/.changes/next-release/enhancement-CodeArtifact-66239.json @@ -0,0 +1,5 @@ +{ + "type": "enhancement", + "category": "CodeArtifact", + "description": "Updated npm login to write configuration directly to .npmrc, consistent with how other package manager integrations handle their config files." +} diff --git a/awscli/customizations/codeartifact/login.py b/awscli/customizations/codeartifact/login.py index ff8818b0ba8a..41b1a09d714c 100644 --- a/awscli/customizations/codeartifact/login.py +++ b/awscli/customizations/codeartifact/login.py @@ -2,6 +2,7 @@ import os import platform import re +import secrets import subprocess import sys from configparser import RawConfigParser @@ -460,6 +461,14 @@ def login(self, dry_run=False): commands = self.get_commands( self.repository_endpoint, self.auth_token, scope=scope ) + + if not dry_run: + repo_uri = urlsplit(self.repository_endpoint) + auth_token_key = f'//{repo_uri.netloc}{repo_uri.path}:_authToken' + self._write_npmrc_value( + auth_token_key, self.auth_token, self.get_npmrc_path() + ) + self._run_commands('npm', commands, dry_run) def _run_command(self, tool, command): @@ -488,6 +497,69 @@ def get_scope(cls, namespace): return scope + @classmethod + def get_npmrc_path(cls): + custom = os.environ.get('NPM_CONFIG_USERCONFIG') + if custom: + return os.path.expanduser(custom) + return os.path.join(os.path.expanduser('~'), '.npmrc') + + def _write_npmrc_value(self, key, value, npmrc_path): + new_entry = f'{key}={value}' + pattern = re.compile( + r'^' + re.escape(key) + r'=.*$', re.M + ) + if not os.path.isfile(npmrc_path): + self._create_npmrc_file(npmrc_path, new_entry) + else: + with open(npmrc_path) as f: + contents = f.read() + + if pattern.search(contents): + new_contents = pattern.sub(lambda _: new_entry, contents) + else: + new_contents = self._append_npmrc_entry( + contents, new_entry + ) + + dirname = os.path.dirname(npmrc_path) or '.' + fd, tmp_path = self._create_tmp_file(dirname) + + try: + with os.fdopen(fd, 'w') as f: + f.write(new_contents) + os.replace(tmp_path, npmrc_path) + except BaseException: + if os.path.exists(tmp_path): + os.unlink(tmp_path) + raise + + def _create_tmp_file(self, dirname): + for _ in range(10): + suffix = secrets.token_hex(8) + tmp_path = os.path.join(dirname, f'.npmrc.tmp.{suffix}') + try: + fd = os.open(tmp_path, + os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + return fd, tmp_path + except FileExistsError: + continue + raise RuntimeError('Unable to create temporary file for .npmrc') + + def _create_npmrc_file(self, npmrc_path, new_entry): + dirname = os.path.split(npmrc_path)[0] or '.' + os.makedirs(dirname, exist_ok=True) + with os.fdopen( + os.open(npmrc_path, os.O_WRONLY | os.O_CREAT, 0o600), 'w' + ) as f: + f.write(new_entry + '\n') + + def _append_npmrc_entry(self, contents, new_entry): + if contents.endswith('\n'): + return contents + new_entry + '\n' + else: + return contents + '\n' + new_entry + '\n' + @classmethod def get_commands(cls, endpoint, auth_token, **kwargs): commands = [] @@ -507,12 +579,6 @@ def get_commands(cls, endpoint, auth_token, **kwargs): [cls.NPM_CMD, 'config', 'set', always_auth_config, 'true'] ) - # set auth info for the repository. - auth_token_config = f'//{repo_uri.netloc}{repo_uri.path}:_authToken' - commands.append( - [cls.NPM_CMD, 'config', 'set', auth_token_config, auth_token] - ) - return commands diff --git a/tests/functional/codeartifact/test_codeartifact_login.py b/tests/functional/codeartifact/test_codeartifact_login.py index 8b0b76e33d5b..4df8d0f1ee27 100644 --- a/tests/functional/codeartifact/test_codeartifact_login.py +++ b/tests/functional/codeartifact/test_codeartifact_login.py @@ -60,12 +60,22 @@ def setUp(self): self.subprocess_check_out_mock = ( self.subprocess_check_output_patch.start() ) + + self.test_npmrc_path = self.file_creator.full_path('.npmrc') + self.get_npmrc_path_patch = mock.patch( + 'awscli.customizations.codeartifact.login.NpmLogin' + '.get_npmrc_path' + ) + self.get_npmrc_path_mock = self.get_npmrc_path_patch.start() + self.get_npmrc_path_mock.return_value = self.test_npmrc_path + self.cli_runner = CLIRunner() def tearDown(self): self.pypi_rc_path_patch.stop() self.subprocess_check_output_patch.stop() self.get_netrc_path_patch.stop() + self.get_npmrc_path_patch.stop() self.subprocess_patch.stop() self.file_creator.remove_all() @@ -208,7 +218,6 @@ def _get_npm_commands(self, **kwargs): repo_uri = urlsplit(self.endpoint) always_auth_config = f'//{repo_uri.netloc}{repo_uri.path}:always-auth' - auth_token_config = f'//{repo_uri.netloc}{repo_uri.path}:_authToken' scope = kwargs.get('scope') registry = f'{scope}:registry' if scope else 'registry' @@ -216,9 +225,6 @@ def _get_npm_commands(self, **kwargs): commands = [] commands.append([npm_cmd, 'config', 'set', registry, self.endpoint]) commands.append([npm_cmd, 'config', 'set', always_auth_config, 'true']) - commands.append( - [npm_cmd, 'config', 'set', auth_token_config, self.auth_token] - ) return commands diff --git a/tests/unit/customizations/codeartifact/test_adapter_login.py b/tests/unit/customizations/codeartifact/test_adapter_login.py index 553d977647c2..31245ce92e46 100644 --- a/tests/unit/customizations/codeartifact/test_adapter_login.py +++ b/tests/unit/customizations/codeartifact/test_adapter_login.py @@ -1010,9 +1010,7 @@ def setUp(self): self.commands.append( [self.NPM_CMD, 'config', 'set', always_auth_config, 'true'] ) - self.commands.append( - [self.NPM_CMD, 'config', 'set', auth_token_config, self.auth_token] - ) + self.auth_token_key = auth_token_config self.subprocess_utils = mock.Mock() @@ -1025,7 +1023,8 @@ def setUp(self): self.subprocess_utils, ) - def test_login(self): + @mock.patch('awscli.customizations.codeartifact.login.NpmLogin._write_npmrc_value') + def test_login(self, mock_write_npmrc): self.test_subject.login() expected_calls = [ mock.call(command, capture_output=True, check=True) @@ -1034,8 +1033,12 @@ def test_login(self): self.subprocess_utils.run.assert_has_calls( expected_calls, any_order=True ) + mock_write_npmrc.assert_called_once_with( + self.auth_token_key, self.auth_token, mock.ANY + ) - def test_login_always_auth_error_ignored(self): + @mock.patch('awscli.customizations.codeartifact.login.NpmLogin._write_npmrc_value') + def test_login_always_auth_error_ignored(self, mock_write_npmrc): """Test login ignores error for always-auth. This test is for NPM version >= 9 where the support of 'always-auth' @@ -1052,17 +1055,17 @@ def side_effect(command, capture_output, check): return mock.DEFAULT self.subprocess_utils.run.side_effect = side_effect - expected_calls = [] - - for command in self.commands: - expected_calls.append( - mock.call(command, capture_output=True, check=True) - ) self.test_subject.login() - + expected_calls = [ + mock.call(command, capture_output=True, check=True) + for command in self.commands + ] self.subprocess_utils.run.assert_has_calls( expected_calls, any_order=True ) + mock_write_npmrc.assert_called_once_with( + self.auth_token_key, self.auth_token, mock.ANY + ) def test_get_scope(self): expected_value = f'@{self.namespace}' @@ -1088,19 +1091,240 @@ def test_get_commands(self): self.endpoint, self.auth_token ) self.assertCountEqual(commands, self.commands) + for cmd in commands: + self.assertNotIn(self.auth_token, cmd) def test_get_commands_with_scope(self): commands = self.test_subject.get_commands( self.endpoint, self.auth_token, scope=self.namespace ) - self.commands[0][3] = f'{self.namespace}:registry' - self.assertCountEqual(commands, self.commands) + expected = list(self.commands) + expected[0][3] = f'{self.namespace}:registry' + self.assertCountEqual(commands, expected) def test_login_dry_run(self): self.test_subject.login(dry_run=True) self.subprocess_utils.assert_not_called() + +class TestNpmWriteNpmrcValue(unittest.TestCase): + + def setUp(self): + self.domain = 'domain' + self.domain_owner = 'domain-owner' + self.package_format = 'npm' + self.repository = 'repository' + self.auth_token = 'auth-token' + self.expiration = (datetime.now(tzlocal()) + relativedelta(hours=10) + + relativedelta(minutes=9)).replace(microsecond=0) + self.endpoint = 'https://{domain}-{domainOwner}.codeartifact.aws.' \ + 'a2z.com/{format}/{repository}/'.format( + domain=self.domain, + domainOwner=self.domain_owner, + format=self.package_format, + repository=self.repository + ) + + repo_uri = urlparse.urlsplit(self.endpoint) + self.auth_token_key = '//{}{}:_authToken'.format( + repo_uri.netloc, repo_uri.path + ) + + self.file_creator = FileCreator() + self.test_npmrc_path = self.file_creator.full_path('.npmrc') + + self.subprocess_utils = mock.Mock() + + self.test_subject = NpmLogin( + self.auth_token, self.expiration, self.endpoint, + self.domain, self.repository, self.subprocess_utils + ) + + def tearDown(self): + self.file_creator.remove_all() + + def test_creates_new_file_when_not_exists(self): + npmrc_path = os.path.join(self.file_creator.rootdir, 'subdir', '.npmrc') + self.test_subject._write_npmrc_value( + self.auth_token_key, self.auth_token, npmrc_path) + self.assertTrue(os.path.isfile(npmrc_path)) + with open(npmrc_path, 'r') as f: + contents = f.read() + self.assertEqual( + contents, '{}={}\n'.format(self.auth_token_key, self.auth_token)) + + @skip_if_windows("Unix file permissions are not supported on Windows.") + @mock.patch.object(NpmLogin, 'get_npmrc_path') + def test_login_sets_secure_permissions_on_new_file(self, mock_path): + npmrc_path = os.path.join(self.file_creator.rootdir, 'newdir', '.npmrc') + mock_path.return_value = npmrc_path + self.test_subject.login() + file_mode = os.stat(npmrc_path).st_mode + self.assertEqual(stat.S_IMODE(file_mode), 0o600) + + def test_appends_to_existing_file_without_key(self): + self.file_creator.create_file( + '.npmrc', 'registry=https://example.com/\n') + self.test_subject._write_npmrc_value(self.auth_token_key, self.auth_token, self.test_npmrc_path) + with open(self.test_npmrc_path, 'r') as f: + contents = f.read() + self.assertIn('registry=https://example.com/', contents) + self.assertIn('{}={}'.format(self.auth_token_key, self.auth_token), + contents) + + def test_replaces_existing_key(self): + self.file_creator.create_file( + '.npmrc', + 'registry=https://example.com/\n' + '{}=old-token\n'.format(self.auth_token_key) + + '//host/path/:always-auth=true\n' + ) + self.test_subject._write_npmrc_value(self.auth_token_key, 'new-token', self.test_npmrc_path) + with open(self.test_npmrc_path, 'r') as f: + contents = f.read() + self.assertIn('{}=new-token'.format(self.auth_token_key), contents) + self.assertNotIn('old-token', contents) + self.assertIn('registry=https://example.com/', contents) + self.assertIn('//host/path/:always-auth=true', contents) + + def test_preserves_other_entries(self): + self.file_creator.create_file( + '.npmrc', + '//other-host/path/:_authToken=other-token\n' + '{}=old-token\n'.format(self.auth_token_key) + ) + self.test_subject._write_npmrc_value(self.auth_token_key, 'new-token', self.test_npmrc_path) + with open(self.test_npmrc_path, 'r') as f: + contents = f.read() + self.assertIn('//other-host/path/:_authToken=other-token', contents) + self.assertIn('{}=new-token'.format(self.auth_token_key), contents) + self.assertNotIn('old-token', contents) + + @skip_if_windows("Unix file permissions are not supported on Windows.") + @mock.patch.object(NpmLogin, 'get_npmrc_path') + def test_login_adjusts_permissions_on_preexisting_file(self, mock_path): + mock_path.return_value = self.test_npmrc_path + self.file_creator.create_file('.npmrc', 'registry=https://example.com/\n') + os.chmod(self.test_npmrc_path, 0o644) + self.test_subject.login() + file_mode = os.stat(self.test_npmrc_path).st_mode + self.assertEqual(stat.S_IMODE(file_mode), 0o600) + + def test_handles_file_without_trailing_newline(self): + self.file_creator.create_file('.npmrc', 'registry=https://example.com/') + self.test_subject._write_npmrc_value(self.auth_token_key, self.auth_token, self.test_npmrc_path) + with open(self.test_npmrc_path, 'r') as f: + contents = f.read() + lines = contents.strip().split('\n') + self.assertEqual(len(lines), 2) + self.assertEqual(lines[0], 'registry=https://example.com/') + self.assertEqual( + lines[1], '{}={}'.format(self.auth_token_key, self.auth_token)) + + @mock.patch.object(NpmLogin, 'get_npmrc_path') + def test_token_not_passed_to_subprocess(self, mock_path): + mock_path.return_value = self.test_npmrc_path + self.file_creator.create_file('.npmrc', '') + self.test_subject.login() + for call in self.subprocess_utils.run.call_args_list: + command = call[0][0] if call[0] else call[1].get('command', []) + self.assertNotIn(self.auth_token, ' '.join(command)) + + @mock.patch.object(NpmLogin, 'get_npmrc_path') + def test_dry_run_does_not_write_file(self, mock_path): + npmrc_path = os.path.join(self.file_creator.rootdir, 'noexist', '.npmrc') + mock_path.return_value = npmrc_path + self.test_subject.login(dry_run=True) + self.assertFalse(os.path.exists(npmrc_path)) + + def test_special_chars_in_key_escaped(self): + npmrc_path = os.path.join(self.file_creator.rootdir, 'esctest', '.npmrc') + self.test_subject._write_npmrc_value( + self.auth_token_key, self.auth_token, npmrc_path) + with open(npmrc_path, 'r') as f: + contents = f.read() + self.assertEqual( + contents, '{}={}\n'.format(self.auth_token_key, self.auth_token)) + + def test_get_npmrc_path_respects_env_var(self): + custom_path = '/tmp/custom/.npmrc' + with mock.patch.dict(os.environ, {'NPM_CONFIG_USERCONFIG': custom_path}): + result = NpmLogin.get_npmrc_path() + self.assertEqual(result, custom_path) + + def test_get_npmrc_path_default(self): + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop('NPM_CONFIG_USERCONFIG', None) + result = NpmLogin.get_npmrc_path() + expected = os.path.join(os.path.expanduser('~'), '.npmrc') + self.assertEqual(result, expected) + + def test_atomic_write_preserves_original_on_replace_failure(self): + self.file_creator.create_file( + '.npmrc', 'registry=https://example.com/\n') + original_content = 'registry=https://example.com/\n' + with mock.patch('os.replace', side_effect=OSError('replace failed')): + with self.assertRaises(OSError): + self.test_subject._write_npmrc_value( + self.auth_token_key, self.auth_token, self.test_npmrc_path) + with open(self.test_npmrc_path, 'r') as f: + contents = f.read() + self.assertEqual(contents, original_content) + + @mock.patch('os.unlink') + def test_atomic_write_cleans_up_tmp_on_replace_failure(self, mock_unlink): + self.file_creator.create_file( + '.npmrc', 'registry=https://example.com/\n') + with mock.patch('os.replace', side_effect=OSError('replace failed')): + with self.assertRaises(OSError): + self.test_subject._write_npmrc_value( + self.auth_token_key, self.auth_token, self.test_npmrc_path) + mock_unlink.assert_called_once() + unlinked_path = mock_unlink.call_args[0][0] + self.assertIn('.npmrc.tmp.', unlinked_path) + + def test_atomic_write_no_tmp_file_left_after_success(self): + self.file_creator.create_file( + '.npmrc', 'registry=https://example.com/\n') + dirname = os.path.dirname(self.test_npmrc_path) + self.test_subject._write_npmrc_value( + self.auth_token_key, self.auth_token, self.test_npmrc_path) + tmp_files = [f for f in os.listdir(dirname) + if '.npmrc.tmp.' in f] + self.assertEqual(tmp_files, []) + + @skip_if_windows("Unix file permissions are not supported on Windows.") + def test_create_tmp_file_returns_fd_and_path(self): + dirname = self.file_creator.rootdir + fd, tmp_path = self.test_subject._create_tmp_file(dirname) + try: + self.assertTrue(tmp_path.startswith(dirname)) + self.assertIn('.npmrc.tmp.', tmp_path) + file_mode = os.stat(tmp_path).st_mode + self.assertEqual(stat.S_IMODE(file_mode), 0o600) + finally: + os.close(fd) + os.unlink(tmp_path) + + def test_create_tmp_file_retries_on_collision(self): + dirname = self.file_creator.rootdir + fd1, path1 = self.test_subject._create_tmp_file(dirname) + fd2, path2 = self.test_subject._create_tmp_file(dirname) + try: + self.assertNotEqual(path1, path2) + finally: + os.close(fd1) + os.close(fd2) + os.unlink(path1) + os.unlink(path2) + + def test_create_tmp_file_raises_after_max_retries(self): + with mock.patch('os.open', side_effect=FileExistsError('exists')): + with self.assertRaises(RuntimeError): + self.test_subject._create_tmp_file(self.file_creator.rootdir) + + class TestPipLogin(unittest.TestCase): PIP_INDEX_URL_FMT = PipLogin.PIP_INDEX_URL_FMT