Merge branch 'feat/14-tid-webui' into 'develop' (0f829776) · Commits · TFS / nsc

app.py

+13 −3

Original line number	Diff line number	Diff line
		@@ -14,11 +14,14 @@

		# This file is an original contribution from Telefonica Innovación Digital S.L.

		import os
		from flask import Flask
		from flask_restx import Api
		from flask_cors import CORS
		from swagger.slice_namespace import slice_ns
		from src.Constants import NSC_PORT
		from swagger.tfs_namespace import tfs_ns
		from swagger.ixia_namespace import ixia_ns
		from src.Constants import NSC_PORT, WEBUI_DEPLOY
		from src.webui.gui import gui_bp

		app = Flask(__name__)
		CORS(app)
		@@ -33,7 +36,14 @@ api = Api(
		)

		# Register namespaces
		api.add_namespace(slice_ns, path="/slice")
		api.add_namespace(tfs_ns, path="/tfs")
		api.add_namespace(ixia_ns, path="/ixia")
		#gui_bp = Blueprint('gui', __name__, template_folder='templates')

		if WEBUI_DEPLOY:
		app.secret_key = 'clave-secreta-dev'
		app.register_blueprint(gui_bp)


		if __name__ == "__main__":
		app.run(host="0.0.0.0", port=NSC_PORT, debug=True)

src/Constants.py

+15 −6

Original line number	Diff line number	Diff line
		@@ -14,7 +14,8 @@

		# This file is an original contribution from Telefonica Innovación Digital S.L.

		import logging, os
		import logging, os, json

		# Default logging level
		DEFAULT_LOGGING_LEVEL = logging.INFO

		@@ -24,14 +25,22 @@ NSC_PORT = 8081
		# Paths
		# Obtain the absolute path of the current file
		SRC_PATH = os.path.dirname(os.path.abspath(__file__))
		with open(os.path.join(SRC_PATH, 'IPs.json')) as f:
		ips = json.load(f)

		# Create the path to the desired file relative to the current file
		TEMPLATES_PATH = os.path.join(SRC_PATH, "templates")

		# TFS Flags
		# Flag to determine if configurations should be uploaded to Teraflow
		TFS_UPLOAD = False
		# Controller Flags
		# If True, config is not sent to controllers
		DUMMY_MODE = True
		# Teraflow IP
		TFS_IP = "192.168.165.10"
		TFS_IP = ips.get('TFS_IP')
		# Flag to determine if additional L2VPN configuration support is required for deploying L2VPNs with path selection
		TFS_L2VPN_SUPPORT = False
		# IXIA NEII IP
		IXIA_IP = ips.get('IXIA_IP')

		# WebUI
		# Flag to deploy the WebUI
		WEBUI_DEPLOY = True
		No newline at end of file

src/IPs.json

0 → 100644

+4 −0

Original line number	Diff line number	Diff line
		{
		"TFS_IP": "192.168.27.165",
		"IXIA_IP": "192.168.27.59"
		}
		No newline at end of file

src/network_slice_controller.py

+239 −73

File changed.

Preview size limit exceeded, changes collapsed.

src/realizers/ixia/NEII_V4.py

0 → 100644

+362 −0

Original line number	Diff line number	Diff line
		###SPDX-FileCopyrightText: © 2024 Telefónica Innovación Digital S.L.
		###SPDX-License-Identifier: AGPL-3.0-or-later

		from .automatizacion_ne2v4 import automatizacion
		import ipaddress, logging
		from src.Constants import IXIA_IP

		class NEII_controller:
		def __init__(self, ixia_ip=IXIA_IP):
		self.ixia_ip = ixia_ip

		def menu_principal(self, ip=IXIA_IP):
		'''
		Inputs:
		Outputs:
		Work: The main menu of the application.
		Notes: If the file is executed from the terminal, ensure that the import
		of "automatizacion_ne2v4" does not include a dot at the beggining.
		'''
		ip=input("¿Cuál es la IP del network emulator?: ")
		accion=input("¿Qué deseas hacer?: \n(1) Ver información de IP\n(2) Consultar información del Hardware\n(3) Configurar un perfil nuevo\n(4) Consultar perfiles existentes\nSelecciona una opción: ")
		if accion=="1":
		self.ver_info(ip)
		if accion=="2":
		self.hardware(ip)
		if accion=="3":
		self.nuevo_perfil(ip)
		if accion=="4":
		self.existentes(ip)
		return

		## FUNCIONES MENÚ PRINCIPAL ##

		def ver_info(self,ip):
		'''
		Inputs: -ip: the ip where the Axia API is located.
		Outputs:
		Work: It gives the information of the API.

		'''
		informacion_ip=automatizacion.obtener_informacion_ip(ip)
		if informacion_ip:
		print(informacion_ip)

		def hardware(self,ip):
		'''
		Inputs: -ip: the ip where the Axia API is located.
		Outputs:
		Work: It gives the information of the hardware.

		'''
		informacion_hardware=automatizacion.obtener_informacion_hardware(ip)
		if informacion_hardware:
		print(informacion_hardware)

		def nuevo_perfil(self,ip):
		'''
		Inputs: -ip: the ip where the Axia API is located.
		Outputs:
		Work: Creates and configures the profiles requested in the Axia API on the specified port.
		Notes: It is NOT required to fill all the information requested.

		'''
		puerto=input("¿Qué puerto quieres configurar?: ")
		num_perfiles=int(input("¿Cuantos perfiles quieres añadir?: "))
		configuraciones_totales=[]
		for i in range(num_perfiles):
		nombre=f"perfil{i+1}"
		accion=input("¿Qué deseas configurar?\n1)IPv4\t2)IPv6\n3)VLAN\t4)Delay\n5)Packet Drop\t6)Policer (Rx Bandwidth)\n7)Shaper (Rx Bandwidth)\nPor favor, separa las opciones con comas: ")
		opciones=accion.split(',')
		configuraciones={}
		for opcion in opciones:
		opcion=opcion.strip()
		if opcion=="1":
		source_ip=input("Introduce la IP de origen (IPv4): ")
		destination_ip=input("Introduce la IP destino (IPv4): ")
		prt=input("¿Qué protocolo quieres usar, IP o TCP? ")
		configuraciones['ipv4']=self.ipv4(source_ip,destination_ip,prt)
		elif opcion=="2":
		source=input("Introduce la IP de origen (IPv6): ")
		destination=input("Introduce la IP destino (IPv6): ")
		configuraciones['ipv6']=self.ipv6(source, destination)
		elif opcion=="3":
		vlan_id=int(input("Introduce identificador de VLAN: "))
		configuraciones['vlan']=self.vlan(vlan_id)
		elif opcion=="4":
		delay_perfil=input("Introduce el delay que quieres introducir en el perfil test_api: ")
		configuraciones['ethernetDelay']=self.delay(delay_perfil)
		elif opcion=="5":
		configuraciones['packetDrop']=self.packetDrop()
		elif opcion=="6":
		bandwidth=int(input('Introduzca el ancho de banda (Kbps): '))
		configuraciones['policer']=self.policer(bandwidth)
		elif opcion=="7":
		configuraciones['shaper']=self.shaper()
		else:
		print(f"Opción '{opcion}' no es válida.")
		configuracion_perfil=self.configuracion_total(configuraciones, nombre)
		configuraciones_totales.append(configuracion_perfil)
		perfil_final = {'profiles': configuraciones_totales}
		configuracion_puerto=automatizacion.envio_peticion(ip, puerto, perfil_final)
		if configuracion_puerto:
		print(configuracion_puerto)

		def existentes(self,ip):
		'''
		Inputs: -ip: the ip where the Axia API is located.
		Outputs:
		Work: Shows the information of a given port-

		'''
		puerto=input("¿Qué puerto quieres consultar?: ")
		informacion_puerto=automatizacion.obtener_informacion_puerto(ip, puerto)
		if informacion_puerto:
		print(informacion_puerto)

		def existentes_auto(self,ip,puerto):
		'''
		Inputs: -ip: the ip where the Axia API is located.
		- puerto: the port we want to get the information.
		Outputs:
		Work: Creates and configures the profiles requested in the Axia API on the specified port.

		'''
		informacion_puerto=automatizacion.obtener_informacion_puerto(ip, puerto)
		if informacion_puerto:
		print(f'info puerto\n{informacion_puerto}')
		return informacion_puerto

		## FUNCION PARA LA GUI DEL NSC ##

		import ipaddress

		def nscNEII(self, json_data):
		configuraciones = {}
		ip = self.ixia_ip
		puerto = "5"
		dataProfile = self.existentes_auto(ip, puerto)

		print(f'\n\n{json_data}\n')

		ip_version = json_data.get("ip_version", None)
		src_node_ip = json_data.get("src_node_ip", None)
		dst_node_ip = json_data.get("dst_node_ip", None)
		src_node_ipv6 = json_data.get("src_node_ipv6", None)
		dst_node_ipv6 = json_data.get("dst_node_ipv6", None)
		vlan_id = json_data.get("vlan_id", None)
		bandwidth = json_data.get("bandwidth", None)
		latency = json_data.get("latency", None)
		latency_version = json_data.get("latency_version", None)
		reliability = json_data.get("reliability", None)
		tolerance = json_data.get("tolerance", None)
		packet_reorder = json_data.get("packet_reorder", None)
		num_pack = json_data.get("num_pack", None)
		pack_reorder = json_data.get("pack_reorder", None)
		num_reorder = json_data.get("num_reorder", None)
		max_reorder = json_data.get("max_reorder", None)
		drop_version = json_data.get("drop_version", None)
		desv_reorder = json_data.get("desv_reorder", None)
		packets_drop = json_data.get("packets_drop", None)
		drops = json_data.get("drops", None)
		desv_drop = json_data.get("desv_drop", None)

		# --- Variables de configuración ---

		# Configuración de IPv4 / IPv6
		if src_node_ip and dst_node_ip:
		if isinstance(ipaddress.ip_address(src_node_ip), ipaddress.IPv4Address) and isinstance(ipaddress.ip_address(dst_node_ip), ipaddress.IPv4Address):
		configuraciones['ipv4'] = self.ipv4(src_node_ip, dst_node_ip, 5)
		if src_node_ipv6 and dst_node_ipv6:
		if isinstance(ipaddress.ip_address(src_node_ipv6), ipaddress.IPv6Address) and isinstance(ipaddress.ip_address(dst_node_ipv6), ipaddress.IPv6Address):
		configuraciones['ipv6'] = self.ipv6(src_node_ipv6, dst_node_ipv6)

		# VLAN
		if vlan_id:
		configuraciones['vlan'] = self.vlan(int(vlan_id))

		# Policer
		if bandwidth:
		configuraciones['policer'] = self.policer(bandwidth)

		# Latencia
		if latency:
		if float(latency) > 0:
		configuraciones['ethernetDelay'] = self.delay_gui(float(latency), latency_version, float(tolerance))

		# Packet Reorder
		if packet_reorder:
		configuraciones['reorder'] = self.packetReorder(num_reorder, pack_reorder, num_pack, max_reorder, packet_reorder, desv_reorder)

		# Packet Reorder when reliability
		if reliability:
		configuraciones['reorder'] = self.packetReorder(int(reliability))

		#Dropper
		if drop_version:
		configuraciones['packetDrop'] = self.packetDrop(drops, packets_drop, drop_version, desv_drop)

		# Agregar perfil
		num_profiles = len(dataProfile.get("profiles", []))
		configuracion_perfil = self.configuracion_total(configuraciones, f"profile{num_profiles + 1}")
		dataProfile['profiles'].append(configuracion_perfil)
		logging.info(f"Configuración del perfil: {configuracion_perfil}")

		# Enviar la configuración
		automatizacion.envio_peticion(ip, puerto, dataProfile)
		return automatizacion.obtener_informacion_puerto(ip, puerto)


		## FUNCIONES DE CONFIGURACIÓN DE PUERTO ##

		def delay(self,delay_perfil):
		'''
		Inputs: -delay_perfil: the delay we want to configurate.
		Outputs: the information of the delay for the controller.
		Work: Creates the configuration JSON for delay of the controller.

		'''
		delay_perfil = input("Enter the delay you want to set in the test_api profile: ")
		print(f"delay en main: {delay_perfil}")
		delay_type = input("Select one option of stadistics:\n1)None\t2)Gaussian\n3)Internet\n(Type the number)")
		configuracion_delay = automatizacion.añadir_configuracion_puerto_delay(delay_perfil,delay_type)
		print(f"Config delay:\n{configuracion_delay}")
		return configuracion_delay

		def delay_gui(self,delay_perfil, latency_version, max_latency):
		'''
		Inputs: -delay_perfil: the delay we want to configurate.
		Outputs: the information of the delay for the controller.
		Work: Creates the configuration JSON for delay of the controller.

		'''
		print(f'\nPero: {max_latency}\n')
		configuracion_delay = automatizacion.añadir_configuracion_puerto_delay(delay_perfil,latency_version,max_latency)
		return configuracion_delay

		def ipv4(self,source_ip,destination_ip,prt):
		'''
		Inputs: -source_ip: the source IPv4 we want to configurate.
		-destination_ip: the destination IPv4 we want to configurate.
		-prt: the protocol we want to configurate
		Outputs: the information of the IPv4 for the controller.
		Work: Creates the configuration JSON for IPv4 of the controller.
		Notes: by default, the protocol is TCP (6).

		'''
		source_hx=None
		destination_hx=None
		protocolo=None
		if source_ip:
		source_hx=hex(int(ipaddress.IPv4Address(source_ip)))[2:]
		if destination_ip:
		destination_hx=hex(int(ipaddress.IPv4Address(destination_ip)))[2:]
		if prt:
		protocolo="4" if prt=="IP" else "6"
		configuracion_puerto=automatizacion.añadir_configuracion_puerto_ipv4(source_hx, destination_hx, protocolo)
		return configuracion_puerto

		def ipv6(self,source, destination):
		'''
		Inputs: -source: the source IPv6 we want to configurate.
		-destination: the destination IPv6 we want to configurate.
		Outputs: the information of the IPv6 for the controller.
		Work: Creates the configuration JSON for IPv6 of the controller.

		'''
		if source:
		source=ipaddress.IPv6Address(source).exploded.replace(":", "")
		elif not source: source=None
		if destination:
		destination=ipaddress.IPv6Address(destination).exploded.replace(":", "")
		elif not destination: destination=None
		configuracion_puerto=automatizacion.añadir_configuracion_puerto_ipv6(source, destination)
		return configuracion_puerto

		def vlan(self,vlan_id):
		'''
		Inputs: -vlan_id: the VLAN we want to configurate.
		Outputs: the information of the VLAN for the controller.
		Work: Creates the configuration JSON for VLAN of the controller.

		'''
		configuracion_vlan=automatizacion.añadir_configuracion_VLAN(vlan_id)
		return configuracion_vlan

		def packetDrop(self, drop, total,version,dev):
		'''
		Inputs:
		Outputs: the information of the packet drop configuration for the controller.
		Work: Creates the configuration JSON for packet drop of the controller.

		'''
		configuracionPD=automatizacion.añadir_configuración_packetDrop(drop,total,version,dev)
		return configuracionPD

		def policer(self,bandwidth):
		'''
		Inputs: -bandwidth: the TX bandwdth we want to configurate.
		Outputs: the information of the TX bandwidth for the controller.
		Work: Creates the configuration JSON for the TX bandwidth of the controller.

		'''
		configuracion_policer=automatizacion.añadir_configuracion_policer(bandwidth)
		return configuracion_policer

		def shaper(self):
		'''
		Inputs:
		Outputs: the information of the RX bandwidth for the controller.
		Work: Creates the configuration JSON for the RX bandwidth of the controller.

		'''
		bandwidth=int(input('Introduzca el ancho de banda (Kbps): '))
		configuracion_policer=automatizacion.añadir_configuracion_shaper(bandwidth)
		return configuracion_policer

		def packetReorder(self,reorder,packages=None, npackagesreorder=None, maxreord=None, version=None, stev=None):
		print(f"\n\n\nNEII ANTES DEL PASO\nPACKAGES:{packages}\nREORDER:{reorder}\nMAXREORD:{maxreord}\nNPACKAGESORDER:{npackagesreorder}\n\n\n")
		configuracion_reorder=automatizacion.añadir_configuracion_reorder(packages,reorder,npackagesreorder, maxreord, version, stev)
		return configuracion_reorder

		def configuracion_total(self,configuraciones, nombre_perfil):
		'''
		Inputs: -configuraciones: the informtion of all the configurations for the controller.
		-nombre_perfil: the name of the profile where the information is going to be allocated.
		Outputs: the information of a configurated profile.
		Work: Creates the configuration JSON for a profile for the controller.

		'''
		perfil = {
		'tag': nombre_perfil,
		'dramAllocation': {},
		'rules': [],
		'ethernetDelay': {'enabled': False},
		'packetDrop': {'enabled': False},
		'enabled':True
		}
		if 'ipv4' in configuraciones:
		ipv4_config = configuraciones['ipv4']
		perfil['dramAllocation'] = ipv4_config['profiles'][0]['dramAllocation']
		perfil['rules'].extend(ipv4_config['profiles'][0]['rules'])
		if 'ipv6' in configuraciones:
		ipv6_config = configuraciones['ipv6']
		perfil['rules'].extend(ipv6_config['profiles'][0]['rules'])
		if 'vlan' in configuraciones:
		vlan_config = configuraciones['vlan']
		perfil['rules'].extend(vlan_config['profiles'][0]['rules'])
		if 'ethernetDelay' in configuraciones:
		perfil['ethernetDelay'] = configuraciones['ethernetDelay']['ethernetDelay']
		if 'packetDrop' in configuraciones:
		perfil['packetDrop'] = configuraciones['packetDrop']['packetDrop']
		if 'policer' in configuraciones:
		perfil['policer']=configuraciones['policer']['policer']
		if 'shaper' in configuraciones:
		perfil['shaper']=configuraciones['shaper']['shaper']
		if 'reorder' in configuraciones:
		perfil['reorder']=configuraciones['reorder']['reorder']
		return perfil

		if __name__=="__main__":
		controller=NEII_controller()
		controller.menu_principal()
		No newline at end of file

Admin message