Restrict the CodeDeploy on-premises config file on Windows

.changes/next-release/bugfix-codedeploy-61873.json

@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "codedeploy",
+  "description": "Restrict access to the on-premises instance configuration file written by ``aws deploy install`` on Windows"
+}

awscli/customizations/codedeploy/install.py

@@ -120,16 +120,29 @@ def _create_config(self, params):
         except OSError as e:
             if e.errno != errno.EEXIST:
                 raise e
-        if params.config_file != params.system.CONFIG_PATH:
+        # On Windows the file is always replaced, even when it is already the
+        # installed config, so it never keeps an ACL it had before.
+        if (
+            params.config_file != params.system.CONFIG_PATH
+            or params.system.RECREATE_CONFIG_FILE
+        ):
+            # Read the contents first, since the source may be the file that
+            # is about to be replaced.
             with open(params.config_file, 'rb') as f:
                 contents = f.read()
-            fd = os.open(
-                params.system.CONFIG_PATH,
-                os.O_WRONLY | os.O_CREAT | os.O_TRUNC | _O_BINARY,
-                0o600,
-            )
+            if params.system.RECREATE_CONFIG_FILE:
+                fd = params.system.create_config_file()
+            else:
+                fd = os.open(
+                    params.system.CONFIG_PATH,
+                    os.O_WRONLY | os.O_CREAT | os.O_TRUNC | _O_BINARY,
+                    0o600,
+                )
             with os.fdopen(fd, 'wb') as f:
-                os.chmod(params.system.CONFIG_PATH, 0o600)
+                # create_config_file sets the file's ACL itself, and Windows
+                # ignores the mode, so chmod is only needed for os.open.
+                if not params.system.RECREATE_CONFIG_FILE:
+                    os.chmod(params.system.CONFIG_PATH, 0o600)
                 f.write(contents)
         sys.stdout.write('DONE\n')
 

awscli/customizations/codedeploy/systems.py

@@ -13,17 +13,112 @@
 
 import ctypes
 import os
+import stat
 import subprocess
 
 DEFAULT_CONFIG_FILE = 'codedeploy.onpremises.yml'
 
+# Security descriptor for the Windows configuration file: a protected DACL
+# (nothing inherited from the directory) granting full control only to
+# LocalSystem and Administrators.
+_WINDOWS_CONFIG_FILE_SDDL = 'D:P(A;;FA;;;SY)(A;;FA;;;BA)'
+
+
+def _create_windows_file(path, sddl):
+    """Create ``path`` with the DACL in ``sddl`` and return a writable fd.
+
+    Windows ignores the mode passed to os.open, so the file would take its
+    ACL from its directory. Here CreateFileW sets the DACL when it creates
+    the file, so the file is never accessible with any other ACL.
+    CREATE_NEW makes this fail if ``path`` already exists.
+    """
+    import msvcrt
+    from ctypes import wintypes
+
+    class SecurityAttributes(ctypes.Structure):
+        _fields_ = [
+            ('nLength', wintypes.DWORD),
+            ('lpSecurityDescriptor', ctypes.c_void_p),
+            ('bInheritHandle', wintypes.BOOL),
+        ]
+
+    advapi32 = ctypes.WinDLL('advapi32', use_last_error=True)
+    kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
+
+    convert = advapi32.ConvertStringSecurityDescriptorToSecurityDescriptorW
+    convert.argtypes = [
+        wintypes.LPCWSTR,
+        wintypes.DWORD,
+        ctypes.POINTER(ctypes.c_void_p),
+        ctypes.POINTER(wintypes.ULONG),
+    ]
+    convert.restype = wintypes.BOOL
+
+    create_file = kernel32.CreateFileW
+    create_file.argtypes = [
+        wintypes.LPCWSTR,
+        wintypes.DWORD,
+        wintypes.DWORD,
+        ctypes.c_void_p,
+        wintypes.DWORD,
+        wintypes.DWORD,
+        wintypes.HANDLE,
+    ]
+    create_file.restype = wintypes.HANDLE
+
+    local_free = kernel32.LocalFree
+    local_free.argtypes = [ctypes.c_void_p]
+    local_free.restype = ctypes.c_void_p
+
+    close_handle = kernel32.CloseHandle
+    close_handle.argtypes = [wintypes.HANDLE]
+    close_handle.restype = wintypes.BOOL
+
+    sddl_revision_1 = 1
+    generic_write = 0x40000000
+    # No other handle may read, write or delete the file while it is open.
+    no_sharing = 0
+    create_new = 1
+    file_attribute_normal = 0x80
+    invalid_handle_value = ctypes.c_void_p(-1).value
+
+    descriptor = ctypes.c_void_p()
+    if not convert(sddl, sddl_revision_1, ctypes.byref(descriptor), None):
+        raise ctypes.WinError(ctypes.get_last_error())
+    try:
+        attributes = SecurityAttributes(
+            ctypes.sizeof(SecurityAttributes), descriptor.value, False
+        )
+        handle = create_file(
+            path,
+            generic_write,
+            no_sharing,
+            ctypes.byref(attributes),
+            create_new,
+            file_attribute_normal,
+            None,
+        )
+        if handle is None or handle == invalid_handle_value:
+            raise ctypes.WinError(ctypes.get_last_error())
+    finally:
+        local_free(descriptor)
+    try:
+        return msvcrt.open_osfhandle(handle, os.O_WRONLY | os.O_BINARY)
+    except Exception:
+        close_handle(handle)
+        raise
+
 
 class System:
     UNSUPPORTED_SYSTEM_MSG = (
         'Only Ubuntu Server, Red Hat Enterprise Linux Server and '
         'Windows Server operating systems are supported.'
     )
 
+    # Whether install replaces an existing configuration file with one from
+    # create_config_file, rather than truncating it in place.
+    RECREATE_CONFIG_FILE = False
+
     def __init__(self, params):
         self.session = params.session
         self.s3 = self.session.create_client('s3', region_name=params.region)
@@ -43,13 +138,41 @@ class Windows(System):
     CONFIG_FILE = 'conf.onpremises.yml'
     CONFIG_PATH = rf'{CONFIG_DIR}\{CONFIG_FILE}'
     INSTALLER = 'codedeploy-agent.msi'
+    # An existing file is replaced rather than truncated, which would keep
+    # its ACL and any handles opened to it before.
+    RECREATE_CONFIG_FILE = True
 
     def validate_administrator(self):
         if not ctypes.windll.shell32.IsUserAnAdmin():
             raise RuntimeError(
                 'You must run this command as an Administrator.'
             )
 
+    def create_config_file(self):
+        """Replace CONFIG_PATH with a new, empty file and return its fd.
+
+        The new file grants access only to LocalSystem and Administrators,
+        whatever the ACL of CONFIG_DIR.
+        """
+        try:
+            # os.remove fails on read-only files, so clear the attribute.
+            os.chmod(self.CONFIG_PATH, stat.S_IWRITE)
+            os.remove(self.CONFIG_PATH)
+        except FileNotFoundError:
+            pass
+        except PermissionError as e:
+            raise RuntimeError(
+                'Failed to replace the existing on-premises instance '
+                f'configuration file {self.CONFIG_PATH}: {e}. Make sure no '
+                'other process has the file open, then run install again.'
+            )
+        try:
+            return _create_windows_file(
+                self.CONFIG_PATH, _WINDOWS_CONFIG_FILE_SDDL
+            )
+        except OSError as e:
+            raise RuntimeError(f'Failed to create {self.CONFIG_PATH}: {e}')
+
     def install(self, params):
         if 'installer' in params:
             self.INSTALLER = params.installer