Prowler Providers

# Regional provider execution pattern
for region in provider.get_regions():
    regional_client = service.get_regional_client(region)
    regional_resources = regional_client.discover_resources()
    # Process regional resources

# Non-regional provider execution pattern
global_client = service.get_client()
global_resources = global_client.discover_resources()
# Process all resources in single iteration

# File: prowler/providers/aws/aws_provider.py
def get_aws_enabled_regions(self, current_session: Session) -> set:
    """get_aws_enabled_regions returns a set of enabled AWS regions"""
    try:
        # EC2 Client to check enabled regions
        service = "ec2"
        default_region = self.get_default_region(service)
        ec2_client = current_session.client(service, region_name=default_region)

        enabled_regions = set()
        # With AllRegions=False we only get the enabled regions for the account
        for region in ec2_client.describe_regions(AllRegions=False).get("Regions", []):
            enabled_regions.add(region.get("RegionName"))

        return enabled_regions
    except Exception as error:
        logger.error(f"{error.__class__.__name__}: {error}")
        return set()

# File: prowler/providers/aws/aws_regions_by_service.json (extract)
{
  "services": {
    "ec2": {
      "regions": {
        "aws": [
          "af-south-1", "ap-east-1", "ap-northeast-1", "ap-northeast-2",
          "ap-northeast-3", "ap-south-1", "ap-southeast-1", "ap-southeast-2",
          "ca-central-1", "eu-central-1", "eu-north-1", "eu-south-1",
          "eu-west-1", "eu-west-2", "eu-west-3", "me-south-1",
          "sa-east-1", "us-east-1", "us-east-2", "us-west-1", "us-west-2"
        ],
        "aws-cn": ["cn-north-1", "cn-northwest-1"],
        "aws-us-gov": ["us-gov-east-1", "us-gov-west-1"]
      }
    }
  }
}

prowler/providers/<provider_name>/
├── __init__.py
├── <provider_name>_provider.py
├── models.py
├── exceptions/
│   ├── __init__.py
│   └── exceptions.py
├── services/
│   ├── service_name1/
│   └── service_name2/
└── lib/
    ├── __init__.py
    ├── arguments/
    │   ├── __init__.py
    │   └── arguments.py
    ├── mutelist/
    │   ├── __init__.py
    │   └── mutelist.py
    ├── regions/
    │   ├── __init__.py
    │   └── <provider_name>_regions.py
    └── service/
        ├── __init__.py
        └── service.py

import os
from typing import Optional, Union
from prowler.providers.common.provider import Provider
from prowler.providers.common.models import Audit_Metadata, Connection
from prowler.config.config import load_and_validate_config_file, get_default_mute_file_path
from prowler.lib.logger import logger
from prowler.lib.utils.utils import print_boxes

# Import your SDK and all the needed libraries for the provider.
import your_sdk_library
from your_sdk_library.auth_methods import ClientSecretCredential, ProfileCredential, DefaultCredential

# Import the needed exceptions, mutelist and models for the provider.
from prowler.providers.<provider_name>.exceptions.exceptions import <ProviderName>Exceptions
from prowler.providers.<provider_name>.mutelist.mutelist import <ProviderName>Mutelist
from prowler.providers.<provider_name>.models import <ProviderName>NeededModels

class YourProvider(Provider):
    """
    YourProvider class is the main class for the Your Provider.

    This class is responsible for initializing the provider, setting up the session,
    validating credentials, and managing identity information.

    Attributes:
        _type (str): The provider type.
        _session (YourSDKSession): The provider session.
        _identity (YourProviderIdentityInfo): The provider identity information.
        _audit_config (dict): The audit configuration.
        _mutelist (YourProviderMutelist): The provider mutelist.
        audit_metadata (Audit_Metadata): The audit metadata.
    """

    _type: str = "your_provider"
    _session: your_sdk_library.Session
    _identity: YourProviderIdentityInfo
    _audit_config: dict
    _mutelist: YourProviderMutelist
    audit_metadata: Audit_Metadata

    def __init__(
        self,
        # Authentication parameters
        client_id: str = None,
        client_secret: str = None,
        tenant_id: str = None,
        # Configuration
        config_path: str = None,
        config_content: dict = None,
        mutelist_path: str = None,
        mutelist_content: dict = None,
        # Additional provider-specific parameters
        region: str = None,
        profile: str = None,
    ):
        """
        Initializes the YourProvider instance.

        Args:
            client_id: The client ID for authentication
            client_secret: The client secret for authentication
            tenant_id: The tenant ID for authentication
            config_path: Path to the configuration file
            config_content: Configuration content as dictionary
            mutelist_path: Path to the mutelist file
            mutelist_content: Mutelist content as dictionary
            region: The region to use
            profile: The profile to use

        Raises:
            YourProviderSetUpSessionError: If session setup fails
            YourProviderInvalidCredentialsError: If credentials are invalid
        """
        logger.info("Initializing YourProvider ...")

        # Setup session using SDK
        self._session = self.setup_session(
            client_id, client_secret, tenant_id, region, profile
        )

        # Get identity information
        self._identity = self.setup_identity(self._session)

        # Load configuration
        if config_content:
            self._audit_config = config_content
        else:
            if not config_path:
                config_path = default_config_file_path
            self._audit_config = load_and_validate_config_file(self._type, config_path)

        # Setup mutelist
        if mutelist_content:
            self._mutelist = YourProviderMutelist(mutelist_content=mutelist_content)
        else:
            if not mutelist_path:
                mutelist_path = get_default_mute_file_path(self._type)
            self._mutelist = YourProviderMutelist(mutelist_path=mutelist_path)

        Provider.set_global_provider(self)

    @staticmethod
    def setup_session(
        client_id: str = None,
        client_secret: str = None,
        tenant_id: str = None,
        region: str = None,
        profile: str = None,
    ) -> your_sdk_library.Session:
        """
        Sets up the provider session using the provided credentials.

        This method handles the authentication flow and creates a session object
        that can be used to interact with the provider's services.

        Args:
            client_id: The client ID for authentication
            client_secret: The client secret for authentication
            tenant_id: The tenant ID for authentication
            region: The region to use
            profile: The profile to use

        Returns:
            YourSDKSession: The authenticated session object

        Raises:
            YourProviderSetUpSessionError: If session setup fails
        """
        try:
            logger.debug("Creating session ...")

            # Determine authentication method based on provided parameters
            if client_id and client_secret and tenant_id:
                # Use client credentials authentication
                credentials = your_sdk_library.ClientSecretCredential(
                    tenant_id=tenant_id,
                    client_id=client_id,
                    client_secret=client_secret
                )
                auth_method = "Client Credentials"
            elif profile:
                # Use profile-based authentication
                credentials = your_sdk_library.ProfileCredential(profile=profile)
                auth_method = "Profile"
            else:
                # Use default authentication (environment variables, etc.)
                credentials = your_sdk_library.DefaultCredential()
                auth_method = "Default"

            # Create session with credentials
            session = your_sdk_library.Session(
                credentials=credentials,
                region=region
            )

            logger.debug(f"Session created using {auth_method} authentication")
            return session

        except Exception as error:
            logger.critical(f"Failed to setup session: {error}")
            raise YourProviderSetUpSessionError(
                original_exception=error,
                file=os.path.basename(__file__),
            )

    def setup_identity(self, session: your_sdk_library.Session) -> YourProviderIdentityInfo:
        """
        Gets identity information from the provider session.

        This method retrieves account information, user details, and other
        identity-related data from the provider.

        Args:
            session: The authenticated session object

        Returns:
            YourProviderIdentityInfo: The identity information

        Raises:
            YourProviderSetUpIdentityError: If identity setup fails
        """
        try:
            # Use SDK to get account/identity information
            identity_info = session.get_identity()

            return YourProviderIdentityInfo(
                account_id=identity_info.account_id,
                account_name=identity_info.account_name,
                region=identity_info.region,
                user_id=identity_info.user_id,
                # Add other identity fields as needed
            )
        except Exception as e:
            logger.error(f"Failed to get identity information: {e}")
            raise YourProviderSetUpIdentityError(
                original_exception=e,
                file=os.path.basename(__file__),
            )

    @property
    def identity(self):
        """Returns the provider identity information."""
        return self._identity

    @property
    def session(self):
        """Returns the provider session object."""
        return self._session

    @property
    def type(self):
        """Returns the provider type."""
        return self._type

    @property
    def audit_config(self):
        """Returns the audit configuration."""
        return self._audit_config

    @property
    def mutelist(self):
        """Returns the provider mutelist."""
        return self._mutelist

    def print_credentials(self):
        """
        Display account information with color formatting.

        This method prints the provider credentials and account information
        in a formatted way using colorama for better readability.
        """
        from colorama import Fore, Style
        from prowler.lib.utils.utils import print_boxes

        report_lines = [
            f"  Account ID: {Fore.YELLOW}{self._identity.account_id}{Style.RESET_ALL}",
            f"  Account Name: {Fore.YELLOW}{self._identity.account_name}{Style.RESET_ALL}",
            f"  Region: {Fore.YELLOW}{self._identity.region}{Style.RESET_ALL}",
            f"  User ID: {Fore.YELLOW}{self._identity.user_id}{Style.RESET_ALL}",
        ]
        report_title = f"{Style.BRIGHT}Using the {self._type.upper()} credentials below:{Style.RESET_ALL}"
        print_boxes(report_lines, report_title)

    @staticmethod
    def test_connection(
        client_id: str = None,
        client_secret: str = None,
        tenant_id: str = None,
        region: str = None,
        profile: str = None,
        raise_on_exception: bool = True,
        provider_id: str = None,
    ) -> Connection:
        """
        Test connection to the provider.

        This method validates the provided credentials and tests the connection
        to the provider's services.

        Args:
            client_id: The client ID for authentication
            client_secret: The client secret for authentication
            tenant_id: The tenant ID for authentication
            region: The region to test
            profile: The profile to use
            raise_on_exception: Whether to raise exceptions or return Connection object
            provider_id: The provider ID to validate against

        Returns:
            Connection: Connection test result

        Raises:
            YourProviderSetUpSessionError: If session setup fails
            YourProviderInvalidCredentialsError: If credentials are invalid
        """
        try:
            # Create temporary session for testing
            test_session = YourProvider.setup_session(
                client_id, client_secret, tenant_id, region, profile
            )

            # Test the connection by getting identity
            identity = YourProvider.setup_identity(test_session)

            # Validate provider ID if provided
            if provider_id and identity.account_id != provider_id:
                raise YourProviderInvalidProviderIdError(
                    file=os.path.basename(__file__),
                )

            return Connection(
                status=True,
                message=f"Successfully connected to {provider_id or 'provider'}",
                error=None,
            )
        except Exception as e:
            if raise_on_exception:
                raise e
            return Connection(
                status=False,
                message="Failed to connect",
                error=str(e),
            )

    def get_regions(self) -> set:
        """
        Get available regions for the provider.

        Returns:
            set: Set of available region names
        """
        # Implementation depends on your provider
        # Example for cloud providers that support regions
        return {"region1", "region2", "region3"}

    def get_services(self) -> list:
        """
        Get available services for the provider.

        Returns:
            list: List of available service names
        """
        # Implementation depends on your provider
        return ["service1", "service2", "service3"]

# Import the needed generic libraries for the provider.
from pydantic import BaseModel
from dataclasses import dataclass
from typing import Optional, List

# Import the needed Prowler libraries for the provider.
from prowler.providers.common.models import ProviderOutputOptions
from prowler.config.config import output_file_timestamp

class YourProviderIdentityInfo:
    """
    Identity information for the provider.

    This class holds all the identity-related information retrieved
    from the provider, including account details and user information.
    """
    account_id: str
    account_name: str
    region: str
    user_id: str
    # Add other identity fields as needed

class YourProviderSession:
    """
    Session object that contains the credentials and authentication details for the provider.

    This class holds the actual credentials and authentication information needed
    to establish a connection with the provider's services.
    """
    # Authentication credentials
    access_key: str
    secret_key: str
    # Or for other providers:
    # client_id: str
    # client_secret: str
    # tenant_id: str

    # Connection details
    region: str

def init_parser(self):
    """Init the <provider_name> Provider CLI parser"""
    <provider_name>_parser = self.subparsers.add_parser(
        "<provider_name>", parents=[self.common_providers_parser], help="<provider_name> Provider"
    )
    # Authentication Modes
    <provider_name>_auth_subparser = <provider_name>_parser.add_argument_group("Authentication Modes")
    <provider_name>_auth_modes_group = <provider_name>_auth_subparser.add_mutually_exclusive_group()
    <provider_name>_auth_modes_group.add_argument(
        "--credentials-file",
        nargs="?",
        metavar="FILE_PATH",
        help="Authenticate using a <provider_name> Service Account Application Credentials JSON file",
    )
    <provider_name>_auth_modes_group.add_argument(
        "--impersonate-service-account",
        nargs="?",
        metavar="SERVICE_ACCOUNT",
        help="Impersonate a <provider_name> Service Account",
    )
    <provider_name>_parser.add_argument(
        "--your-provider-region",
        help="Your Provider Region",
        type=str,
    )
    <provider_name>_parser.add_argument(
        "--env-auth",
        action="store_true",
        help="Use User and Password environment variables authentication to log in against <provider_name>",
    )
    # More arguments for the provider.

from prowler.lib.mutelist.mutelist import Mutelist
from prowler.lib.check.models import CheckReportYourProvider

class YourProviderMutelist(Mutelist):
    """
    Mutelist implementation for YourProvider.

    This class handles the muting functionality for the provider,
    allowing users to exclude specific checks or resources from audits.
    """

    def is_finding_muted(self, finding: CheckReportYourProvider) -> bool:
        """
        Check if a specific finding is muted.

        Args:
            finding: The finding to check
        """
        return self.is_muted(finding.check_id, finding.resource_id)

from typing import List, Set

def get_regions() -> List[str]:
    """
    Get list of available regions for the provider.

    Returns:
        List[str]: List of available region names
    """
    return [
        "region1",
        "region2",
        "region3",
        # ... other regions
    ]

def validate_region(region: str) -> bool:
    """
    Validate if a region is supported.

    Args:
        region: The region to validate

    Returns:
        bool: True if the region is valid, False otherwise
    """
    return region in get_regions()

def get_default_region() -> str:
    """
    Get the default region for the provider.

    Returns:
        str: The default region name
    """
    return "region1"

def get_global_region() -> str:
    """
    Get the global region for the provider.

    Returns:
        str: The global region name
    """
    return "global"

from prowler.exceptions.exceptions import ProwlerException


# Exceptions codes from 7000 to 7999 are reserved for YourProvider exceptions (Numbers as example)
class YourProviderBaseException(ProwlerException):
    """Base class for YourProvider Errors."""

    YOUR_PROVIDER_ERROR_CODES = {
        (7001, "YourProviderCredentialsError"): {
            "message": "Error loading credentials for YourProvider",
            "remediation": "Check the credentials and ensure they are properly set up. API_KEY and API_SECRET are required.",
        },
        (7002, "YourProviderAuthenticationError"): {
            "message": "Authentication failed with YourProvider",
            "remediation": "Check the API credentials and ensure they are valid and have proper permissions.",
        },
        (7003, "YourProviderInvalidRegionError"): {
            "message": "Invalid region provided for YourProvider",
            "remediation": "Check the region and ensure it is a valid region for YourProvider.",
        },
        (7004, "YourProviderSetUpSessionError"): {
            "message": "Error setting up session",
            "remediation": "Check the session setup and ensure it is properly configured.",
        },
        (7005, "YourProviderInvalidProviderIdError"): {
            "message": "Provider does not match with the expected account_id",
            "remediation": "Check the provider and ensure it matches the expected account_id.",
        },
    }

    def __init__(self, code, file=None, original_exception=None, message=None):
        provider = "YourProvider"
        error_info = self.YOUR_PROVIDER_ERROR_CODES.get((code, self.__class__.__name__))
        if message:
            error_info["message"] = message
        super().__init__(
            code=code,
            source=provider,
            file=file,
            original_exception=original_exception,
            error_info=error_info,
        )


class YourProviderCredentialsError(YourProviderBaseException):
    """Base class for YourProvider credentials errors."""

    def __init__(self, file=None, original_exception=None, message=None):
        super().__init__(
            7001, file=file, original_exception=original_exception, message=message
        )


class YourProviderAuthenticationError(YourProviderCredentialsError):
    def __init__(self, file=None, original_exception=None, message=None):
        super().__init__(
            7002, file=file, original_exception=original_exception, message=message
        )


class YourProviderInvalidRegionError(YourProviderBaseException):
    def __init__(self, file=None, original_exception=None, message=None):
        super().__init__(
            7003, file=file, original_exception=original_exception, message=message
        )


class YourProviderSetUpSessionError(YourProviderCredentialsError):
    def __init__(self, file=None, original_exception=None, message=None):
        super().__init__(
            7004, file=file, original_exception=original_exception, message=message
        )


class YourProviderInvalidProviderIdError(YourProviderBaseException):
    def __init__(self, file=None, original_exception=None, message=None):
        super().__init__(
            7005, file=file, original_exception=original_exception, message=message
        )

from prowler.providers.<provider_name>.<provider_name>_provider import <ProviderName>Provider

class YourProviderService(BaseService):
    """
    Base service class for YourProvider services.

    This class provides common functionality for all services
    within the provider, including session management and error handling.
    """

    def __init__(self, provider: <ProviderName>Provider):
        """
        Initialize the service.

        Args:
            provider: The provider instance
        """
        self.client = provider.session.get_client(self.service_name)
        self.audit_config = provider.audit_config
        self.fixer_config = provider.fixer_config

class ProwlerArgumentParser:
    # Set the default parser
    def __init__(self):
        # CLI Arguments
        self.parser = argparse.ArgumentParser(
            prog="prowler",
            formatter_class=RawTextHelpFormatter,
            usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,github,nhn,dashboard,iac,your_provider} ...",
            epilog="""
Available Cloud Providers:
  {aws,azure,gcp,kubernetes,m365,github,iac,nhn,your_provider}
    aws                 AWS Provider
    azure               Azure Provider
    gcp                 GCP Provider
    kubernetes          Kubernetes Provider
    m365                Microsoft 365 Provider
    github              GitHub Provider
    iac                 IaC Provider (Preview)
    nhn                 NHN Provider (Unofficial)
    your_provider       Your Provider

Available components:
    dashboard           Local dashboard

To see the different available options on a specific component, run:
    prowler {provider|dashboard} -h|--help

Detailed documentation at https://docs.prowler.com
""",

# In the prowler setup output options section
    if provider == "aws":
        output_options = AWSOutputOptions(
            args, bulk_checks_metadata, global_provider.identity
        )
    elif provider == "azure":
        output_options = AzureOutputOptions(
            args, bulk_checks_metadata, global_provider.identity
        )
    elif provider == "gcp":
        output_options = GCPOutputOptions(
            args, bulk_checks_metadata, global_provider.identity
        )
    elif provider == "kubernetes":
        output_options = KubernetesOutputOptions(
            args, bulk_checks_metadata, global_provider.identity
        )
    elif provider == "github":
        output_options = GithubOutputOptions(
            args, bulk_checks_metadata, global_provider.identity
        )
    elif provider == "m365":
        output_options = M365OutputOptions(
            args, bulk_checks_metadata, global_provider.identity
        )
    elif provider == "nhn":
        output_options = NHNOutputOptions(
            args, bulk_checks_metadata, global_provider.identity
        )
    elif provider == "iac":
        output_options = IACOutputOptions(
            args, bulk_checks_metadata
        )
    elif provider == "your_provider":
        output_options = YourProviderOutputOptions(
            args, bulk_checks_metadata, global_provider.identity
        )


    # Setup Compliance Options
    elif provider == "your_provider":
        for compliance_name in input_compliance_frameworks:
            if compliance_name.startswith("cis_"):
                # Generate CIS Finding Object (example of compliance with CIS framework)
                filename = (
                    f"{output_options.output_directory}/compliance/"
                    f"{output_options.output_filename}_{compliance_name}.csv"
                )
                cis = YourProviderCIS(
                    findings=finding_outputs,
                    compliance=bulk_compliance_frameworks[compliance_name],
                    file_path=filename,
                )
                generated_outputs["compliance"].append(cis)
                cis.batch_write_data_to_file()

elif "your_provider" in provider_class_name.lower():
    provider_class(
        username=arguments.your_provider_username,
        password=arguments.your_provider_password,
        tenant_id=arguments.your_provider_tenant_id,
        config_path=arguments.config_file,
        mutelist_path=arguments.mutelist_file,
        fixer_config=fixer_config,
    )

class Provider(str, Enum):
    AWS = "aws"
    AZURE = "azure"
    GCP = "gcp"
    KUBERNETES = "kubernetes"
    M365 = "m365"
    GITHUB = "github"
    YOUR_PROVIDER = "your_provider"  # Add your provider here

{
  "Framework": "CIS",
  "Version": "1.0",
  "Provider": "your_provider",
  "Description": "Description of the compliance framework",
  # The requirements depends on the framework, for example, CIS has a requirements section with the checks and attributes.
  "Requirements": [
    {
      "Id": "1.1.1",
      "Description": "Description of the requirement",
      "Checks": ["your_provider_check_1", "your_provider_check_2"],
      "Attributes": []
    }
  ]
}

# Add your provider case in the display_summary_table function
elif provider.type == "your_provider":
    entity_type = "Your Entity Type"
    audited_entities = provider.identity.your_entity_field

# Add your provider case in the fill_common_finding_data function
elif provider.type == "your_provider":
    output_data["auth_method"] = f"Your Auth Method: {get_nested_attribute(provider, 'identity.auth_type')}"
    output_data["account_uid"] = get_nested_attribute(provider, "identity.account_id")
    output_data["account_name"] = get_nested_attribute(provider, "identity.account_name")
    output_data["resource_name"] = check_output.resource_name
    output_data["resource_uid"] = check_output.resource_id
    output_data["region"] = check_output.location  # or your location field

# Add your provider case in the stdout_report function
if finding.check_metadata.Provider == "your_provider":
    details = finding.your_location_field  # e.g., finding.location, finding.namespace, etc.

@staticmethod
def get_your_provider_assessment_summary(provider: Provider) -> str:
    """
    get_your_provider_assessment_summary gets the HTML assessment summary for your provider

    Args:
        provider (Provider): the provider object

    Returns:
        str: the HTML assessment summary
    """
    try:
        return f"""
            <div class="col-md-2">
                <div class="card">
                    <div class="card-header">
                        Your Provider Assessment Summary
                    </div>
                    <ul class="list-group list-group-flush">
                        <li class="list-group-item">
                            <b>Your Entity Type:</b> {provider.identity.your_entity_field}
                        </li>
                        <li class="list-group-item">
                            <b>Your Location Field:</b> {provider.identity.your_location_field}
                        </li>
                    </ul>
                </div>
            </div>
            <div class="col-md-4">
                <div class="card">
                    <div class="card-header">
                        Your Provider Credentials
                    </div>
                    <ul class="list-group list-group-flush">
                        <li class="list-group-item">
                            <b>Authentication Method:</b> {provider.auth_method}
                        </li>
                        <li class="list-group-item">
                            <b>Identity ID:</b> {provider.identity.identity_id}
                        </li>
                    </ul>
                </div>
            </div>"""
    except Exception as error:
        logger.error(
            f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
        )
        return ""

@dataclass
class CheckReportYourProvider(CheckReport):
    """
    Check report for YourProvider.
    """
    resource_name: str
    resource_id: str

    def _init_(self, metadata: Dict, resource: Any) -> None:
        super()._init_(metadata, resource)
        self.resource_name = resource.name
        self.resource_id = resource.id

[tool.poetry.dependencies]
python = "^3.9"
# ... other dependencies
your-sdk-library = "^1.0.0"  # Add your SDK dependency

import pytest
from prowler.providers.your_provider.your_provider import YourProvider

class TestYourProvider:
    """Test cases for YourProvider."""

    def test_provider_initialization_with_client_credentials(self):
        """Test provider initialization with client credentials."""
        provider = YourProvider(
            client_id="test_client_id",
            client_secret="test_client_secret",
            tenant_id="test_tenant_id"
        )
        assert provider.type == "your_provider"
        assert provider.identity is not None
        assert provider.session is not None

    def test_provider_initialization_with_profile(self):
        """Test provider initialization with profile."""
        provider = YourProvider(
            profile="test_profile"
        )
        assert provider.type == "your_provider"
        assert provider.identity is not None

    def test_connection_test(self):
        """Test connection functionality."""
        result = YourProvider.test_connection(
            client_id="test_client_id",
            client_secret="test_client_secret",
            tenant_id="test_tenant_id"
        )
        # Add assertions based on expected behavior

    def test_identity_retrieval(self):
        """Test identity information retrieval."""
        provider = YourProvider(
            client_id="test_client_id",
            client_secret="test_client_secret",
            tenant_id="test_tenant_id"
        )
        assert provider.identity.account_id is not None
        assert provider.identity.account_name is not None

    def test_argument_validation(self):
        """Test argument validation."""
        from prowler.providers.your_provider.lib.arguments.arguments import (
            validate_your_provider_arguments
        )

        # Valid arguments
        validate_your_provider_arguments(
            client_id="test_client_id",
            client_secret="test_client_secret",
            tenant_id="test_tenant_id"
        )

        # Invalid arguments
        with pytest.raises(ValueError, match="at least one authentication method"):
            validate_your_provider_arguments()

prowler/providers/<provider_name>/
├── __init__.py
├── <provider_name>_provider.py
├── models.py
├── exceptions/
│   ├── __init__.py
│   └── exceptions.py
├── services/
│   ├── service_name1/
│   └── service_name2/
└── lib/
    ├── __init__.py
    ├── arguments/
    │   ├── __init__.py
    │   └── arguments.py
    ├── mutelist/
    │   ├── __init__.py
    │   └── mutelist.py
    ├── regions/
    │   ├── __init__.py
    │   └── regions.py
    └── service/
        ├── __init__.py
        └── service.py

import os
from typing import Optional
import requests
from prowler.providers.common.provider import Provider
from prowler.providers.common.models import Audit_Metadata, Connection
from prowler.config.config import load_and_validate_config_file, get_default_mute_file_path
from prowler.lib.logger import logger
from prowler.lib.utils.utils import print_boxes

# Import the needed exceptions, mutelist and models for the provider.
from prowler.providers.<provider_name>.exceptions.exceptions import <ProviderName>Exceptions
from prowler.providers.<provider_name>.lib.mutelist.mutelist import <ProviderName>Mutelist
from prowler.providers.<provider_name>.models import <ProviderName>NeededModels

class APIProvider(Provider):
    """
    APIProvider class is the main class for the API Provider.

    This class is responsible for initializing the provider, setting up the HTTP session,
    validating credentials, and managing identity information through direct API calls.

    Attributes:
        _type (str): The provider type.
        _session (requests.Session): The HTTP session for API calls.
        _identity (APIIdentityInfo): The provider identity information.
        _audit_config (dict): The audit configuration.
        _mutelist (APIMutelist): The provider mutelist.
        audit_metadata (Audit_Metadata): The audit metadata.
    """

    _type: str = "api_provider"
    _session: Optional[requests.Session]
    _identity: APIIdentityInfo
    _audit_config: dict
    _mutelist: APIMutelist
    audit_metadata: Audit_Metadata

    def __init__(
        self,
        # Authentication parameters
        username: str = None,
        password: str = None,
        tenant_id: str = None,
        # Configuration
        config_path: str = None,
        config_content: dict = None,
        mutelist_path: str = None,
        mutelist_content: dict = None,
        fixer_config: dict = None,
    ):
        """
        Initializes the APIProvider instance.

        Args:
            username: The API username for authentication
            password: The API password for authentication
            tenant_id: The tenant ID for authentication
            config_path: Path to the configuration file
            config_content: Configuration content as dictionary
            mutelist_path: Path to the mutelist file
            mutelist_content: Mutelist content as dictionary
            fixer_config: Fixer configuration dictionary

        Raises:
            ValueError: If required authentication parameters are missing
        """
        logger.info("Initializing APIProvider ...")

        # 1) Store argument values with environment variable fallback
        self._username = username or os.getenv("YOUR_PROVIDER_USERNAME")
        self._password = password or os.getenv("YOUR_PROVIDER_PASSWORD")
        self._tenant_id = tenant_id or os.getenv("YOUR_PROVIDER_TENANT_ID")

        # Validate required parameters
        if not all([self._username, self._password, self._tenant_id]):
            raise ValueError("APIProvider requires username, password and tenant_id")

        # 2) Load audit_config, fixer_config, mutelist
        self._fixer_config = fixer_config if fixer_config else {}

        if config_content:
            self._audit_config = config_content
        else:
            if not config_path:
                config_path = default_config_file_path
            self._audit_config = load_and_validate_config_file(self._type, config_path)

        if mutelist_content:
            self._mutelist = APIMutelist(mutelist_content=mutelist_content)
        else:
            if not mutelist_path:
                mutelist_path = get_default_mute_file_path(self._type)
            self._mutelist = APIMutelist(mutelist_path=mutelist_path)

        # 3) Initialize session/token
        self._token = None
        self._session = None
        self.setup_session()

        # 4) Create identity object
        self._identity = APIIdentityInfo(
            tenant_id=self._tenant_id,
            username=self._username,
        )

        Provider.set_global_provider(self)

    @property
    def type(self) -> str:
        """Returns the type of the provider."""
        return self._type

    @property
    def identity(self) -> APIIdentityInfo:
        """Returns the provider identity information."""
        return self._identity

    @property
    def session(self) -> requests.Session:
        """Returns the HTTP session for API calls."""
        return self._session

    @property
    def audit_config(self) -> dict:
        """Returns the audit configuration."""
        return self._audit_config

    @property
    def fixer_config(self) -> dict:
        """Returns the fixer configuration."""
        return self._fixer_config

    @property
    def mutelist(self) -> APIMutelist:
        """Returns the provider mutelist."""
        return self._mutelist

    def print_credentials(self) -> None:
        """
        Display account information with color formatting.

        This method prints the provider credentials and account information
        in a formatted way using colorama for better readability.
        """
        from colorama import Style

        report_lines = [
            f"  Username: {self._username}",
            f"  TenantID: {self._tenant_id}",
        ]
        report_title = f"{Style.BRIGHT}Using the {self._type.upper()} credentials below:{Style.RESET_ALL}"
        print_boxes(report_lines, report_title)

    def setup_session(self) -> None:
        """
        Implement API authentication method by calling the provider's authentication endpoint.

        This method performs the authentication flow to obtain an access token
        and creates a requests.Session with the appropriate headers for API calls.
        """
        # Example for a Keystone-like authentication
        url = "https://api.your-provider.com/v2.0/tokens"
        data = {
            "auth": {
                "tenantId": self._tenant_id,
                "passwordCredentials": {
                    "username": self._username,
                    "password": self._password,
                },
            }
        }

        try:
            response = requests.post(url, json=data, timeout=10)
            if response.status_code == 200:
                resp_json = response.json()
                self._token = resp_json["access"]["token"]["id"]

                # Create session with authentication headers
                sess = requests.Session()
                sess.headers.update({
                    "X-Auth-Token": self._token,
                    "Content-Type": "application/json"
                })
                self._session = sess
                logger.info("API token acquired successfully and session is set up.")
            else:
                logger.critical(
                    f"Failed to get token. Status: {response.status_code}, Body: {response.text}"
                )
                raise ValueError("Failed to get API token")
        except Exception as e:
            logger.critical(f"[setup_session] Error: {e}")
            raise e

    @staticmethod
    def test_connection(
        username: str,
        password: str,
        tenant_id: str,
        raise_on_exception: bool = True,
    ) -> Connection:
        """
        Test connection to the API provider by performing:
          1) Authentication token request
          2) (Optional) a small test API call to confirm credentials are valid

        Args:
            username: The API username
            password: The API password
            tenant_id: The tenant ID
            raise_on_exception: If True, raise the caught exception;
                               if False, return Connection(error=exception).

        Returns:
            Connection: Connection test result
        """
        try:
            # 1) Validate arguments
            if not username or not password or not tenant_id:
                error_msg = "API test_connection error: missing username/password/tenant_id"
                logger.error(error_msg)
                raise ValueError(error_msg)

            # 2) Request authentication token
            token_url = "https://api.your-provider.com/v2.0/tokens"
            data = {
                "auth": {
                    "tenantId": tenant_id,
                    "passwordCredentials": {
                        "username": username,
                        "password": password,
                    },
                }
            }

            resp = requests.post(token_url, json=data, timeout=10)
            if resp.status_code != 200:
                error_msg = f"Failed to get token. Status: {resp.status_code}, Body: {resp.text}"
                logger.error(error_msg)
                if raise_on_exception:
                    raise Exception(error_msg)
                return Connection(error=Exception(error_msg))

            # Success
            token_json = resp.json()
            api_token = token_json["access"]["token"]["id"]
            logger.info("API test_connection: Successfully acquired token.")

            # 3) (Optional) Test API call to confirm credentials are valid
            test_endpoint = f"https://api.your-provider.com/v2/{tenant_id}/test"
            headers = {
                "X-Auth-Token": api_token,
                "Content-Type": "application/json",
            }

            test_resp = requests.get(test_endpoint, headers=headers, timeout=10)
            if test_resp.status_code == 200:
                logger.info("API test_connection: Test call success. Credentials valid.")
                return Connection(is_connected=True)
            else:
                error_msg = f"Test call failed. Status: {test_resp.status_code}, Body: {test_resp.text}"
                logger.error(error_msg)
                if raise_on_exception:
                    raise Exception(error_msg)
                return Connection(error=Exception(error_msg))

        except Exception as e:
            logger.critical(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}]: {e}")
            if raise_on_exception:
                raise e
            return Connection(error=e)

    @staticmethod
    def validate_arguments(username: str, password: str, tenant_id: str) -> None:
        """
        Ensures that username, password, and tenant_id are not empty.

        Args:
            username: The username to validate
            password: The password to validate
            tenant_id: The tenant ID to validate

        Raises:
            ValueError: If any required parameter is missing
        """
        if not username or not password or not tenant_id:
            raise ValueError("API Provider requires username, password and tenant_id.")

from prowler.exceptions.exceptions import ProwlerException


# Exceptions codes from 8000 to 8999 are reserved for API Provider exceptions (example numbers)
class APIProviderBaseException(ProwlerException):
    """Base class for API Provider Errors."""

    APIProvider_ERROR_CODES = {
        (8000, "APIProviderCredentialsError"): {
            "message": "API Provider credentials not found or invalid",
            "remediation": "Check the API Provider API credentials and ensure they are properly set.",
        },
        (8001, "APIProviderAuthenticationError"): {
            "message": "API Provider authentication failed",
            "remediation": "Check the API Provider API credentials and ensure they are valid.",
        },
        (8002, "APIProviderSessionError"): {
            "message": "API Provider session setup failed",
            "remediation": "Check the session setup and ensure it is properly configured.",
        },
        (8003, "APIProviderIdentityError"): {
            "message": "API Provider identity setup failed",
            "remediation": "Check credentials and ensure they are properly set up for API Provider.",
        },
        (8004, "APIProviderAPIError"): {
            "message": "API Provider API call failed",
            "remediation": "Check the API request and ensure it is properly formatted.",
        },
        (8005, "APIProviderRateLimitError"): {
            "message": "API Provider API rate limit exceeded",
            "remediation": "Reduce the number of API requests or wait before making more requests.",
        },
    }

    def __init__(self, code, file=None, original_exception=None, message=None):
        provider = "API Provider"
        error_info = self.APIProvider_ERROR_CODES.get((code, self.__class__.__name__))
        if message:
            error_info["message"] = message
        super().__init__(
            code=code,
            source=provider,
            file=file,
            original_exception=original_exception,
            error_info=error_info,
        )


class APIProviderCredentialsError(APIProviderBaseException):
    """Exception for API Provider credentials errors"""

    def __init__(self, file=None, original_exception=None, message=None):
        super().__init__(
            code=8000,
            file=file,
            original_exception=original_exception,
            message=message,
        )


class APIProviderAuthenticationError(APIProviderBaseException):
    """Exception for API Provider authentication errors"""

    def __init__(self, file=None, original_exception=None, message=None):
        super().__init__(
            code=8001,
            file=file,
            original_exception=original_exception,
            message=message,
        )


class APIProviderSessionError(APIProviderBaseException):
    """Exception for API Provider session setup errors"""

    def __init__(self, file=None, original_exception=None, message=None):
        super().__init__(
            code=8002,
            file=file,
            original_exception=original_exception,
            message=message,
        )


class APIProviderIdentityError(APIProviderBaseException):
    """Exception for API Provider identity setup errors"""

    def __init__(self, file=None, original_exception=None, message=None):
        super().__init__(
            code=8003,
            file=file,
            original_exception=original_exception,
            message=message,
        )


class APIProviderAPIError(APIProviderBaseException):
    """Exception for API Provider API errors"""

    def __init__(self, file=None, original_exception=None, message=None):
        super().__init__(
            code=8004,
            file=file,
            original_exception=original_exception,
            message=message,
        )


class APIProviderRateLimitError(APIProviderBaseException):
    """Exception for API Provider rate limit errors"""

    def __init__(self, file=None, original_exception=None, message=None):
        super().__init__(
            code=8005,
            file=file,
            original_exception=original_exception,
            message=message,
        )

from prowler.providers.<provider_name>.<provider_name>_provider import <ProviderName>Provider

class APIProviderService(BaseService):
    """
    Base service class for API Provider services.

    This class provides common functionality for all services
    within the provider, including session management and error handling.
    """

    def __init__(self, provider: <ProviderName>Provider):
        """
        Initialize the service.

        Args:
            provider: The provider instance
        """
        self.client = provider.session.get_client(self.service_name)
        self.audit_config = provider.audit_config
        self.fixer_config = provider.fixer_config
        self.session = provider.session
        self.base_url = provider.session.base_url
        self.auth = HTTPDigestAuth(
            provider.session.public_key,
            provider.session.private_key,
        )
        self.headers = {
            "Authorization": self.auth.encode(),
            "Content-Type": "application/json",
        }

import pytest
from prowler.providers.api_provider.api_provider import APIProvider

class TestAPIProvider:
    """Test cases for APIProvider."""

    def test_provider_initialization(self):
        """Test provider initialization with valid credentials."""
        provider = APIProvider(
            username="test_user",
            password="test_password",
            tenant_id="test_tenant"
        )
        assert provider.type == "your_api_provider"
        assert provider.identity is not None
        assert provider.session is not None

    def test_connection_test(self):
        """Test connection functionality."""
        result = APIProvider.test_connection(
            username="test_user",
            password="test_password",
            tenant_id="test_tenant"
        )
        # Add assertions based on expected behavior

    def test_argument_validation(self):
        """Test argument validation."""
        from prowler.providers.api_provider.api_provider import (
            APIProvider
        )

        # Valid arguments
        APIProvider.validate_arguments(
            username="test_user",
            password="test_password",
            tenant_id="test_tenant"
        )

        # Invalid arguments
        with pytest.raises(ValueError, match="requires username, password and tenant_id"):
            APIProvider.validate_arguments("", "", "")

    def test_session_setup(self):
        """Test session setup."""
        provider = APIProvider(
            username="test_user",
            password="test_password",
            tenant_id="test_tenant"
        )
        assert provider.session is not None
        assert "X-Auth-Token" in provider.session.headers

prowler/providers/<provider_name>/
├── __init__.py
├── <provider_name>_provider.py
├── models.py
└── lib/
    ├── __init__.py
    └── arguments/
        ├── __init__.py
        └── arguments.py

import json
import subprocess
import sys
from typing import List

from colorama import Fore, Style

from prowler.config.config import (
    default_config_file_path,
    load_and_validate_config_file,
)
from prowler.lib.check.models import CheckReportYourTool
from prowler.lib.logger import logger
from prowler.lib.utils.utils import print_boxes
from prowler.providers.common.models import Audit_Metadata
from prowler.providers.common.provider import Provider

class ToolProvider(Provider):
    """
    ToolProvider class is the main class for the Your Tool Provider.

    This class is responsible for initializing the provider, executing the external tool,
    parsing tool output, and converting results to Prowler's standard format.

    Attributes:
        _type (str): The provider type.
        _session: Not used for tool providers.
        _identity (str): Simple identity for tool providers.
        _audit_config (dict): The audit configuration.
        audit_metadata (Audit_Metadata): The audit metadata.
    """

    _type: str = "your_tool_provider"
    audit_metadata: Audit_Metadata

    def __init__(
        self,
        # Tool-specific parameters
        scan_path: str = ".",
        tool_specific_arg: str = "default_value",
        exclude_path: list[str] = [],
        # Configuration
        config_path: str = None,
        config_content: dict = None,
        fixer_config: dict = {},
        # Authentication (if needed for your tool)
        auth_token: str = None,
        auth_username: str = None,
    ):
        """
        Initializes the ToolProvider instance.

        Args:
            scan_path: Path to the folder containing files to scan
            tool_specific_arg: Tool-specific argument for your external tool
            exclude_path: List of paths to exclude from scan
            config_path: Path to the configuration file
            config_content: Configuration content as dictionary
            fixer_config: Fixer configuration dictionary
            auth_token: Authentication token for your tool (if needed)
            auth_username: Username for your tool (if needed)

        Raises:
            ValueError: If required parameters are missing
        """
        logger.info("Instantiating YourTool Provider...")

        # Store tool-specific parameters
        self.scan_path = scan_path
        self.tool_specific_arg = tool_specific_arg
        self.exclude_path = exclude_path
        self.region = "global"
        self.audited_account = "local-tool"
        self._session = None
        self._identity = "prowler"
        self._auth_method = "No auth"

        # Handle tool authentication if needed
        if auth_token:
            self.auth_token = auth_token
            self._auth_method = "Token"
            logger.info("Using token for tool authentication")
        elif auth_username:
            self.auth_username = auth_username
            self._auth_method = "Username"
            logger.info("Using username for tool authentication")
                logger.info("Using username for tool authentication")
            else:
                logger.debug("No authentication method provided; proceeding without authentication.")

        # Audit Config
        if config_content:
            self._audit_config = config_content
        else:
            if not config_path:
                config_path = default_config_file_path
            self._audit_config = load_and_validate_config_file(self._type, config_path)

        # Fixer Config
        self._fixer_config = fixer_config

        # Mutelist (not needed for tool providers since tools have their own mutelist logic)
        self._mutelist = None

        Provider.set_global_provider(self)

    @property
    def auth_method(self):
        """Returns the authentication method used."""
        return self._auth_method

    @property
    def type(self):
        """Returns the type of the provider."""
        return self._type

    @property
    def identity(self):
        """Returns the provider identity."""
        return self._identity

    @property
    def session(self):
        """Returns the session (not used for tool providers)."""
        return self._session

    @property
    def audit_config(self):
        """Returns the audit configuration."""
        return self._audit_config

    @property
    def fixer_config(self):
        """Returns the fixer configuration."""
        return self._fixer_config

    def setup_session(self):
        """Tool providers don't need a session since they use external tools directly"""
        return None

    def _process_check(self, finding: dict, check: dict, status: str) -> CheckReportYourTool:
        """
        Process a single check (failed or passed) and create a CheckReportYourTool object.

        Args:
            finding: The finding object from tool output
            check: The individual check data
            status: The status of the check ("FAIL", "PASS", or "MUTED")

        Returns:
            CheckReportYourTool: The processed check report
        """
        try:
            metadata_dict = {
                "Provider": "your_tool_provider",
                "CheckID": check.get("check_id", ""),
                "CheckTitle": check.get("check_name", ""),
                "CheckType": ["Your Tool Provider"],
                "ServiceName": finding["check_type"],
                "SubServiceName": "",
                "ResourceIdTemplate": "",
                "Severity": (
                    check.get("severity", "low").lower()
                    if check.get("severity")
                    else "low"
                ),
                "ResourceType": "your_tool",
                "Description": check.get("check_name", ""),
                "Risk": "",
                "RelatedUrl": (
                    check.get("guideline", "") if check.get("guideline") else ""
                ),
                "Remediation": {
                    "Code": {
                        "NativeIaC": "",
                        "Terraform": "",
                        "CLI": "",
                        "Other": "",
                    },
                    "Recommendation": {
                        "Text": "",
                        "Url": (
                            check.get("guideline", "") if check.get("guideline") else ""
                        ),
                    },
                },
                "Categories": [],
                "DependsOn": [],
                "RelatedTo": [],
                "Notes": "",
            }

            # Convert metadata dict to JSON string
            metadata = json.dumps(metadata_dict)

            report = CheckReportYourTool(metadata=metadata, finding=check)
            report.status = status
            report.resource_tags = check.get("entity_tags", {})
            report.status_extended = check.get("check_name", "")
            if status == "MUTED":
                report.muted = True
            return report
        except Exception as error:
            logger.critical(
                f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
            )
            sys.exit(1)

    def run(self) -> List[CheckReportYourTool]:
        """
        Main execution method that handles tool execution.

        Returns:
            List[CheckReportYourTool]: List of check reports from the tool scan
        """
        return self.run_scan(self.scan_path, self.exclude_path)

    def run_scan(
        self, directory: str, exclude_path: list[str]
    ) -> List[CheckReportYourTool]:
        """
        Execute the external tool and parse its output.

        Args:
            directory: Directory to scan
            frameworks: List of frameworks to scan
            exclude_path: List of paths to exclude

        Returns:
            List[CheckReportYourTool]: List of check reports
        """
        try:
            logger.info(f"Running YourTool scan on {directory} ...")

            # Build the tool command
            tool_command = [
                "your_tool_command",
                # Add your tool-specific arguments here, this are just examples
                "-d",
                directory,
                "-o",
                "json",
                "-f",
                ",".join(frameworks),
            ]
            if exclude_path:
                tool_command.extend(["--skip-path", ",".join(exclude_path)])

            # Run the tool with JSON output
            process = subprocess.run(
                tool_command,
                capture_output=True,
                text=True,
            )

            # Log tool's error output if any
            if process.stderr:
                logger.error(process.stderr)

            try:
                output = json.loads(process.stdout)
                if not output:
                    logger.warning("No findings returned from YourTool scan")
                    return []
            except Exception as error:
                logger.critical(
                    f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
                )
                sys.exit(1)

            reports = []

            # If only one framework has findings, the output is a dict, otherwise it's a list of dicts
            if isinstance(output, dict):
                output = [output]

            # Process all frameworks findings
            for finding in output:
                results = finding.get("results", {})

                # Process failed checks
                failed_checks = results.get("failed_checks", [])
                for failed_check in failed_checks:
                    report = self._process_check(finding, failed_check, "FAIL")
                    reports.append(report)

                # Process passed checks
                passed_checks = results.get("passed_checks", [])
                for passed_check in passed_checks:
                    report = self._process_check(finding, passed_check, "PASS")
                    reports.append(report)

                # Process skipped checks (muted)
                skipped_checks = results.get("skipped_checks", [])
                for skipped_check in skipped_checks:
                    report = self._process_check(finding, skipped_check, "MUTED")
                    reports.append(report)

            return reports

        except Exception as error:
            if "No such file or directory: 'your_tool_command'" in str(error):
                logger.critical("Please, install your_tool using 'pip install your_tool'")
                sys.exit(1)
            logger.critical(
                f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
            )
            sys.exit(1)

    def print_credentials(self):
        """
        Display scan information with color formatting.

        This method prints the tool scan information in a formatted way
        using colorama for better readability.
        """
        if self.scan_repository_url:
            report_title = (
                f"{Style.BRIGHT}Scanning remote repository:{Style.RESET_ALL}"
            )
            report_lines = [
                f"Repository: {Fore.YELLOW}{self.scan_repository_url}{Style.RESET_ALL}",
            ]
        else:
            report_title = (
                f"{Style.BRIGHT}Scanning local directory:{Style.RESET_ALL}"
            )
            report_lines = [
                f"Directory: {Fore.YELLOW}{self.scan_path}{Style.RESET_ALL}",
            ]

        if self.exclude_path:
            report_lines.append(
                f"Excluded paths: {Fore.YELLOW}{', '.join(self.exclude_path)}{Style.RESET_ALL}"
            )

        report_lines.append(
            f"Frameworks: {Fore.YELLOW}{', '.join(self.frameworks)}{Style.RESET_ALL}"
        )

        report_lines.append(
            f"Authentication method: {Fore.YELLOW}{self.auth_method}{Style.RESET_ALL}"
        )

        print_boxes(report_lines, report_title)