shaba/openuds
1
0
Fork 1
mirror of https://github.com/dkmstr/openuds.git synced 2025-01-26 10:03:50 +03:00
Code Issues Releases Wiki Activity

more test prepartion and fixes

Browse Source
This commit is contained in:
Adolfo Gómez García 2022-08-19 20:20:29 +02:00
parent 07031850e5
commit b12730d160
6 changed files with 92 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
server/src/tests/REST/test_actor_register.py
View File

@ -42,7 +42,7 @@ from ..utils import rest, constants
logger = logging.getLogger(__name__)
class RESTActorRegister(rest.test.RESTTestCase):
class ActorRegisterV3(rest.test.RESTTestCase):
"""
Test actor functionality
"""

68
server/src/tests/REST/test_actor_v3.py
View File

@ -44,41 +44,71 @@ class TestActorV3(rest.test.RESTTestCase):
Test actor functionality
"""
def test_test_managed(self) -> None:
def do_test(self, type_: str, token: str) -> None:
"""
Test actorv3 initialization
Test actorv3 test managed
"""
rest_token, actor_token = self.login_and_register()
# Auth token already set in client headers
# No actor token, will fail
response = self.client.post(
'/uds/rest/actor/v3/test',
data={'type': MANAGED},
data={'type': type_},
content_type='application/json',
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()['result'], 'invalid token')
# Helpers
success = lambda: self.client.post(
'/uds/rest/actor/v3/test',
data={'type': type_, 'token': token},
content_type='application/json',
)
invalid = lambda: self.client.post(
'/uds/rest/actor/v3/test',
data={'type': type_, 'token': 'invalid'},
content_type='application/json',
)
# Invalid actor token also fails
response = self.client.post(
'/uds/rest/actor/v3/test',
data={'type': MANAGED, 'token': 'invalid'},
content_type='application/json',
)
response = invalid()
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()['result'], 'invalid token')
# Without header, test will success because its not authenticated
self.client.add_header(AUTH_TOKEN_HEADER, 'invalid')
response = self.client.post(
'/uds/rest/actor/v3/test',
data={'type': MANAGED, 'token': actor_token},
content_type='application/json',
)
# This one works
response = success()
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()['result'], 'ok')
# And this one too, without authentication token
# Without header, test will success because its not authenticated
self.client.add_header(AUTH_TOKEN_HEADER, 'invalid')
response = success()
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()['result'], 'ok')
# We have ALLOWED_FAILS until we get blocked for a while
# Next one will give 403
for a in range(ALLOWED_FAILS):
response = invalid()
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()['result'], 'invalid token')
# And this one will give 403
response = invalid()
self.assertEqual(response.status_code, 403)
def test_test_managed(self) -> None:
rest_token, actor_token = self.login_and_register()
self.do_test(MANAGED, actor_token)
def test_test_unmanaged(self) -> None:
# try for a first few services
service = self.user_service.deployed_service.service
rest_token, actor_token = self.login_and_register()
# Get service token
self.do_test(UNMANAGED, service.token)
# We have 2 attempts failed

4
server/src/tests/fixtures/services.py vendored
View File

@ -63,12 +63,12 @@ def createSingleTestingUserServiceStructure(
from uds.transports.Test import TestTransport
service: 'models.Service' = provider.services.create(
name='Service %d' % (glob['service_id']),
name='Service {}'.format(glob['service_id']),
data_type=ServiceTestCache.typeType,
data=ServiceTestCache(
environment.Environment(str(glob['service_id'])), provider.getInstance()
).serialize(),
token='token%d' % (glob['service_id']),
token='token{}'.format(glob['service_id']),
)
glob['service_id'] += 1 # In case we generate a some more services elsewhere

43
server/src/tests/utils/rest/test.py
View File

@ -42,43 +42,42 @@ from uds.REST.handlers import AUTH_TOKEN_HEADER
NUMBER_OF_ITEMS_TO_CREATE = 4
class RESTTestCase(test.UDSTestCase):
class RESTTestCase(test.UDSTransactionTestCasse):
# Authenticators related
auth: typing.ClassVar[models.Authenticator]
groups: typing.ClassVar[typing.List[models.Group]]
admins: typing.ClassVar[typing.List[models.User]]
staffs: typing.ClassVar[typing.List[models.User]]
plain_users: typing.ClassVar[typing.List[models.User]]
auth: models.Authenticator
groups: typing.List[models.Group]
admins: typing.List[models.User]
staffs: typing.List[models.User]
plain_users: typing.List[models.User]
user_service: typing.ClassVar[models.UserService]
user_service: models.UserService
@classmethod
def setUpTestData(cls: typing.Type['RESTTestCase']) -> None:
def setUp(self) -> None:
# Set up data for REST Test cases
# First, the authenticator related
cls.auth = fixtures.authenticators.createAuthenticator()
cls.groups = fixtures.authenticators.createGroups(
cls.auth, NUMBER_OF_ITEMS_TO_CREATE
self.auth = fixtures.authenticators.createAuthenticator()
self.groups = fixtures.authenticators.createGroups(
self.auth, NUMBER_OF_ITEMS_TO_CREATE
)
# Create some users, one admin, one staff and one user
cls.admins = fixtures.authenticators.createUsers(
cls.auth,
self.admins = fixtures.authenticators.createUsers(
self.auth,
number_of_users=NUMBER_OF_ITEMS_TO_CREATE,
is_admin=True,
groups=cls.groups,
groups=self.groups,
)
cls.staffs = fixtures.authenticators.createUsers(
cls.auth,
self.staffs = fixtures.authenticators.createUsers(
self.auth,
number_of_users=NUMBER_OF_ITEMS_TO_CREATE,
is_staff=True,
groups=cls.groups,
groups=self.groups,
)
cls.plain_users = fixtures.authenticators.createUsers(
cls.auth, number_of_users=NUMBER_OF_ITEMS_TO_CREATE, groups=cls.groups
self.plain_users = fixtures.authenticators.createUsers(
self.auth, number_of_users=NUMBER_OF_ITEMS_TO_CREATE, groups=self.groups
)
cls.user_service = fixtures.services.createSingleTestingUserServiceStructure(
cls.admins[0], cls.groups
self.user_service = fixtures.services.createSingleTestingUserServiceStructure(
self.admins[0], self.groups
)
@staticmethod

24
server/src/uds/REST/methods/actor_v3.py
View File

@ -32,6 +32,7 @@ import secrets
import time
import logging
import typing
import functools
from uds.models import (
getSqlDatetimeAsUnix,
@ -79,9 +80,8 @@ def fixIdsList(idsList: typing.List[str]) -> typing.List[str]:
def checkBlockedIp(ip: str) -> None:
if GlobalConfig.BLOCK_ACTOR_FAILURES.getBool() is False:
return
cache = Cache('actorv3')
fails = cache.get(ip) or 0
if fails > ALLOWED_FAILS:
if fails >= ALLOWED_FAILS:
logger.info(
'Access to actor from %s is blocked for %s seconds since last fail',
ip,
@ -97,6 +97,18 @@ def incFailedIp(ip: str) -> None:
cache.put(ip, fails, GlobalConfig.LOGIN_BLOCK.getInt())
# Decorator that clears failed counter for the IP if succeeds
def clearIfSuccess(func: typing.Callable) -> typing.Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
_self = typing.cast('ActorV3Action', args[0])
result = func(*args, **kwargs) # If raises any exception, it will be raised and we will not clear the counter
clearFailedIp(_self._request.ip)
return result
return wrapper
def clearFailedIp(ip: str) -> None:
cache.remove(ip)
@ -137,8 +149,6 @@ class ActorV3Action(Handler):
checkBlockedIp(self._request.ip)
result = self.action()
logger.debug('Action result: %s', result)
# Result was ok, clear the failed requests for this ip
clearFailedIp(self._request.ip)
return result
except (BlockAccess, KeyError):
# For blocking attacks
@ -156,7 +166,7 @@ class Test(ActorV3Action):
name = 'test'
def post(self) -> typing.MutableMapping[str, typing.Any]:
def action(self) -> typing.MutableMapping[str, typing.Any]:
# First, try to locate an user service providing this token.
try:
if self._params.get('type') == UNMANAGED:
@ -165,13 +175,13 @@ class Test(ActorV3Action):
ActorToken.objects.get(
token=self._params['token']
) # Not assigned, because only needs check
clearFailedIp(self._request.ip)
except Exception:
# Increase failed attempts
incFailedIp(self._request.ip)
# And return error
# And return test failed
return ActorV3Action.actorResult('invalid token')
clearFailedIp(self._request.ip)
return ActorV3Action.actorResult('ok')

3
server/src/uds/core/util/cache.py
View File

@ -96,7 +96,8 @@ class Cache:
Cache.misses += 1
# logger.debug('key not found: %s', skey)
return defValue
except Exception:
except Exception as e:
logger.exception('Error getting cache key: %s', skey)
Cache.misses += 1
# logger.debug('Cache inaccesible: %s:%s', skey, e)
return defValue