python-x2go 0.1.1.8-0ubuntu1 / usr / share / pyshared / x2go / utils.py

# -*- coding: utf-8 -*-

# Copyright (C) 2010-2011 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 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the
# Free Software Foundation, Inc.,
# 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.

"""\
Python X2go helper functions, constants etc.

"""
__NAME__ = 'x2goutils-pylib'

import sys
import os
import locale
import re
import types
import copy
import paramiko
import socket
import gevent
import string
import re
import subprocess

# Python X2go modules
from defaults import X2GOCLIENT_OS as _X2GOCLIENT_OS
from defaults import X2GO_SESSIONPROFILE_DEFAULTS as _X2GO_SESSIONPROFILE_DEFAULTS
from defaults import X2GO_MIMEBOX_ACTIONS as _X2GO_MIMEBOX_ACTIONS
from defaults import pack_methods_nx3

if _X2GOCLIENT_OS == 'Windows':
    import win32api

def is_in_nx3packmethods(method):

    """\
    Test if a given compression method is valid for NX3 Proxy.

    """
    return method in pack_methods_nx3


def find_session_line_in_x2golistsessions(session_name, x2go_stdout):
    """\
    Return the X2go session meta information as returned by the 
    C{x2golistsessions} server command for session C{session_name}.

    """
    sessions = stdout.read().split("\n")
    for line in sessions:
        # skip empty lines
        if not line:
            continue
        if session_name == line.split("|")[1]:
            return line
    return None


def slugify(value):
    """\
    Normalizes string, converts to lowercase, removes non-alpha characters,
    converts spaces to hyphens and replaces round brackets by pointed brackets.

    """
    import unicodedata
    value = unicodedata.normalize('NFKD', unicode(value)).encode('ascii', 'ignore')
    value = re.sub('[^\w\s-]', '', value).strip().lower()
    value = re.sub('[(]', '<', value).strip().lower()
    value = re.sub('[)]', '>', value).strip().lower()
    return value

def _genSessionProfileId():
    """\
    Generate a session profile ID as used in x2goclient's sessions config file.

    """
    import datetime
    return datetime.datetime.utcnow().strftime('%Y%m%d%H%m%S%f')


def _checkIniFileDefaults(defaults):
    """\
    Check an ini file data structure passed on by a user app or class.

    """
    if defaults is None:
        return False
    if type(defaults) is not types.DictType:
        return False
    for sub_dict in defaults.values():
        if type(sub_dict) is not types.DictType:
            return False
    return True


def _checkSessionProfileDefaults(defaults):
    """\
    Check the data structure of a default session profile passed by a user app.

    """
    if defaults is None:
        return False
    if type(defaults) is not types.DictType:
        return False
    return True


def _convert_SessionProfileOptions_2_SessionParams(_options):
    """\
    Convert session profile options as used in x2goclient's sessions file to
    Python X2go session parameters.

    """

    _params = copy.deepcopy(_options)

    # get rid of unknown session profile options
    _known_options = _X2GO_SESSIONPROFILE_DEFAULTS.keys()
    for p in _params.keys():
        if p not in _known_options:
            del _params[p]

    _rename_dict = {
            'host': 'server',
            'user': 'username',
            'soundsystem': 'snd_system',
            'sndport': 'snd_port',
            'type': 'kbtype',
            'layout': 'kblayout',
            'speed': 'link',
            'sshport': 'port',
            'useexports': 'allow_share_local_folders',
            'usemimebox': 'allow_mimebox',
            'mimeboxextensions': 'mimebox_extensions',
            'mimeboxaction': 'mimebox_action',
            'print': 'printing',
            'name': 'profile_name',
            'key': 'key_filename',
            'command': 'cmd',
            'rdpserver': 'rdp_server',
            'rdpoptions': 'rdp_options',
            'xdmcpserver': 'xdmcp_server',
            'useiconv': 'convert_encoding',
            'iconvto': 'server_encoding',
            'iconvfrom': 'client_encoding',
            'usesshproxy': 'use_sshproxy',
            'sshproxyhost': 'sshproxy_host',
            'sshproxyuser': 'sshproxy_user',
            'sshproxykeyfile': 'sshproxy_key_filename',
            'sshproxytunnel': 'sshproxy_tunnel',
    }
    _speed_dict = {
            '0': 'modem',
            '1': 'isdn',
            '2': 'adsl',
            '3': 'wan',
            '4': 'lan',
    }

    for opt, val in _options.iteritems():

        # rename options if necessary
        if opt in _rename_dict.keys():
            del _params[opt]
            opt = _rename_dict[opt]
            _params[opt] = val

        # translate integer values for connection speed to readable strings
        if opt == 'link':
            val = str(val).lower()
            if val in _speed_dict.keys():
                val = _speed_dict[val]
            val = val.lower()
            _params['link'] = val

        # share_local_folders is a list
        if opt in ('share_local_folders', 'mimebox_extensions'):
            if type(val) is types.StringType:
                if val:
                    _params[opt] = val.split(',')
                else:
                    _params[opt] = []

    # append value for quality to value for pack method
    if _params['quality']:
        _params['pack'] = '%s-%s' % (_params['pack'], _params['quality'])
    # delete quality in any case...
    del _params['quality']

    del _params['fstunnel']

    if _params.has_key('export'):

        _export = _params['export']
        del _params['export']
        # fix for wrong export field usage in PyHoca-GUI/CLI and python-x2go before 20110923
        _export = _export.replace(",", ";")

        _export = _export.strip().strip('"').strip().strip(';').strip()
        _export_list = [ f for f in _export.split(';') if f ]

        _params['share_local_folders'] = []
        for _shared_folder in _export_list:
            # fix for wrong export field usage in PyHoca-GUI/CLI and python-x2go before 20110923
            if not ":" in _shared_folder: _shared_folder = "%s:1" % _shared_folder
            if _shared_folder.split(":")[1] == "1":
                _params['share_local_folders'].append(_shared_folder.split(":")[0])

    if not _options['fullscreen']:
        _params['geometry'] = '%sx%s' % (_options['width'], _options['height'])
    else:
        _params['geometry'] = 'fullscreen'
    del _params['width']
    del _params['height']
    del _params['fullscreen']

    if not _options['sound']:
        _params['snd_system'] = 'none'
    del _params['sound']

    if _options['rootless']:
        _params['session_type'] = 'application'
    else:
        _params['session_type'] = 'desktop'
    del _params['rootless']

    if _params['mimebox_action'] not in _X2GO_MIMEBOX_ACTIONS.keys():
        _params['mimebox_action'] = 'OPEN'

    if not _options['usekbd']:
        _params['kbtype'] = 'null/null'
        _params['kblayout'] = 'null'
    del _params['usekbd']

    # currently known but ignored in Python X2go
    _ignored_options = [
            'dpi',
            'setdpi',
            'startsoundsystem',
            'soundtunnel',
            'defsndport',
            'icon',
            'applications',
    ]
    for i in _ignored_options:
        del _params[i]

    return _params


def session_names_by_timestamp(session_infos):
    """\
    Sorts session profile names by their timestamp (as used in the file format's section name).

    """
    session_names = session_infos.keys()
    sortable_session_names = [ '%s|%s' % (session_name.split('-')[2].split('_')[0], session_name) for session_name in session_names ]
    sortable_session_names.sort()
    return [ session_name.split('|')[1] for session_name in sortable_session_names ]


def touch_file(filename, mode='a'):
    """\
    Imitates the behaviour of the GNU/touch command.

    @param filename: name of the file to touch
    @type filename: C{str}
    @param mode: the file mode (as used for Python file objects)
    @type mode: C{str}
    """
    if not os.path.isdir(os.path.dirname(filename)):
        os.makedirs(os.path.dirname(filename), mode=00700)
    f = open(filename, mode=mode)
    f.close()


def unique(seq):
    """\
    Imitates the behaviour of the GNU/uniq command.

    @param seq: a list/sequence containing consecutive duplicates.
    @type seq: C{list}

    @return: list that has been clean up from the consecutive duplicates
    @rtype: C{list}
    """
    # order preserving
    noDupes = []
    [noDupes.append(i) for i in seq if not noDupes.count(i)]
    return noDupes


def known_encodings():
    """\
    Render a list of all-known-to-Python character encodings (including 
    all known aliases)

    """
    from encodings.aliases import aliases
    _raw_encname_list = []
    _raw_encname_list.extend(aliases.keys())
    _raw_encname_list.extend(aliases.values())
    _raw_encname_list.sort()
    _encname_list = []
    for _raw_encname in _raw_encname_list:
        _encname = _raw_encname.upper()
        _encname = _encname.replace('_', '-')
        _encname_list.append(_encname)
    _encname_list.sort()
    _encname_list = unique(_encname_list)
    return _encname_list


def patiently_remove_file(dirname, filename):
    """\
    Try to remove a file, wait for unlocking, remove it once removing is possible...

    @param dirname: directory name the file is in
    @type dirname: C{str}
    @param filename: name of the file to be removed
    @type filename: C{str}
    """
    _not_removed = True
    while _not_removed:
        try:
            os.remove(os.path.join(dirname, filename))
            _not_removed = False
        except:
            # file is probably locked
            gevent.sleep(5)

def detect_unused_port(bind_address='', preferred_port=None):
    """\
    Detect an unused IP socket.

    @param bind_address: IP address to bind to
    @type bind_address: C{str}
    @param preferred_port: IP socket port that shall be tried first for availability
    @type preferred_port: C{str}

    @return: free local IP socket port that can be used for binding
    @rtype: C{str}
    """

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
    try:
        if preferred_port:
            sock.bind((bind_address, preferred_port))
            ipaddr, port = sock.getsockname()
        else:
            raise
    except:
        sock.bind(('', 0))
        ipaddr, port = sock.getsockname()
    return port

def get_encoding():
    """\
    Detect systems default character encoding.

    @return: The system's local character encoding.
    @rtype: C{str}
    """
    try:
        encoding = locale.getdefaultlocale()[1]
        if encoding is None:
            raise BaseException
    except:
        try:
            encoding = sys.getdefaultencoding()
        except:
            encoding = 'ascii'
    return encoding

def is_abs_path(path):
    """\
    Test if a given path is an absolute path name.

    @param path: test this path for absolutism...
    @type path: C{str}

    @return: Returns C{True} if path is an absolute path name
    @rtype: C{bool}
    """
    return bool((path.startswith('/') or re.match('^[%s]\:\\\\' % string.ascii_letters, path)))

def xkb_rules_names():
    """\
    Wrapper for: xprop -root _XKB_RULES_NAMES

    @return: A Python dictionary that contains the current X11 keyboard rules.
    @rtype: C{dict}

    """
    p = subprocess.Popen(['xprop', '-root', '_XKB_RULES_NAMES',], stdout=subprocess.PIPE, )
    _rn_list = p.stdout.read().split('"')
    _rn_dict = {
        'rules': _rn_list[1],
        'model': _rn_list[3],
        'layout': _rn_list[5],
        'variant': _rn_list[7],
        'options': _rn_list[9],
    }
    return _rn_dict

def local_color_depth():
    """\
    Detect the current local screen's color depth.

    """
    if _X2GOCLIENT_OS != 'Windows':
        try:
            p = subprocess.Popen(['xwininfo', '-root',], stdout=subprocess.PIPE, )
            _depth_line = [ _info.strip() for _info in p.stdout.read().split('\n') if 'Depth:' in _info ][0]
            _depth = _depth_line.split(' ')[1]
            return int(_depth)
        except IndexError:
            # a sensible default value
            return 24
        except OSError:
            # for building your package...
            return 24

    else:
        return win32api.GetSystemMetrics(2)

def is_color_depth_ok(depth_session, depth_local):
    """\
    Test if color depth of this session is compatible with the
    local screen's color depth.

    @param depth_session: color depth of the session
    @type depth_session: C{int}
    @param depth_local: color depth of local screen
    @type depth_local: C{int}

    @return: Does the session color depth work with the local display?
    @rtype: C{bool}

    """
    if depth_session == 0:
        return True
    if depth_session == depth_local:
        return True
    if ( ( depth_session == 24 or depth_session == 32 ) and ( depth_local == 24 or depth_local == 32 ) ):
        return True;
    return False