diff --git a/app/configs/app_config.py b/app/configs/app_config.py index 964c7f8c..52646fe4 100644 --- a/app/configs/app_config.py +++ b/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 diff --git a/app/configs/user_config.py b/app/configs/user_config.py index 3893de1c..7b417553 100644 --- a/app/configs/user_config.py +++ b/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): diff --git a/app/resources/custom_error.py b/app/resources/custom_error.py index ed47b7be..bb093d1c 100644 --- a/app/resources/custom_error.py +++ b/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.' ), diff --git a/app/services/output_manager/error_handler.py b/app/services/output_manager/error_handler.py index 4f9ea3f4..bf1d02ae 100644 --- a/app/services/output_manager/error_handler.py +++ b/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): diff --git a/tests/app/configs/__init__.py b/tests/app/configs/__init__.py new file mode 100644 index 00000000..950d1b40 --- /dev/null +++ b/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. diff --git a/tests/app/configs/test_user_config.py b/tests/app/configs/test_user_config.py new file mode 100644 index 00000000..24a483be --- /dev/null +++ b/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) diff --git a/tests/conftest.py b/tests/conftest.py index ca3f66de..ea0c6b85 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -13,7 +13,7 @@ @pytest.fixture(autouse=True) def reset_singletons(): - Singleton._instance = {} + Singleton._instances = {} @pytest.fixture(autouse=True)