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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
import json
import logging
import requests
import urllib3
import emulator_utils
from requests.auth import HTTPBasicAuth
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
logging.basicConfig(
level=logging.INFO, # Nivel mínimo de severidad a registrar
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', # Formato del mensaje de log
handlers=[
logging.FileHandler("register_logs.log"), # Registra en un archivo
logging.StreamHandler() # También muestra en la consola
]
)
def main():
variables=__load_config_file(config_file=emulator_utils.get_register_file())
log_result = __log_to_capif(variables)
admintoken = log_result["access_token"]
postcreation = __create_user(admintoken,variables)
uuid = postcreation["uuid"]
__write_to_file(uuid,variables)
logger.info(uuid)
def __log_to_capif(variables):
logger.info("Logging in to CAPIF")
capif_register_url="https://" + variables["register_host"].strip()+ ":" + variables["capif_register_port"] + "/"
try:
url = capif_register_url + "login"
response = requests.request(
"POST",
url,
headers={"Content-Type": "application/json"},
auth=HTTPBasicAuth(variables["capif_register_username"], variables["capif_register_password"]),
verify=False,
)
response.raise_for_status()
response_payload = json.loads(response.text)
logger.info("Logged in to CAPIF successfully")
return response_payload
except Exception as e:
logger.error(f"Error during login to CAPIF: {e}")
raise
def __create_user(admin_token,variables):
logger.info("Creating user in CAPIF")
capif_register_url="https://" + variables["register_host"].strip()+ ":" + variables["capif_register_port"] + "/"
try:
url = capif_register_url + "createUser"
payload = {
"username": variables["capif_username"],
"password": variables["capif_password"],
"description": "description",
"email": "csr_email_address@tid.es",
"enterprise": "csr_organization",
"country": "crs_locality",
"purpose": "SDK for SAFE 6G",
}
headers = {
"Authorization": "Bearer {}".format(admin_token),
"Content-Type": "application/json",
}
response = requests.request(
"POST", url, headers=headers, data=json.dumps(payload), verify=False
)
response.raise_for_status()
response_payload = json.loads(response.text)
logger.info("User created successfully")
return response_payload
except Exception as e:
logger.error(f"Error during user creation in CAPIF: {e}")
raise
def __load_config_file(config_file: str):
"""Carga el archivo de configuración."""
try:
with open(config_file, 'r') as file:
return json.load(file)
except FileNotFoundError:
logger.warning(f"Configuration file {config_file} not found. Using defaults or environment variables.")
return {}
def __write_to_file(uuid, variables):
logger.info("Saving uuid in config.json")
# Abrimos el archivo y leemos su contenido
with open(variables["config_path"] + "config.json", "r") as infile:
data = json.load(infile)
# Modificamos el contenido del archivo para incluir el nuevo UUID
data["uuid"] = uuid
# Escribimos el contenido actualizado de nuevo en el archivo
with open(variables["config_path"] + "config.json", "w") as outfile:
json.dump(data, outfile, indent=4)
logger.info("Data saved")
if __name__ == "__main__":
logger = logging.getLogger("CAPIF Register")
logger.info("Initializing CAPIF Register")
main()