[Python-checkins] gh-96684: Silently suppress COM security errors in _wmi module (GH-96690)

[zooba]

https://github.com/python/cpython/commit/95d6330a3edacacd081f55699bd6f0a787908924
commit: 95d6330a3edacacd081f55699bd6f0a787908924
branch: main
author: Steve Dower <steve.dower at python.org>
committer: zooba <steve.dower at microsoft.com>
date: 2022-09-08T22:02:04+01:00
summary:

gh-96684: Silently suppress COM security errors in _wmi module (GH-96690)

files:
M Lib/test/test_wmi.py
M PC/_wmimodule.cpp

diff --git a/Lib/test/test_wmi.py b/Lib/test/test_wmi.py
index af2a453dcb52..3f5795952905 100644
--- a/Lib/test/test_wmi.py
+++ b/Lib/test/test_wmi.py
@@ -1,10 +1,9 @@
 # Test the internal _wmi module on Windows
 # This is used by the platform module, and potentially others
 
-import re
 import sys
 import unittest
-from test.support import import_helper
+from test.support import import_helper, requires_resource
 
 
 # Do this first so test will be skipped if module doesn't exist
@@ -20,7 +19,7 @@ def test_wmi_query_os_version(self):
         self.assertEqual("Version", k, r[0])
         # Best we can check for the version is that it's digits, dot, digits, anything
         # Otherwise, we are likely checking the result of the query against itself
-        self.assertTrue(re.match(r"\d+\.\d+.+$", v), r[0])
+        self.assertRegex(v, r"\d+\.\d+.+$", r[0])
 
     def test_wmi_query_repeated(self):
         # Repeated queries should not break
@@ -46,6 +45,7 @@ def test_wmi_query_not_select(self):
         with self.assertRaises(ValueError):
             _wmi.exec_query("not select, just in case someone tries something")
 
+    @requires_resource('cpu')
     def test_wmi_query_overflow(self):
         # Ensure very big queries fail
         # Test multiple times to ensure consistency
@@ -61,7 +61,15 @@ def test_wmi_query_multiple_rows(self):
         it = iter(r.split("\0"))
         try:
             while True:
-                self.assertTrue(re.match(r"ProcessId=\d+", next(it)))
+                self.assertRegex(next(it), r"ProcessId=\d+")
                 self.assertEqual("", next(it))
         except StopIteration:
             pass
+
+    def test_wmi_query_threads(self):
+        from concurrent.futures import ThreadPoolExecutor
+        query = "SELECT ProcessId FROM Win32_Process WHERE ProcessId < 1000"
+        with ThreadPoolExecutor(4) as pool:
+            task = [pool.submit(_wmi.exec_query, query) for _ in range(32)]
+            for t in task:
+                self.assertRegex(t.result(), "ProcessId=")
diff --git a/PC/_wmimodule.cpp b/PC/_wmimodule.cpp
index fbaadcc51100..de22049dd33f 100644
--- a/PC/_wmimodule.cpp
+++ b/PC/_wmimodule.cpp
@@ -63,6 +63,12 @@ _query_thread(LPVOID param)
         RPC_C_IMP_LEVEL_IMPERSONATE,
         NULL, EOAC_NONE, NULL
     );
+    // gh-96684: CoInitializeSecurity will fail if another part of the app has
+    // already called it. Hopefully they passed lenient enough settings that we
+    // can complete the WMI query, so keep going.
+    if (hr == RPC_E_TOO_LATE) {
+        hr = 0;
+    }
     if (SUCCEEDED(hr)) {
         hr = CoCreateInstance(
             CLSID_WbemLocator, 0, CLSCTX_INPROC_SERVER,