shaba/openuds
1
0
Fork 1
mirror of https://github.com/dkmstr/openuds.git synced 2025-01-11 05:17:55 +03:00
Code Issues Releases Wiki Activity

* Renamed model file

Browse Source 
* Added __str__ and __repr__ to user_interface (with some minor type fixes)
* Added first migrator for HTML5RDP Transport to new model
This commit is contained in:
Adolfo Gómez García 2023-08-05 05:39:30 +02:00
parent a45baffa51
commit 8ce84f3c10
No known key found for this signature in database
GPG Key ID: DD1ABF20724CDA23
9 changed files with 186 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
server/src/tests/core/ui/test_userinterface.py
View File

@ -86,7 +86,7 @@ def oldSerializeForm(ui) -> bytes:
):
# logger.debug('Serializing value {0}'.format(v.value))
val = MULTIVALUE_FIELD + pickle.dumps(v.value, protocol=0)
elif v.isType(gui.InfoField.Types.PASSWORD):
elif v.isType(gui.InputField.Types.PASSWORD):
val = PASSWORD_FIELD + CryptoManager().AESCrypt(
v.value.encode('utf8'), UDSK, True
)

131
server/src/uds/core/ui/user_interface.py
View File

@ -154,9 +154,7 @@ class gui:
] = {}
@staticmethod
def choiceItem(
id_: typing.Union[str, int], text: typing.Union[str, int]
) -> 'gui.ChoiceType':
def choiceItem(id_: typing.Union[str, int], text: typing.Union[str, int]) -> 'gui.ChoiceType':
"""
Helper method to create a single choice item.
@ -191,9 +189,7 @@ class gui:
return []
# Helper to convert an item to a dict
def choiceFromValue(
val: typing.Union[str, int, typing.Dict[str, str]]
) -> 'gui.ChoiceType':
def choiceFromValue(val: typing.Union[str, int, typing.Dict[str, str]]) -> 'gui.ChoiceType':
if isinstance(val, dict):
if 'id' not in val or 'text' not in val:
raise ValueError(f'Invalid choice dict: {val}')
@ -217,9 +213,7 @@ class gui:
raise ValueError(f'Invalid type for convertToChoices: {vals}')
@staticmethod
def convertToList(
vals: typing.Union[str, int, typing.Iterable]
) -> typing.List[str]:
def convertToList(vals: typing.Union[str, int, typing.Iterable]) -> typing.List[str]:
if vals:
if isinstance(vals, (str, int)):
return [str(vals)]
@ -227,9 +221,7 @@ class gui:
return []
@staticmethod
def choiceImage(
id_: typing.Union[str, int], text: str, img: str
) -> typing.Dict[str, str]:
def choiceImage(id_: typing.Union[str, int], text: str, img: str) -> typing.Dict[str, str]:
return {'id': str(id_), 'text': str(text), 'img': img}
@staticmethod
@ -337,9 +329,7 @@ class gui:
if 'type' in options:
self.type = options['type'] # set type first
defvalue = options.get(
'defvalue', options.get('defaultValue', options.get('defValue', ''))
)
defvalue = options.get('defvalue', options.get('defaultValue', options.get('defValue', '')))
self._data.update(
{
'length': options.get(
@ -399,11 +389,7 @@ class gui:
"""
if callable(self._data['value']):
return self._data['value']()
return (
self._data['value']
if self._data['value'] is not None
else self.defValue
)
return self._data['value'] if self._data['value'] is not None else self.defValue
@value.setter
def value(self, value: typing.Any) -> None:
@ -438,11 +424,7 @@ class gui:
"""
Returns the default value for this field
"""
return (
self._data['defvalue']
if not callable(self._data['defvalue'])
else self._data['defvalue']()
)
return self._data['defvalue'] if not callable(self._data['defvalue']) else self._data['defvalue']()
@defValue.setter
def defValue(self, defValue: typing.Any) -> None:
@ -479,6 +461,13 @@ class gui:
"""
return True
def __str__(self):
return str(self.value)
def __repr__(self):
args = ','.join([f'{k}="{v}"' for k,v in self._data.items()])
return f'{self.__class__.__name__}({args})'
class TextField(InputField):
"""
This represents a text field.
@ -511,7 +500,7 @@ class gui:
"""
class PatternType(enum.Enum):
class PatternType(enum.StrEnum):
IPV4 = 'ipv4'
IPV6 = 'ipv6'
IP = 'ip'
@ -541,9 +530,7 @@ class gui:
# - 'path' # Path (absolute or relative, Windows or Unix)
# Note:
# Checks are performed on admin side, so they are not 100% reliable.
self._data['pattern'] = options.get(
'pattern', gui.TextField.PatternType.NONE
)
self._data['pattern'] = options.get('pattern', gui.TextField.PatternType.NONE)
if isinstance(self._data['pattern'], str):
self._data['pattern'] = gui.TextField.PatternType(self._data['pattern'])
@ -682,9 +669,7 @@ class gui:
datetime.date: the date that this object holds, or "min" | "max" on error
"""
try:
return datetime.datetime.strptime(
self.value, '%Y-%m-%d'
).date() # ISO Format
return datetime.datetime.strptime(self.value, '%Y-%m-%d').date() # ISO Format
except Exception:
return datetime.date.min if useMin else datetime.date.max
@ -704,11 +689,10 @@ class gui:
return datetime.datetime.min if useMin else datetime.datetime.max
def stamp(self) -> int:
return int(
time.mktime(
datetime.datetime.strptime(self.value, '%Y-%m-%d').timetuple()
)
)
return int(time.mktime(datetime.datetime.strptime(self.value, '%Y-%m-%d').timetuple()))
def __str__(self):
return str(self.datetime())
class PasswordField(InputField):
"""
@ -738,6 +722,9 @@ class gui:
def cleanStr(self):
return str(self.value).strip()
def __str__(self):
return '********'
class HiddenField(InputField):
"""
This represents a hidden field. It is not displayed to the user. It use
@ -823,6 +810,9 @@ class gui:
"""
return self.isTrue()
def __str__(self):
return str(self.isTrue())
class ChoiceField(InputField):
"""
This represents a simple combo box with single selection.
@ -1074,9 +1064,7 @@ class UserInterfaceType(type):
_gui[attrName]._data = copy.deepcopy(attr._data)
newClassDict[attrName] = attr
newClassDict['_base_gui'] = _gui
return typing.cast(
'UserInterfaceType', type.__new__(mcs, classname, bases, newClassDict)
)
return typing.cast('UserInterfaceType', type.__new__(mcs, classname, bases, newClassDict))
class UserInterface(metaclass=UserInterfaceType):
@ -1116,9 +1104,7 @@ class UserInterface(metaclass=UserInterfaceType):
self._gui = copy.deepcopy(self._base_gui)
for key, val in self._gui.items(): # And refresh self references to them
setattr(
self, key, val
) # val is an InputField instance, so it is a reference to self._gui[key]
setattr(self, key, val) # val is an InputField instance, so it is a reference to self._gui[key]
if values is not None:
for k, v in self._gui.items():
@ -1201,7 +1187,7 @@ class UserInterface(metaclass=UserInterfaceType):
return serializer.serialize(value)
fw_converters: typing.Mapping[
gui.InfoField.Types, typing.Callable[[gui.InputField], typing.Optional[str]]
gui.InputField.Types, typing.Callable[[gui.InputField], typing.Optional[str]]
] = {
gui.InputField.Types.TEXT: lambda x: x.value,
gui.InputField.Types.TEXT_AUTOCOMPLETE: lambda x: x.value,
@ -1209,19 +1195,11 @@ class UserInterface(metaclass=UserInterfaceType):
gui.InputField.Types.PASSWORD: lambda x: (
CryptoManager().AESCrypt(x.value.encode('utf8'), UDSK, True).decode()
),
gui.InputField.Types.HIDDEN: (
lambda x: None if not x.isSerializable() else x.value
),
gui.InfoField.Types.CHOICE: lambda x: x.value,
gui.InputField.Types.MULTI_CHOICE: lambda x: codecs.encode(
serialize(x.value), 'base64'
).decode(),
gui.InputField.Types.EDITABLE_LIST: lambda x: codecs.encode(
serialize(x.value), 'base64'
).decode(),
gui.InputField.Types.CHECKBOX: lambda x: gui.TRUE
if x.isTrue()
else gui.FALSE,
gui.InputField.Types.HIDDEN: (lambda x: None if not x.isSerializable() else x.value),
gui.InputField.Types.CHOICE: lambda x: x.value,
gui.InputField.Types.MULTI_CHOICE: lambda x: codecs.encode(serialize(x.value), 'base64').decode(),
gui.InputField.Types.EDITABLE_LIST: lambda x: codecs.encode(serialize(x.value), 'base64').decode(),
gui.InputField.Types.CHECKBOX: lambda x: gui.TRUE if x.isTrue() else gui.FALSE,
gui.InputField.Types.IMAGE_CHOICE: lambda x: x.value,
gui.InputField.Types.IMAGE: lambda x: x.value,
gui.InputField.Types.DATE: lambda x: x.value,
@ -1265,8 +1243,7 @@ class UserInterface(metaclass=UserInterfaceType):
# For future use, right now we only have one version
version = values[ # pylint: disable=unused-variable
len(SERIALIZATION_HEADER) : len(SERIALIZATION_HEADER)
+ len(SERIALIZATION_VERSION)
len(SERIALIZATION_HEADER) : len(SERIALIZATION_HEADER) + len(SERIALIZATION_VERSION)
]
values = values[len(SERIALIZATION_HEADER) + len(SERIALIZATION_VERSION) :]
@ -1278,17 +1255,12 @@ class UserInterface(metaclass=UserInterfaceType):
# Set all values to defaults ones
for k in self._gui:
if (
self._gui[k].isType(gui.InputField.Types.HIDDEN)
and self._gui[k].isSerializable() is False
):
if self._gui[k].isType(gui.InputField.Types.HIDDEN) and self._gui[k].isSerializable() is False:
# logger.debug('Field {0} is not unserializable'.format(k))
continue
self._gui[k].value = self._gui[k].defValue
converters: typing.Mapping[
gui.InfoField.Types, typing.Callable[[str], typing.Any]
] = {
converters: typing.Mapping[gui.InputField.Types, typing.Callable[[str], typing.Any]] = {
gui.InputField.Types.TEXT: lambda x: x,
gui.InputField.Types.TEXT_AUTOCOMPLETE: lambda x: x,
gui.InputField.Types.NUMERIC: int,
@ -1296,13 +1268,9 @@ class UserInterface(metaclass=UserInterfaceType):
CryptoManager().AESDecrypt(x.encode(), UDSK, True).decode()
),
gui.InputField.Types.HIDDEN: lambda x: None,
gui.InfoField.Types.CHOICE: lambda x: x,
gui.InputField.Types.MULTI_CHOICE: lambda x: deserialize(
codecs.decode(x.encode(), 'base64')
),
gui.InputField.Types.EDITABLE_LIST: lambda x: deserialize(
codecs.decode(x.encode(), 'base64')
),
gui.InputField.Types.CHOICE: lambda x: x,
gui.InputField.Types.MULTI_CHOICE: lambda x: deserialize(codecs.decode(x.encode(), 'base64')),
gui.InputField.Types.EDITABLE_LIST: lambda x: deserialize(codecs.decode(x.encode(), 'base64')),
gui.InputField.Types.CHECKBOX: lambda x: x,
gui.InputField.Types.IMAGE_CHOICE: lambda x: x,
gui.InputField.Types.IMAGE: lambda x: x,
@ -1343,10 +1311,7 @@ class UserInterface(metaclass=UserInterfaceType):
try:
# Set all values to defaults ones
for k in self._gui:
if (
self._gui[k].isType(gui.InputField.Types.HIDDEN)
and self._gui[k].isSerializable() is False
):
if self._gui[k].isType(gui.InputField.Types.HIDDEN) and self._gui[k].isSerializable() is False:
# logger.debug('Field {0} is not unserializable'.format(k))
continue
self._gui[k].value = self._gui[k].defValue
@ -1403,15 +1368,11 @@ class UserInterface(metaclass=UserInterfaceType):
return res
def errors(self) -> typing.List[ValidationFieldInfo]:
found_erros: typing.List[UserInterface.ValidationFieldInfo] = []
found_errors: typing.List[UserInterface.ValidationFieldInfo] = []
for key, val in self._gui.items():
if val.required and not val.value:
found_erros.append(
UserInterface.ValidationFieldInfo(key, 'Field is required')
)
found_errors.append(UserInterface.ValidationFieldInfo(key, 'Field is required'))
if not val.validate():
found_erros.append(
UserInterface.ValidationFieldInfo(key, 'Field is not valid')
)
found_errors.append(UserInterface.ValidationFieldInfo(key, 'Field is not valid'))
return found_erros
return found_errors

22
server/src/uds/migrations/0046_registered_server_migration.py
View File

@ -1,12 +1,11 @@
import typing
from django.db import migrations, models
from uds.core.util.os_detector import KnownOS
from django.db import migrations, models
import django.db.models.deletion
import uds.core.util.model
import uds.core.types.servers
from django.db import migrations, models
import uds.core.types.servers
import uds.core.util.model
from uds.core.util.os_detector import KnownOS
ACTOR_TYPE: typing.Final[int] = uds.core.types.servers.ServerType.ACTOR.value
@ -14,13 +13,6 @@ if typing.TYPE_CHECKING:
import uds.models
def migrate_html5rdp_transport(apps, schema_editor) -> None:
try:
Transport: 'typing.Type[uds.models.Transport]' = apps.get_model('uds', 'Transport')
except Exception: # nosec: ignore this
pass
def migrate_old_data(apps, schema_editor) -> None:
try:
RegisteredServer: 'typing.Type[uds.models.RegisteredServer]' = apps.get_model('uds', 'RegisteredServer')
@ -126,6 +118,12 @@ class Migration(migrations.Migration):
"abstract": False,
},
),
migrations.AddConstraint(
model_name="registeredservergroup",
constraint=models.UniqueConstraint(
fields=("host", "port"), name="unique_host_port_group"
),
),
migrations.AddField(
model_name="registeredserver",
name="tags",

1
server/src/uds/migrations/__init__.py
View File

@ -1 +0,0 @@
# Removed old south support

0
server/src/uds/migrations/fixers/__init__.py Normal file
View File

119
server/src/uds/migrations/fixers/html5rdp.py Normal file
View File

@ -0,0 +1,119 @@
import typing
import logging
from uds.core.ui import gui
from uds.core import transports
from uds.core.environment import Environment
from uds.core.types import servers
if typing.TYPE_CHECKING:
import uds.models
logger = logging.getLogger(__name__)
# Copy for migration
class HTML5RDPTransport(transports.Transport):
"""
Provides access via RDP to service.
This transport can use an domain. If username processed by authenticator contains '@', it will split it and left-@-part will be username, and right password
"""
typeName = 'HTML5 RDP' # Not important here, just for migrations
typeType = 'HTML5RDPTransport'
guacamoleServer = gui.TextField()
useGlyptodonTunnel = gui.CheckBoxField()
useEmptyCreds = gui.CheckBoxField()
fixedName = gui.TextField()
fixedPassword = gui.PasswordField()
withoutDomain = gui.CheckBoxField()
fixedDomain = gui.TextField()
wallpaper = gui.CheckBoxField()
desktopComp = gui.CheckBoxField()
smooth = gui.CheckBoxField()
enableAudio = gui.CheckBoxField(
defvalue=gui.TRUE,
)
enableAudioInput = gui.CheckBoxField()
enablePrinting = gui.CheckBoxField()
enableFileSharing = gui.ChoiceField(
defvalue='false',
)
enableClipboard = gui.ChoiceField(
defvalue='enabled',
)
serverLayout = gui.ChoiceField(
defvalue='-',
)
ticketValidity = gui.NumericField(
defvalue='60',
)
forceNewWindow = gui.ChoiceField(
defvalue=gui.FALSE,
)
security = gui.ChoiceField(
defvalue='any',
)
rdpPort = gui.NumericField(
defvalue='3389',
)
customGEPath = gui.TextField(
defvalue='/',
)
# This value is the new "tunnel server"
# Old guacamoleserver value will be stored also on database, but will be ignored
tunnelServer = gui.ChoiceField()
def migrate_html5rdp_transport(apps, schema_editor) -> None:
try:
# Transport: 'typing.Type[uds.models.Transport]' = apps.get_model('uds', 'Transport')
# RegisteredServerGroup: 'typing.Type[uds.models.RegisteredServerGroup]' = apps.get_model('uds', 'RegisteredServerGroup')
# RegisteredServer: 'typing.Type[uds.models.RegisteredServer]' = apps.get_model('uds', 'RegisteredServer')
# For testing
from uds.models import Transport, RegisteredServerGroup, RegisteredServer
for t in Transport.objects.filter(data_type=HTML5RDPTransport.typeType):
print(t)
# Extranct data
obj = HTML5RDPTransport(Environment(t.uuid), None)
obj.deserialize(t.data)
# Guacamole server is https://<host>:<port>
if not obj.guacamoleServer.value.startswith('https://'):
# Skip if not https found
logger.error('Skipping HTML5RDP transport %s as it does not starts with https://', t.name)
continue
host, port = (obj.guacamoleServer.value+':443').split('https://')[1].split(':')[:2]
# Look for an existing tunnel server (RegisteredServerGroup)
tunnelServer = RegisteredServerGroup.objects.filter(
host=host, port=port, kind=servers.ServerType.TUNNEL
).first()
if tunnelServer is None:
logger.info('Creating new tunnel server for HTML5RDP: %s:%s', host, port)
# Create a new one, adding all tunnel servers to it
tunnelServer = RegisteredServerGroup.objects.create(
name=f'HTML5RDP Tunnel on {host}:{port}',
comments='Tunnel server for HTML5 RDP (migration)',
host=host,
port=port,
kind=servers.ServerType.TUNNEL,
)
tunnelServer.servers.set(RegisteredServer.objects.filter(kind=servers.ServerType.TUNNEL))
# Set tunnel server on transport
logger.info('Setting tunnel server %s on transport %s', tunnelServer.name, t.name)
obj.tunnelServer.value = tunnelServer.uuid
# Save transport
t.data = obj.serialize()
t.save(update_fields=['data'])
except Exception as e: # nosec: ignore this
print(e)
logger.exception('Exception found while migrating HTML5RDP transports')

2
server/src/uds/models/__init__.py
View File

@ -103,7 +103,7 @@ from .account_usage import AccountUsage
from .tag import Tag, TaggingMixin
# Tokens
from .registered_servers import RegisteredServer, RegisteredServerGroup
from .servers import RegisteredServer, RegisteredServerGroup
# Notifications & Alerts
from .notifications import Notification, Notifier, LogLevel

8
server/src/uds/models/registered_servers.py → server/src/uds/models/servers.py
View File

@ -76,6 +76,14 @@ class RegisteredServerGroup(UUIDModel, TaggingMixin): # type: ignore # Mypy co
transports: 'models.manager.RelatedManager[Transport]'
servers: 'models.manager.RelatedManager[RegisteredServer]'
class Meta:
# Unique for host and port, so we can have only one group for each host:port
constraints = [
models.UniqueConstraint(
fields=['host', 'port'], name='unique_host_port_group'
)
]
@property
def pretty_host(self) -> str:
if self.port == 0:

2
server/src/uds/transports/HTML5RDP/html5rdp.py
View File

@ -73,7 +73,7 @@ class HTML5RDPTransport(transports.Transport):
guacamoleServer = gui.TextField(
label=_('Tunnel Server'),
order=1,
tooltip=_('Host of the tunnel server (use http/https & port if needed) as accesible from users'),
tooltip=_('Host of the tunnel server (use https & port if needed) as accesible from users'),
defvalue='https://',
length=64,
required=True,