Create secure custom PBA directory on Windows

monkey/monkey_island/cc/services/post_breach_files.py

@@ -1,6 +1,7 @@
 import logging
 import os
-from pathlib import Path
+
+from monkey_island.cc.server_utils.file_utils import create_secure_directory
 
 logger = logging.getLogger(__name__)
 
@@ -15,7 +16,8 @@ class PostBreachFilesService:
     @classmethod
     def initialize(cls, data_dir):
         cls.DATA_DIR = data_dir
-        Path(cls.get_custom_pba_directory()).mkdir(mode=0o0700, parents=True, exist_ok=True)
+        custom_pba_dir = cls.get_custom_pba_directory()
+        create_secure_directory(custom_pba_dir)
 
     @staticmethod
     def save_file(filename: str, file_contents: bytes):

monkey/tests/unit_tests/monkey_island/cc/services/test_post_breach_files.py

@@ -2,8 +2,17 @@
 
 import pytest
 
+from monkey_island.cc.server_utils.file_utils import is_windows_os
 from monkey_island.cc.services.post_breach_files import PostBreachFilesService
 
+if is_windows_os():
+    import win32api
+    import win32security
+
+    FULL_CONTROL = 2032127
+    ACE_ACCESS_MODE_GRANT_ACCESS = win32security.GRANT_ACCESS
+    ACE_INHERIT_OBJECT_AND_CONTAINER = 3
+
 
 def raise_(ex):
     raise ex
@@ -32,13 +41,42 @@ def dir_is_empty(dir_path):
     return len(dir_contents) == 0
 
 
-@pytest.mark.skipif(os.name != "posix", reason="Tests Posix (not Windows) permissions.")
-def test_custom_pba_dir_permissions():
+@pytest.mark.skipif(is_windows_os(), reason="Tests Posix (not Windows) permissions.")
+def test_custom_pba_dir_permissions_linux():
     st = os.stat(PostBreachFilesService.get_custom_pba_directory())
 
     assert st.st_mode == 0o40700
 
 
+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
+
+
+@pytest.mark.skipif(not is_windows_os(), reason="Tests Windows (not Posix) permissions.")
+def test_custom_pba_dir_permissions_windows():
+    pba_dir = PostBreachFilesService.get_custom_pba_directory()
+
+    acl, user_sid = _get_acl_and_sid_from_path(pba_dir)
+
+    assert acl.GetAceCount() == 1
+
+    ace = acl.GetExplicitEntriesFromAcl()[0]
+
+    ace_access_mode = ace["AccessMode"]
+    ace_permissions = ace["AccessPermissions"]
+    ace_inheritance = ace["Inheritance"]
+    ace_sid = ace["Trustee"]["Identifier"]
+
+    assert ace_sid == user_sid
+    assert ace_permissions == FULL_CONTROL and ace_access_mode == ACE_ACCESS_MODE_GRANT_ACCESS
+    assert ace_inheritance == ACE_INHERIT_OBJECT_AND_CONTAINER
+
+
 def test_remove_failure(monkeypatch):
     monkeypatch.setattr(os, "remove", lambda x: raise_(OSError("Permission denied")))