Added webui (b6ab2147) · Commits · TFS / nsc

app.py

+13 −3

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

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

			import os
	from flask import Flask		from flask import Flask
	from flask_restx import Api		from flask_restx import Api
	from flask_cors import CORS		from flask_cors import CORS
	from swagger.slice_namespace import slice_ns		from swagger.tfs_namespace import tfs_ns
	from src.Constants import NSC_PORT		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__)		app = Flask(__name__)
	CORS(app)		CORS(app)
	@@ -33,7 +36,14 @@ api = Api(
	)		)

	# Register namespaces		# 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__":		if __name__ == "__main__":
	app.run(host="0.0.0.0", port=NSC_PORT, debug=True)		app.run(host="0.0.0.0", port=NSC_PORT, debug=True)

src/Constants.py

+15 −6

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

	# This file is an original contribution from Telefonica Innovación Digital S.L.		# 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
	DEFAULT_LOGGING_LEVEL = logging.INFO		DEFAULT_LOGGING_LEVEL = logging.INFO

	@@ -24,14 +25,22 @@ NSC_PORT = 8081
	# Paths		# Paths
	# Obtain the absolute path of the current file		# Obtain the absolute path of the current file
	SRC_PATH = os.path.dirname(os.path.abspath(__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		# Create the path to the desired file relative to the current file
	TEMPLATES_PATH = os.path.join(SRC_PATH, "templates")		TEMPLATES_PATH = os.path.join(SRC_PATH, "templates")

	# TFS Flags		# Controller Flags
	# Flag to determine if configurations should be uploaded to Teraflow		# If True, config is not sent to controllers
	TFS_UPLOAD = False		DUMMY_MODE = True
	# Teraflow IP		# 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		# Flag to determine if additional L2VPN configuration support is required for deploying L2VPNs with path selection
	TFS_L2VPN_SUPPORT = False		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	Original line	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	Original line	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