capif_invoker_connector.py

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
# Basic configuration of the logger functionality

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,  # Minimum severity level to log
    # Log message format
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(log_path),  # Log to a file
        logging.StreamHandler()  # Also display in the console
    ]
)


class capif_invoker_connector:
    """
    Τ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)
        # Load configuration from file if necessary
        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
        else:
            debug_mode = True

        # Initialize logger for this class
        self.logger = logging.getLogger(self.__class__.__name__)
        if debug_mode:
            self.logger.setLevel(logging.DEBUG)
        else:
            self.logger.setLevel(logging.WARNING)

        # Set logging level for urllib based on debug_mode
        urllib_logger = logging.getLogger("urllib3")
        if not debug_mode:
            urllib_logger.setLevel(logging.WARNING)
        else:
            urllib_logger.setLevel(logging.DEBUG)

        self.logger.info("Initializing capif_invoker_connector")

        # Assign values from environment variables or JSON configuration
        invoker_config = config.get('invoker', {})
        invoker_general_folder = os.path.abspath(os.getenv('invoker_folder', invoker_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_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('INVOKER_CAPIF_CALLBACK_URL', invoker_config.get('capif_callback_url', '')).strip()
        supported_features = os.getenv('INVOKER_SUPPORTED_FEATURES', invoker_config.get('supported_features', '')).strip()
        check_authentication_data = invoker_config.get('check_authentication_data', {})
        self.check_authentication = {
            "ip": os.getenv('INVOKER_CHECK_AUTHENTICATION_DATA_IP', check_authentication_data.get('ip', '')).strip(),
            "port":  os.getenv('INVOKER_CHECK_AUTHENTICATION_DATA_PORT', check_authentication_data.get('port', '')).strip()
        }

        # Extract CSR configuration from the JSON
        csr_config = invoker_config.get('cert_generation', {})
        csr_common_name = os.getenv('INVOKER_CSR_COMMON_NAME', csr_config.get('csr_common_name', '')).strip()
        csr_organizational_unit = os.getenv('INVOKER_CSR_ORGANIZATIONAL_UNIT', csr_config.get('csr_organizational_unit', '')).strip()
        csr_organization = os.getenv('INVOKER_CSR_ORGANIZATION', csr_config.get('csr_organization', '')).strip()
        csr_locality = os.getenv('INVOKER_CSR_LOCALITY', csr_config.get('csr_locality', '')).strip()
        csr_state_or_province_name = os.getenv('INVOKER_CSR_STATE_OR_PROVINCE_NAME', csr_config.get('csr_state_or_province_name', '')).strip()
        csr_country_name = os.getenv('INVOKER_CSR_COUNTRY_NAME', csr_config.get('csr_country_name', '')).strip()
        csr_email_address = os.getenv('INVOKER_CSR_EMAIL_ADDRESS', csr_config.get('csr_email_address', '')).strip()

        # Events configuration
        events_config = invoker_config.get('events', {})
        self.events_description = os.getenv('INVOKER_EVENTS_DESCRIPTION', events_config.get('description', ''))
        self.events_filter = os.getenv('INVOKER_EVENTS_FILTERS', events_config.get('eventFilters', ''))

        # Define the invoker folder path and create it if it doesn't exist
        self.invoker_folder = os.path.join(invoker_general_folder, capif_username)
        os.makedirs(self.invoker_folder, exist_ok=True)
        if supported_features is None:
            supported_features = 0
        self.supported_features = supported_features

        # Configure URLs for CAPIF HTTPS and register services
        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() + "/"

        # Ensure the callback URL ends with a slash
        self.capif_callback_url = self.__add_trailing_slash_to_url_if_missing(capif_callback_url.strip())

        # Assign final attributes for CAPIF connection and CSR details
        self.capif_username = capif_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.csr_locality = csr_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.invoker_capif_details_filename = "capif_api_security_context_details-" + self.capif_username + ".json"

        path = os.path.join(
            self.invoker_folder,
            self.invoker_capif_details_filename
        )
        if os.path.exists(path):
            self.invoker_capif_details = self.__load_invoker_api_details()

        self.signed_key_crt_path = os.path.join(
            self.invoker_folder,
            self.capif_username + ".crt"
        )

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

        self.pathca = os.path.join(self.invoker_folder, "ca.crt")

        self.logger.info("capif_invoker_connector initialized with the JSON parameters")

    def _load_config_file(self, config_file: str):
        """Loads the configuration file."""
        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, supp_features="0") -> 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, supp_features
            )
            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.invoker_capif_details_filename
        )
        with open(
            path, "r"
        ) as openfile:
            return json.load(openfile)

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

            response = requests.request(
                "DELETE",
                url,
                cert=(self.signed_key_crt_path, self.private_key_path),
                verify=self.pathca,
            )
            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:

            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.csr_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(self.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):
                # Removes all the content within the folder
                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_username,
                                   self.capif_invoker_password),
                verify=False,
            )

            response.raise_for_status()
            response_payload = json.loads(response.text)
            ca_root_file_path = self.pathca
            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, supp_features
    ):
        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": supp_features,
                "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",
            }
            response = requests.request(
                "POST",
                url,
                headers=headers,
                data=payload,
                verify=self.pathca,
            )
            response.raise_for_status()
            response_payload = json.loads(response.text)
            name = self.capif_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.invoker_capif_details_filename)
        try:
            with open(
                path, "w"
            ) as outfile:
                json.dump(
                    {
                        "user_name": self.capif_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,supp_features="0"):
        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, supp_features
            )

            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, supp_features
    ):
        self.logger.info(
            "Updating Invoker to CAPIF and creating signed certificate by giving our public key to CAPIF")
        try:
            path = self.invoker_folder + "/" + self.invoker_capif_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": supp_features,
                "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",
            }

            response = requests.request(
                "PUT",
                url,
                headers=headers,
                data=payload,
                cert=(self.signed_key_crt_path, self.private_key_path),
                verify=self.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

    def _create_or_update_file(self, file_name, file_type, content, mode="w"):
        """
        Create or update a file with the specified content.

        :param file_name: Name of the file (without extension).
        :param file_type: File type or extension (e.g., "txt", "json", "html").
        :param content: Content to write into the file. Can be a string, dictionary, or list.
        :param mode: Write mode ('w' to overwrite, 'a' to append). Default is 'w'.
        """
        # Validate the mode
        if mode not in ["w", "a"]:
            raise ValueError("Mode must be 'w' (overwrite) or 'a' (append).")

        # Construct the full file name
        full_file_name = f"{file_name}.{file_type}"
        full_path = os.path.join(self.invoker_folder, full_file_name)

        # Ensure the content is properly formatted
        if isinstance(content, (dict, list)):
            if file_type == "json":
                try:
                    # Serialize content to JSON
                    content = json.dumps(content, indent=4)
                except TypeError as e:
                    raise ValueError(f"Failed to serialize content to JSON: {e}")
            else:
                raise TypeError("Content must be a string when the file type is not JSON.")
        elif not isinstance(content, str):
            raise TypeError("Content must be a string, dictionary, or list.")

        try:
            # Open the file in the specified mode
            with open(full_path, mode, encoding="utf-8") as file:
                file.write(content)

            # Log success based on the mode
            if mode == "w":
                self.logger.info(f"File '{full_file_name}' created or overwritten successfully.")
            elif mode == "a":
                self.logger.info(f"Content appended to file '{full_file_name}' successfully.")
        except Exception as e:
            self.logger.error(f"Error handling the file '{full_file_name}': {e}")
            raise