PILOT-3313: Add permissions check for config folder and file

app/configs/app_config.py

@@ -11,7 +11,7 @@ class Env:
         project = ConfigClass.project
         user_config_path = ConfigClass.config_path
         msg_path = ConfigClass.custom_path
-        user_config_file = f'{user_config_path}/config.ini'
+        user_config_file = 'config.ini'
         token_warn_need_refresh = 120  # refresh token if token is about to expire
         token_refresh_interval = 90  # auto refresh token every 40 seconds
 

app/configs/user_config.py

@@ -4,14 +4,19 @@
 
 import configparser
 import os
+import stat
 import time
 from pathlib import Path
+from typing import Iterable
+from typing import Union
 
 from app.configs.app_config import AppConfig
 from app.models.singleton import Singleton
 from app.services.crypto.crypto import decryption
 from app.services.crypto.crypto import encryption
 from app.services.crypto.crypto import generate_secret
+from app.services.output_manager.error_handler import ECustomizedError
+from app.services.output_manager.error_handler import SrvErrorHandler
 
 
 class UserConfig(metaclass=Singleton):
@@ -21,13 +26,35 @@ class UserConfig(metaclass=Singleton):
     This user config is global.
     """
 
-    def __init__(self):
-        if not os.path.exists(AppConfig.Env.user_config_path):
-            os.makedirs(AppConfig.Env.user_config_path)
-        if not os.path.exists(AppConfig.Env.user_config_file):
-            Path.touch(Path(AppConfig.Env.user_config_file))
+    def __init__(self, config_path: Union[str, Path, None] = None, config_filename: Union[str, None] = None) -> None:
+        if config_path is None:
+            config_path = AppConfig.Env.user_config_path
+        if config_filename is None:
+            config_filename = AppConfig.Env.user_config_file
+
+        config_path = Path(config_path)
+        if not config_path.exists():
+            config_path.mkdir(mode=0o0700, exist_ok=False)
+
+        current_user_id = os.geteuid()
+
+        error = self._check_user_permissions(config_path, current_user_id, (0o0500, 0o0700))
+        if error:
+            SrvErrorHandler.customized_handle(ECustomizedError.CONFIG_INVALID_PERMISSIONS, True, error)
+            return
+
+        config_file = config_path / config_filename
+        if not config_file.exists():
+            config_file.touch(mode=0o0600, exist_ok=False)
+
+        error = self._check_user_permissions(config_file, current_user_id, (0o0400, 0o0600))
+        if error:
+            SrvErrorHandler.customized_handle(ECustomizedError.CONFIG_INVALID_PERMISSIONS, True, error)
+            return
+
+        self.config_file = config_file
         self.config = configparser.ConfigParser()
-        self.config.read(AppConfig.Env.user_config_file)
+        self.config.read(self.config_file)
         if not self.config.has_section('USER'):
             self.config['USER'] = {
                 'username': '',
@@ -42,8 +69,28 @@ def __init__(self):
             }
             self.save()
 
+    def _check_user_permissions(self, path: Path, expected_uid: int, expected_bits: Iterable[int]) -> Union[str, None]:
+        """Check if file or folder is owned by the user and has proper access mode."""
+
+        path_stat = path.stat()
+
+        path_uid = path_stat.st_uid
+        if path_uid != expected_uid:
+            return f'"{path}" is owned by the user id {path_uid}. Expected user id is {expected_uid}.'
+
+        path_protection_bits = stat.S_IMODE(path_stat.st_mode)
+        if path_protection_bits not in expected_bits:
+            existing_permissions = oct(path_protection_bits).replace('0o', '')
+            expected_permissions = ', '.join(map(oct, expected_bits)).replace('0o', '')
+            return (
+                f'Permissions {existing_permissions} for "{path}" are too open. '
+                f'Expected permissions are {expected_permissions}.'
+            )
+
+        return None
+
     def save(self):
-        with open(AppConfig.Env.user_config_file, 'w') as configfile:
+        with open(self.config_file, 'w') as configfile:
             self.config.write(configfile)
 
     def clear(self):

app/resources/custom_error.py

@@ -119,6 +119,7 @@ class Error:
             'This cli has been configured already.'
             'If you want to re-config this cli please remove previous file first'
         ),
+        'CONFIG_INVALID_PERMISSIONS': 'Cannot proceed with current config permissions.\n%s',
         'CONTAINER_REGISTRY_NO_URL': (
             'Container registry has not yet been configured. Related commands cannot be used at this time.'
         ),

app/services/output_manager/error_handler.py

@@ -81,6 +81,7 @@ class ECustomizedError(enum.Enum):
     CONTAINER_REGISTRY_NO_URL = 'CONTAINER_REGISTRY_NO_URL'
     CONFIG_NOT_FOUND = 'CONFIG_NOT_FOUND'
     CONFIG_EXIST = 'CONFIG_EXIST'
+    CONFIG_INVALID_PERMISSIONS = 'CONFIG_INVALID_PERMISSIONS'
 
 
 def customized_error_msg(customized_error: ECustomizedError):

tests/app/configs/__init__.py

@@ -0,0 +1,3 @@
+# Copyright (C) 2023 Indoc Research
+#
+# Contact Indoc Research for any questions regarding the use of this source code.

tests/app/configs/test_user_config.py

@@ -0,0 +1,75 @@
+# Copyright (C) 2023 Indoc Research
+#
+# Contact Indoc Research for any questions regarding the use of this source code.
+
+import os
+import stat
+
+import pytest
+
+from app.configs.user_config import UserConfig
+
+
+@pytest.fixture
+def error_log(mocker):
+    return mocker.patch('app.services.logger_services.log_functions.error')
+
+
+class TestUserConfig:
+    def test__init__creates_config_folder_with_0700_and_file_with_0600_access_modes(self, tmp_path, fake):
+        config_folder = tmp_path / fake.pystr()
+        file_name = fake.pystr()
+
+        UserConfig(config_folder, file_name)
+
+        config_folder_mode = stat.S_IMODE(config_folder.stat().st_mode)
+        assert config_folder_mode == 0o0700
+
+        config_file_mode = stat.S_IMODE((config_folder / file_name).stat().st_mode)
+        assert config_file_mode == 0o0600
+
+    def test__init__exits_with_error_when_config_folder_does_not_belong_to_user(self, error_log):
+        with pytest.raises(SystemExit):
+            UserConfig('/')
+
+        expected_message = (
+            'Cannot proceed with current config permissions.\n'
+            f'"/" is owned by the user id 0. Expected user id is {os.geteuid()}.'
+        )
+
+        error_log.assert_called_with(expected_message)
+
+    def test__init__exits_with_error_when_config_folder_does_not_have_expected_access_mode(
+        self, error_log, tmp_path, fake
+    ):
+        config_folder = tmp_path / fake.pystr()
+        config_folder.mkdir(mode=0o0755)
+
+        with pytest.raises(SystemExit):
+            UserConfig(config_folder)
+
+        expected_message = (
+            'Cannot proceed with current config permissions.\n'
+            f'Permissions 755 for "{config_folder}" are too open. Expected permissions are 500, 700.'
+        )
+
+        error_log.assert_called_with(expected_message)
+
+    def test__init__exits_with_error_when_config_file_does_not_have_expected_access_mode(
+        self, error_log, tmp_path, fake
+    ):
+        config_folder = tmp_path / fake.pystr()
+        file_name = fake.pystr()
+        config_file = config_folder / file_name
+        config_folder.mkdir(mode=0o0700)
+        config_file.touch(mode=0o0644)
+
+        with pytest.raises(SystemExit):
+            UserConfig(config_folder, file_name)
+
+        expected_message = (
+            'Cannot proceed with current config permissions.\n'
+            f'Permissions 644 for "{config_file}" are too open. Expected permissions are 400, 600.'
+        )
+
+        error_log.assert_called_with(expected_message)

tests/conftest.py

@@ -13,7 +13,7 @@
 
 @pytest.fixture(autouse=True)
 def reset_singletons():
-    Singleton._instance = {}
+    Singleton._instances = {}
 
 
 @pytest.fixture(autouse=True)