Harden credential store file permissions

azure-devops/azext_devops/dev/common/_credentials.py

@@ -60,15 +60,16 @@ def clear_credential(organization):
         else:
             raise CLIError('No credentials were found.')
     else:
-        # delete particular organization from the list
+        cred_store.clear_password(key)
+
+        # delete particular organization from the list after clearing its credential
         if os.path.isfile(ORGANIZATION_LIST_FILE):
             with open(ORGANIZATION_LIST_FILE, "r") as input_file:
                 orgs = input_file.readlines()
             with open(ORGANIZATION_LIST_FILE, "w") as output_file:
                 for line in orgs:
                     if line.rstrip() != key:
                         output_file.write(line)
-        cred_store.clear_password(key)
 
 
 def _get_service_name(organization):

azure-devops/azext_devops/dev/common/config.py

@@ -6,7 +6,6 @@
 import os
 
 from knack.config import CLIConfig, get_config_parser
-from knack.util import ensure_dir
 from .const import (AZ_DEVOPS_CONFIG_DIR_ENVKEY,
                     AZ_DEVOPS_DEFAULT_CONFIG_DIR,
                     CLI_ENV_VARIABLE_PREFIX,
@@ -16,10 +15,21 @@
 _UNSET = object()
 
 
+def _ensure_private_config_dir(config_dir):
+    try:
+        os.makedirs(config_dir, mode=0o700)
+    except OSError:
+        if not os.path.isdir(config_dir):
+            raise
+    else:
+        if os.name != 'nt':
+            os.chmod(config_dir, 0o700)
+
+
 def _get_config_dir():
-    azure_devops_config_dir = os.getenv(AZ_DEVOPS_CONFIG_DIR_ENVKEY, None) or AZ_DEVOPS_DEFAULT_CONFIG_DIR
-    # Create a directory if it doesn't exist
-    ensure_dir(azure_devops_config_dir)
+    configured_dir = os.getenv(AZ_DEVOPS_CONFIG_DIR_ENVKEY, None)
+    azure_devops_config_dir = configured_dir or AZ_DEVOPS_DEFAULT_CONFIG_DIR
+    _ensure_private_config_dir(azure_devops_config_dir)
     return azure_devops_config_dir
 
 

azure-devops/azext_devops/dev/common/credential_store.py

@@ -37,6 +37,8 @@ def set_password(self, key, token):
             # store credentials in azuredevops config directory if keyring is missing or malfunctioning
             if sys.platform.startswith(self._LINUX_PLATFORM):
                 logger.warning('Failed to store PAT using keyring; falling back to file storage.')
+                logger.warning('The PAT will be stored at %s with access limited to the current user.',
+                               self._PAT_FILE)
                 logger.warning('You can clear the stored credential by running az devops logout.')
                 logger.warning('Refer https://aka.ms/azure-devops-cli-auth to know more on sign in with PAT.')
                 logger.debug('Keyring failed. ERROR :%s', ex)
@@ -135,7 +137,13 @@ def _get_credentials_list():
 
     @staticmethod
     def _commit_change(credential_list):
-        with open(CredentialStore._PAT_FILE, 'w+') as creds_file:
+        flags = os.O_WRONLY | os.O_CREAT
+        with os.fdopen(os.open(CredentialStore._PAT_FILE, flags, 0o600), 'w') as creds_file:
+            if hasattr(os, 'fchmod'):
+                os.fchmod(creds_file.fileno(), 0o600)
+            else:
+                os.chmod(CredentialStore._PAT_FILE, 0o600)
+            creds_file.truncate()
             credential_list.write(creds_file)
 
     @staticmethod

azure-devops/azext_devops/tests/latest/common/test_config.py

@@ -0,0 +1,41 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import os
+import stat
+import tempfile
+import unittest
+
+from azext_devops.dev.common.config import _ensure_private_config_dir
+
+
+@unittest.skipIf(os.name == 'nt', 'POSIX directory permissions are not supported on Windows.')
+class TestConfig(unittest.TestCase):
+
+    def test_ensure_private_config_dir_creates_owner_only_directory(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            config_dir = os.path.join(temp_dir, 'azuredevops')
+
+            previous_umask = os.umask(0o777)
+            try:
+                _ensure_private_config_dir(config_dir)
+            finally:
+                os.umask(previous_umask)
+
+            self.assertEqual(0o700, stat.S_IMODE(os.stat(config_dir).st_mode))
+
+    def test_ensure_private_config_dir_preserves_existing_directory_permissions(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            config_dir = os.path.join(temp_dir, 'azuredevops')
+            os.makedirs(config_dir, mode=0o755)
+            os.chmod(config_dir, 0o755)
+
+            _ensure_private_config_dir(config_dir)
+
+            self.assertEqual(0o755, stat.S_IMODE(os.stat(config_dir).st_mode))
+
+
+if __name__ == '__main__':
+    unittest.main()

azure-devops/azext_devops/tests/latest/common/test_credential_store.py

@@ -0,0 +1,82 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import os
+import stat
+import tempfile
+import unittest
+
+try:
+    from unittest.mock import patch
+except ImportError:
+    from mock import patch
+
+from azext_devops.dev.common.credential_store import CredentialStore
+
+
+class TestCredentialStore(unittest.TestCase):
+
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.pat_file = os.path.join(self.temp_dir.name, 'personalAccessTokens')
+        self.pat_file_patcher = patch.object(CredentialStore, '_PAT_FILE', self.pat_file)
+        self.pat_file_patcher.start()
+
+    def tearDown(self):
+        self.pat_file_patcher.stop()
+        self.temp_dir.cleanup()
+
+    def test_commit_change_creates_owner_only_file(self):
+        if os.name == 'nt':
+            self.skipTest('POSIX file permissions are not supported on Windows.')
+
+        CredentialStore._commit_change(self._get_credentials())
+
+        self.assertEqual(0o600, stat.S_IMODE(os.stat(self.pat_file).st_mode))
+
+    def test_commit_change_hardens_existing_file(self):
+        if os.name == 'nt':
+            self.skipTest('POSIX file permissions are not supported on Windows.')
+
+        with open(self.pat_file, 'w') as creds_file:
+            creds_file.write('existing content')
+        os.chmod(self.pat_file, 0o644)
+
+        CredentialStore._commit_change(self._get_credentials())
+
+        self.assertEqual(0o600, stat.S_IMODE(os.stat(self.pat_file).st_mode))
+
+    def test_commit_change_preserves_existing_file_when_hardening_fails(self):
+        if os.name == 'nt' or not hasattr(os, 'fchmod'):
+            self.skipTest('POSIX file permissions are not supported on Windows.')
+
+        existing_content = 'existing content'
+        with open(self.pat_file, 'w') as creds_file:
+            creds_file.write(existing_content)
+
+        with patch.object(os, 'fchmod', side_effect=PermissionError):
+            with self.assertRaises(PermissionError):
+                CredentialStore._commit_change(self._get_credentials())
+
+        with open(self.pat_file, 'r') as creds_file:
+            self.assertEqual(existing_content, creds_file.read())
+
+    def test_commit_change_writes_credentials(self):
+        CredentialStore._commit_change(self._get_credentials())
+
+        credentials = CredentialStore._get_config_parser()
+        credentials.read(self.pat_file)
+        self.assertEqual('token', credentials.get('organization', CredentialStore._USERNAME))
+
+    @staticmethod
+    def _get_credentials():
+        credentials = CredentialStore._get_config_parser()
+        credentials.add_section('organization')
+        credentials.set('organization', CredentialStore._USERNAME, 'token')
+        return credentials
+
+
+if __name__ == '__main__':
+    unittest.main()

azure-devops/azext_devops/tests/latest/common/test_credentials.py

@@ -6,6 +6,12 @@
 import os
 import unittest
 import uuid
+
+try:
+    from unittest.mock import patch
+except ImportError:
+    from mock import patch
+
 from azext_devops.dev.common._credentials import (get_credential,
                                    set_credential,
                                    clear_credential,
@@ -170,5 +176,25 @@ def test_organization_list_file(self):
         retrieved_token = get_credential(organization3, fall_back_to_default=False)
         self.assertEqual(None, retrieved_token)
 
+    def test_clear_credential_preserves_organization_when_credential_deletion_fails(self):
+        organization = 'https://' + str(uuid.uuid4()) + '.visualstudio.com/'
+        organization_key = 'azdevops-cli:' + normalize_url_for_key(organization)
+        other_organization_key = 'azdevops-cli:https://example.visualstudio.com'
+
+        with open(ORGANIZATION_LIST_FILE, 'w') as org_list:
+            org_list.write(organization_key + '\n')
+            org_list.write(other_organization_key + '\n')
+
+        with patch('azext_devops.dev.common._credentials.CredentialStore') as credential_store:
+            credential_store.return_value.clear_password.side_effect = OSError('permission denied')
+
+            with self.assertRaises(OSError):
+                clear_credential(organization)
+
+        with open(ORGANIZATION_LIST_FILE, 'r') as org_list:
+            self.assertEqual(
+                [organization_key + '\n', other_organization_key + '\n'],
+                org_list.readlines())
+
 if __name__ == '__main__':
     unittest.main()