protocol_interfaces.py

"""
This program is free software: you can redistribute it under the terms
of the GNU General Public License, v. 3.0. If a copy of the GNU General
Public License was not distributed with this file, see <https://www.gnu.org/licenses/>.

*********************************************************************************
* WARNING: Do not edit this file unless absolutely necessary.                   *
* Modifying this file may break communication with the Publisher (host program).*
*                                                                               *
* To ensure compatibility, update this file using the following command:        *
* curl -o protocol_interfaces.py \                                              *
* https://raw.githubusercontent.com/smswithoutborders/RelaySMS-Publisher/ \     *
* feat/plugable-platforms/platforms/protocol_interfaces.py                      *
*********************************************************************************
"""

from abc import ABC, abstractmethod
from typing import Any, Dict
import configparser
import os


class BaseProtocolInterface(ABC):
    """Base protocol interface."""

    @property
    def manifest(self) -> Dict[str, Any]:
        """
        Get the manifest data.

        Returns:
            Dict[str, Any]: A dictionary containing the manifest data.
        """
        manifest_path = os.path.join(os.path.dirname(__file__), "manifest.ini")
        if not os.path.exists(manifest_path):
            raise FileNotFoundError(f"Manifest file not found at {manifest_path}")

        config = configparser.ConfigParser()
        config.read(manifest_path)
        return {section: dict(config[section]) for section in config.sections()}

    @property
    def config(self) -> Dict[str, Any]:
        """
        Get the configuration data.

        Returns:
            Dict[str, Any]: A dictionary containing the configuration data.
        """
        config_path = os.path.join(os.path.dirname(__file__), "config.ini")
        if not os.path.exists(config_path):
            raise FileNotFoundError(f"Config file not found at {config_path}")

        config = configparser.ConfigParser()
        config.read(config_path)
        return {section: dict(config[section]) for section in config.sections()}


class OAuth2ProtocolInterface(BaseProtocolInterface):
    """Abstract base class for all oauth2 protocols."""

    @abstractmethod
    def get_authorization_url(self, **kwargs) -> Dict[str, Any]:
        """
        Get the authorization URL for the OAuth2 flow.

        This method should generate a dictionary containing the authorization URL
        and additional metadata required for the OAuth2 flow.

        Args:
            kwargs: Additional parameters required for the OAuth2 flow.

        Returns:
            Dict[str, Any]: A dictionary containing the following keys:
                - authorization_url (str): The generated authorization URL.
                - state (str): The state parameter for CSRF protection.
                - code_verifier (str or None): The generated code verifier for PKCE if
                  applicable, otherwise None.
                - client_id (str): The client ID for the OAuth2 application.
                - scope (str): The scope of the authorization request, as a
                  comma-separated string.
                - redirect_uri (str): The redirect URI for the OAuth2 application.
        """

    @abstractmethod
    def exchange_code_and_fetch_user_info(
        self, code: str, **kwargs
    ) -> Dict[str, Dict[str, Any]]:
        """
        Exchange the authorization code for an access token and retrieve user information.

        Args:
            code (str): The authorization code received from the OAuth2 provider.
            kwargs: Additional parameters required for the process.

        Returns:
            Dict[str, Dict[str, Any]]: A dictionary containing the following keys:
                - token (Dict[str, Any]): A dictionary containing:
                    - access_token (str): The access token for the user.
                    - refresh_token (str): The refresh token for the user.
                    - id_token (str, optional): The ID token for the user, if applicable.
                    - other metadata as provided by the platform.
                - userinfo (Dict[str, Any]): A dictionary containing:
                    - account_identifier (str): A unique identifier for the user, such as
                      an email address or username.
                    - name (str, optional): The full name of the user, if available.
        """

    @abstractmethod
    def revoke_token(self, token: Dict[str, str], **kwargs) -> bool:
        """
        Revoke the access token.

        This method invalidates the provided tokens, ensuring they can no longer
        be used for authentication.

        Args:
            token (Dict[str, str]): A dictionary containing token details:
                - access_token (str): The token to be revoked.
                - refresh_token (str): The refresh token to be invalidated.
                - id_token (str, optional): The ID token, if applicable.
                - other metadata as provided by the platform.
            kwargs: Additional parameters required for token revocation.

        Returns:
            bool: True if the token was successfully revoked, False otherwise.
        """

    @abstractmethod
    def send_message(
        self, token: Dict[str, str], message: str, **kwargs
    ) -> Dict[str, Any]:
        """
        Send a message to the specified recipient.

        Args:
            token (Dict[str, str]): A dictionary containing token details:
                - access_token (str): The token to be revoked.
                - refresh_token (str): The refresh token to be invalidated.
                - id_token (str, optional): The ID token, if applicable.
                - other metadata as provided by the platform.
            message (str): The content of the message to be sent.
            kwargs: Additional parameters required for sending the message.

        Returns:
            Dict[str, Any]: A dictionary containing:
                - success (bool): True if the message was sent successfully, False otherwise.
                - refreshed_token (Dict[str, Any]): A dictionary containing:
                    - access_token (str): The access token for the user.
                    - refresh_token (str): The refresh token for the user.
                    - id_token (str, optional): The ID token for the user, if applicable.
                    - other metadata as provided by the platform.
        """


class PNBAProtocolInterface(BaseProtocolInterface):
    """Abstract base class for all PNBA protocols."""

    @abstractmethod
    def send_authorization_code(self, phone_number: str, **kwargs) -> Dict[str, Any]:
        """
        Send an authorization code to the specified phone number.

        Args:
            phone_number (str): The phone number to which the authorization code is sent.
            kwargs: Additional parameters required for the process.

        Returns:
            Dict[str, Any]: A dictionary containing:
                - success (bool): True if the code was sent successfully, False otherwise.
                - message (str): A response message.
        """

    @abstractmethod
    def validate_code_and_fetch_user_info(
        self, phone_number: str, code: str, **kwargs
    ) -> Dict[str, Any]:
        """
        Validate the authorization code sent to the phone number and retrieve user information.

        Args:
            phone_number (str): The phone number to which the code was sent.
            code (str): The authorization code received.
            kwargs: Additional parameters required for the process.

        Returns:
            Dict[str, Any]: A dictionary containing:
                - two_step_verification_enabled (bool): True if two-step verification is
                  enabled, False otherwise.
                - userinfo (Dict[str, Any]): A dictionary containing:
                    - account_identifier (str): A unique identifier for the user, such as
                      a phonenumber or username.
                    - name (str, optional): The full name of the user, if available.
        """

    @abstractmethod
    def validate_password_and_fetch_user_info(
        self, phone_number: str, password: str, **kwargs
    ) -> Dict[str, Any]:
        """
        Validate the password for two-step verification and retrieve user information.

        Args:
            phone_number (str): The phone number associated with the account.
            password (str): The password for two-step verification.
            kwargs: Additional parameters required for the process.

        Returns:
            Dict[str, Any]: A dictionary containing:
                - userinfo (Dict[str, Any]): A dictionary containing:
                    - account_identifier (str): A unique identifier for the user, such as
                      a phonenumber or username.
                    - name (str, optional): The full name of the user, if available.
        """

    @abstractmethod
    def invalidate_session(self, phone_number: str, **kwargs) -> bool:
        """
        Invalidate the session associated with the phone number.

        Args:
            phone_number (str): The phone number associated with the session.
            kwargs: Additional parameters required for the process.

        Returns:
            bool: True if the session was successfully invalidated, False otherwise.
        """

    @abstractmethod
    def send_message(
        self, phone_number: str, recipient: str, message: str, **kwargs
    ) -> bool:
        """
        Send a message to the specified recipient.

        Args:
            phone_number (str): The phone number associated with the account.
            recipient (str): The recipient's phone number.
            message (str): The content of the message to be sent.

        Returns:
            bool: True if the message was sent successfully, False otherwise.
        """