mirror of https://github.com/python/cpython
gh-96684: Silently suppress COM security errors in _wmi module (GH-96690)
This commit is contained in:
parent
9c8f379433
commit
95d6330a3e
|
@ -1,10 +1,9 @@
|
||||||
# Test the internal _wmi module on Windows
|
# Test the internal _wmi module on Windows
|
||||||
# This is used by the platform module, and potentially others
|
# This is used by the platform module, and potentially others
|
||||||
|
|
||||||
import re
|
|
||||||
import sys
|
import sys
|
||||||
import unittest
|
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
|
# Do this first so test will be skipped if module doesn't exist
|
||||||
|
@ -20,7 +19,7 @@ class WmiTests(unittest.TestCase):
|
||||||
self.assertEqual("Version", k, r[0])
|
self.assertEqual("Version", k, r[0])
|
||||||
# Best we can check for the version is that it's digits, dot, digits, anything
|
# 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
|
# 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):
|
def test_wmi_query_repeated(self):
|
||||||
# Repeated queries should not break
|
# Repeated queries should not break
|
||||||
|
@ -46,6 +45,7 @@ class WmiTests(unittest.TestCase):
|
||||||
with self.assertRaises(ValueError):
|
with self.assertRaises(ValueError):
|
||||||
_wmi.exec_query("not select, just in case someone tries something")
|
_wmi.exec_query("not select, just in case someone tries something")
|
||||||
|
|
||||||
|
@requires_resource('cpu')
|
||||||
def test_wmi_query_overflow(self):
|
def test_wmi_query_overflow(self):
|
||||||
# Ensure very big queries fail
|
# Ensure very big queries fail
|
||||||
# Test multiple times to ensure consistency
|
# Test multiple times to ensure consistency
|
||||||
|
@ -61,7 +61,15 @@ class WmiTests(unittest.TestCase):
|
||||||
it = iter(r.split("\0"))
|
it = iter(r.split("\0"))
|
||||||
try:
|
try:
|
||||||
while True:
|
while True:
|
||||||
self.assertTrue(re.match(r"ProcessId=\d+", next(it)))
|
self.assertRegex(next(it), r"ProcessId=\d+")
|
||||||
self.assertEqual("", next(it))
|
self.assertEqual("", next(it))
|
||||||
except StopIteration:
|
except StopIteration:
|
||||||
pass
|
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=")
|
||||||
|
|
|
@ -63,6 +63,12 @@ _query_thread(LPVOID param)
|
||||||
RPC_C_IMP_LEVEL_IMPERSONATE,
|
RPC_C_IMP_LEVEL_IMPERSONATE,
|
||||||
NULL, EOAC_NONE, NULL
|
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)) {
|
if (SUCCEEDED(hr)) {
|
||||||
hr = CoCreateInstance(
|
hr = CoCreateInstance(
|
||||||
CLSID_WbemLocator, 0, CLSCTX_INPROC_SERVER,
|
CLSID_WbemLocator, 0, CLSCTX_INPROC_SERVER,
|
||||||
|
|
Loading…
Reference in New Issue