fix: improve response time on windows when using deadline config GUI

src/deadline/client/config/config_file.py

@@ -16,7 +16,6 @@
 import getpass
 import os
 import platform
-import subprocess
 from configparser import ConfigParser
 from pathlib import Path
 from typing import Any, Dict, List, Optional
@@ -26,7 +25,6 @@
 from deadline.job_attachments.models import FileConflictResolution
 
 from ..exceptions import DeadlineOperationError
-import re
 
 # Default path where AWS Deadline Cloud's configuration lives
 CONFIG_FILE_PATH = os.path.join("~", ".deadline", "config")
@@ -182,80 +180,50 @@ def read_config() -> ConfigParser:
     return __config
 
 
-def _get_grant_args(principal: str, permissions: str) -> List[str]:
-    return [
-        "/grant",
-        f"{principal}:{permissions}",
-        # Apply recursively
-        "/T",
-    ]
-
-
-RE_ICACLS_OUTPUT = re.compile(r"^(.+?(?=\\))?(?:\\)?(.+?(?=:)):(.*)$")
-
-
-def _reset_directory_permissions_windows(directory: Path, username: str, permissions: str) -> None:
+def _reset_directory_permissions_windows(directory: Path) -> None:
     if platform.system() != "Windows":
         return
+    import win32security
+    import ntsecuritycon
+
+    # We don't want to propagate existing permissions, so create a new DACL
+    dacl = win32security.ACL()
+
+    # On Windows, both SYSTEM and the Administrators group normally
+    # have Full Access to files in the user's home directory.
+    # Use SIDs to represent the Administrators and SYSTEM to
+    # support multi-language operating systems
+    # Administrator(S-1-5-32-544), SYSTEM(S-1-5-18)
+    # https://learn.microsoft.com/en-us/windows/win32/secauthz/well-known-sids
+    system_sid = win32security.ConvertStringSidToSid("S-1-5-18")
+    admin_sid = win32security.ConvertStringSidToSid("S-1-5-32-544")
+
+    username = getpass.getuser()
+    user_sid, _, _ = win32security.LookupAccountName(None, username)
+
+    for sid in [user_sid, admin_sid, system_sid]:
+        dacl.AddAccessAllowedAceEx(
+            win32security.ACL_REVISION,
+            ntsecuritycon.OBJECT_INHERIT_ACE | ntsecuritycon.CONTAINER_INHERIT_ACE,
+            ntsecuritycon.GENERIC_ALL,
+            sid,
+        )
 
-    result = subprocess.run(
-        [
-            "icacls",
-            str(directory),
-        ],
-        check=True,
-        capture_output=True,
-        text=True,
+    # Get the security descriptor of the object
+    sd = win32security.GetFileSecurity(
+        str(directory.resolve()), win32security.DACL_SECURITY_INFORMATION
     )
 
-    icacls_output = result.stdout
-
-    principals_to_remove = []
-
-    for line in icacls_output.splitlines():
-        if line.startswith(str(directory)):
-            permission_line = line[len(str(directory)) :].strip()
-        else:
-            permission_line = line.strip()
-
-        permissions_match = RE_ICACLS_OUTPUT.match(permission_line)
-        if permissions_match:
-            ad_group = permissions_match.group(1)
-            ad_user = permissions_match.group(2)
-            principal = f"{ad_group}\\{ad_user}"
-            if (
-                ad_user != username
-                and principal != "BUILTIN\\Administrators"
-                and principal != "NT AUTHORITY\\SYSTEM"
-            ):
-                principals_to_remove.append(ad_user)
-
-    for principal in principals_to_remove:
-        subprocess.run(
-            [
-                "icacls",
-                str(directory),
-                "/remove",
-                principal,
-            ],
-            check=True,
-        )
-
-    subprocess.run(
-        [
-            "icacls",
-            str(directory),
-            *_get_grant_args(username, permissions),
-            # On Windows, both SYSTEM and the Administrators group normally
-            # have Full Access to files in the user's home directory.
-            # Use SIDs to represent the Administrators and SYSTEM to
-            # support multi-language operating systems
-            # Administrator(S-1-5-32-544), SYSTEM(S-1-5-18)
-            *_get_grant_args("*S-1-5-32-544", permissions),
-            *_get_grant_args("*S-1-5-18", permissions),
-        ],
-        check=True,
-        capture_output=True,
+    # Set the security descriptor's DACL to the newly-created DACL
+    # Arguments:
+    # 1. bDaclPresent = 1: Indicates that the DACL is present in the security descriptor.
+    #    If set to 0, this method ignores the provided DACL and allows access to all principals.
+    # 2. dacl: The discretionary access control list (DACL) to be set in the security descriptor.
+    # 3. bDaclDefaulted = 0: Indicates the DACL was provided and not defaulted.
+    #    If set to 1, indicates the DACL was defaulted, as in the case of permissions inherited from a parent directory.
+    sd.SetSecurityDescriptorDacl(1, dacl, 0)
+    win32security.SetFileSecurity(
+        str(directory.resolve()), win32security.DACL_SECURITY_INFORMATION, sd
     )
 
 
@@ -268,15 +236,10 @@ def write_config(config: ConfigParser) -> None:
             a modified value from what `read_config` returns.
     """
     config_file_path = get_config_file_path()
-    config_file_path.parent.mkdir(parents=True, exist_ok=True)
-
-    if platform.system() == "Windows":
-        username = getpass.getuser()
-        config_file_parent_path = config_file_path.parent.absolute()
-        # OI - Contained objects will inherit
-        # CI - Sub-directories will inherit
-        # F  - Full control
-        _reset_directory_permissions_windows(config_file_parent_path, username, "(OI)(CI)(F)")
+    if not config_file_path.parent.exists():
+        config_file_path.parent.mkdir(parents=True, exist_ok=True)
+        if platform.system() == "Windows":
+            _reset_directory_permissions_windows(config_file_path.parent)
 
     # Using the config file path as the prefix ensures that the tmpfile and real file are
     # on the same filesystem. This is a requirement for os.replace to be atomic.

test/unit/deadline_client/config/test_config_file.py

@@ -6,7 +6,8 @@
 
 import os
 import platform
-import subprocess
+import getpass
+import tempfile
 from unittest.mock import patch, MagicMock
 from pathlib import Path
 
@@ -273,32 +274,92 @@ def test_default_log_level():
     platform.system() != "Windows",
     reason="This test is for testing file permission changes in Windows.",
 )
-def test_windows_config_file_permissions(fresh_deadline_config) -> None:
-    config_file_path = config_file.get_config_file_path()
-    parent_dir = config_file_path.parent
-    subprocess.run(
-        [
-            "icacls",
-            str(parent_dir),
-            "/grant",
-            "Everyone:(OI)(CI)(F)",
-            "/T",
-        ],
-        check=True,
-    )
+def test_reset_directory_permissions_windows() -> None:
+    """
+    Asserts the _reset_directory_permissions_windows configures the provided
+    folder with access only to the active user, the domain admin, and SYSTEM.
+    """
+    # GIVEN
+    import ntsecuritycon
+    import win32security
+
+    path = Path(tempfile.gettempdir())
+    system_sid = win32security.ConvertStringSidToSid("S-1-5-18")
+    admin_sid = win32security.ConvertStringSidToSid("S-1-5-32-544")
+    user_sid, _, _ = win32security.LookupAccountName(None, getpass.getuser())
+    sids = [system_sid, admin_sid, user_sid]
+
+    # WHEN
+    config_file._reset_directory_permissions_windows(path)
+
+    # THEN
+    sd = win32security.GetFileSecurity(str(path.resolve()), win32security.DACL_SECURITY_INFORMATION)
+    dacl = sd.GetSecurityDescriptorDacl()
+    assert dacl.GetAceCount() == 3
+    assert dacl.GetAclRevision() == win32security.ACL_REVISION
+    for i in range(3):
+        (acetype, aceflags), access, sid = dacl.GetAce(i)
+        assert acetype == win32security.ACCESS_ALLOWED_ACE_TYPE
+        assert aceflags == ntsecuritycon.OBJECT_INHERIT_ACE | ntsecuritycon.CONTAINER_INHERIT_ACE
+        assert access == ntsecuritycon.FILE_ALL_ACCESS
+        try:
+            sids.remove(sid)
+        except ValueError:
+            assert False, f"Unexpected SID: {win32security.ConvertSidToStringSid(sid)}"
 
-    config_file.set_setting("defaults.aws_profile_name", "goodguyprofile")
 
-    result = subprocess.run(
-        [
-            "icacls",
-            str(config_file_path),
-        ],
-        check=True,
-        capture_output=True,
-        text=True,
+@pytest.mark.skipif(
+    platform.system() != "Windows",
+    reason="This test is for testing file permission changes in Windows.",
+)
+@patch.object(config_file, "get_config_file_path")
+def test_write_config_directory_permission_windows(
+    mock_get_config_file_path,
+):
+    """
+    Tests that the config directory permissions are not modified when writing to the config file
+    """
+    # GIVEN
+    path = Path(tempfile.gettempdir())
+    config_path = path / "config"
+    mock_get_config_file_path.return_value = config_path
+
+    # ----------------------------------------------------------------------------------------------
+    # Sets up a directory with an added full access entry for domain guests. Since this is not a
+    # typically expected entry, it can be used to validate existing permissions were not overwritten
+    import win32security
+    import ntsecuritycon
+
+    sd = win32security.GetFileSecurity(str(path.resolve()), win32security.DACL_SECURITY_INFORMATION)
+    guest_sid = win32security.ConvertStringSidToSid("S-1-5-32-546")  # Domain Guests
+    dacl = sd.GetSecurityDescriptorDacl()
+    dacl.AddAccessAllowedAceEx(
+        win32security.ACL_REVISION,
+        ntsecuritycon.OBJECT_INHERIT_ACE | ntsecuritycon.CONTAINER_INHERIT_ACE,
+        ntsecuritycon.FILE_ALL_ACCESS,
+        guest_sid,
     )
-    assert "Everyone" not in result.stdout
+    sd.SetSecurityDescriptorDacl(1, dacl, 0)
+    win32security.SetFileSecurity(str(path.resolve()), win32security.DACL_SECURITY_INFORMATION, sd)
+    # ----------------------------------------------------------------------------------------------
+
+    # WHEN
+    config_file.write_config(MagicMock())
+
+    # THEN
+    new_dacl = win32security.GetFileSecurity(
+        str(path.resolve()), win32security.DACL_SECURITY_INFORMATION
+    ).GetSecurityDescriptorDacl()
+
+    assert new_dacl.GetAceCount() == dacl.GetAceCount()
+    for i in range(dacl.GetAceCount()):
+        # Assert the access control entries are identical
+        (acetype, aceflags), access, sid = dacl.GetAce(i)
+        (new_acetype, new_aceflags), new_access, new_sid = new_dacl.GetAce(i)
+        assert acetype == new_acetype
+        assert aceflags == new_aceflags
+        assert access == new_access
+        assert sid == new_sid
 
 
 @pytest.mark.skipif(