Merge pull request from GHSA-72mj-mqhj-qh4w

GHSA-72mj-mqhj-qh4w

Session tracking allows NVDA notifies of session lock state
changes. However, a transition is required to set the initial
state. This approach relied on the assumption that NVDA can't
start while already on the lock screen. While the assumption is
typically valid, it is possible (but unlikely) for NVDA to start
after the session is locked.

No user observable changes.

Initialize the lock state by querying the session info.

Author: feerrenrut

source/core.py

@@ -580,35 +580,16 @@ def __init__(self, windowName=None):
 			self.handlePowerStatusChange()
 
 			# Call must be paired with a call to sessionTracking.unregister
-			self._isSessionTrackingRegistered = sessionTracking.register(self.handle)
-
-		def warnIfSessionTrackingNotRegistered(self) -> None:
-			if self._isSessionTrackingRegistered:
-				return
-			failedToRegisterMsg = _(
-				# Translators: This is a warning to users, shown if NVDA cannot determine if
-				# Windows is locked.
-				"NVDA failed to register session tracking. "
-				"While this instance of NVDA is running, "
-				"your desktop will not be secure when Windows is locked. "
-				"Restart NVDA? "
-			)
-			if wx.YES == gui.messageBox(
-				failedToRegisterMsg,
-				# Translators: This is a warning to users, shown if NVDA cannot determine if
-				# Windows is locked.
-				caption=_("NVDA could not start securely."),
-				style=wx.ICON_ERROR | wx.YES_NO,
-			):
-				restart()
+			if not sessionTracking.register(self.handle):
+				import utils.security
+				wx.CallAfter(utils.security.warnSessionLockStateUnknown)
 
 		def destroy(self):
 			"""
 			NVDA must unregister session tracking before destroying the message window.
 			"""
-			if self._isSessionTrackingRegistered:
-				# Requires an active message window and a handle to unregister.
-				sessionTracking.unregister(self.handle)
+			# Requires an active message window and a handle to unregister.
+			sessionTracking.unregister(self.handle)
 			super().destroy()
 
 		def windowProc(self, hwnd, msg, wParam, lParam):
@@ -618,7 +599,6 @@ def windowProc(self, hwnd, msg, wParam, lParam):
 			elif msg == winUser.WM_DISPLAYCHANGE:
 				self.handleScreenOrientationChange(lParam)
 			elif msg == WindowMessage.WTS_SESSION_CHANGE:
-				# If we are receiving WTS_SESSION_CHANGE events, _isSessionTrackingRegistered should be True
 				sessionTracking.handleSessionChange(sessionTracking.WindowsTrackedSession(wParam), lParam)
 
 		def handleScreenOrientationChange(self, lParam):
@@ -818,8 +798,6 @@ def _doPostNvdaStartupAction():
 	queueHandler.queueFunction(queueHandler.eventQueue, _doPostNvdaStartupAction)
 
 	log.debug("entering wx application main loop")
-	# Warn here as NVDA must be ready before providing a gui message
-	messageWindow.warnIfSessionTrackingNotRegistered()
 	app.MainLoop()
 
 	log.info("Exiting")

source/utils/security.py

@@ -170,3 +170,50 @@ def isObjectAboveLockScreen(obj: "NVDAObjects.NVDAObject") -> bool:
 		return True
 
 	return False
+
+
+_hasSessionLockStateUnknownWarningBeenGiven = False
+"""Track whether the user has been notified.
+"""
+
+
+def warnSessionLockStateUnknown() -> None:
+	""" Warn the user that the lock state of the computer can not be determined.
+	NVDA will not be able to determine if Windows is on the lock screen
+	(LockApp on Windows 10/11), and will not be able to ensure privacy/security
+	of the signed-in user against unauthenticated users.
+	@note Only warn the user once.
+	"""
+	global _hasSessionLockStateUnknownWarningBeenGiven
+	if _hasSessionLockStateUnknownWarningBeenGiven:
+		return
+	_hasSessionLockStateUnknownWarningBeenGiven = True
+
+	log.warning(
+		"NVDA is unable to determine if Windows is locked."
+		" While this instance of NVDA is running,"
+		" your desktop will not be secure when Windows is locked."
+		" Restarting Windows may address this."
+		" If this error is ongoing then disabling the Windows lock screen is recommended."
+	)
+
+	unableToDetermineSessionLockStateMsg = _(
+		# Translators: This is the message for a warning shown if NVDA cannot determine if
+		# Windows is locked.
+		"NVDA is unable to determine if Windows is locked."
+		" While this instance of NVDA is running,"
+		" your desktop will not be secure when Windows is locked."
+		" Restarting Windows may address this."
+		" If this error is ongoing then disabling the Windows lock screen is recommended."
+	)
+
+	import wx  # Late import to prevent circular dependency.
+	import gui  # Late import to prevent circular dependency.
+	log.debug("Presenting session lock tracking failure warning.")
+	gui.messageBox(
+		unableToDetermineSessionLockStateMsg,
+		# Translators: This is the title for a warning dialog, shown if NVDA cannot determine if
+		# Windows is locked.
+		caption=_("Lock screen not secure while using NVDA"),
+		style=wx.ICON_ERROR | wx.OK,
+	)

source/winAPI/sessionTracking.py

@@ -16,7 +16,10 @@
 https://docs.microsoft.com/en-us/windows/win32/api/wtsapi32/nf-wtsapi32-wtsregistersessionnotification
 """
 
+from __future__ import annotations
+import contextlib
 import ctypes
+from contextlib import contextmanager
 from ctypes.wintypes import (
 	HANDLE,
 	HWND,
@@ -25,17 +28,36 @@
 from typing import (
 	Dict,
 	Set,
+	Optional,
+)
+from winAPI.wtsApi32 import (
+	WTSINFOEXW,
+	WTSQuerySessionInformation,
+	WTS_CURRENT_SERVER_HANDLE,
+	WTS_CURRENT_SESSION,
+	WTS_INFO_CLASS,
+	WTSFreeMemory,
+	WTS_LockState,
+	WTSINFOEX_LEVEL1_W,
 )
-
 from logHandler import log
 
-
 _currentSessionStates: Set["WindowsTrackedSession"] = set()
 """
 Current state of the Windows session associated with this instance of NVDA.
 Maintained via receiving session notifications via the NVDA MessageWindow.
+Initial state will be set by querying the current status.
 """
 
+_sessionQueryLockStateHasBeenUnknown = False
+"""
+Track if any 'Unknown' Value when querying the Session Lock status has been encountered.
+"""
+
+_isSessionTrackingRegistered = False
+"""
+Session tracking is required for NVDA to be notified of lock state changes for security purposes.
+"""
 
 RPC_S_INVALID_BINDING = 0x6A6
 """
@@ -61,7 +83,8 @@ class WindowsTrackedSession(enum.IntEnum):
 	Windows Tracked Session notifications.
 	Members are states which form logical pairs,
 	except SESSION_REMOTE_CONTROL which requires more complex handling.
-
+	Values from: https://learn.microsoft.com/en-us/windows/win32/termserv/wm-wtssession-change
+	Context:
 	https://docs.microsoft.com/en-us/windows/win32/api/wtsapi32/nf-wtsapi32-wtsregistersessionnotification
 	"""
 	CONSOLE_CONNECT = 1
@@ -95,13 +118,91 @@ class WindowsTrackedSession(enum.IntEnum):
 """
 
 
+def _hasLockStateBeenTracked() -> bool:
+	"""
+	Checks if NVDA is aware of a session lock state change since NVDA started.
+	"""
+	return bool(_currentSessionStates.intersection({
+		WindowsTrackedSession.SESSION_LOCK,
+		WindowsTrackedSession.SESSION_UNLOCK
+	}))
+
+
+def _recordLockStateTrackingFailure(error: Optional[Exception] = None):
+	log.error(
+		"Unknown lock state, unexpected, potential security issue, please report.",
+		exc_info=error
+	)  # Report error repeatedly, attention is required.
+	##
+	# For security it would be best to treat unknown as locked.
+	# However, this would make NVDA unusable.
+	# Instead, the user should be warned, and allowed to mitigate the impact themselves.
+	# Reporting is achieved via _sessionQueryLockStateHasBeenUnknown exposed with
+	# L{hasLockStateBeenUnknown}.
+	global _sessionQueryLockStateHasBeenUnknown
+	_sessionQueryLockStateHasBeenUnknown = True
+
+
 def isWindowsLocked() -> bool:
 	"""
 	Checks if the Window lockscreen is active.
-
 	Not to be confused with the Windows sign-in screen, a secure screen.
 	"""
-	return WindowsTrackedSession.SESSION_LOCK in _currentSessionStates
+	lockStateTracked = _hasLockStateBeenTracked()
+	if lockStateTracked:
+		return WindowsTrackedSession.SESSION_LOCK in _currentSessionStates
+	else:
+		_recordLockStateTrackingFailure()  # Report error repeatedly, attention is required.
+		##
+		# For security it would be best to treat unknown as locked.
+		# However, this would make NVDA unusable.
+		# Instead, the user should be warned via UI, and allowed to mitigate the impact themselves.
+		# See usage of L{hasLockStateBeenUnknown}.
+		return False  # return False, indicating unlocked, to allow NVDA to be used
+
+
+def _setInitialWindowLockState() -> None:
+	"""Ensure that session tracking state is initialized.
+	If NVDA has started on a lockScreen, it needs to be aware of this.
+	"""
+	lockStateTracked = _hasLockStateBeenTracked()
+	if lockStateTracked:
+		raise RuntimeError("Can't set initial state if it is already tracked.")
+	# Fall back to explicit query
+	try:
+		isLocked = _isWindowsLocked_checkViaSessionQuery()
+		_currentSessionStates.add(
+			WindowsTrackedSession.SESSION_LOCK
+			if isLocked
+			else WindowsTrackedSession.SESSION_UNLOCK
+		)
+	except RuntimeError as error:
+		_recordLockStateTrackingFailure(error)
+
+
+def _isWindowsLocked_checkViaSessionQuery() -> bool:
+	""" Use a session query to check if the session is locked
+	@return: True is the session is locked.
+	@raise: Runtime error if the lock state can not be determined via a Session Query.
+	"""
+	sessionQueryLockState = _getSessionLockedValue()
+	if sessionQueryLockState == WTS_LockState.WTS_SESSIONSTATE_UNKNOWN:
+		raise RuntimeError(
+			"Unable to determine lock state via Session Query."
+			f" Lock state value: {sessionQueryLockState!r}"
+		)
+	return sessionQueryLockState == WTS_LockState.WTS_SESSIONSTATE_LOCK
+
+
+def isLockStateSuccessfullyTracked() -> bool:
+	"""Check if the lock state is successfully tracked.
+	I.E. Registered for session tracking AND initial value set correctly.
+	@return: True when successfully tracked.
+	"""
+	return (
+		not _sessionQueryLockStateHasBeenUnknown
+		or not _isSessionTrackingRegistered
+	)
 
 
 def register(handle: HWND) -> bool:
@@ -121,6 +222,11 @@ def register(handle: HWND) -> bool:
 
 	https://docs.microsoft.com/en-us/windows/win32/api/wtsapi32/nf-wtsapi32-wtsregistersessionnotification
 	"""
+	##
+	# Ensure that an initial state is set,
+	# do this first to prevent a race condition with session tracking / message processing.
+	_setInitialWindowLockState()
+
 	# OpenEvent handle must be closed with CloseHandle.
 	eventObjectHandle: HANDLE = ctypes.windll.kernel32.OpenEventW(
 		# Blocks until WTS session tracking can be registered.
@@ -154,7 +260,9 @@ def register(handle: HWND) -> bool:
 		else:
 			log.error("Unexpected error registering session tracking.", exc_info=error)
 
-	return registrationSuccess
+	global _isSessionTrackingRegistered
+	_isSessionTrackingRegistered = registrationSuccess
+	return isLockStateSuccessfullyTracked()
 
 
 def unregister(handle: HWND) -> None:
@@ -164,6 +272,9 @@ def unregister(handle: HWND) -> None:
 
 	https://docs.microsoft.com/en-us/windows/win32/api/wtsapi32/nf-wtsapi32-wtsunregistersessionnotification
 	"""
+	if not _isSessionTrackingRegistered:
+		log.info("Not unregistered session tracking, it was not registered.")
+		return
 	if ctypes.windll.wtsapi32.WTSUnRegisterSessionNotification(handle):
 		log.debug("Unregistered session tracking")
 	else:
@@ -187,6 +298,9 @@ def handleSessionChange(newState: WindowsTrackedSession, sessionId: int) -> None
 
 	log.debug(f"Windows Session state notification received: {newState.name}")
 
+	if not _isSessionTrackingRegistered:
+		log.debugWarning("Session tracking not registered, unexpected session change message")
+
 	if newState not in _toggleWindowsSessionStatePair:
 		log.debug(f"Ignoring {newState} event as tracking is not required.")
 		return
@@ -207,3 +321,81 @@ def handleSessionChange(newState: WindowsTrackedSession, sessionId: int) -> None
 		log.debug(f"NVDA started in state {oppositeState.name} or dropped a state change event")
 
 	log.debug(f"New Windows Session state: {_currentSessionStates}")
+
+
+@contextmanager
+def WTSCurrentSessionInfoEx() -> contextlib.AbstractContextManager[ctypes.pointer[WTSINFOEXW]]:
+	"""Context manager to get the WTSINFOEXW for the current server/session or raises a RuntimeError.
+	Handles freeing the memory when usage is complete.
+	"""
+	info = _getCurrentSessionInfoEx()
+	try:
+		yield info
+	finally:
+		WTSFreeMemory(
+			ctypes.cast(info, ctypes.c_void_p),
+		)
+
+
+def _getCurrentSessionInfoEx() -> ctypes.POINTER(WTSINFOEXW):
+	""" Gets the WTSINFOEXW for the current server/session or raises a RuntimeError
+	on failure.
+	On RuntimeError memory is first freed.
+	In other cases use WTSFreeMemory.
+	Ideally use the WTSCurrentSessionInfoEx context manager which will handle freeing the memory.
+	"""
+	ppBuffer = ctypes.wintypes.LPWSTR(None)
+	pBytesReturned = ctypes.wintypes.DWORD(0)
+
+	res = WTSQuerySessionInformation(
+		WTS_CURRENT_SERVER_HANDLE,  # WTS_CURRENT_SERVER_HANDLE to indicate the RD Session Host server on
+		# which the application is running.
+		WTS_CURRENT_SESSION,  # To indicate the session in which the calling application is running
+		# (or the current session) specify WTS_CURRENT_SESSION
+		WTS_INFO_CLASS.WTSSessionInfoEx,  # Indicates the type of session information to retrieve
+		# Fetch a WTSINFOEXW containing a WTSINFOEX_LEVEL1 structure.
+		ctypes.pointer(ppBuffer),  # A pointer to a variable that receives a pointer to the requested information.
+		# The format and contents of the data depend on the information class specified in the WTSInfoClass
+		# parameter.
+		# To free the returned buffer, call the WTSFreeMemory function.
+		ctypes.pointer(pBytesReturned),  # A pointer to a variable that receives the size, in bytes, of the data
+		# returned in ppBuffer.
+	)
+	try:
+		if not res:
+			raise RuntimeError(f"Failure calling WTSQuerySessionInformationW: {res}")
+		elif ctypes.sizeof(WTSINFOEXW) != pBytesReturned.value:
+			raise RuntimeError(
+				f"Returned data size failure, got {pBytesReturned.value}, expected {ctypes.sizeof(WTSINFOEXW)}"
+			)
+		info = ctypes.cast(
+			ppBuffer,
+			ctypes.POINTER(WTSINFOEXW)
+		)
+		if (
+			not info.contents
+			or info.contents.Level != 1
+			##
+			# Level value must be 1, see:
+			# https://learn.microsoft.com/en-us/windows/win32/api/wtsapi32/ns-wtsapi32-wtsinfoexa
+		):
+			raise RuntimeError(
+				f"Unexpected Level data, got {info.contents.Level}."
+			)
+		return info
+	except Exception as e:
+		log.exception("Unexpected WTSQuerySessionInformation value:", exc_info=e)
+		WTSFreeMemory(
+			ctypes.cast(ppBuffer, ctypes.c_void_p),
+		)
+
+
+def _getSessionLockedValue() -> WTS_LockState:
+	"""Get the WTS_LockState for the current server/session or raises a RuntimeError
+	"""
+	with WTSCurrentSessionInfoEx() as info:
+		infoEx: WTSINFOEX_LEVEL1_W = info.contents.Data.WTSInfoExLevel1
+		sessionFlags: ctypes.wintypes.LONG = infoEx.SessionFlags
+		lockState = WTS_LockState(sessionFlags)
+		log.debug(f"Query Lock state result: {lockState!r}")
+		return lockState