# -*- coding: utf-8 -*- # Copyright (C) 2010-2023 by Mike Gabriel # # Python X2Go is free software; you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation; either version 3 of the License, or # (at your option) any later version. # # Python X2Go is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program; if not, write to the # Free Software Foundation, Inc., # 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA. """\ Python X2Go helper functions, constants etc. """ __NAME__ = 'x2goutils-pylib' __package__ = 'x2go' __name__ = 'x2go.utils' import sys import os import locale import re import types import copy import socket import gevent import string import subprocess import distutils.version import paramiko # Python X2Go modules from .defaults import X2GOCLIENT_OS as _X2GOCLIENT_OS from .defaults import X2GO_SESSIONPROFILE_DEFAULTS as _X2GO_SESSIONPROFILE_DEFAULTS from .defaults import X2GO_MIMEBOX_ACTIONS as _X2GO_MIMEBOX_ACTIONS from .defaults import pack_methods_nx3 from .defaults import BACKENDS as _BACKENDS from . import x2go_exceptions if _X2GOCLIENT_OS != 'Windows': import Xlib from .defaults import X_DISPLAY as _X_DISPLAY if _X2GOCLIENT_OS == 'Windows': import win32gui import win32print import win32con def is_in_nx3packmethods(method): """\ Test if a given compression method is valid for NX3 Proxy. :param method: name of an NXv3 pack method :type method: ``str`` :returns: ``True`` if ``method`` is in the hard-coded list of NX3 compression methods. :rtype: ``bool`` """ return method in pack_methods_nx3 def find_session_line_in_x2golistsessions(session_name, stdout): """\ Return the X2Go session meta information as returned by the ``x2golistsessions`` server command for session ``session_name``. :param session_name: name of a session :type session_name: ``str`` :param stdout: raw output from the ,,x2golistsessions'' command, as list of strings :type stdout: ``list`` :returns: the output line that contains ```` :rtype: ``str`` or ``None`` """ sessions = stdout.read().split("\n") for line in sessions: # skip empty lines if not line: continue if session_name == line.split("|")[1]: return line return None def slugify(value): """\ Normalizes string, converts to lowercase, removes non-alpha characters, converts spaces to hyphens and replaces round brackets by pointed brackets. :param value: a string that shall be sluggified :type value: ``str`` :returns: the sluggified string :rtype: ``str`` """ import unicodedata value = unicodedata.normalize('NFKD', str(value)).encode('ascii', 'ignore') value = re.sub('[^\w\s-]', '', value).strip().lower() value = re.sub('[(]', '<', value).strip().lower() value = re.sub('[)]', '>', value).strip().lower() return value def _genSessionProfileId(): """\ Generate a session profile ID as used in x2goclient's sessions config file. :returns: profile ID :rtype: ``str`` """ import datetime return datetime.datetime.utcnow().strftime('%Y%m%d%H%m%S%f') def _checkIniFileDefaults(data_structure): """\ Check an ini file data structure passed on by a user app or class. :param data_structure: an ini file date structure :type data_structure: ``dict`` of ``dict``s :returns: ``True`` if ``data_structure`` matches that of an ini file data structure :rtype: ``bool`` """ if data_structure is None: return False if type(data_structure) is not dict: return False for sub_dict in list(data_structure.values()): if type(sub_dict) is not dict: return False return True def _checkSessionProfileDefaults(data_structure): """\ Check the data structure of a default session profile passed by a user app. :param data_structure: an ini file date structure :type data_structure: ``dict`` of ``dict``s :returns: ``True`` if ``data_structure`` matches that of an ini file data structure :rtype: ``bool`` """ if data_structure is None: return False if type(data_structure) is not dict: return False return True def _convert_SessionProfileOptions_2_SessionParams(options): """\ Convert session profile options as used in x2goclient's sessions file to Python X2Go session parameters. :param options: a dictionary of options, parameter names as in the X2Go ,,sessions'' file :type options: ``dict`` :returns: session options as used in ``X2GoSession`` instances :rtype: ``dict`` """ _params = copy.deepcopy(options) # get rid of unknown session profile options _known_options = list(_X2GO_SESSIONPROFILE_DEFAULTS.keys()) for p in list(_params.keys()): if p not in _known_options: del _params[p] _rename_dict = { 'host': 'server', 'user': 'username', 'soundsystem': 'snd_system', 'sndport': 'snd_port', 'type': 'kbtype', 'layout': 'kblayout', 'variant': 'kbvariant', 'speed': 'link', 'sshport': 'port', 'useexports': 'allow_share_local_folders', 'restoreexports': 'restore_shared_local_folders', 'usemimebox': 'allow_mimebox', 'mimeboxextensions': 'mimebox_extensions', 'mimeboxaction': 'mimebox_action', 'print': 'printing', 'name': 'profile_name', 'key': 'key_filename', 'command': 'cmd', 'rdpserver': 'rdp_server', 'rdpoptions': 'rdp_options', 'xdmcpserver': 'xdmcp_server', 'useiconv': 'convert_encoding', 'iconvto': 'server_encoding', 'iconvfrom': 'client_encoding', 'usesshproxy': 'use_sshproxy', 'sshproxyhost': 'sshproxy_host', 'sshproxyport': 'sshproxy_port', 'sshproxyuser': 'sshproxy_user', 'sshproxykeyfile': 'sshproxy_key_filename', 'sessiontitle': 'session_title', 'setsessiontitle': 'set_session_title', 'published': 'published_applications', 'autostart': 'auto_start_or_resume', 'autoconnect': 'auto_connect', 'forwardsshagent': 'forward_sshagent', 'autologin': 'look_for_keys', 'sshproxyautologin': 'sshproxy_look_for_keys', 'uniquehostkeyaliases': 'unique_hostkey_aliases', } _speed_dict = { '0': 'modem', '1': 'isdn', '2': 'adsl', '3': 'wan', '4': 'lan', } for opt, val in list(options.items()): # rename options if necessary if opt in list(_rename_dict.keys()): del _params[opt] opt = _rename_dict[opt] if opt in _known_options: _type = type(_known_options[opt]) _params[opt] = _type(val) else: _params[opt] = val # translate integer values for connection speed to readable strings if opt == 'link': val = str(val).lower() if val in list(_speed_dict.keys()): val = _speed_dict[val] val = val.lower() _params['link'] = val # share_local_folders is a list if opt in ('share_local_folders', 'mimebox_extensions'): if type(val) is bytes: if val: _params[opt] = val.split(',') else: _params[opt] = [] if _params['cmd'] == "XFCE4": _params['cmd'] = "XFCE" if _params['look_for_keys']: _params['allow_agent'] = True if _params['sshproxy_look_for_keys']: _params['sshproxy_allow_agent'] = True # append value for quality to value for pack method if _params['quality']: _params['pack'] = '%s-%s' % (_params['pack'], _params['quality']) # delete quality in any case... del _params['quality'] del _params['fstunnel'] if 'export' in _params: _export = _params['export'] del _params['export'] if type(_export) is dict: # since Python X2Go 0.5.0.0 _params['share_local_folders'] = list(_export.keys()) else: # before Python X2Go 0.5.0.0 (analysing the INI file's format for the export field) # fix for wrong export field usage in PyHoca-GUI/CLI and python-x2go before 20110923 _export = _export.replace(",", ";") _export = _export.strip().strip('"').strip().strip(';').strip() _export_list = [ f for f in _export.split(';') if f ] _params['share_local_folders'] = [] for _shared_folder in _export_list: # fix for wrong export field usage in PyHoca-GUI/CLI and python-x2go before 20110923 if not ":" in _shared_folder: _shared_folder = "%s:1" % _shared_folder if _shared_folder.split(":")[-1] == "1": _params['share_local_folders'].append(":".join(_shared_folder.split(":")[:-1])) if options['fullscreen']: _params['geometry'] = 'fullscreen' elif options['maxdim']: _params['geometry'] = 'maximize' else: _params['geometry'] = '%sx%s' % (options['width'], options['height']) del _params['width'] del _params['height'] del _params['fullscreen'] del _params['maxdim'] if not options['sound']: _params['snd_system'] = 'none' del _params['sound'] if not options['rootless']: _params['session_type'] = 'desktop' else: _params['session_type'] = 'application' del _params['rootless'] if _params['mimebox_action'] not in list(_X2GO_MIMEBOX_ACTIONS.keys()): _params['mimebox_action'] = 'OPEN' if not options['usekbd']: _params['kbtype'] = 'null/null' _params['kblayout'] = 'null' _params['kbvariant'] = 'null' del _params['usekbd'] if not _params['kbtype'].strip(): _params['kbtype'] = 'null/null' if not _params['kblayout'].strip(): _params['kblayout'] = 'null' if not _params['kbvariant'].strip(): _params['kbvariant'] = 'null' # This might fire back on us some time... # We need to set 'dpi' in _params as otherwise, it won't be overridable via method parameters # X2GoSessionRegistry.register(). However, setting 'dpi' to a default value is different from # not setting it... Let's see... if not options['setdpi']: _params['dpi'] = _X2GO_SESSIONPROFILE_DEFAULTS['dpi'] del _params['setdpi'] if options['sshproxysameuser']: _params['sshproxy_user'] = _params['username'] del _params['sshproxysameuser'] if options['sshproxysamepass']: _params['sshproxy_reuse_authinfo'] = True _params['sshproxy_key_filename'] = _params['key_filename'] del _params['sshproxysamepass'] if _params['use_sshproxy']: # compat code for Python X2Go 0.2.1.0 -> 0.2.2.0 if 'sshproxytunnel' in options: if not options['sshproxytunnel'].startswith('DEPRECATED'): _params['server'] = options['sshproxytunnel'].split(":")[-2] _params['port'] = options['sshproxytunnel'].split(":")[-1] try: del _params['sshproxytunnel'] except KeyError: pass # currently known but ignored in Python X2Go _ignored_options = [ 'startsoundsystem', 'soundtunnel', 'defsndport', 'icon', 'multidisp', 'display', 'krblogin', 'directrdp', 'directrdpsettings', 'rdpclient', 'rdpport', 'sshproxytype', ] for i in _ignored_options: del _params[i] return _params def session_names_by_timestamp(session_infos): """\ Sorts session profile names by their timestamp (as used in the file format's section name). :param session_infos: a dictionary of session infos as reported by :func:`X2GoClient.list_sessions() ` :type session_infos: ``dict`` :returns: a timestamp-sorted list of session names found in ``session_infos`` :rtype: ``list`` """ session_names = list(session_infos.keys()) sortable_session_names = [ '%s|%s' % (session_name.split('-')[-1].split('_')[0], session_name) for session_name in session_names ] sortable_session_names.sort() return [ session_name.split('|')[1] for session_name in sortable_session_names ] def touch_file(filename, mode='a'): """\ Imitates the behaviour of the GNU/touch command. :param filename: name of the file to touch :type filename: ``str`` :param mode: the file mode (as used for Python file objects) (Default value = 'a') :type mode: ``str`` """ if not os.path.isdir(os.path.dirname(filename)): os.makedirs(os.path.dirname(filename), mode=0o0700) f = open(filename, mode=mode) f.close() def unique(seq): """\ Imitates the behaviour of the GNU/uniq command. :param seq: a list/sequence containing consecutive duplicates. :type seq: ``list`` :returns: list that has been clean up from the consecutive duplicates :rtype: ``list`` """ # order preserving noDupes = [] [noDupes.append(i) for i in seq if not noDupes.count(i)] return noDupes def known_encodings(): """\ Render a list of all-known-to-Python character encodings (including all known aliases) """ from encodings.aliases import aliases _raw_encname_list = [] _raw_encname_list.extend(list(aliases.keys())) _raw_encname_list.extend(list(aliases.values())) _raw_encname_list.sort() _encname_list = [] for _raw_encname in _raw_encname_list: _encname = _raw_encname.upper() _encname = _encname.replace('_', '-') _encname_list.append(_encname) _encname_list.sort() _encname_list = unique(_encname_list) return _encname_list def patiently_remove_file(dirname, filename): """\ Try to remove a file, wait for unlocking, remove it once removing is possible... :param dirname: directory name the file is in :type dirname: ``str`` :param filename: name of the file to be removed :type filename: ``str`` """ _not_removed = True while _not_removed: try: os.remove(os.path.join(dirname, filename)) _not_removed = False except: # file is probably locked gevent.sleep(5) def detect_unused_port(bind_address='127.0.0.1', preferred_port=None): """\ Detect an unused IP socket. :param bind_address: IP address to bind to (Default value = '127.0.0.1') :type bind_address: ``str`` :param preferred_port: IP socket port that shall be tried first for availability (Default value = None) :type preferred_port: ``str`` :returns: free local IP socket port that can be used for binding :rtype: ``str`` """ sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) try: if preferred_port: sock.bind((bind_address, preferred_port)) ipaddr, port = sock.getsockname() else: raise except: sock.bind(('', 0)) ipaddr, port = sock.getsockname() return port def get_encoding(): """\ Detect systems default character encoding. :returns: The system's local character encoding. :rtype: ``str`` """ try: encoding = locale.getdefaultlocale()[1] if encoding is None: raise BaseException except: try: encoding = sys.getdefaultencoding() except: encoding = 'ascii' return encoding def is_abs_path(path): """\ Test if a given path is an absolute path name. :param path: test this path for absolutism... :type path: ``str`` :returns: Returns ``True`` if path is an absolute path name :rtype: ``bool`` """ return bool((path.startswith('/') or re.match('^[%s]\:\\\\' % string.ascii_letters, path))) def xkb_rules_names(): """\ Wrapper for: xprop -root _XKB_RULES_NAMES :returns: A Python dictionary that contains the current X11 keyboard rules. :rtype: ``dict`` """ p = subprocess.Popen(['xprop', '-root', '_XKB_RULES_NAMES',], stdout=subprocess.PIPE, ) _rn_list = p.stdout.read().split('"') _rn_dict = { 'rules': _rn_list[1], 'model': _rn_list[3], 'layout': _rn_list[5], 'variant': _rn_list[7], 'options': _rn_list[9], } return _rn_dict def local_color_depth(): """\ Detect the current local screen's color depth. :returns: the local color depth in bits :rtype: ``int`` """ if _X2GOCLIENT_OS != 'Windows': try: p = subprocess.Popen(['xwininfo', '-root',], stdout=subprocess.PIPE, ) _depth_line = [ _info.strip() for _info in p.stdout.read().decode().split('\n') if 'Depth:' in _info ][0] _depth = _depth_line.split(' ')[1] return int(_depth) except IndexError: # a sensible default value return 24 except OSError: # for building your package... return 24 else: # This gets the color depth of the primary monitor. All monitors need not have the same color depth. dc = win32gui.GetDC(None) _depth = win32print.GetDeviceCaps(dc, win32con.BITSPIXEL) * win32print.GetDeviceCaps(dc, win32con.PLANES) win32gui.ReleaseDC(None, dc) return _depth def is_color_depth_ok(depth_session, depth_local): """\ Test if color depth of this session is compatible with the local screen's color depth. :param depth_session: color depth of the session :type depth_session: ``int`` :param depth_local: color depth of local screen :type depth_local: ``int`` :returns: Does the session color depth work with the local display? :rtype: ``bool`` """ if depth_session == 0: return True if depth_session == depth_local: return True if ( ( depth_session == 24 or depth_session == 32 ) and ( depth_local == 24 or depth_local == 32 ) ): return True; if ( ( depth_session == 16 or depth_session == 17 ) and ( depth_local == 16 or depth_local == 17 ) ): return True; return False def find_session_window(session_name): """\ Find a session window by its X2GO session ID. :param session_name: session name/ID of an X2Go session window :type session_name: ``str`` :returns: the window object (or ID) of the searched for session window :rtype: ``obj`` on Unix, ``int`` on Windows """ if _X2GOCLIENT_OS != 'Windows': # establish connection to the win API in use... display = _X_DISPLAY if display: root = display.screen().root success = False windowIDs_obj = root.get_full_property(display.intern_atom('_NET_CLIENT_LIST'), Xlib.X.AnyPropertyType) if windowIDs_obj is None: # works with i3 session manager... windowIDs_obj = root.get_full_property(display.intern_atom('_NET_CLIENT_LIST_STACKING'), Xlib.X.AnyPropertyType) if windowIDs_obj is not None: windowIDs = windowIDs_obj.value for windowID in windowIDs: window = display.create_resource_object('window', windowID) try: name = window.get_wm_name() except Xlib.error.BadWindow: continue if name is not None and "X2GO-{session_name}".format(session_name=session_name) == name: success = True break if success: return window else: windows = [] window = None def _callback(hwnd, extra): if win32gui.GetWindowText(hwnd) == "X2GO-%s" % session_name: windows.append(hwnd) win32gui.EnumWindows(_callback, None) if len(windows): window = windows[0] return window def get_desktop_geometry(): """\ Get the geometry of the current screen's desktop. :returns: a (, ) tuple will be returned :rtype: ``tuple`` """ if _X2GOCLIENT_OS != 'Windows': display = _X_DISPLAY if display: root = display.screen().root return (root.get_geometry().width, root.get_geometry().height) return None def get_workarea_geometry(): """\ Get the geometry of the current screen's work area by wrapping around:: xprop -root '_NET_WORKAREA' :returns: a (, ) tuple will be returned :rtype: ``tuple`` """ if _X2GOCLIENT_OS != 'Windows': p = subprocess.Popen(['xprop', '-root', '_NET_WORKAREA',], stdout=subprocess.PIPE, ) if sys.version_info[0] >= 3: _list = p.stdout.read().decode().rstrip('\n').split(',') else: _list = p.stdout.read().rstrip('\n').split(',') if len(_list) == 4: return (_list[2].strip(), _list[3].strip()) else: return None else: return None def set_session_window_title(session_window, session_title): """\ Set title of session window. :param session_window: session window instance :type session_window: ``obj`` :param session_title: session title to be set for ``session_window`` :type session_title: ``str`` """ if _X2GOCLIENT_OS != 'Windows': try: session_window.set_wm_name(str(session_title)) session_window.set_wm_icon_name(str(session_title)) _X_DISPLAY.sync() except Xlib.error.BadWindow: pass else: win32gui.SetWindowText(session_window, session_title) def raise_session_window(session_window): """\ Raise session window. Not functional for Unix-like operating systems. :param session_window: session window instance :type session_window: ``obj`` """ if _X2GOCLIENT_OS != 'Windows': pass else: if session_window is not None: win32gui.SetForegroundWindow(session_window) def merge_ordered_lists(l1, l2): """\ Merge sort two sorted lists :param l1: first sorted list :type l1: ``list`` :param l2: second sorted list :type l2: ``list`` :returns: the merge result of both sorted lists :rtype: ``list`` """ ordered_list = [] # Copy both the args to make sure the original lists are not # modified l1 = l1[:] l2 = l2[:] while (l1 and l2): if l1[0] not in l2: item = l1.pop(0) elif l2[0] not in l1: item = l2.pop(0) elif l1[0] in l2: item = l1.pop(0) l2.remove(item) if item not in ordered_list: ordered_list.append(item) # Add the remaining of the lists ordered_list.extend(l1 if l1 else l2) return ordered_list def compare_versions(version_a, op, version_b): """\ Compare with using operator . In the background ``distutils.version.LooseVersion`` is used for the comparison operation. :param version_a: a version string :type version_a: ``str`` :param op: an operator provide as string (e.g. '<', '>', '==', '>=' etc.) :type op: ``str`` :param version_b: another version string that is to be compared with :type version_b: ``str`` """ ### FIXME: this comparison is not reliable with beta et al. version strings ver_a = distutils.version.LooseVersion(version_a) ver_b = distutils.version.LooseVersion(version_b) return eval("ver_a %s ver_b" % op) class ProgressStatus(object): """\ A simple progress status iterator class. """ def __init__(self, progress_event, progress_func=list(range(0, 100, 10))): """\ :param progress_event: a threading.Event() object that gets notified on progress :type progress_event: ``obj`` :param progress_func: a function that delivers a value between 0 and 100 (progress percentage value) :type progress_func: ``func`` """ self.ev = progress_event self.progress_func = progress_func def __iter__(self): """\ Intialize the :class:`x2go.utils.ProgressStatus` iterator object. """ self.status = self.progress_func() return self def __next__(self): """\ On each iteration wait for the progress event to get triggered from an outside part of the application. Once the event fires read the progress status from the progress retrieval function and clear the event afterwards (so we wait for the next firing of the event). """ if self.status < 100 and self.status != -1: self.ev.wait() self.status = self.progress_func() self.ev.clear() return self.status else: raise StopIteration def _get_backend_class(backend, class_name): # silence pyflakes, the _this_class var will be assigned in an exec() statement below... _this_class = None if type(backend) not in (bytes, str): return backend if backend != "default": backend = backend.upper( ) available_backends = [ k for k in list(_BACKENDS[class_name].keys()) if k != 'default' ] # if for backend is given 'default' use the default backend module if backend == 'default': backend = _BACKENDS[class_name]['default'] if backend in available_backends: namespace = {} exec("from {backend} import {class_name} as _this_class".format(backend=_BACKENDS[class_name][backend], class_name=class_name), namespace) _this_class = namespace['_this_class'] else: raise x2go_exceptions.X2GoBackendException('unknown backend name %s for class %s' % (backend, class_name)) return _this_class def genkeypair(local_username, client_address, key_type='RSA'): """\ Generate an SSH pub/priv key pair without writing the private key to file. :param local_username: the key is for this user :type local_username: ``unicode`` :param client_address: the key is only valid for this client :type client_address: ``unicode`` :param key_type: either of: RSA, DSA (Default value = 'RSA') :type key_type: ``unicode`` """ key = None pubkey = None # generate key pair if str(key_type) == 'RSA': key = paramiko.RSAKey.generate(2048) elif str(key_type) == 'DSA': key = paramiko.DSSKey.generate(1024) if key: # assemble the public key if key_type == "RSA": pubkey_type = 'ssh-rsa' elif key_type == "DSA": pubkey_type = 'ssh-dss' # FIXME: the from option does not work properly by some reason. Fix it later #pubkey = "from={client_address},no-X11-forwarding,no-pty,no-user-rc {pubkey_type} {pubkey} {local_username}@{client_address}".format(pubkey=key.get_base64(), pubkey_type=pubkey_type, local_username=local_username, client_address=client_address) pubkey = "no-X11-forwarding,no-pty,no-user-rc {pubkey_type} {pubkey} {local_username}@{client_address}".format(pubkey=key.get_base64(), pubkey_type=pubkey_type, local_username=local_username, client_address=client_address) return (pubkey, key) def which(basename): """\ Python equivalent to the shell command "which". :param basename: the basename of an application to be search for in $PATH :type basename: ``str`` :returns: full path to the application :rtype: ``str`` """ if _X2GOCLIENT_OS == 'Windows': delim = ";" else: delim = ":" for path in os.environ["PATH"].split(delim): if os.path.exists(os.path.join(path, basename)): return os.path.join(path, basename) return None