fix(job-attachments): remove dependency on pywin32 for submission code (

#250)

Signed-off-by: Jericho Tolentino <68654047+jericht@users.noreply.github.com>

pyproject.toml

@@ -116,6 +116,12 @@ known-first-party = [
   "deadline"
 ]
 
+[tool.ruff.lint.per-file-ignores]
+# We need to use a platform assertion to short-circuit mypy type checking on non-Windows platforms
+# https://mypy.readthedocs.io/en/stable/common_issues.html#python-version-and-system-platform-checks
+# This causes imports to come after regular Python statements causing flake8 rule E402 to be flagged
+"src/deadline/job_attachments/_windows/*.py" = ["E402"]
+
 [tool.black]
 line-length = 100
 
@@ -156,6 +162,9 @@ source_pkgs = [ "deadline" ]
 omit = [
     "*/deadline/client/ui/*",
 ]
+plugins = [
+  "coverage_conditional_plugin"
+]
 
 [tool.coverage.paths]
 source = [ "src/" ]
@@ -164,6 +173,24 @@ source = [ "src/" ]
 show_missing = true
 fail_under = 80
 
+# https://github.com/wemake-services/coverage-conditional-plugin
+[tool.coverage.coverage_conditional_plugin.omit]
+"sys_platform != 'win32'" = [
+  "src/deadline/job_attachments/_windows/*.py",
+]
+
+[tool.coverage.coverage_conditional_plugin.rules]
+# This cannot be empty otherwise coverage-conditional-plugin crashes with:
+# AttributeError: 'NoneType' object has no attribute 'items'
+#
+# =========== WARNING TO REVIEWERS ============
+#
+# Any rules added here are ran through Python's
+# eval() function so watch for code injection
+# attacks.
+#
+# =========== WARNING TO REVIEWERS ============
+
 [tool.semantic_release]
 # Can be removed or set to true once we are v1
 major_on_zero = false

requirements-testing.txt

@@ -1,5 +1,6 @@
 coverage[toml] ~= 7.2; python_version == '3.7'
 coverage[toml] == 7.*; python_version > '3.7'
+coverage-conditional-plugin == 0.9.*
 pytest == 7.*
 pytest-cov == 4.*
 pytest-timeout == 2.*

src/deadline/job_attachments/_windows/__init__.py

@@ -0,0 +1 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.

src/deadline/job_attachments/_windows/file.py

@@ -0,0 +1,26 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+import ctypes
+import ctypes.wintypes
+import sys
+
+# This assertion short-circuits mypy from type checking this module on platforms other than Windows
+# https://mypy.readthedocs.io/en/stable/common_issues.html#python-version-and-system-platform-checks
+assert sys.platform == "win32"
+
+kernel32 = ctypes.WinDLL("Kernel32")
+
+# https://learn.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-getfinalpathnamebyhandlew
+kernel32.GetFinalPathNameByHandleW.restype = ctypes.wintypes.DWORD
+kernel32.GetFinalPathNameByHandleW.argtypes = [
+    ctypes.wintypes.HANDLE,  # [in]  HANDLE hFile,
+    ctypes.wintypes.LPWSTR,  # [out] LPWSTR lpszFilePath,
+    ctypes.wintypes.DWORD,  # [in]  DWORD  cchFilePath,
+    ctypes.wintypes.DWORD,  # [in]  DWORD  dwFlags
+]
+GetFinalPathNameByHandleW = kernel32.GetFinalPathNameByHandleW
+
+VOLUME_NAME_DOS = 0
+VOLUME_NAME_GUID = 1
+VOLUME_NAME_NONE = 4
+VOLUME_NAME_NT = 2

src/deadline/job_attachments/upload.py

@@ -491,18 +491,54 @@ def _is_path_win32_final_path_of_file_descriptor(self, path: str, fd: int):
         if sys.platform != "win32":
             raise EnvironmentError("This function can only be executed on Windows systems.")
 
-        import win32con
-        import win32file
+        import ctypes
+        import msvcrt
+        from ._windows import file as win_file
 
-        if sys.platform != "win32" and sys.getwindowsversion()[:2] >= (6, 0):
-            from nt import _getfinalpathname  # type: ignore[attr-defined]
-        else:
-            _getfinalpathname = None
+        # Get the handle from the file descriptor
+        try:
+            h = msvcrt.get_osfhandle(fd)
+        except OSError as e:
+            logger.warning(f"Error resolving file descriptor ({fd}) to '{path}': {e}")
+            return False
+
+        # Get the final path name using Win32 API GetFinalPathNameByHandleW
+        buffer_len = 4096
+        buffer = ctypes.create_unicode_buffer(buffer_len)
+        path_len = win_file.GetFinalPathNameByHandleW(
+            h,
+            buffer,
+            buffer_len,
+            win_file.VOLUME_NAME_DOS,
+        )
+        if path_len == 0:
+            ctypes.WinError()
+        elif path_len > buffer_len:
+            # path_len has the required buffer length (returned by GetFinalPathNameByHandleW)
+            # Create a buffer of this size and call the API again
+            buffer_len = path_len
+            buffer = ctypes.create_unicode_buffer(buffer_len)
+            path_len = win_file.GetFinalPathNameByHandleW(
+                h,
+                buffer,
+                buffer_len,
+                win_file.VOLUME_NAME_DOS,
+            )
+
+            if path_len != buffer_len or path_len == 0:
+                # MS documentation states that if GetFinalPathNameByHandleW returns a positive value
+                # greater than the initial buffer length, it is the required buffer length to fit the
+                # path name. This branch uses the that value to create a new buffer, so this should
+                # never fail unless GetFinalPathNameByHandleW behavior has changed.
+                logger.error(
+                    "GetFinalPathNameByHandleW reported incorrect required buffer length. "
+                    f"Rejecting file at '{path}'"
+                )
+                return False
 
-        h = win32file._get_osfhandle(fd)
-        final_path = win32file.GetFinalPathNameByHandle(h, win32con.VOLUME_NAME_DOS)
+        final_path = ctypes.wstring_at(buffer)
 
-        # GetFinalPathNameByHandle() returns a path that starts with the \\?\
+        # GetFinalPathNameByHandleW() returns a path that starts with the \\?\
         # prefix, which pathlib.Path.resolve() removes.  The following is intended
         # to match the behavior of resolve().
         prefix = r"\\?" "\\"
@@ -514,10 +550,7 @@ def _is_path_win32_final_path_of_file_descriptor(self, path: str, fd: int):
             else:
                 simplified_path = final_path[len(prefix) :]
 
-            if _getfinalpathname and _getfinalpathname(simplified_path) == final_path:
-                final_path = simplified_path
-            if _getfinalpathname is None:
-                final_path = simplified_path
+            final_path = simplified_path
 
         return path == final_path