Merge pull request #1232 from guardicore/secure-mongo-key-file

Create mongo key file securely

monkey/monkey_island/cc/environment/utils.py

@@ -1,51 +0,0 @@
-import logging
-import os
-import platform
-
-
-def is_windows_os() -> bool:
-    return platform.system() == "Windows"
-
-
-if is_windows_os():
-    import win32file
-    import win32security
-
-    import monkey_island.cc.environment.windows_permissions as windows_permissions
-
-LOG = logging.getLogger(__name__)
-
-
-def create_secure_directory(path: str):
-    if not os.path.isdir(path):
-        if is_windows_os():
-            _create_secure_directory_windows(path)
-        else:
-            _create_secure_directory_linux(path)
-
-
-def _create_secure_directory_linux(path: str):
-    try:
-        # Don't split directory creation and permission setting
-        # because it will temporarily create an accessible directory which anyone can use.
-        os.mkdir(path, mode=0o700)
-    except Exception as ex:
-        LOG.error(
-            f'Could not create a directory at "{path}" (maybe environmental variables could not be '
-            f"resolved?): {str(ex)}"
-        )
-        raise ex
-
-
-def _create_secure_directory_windows(path: str):
-    try:
-        security_attributes = win32security.SECURITY_ATTRIBUTES()
-        security_attributes.SECURITY_DESCRIPTOR = (
-            windows_permissions.get_security_descriptor_for_owner_only_perms()
-        )
-        win32file.CreateDirectory(path, security_attributes)
-    except Exception as ex:
-        LOG.error(
-            f'Could not create a directory at "{path}": {str(ex)}'
-        )
-        raise ex

monkey/monkey_island/cc/server_utils/consts.py

@@ -1,8 +1,8 @@
 import os
 from pathlib import Path
 
-from monkey_island.cc.environment.utils import is_windows_os
 from monkey_island.cc.server_utils import file_utils
+from monkey_island.cc.server_utils.file_utils import is_windows_os
 
 __author__ = "itay.mizeretz"
 

monkey/monkey_island/cc/server_utils/encryptor.py

@@ -6,6 +6,8 @@ import os
 from Crypto import Random  # noqa: DUO133  # nosec: B413
 from Crypto.Cipher import AES  # noqa: DUO133  # nosec: B413
 
+from monkey_island.cc.server_utils.file_utils import get_file_descriptor_for_new_secure_file
+
 __author__ = "itay.mizeretz"
 
 _encryptor = None
@@ -25,7 +27,7 @@ class Encryptor:
 
     def _init_key(self, password_file):
         self._cipher_key = Random.new().read(self._BLOCK_SIZE)
-        with open(password_file, "wb") as f:
+        with open(get_file_descriptor_for_new_secure_file(path=password_file), "wb") as f:
             f.write(self._cipher_key)
 
     def _load_existing_key(self, password_file):

monkey/monkey_island/cc/server_utils/file_utils.py

@@ -1,5 +1,110 @@
+import logging
 import os
+import platform
+import stat
+
+LOG = logging.getLogger(__name__)
+
+
+def is_windows_os() -> bool:
+    return platform.system() == "Windows"
+
+
+if is_windows_os():
+    import win32file
+    import win32job
+    import win32security
+
+    import monkey_island.cc.server_utils.windows_permissions as windows_permissions
 
 
 def expand_path(path: str) -> str:
     return os.path.expandvars(os.path.expanduser(path))
+
+
+def create_secure_directory(path: str):
+    if not os.path.isdir(path):
+        if is_windows_os():
+            _create_secure_directory_windows(path)
+        else:
+            _create_secure_directory_linux(path)
+
+
+def _create_secure_directory_linux(path: str):
+    try:
+        # Don't split directory creation and permission setting
+        # because it will temporarily create an accessible directory which anyone can use.
+        os.mkdir(path, mode=stat.S_IRWXU)
+
+    except Exception as ex:
+        LOG.error(f'Could not create a directory at "{path}": {str(ex)}')
+        raise ex
+
+
+def _create_secure_directory_windows(path: str):
+    try:
+        security_attributes = win32security.SECURITY_ATTRIBUTES()
+        security_attributes.SECURITY_DESCRIPTOR = (
+            windows_permissions.get_security_descriptor_for_owner_only_perms()
+        )
+        win32file.CreateDirectory(path, security_attributes)
+
+    except Exception as ex:
+        LOG.error(f'Could not create a directory at "{path}": {str(ex)}')
+        raise ex
+
+
+def get_file_descriptor_for_new_secure_file(path: str) -> int:
+    if is_windows_os():
+        return _get_file_descriptor_for_new_secure_file_windows(path)
+    else:
+        return _get_file_descriptor_for_new_secure_file_linux(path)
+
+
+def _get_file_descriptor_for_new_secure_file_linux(path: str) -> int:
+    try:
+        mode = stat.S_IRUSR | stat.S_IWUSR
+        flags = (
+            os.O_RDWR | os.O_CREAT | os.O_EXCL
+        )  # read/write, create new, throw error if file exists
+        fd = os.open(path, flags, mode)
+
+        return fd
+
+    except Exception as ex:
+        LOG.error(f'Could not create a file at "{path}": {str(ex)}')
+        raise ex
+
+
+def _get_file_descriptor_for_new_secure_file_windows(path: str) -> int:
+    try:
+        file_access = win32file.GENERIC_READ | win32file.GENERIC_WRITE
+        # subsequent open operations on the object will succeed only if read access is requested
+        file_sharing = win32file.FILE_SHARE_READ
+        security_attributes = win32security.SECURITY_ATTRIBUTES()
+        security_attributes.SECURITY_DESCRIPTOR = (
+            windows_permissions.get_security_descriptor_for_owner_only_perms()
+        )
+        file_creation = win32file.CREATE_NEW  # fails if file exists
+        file_attributes = win32file.FILE_FLAG_BACKUP_SEMANTICS
+
+        fd = win32file.CreateFile(
+            path,
+            file_access,
+            file_sharing,
+            security_attributes,
+            file_creation,
+            file_attributes,
+            _get_null_value_for_win32(),
+        )
+
+        return fd
+
+    except Exception as ex:
+        LOG.error(f'Could not create a file at "{path}": {str(ex)}')
+        raise ex
+
+
+def _get_null_value_for_win32() -> None:
+    # https://stackoverflow.com/questions/46800142/in-python-with-pywin32-win32job-the-createjobobject-function-how-do-i-pass-nu  # noqa: E501
+    return win32job.CreateJobObject(None, "")

monkey/monkey_island/cc/setup/config_setup.py

@@ -2,9 +2,9 @@ from typing import Tuple
 
 from monkey_island.cc.arg_parser import IslandCmdArgs
 from monkey_island.cc.environment import server_config_handler
-from monkey_island.cc.environment.utils import create_secure_directory
 from monkey_island.cc.server_utils import file_utils
 from monkey_island.cc.server_utils.consts import DEFAULT_SERVER_CONFIG_PATH
+from monkey_island.cc.server_utils.file_utils import create_secure_directory
 from monkey_island.cc.setup.island_config_options import IslandConfigOptions
 
 

monkey/monkey_island/cc/setup/mongo/mongo_setup.py

@@ -5,7 +5,7 @@ import sys
 import time
 
 from monkey_island.cc.database import get_db_version, is_db_server_up
-from monkey_island.cc.environment.utils import create_secure_directory
+from monkey_island.cc.server_utils.file_utils import create_secure_directory
 from monkey_island.cc.setup.mongo import mongo_connector
 from monkey_island.cc.setup.mongo.mongo_connector import MONGO_DB_HOST, MONGO_DB_NAME, MONGO_DB_PORT
 from monkey_island.cc.setup.mongo.mongo_db_process import MongoDbProcess

monkey/tests/unit_tests/monkey_island/cc/environment/test_utils.py

@@ -1,65 +0,0 @@
-import os
-import stat
-
-import pytest
-
-from monkey_island.cc.environment.utils import create_secure_directory, is_windows_os
-
-
-@pytest.fixture
-def test_path_nested(tmpdir):
-    path = os.path.join(tmpdir, "test1", "test2", "test3")
-    return path
-
-
-@pytest.fixture
-def test_path(tmpdir):
-    test_path = "test1"
-    path = os.path.join(tmpdir, test_path)
-
-    return path
-
-
-def test_create_secure_directory__already_created(test_path):
-    os.mkdir(test_path)
-    assert os.path.isdir(test_path)
-    create_secure_directory(test_path)
-
-
-def test_create_secure_directory__no_parent_dir(test_path_nested):
-    with pytest.raises(Exception):
-        create_secure_directory(test_path_nested)
-
-
-@pytest.mark.skipif(is_windows_os(), reason="Tests Posix (not Windows) permissions.")
-def test_create_secure_directory__perm_linux(test_path):
-    create_secure_directory(test_path)
-    st = os.stat(test_path)
-    return bool(st.st_mode & stat.S_IRWXU)
-
-
-@pytest.mark.skipif(not is_windows_os(), reason="Tests Windows (not Posix) permissions.")
-def test_create_secure_directory__perm_windows(test_path):
-    import win32api
-    import win32security
-
-    FULL_CONTROL = 2032127
-    ACE_TYPE_ALLOW = 0
-
-    create_secure_directory(test_path)
-
-    user_sid, _, _ = win32security.LookupAccountName("", win32api.GetUserName())
-    security_descriptor = win32security.GetNamedSecurityInfo(
-        test_path, win32security.SE_FILE_OBJECT, win32security.DACL_SECURITY_INFORMATION
-    )
-    acl = security_descriptor.GetSecurityDescriptorDacl()
-
-    assert acl.GetAceCount() == 1
-
-    ace = acl.GetAce(0)
-    ace_type, _ = ace[0]  # 0 for allow, 1 for deny
-    permissions = ace[1]
-    sid = ace[-1]
-
-    assert sid == user_sid
-    assert permissions == FULL_CONTROL and ace_type == ACE_TYPE_ALLOW

monkey/tests/unit_tests/monkey_island/cc/server_utils/test_file_utils.py

@@ -1,17 +1,136 @@
 import os
+import stat
 
-from monkey_island.cc.server_utils import file_utils
+import pytest
+
+from monkey_island.cc.server_utils.file_utils import (
+    create_secure_directory,
+    expand_path,
+    get_file_descriptor_for_new_secure_file,
+    is_windows_os,
+)
+
+if is_windows_os():
+    import win32api
+    import win32file
+    import win32security
+
+    FULL_CONTROL = 2032127
+    ACE_TYPE_ALLOW = 0
 
 
 def test_expand_user(patched_home_env):
     input_path = os.path.join("~", "test")
     expected_path = os.path.join(patched_home_env, "test")
 
-    assert file_utils.expand_path(input_path) == expected_path
+    assert expand_path(input_path) == expected_path
 
 
 def test_expand_vars(patched_home_env):
     input_path = os.path.join("$HOME", "test")
     expected_path = os.path.join(patched_home_env, "test")
 
-    assert file_utils.expand_path(input_path) == expected_path
+    assert expand_path(input_path) == expected_path
+
+
+@pytest.fixture
+def test_path_nested(tmpdir):
+    path = os.path.join(tmpdir, "test1", "test2", "test3")
+    return path
+
+
+@pytest.fixture
+def test_path(tmpdir):
+    test_path = "test1"
+    path = os.path.join(tmpdir, test_path)
+
+    return path
+
+
+def _get_acl_and_sid_from_path(path: str):
+    sid, _, _ = win32security.LookupAccountName("", win32api.GetUserName())
+    security_descriptor = win32security.GetNamedSecurityInfo(
+        path, win32security.SE_FILE_OBJECT, win32security.DACL_SECURITY_INFORMATION
+    )
+    acl = security_descriptor.GetSecurityDescriptorDacl()
+    return acl, sid
+
+
+def test_create_secure_directory__already_exists(test_path):
+    os.mkdir(test_path)
+    assert os.path.isdir(test_path)
+    create_secure_directory(test_path)
+
+
+def test_create_secure_directory__no_parent_dir(test_path_nested):
+    with pytest.raises(Exception):
+        create_secure_directory(test_path_nested)
+
+
+@pytest.mark.skipif(is_windows_os(), reason="Tests Posix (not Windows) permissions.")
+def test_create_secure_directory__perm_linux(test_path):
+    create_secure_directory(test_path)
+    st = os.stat(test_path)
+
+    expected_mode = stat.S_IRWXU
+    actual_mode = st.st_mode & (stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)
+
+    assert expected_mode == actual_mode
+
+
+@pytest.mark.skipif(not is_windows_os(), reason="Tests Windows (not Posix) permissions.")
+def test_create_secure_directory__perm_windows(test_path):
+    create_secure_directory(test_path)
+
+    acl, user_sid = _get_acl_and_sid_from_path(test_path)
+
+    assert acl.GetAceCount() == 1
+
+    ace = acl.GetAce(0)
+    ace_type, _ = ace[0]  # 0 for allow, 1 for deny
+    permissions = ace[1]
+    sid = ace[-1]
+
+    assert sid == user_sid
+    assert permissions == FULL_CONTROL and ace_type == ACE_TYPE_ALLOW
+
+
+def test_get_file_descriptor_for_new_secure_file__already_exists(test_path):
+    os.close(os.open(test_path, os.O_CREAT, stat.S_IRWXU))
+    assert os.path.isfile(test_path)
+
+    with pytest.raises(Exception):
+        get_file_descriptor_for_new_secure_file(test_path)
+
+
+def test_get_file_descriptor_for_new_secure_file__no_parent_dir(test_path_nested):
+    with pytest.raises(Exception):
+        get_file_descriptor_for_new_secure_file(test_path_nested)
+
+
+@pytest.mark.skipif(is_windows_os(), reason="Tests Posix (not Windows) permissions.")
+def test_get_file_descriptor_for_new_secure_file__perm_linux(test_path):
+    os.close(get_file_descriptor_for_new_secure_file(test_path))
+    st = os.stat(test_path)
+
+    expected_mode = stat.S_IRUSR | stat.S_IWUSR
+    actual_mode = st.st_mode & (stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)
+
+    assert expected_mode == actual_mode
+
+
+@pytest.mark.skipif(not is_windows_os(), reason="Tests Windows (not Posix) permissions.")
+def test_get_file_descriptor_for_new_secure_file__perm_windows(test_path):
+    win32file.CloseHandle(get_file_descriptor_for_new_secure_file(test_path))
+
+    acl, user_sid = _get_acl_and_sid_from_path(test_path)
+
+    assert acl.GetAceCount() == 1
+
+    ace = acl.GetAce(0)
+    ace_type, _ = ace[0]  # 0 for allow, 1 for deny
+    permissions = ace[1]
+    sid = ace[-1]
+
+    assert sid == user_sid
+    assert permissions == FULL_CONTROL and ace_type == ACE_TYPE_ALLOW