[VM Repair] Fix Python 3.12+ compatibility: replace deprecated datetime.utcnow()

src/vm-repair/HISTORY.rst

@@ -2,6 +2,10 @@
 Release History
 ===============
 
+2.2.2
+++++++
+Replacing deprecated ``datetime.utcnow()`` with timezone-aware ``datetime.now(timezone.utc)`` for Python 3.12+ forward compatibility. ``datetime.utcnow()`` is deprecated as of Python 3.12 and scheduled for removal in a future release. The generated timestamps (used for repair VM, copied disk, and repair resource group names) are unchanged. Also replacing ``pkgutil.get_loader()``/``loader.load_module()`` (deprecated in Python 3.12, removed in Python 3.14) with ``importlib.util.find_spec()`` when locating the bundled driver scripts, and extending the static Python 3.12+ compatibility guard to cover these APIs.
+
 2.2.1
 ++++++
 Fixing a command injection vulnerability (MSRC 115198 / VULN-185362). Source VM tag values copied via ``--copy-tags`` could contain shell metacharacters that, on Windows, were interpreted by ``cmd.exe`` and executed as arbitrary commands on the operator's workstation. Tag keys and values are now validated and quoted before being interpolated into the ``az`` command, and ``_call_az_command`` quotes every argument so ``cmd.exe`` treats shell metacharacters as literal text. Minimum fixed version: 2.2.1.

src/vm-repair/azext_vm_repair/_validators.py

@@ -3,7 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
-from datetime import datetime
+from datetime import datetime, timezone
 from json import loads
 from re import match, search, findall
 from knack.log import get_logger
@@ -46,7 +46,7 @@ def validate_create(cmd, namespace):
         namespace.repair_vm_name = ('repair-' + namespace.vm_name)[:14] + '_'
 
     # Check copy disk name
-    timestamp = datetime.utcnow().strftime('%Y%m%d%H%M%S')
+    timestamp = datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')
     if namespace.copy_disk_name:
         _validate_disk_name(namespace.copy_disk_name)
     else:
@@ -400,7 +400,7 @@ def validate_repair_and_restore(cmd, namespace):
     logger.info('Repair VM name: %s', namespace.repair_vm_name)
 
     # Check copy disk name
-    timestamp = datetime.utcnow().strftime('%Y%m%d%H%M%S')
+    timestamp = datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')
     if namespace.copy_disk_name:
         _validate_disk_name(namespace.copy_disk_name)
     else:

src/vm-repair/azext_vm_repair/custom.py

@@ -983,7 +983,7 @@ def repair_and_restore(cmd, vm_name, resource_group_name, repair_password=None,
     :param copy_tags: (Optional) Boolean indicating whether to copy tags from the source VM to the repair VM.
     :param size: (Optional) The size of the repair VM.
     """
-    from datetime import datetime
+    from datetime import datetime, timezone
     import secrets
     import string
 
@@ -1001,7 +1001,7 @@ def repair_and_restore(cmd, vm_name, resource_group_name, repair_password=None,
     repair_username = ''.join(secrets.choice(username_characters) for _ in range(username_length))
 
     # Generate unique names for the repair VM, copied disk, and repair resource group
-    timestamp = datetime.utcnow().strftime('%Y%m%d%H%M%S')
+    timestamp = datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')
     repair_vm_name = ('repair-' + vm_name)[:14] + '_'
     copy_disk_name = vm_name + '-DiskCopy-' + timestamp
     repair_group_name = 'repair-' + vm_name + '-' + timestamp
@@ -1088,7 +1088,7 @@ def repair_button(cmd, vm_name, resource_group_name, button_command, repair_pass
     """
     Button-triggered repair operation. Supports tags for the repair VM.
     """
-    from datetime import datetime
+    from datetime import datetime, timezone
     import secrets
     import string
 
@@ -1107,7 +1107,7 @@ def repair_button(cmd, vm_name, resource_group_name, button_command, repair_pass
     username_characters = string.ascii_lowercase + string.digits
     repair_username = ''.join(secrets.choice(username_characters) for i in range(username_length))
 
-    timestamp = datetime.utcnow().strftime('%Y%m%d%H%M%S')
+    timestamp = datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')
     repair_vm_name = ('repair-' + vm_name)[:14] + '_'
     copy_disk_name = vm_name + '-DiskCopy-' + timestamp
     repair_group_name = 'repair-' + vm_name + '-' + timestamp

src/vm-repair/azext_vm_repair/repair_utils.py

@@ -9,7 +9,7 @@
 import os
 import re
 from json import loads
-import pkgutil
+import importlib.util
 import requests
 
 from knack.log import get_logger
@@ -30,9 +30,8 @@ def _get_cloud_init_script():
     SCRIPTS_DIR_NAME = 'scripts'
     CLOUD_INIT = 'linux-build_setup-cloud-init.txt'
     # Build absoulte path of driver script
-    loader = pkgutil.get_loader(REPAIR_DIR_NAME)
-    mod = loader.load_module(REPAIR_DIR_NAME)
-    rootpath = os.path.dirname(mod.__file__)
+    mod_spec = importlib.util.find_spec(REPAIR_DIR_NAME)
+    rootpath = os.path.dirname(mod_spec.origin)
     return os.path.join(rootpath, SCRIPTS_DIR_NAME, CLOUD_INIT)
 
 
@@ -182,9 +181,8 @@ def _invoke_run_command(script_name, vm_name, rg_name, is_linux, parameters=None
     RUN_COMMAND_RUN_PS_ID = 'RunPowerShellScript'
 
     # Build absoulte path of driver script
-    loader = pkgutil.get_loader(REPAIR_DIR_NAME)
-    mod = loader.load_module(REPAIR_DIR_NAME)
-    rootpath = os.path.dirname(mod.__file__)
+    mod_spec = importlib.util.find_spec(REPAIR_DIR_NAME)
+    rootpath = os.path.dirname(mod_spec.origin)
     run_script = os.path.join(rootpath, SCRIPTS_DIR_NAME, script_name)
 
     if is_linux:

src/vm-repair/azext_vm_repair/tests/latest/test_py312_compat.py

@@ -0,0 +1,71 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import os
+import re
+import unittest
+
+
+class Python312CompatTests(unittest.TestCase):
+    """Guard against (re)introducing Python APIs removed or deprecated in Python 3.12+.
+
+    This is a static, dependency-free scan of the shipped extension source (the
+    ``azext_vm_repair`` package, excluding the ``tests`` tree). It fails if any
+    banned pattern is present, which keeps the extension forward-compatible with
+    Python 3.12 and newer.
+    """
+
+    # azext_vm_repair package root. This file lives at
+    # azext_vm_repair/tests/latest/test_py312_compat.py, so three dirname() calls
+    # walk up to the azext_vm_repair/ package directory.
+    PACKAGE_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+    # (compiled regex, human-readable reason) for patterns banned in shipped code.
+    BANNED_PATTERNS = [
+        (re.compile(r'\bdatetime\.utcnow\s*\('),
+         'datetime.utcnow() is deprecated in Python 3.12; use datetime.now(timezone.utc)'),
+        (re.compile(r'\.utcfromtimestamp\s*\('),
+         'datetime.utcfromtimestamp() is deprecated in Python 3.12; use datetime.fromtimestamp(ts, timezone.utc)'),
+        (re.compile(r'\bimport\s+imp\b'),
+         "the 'imp' module was removed in Python 3.12; use importlib"),
+        (re.compile(r'\bimport\s+asyncore\b'),
+         "the 'asyncore' module was removed in Python 3.12; use asyncio"),
+        (re.compile(r'\bimport\s+asynchat\b'),
+         "the 'asynchat' module was removed in Python 3.12; use asyncio"),
+        (re.compile(r'\bfrom\s+distutils\b|\bimport\s+distutils\b'),
+         "'distutils' was removed in Python 3.12; use the 'packaging' library"),
+        (re.compile(r'\bpkgutil\.(?:get_loader|find_loader|ImpImporter)\b'),
+         'pkgutil.get_loader/find_loader/ImpImporter are deprecated in Python 3.12 and '
+         'removed in 3.14; use importlib.util (e.g. importlib.util.find_spec)'),
+        (re.compile(r'\bplatform\.dist\s*\('),
+         "platform.dist() was removed in Python 3.8; use the 'distro' library"),
+    ]
+
+    def _iter_source_files(self):
+        for root, _, files in os.walk(self.PACKAGE_ROOT):
+            # Only scan shipped code, not the test tree itself.
+            if 'tests' in root.split(os.sep):
+                continue
+            for name in files:
+                if name.endswith('.py'):
+                    yield os.path.join(root, name)
+
+    def test_no_python312_removed_or_deprecated_apis(self):
+        offenders = []
+        for path in self._iter_source_files():
+            with open(path, encoding='utf-8') as handle:
+                for lineno, line in enumerate(handle, start=1):
+                    for pattern, reason in self.BANNED_PATTERNS:
+                        if pattern.search(line):
+                            rel = os.path.relpath(path, self.PACKAGE_ROOT)
+                            offenders.append('{}:{}: {}'.format(rel, lineno, reason))
+        self.assertEqual(
+            offenders, [],
+            'Found Python 3.12+ incompatible API usage in shipped code:\n' + '\n'.join(offenders)
+        )
+
+
+if __name__ == '__main__':
+    unittest.main()

src/vm-repair/setup.py

@@ -8,19 +8,18 @@
 from codecs import open
 from setuptools import setup, find_packages
 
-VERSION = "2.2.1"
+VERSION = "2.2.2"
 
 CLASSIFIERS = [
     'Development Status :: 4 - Beta',
     'Intended Audience :: Developers',
     'Intended Audience :: System Administrators',
     'Programming Language :: Python',
-    'Programming Language :: Python :: 2',
-    'Programming Language :: Python :: 2.7',
     'Programming Language :: Python :: 3',
-    'Programming Language :: Python :: 3.4',
-    'Programming Language :: Python :: 3.5',
-    'Programming Language :: Python :: 3.6',
+    'Programming Language :: Python :: 3.9',
+    'Programming Language :: Python :: 3.10',
+    'Programming Language :: Python :: 3.11',
+    'Programming Language :: Python :: 3.12',
     'Programming Language :: Python :: 3.13',
     'License :: OSI Approved :: MIT License',
 ]