shaba/openuds
1
0
Fork 1
mirror of https://github.com/dkmstr/openuds.git synced 2025-03-12 04:58:34 +03:00
Code Issues Releases Wiki Activity

Fixing up LSP on REST classes

Browse Source
This commit is contained in:
Adolfo Gómez García 2023-11-03 04:18:53 +01:00
parent 85e20f1152
commit d4ea0ad8c6
No known key found for this signature in database
GPG Key ID: DD1ABF20724CDA23
30 changed files with 463 additions and 303 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
server/src/uds/REST/methods/accounts.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2017-2019 Virtual Cable S.L.
# Copyright (c) 2017-2023 Virtual Cable S.L.U.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -38,7 +38,7 @@ from django.utils.translation import gettext_lazy as _
from uds.REST.model import ModelHandler
import uds.core.types.permissions
from uds.core.util import permissions
from uds.core.util import permissions, ensure
from uds.models import Account
from .accountsusage import AccountsUsage
@ -70,8 +70,8 @@ class Accounts(ModelHandler):
{'tags': {'title': _('tags'), 'visible': False}},
]
def item_as_dict(self, item_: 'Model'):
item = typing.cast(Account, item_)
def item_as_dict(self, item: 'Model'):
item = ensure.is_instance(item, Account)
return {
'id': item.uuid,
'name': item.name,
@ -84,11 +84,13 @@ class Accounts(ModelHandler):
def getGui(self, type_: str) -> typing.List[typing.Any]:
return self.addDefaultFields([], ['name', 'comments', 'tags'])
def timemark(self, item: Account):
def timemark(self, item: 'Model') -> typing.Any:
item = ensure.is_instance(item, Account)
item.time_mark = datetime.datetime.now()
item.save()
return ''
def clear(self, item: Account):
def clear(self, item: 'Model') -> typing.Any:
item = ensure.is_instance(item, Account)
self.ensureAccess(item, uds.core.types.permissions.PermissionType.MANAGEMENT)
return item.usages.filter(user_service=None).delete()

22
server/src/uds/REST/methods/accountsusage.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2017-2019 Virtual Cable S.L.
# Copyright (c) 2017-2023 Virtual Cable S.L.U.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -35,14 +35,15 @@ import typing
from django.utils.translation import gettext as _
from uds.core.util import permissions
from uds.core.util import ensure, permissions
from uds.core.util.model import processUuid
from uds.models import Account, AccountUsage
from uds.REST import RequestError
from uds.REST.model import DetailHandler
# Not imported at runtime, just for type checking
if typing.TYPE_CHECKING:
from uds.models import Account, AccountUsage
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -75,7 +76,8 @@ class AccountsUsage(DetailHandler): # pylint: disable=too-many-public-methods
return retVal
def getItems(self, parent: 'Account', item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, Account)
# Check what kind of access do we have to parent provider
perm = permissions.getEffectivePermission(self._user, parent)
try:
@ -87,7 +89,7 @@ class AccountsUsage(DetailHandler): # pylint: disable=too-many-public-methods
logger.exception('itemId %s', item)
raise self.invalidItemException()
def getFields(self, parent: 'Account') -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
{'pool_name': {'title': _('Pool name')}},
{'user_name': {'title': _('User name')}},
@ -98,13 +100,14 @@ class AccountsUsage(DetailHandler): # pylint: disable=too-many-public-methods
{'elapsed_timemark': {'title': _('Elapsed timemark')}},
]
def getRowStyle(self, parent: 'Account') -> typing.Dict[str, typing.Any]:
def getRowStyle(self, parent: 'Model') -> typing.Dict[str, typing.Any]:
return {'field': 'running', 'prefix': 'row-running-'}
def saveItem(self, parent: 'Account', item: typing.Optional[str]) -> None:
def saveItem(self, parent: 'Model', item: typing.Optional[str]) -> None:
raise RequestError('Accounts usage cannot be edited')
def deleteItem(self, parent: 'Account', item: str) -> None:
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, Account)
logger.debug('Deleting account usage %s from %s', item, parent)
try:
usage = parent.usages.get(uuid=processUuid(item))
@ -113,7 +116,8 @@ class AccountsUsage(DetailHandler): # pylint: disable=too-many-public-methods
logger.exception('Exception')
raise self.invalidItemException()
def getTitle(self, parent: 'Account') -> str:
def getTitle(self, parent: 'Model') -> str:
parent = ensure.is_instance(parent, Account)
try:
return _('Usages of {0}').format(parent.name)
except Exception:

8
server/src/uds/REST/methods/actor_token.py
View File

@ -42,6 +42,9 @@ from uds.models import Server
from uds.REST.exceptions import NotFound, RequestError
from uds.REST.model import ModelHandler
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
# Enclosed methods under /osm path
@ -51,7 +54,7 @@ class ActorTokens(ModelHandler):
model = Server
model_filter = {'type': types.servers.ServerType.ACTOR}
table_title = _('Actor tokens')
table_title = typing.cast(str, _('Actor tokens'))
table_fields = [
# {'token': {'title': _('Token')}},
{'stamp': {'title': _('Date'), 'type': 'datetime'}},
@ -64,7 +67,8 @@ class ActorTokens(ModelHandler):
{'log_level': {'title': _('Log level')}},
]
def item_as_dict(self, item: Server) -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item_: 'Model') -> typing.Dict[str, typing.Any]:
item = typing.cast(Server, item_)
data = item.data or {}
log_level_int = data.get('log_level', 2)
if log_level_int < 10000: # Old log level

33
server/src/uds/REST/methods/authenticators.py
View File

@ -41,7 +41,7 @@ from uds.core.environment import Environment
from uds.REST import NotFound
from uds.REST.model import ModelHandler
from uds.core.util import permissions
from uds.core.util import permissions, ensure
from uds.core.util.model import processUuid
from uds.core.ui import gui
@ -49,7 +49,7 @@ from .users_groups import Users, Groups
# Not imported at runtime, just for type checking
if typing.TYPE_CHECKING:
from django.db import models
from django.db.models import Model
from uds.core.module import Module
logger = logging.getLogger(__name__)
@ -134,17 +134,12 @@ class Authenticators(ModelHandler):
field,
{
'name': 'mfa_id',
'choices': [gui.choiceItem('', _('None'))]
'choices': [gui.choiceItem('', typing.cast(str, _('None')))]
+ gui.sortedChoices(
[
gui.choiceItem(v.uuid or '', v.name)
for v in MFA.objects.all()
]
[gui.choiceItem(v.uuid or '', v.name) for v in MFA.objects.all()]
),
'label': gettext('MFA Provider'),
'tooltip': gettext(
'MFA provider to use for this authenticator'
),
'tooltip': gettext('MFA provider to use for this authenticator'),
'type': types.ui.FieldType.CHOICE,
'order': 108,
'tab': types.ui.Tab.MFA,
@ -156,7 +151,8 @@ class Authenticators(ModelHandler):
logger.info('Type not found: %s', e)
raise NotFound('type not found') from e
def item_as_dict(self, item: Authenticator) -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, Authenticator)
type_ = item.getType()
return {
'numeric_id': item.id,
@ -177,21 +173,21 @@ class Authenticators(ModelHandler):
'permission': permissions.getEffectivePermission(self._user, item),
}
def afterSave(self, item: Authenticator) -> None:
def afterSave(self, item: 'Model') -> None:
item = ensure.is_instance(item, Authenticator)
try:
networks = self._params['networks']
except Exception: # No networks passed in, this is ok
logger.debug('No networks')
return
if (
networks is None
): # None is not provided, empty list is ok and means no networks
if networks is None: # None is not provided, empty list is ok and means no networks
return
logger.debug('Networks: %s', networks)
item.networks.set(Network.objects.filter(uuid__in=networks)) # type: ignore # set is not part of "queryset"
# Custom "search" method
def search(self, item: Authenticator) -> typing.List[typing.Dict]:
def search(self, item: 'Model') -> typing.List[typing.Dict]:
item = ensure.is_instance(item, Authenticator)
self.ensureAccess(item, types.permissions.PermissionType.READ)
try:
type_ = self._params['type']
@ -250,11 +246,12 @@ class Authenticators(ModelHandler):
# And ensure small_name chars are valid [a-zA-Z0-9:-]+
if fields['small_name'] and not re.match(r'^[a-zA-Z0-9:.-]+$', fields['small_name']):
raise self.invalidRequestException(
_('Label must contain only letters, numbers, or symbols: - : .')
typing.cast(str, _('Label must contain only letters, numbers, or symbols: - : .'))
)
def deleteItem(self, item: Authenticator):
def deleteItem(self, item: 'Model'):
# For every user, remove assigned services (mark them for removal)
item = ensure.is_instance(item, Authenticator)
for user in item.users.all():
for userService in user.userServices.all():

32
server/src/uds/REST/methods/calendarrules.py
View File

@ -34,20 +34,19 @@ import datetime
import logging
import typing
from django.utils.translation import gettext as _
from django.db import IntegrityError
from django.utils.translation import gettext as _
from uds.models.calendar_rule import freqs, CalendarRule
from uds.core.util.model import getSqlDatetime
from uds.core.util import permissions
from uds.core.util.model import processUuid
from uds.REST.model import DetailHandler
from uds.core.util import ensure, permissions
from uds.core.util.model import getSqlDatetime, processUuid
from uds.models.calendar_rule import CalendarRule, freqs
from uds.REST import RequestError
from uds.REST.model import DetailHandler
# Not imported at runtime, just for type checking
if typing.TYPE_CHECKING:
from django.db.models import Model
from uds.models import Calendar
logger = logging.getLogger(__name__)
@ -80,7 +79,8 @@ class CalendarRules(DetailHandler): # pylint: disable=too-many-public-methods
return retVal
def getItems(self, parent: 'Calendar', item: typing.Optional[str]) -> typing.Any:
def getItems(self, parent: 'Model', item: typing.Optional[str]) -> typing.Any:
parent = ensure.is_instance(parent, Calendar)
# Check what kind of access do we have to parent provider
perm = permissions.getEffectivePermission(self._user, parent)
try:
@ -92,7 +92,9 @@ class CalendarRules(DetailHandler): # pylint: disable=too-many-public-methods
logger.exception('itemId %s', item)
raise self.invalidItemException() from e
def getFields(self, parent: 'Calendar') -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
parent = ensure.is_instance(parent, Calendar)
return [
{'name': {'title': _('Rule name')}},
{'start': {'title': _('Starts'), 'type': 'datetime'}},
@ -109,7 +111,9 @@ class CalendarRules(DetailHandler): # pylint: disable=too-many-public-methods
{'comments': {'title': _('Comments')}},
]
def saveItem(self, parent: 'Calendar', item: typing.Optional[str]) -> None:
def saveItem(self, parent: 'Model', item: typing.Optional[str]) -> None:
parent = ensure.is_instance(parent, Calendar)
# Extract item db fields
# We need this fields for all
logger.debug('Saving rule %s / %s', parent, item)
@ -150,7 +154,8 @@ class CalendarRules(DetailHandler): # pylint: disable=too-many-public-methods
logger.exception('Saving calendar')
raise RequestError(f'incorrect invocation to PUT: {e}') from e
def deleteItem(self, parent: 'Calendar', item: str) -> None:
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, Calendar)
logger.debug('Deleting rule %s from %s', item, parent)
try:
calRule = parent.rules.get(uuid=processUuid(item))
@ -161,7 +166,8 @@ class CalendarRules(DetailHandler): # pylint: disable=too-many-public-methods
logger.exception('Exception')
raise self.invalidItemException() from e
def getTitle(self, parent: 'Calendar') -> str:
def getTitle(self, parent: 'Model') -> str:
parent = ensure.is_instance(parent, Calendar)
try:
return _('Rules of {0}').format(parent.name)
except Exception:

11
server/src/uds/REST/methods/calendars.py
View File

@ -35,11 +35,14 @@ import typing
from django.utils.translation import gettext_lazy as _
from uds.models import Calendar
from uds.core.util import permissions
from uds.core.util import permissions, ensure
from uds.REST.model import ModelHandler
from .calendarrules import CalendarRules
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -56,7 +59,7 @@ class Calendars(ModelHandler):
save_fields = ['name', 'comments', 'tags']
table_title = _('Calendars')
table_title = typing.cast(str, _('Calendars'))
table_fields = [
{
'name': {
@ -74,7 +77,8 @@ class Calendars(ModelHandler):
{'tags': {'title': _('tags'), 'visible': False}},
]
def item_as_dict(self, item: Calendar) -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, Calendar)
return {
'id': item.uuid,
'name': item.name,
@ -84,7 +88,6 @@ class Calendars(ModelHandler):
'number_rules': item.rules.count(),
'number_access': item.calendaraccess_set.all().values('service_pool').distinct().count(),
'number_actions': item.calendaraction_set.all().values('service_pool').distinct().count(),
'permission': permissions.getEffectivePermission(self._user, item),
}

14
server/src/uds/REST/methods/images.py
View File

@ -37,9 +37,12 @@ from django.utils.translation import gettext_lazy as _, gettext
from uds.models import Image
from uds.core import types
from uds.core.ui import gui
from uds.core.util import ensure
from uds.REST.model import ModelHandler
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -55,7 +58,7 @@ class Images(ModelHandler):
model = Image
save_fields = ['name', 'data']
table_title = _('Image Gallery')
table_title = typing.cast(str, _('Image Gallery'))
table_fields = [
{
'thumb': {
@ -72,7 +75,8 @@ class Images(ModelHandler):
def beforeSave(self, fields: typing.Dict[str, typing.Any]) -> None:
fields['data'] = Image.prepareForDb(Image.decode64(fields['data']))[2]
def afterSave(self, item: Image) -> None:
def afterSave(self, item: 'Model') -> None:
item = ensure.is_instance(item, Image)
# Updates the thumbnail and re-saves it
logger.debug('After save: item = %s', item)
item.updateThumbnail()
@ -91,14 +95,16 @@ class Images(ModelHandler):
},
)
def item_as_dict(self, item: Image) -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, Image)
return {
'id': item.uuid,
'name': item.name,
'data': item.data64,
}
def item_as_dict_overview(self, item: Image) -> typing.Dict[str, typing.Any]:
def item_as_dict_overview(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, Image)
return {
'id': item.uuid,
'size': '{}x{}, {} bytes (thumb {} bytes)'.format(

67
server/src/uds/REST/methods/meta_pools.py
View File

@ -37,9 +37,9 @@ from django.utils.translation import gettext
from django.utils.translation import gettext_lazy as _
from uds.core import types
from uds.core.ui import gui
from uds.core.consts.images import DEFAULT_THUMB_BASE64
from uds.core.util import permissions
from uds.core.ui import gui
from uds.core.util import ensure, permissions
from uds.core.util.model import processUuid
from uds.core.util.state import State
from uds.models import Image, MetaPool, ServicePoolGroup
@ -50,6 +50,9 @@ from uds.REST.model import ModelHandler
from .meta_service_pools import MetaAssignedService, MetaServicesPool
from .user_services import Groups
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -80,12 +83,24 @@ class MetaPools(ModelHandler):
'transport_grouping',
]
table_title = _('Meta Pools')
table_title = typing.cast(str, _('Meta Pools'))
table_fields = [
{'name': {'title': _('Name')}},
{'comments': {'title': _('Comments')}},
{'policy': {'title': _('Policy'), 'type': 'dict', 'dict': dict(types.pools.LoadBalancingPolicy.enumerate())}},
{'ha_policy': {'title': _('HA Policy'), 'type': 'dict', 'dict': dict(types.pools.HighAvailabilityPolicy.enumerate())}},
{
'policy': {
'title': _('Policy'),
'type': 'dict',
'dict': dict(types.pools.LoadBalancingPolicy.enumerate()),
}
},
{
'ha_policy': {
'title': _('HA Policy'),
'type': 'dict',
'dict': dict(types.pools.HighAvailabilityPolicy.enumerate()),
}
},
{'user_services_count': {'title': _('User services'), 'type': 'number'}},
{'user_services_in_preparation': {'title': _('In Preparation')}},
{'visible': {'title': _('Visible'), 'type': 'callback'}},
@ -96,11 +111,12 @@ class MetaPools(ModelHandler):
custom_methods = [('setFallbackAccess', True), ('getFallbackAccess', True)]
def item_as_dict(self, item: MetaPool) -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, MetaPool)
# if item does not have an associated service, hide it (the case, for example, for a removed service)
# Access from dict will raise an exception, and item will be skipped
poolGroupId = None
poolGroupName = _('Default')
poolGroupName = typing.cast(str, _('Default'))
poolGroupThumb = DEFAULT_THUMB_BASE64
if item.servicesPoolGroup is not None:
poolGroupId = item.servicesPoolGroup.uuid
@ -110,14 +126,10 @@ class MetaPools(ModelHandler):
allPools = item.members.all()
userServicesCount = sum(
(
i.pool.userServices.exclude(state__in=State.INFO_STATES).count()
for i in allPools
)
(i.pool.userServices.exclude(state__in=State.INFO_STATES).count() for i in allPools)
)
userServicesInPreparation = sum(
(i.pool.userServices.filter(state=State.PREPARING).count())
for i in allPools
(i.pool.userServices.filter(state=State.PREPARING).count()) for i in allPools
)
val = {
@ -126,9 +138,7 @@ class MetaPools(ModelHandler):
'short_name': item.short_name,
'tags': [tag.tag for tag in item.tags.all()],
'comments': item.comments,
'thumb': item.image.thumb64
if item.image is not None
else DEFAULT_THUMB_BASE64,
'thumb': item.image.thumb64 if item.image is not None else DEFAULT_THUMB_BASE64,
'image_id': item.image.uuid if item.image is not None else None,
'servicesPoolGroup_id': poolGroupId,
'pool_group_name': poolGroupName,
@ -162,9 +172,7 @@ class MetaPools(ModelHandler):
},
{
'name': 'policy',
'choices': [
gui.choiceItem(k, str(v)) for k, v in types.pools.LoadBalancingPolicy.enumerate()
],
'choices': [gui.choiceItem(k, str(v)) for k, v in types.pools.LoadBalancingPolicy.enumerate()],
'label': gettext('Load balancing policy'),
'tooltip': gettext('Service pool load balancing policy'),
'type': types.ui.FieldType.CHOICE,
@ -176,7 +184,9 @@ class MetaPools(ModelHandler):
gui.choiceItem(k, str(v)) for k, v in types.pools.HighAvailabilityPolicy.enumerate()
],
'label': gettext('HA Policy'),
'tooltip': gettext('Service pool High Availability policy. If enabled and a pool fails, it will be restarted in another pool. Enable with care!.'),
'tooltip': gettext(
'Service pool High Availability policy. If enabled and a pool fails, it will be restarted in another pool. Enable with care!.'
),
'type': types.ui.FieldType.CHOICE,
'order': 101,
},
@ -184,10 +194,7 @@ class MetaPools(ModelHandler):
'name': 'image_id',
'choices': [gui.choiceImage(-1, '--------', DEFAULT_THUMB_BASE64)]
+ gui.sortedChoices(
[
gui.choiceImage(v.uuid, v.name, v.thumb64) # type: ignore
for v in Image.objects.all()
]
[gui.choiceImage(v.uuid, v.name, v.thumb64) for v in Image.objects.all()] # type: ignore
),
'label': gettext('Associated Image'),
'tooltip': gettext('Image assocciated with this service'),
@ -197,7 +204,7 @@ class MetaPools(ModelHandler):
},
{
'name': 'servicesPoolGroup_id',
'choices': [gui.choiceImage(-1, _('Default'), DEFAULT_THUMB_BASE64)]
'choices': [gui.choiceImage(-1, typing.cast(str, _('Default')), DEFAULT_THUMB_BASE64)]
+ gui.sortedChoices(
[
gui.choiceImage(v.uuid, v.name, v.thumb64) # type: ignore
@ -205,9 +212,7 @@ class MetaPools(ModelHandler):
]
),
'label': gettext('Pool group'),
'tooltip': gettext(
'Pool group for this pool (for pool classify on display)'
),
'tooltip': gettext('Pool group for this pool (for pool classify on display)'),
'type': types.ui.FieldType.IMAGECHOICE,
'order': 121,
'tab': types.ui.Tab.DISPLAY,
@ -235,8 +240,7 @@ class MetaPools(ModelHandler):
{
'name': 'transport_grouping',
'choices': [
gui.choiceItem(k, str(v))
for k, v in types.pools.TransportSelectionPolicy.enumerate()
gui.choiceItem(k, str(v)) for k, v in types.pools.TransportSelectionPolicy.enumerate()
],
'label': gettext('Transport Selection'),
'tooltip': gettext('Transport selection policy'),
@ -281,7 +285,8 @@ class MetaPools(ModelHandler):
logger.debug('Fields: %s', fields)
def deleteItem(self, item: MetaPool) -> None:
def deleteItem(self, item: 'Model') -> None:
item = ensure.is_instance(item, MetaPool)
item.delete()
# Set fallback status

41
server/src/uds/REST/methods/meta_service_pools.py
View File

@ -43,10 +43,13 @@ from uds import models
from uds.core.util.state import State
from uds.core.util.model import processUuid
from uds.core.util import log
from uds.core.util import log, ensure
from uds.REST.model import DetailHandler
from .user_services import AssignedService
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -56,7 +59,7 @@ class MetaServicesPool(DetailHandler):
"""
@staticmethod
def as_dict(item: models.MetaPoolMember):
def as_dict(item: models.MetaPoolMember) -> typing.Dict[str, typing.Any]:
return {
'id': item.uuid,
'pool_id': item.pool.uuid,
@ -68,7 +71,8 @@ class MetaServicesPool(DetailHandler):
'user_services_in_preparation': item.pool.userServices.filter(state=State.PREPARING).count(),
}
def getItems(self, parent: models.MetaPool, item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, models.MetaPool)
try:
if not item:
return [MetaServicesPool.as_dict(i) for i in parent.members.all()]
@ -78,17 +82,18 @@ class MetaServicesPool(DetailHandler):
logger.exception('err: %s', item)
raise self.invalidItemException()
def getTitle(self, parent: models.MetaPool) -> str:
def getTitle(self, parent: 'Model') -> str:
return _('Service pools')
def getFields(self, parent: models.MetaPool) -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
{'priority': {'title': _('Priority'), 'type': 'numeric', 'width': '6em'}},
{'name': {'title': _('Service Pool name')}},
{'enabled': {'title': _('Enabled')}},
]
def saveItem(self, parent: models.MetaPool, item: typing.Optional[str]):
def saveItem(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, models.MetaPool)
# If already exists
uuid = processUuid(item) if item else None
@ -116,7 +121,8 @@ class MetaServicesPool(DetailHandler):
return self.success()
def deleteItem(self, parent: models.MetaPool, item: str):
def deleteItem(self, parent: 'Model', item: str):
parent = ensure.is_instance(parent, models.MetaPool)
member = parent.members.get(uuid=processUuid(self._args[0]))
logStr = "Removed meta pool member {} by {}".format(member.pool.name, self._user.pretty_name)
@ -155,7 +161,8 @@ class MetaAssignedService(DetailHandler):
except Exception:
raise self.invalidItemException()
def getItems(self, parent: models.MetaPool, item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, models.MetaPool)
def assignedUserServicesForPools() -> (
typing.Generator[
typing.Tuple[models.UserService, typing.Optional[typing.Dict[str, typing.Any]]], None, None
@ -198,10 +205,12 @@ class MetaAssignedService(DetailHandler):
logger.exception('getItems')
raise self.invalidItemException()
def getTitle(self, parent: 'models.MetaPool') -> str:
def getTitle(self, parent: 'Model') -> str:
parent = ensure.is_instance(parent, models.MetaPool)
return _('Assigned services')
def getFields(self, parent: 'models.MetaPool') -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
parent = ensure.is_instance(parent, models.MetaPool)
return [
{'creation_date': {'title': _('Creation date'), 'type': 'datetime'}},
{'pool_name': {'title': _('Pool')}},
@ -222,10 +231,12 @@ class MetaAssignedService(DetailHandler):
{'actor_version': {'title': _('Actor version')}},
]
def getRowStyle(self, parent: 'models.MetaPool') -> typing.Dict[str, typing.Any]:
def getRowStyle(self, parent: 'Model') -> typing.Dict[str, typing.Any]:
parent = ensure.is_instance(parent, models.MetaPool)
return {'field': 'state', 'prefix': 'row-state-'}
def getLogs(self, parent: 'models.MetaPool', item: str) -> typing.List[typing.Any]:
def getLogs(self, parent: 'Model', item: str) -> typing.List[typing.Any]:
parent = ensure.is_instance(parent, models.MetaPool)
try:
asignedService = self._getAssignedService(parent, item)
logger.debug('Getting logs for %s', asignedService)
@ -233,7 +244,8 @@ class MetaAssignedService(DetailHandler):
except Exception:
raise self.invalidItemException()
def deleteItem(self, parent: 'models.MetaPool', item: str) -> None:
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, models.MetaPool)
userService = self._getAssignedService(parent, item)
if userService.user:
@ -257,7 +269,8 @@ class MetaAssignedService(DetailHandler):
log.doLog(parent, log.LogLevel.INFO, logStr, log.LogSource.ADMIN)
# Only owner is allowed to change right now
def saveItem(self, parent: 'models.MetaPool', item: typing.Optional[str]):
def saveItem(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, models.MetaPool)
if item is None:
raise self.invalidItemException()

27
server/src/uds/REST/methods/mfas.py
View File

@ -33,14 +33,17 @@
import logging
import typing
from django.utils.translation import gettext_lazy as _, gettext
from django.utils.translation import gettext
from django.utils.translation import gettext_lazy as _
from uds import models
from uds.core import mfas, types
from uds.core.environment import Environment
from uds.core.util import permissions
from uds.core.util import ensure, permissions
from uds.REST.model import ModelHandler
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -51,7 +54,7 @@ class MFA(ModelHandler):
model = models.MFA
save_fields = ['name', 'comments', 'tags', 'remember_device', 'validity']
table_title = _('Multi Factor Authentication')
table_title = typing.cast(str, _('Multi Factor Authentication'))
table_fields = [
{'name': {'title': _('Name'), 'visible': True, 'type': 'iconType'}},
{'type_name': {'title': _('Type')}},
@ -71,9 +74,7 @@ class MFA(ModelHandler):
# Create a temporal instance to get the gui
mfa = mfaType(Environment.getTempEnv(), None)
localGui = self.addDefaultFields(
mfa.guiDescription(), ['name', 'comments', 'tags']
)
localGui = self.addDefaultFields(mfa.guiDescription(), ['name', 'comments', 'tags'])
self.addField(
localGui,
{
@ -81,9 +82,7 @@ class MFA(ModelHandler):
'value': '0',
'minValue': '0',
'label': gettext('Device Caching'),
'tooltip': gettext(
'Time in hours to cache device so MFA is not required again. User based.'
),
'tooltip': gettext('Time in hours to cache device so MFA is not required again. User based.'),
'type': types.ui.FieldType.NUMERIC,
'order': 111,
},
@ -95,18 +94,16 @@ class MFA(ModelHandler):
'value': '5',
'minValue': '0',
'label': gettext('MFA code validity'),
'tooltip': gettext(
'Time in minutes to allow MFA code to be used.'
),
'tooltip': gettext('Time in minutes to allow MFA code to be used.'),
'type': types.ui.FieldType.NUMERIC,
'order': 112,
},
)
return localGui
def item_as_dict(self, item: models.MFA) -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, models.MFA)
type_ = item.getType()
return {
'id': item.uuid,

12
server/src/uds/REST/methods/networks.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2014-2019 Virtual Cable S.L.
# Copyright (c) 2014-2023 Virtual Cable S.L.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -37,10 +37,13 @@ from django.utils.translation import gettext_lazy as _, gettext
from uds.models import Network
from uds.core import types
from uds.core.util import permissions
from uds.core.util import permissions, ensure
from ..model import ModelHandler
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
# Enclosed methods under /item path
@ -55,7 +58,7 @@ class Networks(ModelHandler):
model = Network
save_fields = ['name', 'net_string', 'tags']
table_title = _('Networks')
table_title = typing.cast(str, _('Networks'))
table_fields = [
{
'name': {
@ -98,7 +101,8 @@ class Networks(ModelHandler):
},
)
def item_as_dict(self, item: Network) -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, Network)
return {
'id': item.uuid,
'name': item.name,

21
server/src/uds/REST/methods/notifiers.py
View File

@ -33,15 +33,19 @@
import logging
import typing
from django.utils.translation import gettext_lazy as _, gettext
from uds.core.environment import Environment
from uds.models import Notifier, LogLevel
from uds.core import messaging, types
from uds.core.ui import gui
from uds.core.util import permissions
from django.utils.translation import gettext
from django.utils.translation import gettext_lazy as _
from uds.core import messaging, types
from uds.core.environment import Environment
from uds.core.ui import gui
from uds.core.util import ensure, permissions
from uds.models import LogLevel, Notifier
from uds.REST.model import ModelHandler
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -58,7 +62,7 @@ class Notifiers(ModelHandler):
'tags',
]
table_title = _('Notifiers')
table_title = typing.cast(str, _('Notifiers'))
table_fields = [
{'name': {'title': _('Name'), 'visible': True, 'type': 'iconType'}},
{'type_name': {'title': _('Type')}},
@ -96,7 +100,8 @@ class Notifiers(ModelHandler):
return localGui
def item_as_dict(self, item: Notifier) -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, Notifier)
type_ = item.getType()
return {
'id': item.uuid,

35
server/src/uds/REST/methods/op_calendars.py
View File

@ -37,15 +37,15 @@ import typing
from django.utils.translation import gettext as _
from uds.core import types
from uds.core.util import log
from uds.core.util import log, ensure
from uds.core.util.model import processUuid
from uds.models import Calendar, CalendarAction
from uds.models import Calendar, CalendarAction, CalendarAccess, ServicePool
from uds.models.calendar_action import CALENDAR_ACTION_DICT
from uds.REST.model import DetailHandler
# Not imported at runtime, just for type checking
if typing.TYPE_CHECKING:
from uds.models import CalendarAccess, ServicePool
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -64,7 +64,8 @@ class AccessCalendars(DetailHandler):
'priority': item.priority,
}
def getItems(self, parent: 'ServicePool', item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, ServicePool)
try:
if not item:
return [AccessCalendars.as_dict(i) for i in parent.calendarAccess.all()]
@ -75,17 +76,18 @@ class AccessCalendars(DetailHandler):
logger.exception('err: %s', item)
raise self.invalidItemException() from e
def getTitle(self, parent: 'ServicePool'):
def getTitle(self, parent: 'Model'):
return _('Access restrictions by calendar')
def getFields(self, parent: 'ServicePool') -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
{'priority': {'title': _('Priority'), 'type': 'numeric', 'width': '6em'}},
{'calendar': {'title': _('Calendar')}},
{'access': {'title': _('Access')}},
]
def saveItem(self, parent: 'ServicePool', item: typing.Optional[str]) -> None:
def saveItem(self, parent: 'Model', item: typing.Optional[str]) -> None:
parent = ensure.is_instance(parent, ServicePool)
# If already exists
uuid = processUuid(item) if item is not None else None
@ -121,7 +123,8 @@ class AccessCalendars(DetailHandler):
log.LogSource.ADMIN,
)
def deleteItem(self, parent: 'ServicePool', item: str) -> None:
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, ServicePool)
calendarAccess = parent.calendarAccess.get(uuid=processUuid(self._args[0]))
logStr = f'Removed access calendar {calendarAccess.calendar.name} by {self._user.pretty_name}'
calendarAccess.delete()
@ -156,7 +159,8 @@ class ActionsCalendars(DetailHandler):
'lastExecution': item.last_execution,
}
def getItems(self, parent: 'ServicePool', item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, ServicePool)
try:
if item is None:
return [
@ -167,10 +171,10 @@ class ActionsCalendars(DetailHandler):
except Exception as e:
raise self.invalidItemException() from e
def getTitle(self, parent: 'ServicePool'):
def getTitle(self, parent: 'Model'):
return _('Scheduled actions')
def getFields(self, parent: 'ServicePool') -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
{'calendar': {'title': _('Calendar')}},
{'actionDescription': {'title': _('Action')}},
@ -181,7 +185,8 @@ class ActionsCalendars(DetailHandler):
{'lastExecution': {'title': _('Last execution'), 'type': 'datetime'}},
]
def saveItem(self, parent: 'ServicePool', item: typing.Optional[str]) -> None:
def saveItem(self, parent: 'Model', item: typing.Optional[str]) -> None:
parent = ensure.is_instance(parent, ServicePool)
# If already exists
uuid = processUuid(item) if item is not None else None
@ -221,7 +226,8 @@ class ActionsCalendars(DetailHandler):
log.doLog(parent, log.LogLevel.INFO, logStr, log.LogSource.ADMIN)
def deleteItem(self, parent: 'ServicePool', item: str) -> None:
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, ServicePool)
calendarAction = CalendarAction.objects.get(uuid=processUuid(self._args[0]))
logStr = (
f'Removed scheduled action "{calendarAction.calendar.name},'
@ -234,7 +240,8 @@ class ActionsCalendars(DetailHandler):
log.doLog(parent, log.LogLevel.INFO, logStr, log.LogSource.ADMIN)
def execute(self, parent: 'ServicePool', item: str):
def execute(self, parent: 'Model', item: str):
parent = ensure.is_instance(parent, ServicePool)
logger.debug('Launching action')
uuid = processUuid(item)
calendarAction: CalendarAction = CalendarAction.objects.get(uuid=uuid)

13
server/src/uds/REST/methods/osmanagers.py
View File

@ -37,11 +37,14 @@ from django.utils.translation import gettext, gettext_lazy as _
from uds.core import osmanagers
from uds.core.environment import Environment
from uds.core.util import permissions
from uds.core.util import permissions, ensure
from uds.models import OSManager
from uds.REST import NotFound, RequestError
from uds.REST.model import ModelHandler
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
# Enclosed methods under /osm path
@ -51,7 +54,7 @@ class OsManagers(ModelHandler):
model = OSManager
save_fields = ['name', 'comments', 'tags']
table_title = _('OS Managers')
table_title = typing.cast(str, _('OS Managers'))
table_fields = [
{'name': {'title': _('Name'), 'visible': True, 'type': 'iconType'}},
{'type_name': {'title': _('Type')}},
@ -74,10 +77,12 @@ class OsManagers(ModelHandler):
'permission': permissions.getEffectivePermission(self._user, osm),
}
def item_as_dict(self, item: OSManager) -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, OSManager)
return self.osmToDict(item)
def checkDelete(self, item: OSManager) -> None:
def checkDelete(self, item: 'Model') -> None:
item = ensure.is_instance(item, OSManager)
# Only can delete if no ServicePools attached
if item.deployedServices.count() > 0:
raise RequestError(

9
server/src/uds/REST/methods/permissions.py
View File

@ -32,15 +32,12 @@
"""
import logging
import typing
import uds.core.types.permissions
import uds.core.types.permissions
from uds import models
from uds.core.util import permissions
from uds.core.util.rest.tools import match
from uds import models
from uds.REST import Handler
from uds.REST import RequestError
from uds.REST import Handler, RequestError
# Not imported at runtime, just for type checking
if typing.TYPE_CHECKING:

22
server/src/uds/REST/methods/providers.py
View File

@ -33,15 +33,15 @@
import logging
import typing
from django.utils.translation import gettext, gettext_lazy as _
from uds.core.environment import Environment
from django.utils.translation import gettext
from django.utils.translation import gettext_lazy as _
import uds.core.types.permissions
from uds.models import Provider, Service, UserService
from uds.core import services
from uds.core.environment import Environment
from uds.core.util import ensure, permissions
from uds.core.util.state import State
from uds.core.util import permissions
from uds.models import Provider, Service, UserService
from uds.REST import NotFound, RequestError
from uds.REST.model import ModelHandler
@ -51,7 +51,7 @@ from .services_usage import ServicesUsage
logger = logging.getLogger(__name__)
if typing.TYPE_CHECKING:
from django.db import models
from django.db.models import Model
class Providers(ModelHandler):
@ -66,7 +66,7 @@ class Providers(ModelHandler):
save_fields = ['name', 'comments', 'tags']
table_title = _('Service providers')
table_title = typing.cast(str, _('Service providers'))
# Table info fields
table_fields = [
@ -111,7 +111,8 @@ class Providers(ModelHandler):
'permission': permissions.getEffectivePermission(self._user, item),
}
def checkDelete(self, item: Provider) -> None:
def checkDelete(self, item: 'Model') -> None:
item = ensure.is_instance(item, Provider)
if item.services.count() > 0:
raise RequestError(gettext('Can\'t delete providers with services'))
@ -152,11 +153,12 @@ class Providers(ModelHandler):
# logger.exception('Exception')
return {}
def maintenance(self, item: Provider) -> typing.Dict:
def maintenance(self, item: 'Model') -> typing.Dict:
"""
Custom method that swaps maintenance mode state for a provider
:param item:
"""
item = ensure.is_instance(item, Provider)
self.ensureAccess(item, uds.core.types.permissions.PermissionType.MANAGEMENT)
item.maintenance_mode = not item.maintenance_mode
item.save()

2
server/src/uds/REST/methods/reports.py
View File

@ -65,7 +65,7 @@ class Reports(model.BaseModelHandler):
needs_admin = True # By default, staff is lower level needed
table_title = _('Available reports')
table_title = typing.cast(str, _('Available reports'))
table_fields = [
{'group': {'title': _('Group')}},
{'name': {'title': _('Name')}},

38
server/src/uds/REST/methods/servers_management.py
View File

@ -37,7 +37,7 @@ from django.utils.translation import gettext_lazy as _
from uds import models
from uds.core import consts, types, ui
from uds.core.util import permissions
from uds.core.util import permissions, ensure
from uds.core.util.model import getSqlDatetime, processUuid
from uds.REST.exceptions import NotFound, RequestError
from uds.REST.model import DetailHandler, ModelHandler
@ -140,15 +140,15 @@ class ServersServers(DetailHandler):
logger.exception('REST servers')
raise self.invalidItemException() from e
def getTitle(self, parent_: 'Model') -> str:
parent = typing.cast('models.ServerGroup', parent_)
def getTitle(self, parent: 'Model') -> str:
parent = ensure.is_instance(parent, models.ServerGroup)
try:
return _('Servers of {0}').format(parent.name)
except Exception:
return str(_('Servers'))
def getFields(self, parent_: 'Model') -> typing.List[typing.Any]:
parent = typing.cast('models.ServerGroup', parent_)
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
parent = ensure.is_instance(parent, models.ServerGroup)
return [
{
'hostname': {
@ -166,12 +166,12 @@ class ServersServers(DetailHandler):
},
]
def getRowStyle(self, parent_: 'Model') -> typing.Dict[str, typing.Any]:
parent = typing.cast('models.ServerGroup', parent_)
def getRowStyle(self, parent: 'Model') -> typing.Dict[str, typing.Any]:
parent = ensure.is_instance(parent, models.ServerGroup)
return {'field': 'maintenance_mode', 'prefix': 'row-maintenance-'}
def getGui(self, parent_: 'Model', forType: str = '') -> typing.List[typing.Any]:
parent = typing.cast('models.ServerGroup', parent_)
def getGui(self, parent: 'Model', forType: str = '') -> typing.List[typing.Any]:
parent = ensure.is_instance(parent, models.ServerGroup)
kind, subkind = parent.server_type, parent.subtype
title = _('of type') + f' {subkind.upper()} {kind.name.capitalize()}'
if kind == types.servers.ServerType.UNMANAGED:
@ -234,8 +234,8 @@ class ServersServers(DetailHandler):
],
)
def saveItem(self, parent_: 'Model', item: typing.Optional[str]) -> None:
parent = typing.cast('models.ServerGroup', parent_)
def saveItem(self, parent: 'Model', item: typing.Optional[str]) -> None:
parent = ensure.is_instance(parent, models.ServerGroup)
# Item is the uuid of the server to add
server: typing.Optional['models.Server'] = None # Avoid warning on reference before assignment
@ -276,8 +276,8 @@ class ServersServers(DetailHandler):
raise self.invalidRequestException() from None
def deleteItem(self, parent_: 'Model', item: str) -> None:
parent = typing.cast('models.ServerGroup', parent_)
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, models.ServerGroup)
try:
server = models.Server.objects.get(uuid=processUuid(item))
if parent.server_type == types.servers.ServerType.UNMANAGED:
@ -289,8 +289,8 @@ class ServersServers(DetailHandler):
raise self.invalidItemException() from None
# Custom methods
def maintenance(self, parent_: 'Model', id: str) -> typing.Any:
parent = typing.cast('models.ServerGroup', parent_)
def maintenance(self, parent: 'Model', id: str) -> typing.Any:
parent = ensure.is_instance(parent, models.ServerGroup)
"""
Custom method that swaps maintenance mode state for a tunnel server
:param item:
@ -368,8 +368,8 @@ class ServersGroups(ModelHandler):
fields['subtype'] = subtype
return super().beforeSave(fields)
def item_as_dict(self, item_: 'Model') -> typing.Dict[str, typing.Any]:
item = typing.cast('models.ServerGroup', item_) # We will receive for sure
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, models.ServerGroup)
return {
'id': item.uuid,
'name': item.name,
@ -382,8 +382,8 @@ class ServersGroups(ModelHandler):
'permission': permissions.getEffectivePermission(self._user, item),
}
def deleteItem(self, item_: 'Model') -> None:
item = typing.cast('models.ServerGroup', item_) # We will receive for sure
def deleteItem(self, item: 'Model') -> None:
item = ensure.is_instance(item, models.ServerGroup)
"""
Processes a DELETE request
"""

34
server/src/uds/REST/methods/services.py
View File

@ -40,8 +40,7 @@ from uds import models
from uds.core import exceptions, types
import uds.core.types.permissions
from uds.core.util import log
from uds.core.util import permissions
from uds.core.util import log, permissions, ensure
from uds.core.util.model import processUuid
from uds.core.environment import Environment
from uds.core.consts.images import DEFAULT_THUMB_BASE64
@ -54,7 +53,7 @@ from uds.REST import NotFound, ResponseError, RequestError, AccessDenied
# Not imported at runtime, just for type checking
if typing.TYPE_CHECKING:
from uds.models import Provider
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -118,7 +117,8 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
return retVal
def getItems(self, parent: 'Provider', item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, models.Provider)
# Check what kind of access do we have to parent provider
perm = permissions.getEffectivePermission(self._user, parent)
try:
@ -131,7 +131,8 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
logger.error('Error getting services for %s: %s', parent, e)
raise self.invalidItemException() from e
def getRowStyle(self, parent: 'Provider') -> typing.Dict[str, typing.Any]:
def getRowStyle(self, parent: 'Model') -> typing.Dict[str, typing.Any]:
parent = ensure.is_instance(parent, models.Provider)
return {'field': 'maintenance_mode', 'prefix': 'row-maintenance-'}
def _deleteIncompleteService(
@ -147,7 +148,8 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
except Exception: # nosec: This is a delete, we don't care about exceptions
pass
def saveItem(self, parent: 'Provider', item: typing.Optional[str]) -> None:
def saveItem(self, parent: 'Model', item: typing.Optional[str]) -> None:
parent = ensure.is_instance(parent, models.Provider)
# Extract item db fields
# We need this fields for all
logger.debug('Saving service for %s / %s', parent, item)
@ -210,7 +212,8 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
logger.exception('Saving Service')
raise RequestError('incorrect invocation to PUT: {0}'.format(e)) from e
def deleteItem(self, parent: 'Provider', item: str) -> None:
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, models.Provider)
try:
service = parent.services.get(uuid=processUuid(item))
if service.deployedServices.count() == 0:
@ -222,13 +225,14 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
raise RequestError('Item has associated deployed services')
def getTitle(self, parent: 'Provider') -> str:
def getTitle(self, parent: 'Model') -> str:
parent = ensure.is_instance(parent, models.Provider)
try:
return _('Services of {}').format(parent.name)
except Exception:
return _('Current services')
def getFields(self, parent: 'Provider') -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
{'name': {'title': _('Service name'), 'visible': True, 'type': 'iconType'}},
{'comments': {'title': _('Comments')}},
@ -251,8 +255,9 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
]
def getTypes(
self, parent: 'Provider', forType: typing.Optional[str]
self, parent: 'Model', forType: typing.Optional[str]
) -> typing.Iterable[typing.Dict[str, typing.Any]]:
parent = ensure.is_instance(parent, models.Provider)
logger.debug('getTypes parameters: %s, %s', parent, forType)
offers: typing.List[typing.Dict[str, typing.Any]] = []
if forType is None:
@ -282,7 +287,8 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
return offers # Default is that details do not have types
def getGui(self, parent: 'Provider', forType: str) -> typing.Iterable[typing.Any]:
def getGui(self, parent: 'Model', forType: str) -> typing.Iterable[typing.Any]:
parent = ensure.is_instance(parent, models.Provider)
try:
logger.debug('getGui parameters: %s, %s', parent, forType)
parentInstance = parent.getInstance()
@ -320,7 +326,8 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
logger.exception('getGui')
raise ResponseError(str(e)) from e
def getLogs(self, parent: 'Provider', item: str) -> typing.List[typing.Any]:
def getLogs(self, parent: 'Model', item: str) -> typing.List[typing.Any]:
parent = ensure.is_instance(parent, models.Provider)
try:
service = parent.services.get(uuid=processUuid(item))
logger.debug('Getting logs for %s', item)
@ -328,7 +335,8 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
except Exception:
raise self.invalidItemException() from None
def servicesPools(self, parent: 'Provider', item: str) -> typing.Any:
def servicesPools(self, parent: 'Model', item: str) -> typing.Any:
parent = ensure.is_instance(parent, models.Provider)
service = parent.services.get(uuid=processUuid(item))
logger.debug('Got parameters for servicepools: %s, %s', parent, item)
res = []

14
server/src/uds/REST/methods/services_pool_groups.py
View File

@ -37,14 +37,18 @@ from django.utils.translation import gettext
from django.utils.translation import gettext_lazy as _
from uds.core import types
from uds.core.ui import gui
from uds.core.consts.images import DEFAULT_THUMB_BASE64
from uds.core.ui import gui
from uds.core.util import ensure
from uds.core.util.model import processUuid
from uds.models import Image, ServicePoolGroup
from uds.REST.model import ModelHandler
logger = logging.getLogger(__name__)
if typing.TYPE_CHECKING:
from django.db.models import Model
# Enclosed methods under /item path
@ -59,7 +63,7 @@ class ServicesPoolGroups(ModelHandler):
model = ServicePoolGroup
save_fields = ['name', 'comments', 'image_id', 'priority']
table_title = _('Services Pool Groups')
table_title = typing.cast(str, _('Services Pool Groups'))
table_fields = [
{'priority': {'title': _('Priority'), 'type': 'numeric', 'width': '6em'}},
{
@ -109,7 +113,8 @@ class ServicesPoolGroups(ModelHandler):
return localGui
def item_as_dict(self, item: ServicePoolGroup) -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, ServicePoolGroup)
return {
'id': item.uuid,
'priority': item.priority,
@ -119,8 +124,9 @@ class ServicesPoolGroups(ModelHandler):
}
def item_as_dict_overview(
self, item: ServicePoolGroup
self, item: 'Model'
) -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, ServicePoolGroup)
return {
'id': item.uuid,
'priority': item.priority,

38
server/src/uds/REST/methods/services_pools.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2014-2019 Virtual Cable S.L.
# Copyright (c) 2014-2023 Virtual Cable S.L.U.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -42,7 +42,7 @@ from uds.core import types
from uds.core.managers.user_service import UserServiceManager
from uds.core.ui import gui
from uds.core.consts.images import DEFAULT_THUMB_BASE64
from uds.core.util import log, permissions
from uds.core.util import log, permissions, ensure
from uds.core.util.config import GlobalConfig
from uds.core.util.model import getSqlDatetime, processUuid
from uds.core.util.state import State
@ -65,6 +65,9 @@ from .services import Services
from .user_services import (AssignedService, CachedService, Changelog, Groups,
Publications, Transports)
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -109,7 +112,7 @@ class ServicesPools(ModelHandler):
remove_fields = ['osmanager_id', 'service_id']
table_title = _('Service Pools')
table_title = typing.cast(str, _('Service Pools'))
table_fields = [
{'name': {'title': _('Name')}},
{'state': {'title': _('Status'), 'type': 'dict', 'dict': State.dictionary()}},
@ -133,7 +136,7 @@ class ServicesPools(ModelHandler):
('createFromAssignable', True),
]
def getItems(self, *args, **kwargs):
def getItems(self, *args, **kwargs) -> typing.Generator[typing.Any, None, None]:
# Optimized query, due that there is a lot of info needed for theee
d = getSqlDatetime() - datetime.timedelta(seconds=GlobalConfig.RESTRAINT_TIME.getInt())
return super().getItems(
@ -180,7 +183,8 @@ class ServicesPools(ModelHandler):
# return super().getItems(overview=kwargs.get('overview', True), prefetch=['service', 'service__provider', 'servicesPoolGroup', 'image', 'tags'])
# return super(ServicesPools, self).getItems(*args, **kwargs)
def item_as_dict(self, item: ServicePool) -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, ServicePool)
summary = 'summarize' in self._params
# if item does not have an associated service, hide it (the case, for example, for a removed service)
# Access from dict will raise an exception, and item will be skipped
@ -576,14 +580,16 @@ class ServicesPools(ModelHandler):
except Exception as e:
raise RequestError(str(e)) from e
def afterSave(self, item: ServicePool) -> None:
def afterSave(self, item: 'Model') -> None:
item = ensure.is_instance(item, ServicePool)
if self._params.get('publish_on_save', False) is True:
try:
item.publish()
except Exception as e:
logger.error('Could not publish service pool %s: %s', item.name, e)
def deleteItem(self, item: ServicePool) -> None:
def deleteItem(self, item: 'Model') -> None:
item = ensure.is_instance(item, ServicePool)
try:
logger.debug('Deleting %s', item)
item.remove() # This will mark it for deletion, but in fact will not delete it directly
@ -592,14 +598,16 @@ class ServicesPools(ModelHandler):
logger.exception('deleting service pool')
# Logs
def getLogs(self, item: ServicePool) -> typing.List[typing.Dict]:
def getLogs(self, item: 'Model') -> typing.List[typing.Dict]:
item = ensure.is_instance(item, ServicePool)
try:
return log.getLogs(item)
except Exception:
return []
# Set fallback status
def setFallbackAccess(self, item: ServicePool):
def setFallbackAccess(self, item: 'Model'):
item = ensure.is_instance(item, ServicePool)
self.ensureAccess(item, types.permissions.PermissionType.MANAGEMENT)
fallback = self._params.get('fallbackAccess')
@ -609,11 +617,13 @@ class ServicesPools(ModelHandler):
item.save()
return item.fallbackAccess
def getFallbackAccess(self, item: ServicePool):
def getFallbackAccess(self, item: 'Model'):
item = ensure.is_instance(item, ServicePool)
return item.fallbackAccess
# Returns the action list based on current element, for calendar
def actionsList(self, item: ServicePool) -> typing.Any:
def actionsList(self, item: 'Model') -> typing.Any:
item = ensure.is_instance(item, ServicePool)
validActions: typing.Tuple[typing.Dict, ...] = ()
itemInfo = item.service.getType() # type: ignore
if itemInfo.usesCache is True:
@ -646,11 +656,13 @@ class ServicesPools(ModelHandler):
)
return validActions
def listAssignables(self, item: ServicePool) -> typing.Any:
def listAssignables(self, item: 'Model') -> typing.Any:
item = ensure.is_instance(item, ServicePool)
service = item.service.getInstance() # type: ignore
return [gui.choiceItem(i[0], i[1]) for i in service.listAssignables()]
def createFromAssignable(self, item: ServicePool) -> typing.Any:
def createFromAssignable(self, item: 'Model') -> typing.Any:
item = ensure.is_instance(item, ServicePool)
if 'user_id' not in self._params or 'assignable_id' not in self._params:
return self.invalidRequestException('Invalid parameters')

19
server/src/uds/REST/methods/services_usage.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012-2019 Virtual Cable S.L.
# Copyright (c) 2012-2023 Virtual Cable S.L.U.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
@ -36,14 +36,15 @@ import typing
from django.utils.translation import gettext as _
from uds.models import UserService
from uds.models import UserService, Provider
from uds.core.util.state import State
from uds.core.util.model import processUuid
from uds.REST.model import DetailHandler
from uds.core.util import ensure
# Not imported at runtime, just for type checking
if typing.TYPE_CHECKING:
from uds.models import Provider
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -88,7 +89,8 @@ class ServicesUsage(DetailHandler):
'in_use': item.in_use,
}
def getItems(self, parent: 'Provider', item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, Provider)
try:
if item is None:
userServicesQuery = UserService.objects.filter(
@ -110,10 +112,10 @@ class ServicesUsage(DetailHandler):
logger.exception('getItems')
raise self.invalidItemException()
def getTitle(self, parent: 'Provider') -> str:
def getTitle(self, parent: 'Model') -> str:
return _('Services Usage')
def getFields(self, parent: 'Provider') -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
# {'creation_date': {'title': _('Creation date'), 'type': 'datetime'}},
{'state_date': {'title': _('Access'), 'type': 'datetime'}},
@ -127,10 +129,11 @@ class ServicesUsage(DetailHandler):
{'source_host': {'title': _('Src Host')}},
]
def getRowStyle(self, parent: 'Provider') -> typing.Dict[str, typing.Any]:
def getRowStyle(self, parent: 'Model') -> typing.Dict[str, typing.Any]:
return {'field': 'state', 'prefix': 'row-state-'}
def deleteItem(self, parent: 'Provider', item: str) -> None:
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, Provider)
userService: UserService
try:
userService = UserService.objects.get(

26
server/src/uds/REST/methods/transports.py
View File

@ -30,19 +30,21 @@
'''
@itemor: Adolfo Gómez, dkmaster at dkmon dot com
'''
import re
import logging
import re
import typing
from django.utils.translation import gettext_lazy as _, gettext
from uds.core.environment import Environment
from uds.models import Transport, Network, ServicePool
from uds.core import transports, types, consts
from uds.core.ui import gui
from uds.core.util import permissions
from django.utils.translation import gettext
from django.utils.translation import gettext_lazy as _
from uds.core import consts, transports, types
from uds.core.environment import Environment
from uds.core.util import ensure, permissions
from uds.models import Network, ServicePool, Transport
from uds.REST.model import ModelHandler
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -61,7 +63,7 @@ class Transports(ModelHandler):
'label',
]
table_title = _('Transports')
table_title = typing.cast(str, _('Transports'))
table_fields = [
{'priority': {'title': _('Priority'), 'type': 'numeric', 'width': '6em'}},
{'name': {'title': _('Name'), 'visible': True, 'type': 'iconType'}},
@ -147,7 +149,8 @@ class Transports(ModelHandler):
return field
def item_as_dict(self, item: Transport) -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, Transport)
type_ = item.getType()
pools = [{'id': x.uuid} for x in item.deployedServices.all()]
return {
@ -178,10 +181,11 @@ class Transports(ModelHandler):
# And ensure small_name chars are valid [ a-zA-Z0-9:-]+
if fields['label'] and not re.match(r'^[a-zA-Z0-9:-]+$', fields['label']):
raise self.invalidRequestException(
_('Label must contain only letters, numbers, ":" and "-"')
gettext('Label must contain only letters, numbers, ":" and "-"')
)
def afterSave(self, item: Transport) -> None:
def afterSave(self, item: 'Model') -> None:
item = ensure.is_instance(item, Transport)
try:
networks = self._params['networks']
except Exception: # No networks passed in, this is ok

34
server/src/uds/REST/methods/tunnels_management.py
View File

@ -37,7 +37,7 @@ from django.utils.translation import gettext_lazy as _
import uds.core.types.permissions
from uds.core import types, consts, ui
from uds.core.util import permissions, validators
from uds.core.util import permissions, validators, ensure
from uds.core.util.model import processUuid
from uds import models
from uds.REST.model import DetailHandler, ModelHandler
@ -45,6 +45,7 @@ from uds.REST.model import DetailHandler, ModelHandler
# Not imported at runtime, just for type checking
if typing.TYPE_CHECKING:
from uds.core.module import Module
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -53,7 +54,8 @@ class TunnelServers(DetailHandler):
# tunnels/[id]/servers
custom_methods = ['maintenance']
def getItems(self, parent: 'models.ServerGroup', item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, models.ServerGroup)
try:
multi = False
if item is None:
@ -81,13 +83,15 @@ class TunnelServers(DetailHandler):
logger.exception('REST groups')
raise self.invalidItemException() from e
def getTitle(self, parent: 'models.ServerGroup') -> str:
def getTitle(self, parent: 'Model') -> str:
parent = ensure.is_instance(parent, models.ServerGroup)
try:
return _('Servers of {0}').format(parent.name)
except Exception:
return _('Servers')
return gettext('Servers')
def getFields(self, parent: 'models.ServerGroup') -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
parent = ensure.is_instance(parent, models.ServerGroup)
return [
{
'hostname': {
@ -105,19 +109,22 @@ class TunnelServers(DetailHandler):
},
]
def getRowStyle(self, parent: 'models.ServerGroup') -> typing.Dict[str, typing.Any]:
def getRowStyle(self, parent: 'Model') -> typing.Dict[str, typing.Any]:
parent = ensure.is_instance(parent, models.ServerGroup)
return {'field': 'maintenance_mode', 'prefix': 'row-maintenance-'}
# Cannot save a tunnel server, it's not editable...
def deleteItem(self, parent: 'models.ServerGroup', item: str) -> None:
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, models.ServerGroup)
try:
parent.servers.remove(models.Server.objects.get(uuid=processUuid(item)))
except Exception:
raise self.invalidItemException() from None
# Custom methods
def maintenance(self, parent: 'models.ServerGroup', id: str) -> typing.Any:
def maintenance(self, parent: 'Model', id: str) -> typing.Any:
parent = ensure.is_instance(parent, models.ServerGroup)
"""
Custom method that swaps maintenance mode state for a tunnel server
:param item:
@ -143,7 +150,7 @@ class Tunnels(ModelHandler):
detail = {'servers': TunnelServers}
save_fields = ['name', 'comments', 'host:', 'port:0']
table_title = _('Tunnels')
table_title = typing.cast(str, _('Tunnels'))
table_fields = [
{'name': {'title': _('Name'), 'visible': True, 'type': 'iconType'}},
{'comments': {'title': _('Comments')}},
@ -181,7 +188,8 @@ class Tunnels(ModelHandler):
],
)
def item_as_dict(self, item: 'models.ServerGroup') -> typing.Dict[str, typing.Any]:
def item_as_dict(self, item: 'Model') -> typing.Dict[str, typing.Any]:
item = ensure.is_instance(item, models.ServerGroup)
return {
'id': item.uuid,
'name': item.name,
@ -200,7 +208,8 @@ class Tunnels(ModelHandler):
# Ensure host is a valid IP(4 or 6) or hostname
validators.validateHost(fields['host'])
def assign(self, parent: 'models.ServerGroup') -> typing.Any:
def assign(self, parent: 'Model') -> typing.Any:
parent = ensure.is_instance(parent, models.ServerGroup)
self.ensureAccess(parent, uds.core.types.permissions.PermissionType.MANAGEMENT)
server: typing.Optional['models.Server'] = None # Avoid warning on reference before assignment
@ -220,7 +229,8 @@ class Tunnels(ModelHandler):
# TODO: implement this
return 'ok'
def tunnels(self, parent: 'models.ServerGroup') -> typing.Any:
def tunnels(self, parent: 'Model') -> typing.Any:
parent = ensure.is_instance(parent, models.ServerGroup)
"""
Custom method that returns all tunnels of a tunnel server NOT already assigned to a group
:param item:

89
server/src/uds/REST/methods/user_services.py
View File

@ -33,18 +33,19 @@
import logging
import typing
from django.db.models import Prefetch, F
from django.utils.translation import gettext as _
from uds import models
import uds.core.types.permissions
from uds.core.util.state import State
from uds.core.util.model import processUuid
from uds.core.util import log, permissions
from uds.core.util import log, permissions, ensure
from uds.core.managers.user_service import UserServiceManager
from uds.REST.model import DetailHandler
from uds.REST import ResponseError
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -115,7 +116,8 @@ class AssignedService(DetailHandler):
)
return val
def getItems(self, parent: models.ServicePool, item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, models.ServicePool)
# Extract provider
try:
if not item:
@ -147,10 +149,10 @@ class AssignedService(DetailHandler):
logger.exception('getItems')
raise self.invalidItemException() from e
def getTitle(self, parent: models.ServicePool) -> str:
def getTitle(self, parent: 'Model') -> str:
return _('Assigned services')
def getFields(self, parent: models.ServicePool) -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
{'creation_date': {'title': _('Creation date'), 'type': 'datetime'}},
{'revision': {'title': _('Revision')}},
@ -172,10 +174,11 @@ class AssignedService(DetailHandler):
{'actor_version': {'title': _('Actor version')}},
]
def getRowStyle(self, parent: models.ServicePool) -> typing.Dict[str, typing.Any]:
def getRowStyle(self, parent: 'Model') -> typing.Dict[str, typing.Any]:
return {'field': 'state', 'prefix': 'row-state-'}
def getLogs(self, parent: models.ServicePool, item: str) -> typing.List[typing.Any]:
def getLogs(self, parent: 'Model', item: str) -> typing.List[typing.Any]:
parent = ensure.is_instance(parent, models.ServicePool)
try:
userService: models.UserService = parent.assignedUserServices().get(uuid=processUuid(item))
logger.debug('Getting logs for %s', userService)
@ -184,7 +187,8 @@ class AssignedService(DetailHandler):
raise self.invalidItemException() from e
# This is also used by CachedService, so we use "userServices" directly and is valid for both
def deleteItem(self, parent: models.ServicePool, item: str) -> None:
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, models.ServicePool)
try:
userService: models.UserService = parent.userServices.get(uuid=processUuid(item))
except Exception as e:
@ -208,7 +212,8 @@ class AssignedService(DetailHandler):
log.doLog(parent, log.LogLevel.INFO, logStr, log.LogSource.ADMIN)
# Only owner is allowed to change right now
def saveItem(self, parent: models.ServicePool, item: typing.Optional[str]) -> None:
def saveItem(self, parent: 'Model', item: typing.Optional[str]) -> None:
parent = ensure.is_instance(parent, models.ServicePool)
if not item:
raise self.invalidItemException('Only modify is allowed')
fields = self.readFieldsFromParams(['auth_id', 'user_id'])
@ -247,7 +252,8 @@ class CachedService(AssignedService):
custom_methods: typing.ClassVar[typing.List[str]] = [] # Remove custom methods from assigned services
def getItems(self, parent: models.ServicePool, item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, models.ServicePool)
# Extract provider
try:
if not item:
@ -263,10 +269,10 @@ class CachedService(AssignedService):
logger.exception('getItems')
raise self.invalidItemException() from e
def getTitle(self, parent: models.ServicePool) -> str:
def getTitle(self, parent: 'Model') -> str:
return _('Cached services')
def getFields(self, parent: models.ServicePool) -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
{'creation_date': {'title': _('Creation date'), 'type': 'datetime'}},
{'revision': {'title': _('Revision')}},
@ -284,7 +290,8 @@ class CachedService(AssignedService):
{'actor_version': {'title': _('Actor version')}},
]
def getLogs(self, parent: models.ServicePool, item: str) -> typing.List[typing.Any]:
def getLogs(self, parent: 'Model', item: str) -> typing.List[typing.Any]:
parent = ensure.is_instance(parent, models.ServicePool)
try:
userService = parent.cachedUserServices().get(uuid=processUuid(item))
logger.debug('Getting logs for %s', item)
@ -298,7 +305,8 @@ class Groups(DetailHandler):
Processes the groups detail requests of a Service Pool
"""
def getItems(self, parent: models.ServicePool, item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, models.ServicePool)
group: models.Group
return [
{
@ -314,10 +322,11 @@ class Groups(DetailHandler):
for group in parent.assignedGroups.all()
]
def getTitle(self, parent: models.ServicePool) -> str:
def getTitle(self, parent: 'Model') -> str:
parent = ensure.is_instance(parent, models.ServicePool)
return _('Assigned groups')
def getFields(self, parent: models.ServicePool) -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
# Note that this field is "self generated" on client table
{
@ -342,10 +351,11 @@ class Groups(DetailHandler):
},
]
def getRowStyle(self, parent: models.ServicePool) -> typing.Dict[str, typing.Any]:
def getRowStyle(self, parent: 'Model') -> typing.Dict[str, typing.Any]:
return {'field': 'state', 'prefix': 'row-state-'}
def saveItem(self, parent: models.ServicePool, item: typing.Optional[str]) -> None:
def saveItem(self, parent: 'Model', item: typing.Optional[str]) -> None:
parent = ensure.is_instance(parent, models.ServicePool)
group: models.Group = models.Group.objects.get(uuid=processUuid(self._params['id']))
parent.assignedGroups.add(group)
log.doLog(
@ -355,7 +365,8 @@ class Groups(DetailHandler):
log.LogSource.ADMIN,
)
def deleteItem(self, parent: models.ServicePool, item: str) -> None:
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, models.ServicePool)
group: models.Group = models.Group.objects.get(uuid=processUuid(self._args[0]))
parent.assignedGroups.remove(group)
log.doLog(
@ -371,7 +382,8 @@ class Transports(DetailHandler):
Processes the transports detail requests of a Service Pool
"""
def getItems(self, parent: models.ServicePool, item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, models.ServicePool)
def getType(trans):
try:
return self.typeAsDict(trans.getType())
@ -391,10 +403,11 @@ class Transports(DetailHandler):
if getType(i)
]
def getTitle(self, parent: models.ServicePool) -> str:
def getTitle(self, parent: 'Model') -> str:
parent = ensure.is_instance(parent, models.ServicePool)
return _('Assigned transports')
def getFields(self, parent: models.ServicePool) -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
{'priority': {'title': _('Priority'), 'type': 'numeric', 'width': '6em'}},
{'name': {'title': _('Name')}},
@ -402,7 +415,8 @@ class Transports(DetailHandler):
{'comments': {'title': _('Comments')}},
]
def saveItem(self, parent: models.ServicePool, item: typing.Optional[str]) -> None:
def saveItem(self, parent: 'Model', item: typing.Optional[str]) -> None:
parent = ensure.is_instance(parent, models.ServicePool)
transport: models.Transport = models.Transport.objects.get(uuid=processUuid(self._params['id']))
parent.transports.add(transport)
log.doLog(
@ -412,7 +426,8 @@ class Transports(DetailHandler):
log.LogSource.ADMIN,
)
def deleteItem(self, parent: models.ServicePool, item: str) -> None:
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, models.ServicePool)
transport: models.Transport = models.Transport.objects.get(uuid=processUuid(self._args[0]))
parent.transports.remove(transport)
log.doLog(
@ -430,11 +445,12 @@ class Publications(DetailHandler):
custom_methods = ['publish', 'cancel'] # We provided these custom methods
def publish(self, parent: models.ServicePool):
def publish(self, parent: 'Model'):
"""
Custom method "publish", provided to initiate a publication of a deployed service
:param parent: Parent service pool
"""
parent = ensure.is_instance(parent, models.ServicePool)
changeLog = self._params['changelog'] if 'changelog' in self._params else None
if (
@ -456,13 +472,14 @@ class Publications(DetailHandler):
return self.success()
def cancel(self, parent: models.ServicePool, uuid: str):
def cancel(self, parent: 'Model', uuid: str):
"""
Invoked to cancel a running publication
Double invocation (this means, invoking cancel twice) will mean that is a "forced cancelation"
:param parent: Parent service pool
:param uuid: uuid of the publication
"""
parent = ensure.is_instance(parent, models.ServicePool)
if (
permissions.hasAccess(self._user, parent, uds.core.types.permissions.PermissionType.MANAGEMENT)
is False
@ -485,7 +502,8 @@ class Publications(DetailHandler):
return self.success()
def getItems(self, parent: models.ServicePool, item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, models.ServicePool)
return [
{
'id': i.uuid,
@ -498,10 +516,11 @@ class Publications(DetailHandler):
for i in parent.publications.all()
]
def getTitle(self, parent: models.ServicePool) -> str:
def getTitle(self, parent: 'Model') -> str:
parent = ensure.is_instance(parent, models.ServicePool)
return _('Publications')
def getFields(self, parent: models.ServicePool) -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
{'revision': {'title': _('Revision'), 'type': 'numeric', 'width': '6em'}},
{'publish_date': {'title': _('Publish date'), 'type': 'datetime'}},
@ -515,7 +534,7 @@ class Publications(DetailHandler):
{'reason': {'title': _('Reason')}},
]
def getRowStyle(self, parent: models.ServicePool) -> typing.Dict[str, typing.Any]:
def getRowStyle(self, parent: 'Model') -> typing.Dict[str, typing.Any]:
return {'field': 'state', 'prefix': 'row-state-'}
@ -524,7 +543,8 @@ class Changelog(DetailHandler):
Processes the transports detail requests of a Service Pool
"""
def getItems(self, parent: models.ServicePool, item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, models.ServicePool)
return [
{
'revision': i.revision,
@ -534,10 +554,11 @@ class Changelog(DetailHandler):
for i in parent.changelog.all()
]
def getTitle(self, parent: models.ServicePool) -> str:
return _('Changelog')
def getTitle(self, parent: 'Model') -> str:
parent = ensure.is_instance(parent, models.ServicePool)
return _(f'Changelog')
def getFields(self, parent: models.ServicePool) -> typing.List[typing.Any]:
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
{'revision': {'title': _('Revision'), 'type': 'numeric', 'width': '6em'}},
{'stamp': {'title': _('Publish date'), 'type': 'datetime'}},

60
server/src/uds/REST/methods/users_groups.py
View File

@ -41,7 +41,7 @@ from uds.core.util.state import State
from uds.core.auths.exceptions import AuthenticatorException
from uds.core.auths.user import User as aUser
from uds.core.util import log
from uds.core.util import log, ensure
from uds.core.util.model import processUuid
from uds.models import Authenticator, User, Group, ServicePool
from uds.core.managers.crypto import CryptoManager
@ -52,6 +52,9 @@ from uds.REST.model import DetailHandler
from .user_services import AssignedService
if typing.TYPE_CHECKING:
from django.db.models import Model
logger = logging.getLogger(__name__)
@ -62,7 +65,7 @@ if typing.TYPE_CHECKING:
from uds.models import UserService
def getGroupsFromMeta(groups):
def getGroupsFromMeta(groups) -> typing.Iterable[Group]:
for g in groups:
if g.is_meta:
for x in g.groups.all():
@ -79,10 +82,11 @@ def getPoolsForGroups(groups):
class Users(DetailHandler):
custom_methods = ['servicesPools', 'userServices', 'cleanRelated']
def getItems(self, parent: Authenticator, item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]) -> typing.Any:
parent = ensure.is_instance(parent, Authenticator)
# processes item to change uuid key for id
def uuid_to_id(
iterable: typing.Iterable[typing.MutableMapping[str, typing.Any]]
iterable: typing.Iterable[typing.Any] # will get values from a queryset
):
for v in iterable:
v['id'] = v['uuid']
@ -148,7 +152,7 @@ class Users(DetailHandler):
# User not found
raise self.invalidItemException() from e
def getTitle(self, parent):
def getTitle(self, parent: 'Model') -> str:
try:
return _('Users of {0}').format(
Authenticator.objects.get(
@ -158,7 +162,7 @@ class Users(DetailHandler):
except Exception:
return _('Current users')
def getFields(self, parent: Authenticator):
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
{
'name': {
@ -181,10 +185,11 @@ class Users(DetailHandler):
{'last_access': {'title': _('Last access'), 'type': 'datetime'}},
]
def getRowStyle(self, parent):
def getRowStyle(self, parent: 'Model') -> typing.Dict[str, typing.Any]:
return {'field': 'state', 'prefix': 'row-state-'}
def getLogs(self, parent: Authenticator, item: str):
def getLogs(self, parent: 'Model', item: str) -> typing.List[typing.Any]:
parent = ensure.is_instance(parent, Authenticator)
user = None
try:
user = parent.users.get(uuid=processUuid(item))
@ -193,7 +198,8 @@ class Users(DetailHandler):
return log.getLogs(user)
def saveItem(self, parent: Authenticator, item):
def saveItem(self, parent: 'Model', item):
parent = ensure.is_instance(parent, Authenticator)
logger.debug('Saving user %s / %s', parent, item)
valid_fields = [
'name',
@ -260,7 +266,8 @@ class Users(DetailHandler):
return self.getItems(parent, user.uuid)
def deleteItem(self, parent: Authenticator, item: str):
def deleteItem(self, parent: 'Model', item: str):
parent = ensure.is_instance(parent, Authenticator)
try:
user = parent.users.get(uuid=processUuid(item))
if not self._user.is_admin and (user.is_admin or user.staff_member):
@ -292,7 +299,8 @@ class Users(DetailHandler):
return 'deleted'
def servicesPools(self, parent: Authenticator, item: str):
def servicesPools(self, parent: 'Model', item: str):
parent = ensure.is_instance(parent, Authenticator)
uuid = processUuid(item)
user = parent.users.get(uuid=processUuid(uuid))
res = []
@ -314,7 +322,8 @@ class Users(DetailHandler):
return res
def userServices(self, parent: Authenticator, item: str):
def userServices(self, parent: 'Authenticator', item: str) -> typing.List[typing.Dict]:
parent = ensure.is_instance(parent, Authenticator)
uuid = processUuid(item)
user = parent.users.get(uuid=processUuid(uuid))
res = []
@ -327,7 +336,7 @@ class Users(DetailHandler):
return res
def cleanRelated(self, parent: Authenticator, item: str) -> typing.Dict:
def cleanRelated(self, parent: 'Authenticator', item: str) -> typing.Dict[str, str]:
uuid = processUuid(item)
user = parent.users.get(uuid=processUuid(uuid))
user.cleanRelated()
@ -337,7 +346,8 @@ class Users(DetailHandler):
class Groups(DetailHandler):
custom_methods = ['servicesPools', 'users']
def getItems(self, parent: Authenticator, item: typing.Optional[str]):
def getItems(self, parent: 'Model', item: typing.Optional[str]):
parent = ensure.is_instance(parent, Authenticator)
try:
multi = False
if item is None:
@ -373,13 +383,14 @@ class Groups(DetailHandler):
logger.exception('REST groups')
raise self.invalidItemException() from e
def getTitle(self, parent: Authenticator) -> str:
def getTitle(self, parent: 'Model') -> str:
parent = ensure.is_instance(parent, Authenticator)
try:
return _('Groups of {0}').format(parent.name)
except Exception:
return _('Current groups')
def getFields(self, parent):
def getFields(self, parent: 'Model') -> typing.List[typing.Any]:
return [
{
'name': {
@ -396,7 +407,8 @@ class Groups(DetailHandler):
},
]
def getTypes(self, parent: Authenticator, forType: typing.Optional[str]):
def getTypes(self, parent: 'Model', forType: typing.Optional[str]):
parent = ensure.is_instance(parent, Authenticator)
tDct = {
'group': {'name': _('Group'), 'description': _('UDS Group')},
'meta': {'name': _('Meta group'), 'description': _('UDS Meta Group')},
@ -419,7 +431,8 @@ class Groups(DetailHandler):
except Exception:
raise self.invalidRequestException() from None
def saveItem(self, parent: Authenticator, item: typing.Optional[str]) -> None:
def saveItem(self, parent: 'Model', item: typing.Optional[str]) -> None:
parent = ensure.is_instance(parent, Authenticator)
group = None # Avoid warning on reference before assignment
try:
is_meta = self._params['type'] == 'meta'
@ -484,7 +497,8 @@ class Groups(DetailHandler):
logger.exception('Saving group')
raise self.invalidRequestException() from e
def deleteItem(self, parent: Authenticator, item: str) -> None:
def deleteItem(self, parent: 'Model', item: str) -> None:
parent = ensure.is_instance(parent, Authenticator)
try:
group = parent.groups.get(uuid=item)
@ -493,8 +507,9 @@ class Groups(DetailHandler):
raise self.invalidItemException() from None
def servicesPools(
self, parent: Authenticator, item: str
self, parent: 'Model', item: str
) -> typing.List[typing.Mapping[str, typing.Any]]:
parent = ensure.is_instance(parent, Authenticator)
uuid = processUuid(item)
group = parent.groups.get(uuid=processUuid(uuid))
res: typing.List[typing.Mapping[str, typing.Any]] = []
@ -516,9 +531,10 @@ class Groups(DetailHandler):
return res
def users(
self, parent: Authenticator, item: str
self, parent: 'Model', item: str
) -> typing.List[typing.Mapping[str, typing.Any]]:
uuid = processUuid(item)
parent = ensure.is_instance(parent, Authenticator)
group = parent.groups.get(uuid=processUuid(uuid))
def info(user):
@ -549,6 +565,6 @@ class Groups(DetailHandler):
users = list(tmpSet or {}) if tmpSet else []
tmpSet = None
else:
users = group.users.all()
users = list(group.users.all())
return [info(i) for i in users]

23
server/src/uds/core/util/ensure.py
View File

@ -66,3 +66,26 @@ def is_iterable(obj: typing.Any) -> typing.Generator[typing.Any, None, None]:
yield from obj
except Exception: # Not iterable
yield obj
T = typing.TypeVar('T')
def is_instance(obj: typing.Any, cls: typing.Type[T]) -> T:
"""Checks if an object is an instance of a class or a list of instances of a class
Args:
obj (typing.Union[T, typing.Iterable[T]]): object to be checked
cls (typing.Type[T]): Class to check
Returns:
T: The object if it's an instance of the class, casted to the class if it's a list of instances of the class
Raises:
ValueError: If the object is not an instance of the class
"""
if not obj:
return obj
if isinstance(obj, cls):
return obj
raise ValueError(f'Object {obj} is not an instance of {cls}')

2
server/src/uds/services/OpenGnsys/jobs.py
View File

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

2
server/src/uds/services/OpenGnsys/og/__init__.py
View File

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

2
server/src/uds/services/OpenGnsys/og/fake.py
View File

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