p41275903
/
monkey
forked from p15670423/monkey
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

BB tests bugfix of not refreshing JWT

This commit is contained in:
VakarisZ 2019-10-30 14:38:17 +02:00
parent ec5de408b2
commit dba52fcbed
1 changed files with 16 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
envs/monkey_zoo/blackbox/island_client/monkey_island_requests.py
View File

@ -1,4 +1,5 @@
import requests import requests
import functools
# SHA3-512 of '1234567890!@#$%^&*()_nothing_up_my_sleeve_1234567890!@#$%^&*()' # SHA3-512 of '1234567890!@#$%^&*()_nothing_up_my_sleeve_1234567890!@#$%^&*()'
import logging import logging
@ -8,6 +9,7 @@ NO_AUTH_CREDS = '55e97c9dcfd22b8079189ddaeea9bce8125887e3237b800c6176c9afa80d206
LOGGER = logging.getLogger(__name__) LOGGER = logging.getLogger(__name__)
# noinspection PyArgumentList
class MonkeyIslandRequests(object): class MonkeyIslandRequests(object):
def __init__(self, server_address): def __init__(self, server_address):
self.addr = "https://{IP}/".format(IP=server_address) self.addr = "https://{IP}/".format(IP=server_address)
@ -21,29 +23,43 @@ class MonkeyIslandRequests(object):
"Unable to connect to island, aborting! Error information: {}. Server: {}".format(err, self.addr)) "Unable to connect to island, aborting! Error information: {}. Server: {}".format(err, self.addr))
assert False assert False
class _Decorators:
@classmethod
def refresh_jwt_token(cls, request_function):
@functools.wraps(request_function)
def request_function_wrapper(self, *args,**kwargs):
self.token = self.try_get_jwt_from_server()
# noinspection PyArgumentList
return request_function(self, *args, **kwargs)
return request_function_wrapper
def get_jwt_from_server(self): def get_jwt_from_server(self):
resp = requests.post(self.addr + "api/auth", resp = requests.post(self.addr + "api/auth",
json={"username": NO_AUTH_CREDS, "password": NO_AUTH_CREDS}, json={"username": NO_AUTH_CREDS, "password": NO_AUTH_CREDS},
verify=False) verify=False)
return resp.json()["access_token"] return resp.json()["access_token"]
@_Decorators.refresh_jwt_token
def get(self, url, data=None): def get(self, url, data=None):
return requests.get(self.addr + url, return requests.get(self.addr + url,
headers=self.get_jwt_header(), headers=self.get_jwt_header(),
params=data, params=data,
verify=False) verify=False)
@_Decorators.refresh_jwt_token
def post(self, url, data): def post(self, url, data):
return requests.post(self.addr + url, return requests.post(self.addr + url,
data=data, data=data,
headers=self.get_jwt_header(), headers=self.get_jwt_header(),
verify=False) verify=False)
@_Decorators.refresh_jwt_token
def post_json(self, url, dict_data): def post_json(self, url, dict_data):
return requests.post(self.addr + url, return requests.post(self.addr + url,
json=dict_data, json=dict_data,
headers=self.get_jwt_header(), headers=self.get_jwt_header(),
verify=False) verify=False)
@_Decorators.refresh_jwt_token
def get_jwt_header(self): def get_jwt_header(self):
return {"Authorization": "JWT " + self.token} return {"Authorization": "JWT " + self.token}