Reformateo de codigo (4a7c4eb2) · Commits · OCF / sdk

.flake8

0 → 100644

+2 −0

Original line number	Diff line number	Diff line
		[flake8]
		max-line-length = 120

netapp-samples/netapp-provider-sample/capif-sdk-config-sample.json

+3 −3

Original line number	Diff line number	Diff line
		@@ -35,10 +35,10 @@
		},
		"publish_req": {
		"service_api_id": "",
		"publisher_apf_id": "APFa839e8f520b35e6128b865076f2326",
		"publisher_apf_id": "APFa21b8720cb9f4625b8cbb182f72912",
		"publisher_aefs_ids": [
		"AEFa38221323a1ac541136b3cc6594da7",
		"AEFdad3120ec180ab4ae0fb3d82b90b56"
		"AEFb02a55977008db414efd42dfe51537",
		"AEF7245fec7c72bb670a70d907aa29430"
		]
		},
		"api_description_path": "./netapp-provider-api-spec.json"

netapp-samples/netapp-provider-sample/netapp-provider-api-spec.json

+2 −2

Original line number	Diff line number	Diff line
		@@ -2,7 +2,7 @@
		"apiName": "Api-de-prueba-2",
		"aefProfiles": [
		{
		"aefId": "AEFa38221323a1ac541136b3cc6594da7",
		"aefId": "AEFb02a55977008db414efd42dfe51537",
		"versions": [
		{
		"apiVersion": "v1",
		@@ -61,7 +61,7 @@
		]
		},
		{
		"aefId": "AEFdad3120ec180ab4ae0fb3d82b90b56",
		"aefId": "AEF7245fec7c72bb670a70d907aa29430",
		"versions": [
		{
		"apiVersion": "v1",

sdk/CAPIFInvokerConnector.py

0 → 100644

+477 −0

Original line number	Diff line number	Diff line
		import os
		import logging
		import shutil
		from requests.auth import HTTPBasicAuth
		import urllib3
		from OpenSSL.SSL import FILETYPE_PEM
		from OpenSSL.crypto import (
		dump_certificate_request,
		dump_privatekey,
		PKey,
		TYPE_RSA,
		X509Req
		)
		import requests
		import json
		import warnings
		from requests.exceptions import RequestsDependencyWarning
		urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
		warnings.filterwarnings("ignore", category=RequestsDependencyWarning)
		# noqa: E501
		# Configuración básica del logger

		log_path = 'logs/sdk_logs.log'

		log_dir = os.path.dirname(log_path)

		if not os.path.exists(log_dir):
		os.makedirs(log_dir)

		logging.basicConfig(
		level=logging.NOTSET, # Nivel mínimo de severidad a registrar
		# Formato del mensaje de log
		format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
		handlers=[
		logging.FileHandler(log_path), # Registra en un archivo
		logging.StreamHandler() # También muestra en la consola
		]
		)


		class CAPIFInvokerConnector:
		"""
		Τhis class is responsbile for onboarding an Invoker (ex. a Invoker) to CAPIF
		"""

		def __init__(self , config_file: str):

		config_file = os.path.abspath(config_file)
		# Cargar configuración desde archivo si es necesario
		config = self.__load_config_file(config_file)

		debug_mode = os.getenv('DEBUG_MODE', config.get(
		'debug_mode', 'False')).strip().lower()
		if debug_mode == "false":
		debug_mode = False

		# Inicializar logger
		self.logger = logging.getLogger(self.__class__.__name__)
		if debug_mode:
		self.logger.setLevel(logging.DEBUG)
		else:
		self.logger.setLevel(logging.WARNING)

		urllib_logger = logging.getLogger("urllib3")
		if not debug_mode:
		urllib_logger.setLevel(logging.WARNING)
		else:
		urllib_logger.setLevel(logging.DEBUG)

		self.logger.info("Initializing CAPIFInvokerConnector")

		# Asignar valores desde variables de entorno o desde configuración

		invoker_general_folder = os.path.abspath(
		os.getenv('invoker_folder', config.get('invoker_folder', '')).strip())

		capif_host = os.getenv(
		'CAPIF_HOST', config.get('capif_host', '')).strip()
		register_host = os.getenv(
		'REGISTER_HOST', config.get('register_host', '')).strip()
		capif_https_port = str(
		os.getenv('CAPIF_HTTPS_PORT', config.get('capif_https_port', '')).strip())
		capif_register_port = str(os.getenv(
		'CAPIF_REGISTER_PORT', config.get('capif_register_port', '')).strip())
		capif_invoker_username = os.getenv(
		'CAPIF_USERNAME', config.get('capif_username', '')).strip()
		capif_invoker_password = os.getenv(
		'CAPIF_PASSWORD', config.get('capif_password', '')).strip()
		capif_callback_url = os.getenv(
		'CAPIF_CALLBACK_URL', config.get('capif_callback_url', '')).strip()

		csr_common_name = os.getenv(
		'CSR_COMMON_NAME', config.get('csr_common_name', '')).strip()
		csr_organizational_unit = os.getenv(
		'CSR_ORGANIZATIONAL_UNIT', config.get('csr_organizational_unit', '')).strip()
		csr_organization = os.getenv(
		'CSR_ORGANIZATION', config.get('csr_organization', '')).strip()
		crs_locality = os.getenv(
		'CRS_LOCALITY', config.get('crs_locality', '')).strip()
		csr_state_or_province_name = os.getenv('CSR_STATE_OR_PROVINCE_NAME', config.get(
		'csr_state_or_province_name', '')).strip()
		csr_country_name = os.getenv(
		'CSR_COUNTRY_NAME', config.get('csr_country_name', '')).strip()
		csr_email_address = os.getenv(
		'CSR_EMAIL_ADDRESS', config.get('csr_email_address', '')).strip()

		self.invoker_folder = os.path.join(
		invoker_general_folder, capif_invoker_username)
		os.makedirs(self.invoker_folder, exist_ok=True)
		# Resto del código original para inicializar URLs y otros atributos

		if len(capif_https_port) == 0 or int(capif_https_port) == 443:
		self.capif_https_url = "https://" + capif_host.strip() + "/"
		else:
		self.capif_https_url = (
		"https://" + capif_host.strip() + ":" + capif_https_port.strip() + "/"
		)

		if len(capif_register_port) == 0:
		self.capif_register_url = "https://" + register_host.strip() + ":8084/"
		else:
		self.capif_register_url = (
		"https://" + register_host.strip() + ":" + capif_register_port.strip() + "/"
		)

		self.capif_callback_url = self.__add_trailing_slash_to_url_if_missing(
		capif_callback_url.strip()
		)

		self.capif_invoker_username = capif_invoker_username
		self.capif_invoker_password = capif_invoker_password

		self.csr_common_name = "invoker_" + csr_common_name
		self.csr_organizational_unit = csr_organizational_unit
		self.csr_organization = csr_organization
		self.crs_locality = crs_locality
		self.csr_state_or_province_name = csr_state_or_province_name
		self.csr_country_name = csr_country_name
		self.csr_email_address = csr_email_address
		self.capif_api_details_filename = "capif_api_security_context_details-" + \
		self.capif_invoker_username+".json"
		# self.capif_api_details = self.__load_invoker_api_details()

		self.logger.info(
		"CAPIFInvokerConnector initialized with the capif-sdk-config.json parameters")

		def __load_config_file(self, config_file: str):
		"""Carga el archivo de configuración."""
		try:
		with open(config_file, 'r') as file:
		return json.load(file)
		except FileNotFoundError:
		self.logger.warning(
		f"Configuration file {config_file} not found. Using defaults or environment variables.")
		return {}

		def __add_trailing_slash_to_url_if_missing(self, url):
		if url[len(url) - 1] != "/":
		url = url + "/"
		return url

		def onboard_invoker(self) -> None:
		self.logger.info("Registering and onboarding Invoker")
		try:
		public_key = self.__create_private_and_public_keys()
		capif_postauth_info = self.__save_ca_root_and_get_auth()
		capif_onboarding_url = capif_postauth_info["ccf_onboarding_url"]
		capif_discover_url = capif_postauth_info["ccf_discover_url"]
		capif_access_token = capif_postauth_info["access_token"]
		api_invoker_id = self.__onboard_invoker_and_create_certificate(
		public_key, capif_onboarding_url, capif_access_token
		)
		self.__write_to_file(api_invoker_id, capif_discover_url)
		self.logger.info("Invoker registered and onboarded successfully")
		except Exception as e:
		self.logger.error(
		f"Error during Invoker registration and onboarding: {e}")
		raise

		def __load_invoker_api_details(self):
		self.logger.info("Loading Invoker API details")
		path = os.path.join(
		self.invoker_folder,
		self.capif_api_details_filename
		)
		with open(
		path, "r"
		) as openfile:
		return json.load(openfile)

		def __offboard_Invoker(self) -> None:
		self.logger.info("Offboarding Invoker")
		try:
		capif_api_details = self.__load_invoker_api_details()
		url = (
		self.capif_https_url
		+ "api-invoker-management/v1/onboardedInvokers/"
		+ capif_api_details["api_invoker_id"]
		)

		signed_key_crt_path = os.path.join(
		self.invoker_folder,
		capif_api_details["user_name"] + ".crt"
		)

		private_key_path = os.path.join(
		self.invoker_folder,
		"private.key"
		)

		path = os.path.join(
		self.invoker_folder,
		"ca.crt"
		)
		response = requests.request(
		"DELETE",
		url,
		cert=(signed_key_crt_path, private_key_path),
		verify=path,
		)
		response.raise_for_status()
		self.logger.info("Invoker offboarded successfully")
		except Exception as e:
		self.logger.error(
		f"Error during Invoker offboarding: {e} - Response: {response.text}")
		raise

		def offboard_invoker(self) -> None:
		self.logger.info("Offboarding and deregistering Invoker")
		try:
		self.__offboard_Invoker()
		self.__remove_files()
		self.logger.info(
		"Invoker offboarded and deregistered successfully")
		except Exception as e:
		self.logger.error(
		f"Error during Invoker offboarding and deregistering: {e}")
		raise

		def __create_private_and_public_keys(self) -> str:
		self.logger.info(
		"Creating private and public keys for the Invoker cert")
		try:
		private_key_path = os.path.join(self.invoker_folder, "private.key")

		csr_file_path = os.path.join(self.invoker_folder, "cert_req.csr")

		key = PKey()
		key.generate_key(TYPE_RSA, 2048)

		req = X509Req()
		req.get_subject().CN = self.csr_common_name
		req.get_subject().O = self.csr_organization
		req.get_subject().OU = self.csr_organizational_unit
		req.get_subject().L = self.crs_locality
		req.get_subject().ST = self.csr_state_or_province_name
		req.get_subject().C = self.csr_country_name
		req.get_subject().emailAddress = self.csr_email_address
		req.set_pubkey(key)
		req.sign(key, "sha256")

		with open(csr_file_path, "wb+") as f:
		f.write(dump_certificate_request(FILETYPE_PEM, req))
		public_key = dump_certificate_request(FILETYPE_PEM, req)
		with open(private_key_path, "wb+") as f:
		f.write(dump_privatekey(FILETYPE_PEM, key))

		self.logger.info("Keys created successfully")
		return public_key
		except Exception as e:
		self.logger.error(f"Error during key creation: {e}")
		raise

		def __remove_files(self):
		self.logger.info("Removing files generated")
		try:
		folder_path = self.invoker_folder

		if os.path.exists(folder_path):
		# Elimina todo el contenido dentro de la carpeta
		for root, dirs, files in os.walk(folder_path):
		for file in files:
		os.remove(os.path.join(root, file))
		for dir in dirs:
		shutil.rmtree(os.path.join(root, dir))
		os.rmdir(folder_path)
		self.logger.info(
		f"All contents in {folder_path} removed successfully")
		else:
		self.logger.warning(f"Folder {folder_path} does not exist.")
		except Exception as e:
		self.logger.error(f"Error during removing folder contents: {e}")
		raise

		def __save_ca_root_and_get_auth(self):
		self.logger.info(
		"Saving CAPIF CA root file and getting auth token with user and password given by the CAPIF administrator")
		try:
		url = self.capif_register_url + "getauth"

		response = requests.request(
		"GET",
		url,
		headers={"Content-Type": "application/json"},
		auth=HTTPBasicAuth(self.capif_invoker_username,
		self.capif_invoker_password),
		verify=False,
		)

		response.raise_for_status()
		response_payload = json.loads(response.text)
		ca_root_file_path = os.path.join(self.invoker_folder, "ca.crt")
		ca_root_file = open(ca_root_file_path, "wb+")
		ca_root_file.write(bytes(response_payload["ca_root"], "utf-8"))
		self.logger.info(
		"CAPIF CA root file saved and auth token obtained successfully")
		return response_payload
		except Exception as e:
		self.logger.error(
		f"Error during saving CAPIF CA root file and getting auth token: {e} - Response: {response.text}")
		raise

		def __onboard_invoker_and_create_certificate(
		self, public_key, capif_onboarding_url, capif_access_token
		):
		self.logger.info(
		"Onboarding Invoker to CAPIF and creating signed certificate by giving our public key to CAPIF")
		try:
		url = self.capif_https_url + capif_onboarding_url
		payload_dict = {
		"notificationDestination": self.capif_callback_url,
		"supportedFeatures": "fffffff",
		"apiInvokerInformation": self.csr_common_name,
		"websockNotifConfig": {
		"requestWebsocketUri": True,
		"websocketUri": "websocketUri",
		},
		"onboardingInformation": {"apiInvokerPublicKey": str(public_key, "utf-8")},
		"requestTestNotification": True,
		}
		payload = json.dumps(payload_dict)
		headers = {
		"Authorization": "Bearer {}".format(capif_access_token),
		"Content-Type": "application/json",
		}
		pathca = os.path.join(self.invoker_folder, "ca.crt")
		response = requests.request(
		"POST",
		url,
		headers=headers,
		data=payload,
		verify=pathca,
		)
		response.raise_for_status()
		response_payload = json.loads(response.text)
		name = self.capif_invoker_username+".crt"
		pathcsr = os.path.join(self.invoker_folder, name)
		certification_file = open(
		pathcsr, "wb"
		)
		certification_file.write(
		bytes(
		response_payload["onboardingInformation"]["apiInvokerCertificate"],
		"utf-8",
		)
		)
		certification_file.close()
		self.logger.info(
		"Invoker onboarded and signed certificate created successfully")
		return response_payload["apiInvokerId"]
		except Exception as e:
		self.logger.error(
		f"Error during onboarding Invoker to CAPIF: {e} - Response: {response.text}")
		raise

		def __write_to_file(self, api_invoker_id, discover_services_url):
		self.logger.info(
		"Writing API invoker ID and service discovery URL to file")
		path = os.path.join(self.invoker_folder,
		self.capif_api_details_filename)
		try:
		with open(
		path, "w"
		) as outfile:
		json.dump(
		{
		"user_name": self.capif_invoker_username,
		"api_invoker_id": api_invoker_id,
		"discover_services_url": discover_services_url,
		},
		outfile,
		)
		self.logger.info(
		"API invoker ID and service discovery URL written to file successfully")
		except Exception as e:
		self.logger.error(f"Error during writing to file: {e}")
		raise

		def update_invoker(self):
		self.logger.info("Updating Invoker")
		try:

		capif_postauth_info = self.__save_ca_root_and_get_auth()
		capif_onboarding_url = capif_postauth_info["ccf_onboarding_url"]
		capif_access_token = capif_postauth_info["access_token"]
		path = self.invoker_folder + "/cert_req.csr"
		with open(path, "rb") as file:
		public_key = file.read()

		self.__update_invoker_to_capif_and_create_the_signed_certificate(
		public_key, capif_onboarding_url, capif_access_token
		)

		self.logger.info("Invoker updated successfully")
		except Exception as e:
		self.logger.error(f"Error during Invoker updating Invoker: {e}")
		raise

		def __update_invoker_to_capif_and_create_the_signed_certificate(
		self, public_key, capif_onboarding_url, capif_access_token
		):
		self.logger.info(
		"Updating Invoker to CAPIF and creating signed certificate by giving our public key to CAPIF")
		try:
		path = self.invoker_folder + "/" + self.capif_api_details_filename

		with open(path, "r") as file:
		invoker_details = file.read()

		invoker_details = json.loads(invoker_details)

		invokerid = invoker_details["api_invoker_id"]
		url = self.capif_https_url + capif_onboarding_url + "/" + invokerid
		payload_dict = {
		"notificationDestination": self.capif_callback_url,
		"supportedFeatures": "fffffff",
		"apiInvokerInformation": self.csr_common_name,
		"websockNotifConfig": {
		"requestWebsocketUri": True,
		"websocketUri": "websocketUri",
		},
		"onboardingInformation": {"apiInvokerPublicKey": str(public_key, "utf-8")},
		"requestTestNotification": True,
		}
		payload = json.dumps(payload_dict)
		headers = {
		"Authorization": "Bearer {}".format(capif_access_token),
		"Content-Type": "application/json",
		}
		signed_key_crt_path = os.path.join(
		self.invoker_folder,
		self.capif_invoker_username + ".crt"
		)

		private_key_path = os.path.join(
		self.invoker_folder,
		"private.key"
		)
		pathca = os.path.join(self.invoker_folder, "ca.crt")
		response = requests.request(
		"PUT",
		url,
		headers=headers,
		data=payload,
		cert=(signed_key_crt_path, private_key_path),
		verify=pathca,
		)

		response.raise_for_status()

		self.logger.info(
		"Invoker updated and signed certificate updated successfully")

		except Exception as e:
		self.logger.error(
		f"Error during updating Invoker to CAPIF: {e} - Response: {response.text}")
		raise

sdk/CAPIFProviderConnector.py

0 → 100644

+1249 −0

File added.

Preview size limit exceeded, changes collapsed.