diff --git a/xpra/client/gtk3/u2f_tool.py b/xpra/client/gtk3/u2f_tool.py index 290d55cef3..ec2f2afe14 100644 --- a/xpra/client/gtk3/u2f_tool.py +++ b/xpra/client/gtk3/u2f_tool.py @@ -75,8 +75,9 @@ def printmsgs(*msgs): "%s" % (str(e) or type(e))) return 1 + from pyu2f.model import RegisteredKey info("Please activate your U2F device now to generate a new key") - registered_keys = [] + registered_keys : List[RegisteredKey] = [] challenge= b'01234567890123456789012345678901' #unused rr = dev.Register(APP_ID, challenge, registered_keys) b = rr.registration_data diff --git a/xpra/platform/posix/fd_portal_shadow.py b/xpra/platform/posix/fd_portal_shadow.py index 96f8843277..93cb7701a7 100755 --- a/xpra/platform/posix/fd_portal_shadow.py +++ b/xpra/platform/posix/fd_portal_shadow.py @@ -39,6 +39,11 @@ class PipewireWindowModel(RootWindowModel): __slots__ = ("pipewire_id", "pipewire_props") + def __init__(self, root_window, capture, title:str, geometry, node_id:int, props:typedict): + super().__init__(root_window=root_window, capture=capture, title=title, geometry=geometry) + self.pipewire_id = node_id + self.pipewire_props = props + class PortalShadow(GTKShadowServerBase): def __init__(self, multi_window=True): @@ -258,7 +263,7 @@ def create_capture_pipeline(self, fd : int, node_id : int, w : int, h : int) -> log.info(f"cannot use {encoding}: {e}") return Capture(el, pixel_format="BGRX", width=w, height=h) - def start_pipewire_capture(self, node_id, props) -> None: + def start_pipewire_capture(self, node_id:int, props:typedict) -> None: log(f"start_pipewire_capture({node_id}, {props})") if not isinstance(node_id, int): raise ValueError(f"node-id is a {type(node_id)}, must be an int") @@ -282,9 +287,7 @@ def start_pipewire_capture(self, node_id, props) -> None: source_type = props.intget("source_type") title = f"{AvailableSourceTypes(source_type)} {node_id}" geometry = (x, y, w, h) - model = PipewireWindowModel(self.root, self.capture, title, geometry) - model.pipewire_id = node_id - model.pipewire_props = props + model = PipewireWindowModel(self.root, self.capture, title, geometry, node_id, props) #must be called from the main thread: log(f"new model: {model}") self.do_add_new_window_common(node_id, model) diff --git a/xpra/platform/posix/gl_context.py b/xpra/platform/posix/gl_context.py index 386f36eda4..e1e66e639a 100644 --- a/xpra/platform/posix/gl_context.py +++ b/xpra/platform/posix/gl_context.py @@ -53,7 +53,8 @@ def c_attrs(props): def get_xdisplay() -> int: ptr = get_display_ptr() - assert ptr, "no X11 display registered" + if not ptr: + raise RuntimeError("no X11 display registered") # pylint: disable=import-outside-toplevel from OpenGL.raw.GLX._types import struct__XDisplay return cast(ptr, POINTER(struct__XDisplay)) diff --git a/xpra/platform/win32/gl_context.py b/xpra/platform/win32/gl_context.py index 9639010e80..a5eeac847e 100644 --- a/xpra/platform/win32/gl_context.py +++ b/xpra/platform/win32/gl_context.py @@ -22,6 +22,7 @@ ) from xpra.platform.win32.glwin32 import ( wglCreateContext, wglMakeCurrent, wglDeleteContext, + HGLRC, PIXELFORMATDESCRIPTOR, PFD_TYPE_RGBA, PFD_DRAW_TO_WINDOW, PFD_SUPPORT_OPENGL, PFD_DOUBLEBUFFER, PFD_DEPTH_DONTCARE, PFD_SUPPORT_COMPOSITION, PFD_MAIN_PLANE, PAINTSTRUCT, @@ -38,11 +39,11 @@ def DefWndProc(hwnd, msg, wParam, lParam): class WGLWindowContext: - def __init__(self, hwnd, hdc, context): + def __init__(self, hwnd:int, hdc:int, context): self.hwnd = hwnd self.hdc = hdc self.context = context - self.ps = None + self.ps = PAINTSTRUCT() self.paint_hdc = None def __enter__(self): @@ -129,10 +130,10 @@ def check_support(self, force_enable=False): def get_bit_depth(self): return 0 - def is_double_buffered(self): + def is_double_buffered(self) -> bool: return DOUBLE_BUFFERED #self.pixel_format_props.get("double-buffered", False) - def get_paint_context(self, gdk_window): + def get_paint_context(self, gdk_window) -> WGLWindowContext: hwnd = get_window_handle(gdk_window) if self.hwnd!=hwnd: #(this shouldn't happen) @@ -142,7 +143,7 @@ def get_paint_context(self, gdk_window): self.context = self.create_wgl_context(hwnd) return WGLWindowContext(hwnd, self.hdc, self.context) - def create_wgl_context(self, hwnd): + def create_wgl_context(self, hwnd:int) -> HGLRC: bpc = 8 self.hwnd = hwnd self.pixel_format_props = {} diff --git a/xpra/platform/win32/gui.py b/xpra/platform/win32/gui.py index 02e0cd70c9..4a091fe024 100644 --- a/xpra/platform/win32/gui.py +++ b/xpra/platform/win32/gui.py @@ -9,7 +9,7 @@ import os import sys import types -from typing import Optional, Callable, List, Tuple, Dict, Any +from typing import Optional, Callable, List, Tuple, Dict, Any, Type from ctypes import ( WinDLL, WinError, get_last_error, # @UnresolvedImport CDLL, pythonapi, py_object, @@ -119,7 +119,7 @@ FORWARD_WINDOWS_KEY, WHEEL, WHEEL_DELTA) -def do_init(): +def do_init() -> None: from xpra.platform.win32.dpi import init_dpi init_dpi() if APP_ID: @@ -131,7 +131,7 @@ def do_init(): log.error("Error: failed to setup dwm window color matching", exc_info=True) -def init_appid(): +def init_appid() -> None: SetCurrentProcessExplicitAppUserModelID = shell32.SetCurrentProcessExplicitAppUserModelID SetCurrentProcessExplicitAppUserModelID.restype = HRESULT SetCurrentProcessExplicitAppUserModelID.argtypes = [LPCWSTR] @@ -139,7 +139,7 @@ def init_appid(): log.warn("Warning: failed to set process app ID") -def use_stdin(): +def use_stdin() -> bool: if os.environ.get("MSYSCON") or os.environ.get("CYGWIN"): return False stdin = sys.stdin @@ -157,12 +157,12 @@ def use_stdin(): return True -def get_clipboard_native_class(): +def get_clipboard_native_class() -> str: #"xpra.clipboard.translated_clipboard.TranslatedClipboardProtocolHelper" return "xpra.platform.win32.clipboard.Win32Clipboard" -def get_native_notifier_classes(): +def get_native_notifier_classes() -> List[Type]: try: from xpra.platform.win32.win32_notifier import Win32_Notifier return [Win32_Notifier] @@ -205,7 +205,7 @@ def gl_check() -> str: return "" -def get_monitor_workarea_for_window(handle): +def get_monitor_workarea_for_window(handle:int): try: monitor = MonitorFromWindow(handle, win32con.MONITOR_DEFAULTTONEAREST) mi = GetMonitorInfo(monitor) @@ -227,7 +227,7 @@ def get_monitor_workarea_for_window(handle): return None -def get_window_handle(window): +def get_window_handle(window) -> int: """ returns the win32 hwnd from a gtk.Window or gdk.Window """ gdk_window = window try: @@ -241,7 +241,7 @@ def get_window_handle(window): #log("get_window_handle(%s) gpointer=%#x, hwnd=%#x", gpointer, hwnd) return hwnd -def get_desktop_name(): +def get_desktop_name() -> str: try: desktop = OpenInputDesktop(0, True, win32con.MAXIMUM_ALLOWED) if desktop: @@ -254,10 +254,10 @@ def get_desktop_name(): except Exception as e: log.warn("Warning: failed to get desktop name") log.warn(" %s", e) - return None + return "" -def get_session_type(): +def get_session_type() -> str: try: b = c_bool() retcode = dwmapi.DwmIsCompositionEnabled(byref(b)) @@ -309,7 +309,7 @@ def win32_propsys_set_group_leader(self, leader): log.error("Error: failed to set group leader") log.estr(e) -WS_NAMES = { +WS_NAMES : Dict[int,str] = { win32con.WS_BORDER : "BORDER", win32con.WS_CAPTION : "CAPTION", win32con.WS_CHILD : "CHILD", @@ -336,10 +336,10 @@ def win32_propsys_set_group_leader(self, leader): win32con.WS_VSCROLL : "VSCROLL", } -def style_str(style): +def style_str(style) -> str: return csv(s for c,s in WS_NAMES.items() if (c & style)==c) -def pointer_grab(window, *args): +def pointer_grab(window, *args) -> bool: hwnd = get_window_handle(window) grablog("pointer_grab%s window=%s, hwnd=%s", args, window, hwnd) if not hwnd: @@ -369,7 +369,7 @@ def pointer_grab(window, *args): window._client.pointer_grabbed = window.wid return True -def pointer_ungrab(window, *args): +def pointer_ungrab(window, *args) -> bool: hwnd = get_window_handle(window) client = window._client grablog("pointer_ungrab%s window=%s, hwnd=%s, pointer_grabbed=%s", @@ -381,7 +381,7 @@ def pointer_ungrab(window, *args): client.pointer_grabbed = None return True -def fixup_window_style(self, *_args): +def fixup_window_style(self, *_args) -> None: """ a fixup function we want to call from other places """ hwnd = get_window_handle(self) if not hwnd: @@ -431,7 +431,7 @@ def fixup_window_style(self, *_args): except Exception: log.warn("failed to fixup window style", exc_info=True) -def set_decorated(self, decorated): +def set_decorated(self, decorated:bool): """ override method which ensures that we call fixup_window_style whenever decorations are toggled """ self.__set_decorated(decorated) #call the original saved method @@ -442,7 +442,7 @@ def window_state_updated(window): log("window_state_updated(%s)", window) fixup_window_style(window) -def apply_maxsize_hints(window, hints): +def apply_maxsize_hints(window, hints:Dict[str,Any]): """ extracts the max-size hints from the hints, and passes it to the win32hooks class which can implement it (as GTK2 does not honour it properly on win32) @@ -489,7 +489,7 @@ def apply_maxsize_hints(window, hints): hints.pop(x, None) window_state_updated(window) -def apply_geometry_hints(self, hints): +def apply_geometry_hints(self, hints:Dict): log("apply_geometry_hints(%s)", hints) apply_maxsize_hints(self, hints) return self.__apply_geometry_hints(hints) #call the original saved method @@ -508,7 +508,7 @@ def cache_pointer_offset(self, event): def no_set_group(*_args): """ provide a dummy implementation """ -def add_window_hooks(window): +def add_window_hooks(window) -> None: log("add_window_hooks(%s) WINDOW_HOOKS=%s, GROUP_LEADER=%s, UNDECORATED_STYLE=%s", window, WINDOW_HOOKS, GROUP_LEADER, UNDECORATED_STYLE) log(" MAX_SIZE_HINT=%s, MAX_SIZE_HINT=%s", MAX_SIZE_HINT, MAX_SIZE_HINT) @@ -613,7 +613,7 @@ def mousehwheel(_hwnd, _event, wParam, lParam): win32hooks.add_window_event_handler(win32con.WM_MOUSEHWHEEL, mousehwheel) -def remove_window_hooks(window): +def remove_window_hooks(window) -> None: try: win32hooks = getattr(window, "win32hooks", None) if win32hooks: @@ -624,14 +624,14 @@ def remove_window_hooks(window): log.error("remove_window_hooks(%s)", exc_info=True) -def get_xdpi(): +def get_xdpi() -> int: try: return _get_device_caps(win32con.LOGPIXELSX) except Exception as e: log.warn("failed to get xdpi: %s", e) return -1 -def get_ydpi(): +def get_ydpi() -> int: try: return _get_device_caps(win32con.LOGPIXELSY) except Exception as e: @@ -661,7 +661,7 @@ def get_ydpi(): } -def _add_SPI(info:Dict[str,Any], constant:int, name:str, convert:Callable, default:Any=None): +def _add_SPI(info:Dict[str,Any], constant:int, name:str, convert:Callable, default:Any=None) -> None: v = GetIntSystemParametersInfo(constant) if v is not None: info[name] = convert(v) @@ -986,7 +986,7 @@ def getTaskbar(): return taskbar -def set_window_progress(window, pct): +def set_window_progress(window, pct:int): taskbar = getattr(window, "taskbar", None) if not taskbar: taskbar = getTaskbar() @@ -1006,7 +1006,7 @@ def set_window_progress(window, pct): WTS_SESSION_LOCK = 0x7 WTS_SESSION_UNLOCK = 0x8 WTS_SESSION_REMOTE_CONTROL = 0x9 -WTS_SESSION_EVENTS = { +WTS_SESSION_EVENTS : Dict[int, str] = { WTS_CONSOLE_CONNECT : "CONSOLE CONNECT", WTS_CONSOLE_DISCONNECT : "CONSOLE_DISCONNECT", WTS_REMOTE_CONNECT : "REMOTE_CONNECT", @@ -1062,10 +1062,10 @@ def log_screensaver(): from xpra.make_thread import start_thread start_thread(self.init_keyboard_listener, "keyboard-listener", daemon=True) - def ready(self): + def ready(self) -> None: """ nothing specific to do here """ - def cleanup(self): + def cleanup(self) -> None: log("ClientExtras.cleanup()") self._exit = True cha = self._console_handler_added @@ -1092,7 +1092,7 @@ def cleanup(self): log("ClientExtras.cleanup() ended") #self.client = None - def get_keyboard_layout_id(self): + def get_keyboard_layout_id(self) -> int: name_buf = create_string_buffer(win32con.KL_NAMELENGTH) if not GetKeyboardLayoutName(name_buf): return 0 @@ -1104,14 +1104,14 @@ def get_keyboard_layout_id(self): log.warn("Warning: failed to parse keyboard layout code '%s'", name_buf.value) return 0 - def poll_layout(self): + def poll_layout(self) -> None: self.keyboard_poll_timer = 0 klid = self.get_keyboard_layout_id() if klid and klid!=self.keyboard_id: self.keyboard_id = klid self.client.window_keyboard_layout_changed() - def init_keyboard_listener(self): + def init_keyboard_listener(self) -> None: class KBDLLHOOKSTRUCT(Structure): _fields_ = [("vk_code", DWORD), ("scan_code", DWORD), @@ -1126,7 +1126,7 @@ def toString(self): win32con.WM_KEYUP : "KEYUP", win32con.WM_SYSKEYUP : "SYSKEYUP", } - def low_level_keyboard_handler(nCode, wParam, lParam): + def low_level_keyboard_handler(nCode:int, wParam:int, lParam:int): log("WH_KEYBOARD_LL: %s", (nCode, wParam, lParam)) kh = getattr(self.client, "keyboard_helper", None) locked = getattr(kh, "locked", False) @@ -1223,7 +1223,7 @@ def low_level_keyboard_handler(nCode, wParam, lParam): keylog("DispatchMessageA(%#x)=%s", lpmsg, r) - def wm_move(self, wParam, lParam): + def wm_move(self, wParam:int, lParam:int) -> None: c = self.client log("WM_MOVE: %s/%s client=%s", wParam, lParam, c) if c: @@ -1231,7 +1231,7 @@ def wm_move(self, wParam, lParam): #but we do want to process it as such (see window reinit code) c.screen_size_changed() - def end_session(self, wParam, lParam): + def end_session(self, wParam:int, lParam:int): log("WM_ENDSESSION") c = self.client if not c: @@ -1249,8 +1249,8 @@ def end_session(self, wParam, lParam): return c.disconnect_and_quit(ExitCode.OK, reason) - def session_change_event(self, event, session): - event_name = WTS_SESSION_EVENTS.get(event, event) + def session_change_event(self, event:int, session:int): + event_name = WTS_SESSION_EVENTS.get(event) or str(event) log("WM_WTSSESSION_CHANGE: %s on session %#x", event_name, session) c = self.client if not c: @@ -1265,11 +1265,11 @@ def session_change_event(self, event, session): GLib.idle_add(c.unfreeze) - def inputlangchange(self, wParam, lParam): + def inputlangchange(self, wParam:int, lParam:int): keylog("WM_INPUTLANGCHANGE: %i, %i", wParam, lParam) self.poll_layout() - def inichange(self, wParam, lParam): + def inichange(self, wParam:int, lParam:int): if lParam: from ctypes import c_char_p log("WM_WININICHANGE: %#x=%s", lParam, c_char_p(lParam).value) @@ -1277,7 +1277,7 @@ def inichange(self, wParam, lParam): log("WM_WININICHANGE: %i, %i", wParam, lParam) - def activateapp(self, wParam, lParam): + def activateapp(self, wParam:int, lParam:int) -> None: c = self.client log("WM_ACTIVATEAPP: %s/%s client=%s", wParam, lParam, c) if not c: @@ -1294,7 +1294,7 @@ def activateapp(self, wParam, lParam): fixup_window_style() - def power_broadcast_event(self, wParam, lParam): + def power_broadcast_event(self, wParam:int, lParam:int): c = self.client log("WM_POWERBROADCAST: %s/%s client=%s", POWER_EVENTS.get(wParam, wParam), lParam, c) #maybe also "PBT_APMQUERYSUSPEND" and "PBT_APMQUERYSTANDBY"? @@ -1306,7 +1306,7 @@ def power_broadcast_event(self, wParam, lParam): c.resume() - def handle_console_event(self, event): + def handle_console_event(self, event:int) -> int: c = self.client event_name = KNOWN_EVENTS.get(event, event) log("handle_console_event(%s) client=%s, event_name=%s", event, c, event_name) diff --git a/xpra/server/auth/exec_auth.py b/xpra/server/auth/exec_auth.py index 18efeb8589..60ad1fbf2f 100644 --- a/xpra/server/auth/exec_auth.py +++ b/xpra/server/auth/exec_auth.py @@ -44,7 +44,7 @@ def __init__(self, **kwargs): self.proc = None self.timeout_event = False if not self.command: - self.command = get_default_auth_dialog() + self.command = [get_default_auth_dialog(), ] connection = kwargs.get("connection") if not connection: raise ValueError("connection object is missing") diff --git a/xpra/server/auth/pam_auth.py b/xpra/server/auth/pam_auth.py index eec934ef33..fad0da710b 100644 --- a/xpra/server/auth/pam_auth.py +++ b/xpra/server/auth/pam_auth.py @@ -55,7 +55,7 @@ def check_password(self, password:str) -> bool: def get_challenge(self, digests) -> Tuple[bytes,str]: self.req_xor(digests) - return super().get_challenge(["xor"]) + return super().do_get_challenge(["xor"]) def __repr__(self): return "PAM" diff --git a/xpra/server/auth/sqlauthbase.py b/xpra/server/auth/sqlauthbase.py index b8afab84f8..b158ab0aa7 100755 --- a/xpra/server/auth/sqlauthbase.py +++ b/xpra/server/auth/sqlauthbase.py @@ -41,12 +41,12 @@ def get_sessions(self) -> Optional[SessionData]: return self.parse_session_data(data) def parse_session_data(self, data) -> Optional[SessionData]: + displays = [] + env_options = {} + session_options = {} try: uid = int(data[0] or "0") gid = int(data[1] or "0") - displays = [] - env_options = {} - session_options = {} if len(data)>2: displays = [x.strip() for x in str(data[2]).split(",")] if len(data)>3: @@ -57,7 +57,8 @@ def parse_session_data(self, data) -> Optional[SessionData]: log("parse_session_data() error on row %s", data, exc_info=True) log.error("Error: sqlauth database row parsing problem:") log.estr(e) - return None + uid = self.get_uid() + gid = self.get_gid() return uid, gid, displays, env_options, session_options @@ -90,7 +91,7 @@ def add_user(self, username:str, password:str, uid:int=getuid(), gid:int=getgid( def remove_user(self, username:str, password:str="") -> None: sql = "DELETE FROM users WHERE username=%s" % self.param - sqlargs = (username, ) + sqlargs : Tuple[str,...] = (username, ) if password: sql += " AND password=%s" % self.param sqlargs = (username, password) diff --git a/xpra/server/auth/sys_auth_base.py b/xpra/server/auth/sys_auth_base.py index 943662221b..67c5eac341 100644 --- a/xpra/server/auth/sys_auth_base.py +++ b/xpra/server/auth/sys_auth_base.py @@ -105,6 +105,12 @@ def req_challenge(self, digests, required="xor") -> None: raise RuntimeError(f"{self!r} authenticator requires the {required!r} digest") def get_challenge(self, digests) -> Optional[Tuple[bytes,str]]: + if self.salt is not None: + log.error("Error: authentication challenge already sent!") + return None + return self.do_get_challenge(digests) + + def do_get_challenge(self, digests) -> Tuple[bytes,str]: if self.salt is not None: log.error("Error: authentication challenge already sent!") return None diff --git a/xpra/server/auth/win32_auth.py b/xpra/server/auth/win32_auth.py index 5e736e251f..03e72da3e4 100755 --- a/xpra/server/auth/win32_auth.py +++ b/xpra/server/auth/win32_auth.py @@ -44,9 +44,9 @@ def get_password(self) -> str: def get_challenge(self, digests) -> Tuple[bytes,str]: self.req_xor(digests) - return super().get_challenge(["xor"]) + return super().do_get_challenge(["xor"]) - def check(self, password:str) -> bool: + def check(self, password:bytes) -> bool: token = HANDLE() domain = '' #os.environ.get('COMPUTERNAME') if LOG_CREDENTIALS: diff --git a/xpra/x11/gtk_x11/clipboard.py b/xpra/x11/gtk_x11/clipboard.py index b7b9d2175b..d9afc8cdd6 100644 --- a/xpra/x11/gtk_x11/clipboard.py +++ b/xpra/x11/gtk_x11/clipboard.py @@ -361,7 +361,7 @@ def set_selection_response(self, requestor:int, target:str, prop:str, dtype:str, else: #maybe even delete the property? #X11Window.XDeleteProperty(xid, prop) - prop = None + prop = "" X11Window.sendSelectionNotify(requestor, self._selection, target, prop, time) except XError as e: log("failed to set selection", exc_info=True) @@ -397,7 +397,7 @@ def do_selection_notify_event(self, event) -> None: xid = 0 if event.owner: xid = event.owner - self.owned = xid and xid==self.xid + self.owned = bool(xid) and xid==self.xid log("do_selection_notify_event(%s) owned=%s, was %s (owner=%#x, xid=%#x), enabled=%s, can-send=%s", event, self.owned, owned, xid, self.xid, self._enabled, self._can_send) if not self._enabled: