# -*- 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.control.plain.X2GoControlSession` class - core functions for handling your individual X2Go sessions.
This backend handles X2Go server implementations that respond via server-side PLAIN text output.
"""
__NAME__ = 'x2gocontrolsession-pylib'
__package__ = 'x2go.backends.control'
__name__ = 'x2go.backends.control.plain'
# modules
import os
import sys
import types
import paramiko
import gevent
import copy
import string
import random
import re
import locale
import threading
import io
import base64
import uuid
from gevent import socket
# Python X2Go modules
import x2go.sshproxy as sshproxy
import x2go.log as log
import x2go.utils as utils
import x2go.x2go_exceptions as x2go_exceptions
import x2go.defaults as defaults
import x2go.checkhosts as checkhosts
from x2go.defaults import BACKENDS as _BACKENDS
import x2go._paramiko
x2go._paramiko.monkey_patch_paramiko()
def _rerewrite_blanks(cmd):
"""\
In command strings X2Go server scripts expect blanks being rewritten to ,,X2GO_SPACE_CHAR''.
Commands get rewritten in the terminal sessions. This re-rewrite function helps
displaying command string in log output.
:param cmd: command that has to be rewritten for log output
:type cmd: ``str``
:returns: the command with ,,X2GO_SPACE_CHAR'' re-replaced by blanks
:rtype: ``str``
"""
# X2Go run command replace X2GO_SPACE_CHAR string with blanks
if cmd:
cmd = cmd.replace("X2GO_SPACE_CHAR", " ")
return cmd
def _rewrite_password(cmd, user=None, password=None):
"""\
In command strings Python X2Go replaces some macros with actual values:
- X2GO_USER -> the user name under which the user is authenticated via SSH
- X2GO_PASSWORD -> the password being used for SSH authentication
Both macros can be used to on-the-fly authenticate via RDP.
:param cmd: command that is to be sent to an X2Go server script
:type cmd: ``str``
:param user: the SSH authenticated user name (Default value = None)
:type user: ``str``
:param password: the password being used for SSH authentication (Default value = None)
:type password: ``str``
:returns: the command with macros replaced
:rtype: ``str``
"""
# if there is a ,,-u X2GO_USER'' parameter in RDP options then we will replace
# it by our X2Go session password
if cmd and user:
cmd = cmd.replace('X2GO_USER', user)
# if there is a ,,-p X2GO_PASSWORD'' parameter in RDP options then we will replace
# it by our X2Go session password
if cmd and password:
cmd = cmd.replace('X2GO_PASSWORD', password)
return cmd
[docs]
class X2GoControlSession(paramiko.SSHClient):
"""\
In the Python X2Go concept, X2Go sessions fall into two parts: a control session and one to many terminal sessions.
The control session handles the SSH based communication between server and client. It is mainly derived from
``paramiko.SSHClient`` and adds on X2Go related functionality.
"""
def __init__(self,
profile_name='UNKNOWN',
add_to_known_hosts=False,
known_hosts=None,
forward_sshagent=False,
unique_hostkey_aliases=False,
terminal_backend=_BACKENDS['X2GoTerminalSession']['default'],
info_backend=_BACKENDS['X2GoServerSessionInfo']['default'],
list_backend=_BACKENDS['X2GoServerSessionList']['default'],
proxy_backend=_BACKENDS['X2GoProxy']['default'],
client_rootdir=os.path.join(defaults.LOCAL_HOME, defaults.X2GO_CLIENT_ROOTDIR),
sessions_rootdir=os.path.join(defaults.LOCAL_HOME, defaults.X2GO_SESSIONS_ROOTDIR),
ssh_rootdir=os.path.join(defaults.LOCAL_HOME, defaults.X2GO_SSH_ROOTDIR),
logger=None, loglevel=log.loglevel_DEFAULT,
published_applications_no_submenus=0,
low_latency=False,
**kwargs):
"""\
Initialize an X2Go control session. For each connected session profile there will be one SSH-based
control session and one to many terminal sessions that all server-client-communicate via this one common control
session.
A control session normally gets set up by an :class:`x2go.session.X2GoSession` instance. Do not use it directly!!!
:param profile_name: the profile name of the session profile this control session works for
:type profile_name: ``str``
:param add_to_known_hosts: Auto-accept server host validity?
:type add_to_known_hosts: ``bool``
:param known_hosts: the underlying Paramiko/SSH systems ``known_hosts`` file
:type known_hosts: ``str``
:param forward_sshagent: forward SSH agent authentication requests to the X2Go client-side
:type forward_sshagent: ``bool``
:param unique_hostkey_aliases: instead of storing [<hostname>]:<port> in known_hosts file, use the
(unique-by-design) profile ID
:type unique_hostkey_aliases: ``bool``
:param terminal_backend: X2Go terminal session backend to use
:type terminal_backend: ``str``
:param info_backend: backend for handling storage of server session information
:type info_backend: ``X2GoServerSessionInfo*`` instance
:param list_backend: backend for handling storage of session list information
:type list_backend: ``X2GoServerSessionList*`` instance
:param proxy_backend: backend for handling the X-proxy connections
:type proxy_backend: ``X2GoProxy*`` instance
:param client_rootdir: client base dir (default: ~/.x2goclient)
:type client_rootdir: ``str``
:param sessions_rootdir: sessions base dir (default: ~/.x2go)
:type sessions_rootdir: ``str``
:param ssh_rootdir: ssh base dir (default: ~/.ssh)
:type ssh_rootdir: ``str``
:param published_applications_no_submenus: published applications menus with less items than ``published_applications_no_submenus``
are rendered without submenus
:type published_applications_no_submenus: ``int``
:param logger: you can pass an :class:`x2go.log.X2GoLogger` object to the
:class:`x2go.backends.control.plain.X2GoControlSession` 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``
:param low_latency: set this boolean switch for weak connections, it will double all timeout values.
:type low_latency: ``bool``
:param kwargs: catch any non-defined parameters in ``kwargs``
:type kwargs: ``dict``
"""
self.associated_terminals = {}
self.terminated_terminals = []
self.profile_name = profile_name
self.add_to_known_hosts = add_to_known_hosts
self.known_hosts = known_hosts
self.forward_sshagent = forward_sshagent
self.unique_hostkey_aliases = unique_hostkey_aliases
self.hostname = None
self.port = None
self.sshproxy_session = None
self._session_auth_rsakey = None
self._remote_home = None
self._remote_group = {}
self._remote_username = None
self._remote_peername = None
self._server_versions = None
self._server_features = None
if logger is None:
self.logger = log.X2GoLogger(loglevel=loglevel)
else:
self.logger = copy.deepcopy(logger)
self.logger.tag = __NAME__
self._terminal_backend = terminal_backend
self._info_backend = info_backend
self._list_backend = list_backend
self._proxy_backend = proxy_backend
self.client_rootdir = client_rootdir
self.sessions_rootdir = sessions_rootdir
self.ssh_rootdir = ssh_rootdir
self._published_applications_menu = {}
self.agent_chan = None
self.agent_handler = None
paramiko.SSHClient.__init__(self)
if self.add_to_known_hosts:
self.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.session_died = False
self.low_latency = low_latency
self.published_applications_no_submenus = published_applications_no_submenus
self._already_querying_published_applications = threading.Lock()
self._transport_lock = threading.Lock()
[docs]
def get_hostname(self):
"""\
Get the hostname as stored in the properties of this control session.
:returns: the hostname of the connected X2Go server
:rtype: ``str``
"""
return self.hostname
[docs]
def get_port(self):
"""\
Get the port number of the SSH connection as stored in the properties of this control session.
:returns: the server-side port number of the control session's SSH connection
:rtype: ``str``
"""
return self.port
[docs]
def load_session_host_keys(self):
"""\
Load known SSH host keys from the ``known_hosts`` file.
If the file does not exist, create it first.
"""
if self.known_hosts is not None:
utils.touch_file(self.known_hosts)
self.load_host_keys(self.known_hosts)
def __del__(self):
"""\
On instance descruction, do a proper session disconnect from the server.
"""
self.disconnect()
[docs]
def test_sftpclient(self):
ssh_transport = self.get_transport()
try:
self.sftp_client = paramiko.SFTPClient.from_transport(ssh_transport)
except (AttributeError, paramiko.SFTPError):
raise x2go_exceptions.X2GoSFTPClientException('failed to initialize SFTP channel')
def _x2go_sftp_put(self, local_path, remote_path, timeout=20):
"""\
Put a local file on the remote server via sFTP.
During sFTP operations, remote command execution gets blocked.
:param local_path: full local path name of the file to be put on the server
:type local_path: ``str``
:param remote_path: full remote path name of the server-side target location, path names have to be Unix-compliant
:type remote_path: ``str``
:param timeout: this SFTP put action should not take longer then the given value (Default value = 20)
:type timeout: ``int``
:raises X2GoControlSessionException: if the SSH connection dropped out
"""
ssh_transport = self.get_transport()
self._transport_lock.acquire()
if ssh_transport and ssh_transport.is_authenticated():
self.logger('sFTP-put: %s -> %s:%s' % (os.path.normpath(local_path), self.remote_peername(), remote_path), loglevel=log.loglevel_DEBUG)
if self.low_latency: timeout = timeout * 2
timer = gevent.Timeout(timeout)
timer.start()
try:
try:
self.sftp_client = paramiko.SFTPClient.from_transport(ssh_transport)
except paramiko.SFTPError:
self._transport_lock.release()
raise x2go_exceptions.X2GoSFTPClientException('failed to initialize SFTP channel')
try:
self.sftp_client.put(os.path.normpath(local_path), remote_path)
except (x2go_exceptions.SSHException, socket.error, IOError):
# react to connection dropped error for SSH connections
self.session_died = True
self._transport_lock.release()
raise x2go_exceptions.X2GoControlSessionException('The SSH connection was dropped during an sFTP put action.')
except gevent.timeout.Timeout:
self.session_died = True
self._transport_lock.release()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise x2go_exceptions.X2GoControlSessionException('the X2Go control session timed out during an SFTP write command')
finally:
timer.cancel()
self.sftp_client = None
if self._transport_lock.locked():
self._transport_lock.release()
def _x2go_sftp_write(self, remote_path, content, timeout=20):
"""\
Create a text file on the remote server via sFTP.
During sFTP operations, remote command execution gets blocked.
:param remote_path: full remote path name of the server-side target location, path names have to be Unix-compliant
:type remote_path: ``str``
:param content: a text file, multi-line files use Unix-link EOL style
:type content: ``str``
:param timeout: this SFTP write action should not take longer then the given value (Default value = 20)
:type timeout: ``int``
:raises X2GoControlSessionException: if the SSH connection dropped out
"""
ssh_transport = self.get_transport()
self._transport_lock.acquire()
if ssh_transport and ssh_transport.is_authenticated():
self.logger('sFTP-write: opening remote file %s on host %s for writing' % (remote_path, self.remote_peername()), loglevel=log.loglevel_DEBUG)
if self.low_latency: timeout = timeout * 2
timer = gevent.Timeout(timeout)
timer.start()
try:
try:
self.sftp_client = paramiko.SFTPClient.from_transport(ssh_transport)
except paramiko.SFTPError:
self._transport_lock.release()
raise x2go_exceptions.X2GoSFTPClientException('failed to initialize SFTP channel')
try:
remote_fileobj = self.sftp_client.open(remote_path, 'w')
self.logger('sFTP-write: writing content: %s' % content, loglevel=log.loglevel_DEBUG_SFTPXFER)
remote_fileobj.write(content)
remote_fileobj.close()
except (x2go_exceptions.SSHException, socket.error, IOError):
self.session_died = True
self._transport_lock.release()
self.logger('sFTP-write: opening remote file %s on host %s failed' % (remote_path, self.remote_peername()), loglevel=log.loglevel_WARN)
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise x2go_exceptions.X2GoControlSessionException('The SSH connection was dropped during an sFTP write action.')
except gevent.timeout.Timeout:
self.session_died = True
self._transport_lock.release()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise x2go_exceptions.X2GoControlSessionException('the X2Go control session timed out during an SFTP write command')
finally:
timer.cancel()
self.sftp_client = None
if self._transport_lock.locked():
self._transport_lock.release()
def _x2go_sftp_remove(self, remote_path, timeout=20):
"""\
Remote a remote file from the server via sFTP.
During sFTP operations, remote command execution gets blocked.
:param remote_path: full remote path name of the server-side file to be removed, path names have to be Unix-compliant
:type remote_path: ``str``
:param timeout: this SFTP remove action should not take longer then the given value (Default value = 20)
:type timeout: ``int``
:raises X2GoControlSessionException: if the SSH connection dropped out
"""
ssh_transport = self.get_transport()
self._transport_lock.acquire()
if ssh_transport and ssh_transport.is_authenticated():
self.logger('sFTP-write: removing remote file %s on host %s' % (remote_path, self.remote_peername()), loglevel=log.loglevel_DEBUG)
if self.low_latency: timeout = timeout * 2
timer = gevent.Timeout(timeout)
timer.start()
try:
try:
self.sftp_client = paramiko.SFTPClient.from_transport(ssh_transport)
except paramiko.SFTPError:
self._transport_lock.release()
raise x2go_exceptions.X2GoSFTPClientException('failed to initialize SFTP channel')
try:
self.sftp_client.remove(remote_path)
except (x2go_exceptions.SSHException, socket.error, IOError):
self.session_died = True
self._transport_lock.release()
self.logger('sFTP-write: removing remote file %s on host %s failed' % (remote_path, self.remote_peername()), loglevel=log.loglevel_WARN)
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise x2go_exceptions.X2GoControlSessionException('The SSH connection was dropped during an sFTP remove action.')
except gevent.timeout.Timeout:
self.session_died = True
self._transport_lock.release()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise x2go_exceptions.X2GoControlSessionException('the X2Go control session timed out during an SFTP write command')
finally:
timer.cancel()
self.sftp_client = None
if self._transport_lock.locked():
self._transport_lock.release()
def _x2go_exec_command(self, cmd_line, loglevel=log.loglevel_INFO, timeout=20, **kwargs):
"""\
Execute an X2Go server-side command via SSH.
During SSH command executions, sFTP operations get blocked.
:param cmd_line: the command to be executed on the remote server
:type cmd_line: ``str`` or ``list``
:param loglevel: use this loglevel for reporting about remote command execution (Default value = log.loglevel_INFO)
:type loglevel: ``int``
:param timeout: if commands take longer than ``<timeout>`` to be executed, consider the control session connection
to have died. (Default value = 20)
:type timeout: ``int``
:param kwargs: parameters that get passed through to the ``paramiko.SSHClient.exec_command()`` method.
:type kwargs: ``dict``
:returns: ``True`` if the command could be successfully executed on the remote X2Go server
:rtype: ``bool``
:raises X2GoControlSessionException: if the command execution failed (due to a lost connection)
"""
if type(cmd_line) == list:
cmd = " ".join(cmd_line)
else:
cmd = cmd_line
cmd_uuid = str(uuid.uuid1())
cmd = 'echo X2GODATABEGIN:%s; PATH=/usr/local/bin:/usr/bin:/bin sh -c \"%s\"; echo X2GODATAEND:%s' % (cmd_uuid, cmd, cmd_uuid)
if self.session_died:
self.logger("control session seams to be dead, not executing command ,,%s'' on X2Go server %s" % (_rerewrite_blanks(cmd), self.profile_name,), loglevel=loglevel)
if sys.version_info[0] >= 3:
return (io.BytesIO(), io.BytesIO(), io.BytesIO(b'failed to execute command'))
else:
return (io.StringIO(), io.StringIO(), io.StringIO(u'failed to execute command'))
self._transport_lock.acquire()
_retval = None
_password = None
ssh_transport = self.get_transport()
if ssh_transport and ssh_transport.is_authenticated():
if self.low_latency: timeout = timeout * 2
timer = gevent.Timeout(timeout)
timer.start()
try:
self.logger("executing command on X2Go server ,,%s'': %s" % (self.profile_name, _rerewrite_blanks(cmd)), loglevel=loglevel)
if self._session_password:
_password = base64.b64decode(self._session_password)
_retval = self.exec_command(_rewrite_password(cmd, user=self.get_transport().get_username(), password=_password), **kwargs)
except AttributeError:
self.session_died = True
self._transport_lock.release()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise x2go_exceptions.X2GoControlSessionException('the X2Go control session has died unexpectedly')
except EOFError:
self.session_died = True
self._transport_lock.release()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise x2go_exceptions.X2GoControlSessionException('the X2Go control session has died unexpectedly')
except x2go_exceptions.SSHException:
self.session_died = True
self._transport_lock.release()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise x2go_exceptions.X2GoControlSessionException('the X2Go control session has died unexpectedly')
except gevent.timeout.Timeout:
self.session_died = True
self._transport_lock.release()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise x2go_exceptions.X2GoControlSessionException('the X2Go control session command timed out')
except socket.error:
self.session_died = True
self._transport_lock.release()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise x2go_exceptions.X2GoControlSessionException('the X2Go control session has died unexpectedly')
finally:
timer.cancel()
else:
self._transport_lock.release()
raise x2go_exceptions.X2GoControlSessionException('the X2Go control session is not connected (while issuing SSH command=%s)' % cmd)
if self._transport_lock.locked():
self._transport_lock.release()
sanitized_stdout = u''
is_x2go_data = False
# sanitized X2Go relevant data, protect against data injection via .bashrc files
(stdin, stdout, stderr) = _retval
raw_stdout = stdout.read()
# Python 3 needs a decoding from bytestring to string
if sys.version_info[0] >= 3:
raw_stdout = raw_stdout.decode()
else:
if type(raw_stdout) is not types.UnicodeType:
raw_stdout = raw_stdout.decode('utf-8')
for line in raw_stdout.split('\n'):
if line.startswith('X2GODATABEGIN:'+cmd_uuid):
is_x2go_data = True
continue
if not is_x2go_data: continue
if line.startswith('X2GODATAEND:'+cmd_uuid): break
sanitized_stdout += line + "\n"
if sys.version_info[0] >= 3:
_stdout_new = io.BytesIO(sanitized_stdout.encode())
else:
_stdout_new = io.StringIO(sanitized_stdout)
_retval = (stdin, _stdout_new, stderr)
return _retval
@property
def _x2go_server_versions(self):
"""\
Render a dictionary of server-side X2Go components and their versions. Results get cached
once there has been one successful query.
"""
if self._server_versions is None:
self._server_versions = {}
(stdin, stdout, stderr) = self._x2go_exec_command('which x2goversion >/dev/null && x2goversion')
if sys.version_info[0] >= 3:
_lines = stdout.read().decode().split('\n')
else:
_lines = stdout.read().split('\n')
for _line in _lines:
if ':' not in _line: continue
comp = _line.split(':')[0].strip()
version = _line.split(':')[1].strip()
self._server_versions.update({comp: version})
self.logger('server-side X2Go components and their versions are: %s' % self._server_versions, loglevel=log.loglevel_DEBUG)
return self._server_versions
[docs]
def query_server_versions(self, force=False):
"""\
Do a query for the server-side list of X2Go components and their versions.
:param force: do not use the cached component list, really ask the server (again) (Default value = False)
:type force: ``bool``
:returns: dictionary of X2Go components (as keys) and their versions (as values)
:rtype: ``list``
"""
if force:
self._server_versions = None
return self._x2go_server_versions
get_server_versions = query_server_versions
@property
def _x2go_server_features(self):
"""\
Render a list of server-side X2Go features. Results get cached once there has been one successful query.
"""
if self._server_features is None:
(stdin, stdout, stderr) = self._x2go_exec_command('which x2gofeaturelist >/dev/null && x2gofeaturelist')
_stdout = stdout.read()
if sys.version_info[0] >= 3:
_stdout = _stdout.decode()
self._server_features = _stdout.split('\n')
self._server_features = [ f for f in self._server_features if f ]
self._server_features.sort()
self.logger('server-side X2Go features are: %s' % self._server_features, loglevel=log.loglevel_DEBUG)
return self._server_features
[docs]
def query_server_features(self, force=False):
"""\
Do a query for the server-side list of X2Go features.
:param force: do not use the cached feature list, really ask the server (again) (Default value = False)
:type force: ``bool``
:returns: list of X2Go feature names
:rtype: ``list``
"""
if force:
self._server_features = None
return self._x2go_server_features
get_server_features = query_server_features
@property
def _x2go_remote_home(self):
"""\
Retrieve and cache the remote home directory location.
"""
if self._remote_home is None:
(stdin, stdout, stderr) = self._x2go_exec_command('echo $HOME')
_stdout = stdout.read()
if sys.version_info[0] >= 3:
_stdout = _stdout.decode()
if _stdout:
self._remote_home = _stdout.split()[0]
self.logger('remote user\' home directory: %s' % self._remote_home, loglevel=log.loglevel_DEBUG)
return self._remote_home
else:
return self._remote_home
def _x2go_remote_group(self, group):
"""\
Retrieve and cache the members of a server-side POSIX group.
:param group: remote POSIX group name
:type group: ``str``
:returns: list of POSIX group members
:rtype: ``list``
"""
if group not in self._remote_group:
(stdin, stdout, stderr) = self._x2go_exec_command('getent group %s | cut -d":" -f4' % group)
self._remote_group[group] = stdout.read().split('\n')[0].split(',')
self.logger('remote %s group: %s' % (group, self._remote_group[group]), loglevel=log.loglevel_DEBUG)
return self._remote_group[group]
else:
return self._remote_group[group]
[docs]
def is_x2gouser(self, username):
"""\
Is the remote user allowed to launch X2Go sessions?
FIXME: this method is currently non-functional.
:param username: remote user name
:type username: ``str``
:returns: ``True`` if the remote user is allowed to launch X2Go sessions
:rtype: ``bool``
"""
###
### FIXME:
###
# discussion about server-side access restriction based on posix group membership or similar currently
# in process (as of 20110517, mg)
#return username in self._x2go_remote_group('x2gousers')
return True
[docs]
def is_sshfs_available(self):
"""\
Check if the remote user is allowed to use SSHFS mounts.
:returns: ``True`` if the user is allowed to connect client-side shares to the X2Go session
:rtype: ``bool``
"""
(stdin, stdout, stderr) = self._x2go_exec_command('which fusermount')
# if which returns the full path of fusermount, the current use is allowed to execute it
return bool(stdout.read())
[docs]
def remote_username(self):
"""\
Returns (and caches) the control session's remote username.
:returns: SSH transport's user name
:rtype: ``str``
:raises X2GoControlSessionException: on SSH connection loss
"""
if self._remote_username is None:
if self.get_transport() is not None:
try:
self._remote_username = self.get_transport().get_username()
except:
self.session_died = True
raise x2go_exceptions.X2GoControlSessionException('Lost connection to X2Go server')
return self._remote_username
[docs]
def remote_peername(self):
"""\
Returns (and caches) the control session's remote host (name or ip).
:returns: SSH transport's peer name
:rtype: ``tuple``
:raises X2GoControlSessionException: on SSH connection loss
"""
if self._remote_peername is None:
if self.get_transport() is not None:
try:
self._remote_peername = self.get_transport().getpeername()
except:
self.session_died = True
raise x2go_exceptions.X2GoControlSessionException('Lost connection to X2Go server')
return self._remote_peername
@property
def _x2go_session_auth_rsakey(self):
"""\
Generate (and cache) a temporary RSA host key for the lifetime of this control session.
"""
if self._session_auth_rsakey is None:
self._session_auth_rsakey = paramiko.RSAKey.generate(defaults.RSAKEY_STRENGTH)
return self._session_auth_rsakey
[docs]
def set_profile_name(self, profile_name):
"""\
Manipulate the control session's profile name.
:param profile_name: new profile name for this control session
:type profile_name: ``str``
"""
self.profile_name = profile_name
[docs]
def check_host(self, hostname, port=22):
"""\
Wraps around a Paramiko/SSH host key check.
:param hostname: the remote X2Go server's hostname
:type hostname: ``str``
:param port: the SSH port of the remote X2Go server (Default value = 22)
:type port: ``int``
:returns: ``True`` if the host key check succeeded, ``False`` otherwise
:rtype: ``bool``
"""
# trailing whitespace tolerance
hostname = hostname.strip()
# force into IPv4 for localhost connections
if hostname in ('localhost', 'localhost.localdomain'):
hostname = '127.0.0.1'
return checkhosts.check_ssh_host_key(self, hostname, port=port)
[docs]
def connect(self, hostname, port=22, username=None, password=None, passphrase=None, pkey=None,
key_filename=None, timeout=None, allow_agent=False, look_for_keys=False,
use_sshproxy=False, sshproxy_host=None, sshproxy_port=22, sshproxy_user=None, sshproxy_password=None, sshproxy_force_password_auth=False,
sshproxy_key_filename=None, sshproxy_pkey=None, sshproxy_look_for_keys=False, sshproxy_passphrase='', sshproxy_allow_agent=False,
sshproxy_tunnel=None,
add_to_known_hosts=None,
forward_sshagent=None,
unique_hostkey_aliases=None,
force_password_auth=False,
session_instance=None,
):
"""\
Connect to an X2Go server and authenticate to it. This method is directly
inherited from the ``paramiko.SSHClient`` class. The features of the Paramiko
SSH client connect method are recited here. The parameters ``add_to_known_hosts``,
``force_password_auth``, ``session_instance`` and all SSH proxy related parameters
have been added as X2Go specific parameters
The server's host key is checked against the system host keys
(see ``load_system_host_keys``) and any local host keys (``load_host_keys``).
If the server's hostname is not found in either set of host keys, the missing host
key policy is used (see ``set_missing_host_key_policy``). The default policy is
to reject the key and raise an ``SSHException``.
Authentication is attempted in the following order of priority:
- The ``pkey`` or ``key_filename`` passed in (if any)
- Any key we can find through an SSH agent
- Any "id_rsa" or "id_dsa" key discoverable in ``~/.ssh/``
- Plain username/password auth, if a password was given
If a private key requires a password to unlock it, and a password is
passed in, that password will be used to attempt to unlock the key.
:param hostname: the server to connect to
:type hostname: ``str``
:param port: the server port to connect to (Default value = 22)
:type port: ``int``
:param username: the username to authenticate as (defaults to the
current local username)
:type username: ``str``
:param password: a password to use for authentication or for unlocking
a private key (Default value = None)
:type password: ``str``
:param passphrase: a passphrase to use for unlocking
a private key in case the password is already needed for two-factor
authentication (Default value = None)
:type passphrase: ``str``
:param key_filename: the filename, or list of filenames, of optional
private key(s) to try for authentication (Default value = None)
:type key_filename: ``str`` or list(str)
:param pkey: an optional private key to use for authentication (Default value = None)
:type pkey: ``PKey``
:param forward_sshagent: forward SSH agent authentication requests to the X2Go client-side
(will update the class property of the same name) (Default value = None)
:type forward_sshagent: ``bool``
:param unique_hostkey_aliases: update the unique_hostkey_aliases class property (Default value = None)
:type unique_hostkey_aliases: ``bool``
:param timeout: an optional timeout (in seconds) for the TCP connect (Default value = None)
:type timeout: float
:param look_for_keys: set to ``True`` to enable searching for discoverable
private key files in ``~/.ssh/`` (Default value = False)
:type look_for_keys: ``bool``
:param allow_agent: set to ``True`` to enable connecting to a local SSH agent
for acquiring authentication information (Default value = False)
:type allow_agent: ``bool``
:param add_to_known_hosts: non-paramiko option, if ``True`` paramiko.AutoAddPolicy()
is used as missing-host-key-policy. If set to ``False`` paramiko.RejectPolicy()
is used (Default value = None)
:type add_to_known_hosts: ``bool``
:param force_password_auth: non-paramiko option, disable pub/priv key authentication
completely, even if the ``pkey`` or the ``key_filename`` parameter is given (Default value = False)
:type force_password_auth: ``bool``
:param session_instance: an instance :class:`x2go.session.X2GoSession` using this :class:`x2go.backends.control.plain.X2GoControlSession`
instance. (Default value = None)
:type session_instance: ``obj``
:param use_sshproxy: connect through an SSH proxy (Default value = False)
:type use_sshproxy: ``True`` if an SSH proxy is to be used for tunneling the connection
:param sshproxy_host: hostname of the SSH proxy server (Default value = None)
:type sshproxy_host: ``str``
:param sshproxy_port: port of the SSH proxy server (Default value = 22)
:type sshproxy_port: ``int``
:param sshproxy_user: username that we use for authenticating against ``<sshproxy_host>`` (Default value = None)
:type sshproxy_user: ``str``
:param sshproxy_password: a password to use for SSH proxy authentication or for unlocking
a private key (Default value = None)
:type sshproxy_password: ``str``
:param sshproxy_passphrase: a passphrase to use for unlocking
a private key needed for the SSH proxy host in case the sshproxy_password is already needed for
two-factor authentication (Default value = '')
:type sshproxy_passphrase: ``str``
:param sshproxy_force_password_auth: enforce using a given ``sshproxy_password`` even if a key(file) is given (Default value = False)
:type sshproxy_force_password_auth: ``bool``
:param sshproxy_key_filename: local file location of the private key file (Default value = None)
:type sshproxy_key_filename: ``str``
:param sshproxy_pkey: an optional private key to use for SSH proxy authentication (Default value = None)
:type sshproxy_pkey: ``PKey``
:param sshproxy_look_for_keys: set to ``True`` to enable connecting to a local SSH agent
for acquiring authentication information (for SSH proxy authentication) (Default value = False)
:type sshproxy_look_for_keys: ``bool``
:param sshproxy_allow_agent: set to ``True`` to enable connecting to a local SSH agent
for acquiring authentication information (for SSH proxy authentication) (Default value = False)
:type sshproxy_allow_agent: ``bool``
:param sshproxy_tunnel: the SSH proxy tunneling parameters, format is: <local-address>:<local-port>:<remote-address>:<remote-port> (Default value = None)
:type sshproxy_tunnel: ``str``
:returns: ``True`` if an authenticated SSH transport could be retrieved by this method
:rtype: ``bool``
:raises BadHostKeyException: if the server's host key could not be
verified
:raises AuthenticationException: if authentication failed
:raises SSHException: if there was any other error connecting or
establishing an SSH session
:raises socket.error: if a socket error occurred while connecting
:raises X2GoSSHProxyException: any SSH proxy exception is passed through while establishing the SSH proxy connection and tunneling setup
:raises X2GoSSHAuthenticationException: any SSH proxy authentication exception is passed through while establishing the SSH proxy connection and tunneling setup
:raises X2GoRemoteHomeException: if the remote home directory does not exist or is not accessible
:raises X2GoControlSessionException: if the remote peer has died unexpectedly
"""
_fake_hostname = None
if hostname and type(hostname) not in (str, bytes):
hostname = [hostname]
if hostname and type(hostname) is list:
hostname = random.choice(hostname)
if not username:
self.logger('no username specified, cannot connect without username', loglevel=log.loglevel_ERROR)
raise paramiko.AuthenticationException('no username specified, cannot connect without username')
if type(password) not in (bytes, str):
password = ''
if type(sshproxy_password) not in (bytes, str):
sshproxy_password = ''
if unique_hostkey_aliases is None:
unique_hostkey_aliases = self.unique_hostkey_aliases
# prep the fake hostname with the real hostname, so we trigger the corresponding code path in
# x2go.checkhosts and either of its missing host key policies
if unique_hostkey_aliases:
if port != 22: _fake_hostname = "[%s]:%s" % (hostname, port)
else: _fake_hostname = hostname
if add_to_known_hosts is None:
add_to_known_hosts = self.add_to_known_hosts
if forward_sshagent is None:
forward_sshagent = self.forward_sshagent
if look_for_keys:
key_filename = None
pkey = None
_twofactorauth = False
if password and (passphrase is None) and not force_password_auth: passphrase = password
if use_sshproxy and sshproxy_host and sshproxy_user:
try:
if not sshproxy_tunnel:
sshproxy_tunnel = "localhost:44444:%s:%s" % (hostname, port)
self.sshproxy_session = sshproxy.X2GoSSHProxy(known_hosts=self.known_hosts,
add_to_known_hosts=add_to_known_hosts,
sshproxy_host=sshproxy_host,
sshproxy_port=sshproxy_port,
sshproxy_user=sshproxy_user,
sshproxy_password=sshproxy_password,
sshproxy_passphrase=sshproxy_passphrase,
sshproxy_force_password_auth=sshproxy_force_password_auth,
sshproxy_key_filename=sshproxy_key_filename,
sshproxy_pkey=sshproxy_pkey,
sshproxy_look_for_keys=sshproxy_look_for_keys,
sshproxy_allow_agent=sshproxy_allow_agent,
sshproxy_tunnel=sshproxy_tunnel,
session_instance=session_instance,
logger=self.logger,
)
hostname = self.sshproxy_session.get_local_proxy_host()
port = self.sshproxy_session.get_local_proxy_port()
_fake_hostname = self.sshproxy_session.get_remote_host()
_fake_port = self.sshproxy_session.get_remote_port()
if _fake_port != 22:
_fake_hostname = "[%s]:%s" % (_fake_hostname, _fake_port)
except:
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
self.sshproxy_session = None
raise
if self.sshproxy_session is not None:
self.sshproxy_session.start()
# divert port to sshproxy_session's local forwarding port (it might have changed due to
# SSH connection errors
gevent.sleep(.1)
port = self.sshproxy_session.get_local_proxy_port()
if not add_to_known_hosts and session_instance:
self.set_missing_host_key_policy(checkhosts.X2GoInteractiveAddPolicy(caller=self, session_instance=session_instance, fake_hostname=_fake_hostname))
if add_to_known_hosts:
self.set_missing_host_key_policy(checkhosts.X2GoAutoAddPolicy(caller=self, session_instance=session_instance, fake_hostname=_fake_hostname))
# trailing whitespace tolerance in hostname
hostname = hostname.strip()
self.logger('connecting to [%s]:%s' % (hostname, port), loglevel=log.loglevel_NOTICE)
self.load_session_host_keys()
_hostname = hostname
# enforce IPv4 for localhost address
if _hostname in ('localhost', 'localhost.localdomain'):
_hostname = '127.0.0.1'
# update self.forward_sshagent via connect method parameter
if forward_sshagent is not None:
self.forward_sshagent = forward_sshagent
if timeout and self.low_latency:
timeout = timeout * 2
if key_filename and "~" in key_filename:
key_filename = os.path.expanduser(key_filename)
if key_filename or pkey or look_for_keys or allow_agent or (password and force_password_auth):
try:
if password and force_password_auth:
self.logger('trying password based SSH authentication with server', loglevel=log.loglevel_DEBUG)
paramiko.SSHClient.connect(self, _hostname, port=port, username=username, password=password, pkey=None,
key_filename=None, timeout=timeout, allow_agent=False,
look_for_keys=False)
elif (key_filename and os.path.exists(os.path.normpath(key_filename))) or pkey:
self.logger('trying SSH pub/priv key authentication with server', loglevel=log.loglevel_DEBUG)
paramiko.SSHClient.connect(self, _hostname, port=port, username=username, pkey=pkey, password=passphrase,
key_filename=key_filename, timeout=timeout, allow_agent=False,
look_for_keys=False)
else:
self.logger('trying SSH key discovery or agent authentication with server', loglevel=log.loglevel_DEBUG)
paramiko.SSHClient.connect(self, _hostname, port=port, username=username, pkey=None, password=passphrase,
key_filename=None, timeout=timeout, allow_agent=allow_agent,
look_for_keys=look_for_keys)
except (paramiko.PasswordRequiredException, paramiko.SSHException) as e:
self.close()
if type(e) == paramiko.SSHException and str(e).startswith('Two-factor authentication requires a password'):
self.logger('X2Go Server requests two-factor authentication', loglevel=log.loglevel_NOTICE)
_twofactorauth = True
if passphrase is not None:
self.logger('unlock SSH private key file with provided password', loglevel=log.loglevel_INFO)
try:
if not password: password = None
if (key_filename and os.path.exists(os.path.normpath(key_filename))) or pkey:
self.logger('re-trying SSH pub/priv key authentication with server', loglevel=log.loglevel_DEBUG)
try:
paramiko.SSHClient.connect(self, _hostname, port=port, username=username, password=password, passphrase=passphrase, pkey=pkey,
key_filename=key_filename, timeout=timeout, allow_agent=False,
look_for_keys=False)
except TypeError:
if _twofactorauth and password and passphrase and password != passphrase:
self.logger('your version of Paramiko/SSH does not support authentication workflows which require SSH key decryption in combination with two-factor authentication', loglevel=log.loglevel_WARN)
paramiko.SSHClient.connect(self, _hostname, port=port, username=username, password=password, pkey=pkey,
key_filename=key_filename, timeout=timeout, allow_agent=False,
look_for_keys=False)
else:
self.logger('re-trying SSH key discovery now with passphrase for unlocking the key(s)', loglevel=log.loglevel_DEBUG)
try:
paramiko.SSHClient.connect(self, _hostname, port=port, username=username, password=password, passphrase=passphrase, pkey=None,
key_filename=None, timeout=timeout, allow_agent=allow_agent,
look_for_keys=look_for_keys)
except TypeError:
if _twofactorauth and password and passphrase and password != passphrase:
self.logger('your version of Paramiko/SSH does not support authentication workflows which require SSH key decryption in combination with two-factor authentication', loglevel=log.loglevel_WARN)
paramiko.SSHClient.connect(self, _hostname, port=port, username=username, password=password, pkey=None,
key_filename=None, timeout=timeout, allow_agent=allow_agent,
look_for_keys=look_for_keys)
except paramiko.AuthenticationException as auth_e:
# the provided password cannot be used to unlock any private SSH key file (i.e. wrong password)
raise paramiko.AuthenticationException(str(auth_e))
except paramiko.SSHException as auth_e:
if str(auth_e) == 'No authentication methods available':
raise paramiko.AuthenticationException('Interactive password authentication required!')
else:
self.close()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise auth_e
else:
self.close()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise e
except paramiko.AuthenticationException as e:
self.close()
if password:
self.logger('next auth mechanism we\'ll try is password authentication', loglevel=log.loglevel_DEBUG)
try:
paramiko.SSHClient.connect(self, _hostname, port=port, username=username, password=password,
key_filename=None, pkey=None, timeout=timeout, allow_agent=False, look_for_keys=False)
except:
self.close()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise
else:
self.close()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise e
except paramiko.SSHException as e:
if str(e) == 'No authentication methods available':
raise paramiko.AuthenticationException('Interactive password authentication required!')
else:
self.close()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise e
except:
self.close()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise
# if there is no private key (and no agent auth), we will use the given password, if any
else:
# create a random password if password is empty to trigger host key validity check
if not password:
password = "".join([random.choice(string.letters+string.digits) for x in range(1, 20)])
self.logger('performing SSH password authentication with server', loglevel=log.loglevel_DEBUG)
try:
paramiko.SSHClient.connect(self, _hostname, port=port, username=username, password=password,
timeout=timeout, allow_agent=False, look_for_keys=False)
except:
self.close()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise
self.set_missing_host_key_policy(paramiko.RejectPolicy())
self.hostname = hostname
self.port = port
# preparing reverse tunnels
ssh_transport = self.get_transport()
ssh_transport.reverse_tunnels = {}
# mark Paramiko/SSH transport as X2GoControlSession
ssh_transport._x2go_session_marker = True
try:
self._session_password = base64.b64encode(password)
except TypeError:
self._session_password = None
if ssh_transport is not None:
# since Paramiko 1.7.7.1 there is compression available, let's use it if present...
if x2go._paramiko.PARAMIKO_FEATURE['use-compression']:
ssh_transport.use_compression(compress=False)
# enable keep alive callbacks
ssh_transport.set_keepalive(5)
self.session_died = False
self.query_server_features(force=True)
if self.forward_sshagent:
if x2go._paramiko.PARAMIKO_FEATURE['forward-ssh-agent']:
try:
self.agent_chan = ssh_transport.open_session()
self.agent_handler = paramiko.agent.AgentRequestHandler(self.agent_chan)
self.logger('Requesting SSH agent forwarding for control session of connected session profile %s' % self.profile_name, loglevel=log.loglevel_NOTICE)
except EOFError as e:
# if we come across an EOFError here, we must assume the session is dead...
self.session_died = True
raise x2go_exceptions.X2GoControlSessionException('The SSH connection was dropped while setting up SSH agent forwarding socket.')
else:
self.logger('SSH agent forwarding is not available in the Paramiko version used with this instance of Python X2Go', loglevel=log.loglevel_WARN)
else:
self.close()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
self._remote_home = None
if not self.home_exists():
self.close()
if self.sshproxy_session:
self.sshproxy_session.stop_thread()
raise x2go_exceptions.X2GoRemoteHomeException('remote home directory does not exist')
return (self.get_transport() is not None)
[docs]
def dissociate(self, terminal_session):
"""\
Drop an associated terminal session.
:param terminal_session: the terminal session object to remove from the list of associated terminals
:type terminal_session: ``X2GoTerminalSession*``
"""
for t_name in list(self.associated_terminals.keys()):
if self.associated_terminals[t_name] == terminal_session:
del self.associated_terminals[t_name]
if t_name in self.terminated_terminals:
del self.terminated_terminals[t_name]
[docs]
def disconnect(self):
"""\
Disconnect this control session from the remote server.
:returns: report success or failure after having disconnected
:rtype: ``bool``
"""
if self.associated_terminals:
t_names = list(self.associated_terminals.keys())
for t_obj in list(self.associated_terminals.values()):
try:
if not self.session_died:
t_obj.suspend()
except x2go_exceptions.X2GoTerminalSessionException:
pass
except x2go_exceptions.X2GoControlSessionException:
self.session_died
t_obj.__del__()
for t_name in t_names:
try:
del self.associated_terminals[t_name]
except KeyError:
pass
self._remote_home = None
self._remote_group = {}
self._session_auth_rsakey = None
# in any case, release out internal transport lock
if self._transport_lock.locked():
self._transport_lock.release()
# close SSH agent auth forwarding objects
if self.agent_handler is not None:
self.agent_handler.close()
if self.agent_chan is not None:
try:
self.agent_chan.close()
except EOFError:
pass
retval = False
try:
if self.get_transport() is not None:
retval = self.get_transport().is_active()
try:
self.close()
except IOError:
pass
except AttributeError:
# if the Paramiko _transport object has not yet been initialized, ignore it
# but state that this method call did not close the SSH client, but was already closed
pass
# take down sshproxy_session no matter what happened to the control session itself
if self.sshproxy_session is not None:
self.sshproxy_session.stop_thread()
return retval
[docs]
def home_exists(self):
"""\
Test if the remote home directory exists.
:returns: ``True`` if the home directory exists, ``False`` otherwise
:rtype: ``bool``
"""
(stdin, stdout, stderr) = self._x2go_exec_command('stat -tL "%s"' % self._x2go_remote_home, loglevel=log.loglevel_DEBUG)
_stdout = stdout.read()
if _stdout:
return True
return False
[docs]
def is_alive(self):
"""\
Test if the connection to the remote X2Go server is still alive.
:returns: ``True`` if the connection is still alive, ``False`` otherwise
:rtype: ``bool``
"""
try:
if self._x2go_exec_command('echo', loglevel=log.loglevel_DEBUG):
return True
except x2go_exceptions.X2GoControlSessionException:
self.session_died = True
self.disconnect()
return False
[docs]
def has_session_died(self):
"""\
Test if the connection to the remote X2Go server died on the way.
:returns: ``True`` if the connection has died, ``False`` otherwise
:rtype: ``bool``
"""
return self.session_died
[docs]
def get_published_applications(self, lang=None, refresh=False, raw=False, very_raw=False, max_no_submenus=defaults.PUBAPP_MAX_NO_SUBMENUS):
"""\
Retrieve the menu tree of published applications from the remote X2Go server.
The ``raw`` option lets this method return a ``list`` of ``dict`` elements. Each ``dict`` elements has a
``desktop`` key containing a shortened version of the text output of a .desktop file and an ``icon`` key
which contains the desktop base64-encoded icon data.
The {very_raw} lets this method return the output of the ``x2gogetapps`` script as is.
:param lang: locale/language identifier (Default value = None)
:type lang: ``str``
:param refresh: force reload of the menu tree from X2Go server (Default value = False)
:type refresh: ``bool``
:param raw: retrieve a raw output of the server list of published applications (Default value = False)
:type raw: ``bool``
:param very_raw: retrieve a very raw output of the server list of published applications (Default value = False)
:type very_raw: ``bool``
:param max_no_submenus: Number of applications before applications are put into XDG category submenus
(Default value = defaults.PUBAPP_MAX_NO_SUBMENUS)
:type max_no_submenus: ``int``
:returns: an i18n capable menu tree packed as a Python dictionary
:rtype: ``list``
"""
self._already_querying_published_applications.acquire()
if defaults.X2GOCLIENT_OS != 'Windows' and lang is None:
lang = locale.getdefaultlocale()[0]
elif lang is None:
lang = 'en'
if 'X2GO_PUBLISHED_APPLICATIONS' in self.get_server_features():
if self._published_applications_menu is {} or \
lang not in self._published_applications_menu or \
raw or very_raw or refresh or \
(self.published_applications_no_submenus != max_no_submenus):
self.published_applications_no_submenus = max_no_submenus
### STAGE 1: retrieve menu from server
self.logger('querying server (%s) for list of published applications' % self.profile_name, loglevel=log.loglevel_NOTICE)
(stdin, stdout, stderr) = self._x2go_exec_command('which x2gogetapps >/dev/null && x2gogetapps')
_raw_output = stdout.read()
if very_raw:
self.logger('published applications query for %s finished, return very raw output' % self.profile_name, loglevel=log.loglevel_NOTICE)
self._already_querying_published_applications.release()
return _raw_output
### STAGE 2: dissect the text file retrieved from server, cut into single menu elements
_raw_menu_items = _raw_output.split('</desktop>\n')
_raw_menu_items = [ i.replace('<desktop>\n', '') for i in _raw_menu_items ]
_menu = []
for _raw_menu_item in _raw_menu_items:
if '<icon>\n' in _raw_menu_item and '</icon>' in _raw_menu_item:
_menu_item = _raw_menu_item.split('<icon>\n')[0] + _raw_menu_item.split('</icon>\n')[1]
_icon_base64 = _raw_menu_item.split('<icon>\n')[1].split('</icon>\n')[0]
else:
_menu_item = _raw_menu_item
_icon_base64 = None
if _menu_item:
_menu.append({ 'desktop': _menu_item, 'icon': _icon_base64, })
_menu_item = None
_icon_base64 = None
if raw:
self.logger('published applications query for %s finished, returning raw output' % self.profile_name, loglevel=log.loglevel_NOTICE)
self._already_querying_published_applications.release()
return _menu
if len(_menu) > max_no_submenus >= 0:
_render_submenus = True
else:
_render_submenus = False
# STAGE 3: create menu structure in a Python dictionary
_category_map = {
lang: {
'Multimedia': [],
'Development': [],
'Education': [],
'Games': [],
'Graphics': [],
'Internet': [],
'Office': [],
'System': [],
'Utilities': [],
'Other Applications': [],
'TOP': [],
}
}
_empty_menus = list(_category_map[lang].keys())
for item in _menu:
_menu_entry_name = ''
_menu_entry_fallback_name = ''
_menu_entry_comment = ''
_menu_entry_fallback_comment = ''
_menu_entry_exec = ''
_menu_entry_cat = ''
_menu_entry_shell = False
lang_regio = lang
lang_only = lang_regio.split('_')[0]
for line in item['desktop'].split('\n'):
if re.match('^Name\[%s\]=.*' % lang_regio, line) or re.match('Name\[%s\]=.*' % lang_only, line):
_menu_entry_name = line.split("=")[1].strip()
elif re.match('^Name=.*', line):
_menu_entry_fallback_name = line.split("=")[1].strip()
elif re.match('^Comment\[%s\]=.*' % lang_regio, line) or re.match('Comment\[%s\]=.*' % lang_only, line):
_menu_entry_comment = line.split("=")[1].strip()
elif re.match('^Comment=.*', line):
_menu_entry_fallback_comment = line.split("=")[1].strip()
elif re.match('^Exec=.*', line):
_menu_entry_exec = line.split("=")[1].strip()
elif re.match('^Terminal=.*(t|T)(r|R)(u|U)(e|E).*', line):
_menu_entry_shell = True
elif re.match('^Categories=.*', line):
if 'X2Go-Top' in line:
_menu_entry_cat = 'TOP'
elif 'Audio' in line or 'Video' in line:
_menu_entry_cat = 'Multimedia'
elif 'Development' in line:
_menu_entry_cat = 'Development'
elif 'Education' in line:
_menu_entry_cat = 'Education'
elif 'Game' in line:
_menu_entry_cat = 'Games'
elif 'Graphics' in line:
_menu_entry_cat = 'Graphics'
elif 'Network' in line:
_menu_entry_cat = 'Internet'
elif 'Office' in line:
_menu_entry_cat = 'Office'
elif 'Settings' in line:
continue
elif 'System' in line:
_menu_entry_cat = 'System'
elif 'Utility' in line:
_menu_entry_cat = 'Utilities'
else:
_menu_entry_cat = 'Other Applications'
if not _menu_entry_exec:
continue
else:
# FIXME: strip off any noted options (%f, %F, %u, %U, ...), this can be more intelligent
_menu_entry_exec = _menu_entry_exec.replace('%f', '').replace('%F','').replace('%u','').replace('%U','')
if _menu_entry_shell:
_menu_entry_exec = "x-terminal-emulator -e '%s'" % _menu_entry_exec
if not _menu_entry_cat:
_menu_entry_cat = 'Other Applications'
if not _render_submenus:
_menu_entry_cat = 'TOP'
if _menu_entry_cat in _empty_menus:
_empty_menus.remove(_menu_entry_cat)
if not _menu_entry_name: _menu_entry_name = _menu_entry_fallback_name
if not _menu_entry_comment: _menu_entry_comment = _menu_entry_fallback_comment
if not _menu_entry_comment: _menu_entry_comment = _menu_entry_name
_menu_entry_icon = item['icon']
_category_map[lang][_menu_entry_cat].append(
{
'name': _menu_entry_name,
'comment': _menu_entry_comment,
'exec': _menu_entry_exec,
'icon': _menu_entry_icon,
}
)
for _cat in _empty_menus:
del _category_map[lang][_cat]
for _cat in list(_category_map[lang].keys()):
_sorted = sorted(_category_map[lang][_cat], key=lambda k: k['name'])
_category_map[lang][_cat] = _sorted
self._published_applications_menu.update(_category_map)
self.logger('published applications query for %s finished, return menu tree' % self.profile_name, loglevel=log.loglevel_NOTICE)
else:
# FIXME: ignoring the absence of the published applications feature for now, handle it appropriately later
pass
self._already_querying_published_applications.release()
return self._published_applications_menu
[docs]
def start(self, **kwargs):
"""\
Start a new X2Go session.
The :func:`X2GoControlSession.start() <x2go.backends.control.X2GoControlSession.start()>` method accepts any parameter
that can be passed to any of the ``X2GoTerminalSession`` backend class
constructors.
:param kwargs: parameters that get passed through to the control session's
:func:`resume()` method, only the ``session_name`` parameter will get removed
before pass-through
:type kwargs: ``dict``
:returns: return: return value of the cascaded :func:`resume()` method, denoting the success or failure
of the session startup
:rtype: ``bool``
"""
if 'session_name' in list(kwargs.keys()):
del kwargs['session_name']
return self.resume(**kwargs)
[docs]
def resume(self, session_name=None, session_instance=None, session_list=None, **kwargs):
"""\
Resume a running/suspended X2Go session.
The :func:`X2GoControlSession.resume() <x2go.backends.control.X2GoControlSession.resume()>` method accepts any parameter
that can be passed to any of the ``X2GoTerminalSession*`` backend class constructors.
:param session_name: the X2Go session name (Default value = None)
:type session_name: ``str``
:param session_instance: a Python X2Go session instance (Default value = None)
:type session_instance: :class:`x2go.session.X2GoSession`
:param session_list: Default value = None)
:param kwargs: catch any non-defined param in kwargs
:type kwargs: ``dict``
:returns: True if the session could be successfully resumed
:rtype: ``bool``
:raises X2GoUserException: if the remote user is not allowed to launch/resume X2Go sessions.
"""
if self.get_transport() is not None:
if not self.is_x2gouser(self.get_transport().get_username()):
raise x2go_exceptions.X2GoUserException('remote user %s is not allowed to run X2Go commands' % self.get_transport().get_username())
session_info = None
try:
if session_name is not None:
if session_list:
session_info = session_list[session_name]
else:
session_info = self.list_sessions()[session_name]
except KeyError:
_success = False
_terminal = self._terminal_backend(self,
profile_name=self.profile_name,
session_info=session_info,
info_backend=self._info_backend,
list_backend=self._list_backend,
proxy_backend=self._proxy_backend,
client_rootdir=self.client_rootdir,
session_instance=session_instance,
sessions_rootdir=self.sessions_rootdir,
**kwargs)
_success = False
try:
if session_name is not None:
_success = _terminal.resume()
else:
_success = _terminal.start()
except x2go_exceptions.X2GoTerminalSessionException:
_success = False
if _success:
while not _terminal.ok():
gevent.sleep(.2)
if _terminal.ok():
self.associated_terminals[_terminal.get_session_name()] = _terminal
self.get_transport().reverse_tunnels[_terminal.get_session_name()] = {
'sshfs': (0, None),
'snd': (0, None),
}
return _terminal or None
return None
[docs]
def share_desktop(self, desktop=None, user=None, display=None, share_mode=0, **kwargs):
"""\
Share another already running desktop session. Desktop sharing can be run
in two different modes: view-only and full-access mode.
:param desktop: desktop ID of a sharable desktop in format ``<user>@<display>`` (Default value = None)
:type desktop: ``str``
:param user: user name and display number can be given separately, here give the
name of the user who wants to share a session with you (Default value = None)
:type user: ``str``
:param display: user name and display number can be given separately, here give the
number of the display that a user allows you to be shared with (Default value = None)
:type display: ``str``
:param share_mode: desktop sharing mode, 0 stands for VIEW-ONLY, 1 for FULL-ACCESS mode (Default value = 0)
:type share_mode: ``int``
:param kwargs: catch any non-defined param in kwargs
:type kwargs: ``dict``
:returns: True if the session could be successfully shared
:rtype: ``bool``
:raises X2GoDesktopSharingException: if ``username`` and ``dislpay`` do not relate to a
sharable desktop session
"""
if desktop:
user = desktop.split('@')[0]
display = desktop.split('@')[1]
if not (user and display):
raise x2go_exceptions.X2GoDesktopSharingException('Need user name and display number of shared desktop.')
cmd = '%sXSHAD%sXSHAD%s' % (share_mode, user, display)
kwargs['cmd'] = cmd
kwargs['session_type'] = 'shared'
return self.start(**kwargs)
[docs]
def list_desktops(self, raw=False, maxwait=20):
"""\
List all desktop-like sessions of current user (or of users that have
granted desktop sharing) on the connected server.
:param raw: if ``True``, the raw output of the server-side X2Go command
``x2golistdesktops`` is returned. (Default value = False)
:type raw: ``bool``
:param maxwait: time in secs to wait for server query to reply (Default value = 20)
:type maxwait: ``int``
:returns: a list of X2Go desktops available for sharing
:rtype: ``list``
:raises X2GoTimeOutException: on command execution timeouts, with the server-side ``x2golistdesktops``
command this can sometimes happen. Make sure you ignore these time-outs and to try again
"""
if raw:
(stdin, stdout, stderr) = self._x2go_exec_command("export HOSTNAME && x2golistdesktops")
return stdout.read(), stderr.read()
else:
# this _success loop will catch errors in case the x2golistsessions output is corrupt
# this should not be needed and is a workaround for the current X2Go server implementation
if self.low_latency:
maxwait = maxwait * 2
timeout = gevent.Timeout(maxwait)
timeout.start()
try:
(stdin, stdout, stderr) = self._x2go_exec_command("export HOSTNAME && x2golistdesktops")
_stdout = stdout.read()
if sys.version_info[0] >= 3:
_stdout = _stdout.decode()
_listdesktops = _stdout.split('\n')
except gevent.timeout.Timeout:
# if we do not get a reply here after <maxwait> seconds we will raise a time out, we have to
# make sure that we catch this at places where we want to ignore timeouts (e.g. in the
# desktop list cache)
raise x2go_exceptions.X2GoTimeOutException('x2golistdesktop command timed out')
finally:
timeout.cancel()
return _listdesktops
[docs]
def list_mounts(self, session_name, raw=False, maxwait=20):
"""\
List all mounts for a given session of the current user on the connected server.
:param session_name: name of a session to query a list of mounts for
:type session_name: ``str``
:param raw: if ``True``, the raw output of the server-side X2Go command
``x2golistmounts`` is returned. (Default value = False)
:type raw: ``bool``
:param maxwait: stop processing ``x2golistmounts`` after ``<maxwait>`` seconds (Default value = 20)
:type maxwait: ``int``
:returns: a list of client-side mounts for X2Go session ``<session_name>`` on the server
:rtype: ``list``
:raises X2GoTimeOutException: on command execution timeouts, queries with the server-side
``x2golistmounts`` query should normally be processed quickly, a time-out may hint that the
control session has lost its connection to the X2Go server
"""
if raw:
(stdin, stdout, stderr) = self._x2go_exec_command("export HOSTNAME && x2golistmounts %s" % session_name)
return stdout.read(), stderr.read()
else:
if self.low_latency:
maxwait = maxwait * 2
# this _success loop will catch errors in case the x2golistmounts output is corrupt
timeout = gevent.Timeout(maxwait)
timeout.start()
try:
(stdin, stdout, stderr) = self._x2go_exec_command("export HOSTNAME && x2golistmounts %s" % session_name)
_stdout = stdout.read()
if sys.version_info[0] >= 3:
_stdout = _stdout.decode()
_listmounts = {session_name: [ line for line in _stdout.split('\n') if line ] }
except gevent.timeout.Timeout:
# if we do not get a reply here after <maxwait> seconds we will raise a time out, we have to
# make sure that we catch this at places where we want to ignore timeouts
raise x2go_exceptions.X2GoTimeOutException('x2golistmounts command timed out')
finally:
timeout.cancel()
return _listmounts
[docs]
def list_sessions(self, raw=False):
"""\
List all sessions of current user on the connected server.
:param raw: if ``True``, the raw output of the server-side X2Go command
``x2golistsessions`` is returned. (Default value = False)
:type raw: ``bool``
:returns: normally an instance of a ``X2GoServerSessionList*`` backend is returned. However,
if the raw argument is set, the plain text output of the server-side ``x2golistsessions``
command is returned
:rtype: ``X2GoServerSessionList`` instance or str
:raises X2GoControlSessionException: on command execution timeouts, if this happens the control session will
be interpreted as disconnected due to connection loss
"""
if raw:
if 'X2GO_LIST_SHADOWSESSIONS' in self._x2go_server_features:
(stdin, stdout, stderr) = self._x2go_exec_command("export HOSTNAME && { x2golistsessions; x2golistshadowsessions; }")
else:
(stdin, stdout, stderr) = self._x2go_exec_command("export HOSTNAME && x2golistsessions")
return stdout.read(), stderr.read()
else:
# this _success loop will catch errors in case the x2golistsessions output is corrupt
# this should not be needed and is a workaround for the current X2Go server implementation
_listsessions = {}
_success = False
_count = 0
_maxwait = 20
# we will try this 20 times before giving up... we might simply catch the x2golistsessions
# output in the middle of creating a session in the database...
while not _success and _count < _maxwait:
_count += 1
try:
if 'X2GO_LIST_SHADOWSESSIONS' in self._x2go_server_features:
(stdin, stdout, stderr) = self._x2go_exec_command("export HOSTNAME && { x2golistsessions; x2golistshadowsessions; }")
else:
(stdin, stdout, stderr) = self._x2go_exec_command("export HOSTNAME && x2golistsessions")
_stdout = stdout.read()
if sys.version_info[0] >= 3:
_stdout = _stdout.decode()
_l = self._list_backend(_stdout, info_backend=self._info_backend)
_listsessions = _l.get_sessions()
_success = True
except KeyError:
gevent.sleep(1)
except IndexError:
gevent.sleep(1)
except ValueError:
gevent.sleep(1)
if _count >= _maxwait:
self.session_died = True
self.disconnect()
raise x2go_exceptions.X2GoControlSessionException('x2golistsessions command failed after we have tried 20 times')
# update internal variables when list_sessions() is called
if _success and not self.session_died:
for _session_name, _terminal in list(self.associated_terminals.items()):
if _session_name in list(_listsessions.keys()):
# update the whole session_info object within the terminal session
if hasattr(self.associated_terminals[_session_name], 'session_info') and not self.associated_terminals[_session_name].is_session_info_protected():
self.associated_terminals[_session_name].session_info.update(_listsessions[_session_name])
else:
self.associated_terminals[_session_name].__del__()
try: del self.associated_terminals[_session_name]
except KeyError: pass
self.terminated_terminals.append(_session_name)
if _terminal.is_suspended():
self.associated_terminals[_session_name].__del__()
try: del self.associated_terminals[_session_name]
except KeyError: pass
return _listsessions
[docs]
def clean_sessions(self, destroy_terminals=True, published_applications=False):
"""\
Find X2Go terminals that have previously been started by the
connected user on the remote X2Go server and terminate them.
:param destroy_terminals: destroy the terminal session instances after cleanup (Default value = True)
:type destroy_terminals: ``bool``
:param published_applications: also clean up published applications providing sessions (Default value = False)
:type published_applications: ``bool``
"""
session_list = self.list_sessions()
if published_applications:
session_names = list(session_list.keys())
else:
session_names = [ _sn for _sn in list(session_list.keys()) if not session_list[_sn].is_published_applications_provider() ]
for session_name in session_names:
if session_name in self.associated_terminals:
self.associated_terminals[session_name].terminate()
if destroy_terminals:
if self.associated_terminals[session_name] is not None:
self.associated_terminals[session_name].__del__()
try: del self.associated_terminals[session_name]
except KeyError: pass
else:
self.terminate(session_name=session_name)
[docs]
def is_connected(self):
"""\
Returns ``True`` if this control session is connected to the remote server (that
is: if it has a valid Paramiko/SSH transport object).
:returns: X2Go session connected?
:rtype: ``bool``
"""
return self.get_transport() is not None and self.get_transport().is_authenticated()
[docs]
def is_running(self, session_name):
"""\
Returns ``True`` if the given X2Go session is in running state,
``False`` else.
:param session_name: X2Go name of the session to be queried
:type session_name: ``str``
:returns: X2Go session running? If ``<session_name>`` is not listable by the :func:`list_sessions()` method then ``None`` is returned
:rtype: ``bool`` or ``None``
"""
session_infos = self.list_sessions()
if session_name in list(session_infos.keys()):
return session_infos[session_name].is_running()
return None
[docs]
def is_suspended(self, session_name):
"""\
Returns ``True`` if the given X2Go session is in suspended state,
``False`` else.
:param session_name: X2Go name of the session to be queried
:type session_name: ``str``
:returns: X2Go session suspended? If ``<session_name>`` is not listable by the :func:`list_sessions()` method then ``None`` is returned
:rtype: ``bool`` or ``None``
"""
session_infos = self.list_sessions()
if session_name in list(session_infos.keys()):
return session_infos[session_name].is_suspended()
return None
[docs]
def has_terminated(self, session_name):
"""\
Returns ``True`` if the X2Go session with name ``<session_name>`` has been seen
by this control session and--in the meantime--has been terminated.
If ``<session_name>`` has not been seen, yet, the method will return ``None``.
:param session_name: X2Go name of the session to be queried
:type session_name: ``str``
:returns: X2Go session has terminated?
:rtype: ``bool`` or ``None``
"""
session_infos = self.list_sessions()
if session_name in self.terminated_terminals:
return True
if session_name not in list(session_infos.keys()) and session_name in list(self.associated_terminals.keys()):
# do a post-mortem tidy up
self.terminate(session_name)
return True
if self.is_suspended(session_name) or self.is_running(session_name):
return False
return None
[docs]
def suspend(self, session_name):
"""\
Suspend X2Go session with name ``<session_name>`` on the connected
server.
:param session_name: X2Go name of the session to be suspended
:type session_name: ``str``
:returns: ``True`` if the session could be successfully suspended
:rtype: ``bool``
"""
_ret = False
_session_names = [ t.get_session_name() for t in list(self.associated_terminals.values()) ]
if session_name in _session_names:
self.logger('suspending associated terminal session: %s' % session_name, loglevel=log.loglevel_DEBUG)
(stdin, stdout, stderr) = self._x2go_exec_command("x2gosuspend-session %s" % session_name, loglevel=log.loglevel_DEBUG)
stdout.read()
stderr.read()
if session_name in self.associated_terminals:
if self.associated_terminals[session_name] is not None:
self.associated_terminals[session_name].__del__()
try: del self.associated_terminals[session_name]
except KeyError: pass
_ret = True
else:
self.logger('suspending non-associated terminal session: %s' % session_name, loglevel=log.loglevel_DEBUG)
(stdin, stdout, stderr) = self._x2go_exec_command("x2gosuspend-session %s" % session_name, loglevel=log.loglevel_DEBUG)
stdout.read()
stderr.read()
_ret = True
return _ret
[docs]
def terminate(self, session_name, destroy_terminals=True):
"""\
Terminate X2Go session with name ``<session_name>`` on the connected
server.
:param session_name: X2Go name of the session to be terminated
:type session_name: ``str``
:param destroy_terminals: destroy all terminal sessions associated to this control session (Default value = True)
:type destroy_terminals: ``bool``
:returns: ``True`` if the session could be successfully terminated
:rtype: ``bool``
"""
_ret = False
if session_name in list(self.associated_terminals.keys()):
self.logger('terminating associated session: %s' % session_name, loglevel=log.loglevel_DEBUG)
(stdin, stdout, stderr) = self._x2go_exec_command("x2goterminate-session %s" % session_name, loglevel=log.loglevel_DEBUG)
stdout.read()
stderr.read()
if destroy_terminals:
if self.associated_terminals[session_name] is not None:
self.associated_terminals[session_name].__del__()
try: del self.associated_terminals[session_name]
except KeyError: pass
self.terminated_terminals.append(session_name)
_ret = True
else:
self.logger('terminating non-associated session: %s' % session_name, loglevel=log.loglevel_DEBUG)
(stdin, stdout, stderr) = self._x2go_exec_command("x2goterminate-session %s" % session_name, loglevel=log.loglevel_DEBUG)
stdout.read()
stderr.read()
_ret = True
return _ret