# -*- coding: utf-8 -*-
# vim:fenc=utf-8
# Copyright (C) 2012-2020 by Mike Gabriel <mike.gabriel@das-netzwerkteam.de>
#
# X2Go Session Broker 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.
#
# X2Go Session Broker 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.
"""\
This modules provides various helper functions and classes for the
UCCS web frontend of the X2Go Session Broker.
The UCCS protocol was originally brought to life by Canonical and was
part of Unity Greeter in Ubuntu 12.10.
See early blog posts about that topic:
* https://sunweavers.net/blog/how-to-for-hi-jacking-the-rdp-remote-login-feature-introduced-in-ubuntu-12-10-with-x2go
* https://sunweavers.net/blog/thoughts-about-canonical-s-uccs-service
* https://sunweavers.net/blog/unity-greeter-with-x2go-remote-login-support
The UCCS web frontend of the X2Go Session Broker offers remote logon
support to the Unity Greeter fork called Arctica Greeter.
"""
try: import simplejson as json
except ImportError: import json
import re
import copy
from x2gobroker.defaults import X2GOBROKER_LATEST_UCCS_API_VERSION as latest_api_version
[docs]
def convert_to_builtin_type(obj):
"""\
Helper function for converting Python objects to dictionaries.
Used for doing JSON dumps.
"""
d_obj = { }
d_obj.update(obj.__dict__)
d_obj_res = copy.deepcopy(d_obj)
# hide private object items (starting with "_") from the resulting dict
for key, val in d_obj.items():
if re.match("^_[a-zA-Z]{1}.*", key):
del d_obj_res[key]
return d_obj_res
[docs]
class ManagementServer():
"""\
Generate a UCCS compatible JSON object for a UCCS management server.
A :class:`ManagementServer` in UCCS terminology is a host offering a
UCCS compliant API for handling remote logons from clients over the
web.
The :class:`ManagementServer` is the entry point for all clients.
:param url: URL of the UCCS broker server
:type url: ``str``
:param name: human-readable, descriptive server name
:type name: ``str``
:param api_version: API version used between remote logon service and broker
:type api_version: ``int``
"""
def __init__(self, url, name, api_version=latest_api_version):
"""\
Initialize instance.
"""
self._api_version = api_version
self.RemoteDesktopServers = []
self.AdditionalManagementServers = []
self.URL = '{url}/'.format(url=url.rstrip('/'))
self.Name = name
[docs]
def set_default(self, ts_name):
"""\
Define the default (terminal) server instance.
:param ts_name: name of the terminal server that is to be set as default
:type ts_name: ``str``
"""
if isinstance(ts_name, str):
self.DefaultServer = ts_name
elif isinstance(ts_name, str):
self.DefaultServer = ts_name
else:
raise TypeError("set_default expects a string argument")
[docs]
def add_terminalserver(self, server):
"""\
Add a terminal server to this management server object.
:param server: instance of class :class:`RDPServer` or :class:`X2GoServer`
:type server: ``obj``
"""
self.RemoteDesktopServers.append(server)
[docs]
def add_additional_management_server(self, amserver):
"""\
Add / cascade a managemnet server to this management server
object.
:param server: instance of class :class:`AdditionalManagementServer`
:type server: ``obj``
"""
if isinstance(amserver, AdditionalManagementServer):
self.AdditionalManagementServers.append(amserver)
else:
raise TypeError("add_additional_management_server expects a "\
"AdditionalManagementServer argument")
[docs]
def toJson(self):
"""\
Dump this instance as JSON object.
"""
return json.dumps(self, default=convert_to_builtin_type, sort_keys=True, indent=2)
[docs]
class AdditionalManagementServer():
"""\
Instantiate a to-be-cascaded sub-management UCCS server.
In UCCS, other than terminal servers, you can add
:class:`AdditionalManagementServer` instances and cascade UCCS
setups.
:param url: URL of the UCCS broker server
:type url: ``str``
:param name: human-readable, descriptive server name
:type name: ``str``
:param api_version: API version used between remote logon service and broker
:type api_version: ``int``
"""
pass
[docs]
class RDPServer():
"""\
Instantiate a UCCS compatible RDP server session profile object.
:param host: hostname of RDP server host
:type host: ``str``
:param name: session profile name
:type name: ``str``
:param username: username to be used for login
:type username: ``str``
:param password: password to be used for login
:type password: ``str``
:param api_version: API version used between remote logon service and broker
:type api_version: ``int``
"""
def __init__(self, host, name, username='', password='', api_version=latest_api_version):
self._api_version = api_version
self.URL = 'http://{url}/'.format(url=host)
self.Name = name
self.Protocol = 'rdp'
self.DomainRequired = True
self.Username = username
self.Password = password
[docs]
def set_domain(self, domain):
"""\
Set the domain for this RDP server.
:param domain: the domain name to be set
:type domain: ``str``
:raises TypeError: domain has to be ``str``
"""
if isinstance(domain, str):
self.WindowsDomain = domain
elif isinstance(domain, str):
self.WindowsDomain = domain
else:
raise TypeError("set_domain() expects a string or unicode argument")
[docs]
def toJson(self):
"""\
Dump this instance as JSON object.
"""
return json.dumps(self, default=convert_to_builtin_type, sort_keys=True, indent=2)
[docs]
class ICAServer():
"""\
Instantiate a UCCS compatible ICA server session profile object.
:param host: hostname of ICA server host
:type host: ``str``
:param name: session profile name
:type name: ``str``
:param username: username to be used for login
:type username: ``str``
:param password: password to be used for login
:type password: ``str``
:param api_version: API version used between remote logon service and broker
:type api_version: ``int``
"""
def __init__(self, host, name, username='', password='', api_version=latest_api_version):
self._api_version = api_version
self.URL = 'http://{url}/'.format(url=host)
self.Name = name
self.Protocol = 'ica'
self.DomainRequired = True
self.Username = username
self.Password = password
[docs]
def set_domain(self, domain):
"""\
Set the domain for this ICA server.
:param domain: the domain name to be set
:type domain: ``str``
:raises TypeError: domain has to be ``str``
"""
if isinstance(domain, str):
self.WindowsDomain = domain
elif isinstance(domain, str):
self.WindowsDomain = domain
else:
raise TypeError("set_domain() expects a string or unicode argument")
[docs]
def toJson(self):
"""\
Dump this instance as JSON object.
"""
return json.dumps(self, default=convert_to_builtin_type, sort_keys=True, indent=2)
[docs]
class X2GoServer():
"""\
Instantiate a UCCS compatible X2Go Server session profile object.
:param host: hostname of X2Go Server host
:type host: ``str``
:param name: session profile name
:type name: ``str``
:param username: username to be used for login
:type username: ``str``
:param password: password to be used for login
:type password: ``str``
:param api_version: API version used between remote logon service and broker
:type api_version: ``int``
"""
def __init__(self, host, name, username='', password='', api_version=latest_api_version):
self._api_version = api_version
self.URL = 'http://{url}/'.format(url=host)
self.Name = name
self.Protocol = 'x2go'
if self._api_version == 4:
self.SessionTypeRequired = True
else:
self.CommandRequired = True
self.Username = username
self.Password = password
# legacy: API v4 (not used in API v5 and higher)
[docs]
def set_session_type(self, session_type):
"""\
Set the session type to be used on this X2Go Server.
This method is a legacy method formerly used by APIv4 of the UCCS
protocol. In APIv5, the method :func:`set_command()` gets used instead.
:param command: the session type to be set
:type command: ``str``
:raises TypeError: command has to be ``str``
"""
#FIXME: throw an exception when used with API v5 or newer
if isinstance(session_type, str):
self.SessionType = session_type
else:
raise TypeError("set_session_type() expects a string argument")
# since: API v5
[docs]
def set_command(self, command):
"""\
Set the command to be used on this X2Go Server.
Added since APIv5 of the UCCS protocol.
:param command: the session type to be set
:type command: ``str``
:raises TypeError: command has to be ``str``
"""
#FIXME: throw an exception when used with API v4
if isinstance(command, str):
self.Command = command
else:
raise TypeError("set_command() expects a string argument")
[docs]
def toJson(self):
"""\
Dump this instance as JSON object.
"""
return json.dumps(self, default=convert_to_builtin_type, sort_keys=True, indent=2)