# -*- coding: utf-8 -*-
# Copyright (C) 2010-2023 by Mike Gabriel <mike.gabriel@das-netzwerkteam.de>
#
# 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.
"""\
:class:`x2go.backends.profiles.base.X2GoSessionProfiles` class - managing X2Go Client session profiles, base API.
:class:`x2go.backends.profiles.base.X2GoSessionProfiles` is a public API class. Use this class in your Python X2Go based
applications.
"""
__NAME__ = 'x2gosessionprofiles-pylib'
__package__ = 'x2go.backends.profiles'
__name__ = 'x2go.backends.profiles.base'
import copy
import types
import re
# Python X2Go modules
from x2go.defaults import X2GO_SESSIONPROFILE_DEFAULTS as _X2GO_SESSIONPROFILE_DEFAULTS
from x2go.defaults import X2GO_DESKTOPSESSIONS as _X2GO_DESKTOPSESSIONS
import x2go.log as log
import x2go.utils as utils
from x2go.x2go_exceptions import X2GoProfileException
[docs]
class X2GoSessionProfiles(object):
defaultSessionProfile = copy.deepcopy(_X2GO_SESSIONPROFILE_DEFAULTS)
_non_profile_sections = ('embedded')
def __init__(self, session_profile_defaults=None, logger=None, loglevel=log.loglevel_DEFAULT, **kwargs):
"""\
Retrieve X2Go session profiles. Base class for the different specific session profile
configuration backends.
:param session_profile_defaults: a default session profile
:type session_profile_defaults: ``dict``
:param logger: you can pass an :class:`x2go.log.X2GoLogger` object to the
:class:`x2go.backends.profiles.httpbroker.X2GoSessionProfiles` constructor
:type logger: :class:`x2go.log.X2GoLogger` instance
:param loglevel: if no :class:`x2go.log.X2GoLogger` object has been supplied a new one will be
constructed with the given loglevel
:type loglevel: ``int``
"""
self.defaultValues = {}
self._profile_metatypes = {}
self._cached_profile_ids = {}
self.__useexports = {}
self._profiles_need_profile_id_renewal = []
self.write_user_config = False
if logger is None:
self.logger = log.X2GoLogger(loglevel=loglevel)
else:
self.logger = copy.deepcopy(logger)
self.logger.tag = __NAME__
if utils._checkSessionProfileDefaults(session_profile_defaults):
self.defaultSessionProfile = session_profile_defaults
self.populate_session_profiles()
def __call__(self, profile_id_or_name):
"""\
Retrieve the session profile configuration for a given session profile ID (or name)
:param profile_id_or_name: profile ID or profile name
:type profile_id_or_name: ``str``
:returns: the profile ID's / name's profile configuration
:rtype: ``dict``
"""
_profile_id = self.check_profile_id_or_name(self, profile_id_or_name)
return self.get_profile_config(profile_id=_profile_id)
[docs]
def init_profile_cache(self, profile_id_or_name):
"""\
Some session profile backends (e.g. the broker backends cache
dynamic session profile data). On new connections, it is
recommented to (re-)initialize these caches.
:param profile_id_or_name: profile ID or profile name
:type profile_id_or_name: ``str``
"""
profile_id = self.check_profile_id_or_name(profile_id_or_name)
# allow backend specific clean-up
self._init_profile_cache(profile_id)
def _init_profile_cache(self, profile_id):
"""\
Inherit from this class to (re-)initialize profile ID based
cache storage.
:param profile_id: profile ID
:type profile_id: ``str``
"""
pass
[docs]
def populate_session_profiles(self):
"""\
Load a session profile set from the configuration storage
backend and make it available for this class.
:returns: a set of session profiles
:rtype: ``dict``
"""
self.session_profiles = self. _populate_session_profiles()
# scan for duplicate profile names and handle them...
scan_profile_names = {}
for profile_id in list(self.session_profiles.keys()):
profile_name = self.to_profile_name(profile_id)
if profile_name not in list(scan_profile_names.keys()):
scan_profile_names[profile_name] = [profile_id]
else:
scan_profile_names[profile_name].append(profile_id)
_duplicates = {}
for profile_name in list(scan_profile_names.keys()):
if len(scan_profile_names[profile_name]) > 1:
_duplicates[profile_name] = scan_profile_names[profile_name]
for profile_name in list(_duplicates.keys()):
i = 1
for profile_id in _duplicates[profile_name]:
self.update_value(None, 'name', '{name} ({i})'.format(name=profile_name, i=i), profile_id=profile_id)
i += 1
def _populate_session_profiles(self):
"""\
Inherit from this class and provide the backend specific way of loading /
populating a set of session profile via this method.
:returns: a set of session profiles
:rtype: ``dict``
"""
return {}
[docs]
def is_mutable(self, profile_id_or_name=None, profile_id=None):
"""\
Check if a given profile name (or ID) is mutable or not.
:param profile_id_or_name: profile name or profile ID (Default value = None)
:type profile_id_or_name: ``str``
:param profile_id: if the profile ID is known, pass it in directly and skip
the :func:`check_profile_id_or_name()` call (Default value = None)
:type profile_id: ``str``
:returns: ``True`` if the session profile of the specified name/ID is mutable
:rtype: ``bool``
:raises X2GoProfileException: if no such session profile exists
"""
try:
profile_id = profile_id or self.check_profile_id_or_name(profile_id_or_name)
return self._is_mutable(profile_id)
except X2GoProfileException:
return None
def _is_mutable(self, profile_id):
"""\
Inherit from this base class and provide your own decision making
code here if a given profile ID is mutable or not.
:param profile_id: profile ID
:type profile_id: ``str``
:returns: ``True`` if the session profile of the specified ID is mutable
:rtype: ``bool``
"""
return False
[docs]
def supports_mutable_profiles(self):
"""\
Check if the current session profile backend supports
mutable session profiles.
:returns: list of mutable profiles
:rtype: ``list``
"""
return self._supports_mutable_profiles()
def _supports_mutable_profiles(self):
"""\
Inherit from this base class and provide your own decision making
code here if a your session profile backend supports mutable
session profiles or not.
:returns: list of mutable profiles
:rtype: ``list``
"""
return False
[docs]
def mutable_profile_ids(self):
"""\
List all mutable session profiles.
:returns: List up all session profile IDs of mutable session profiles.
:rtype: ``bool``
"""
return [ p for p in self.profile_ids if self._is_mutable(p) ]
[docs]
def write(self):
"""\
Store session profile data to the storage backend.
:returns: ``True`` if the write process has been successfull, ``False`` otherwise
:rtype: ``bool``
"""
# then update profile IDs for profiles that have a renamed host attribute...
for profile_id in self._profiles_need_profile_id_renewal:
_config = self.get_profile_config(profile_id=profile_id)
self._delete_profile(profile_id)
try: del self._cached_profile_ids[profile_id]
except KeyError: pass
self.add_profile(profile_id=None, force_add=True, **_config)
self._profiles_need_profile_id_renewal = []
self._cached_profile_ids = {}
return self._write()
def _write(self):
"""\
Write session profiles back to session profile storage backend. Inherit from this
class and adapt to the session profile backend via this method.
"""
return True
[docs]
def get_profile_option_type(self, option):
"""\
Get the data type for a specific session profile option.
:param option: the option to get the data type for
:type option: will be detected by this method
:returns: the data type of ``option``
:rtype: ``type``
"""
try:
return type(self.defaultSessionProfile[option])
except KeyError:
return bytes
[docs]
def get_profile_config(self, profile_id_or_name=None, parameter=None, profile_id=None):
"""\
The configuration options for a single session profile.
:param profile_id_or_name: either profile ID or profile name is accepted (Default value = None)
:type profile_id_or_name: ``str``
:param parameter: if specified, only the value for the given parameter is returned (Default value = None)
:type parameter: ``str``
:param profile_id: profile ID (faster than specifying ``profile_id_or_name``) (Default value = None)
:type profile_id: ``str``
:returns: the session profile configuration for the given profile ID (or name)
:rtype: ``dict``
"""
_profile_id = profile_id or self.check_profile_id_or_name(profile_id_or_name)
_profile_config = {}
if parameter is None:
parameters = self._get_profile_options(_profile_id)
else:
parameters = [parameter]
for option in parameters:
value = self._get_profile_parameter(_profile_id, option, key_type=self.get_profile_option_type(option))
if type(value) is bytes:
value = str(value)
# the type check is a nasty Python2/Python3 hack around unicodes and strings
if option == 'export' and (type(value) is type(u'') or type(value) is type('')):
_value = value.replace(',', ';').strip().strip('"').strip().strip(';').strip()
value = {}
if _value:
_export_paths = _value.split(';')
for _path in _export_paths:
if not re.match('.*:(0|1)$', _path): _path = '%s:1' % _path
_auto_export_path = re.match('.*:1$', _path) and True or False
_export_path = ':'.join(_path.split(':')[:-1])
value[_export_path] = _auto_export_path
_profile_config[option] = value
if parameter is not None:
if parameter in list(_profile_config.keys()):
value = _profile_config[parameter]
return value
else:
raise X2GoProfileException('no such session profile parameter: %s' % parameter)
return _profile_config
[docs]
def default_profile_config(self):
"""\
Return a default session profile.
:returns: default session profile
:rtype: ``dict``
"""
return copy.deepcopy(self.defaultSessionProfile)
[docs]
def has_profile(self, profile_id_or_name):
"""\
Does a session profile of a given profile ID or profile name exist?
:param profile_id_or_name: profile ID or profile name
:type profile_id_or_name: ``str``
:returns: ``True`` if there is such a session profile, ``False`` otherwise
:rtype: ``bool``
"""
try:
self.check_profile_id_or_name(profile_id_or_name)
return True
except X2GoProfileException:
return False
def _update_profile_ids_cache(self):
for p in self._get_profile_ids():
if p not in self._non_profile_sections:
self._cached_profile_ids[p] = self.to_profile_name(p)
@property
def profile_ids(self):
"""\
Render a list of all profile IDs found in the session profiles configuration.
"""
if not self._cached_profile_ids:
self._update_profile_ids_cache()
return list(self._cached_profile_ids.keys())
def _get_profile_ids(self):
"""\
Inherit from this class and provide a way for actually getting
a list of session profile IDs from the storage backend via this method.
:returns: list of available session profile IDs
:rtype: ``list``
"""
return []
[docs]
def has_profile_id(self, profile_id):
"""\
Does a session profile of a given profile ID exist? (Faster than :func:`has_profile()`.)
:param profile_id: profile ID
:type profile_id: ``str``
:returns: ``True`` if there is such a session profile, ``False`` otherwise
:rtype: ``bool``
"""
return str(profile_id) in self.profile_ids
@property
def profile_names(self):
"""\
Render a list of all profile names found in the session profiles configuration.
"""
if not self._cached_profile_ids:
self._update_profile_ids_cache()
return list(self._cached_profile_ids.values())
[docs]
def has_profile_name(self, profile_name):
"""\
Does a session profile of a given profile name exist? (Faster than :func:`has_profile()`.)
:param profile_name: profile name
:type profile_name: ``str``
:returns: ``True`` if there is such a session profile, ``False`` otherwise
:rtype: ``bool``
"""
return str(profile_name) in self.profile_names
[docs]
def to_profile_id(self, profile_name):
"""\
Convert profile name to profile ID.
:param profile_name: profile name
:type profile_name: ``str``
:returns: profile ID
:rtype: ``str``
"""
_profile_ids = [ p for p in self.profile_ids if self._cached_profile_ids[p] == profile_name ]
if len(_profile_ids) == 1:
return str(_profile_ids[0])
elif len(_profile_ids) == 0:
return None
else:
raise X2GoProfileException('The sessions config file contains multiple session profiles with name: %s' % profile_name)
[docs]
def to_profile_name(self, profile_id):
"""\
Convert profile ID to profile name.
:param profile_id: profile ID
:type profile_id: ``str``
:returns: profile name
:rtype: ``str``
"""
try:
_profile_name = self.get_profile_config(profile_id=profile_id, parameter='name')
return str(_profile_name)
except:
return ''
[docs]
def add_profile(self, profile_id=None, force_add=False, **kwargs):
"""\
Add a new session profile.
:param profile_id: a custom profile ID--if left empty a profile ID will be auto-generated (Default value = None)
:type profile_id: ``str``
:param kwargs: session profile options for this new session profile
:type kwargs: ``dict``
:param force_add: enforce adding of the given profile (Default value = False)
:type force_add: ``bool``
:returns: the (auto-generated) profile ID of the new session profile
:rtype: ``str``
"""
if profile_id is None or profile_id in self.profile_ids:
profile_id = utils._genSessionProfileId()
self.session_profiles[profile_id] = self.default_profile_config()
if 'name' not in list(kwargs.keys()):
raise X2GoProfileException('session profile parameter ,,name\'\' is missing in method parameters')
if kwargs['name'] in self.profile_names and not force_add:
raise X2GoProfileException('a profile of name ,,%s\'\' already exists' % kwargs['name'])
self._cached_profile_ids[profile_id] = kwargs['name']
for key, value in list(kwargs.items()):
self.update_value(None, key, value, profile_id=profile_id)
_default_session_profile = self.default_profile_config()
for key, value in list(_default_session_profile.items()):
if key in kwargs: continue
self.update_value(None, key, value, profile_id=profile_id)
self._cached_profile_ids = {}
return str(profile_id)
[docs]
def delete_profile(self, profile_id_or_name):
"""\
Delete a session profile from the configuration file.
:param profile_id_or_name: profile ID or profile name
:type profile_id_or_name: ``str``
"""
_profile_id = self.check_profile_id_or_name(profile_id_or_name)
self._delete_profile(_profile_id)
self.write_user_config = True
self.write()
self._cached_profile_ids = {}
def _delete_profile(self, profile_id):
"""\
Inherit from this class and provide a way for actually deleting
a complete session profile from the storage backend via this method.
:param profile_id: Profile ID of the profile to be deleted
:type profile_id: ``str``
"""
pass
[docs]
def update_value(self, profile_id_or_name, option, value, profile_id=None):
"""\
Update a value in a session profile.
:param profile_id_or_name: the profile ID
:type profile_id_or_name: ``str``
:param option: the session profile option of the given profile ID
:type option: ``str``
:param value: the value to update the session profile option with
:type value: any type, depends on the session profile option
:param profile_id: if the profile ID is known, pass it in directly and skip
the :func:`check_profile_id_or_name()` call (Default value = None)
:type profile_id: ``str``
"""
try:
profile_id = profile_id or self.check_profile_id_or_name(profile_id_or_name)
except X2GoProfileException:
profile_id = profile_id_or_name
if not self.is_mutable(profile_id=profile_id):
raise X2GoProfileException("session profile cannot be modified, it is marked as immutable")
if option == 'name':
profile_name = value
current_profile_name = self.get_value(profile_id, option)
if not profile_name:
raise X2GoProfileException('profile name for profile id %s must not be empty' % profile_id)
else:
if profile_name != current_profile_name:
try: del self._cached_profile_ids[profile_id]
except KeyError: pass
if profile_name in self.profile_names:
raise X2GoProfileException('a profile of name ,,%s\'\' already exists' % profile_name)
self._cached_profile_ids[profile_id] = profile_name
if option == 'export' and type(value) == dict:
_strvalue = '"'
for folder in list(value.keys()):
_strvalue += "%s:%s;" % (folder, int(value[folder]))
_strvalue += '"'
_strvalue = _strvalue.replace('""', '')
value = _strvalue
if option == 'host':
_host = self.get_profile_config(profile_id=profile_id, parameter='host')
if _host != value and _host is not None:
self._profiles_need_profile_id_renewal.append(profile_id)
if type(value) is tuple:
value = list(value)
if type(value) is not list:
value = value.split(',')
self._update_value(profile_id, option, value)
def _update_value(self, profile_id, option, value):
"""\
Inherit from this class and provide for actually updating
a session profile's value in the storage backend via this method.
:param profile_id: the profile ID of the profile to be updated
:type profile_id: ``str``
:param option: the option to be updated
:type option: ``str``
:param value: the value to be updated for the given option
:type value: ``str`` or ``list`` or ``int`` or ``bool``
"""
pass
[docs]
def check_profile_id_or_name(self, profile_id_or_name):
"""\
Detect the profile ID from a given string which maybe profile ID or profile name.
:param profile_id_or_name: profile ID or profile name
:type profile_id_or_name: ``str``
:returns: profile ID
:rtype: ``str``
:raises X2GoProfileException: if no such session profile exists
"""
_profile_id = None
if self.has_profile_name(profile_id_or_name):
# we were given a sesion profile name...
_profile_id = self.to_profile_id(profile_id_or_name)
elif self.has_profile_id(profile_id_or_name):
# we were given a session profile id...
_profile_id = profile_id_or_name
else:
raise X2GoProfileException('No session profile with id or name ,,%s\'\' exists.' % profile_id_or_name)
if _profile_id is not None:
_profile_id = str(_profile_id)
return _profile_id
[docs]
def to_session_params(self, profile_id_or_name=None, profile_id=None):
"""\
Convert session profile options to :class:`x2go.session.X2GoSession` constructor method parameters.
:param profile_id_or_name: either profile ID or profile name is accepted (Default value = None)
:type profile_id_or_name: ``str``
:param profile_id: profile ID (fast than specifying ``profile_id_or_name``) (Default value = None)
:type profile_id: ``str``
:returns: a dictionary of :class:`x2go.session.X2GoSession` constructor method parameters
:rtype: ``dict``
"""
_profile_id = profile_id or self.check_profile_id_or_name(profile_id_or_name)
return utils._convert_SessionProfileOptions_2_SessionParams(self.get_profile_config(_profile_id))
[docs]
def get_session_param(self, profile_id_or_name, param):
"""\
Get a single :class:`x2go.session.X2GoSession` parameter from a specific session profile.
:param profile_id_or_name: either profile ID or profile name is accepted
:type profile_id_or_name: ``str``
:param param: the parameter name in the :class:`x2go.session.X2GoSession` constructor method
:type param: ``str``
:returns: the value of the session profile option represented by ``param``
:rtype: depends on the session profile option requested
"""
return self.to_session_params(profile_id_or_name)[param]
def _get_profile_parameter(self, profile_id, option, key_type):
"""\
Inherit from this class and provide a way for actually obtaining
the value of a specific profile parameter.
:param profile_id: the profile's unique ID
:type profile_id: ``str``
:param option: the session profile option for which to retrieve its value
:type option: ``str``
:param key_type: type of the value to return
:type key_type: ``typeobject``
:returns: value of a session profile parameter
:rtype: ``various types``
"""
return None
def _get_profile_options(self, profile_id):
"""\
Inherit from this class and provide a way for actually obtaining
a list of available profile options of a given session profile.
:param profile_id: the profile ID of the profile to operate on
:type profile_id: ``str``
:returns: list of available option is the given session profile
:rtype: ``list``
"""
return []
[docs]
def get_server_hostname(self, profile_id):
"""\
Retrieve host name of the X2Go Server configured in a session profile.
:param profile_id: the profile's unique ID
:type profile_id: ``str``
:returns: the host name of the X2Go Server configured by the session profile
of the given profile ID
:rtype: ``list``
"""
return str(self._get_server_hostname(profile_id))
def _get_server_hostname(self, profile_id):
"""\
Inherit from this class and provide a way for actually obtaining
a the server host name for a given profile ID.
:param profile_id: the profile's unique ID
:type profile_id: ``str``
:returns: the host name of the X2Go Server configured by the session profile
of the given profile ID
:rtype: ``list``
"""
return 'localhost'
[docs]
def get_server_port(self, profile_id):
"""\
Retrieve SSH port of the X2Go Server configured in a session profile.
:param profile_id: the profile's unique ID
:type profile_id: ``str``
:returns: the SSH port of the X2Go Server configured by the session profile
of the given profile ID
:rtype: ``list``
"""
return self._get_server_port(profile_id)
def _get_server_port(self, profile_id):
"""\
Inherit from this class and provide a way for actually obtaining
a the server SSH port for a given profile ID.
:param profile_id: the profile's unique ID
:type profile_id: ``str``
:returns: the SSH port of the X2Go Server configured by the session profile
of the given profile ID
:rtype: ``list``
"""
return 22
[docs]
def get_pkey_object(self, profile_id):
"""\
If available, return a PKey (Paramiko/SSH private key) object.
:param profile_id: the profile's unique ID
:type profile_id: ``str``
:returns: a Paramiko/SSH PKey object
:rtype: ``obj``
"""
return self._get_pkey_object(profile_id)
def _get_pkey_object(self, profile_id):
"""\
Inherit from this class and provide a way for actually
providing such a PKey object.
:param profile_id: the profile ID for which to retrieve the PKey object
:type profile_id: ``str``
"""
return None