Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
import requests
import json
import configparser
import os
from termcolor import colored
import warnings
from urllib3.exceptions import InsecureRequestWarning
class refreshAdmin():
warnings.filterwarnings("ignore", category=InsecureRequestWarning)
def __refresh_admin_token__(self, register_ip, register_port, refresh_token, log_level):
url = "https://{}:{}/refresh".format(register_ip,register_port)
headers = {
'Authorization': 'Bearer {}'.format(refresh_token),
}
print(colored("\nRefreshing admin tokens", "yellow"))
if log_level == "debug":
print(colored("''''''''''REQUEST'''''''''''''''''","yellow"))
print(colored(f"Request: to {url}","yellow"))
print(colored(f"Request Headers: {headers}", "yellow"))
print(colored("''''''''''REQUEST'''''''''''''''''","yellow"))
try:
response = requests.request("POST", url, headers=headers, verify=False)
response.raise_for_status()
if log_level == "debug":
print(colored("\n''''''''''RESPONSE'''''''''''''''''","green"))
print(colored(f"Response to: {response.url}","green"))
print(colored(f"Response Headers: {response.headers}","green"))
print(colored(f"Response: {json.dumps(response.json(), indent=2)}","green"))
print(colored(f"Response Status code: {response.status_code}","green"))
print(colored("''''''''''RESPONSE'''''''''''''''''","green"))
response_payload = json.loads(response.text)
return response_payload['access_token']
except requests.exceptions.HTTPError as err:
raise Exception(err.response.status_code)
def execute_refresh_admin(self, log_level=None):
register_ip = os.getenv('REGISTER_HOSTNAME')
register_port = os.getenv('REGISTER_PORT')
with open('capif_ops/config_files/demo_values.json', 'r') as demo_file:
demo_values = json.load(demo_file)
try:
if "refresh_token" not in demo_values:
print(colored(f"\nYou need to log in as an administrator to receive the refresh token.\n", "red"))
return True
access_token = self.__refresh_admin_token__(register_ip, register_port, demo_values["refresh_token"], log_level)
demo_values['access_token']= access_token
with open('capif_ops/config_files/demo_values.json', 'w') as outfile:
json.dump(demo_values, outfile)
print(colored(f"\nAccess token: {access_token} received.\n","green"))
except Exception as e:
status_code = e.args[0]
print(colored(f"\n{e}.\n", "red"))
return True