burmatovan/openuds
1
0
Fork 0
forked from shaba/openuds
Code Issues Pull Requests Projects Releases Wiki Activity

upgrades for 3.7

Browse Source
This commit is contained in:
Adolfo Gómez García 2019-08-05 21:04:36 +02:00
parent 5157dce173
commit e484c31b38
15 changed files with 169 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
server/src/uds/core/osmanagers/osmfactory.py
View File

@ -33,11 +33,12 @@
import logging
import typing
logger = logging.getLogger(__name__)
if typing.TYPE_CHECKING:
from .osmanager import OSManager
logger = logging.getLogger(__name__)
class OSManagersFactory:
_factory: typing.Optional['OSManagersFactory'] = None
_osManagers: typing.Dict[str, typing.Type['OSManager']]
@ -46,7 +47,7 @@ class OSManagersFactory:
self._osManagers = {}
@staticmethod
def factory():
def factory() -> 'OSManagersFactory':
if OSManagersFactory._factory is None:
OSManagersFactory._factory = OSManagersFactory()
return OSManagersFactory._factory
@ -54,7 +55,7 @@ class OSManagersFactory:
def providers(self):
return self._osManagers
def insert(self, type_: typing.Type['OSManager']):
def insert(self, type_: typing.Type['OSManager']) -> None:
logger.debug('Adding OS Manager %s as %s', type_.type(), type_)
typeName = type_.type().lower()
self._osManagers[typeName] = type_

60
server/src/uds/core/services/BasePublication.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012 Virtual Cable S.L.
# Copyright (c) 2012-2019 Virtual Cable S.L.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -30,12 +30,16 @@
"""
.. moduleauthor:: Adolfo Gómez, dkmaster at dkmon dot com
"""
from __future__ import unicode_literals
import typing
from uds.core import Environmentable
from uds.core import Serializable
__updated__ = '2018-06-07'
if typing.TYPE_CHECKING:
from uds.core import services
from uds.core import osmanagers
from uds.core.environment import Environment
from uds.models import DeployedServicePublication
class Publication(Environmentable, Serializable):
@ -49,7 +53,7 @@ class Publication(Environmentable, Serializable):
As always, do not forget to invoke base class __init__ if you override it as this::
super(self.__class__, self).__init__(environment, **kwargs)
super().__init__(environment, **kwargs)
This is a MUST, so internal structured gets filled correctly, so don't forget it!.
@ -77,7 +81,14 @@ class Publication(Environmentable, Serializable):
# : change suggestedTime in your implementation.
suggestedTime = 10
def __init__(self, environment, **kwargs):
_osmanager: typing.Optional['osmanagers.OSManager']
_service: 'services.Service'
_revision: int
_dbPublication: typing.Optional['DeployedServicePublication']
_dsName: str
_uuid: str
def __init__(self, environment: 'Environment', **kwargs):
"""
Do not forget to invoke this in your derived class using "super(self.__class__, self).__init__(environment, values)"
We want to use the env, cache and storage methods outside class. If not called, you must implement your own methods
@ -95,7 +106,7 @@ class Publication(Environmentable, Serializable):
self.initialize()
def initialize(self):
def initialize(self) -> None:
"""
This method will be invoked from __init__ constructor.
This is provided so you don't have to provide your own __init__ method,
@ -103,9 +114,8 @@ class Publication(Environmentable, Serializable):
This will get invoked when all initialization stuff is done, so
you can here access service, osManager, ...
"""
pass
def service(self):
def service(self) -> 'services.Service':
"""
Utility method to access parent service of this publication
@ -115,7 +125,7 @@ class Publication(Environmentable, Serializable):
"""
return self._service
def osManager(self):
def osManager(self) -> 'osmanagers.OSManager':
"""
Utility method to access os manager for this publication.
@ -127,27 +137,26 @@ class Publication(Environmentable, Serializable):
"""
return self._osManager
def revision(self):
def revision(self) -> int:
"""
Utility method to access the revision of this publication
This is a numeric value, and is set by core
"""
return self._revision
def dsName(self):
def dsName(self) -> str:
"""
Utility method to access the declared deployed service name.
This name is set by core, using the administrator provided data
at administration interface.
'''
return self._dsName
def getUuid(self):
"""
return self._dsName
def publish(self):
def getUuid(self) -> str:
return self._uuid
def publish(self) -> str:
"""
This method is invoked whenever the administrator requests a new publication.
@ -180,9 +189,9 @@ class Publication(Environmentable, Serializable):
to the core. Take that into account and handle exceptions inside
this method.
"""
raise Exception('publish method for class {0} not provided! '.format(self.__class__.__name__))
raise NotImplementedError('publish method for class {0} not provided! '.format(self.__class__.__name__))
def checkState(self):
def checkState(self) -> str:
"""
This is a task method. As that, the expected return values are
State values RUNNING, FINISHED or ERROR.
@ -206,9 +215,9 @@ class Publication(Environmentable, Serializable):
to the core. Take that into account and handle exceptions inside
this method.
"""
raise Exception('checkState method for class {0} not provided!!!'.format(self.__class__.__name__))
raise NotImplementedError('checkState method for class {0} not provided!!!'.format(self.__class__.__name__))
def finish(self):
def finish(self) -> None:
"""
Invoked when Publication manager noticed that the publication has finished.
This give us the opportunity of cleaning up things (as stored vars, etc..)
@ -217,9 +226,8 @@ class Publication(Environmentable, Serializable):
Default implementation does nothing. You can leave default method if you
are going to do nothing.
"""
pass
def reasonOfError(self):
def reasonOfError(self) -> str:
"""
If a publication produces an error, here we must return the reason why
it happened. This will be called just after publish or checkPublishingState
@ -232,7 +240,7 @@ class Publication(Environmentable, Serializable):
"""
return 'unknown'
def destroy(self):
def destroy(self) -> str:
"""
This is a task method. As that, the expected return values are
State values RUNNING, FINISHED or ERROR.
@ -250,9 +258,9 @@ class Publication(Environmentable, Serializable):
to the core. Take that into account and handle exceptions inside
this method.
"""
raise Exception('destroy method for class {0} not provided!'.format(self.__class__.__name__))
raise NotImplementedError('destroy method for class {0} not provided!'.format(self.__class__.__name__))
def cancel(self):
def cancel(self) -> str:
"""
This is a task method. As that, the expected return values are
State values RUNNING, FINISHED or ERROR.
@ -270,7 +278,7 @@ class Publication(Environmentable, Serializable):
to the core. Take that into account and handle exceptions inside
this method.
"""
raise Exception('cancel method for class {0} not provided!'.format(self.__class__.__name__))
raise NotImplementedError('cancel method for class {0} not provided!'.format(self.__class__.__name__))
def __str__(self):
"""

38
server/src/uds/core/services/BaseService.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012 Virtual Cable S.L.
# Copyright (c) 2012-2019 Virtual Cable S.L.U.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -40,6 +40,16 @@ from . import types
from .BasePublication import Publication
from .BaseDeployed import UserDeployment
if typing.TYPE_CHECKING:
from uds.core import services
from uds.core import osmanagers
from uds.core.environment import Environment
from uds.core.util.UniqueNameGenerator import UniqueNameGenerator
from uds.core.util.UniqueMacGenerator import UniqueMacGenerator
from uds.core.util.UniqueGIDGenerator import UniqueGIDGenerator
from uds.models import DeployedServicePublication
class Service(Module):
"""
This class is in fact an interface, and represents a service, that is the
@ -178,17 +188,19 @@ class Service(Module):
# : For example, VDI, VAPP, ...
servicesTypeProvided: typing.Iterable = types.ALL
def __init__(self, environment, parent, values=None, uuid=None):
_provider: 'services.ServiceProvider'
def __init__(self, environment, parent: 'services.ServiceProvider', values: Module.ValuesType = None, uuid: typing.Optional[str] = None):
"""
Do not forget to invoke this in your derived class using "super(self.__class__, self).__init__(environment, parent, values)".
Do not forget to invoke this in your derived class using "super().__init__(environment, parent, values)".
We want to use the env, parent methods outside class. If not called, you must implement your own methods
cache and storage are "convenient" methods to access _env.cache and _env.storage
"""
super(Service, self).__init__(environment, values)
Module.__init__(self, environment, values, uuid)
self._provider = parent
self.initialize(values)
def initialize(self, values):
def initialize(self, values: Module.ValuesType) -> None:
"""
This method will be invoked from __init__ constructor.
This is provided so you don't have to provide your own __init__ method,
@ -204,7 +216,7 @@ class Service(Module):
Default implementation does nothing
"""
def parent(self):
def parent(self) -> 'services.ServiceProvider':
"""
Utility method to access parent provider for this service
@ -214,7 +226,7 @@ class Service(Module):
"""
return self._provider
def requestServicesForAssignation(self, **kwargs):
def requestServicesForAssignation(self, **kwargs) -> typing.Iterable[UserDeployment]:
"""
override this if mustAssignManualy is True
@params kwargs: Named arguments
@ -222,26 +234,26 @@ class Service(Module):
We will access the returned array in "name" basis. This means that the service will be assigned by "name", so be care that every single service
returned are not repeated... :-)
"""
raise Exception('The class {0} has been marked as manually asignable but no requestServicesForAssignetion provided!!!'.format(self.__class__.__name__))
raise NotImplementedError('The class {0} has been marked as manually asignable but no requestServicesForAssignetion provided!!!'.format(self.__class__.__name__))
def macGenerator(self):
def macGenerator(self) -> typing.Optional['UniqueMacGenerator']:
"""
Utility method to access provided macs generator (inside environment)
Returns the environment unique mac addresses generator
"""
return self.idGenerators('mac')
return typing.cast('UniqueMacGenerator', self.idGenerators('mac'))
def nameGenerator(self):
def nameGenerator(self) -> typing.Optional['UniqueNameGenerator']:
"""
Utility method to access provided names generator (inside environment)
Returns the environment unique name generator
"""
return self.idGenerators('name')
return typing.cast('UniqueNameGenerator', self.idGenerators('name'))
def __str__(self):
"""
String method, mainly used for debugging purposes
"""
return "Base Service Provider"
return 'Base Service Provider'

2
server/src/uds/core/services/BaseServiceProvider.py
View File

@ -199,4 +199,4 @@ class ServiceProvider(Module):
Basic implementation, mostly used for debuging and testing, never used
at user or admin interfaces.
"""
return "Base Service Provider"
return 'Base Service Provider'

18
server/src/uds/core/services/Exceptions.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012 Virtual Cable S.L.
# Copyright (c) 2012-2019 Virtual Cable S.L.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -30,9 +30,10 @@
"""
.. moduleauthor:: Adolfo Gómez, dkmaster at dkmon dot com
"""
from __future__ import unicode_literals
import typing
__updated__ = '2016-03-16'
if typing.TYPE_CHECKING:
from uds.models import UserService, Transport
class ServiceException(Exception):
@ -44,28 +45,24 @@ class UnsupportedException(ServiceException):
"""
Reflects that we request an operation that is not supported, i.e. Cancel a publication with snapshots
"""
pass
class OperationException(ServiceException):
"""
Reflects that the operation requested can't be acomplished, i.e. remove an snapshot without snapshot reference, cancel non running operation, etc...
"""
pass
class PublishException(ServiceException):
"""
Reflects thate the publication can't be done for causes we don't know in advance
"""
pass
class DeploymentException(ServiceException):
"""
Reflects that a deployment of a service (at cache, or assigned to user) can't be done for causes we don't know in advance
"""
pass
class CancelException(ServiceException):
@ -78,7 +75,6 @@ class InvalidServiceException(ServiceException):
"""
Invalid service specified. The service is not ready
"""
pass
class MaxServicesReachedError(ServiceException):
@ -86,21 +82,18 @@ class MaxServicesReachedError(ServiceException):
Number of maximum services has been reached, and no more services
can be created for users.
"""
pass
class ServiceInMaintenanceMode(ServiceException):
"""
The service is in maintenance mode and can't be accesed right now
"""
pass
class ServiceAccessDeniedByCalendar(ServiceException):
"""
This service can't be accessed right now, probably due to date-time restrictions
"""
pass
class ServiceNotReadyError(ServiceException):
@ -108,6 +101,9 @@ class ServiceNotReadyError(ServiceException):
The service is not ready
Can include an optional code error
"""
code: int
service: 'UserService'
transport: 'Transport'
def __init__(self, *args, **kwargs):
super(ServiceNotReadyError, self).__init__(*args, **kwargs)
self.code = kwargs.get('code', 0x0000)

6
server/src/uds/core/services/types.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012 Virtual Cable S.L.
# Copyright (c) 2012-20019 Virtual Cable S.L.U.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -30,10 +30,6 @@
"""
.. moduleauthor:: Adolfo Gómez, dkmaster at dkmon dot com
"""
from __future__ import unicode_literals
__updated__ = '2015-05-28'
VDI = 'vdi'
VAPP = 'vApp'

109
server/src/uds/core/transports/BaseTransport.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012 Virtual Cable S.L.
# Copyright (c) 2012-2019 Virtual Cable S.L.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -42,14 +42,17 @@ from uds.core.transports import protocols
from uds.core.util import encoders
from uds.core.util import connection
# Not imported in runtime, just for type checking
if typing.TYPE_CHECKING:
from django.http import HttpRequest # pylint: disable=ungrouped-imports
from uds.core.environment import Environment
from uds import models
logger = logging.getLogger(__name__)
DIRECT_GROUP = _('Direct')
TUNNELED_GROUP = _('Tunneled')
# Not imported in runtime, just for type checking
if typing.TYPE_CHECKING:
from uds import models
class Transport(Module):
"""
@ -68,26 +71,26 @@ class Transport(Module):
# Windows
# Macintosh
# Linux
supportedOss = OsDetector.desktopOss # Supported operating systems
supportedOss: typing.Tuple = OsDetector.desktopOss # Supported operating systems
# If this transport is visible via Web, via Thin Client or both
webTransport = False
tcTransport = False
webTransport: bool = False
tcTransport: bool = False
# If the link to use transport is provided by transport itself
ownLink = False
ownLink: bool = False
# Protocol "type". This is not mandatory, but will help
protocol = protocols.NONE
protocol: str = protocols.NONE
# For allowing grouping transport on dashboard "new" menu, and maybe other places
group = DIRECT_GROUP
group: str = DIRECT_GROUP
def __init__(self, environment, values):
super(Transport, self).__init__(environment, values)
def __init__(self, environment: 'Environment', values: Module.ValuesType):
super().__init__(environment, values)
self.initialize(values)
def initialize(self, values):
def initialize(self, values: Module.ValuesType):
"""
This method will be invoked from __init__ constructor.
This is provided so you don't have to provide your own __init__ method,
@ -102,60 +105,59 @@ class Transport(Module):
Default implementation does nothing
"""
pass
def destroy(self):
def destroy(self) -> None:
"""
Invoked when Transport is deleted
"""
pass
def testServer(self, userService: 'models.UserService', ip: str, port: typing.Union[str, int], timeout: int = 4):
proxy = userService.deployed_service.service.proxy
def testServer(self, userService: 'models.UserService', ip: str, port: typing.Union[str, int], timeout: int = 4) -> bool:
proxy: 'models.Proxy' = userService.deployed_service.service.proxy
if proxy is not None:
return proxy.doTestServer(ip, port, timeout)
return connection.testServer(ip, str(port), timeout)
def isAvailableFor(self, userService: 'models.UserService', ip: str):
def isAvailableFor(self, userService: 'models.UserService', ip: str) -> bool:
"""
Checks if the transport is available for the requested destination ip
Override this in yours transports
"""
return False
def getCustomAvailableErrorMsg(self, userService, ip):
def getCustomAvailableErrorMsg(self, userService: 'models.UserService', ip: str) -> str:
"""
Returns a customized error message, that will be used when a service fails to check "isAvailableFor"
Override this in yours transports if needed
"""
return "Not accessible (using service ip {0})".format(ip)
return "Not accessible (using service ip {})".format(ip)
@classmethod
def supportsProtocol(cls, protocol):
if isinstance(protocol, (list, tuple)):
for v in protocol:
if cls.supportsProtocol(v) is True:
return True
return False
return protocol.lower() == cls.protocol.lower()
def supportsProtocol(cls, protocol: typing.Union[typing.Iterable, str]):
if isinstance(protocol, str):
return protocol.lower() == cls.protocol.lower()
# Not string group of strings
for v in protocol:
if cls.supportsProtocol(v):
return True
return False
@classmethod
def supportsOs(cls, osName):
def supportsOs(cls, osName: str) -> bool:
"""
Helper method to check if transport supports requested operating system.
Class method
"""
logger.debug('Checking suported os {0} against {1}'.format(osName, cls.supportedOss))
logger.debug('Checking suported os %s against %s', osName, cls.supportedOss)
return cls.supportedOss.count(osName) > 0
@classmethod
def providesConnetionInfo(cls):
def providesConnetionInfo(cls) -> bool:
"""
Helper method to check if transport provides information about connection
"""
return cls.getConnectionInfo != Transport.getConnectionInfo
def getConnectionInfo(self, service, user, password):
def getConnectionInfo(self, userService: 'models.UserService', user: 'models.User', password: str) -> typing.Dict[str, str]:
"""
This method must provide information about connection.
We don't have to implement it, but if we wont to allow some types of connections
@ -180,7 +182,7 @@ class Transport(Module):
"""
return {'protocol': self.protocol, 'username': '', 'password': '', 'domain': ''}
def processedUser(self, userService, user):
def processedUser(self, userService: 'models.UserService', user: 'models.User') -> None:
"""
Used to "transform" username that will be sent to service
This is used to make the "user" that will receive the service match with that sent in notification
@ -188,7 +190,16 @@ class Transport(Module):
"""
return user.name
def getUDSTransportScript(self, userService, transport, ip, os, user, password, request) -> typing.Tuple[str, str, typing.Dict[str, str]]:
def getUDSTransportScript(
self,
userService: 'models.UserService',
transport: 'models.Transport',
ip: str,
os: typing.Dict[str, str],
user: 'models.User',
password: str,
request: 'HttpRequest'
) -> typing.Tuple[str, str, typing.Dict[str, str]]:
"""
If this is an uds transport, this will return the tranport script needed for executing
this on client
@ -210,20 +221,38 @@ class Transport(Module):
'O6acQZmbjBCqZoo9Qsg7k9cTcalNkc5flEYAk1mULnddgDM6'\
'YGmoJgVnDr0=', {'transport': transport.name}
def getEncodedTransportScript(self, userService, transport, ip, os, user, password, request):
def getEncodedTransportScript(
self,
userService: 'models.UserService',
transport: 'models.Transport',
ip: str,
os: typing.Dict[str, str],
user: 'models.User',
password: str,
request: 'HttpRequest'
) -> typing.Tuple[str, str, typing.Dict[str, str]]:
"""
Encodes the script so the client can understand it
"""
script, signature, params = self.getUDSTransportScript(userService, transport, ip, os, user, password, request)
logger.debug('Transport script: {}'.format(script))
return encoders.encode(encoders.encode(script, 'bz2'), 'base64', asText=True).replace('\n', ''), signature, params
logger.debug('Transport script: %s', script)
return typing.cast(str, encoders.encode(encoders.encode(script, 'bz2'), 'base64', asText=True)).replace('\n', ''), signature, params
def getLink(self, userService, transport, ip, os, user, password, request):
def getLink(
self,
userService: 'models.UserService',
transport: 'models.Transport',
ip: str,
os: typing.Dict[str, str],
user: 'models.User',
password: str,
request: 'HttpRequest'
) -> str:
"""
Must override if transport does provides its own link
If transport provides own link, this method provides the link itself
"""
return None
return 'https://www.udsenterprise.com'
def __str__(self):
return "Base OS Manager"
return 'Base OS Manager'

25
server/src/uds/core/transports/TransportsFactory.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012 Virtual Cable S.L.
# Copyright (c) 2012-2019 Virtual Cable S.L.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -30,24 +30,25 @@
"""
@author: Adolfo Gómez, dkmaster at dkmon dot com
"""
from __future__ import unicode_literals
import logging
import typing
if typing.TYPE_CHECKING:
from .BaseTransport import Transport
logger = logging.getLogger(__name__)
class TransportsFactory(object):
_factory = None
class TransportsFactory:
_factory: typing.Optional['TransportsFactory'] = None
_transports: typing.Dict[str, typing.Type['Transport']]
def __init__(self):
self._transports = {}
@staticmethod
def factory():
"""
Singleton getter
"""
def factory() -> 'TransportsFactory':
if TransportsFactory._factory is None:
TransportsFactory._factory = TransportsFactory()
return TransportsFactory._factory
@ -55,10 +56,10 @@ class TransportsFactory(object):
def providers(self):
return self._transports
def insert(self, type_):
logger.debug('Adding transport {0} as {1}'.format(type_.type(), type_))
def insert(self, type_: typing.Type['Transport']) -> None:
logger.debug('Adding transport %s as %s', type_.type(), type_)
typeName = type_.type().lower()
self._transports[typeName] = type_
def lookup(self, typeName):
def lookup(self, typeName: str) -> typing.Optional[typing.Type['Transport']]:
return self._transports.get(typeName.lower(), None)

4
server/src/uds/core/transports/__init__.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012 Virtual Cable S.L.
# Copyright (c) 2012-2019 Virtual Cable S.L.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -32,8 +32,6 @@ UDS Service modules interfaces and classes.
.. moduleauthor:: Adolfo Gómez, dkmaster at dkmon dot com
"""
from __future__ import unicode_literals
from .BaseTransport import Transport

3
server/src/uds/core/transports/protocols.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012 Virtual Cable S.L.
# Copyright (c) 2012-2019 Virtual Cable S.L.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -30,7 +30,6 @@
"""
@author: Adolfo Gómez, dkmaster at dkmon dot com
"""
from __future__ import unicode_literals
NONE = ''
RDP = 'rdp'

2
server/src/uds/core/ui/images.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012 Virtual Cable S.L.
# Copyright (c) 2012-2019 Virtual Cable S.L.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,

2
server/src/uds/core/ui/theme.py
View File

@ -30,8 +30,6 @@
"""
.. moduleauthor:: Adolfo Gómez, dkmaster at dkmon dot com
"""
from __future__ import unicode_literals
from uds.core.util.Config import GlobalConfig

2
server/src/uds/core/util/connection.py
View File

@ -37,7 +37,7 @@ import socket
logger = logging.getLogger(__name__)
def testServer(host: str, port: typing.Union[int, str], timeOut: int = 4):
def testServer(host: str, port: typing.Union[int, str], timeOut: int = 4) -> bool:
try:
logger.debug('Checking connection to %s:%s with %s seconds timeout', host, port, timeOut)
sock = socket.create_connection((host, int(port)), timeOut)

11
server/src/uds/models/Proxy.py
View File

@ -30,6 +30,7 @@
"""
import json
import logging
import typing
import requests
@ -62,18 +63,18 @@ class Proxy(UUIDModel, TaggingMixin): # type: ignore
app_label = 'uds'
@property
def url(self):
def url(self) -> str:
return 'http{}://{}:{}'.format('s' if self.ssl is True else '', self.host, self.port)
@property
def proxyRequestUrl(self):
def proxyRequestUrl(self) -> str:
return self.url + "/proxyRequest"
@property
def testServerUrl(self):
def testServerUrl(self) -> str:
return self.url + "/testServer"
def doProxyRequest(self, url, data=None, timeout=5):
def doProxyRequest(self, url, data: typing.Optional[typing.Any] = None, timeout: int = 5) -> requests.Response:
d = {
'url': url
}
@ -88,7 +89,7 @@ class Proxy(UUIDModel, TaggingMixin): # type: ignore
timeout=timeout
)
def doTestServer(self, ip, port, timeout=5):
def doTestServer(self, ip: str, port: typing.Union[str, int], timeout=5) -> bool:
try:
url = self.testServerUrl + '?host={}&port={}&timeout={}'.format(ip, port, timeout)
r = requests.get(

8
server/src/uds/transports/RDP/BaseRDPTransport.py
View File

@ -136,8 +136,8 @@ class BaseRDPTransport(Transport):
self.cache.put(ip, 'N', READY_CACHE_TIMEOUT)
return ready == 'Y'
def processedUser(self, userService, userName):
v = self.processUserPassword(userService, userName, '')
def processedUser(self, userService, user):
v = self.processUserPassword(userService, user, '')
return v['username']
def processUserPassword(self, service, user, password):
@ -182,8 +182,8 @@ class BaseRDPTransport(Transport):
return {'protocol': self.protocol, 'username': username, 'password': password, 'domain': domain}
def getConnectionInfo(self, service, user, password):
return self.processUserPassword(service, user, password)
def getConnectionInfo(self, userService, user, password):
return self.processUserPassword(userService, user, password)
def getScript(self, scriptName, osName, params) -> Tuple[str, str, dict]:
# Reads script