Plugin Development

class Plugin(metaclass=ABCMeta):
    """
    Base interface for all Certbot plugins.
    
    Objects providing this interface are called without satisfying
    entry point "extras" dependencies, so make sure they're importable
    without extras.
    """
    
    description: str  # Short plugin description
    name: str        # Unique name of the plugin
    
    def __init__(self, config: Optional[NamespaceConfig], name: str):
        """
        Create a new Plugin.
        
        Args:
            config: Configuration object
            name: Unique plugin name
        """
    
    @classmethod  
    def add_parser_arguments(cls, add: Callable[..., None]):
        """
        Add plugin-specific command line arguments.
        
        Args:
            add: Function to add arguments to parser
        """
    
    def prepare(self):
        """
        Prepare the plugin for use.
        
        Perform any necessary initialization before the plugin
        is used for authentication or installation.
        """
    
    def more_info(self) -> str:
        """
        Return additional information about the plugin.
        
        Returns:
            Human-readable string with more details
        """

class Authenticator(Plugin):
    """
    Interface for ACME challenge authenticator plugins.
    
    Authenticators prove control of domains through various challenge types
    like HTTP-01, DNS-01, or TLS-ALPN-01.
    """
    
    def get_chall_pref(self, domain: str) -> Iterable[type[Challenge]]:
        """
        Return iterable of challenge preferences for domain.
        
        Args:
            domain: Domain name for which to get preferences
            
        Returns:
            Iterable of preferred challenge types (Challenge subclasses) in order of preference
        """
    
    def perform(self, achalls: list[AnnotatedChallenge]) -> list[ChallengeResponse]:
        """
        Perform the given challenge.
        
        Args:
            achalls: List of annotated challenges to perform
            
        Returns:
            List of challenge responses
            
        Raises:
            errors.PluginError: If challenges cannot be performed
        """
    
    def cleanup(self, achalls: list[AnnotatedChallenge]):
        """
        Clean up after challenge completion.
        
        Args:
            achalls: List of annotated challenges to clean up
            
        Raises:
            errors.PluginError: If cleanup fails
        """

class Installer(Plugin):
    """
    Interface for certificate installer plugins.
    
    Installers deploy certificates to web servers and can apply
    enhancements like HTTPS redirects.
    """
    
    def get_all_names(self) -> Iterable[str]:
        """
        Get all domain names that the installer can handle.
        
        Returns:
            Iterable of domain names found in server configuration
        """
    
    def deploy_cert(self, domain: str, cert_path: str, key_path: str, 
                   chain_path: str, fullchain_path: str):
        """
        Deploy certificate for domain.
        
        Args:
            domain: Domain name for certificate
            cert_path: Path to certificate file
            key_path: Path to private key file  
            chain_path: Path to certificate chain file
            fullchain_path: Path to full certificate chain file
            
        Raises:
            errors.PluginError: If deployment fails
        """
    
    def enhance(self, domain: str, enhancement: str, 
               options: Optional[Union[list[str], str]] = None):
        """
        Apply enhancement to domain configuration.
        
        Args:
            domain: Domain name to enhance
            enhancement: Type of enhancement (e.g., 'redirect')
            options: Enhancement-specific options
            
        Raises:
            errors.PluginError: If enhancement fails
        """
    
    def supported_enhancements(self) -> list[str]:
        """
        Get list of supported enhancements.
        
        Returns:
            List of enhancement names supported by this installer
        """
    
    def save(self, title: Optional[str] = None, temporary: bool = False):
        """
        Save current configuration state.
        
        Args:
            title: Title for the save operation
            temporary: Whether this is a temporary save
            
        Raises:
            errors.PluginError: If save fails
        """
    
    def rollback_checkpoints(self, rollback: int = 1):
        """
        Rollback configuration to previous state.
        
        Args:
            rollback: Number of checkpoints to rollback
            
        Raises:
            errors.PluginError: If rollback fails
        """
    
    def recovery_routine(self):
        """
        Revert configuration to most recent finalized checkpoint.
        
        Remove all changes (temporary and permanent) that have not been
        finalized. Useful to protect against crashes and interruptions.
        
        Raises:
            errors.PluginError: If unable to recover the configuration
        """
    
    def config_test(self):
        """
        Test current configuration for validity.
        
        Raises:
            errors.MisconfigurationError: If configuration is invalid
        """
    
    def restart(self):
        """
        Restart the server to apply configuration changes.
        
        Raises:
            errors.MisconfigurationError: If restart fails
        """

class Plugin(interfaces.Plugin):
    """
    Base implementation class for plugins with common functionality.
    
    Provides default implementations and utility methods.
    """
    
    def __init__(self, config: NamespaceConfig, name: str):
        """Initialize plugin with configuration."""
        self.config = config
        self.name = name

class Installer(Plugin, interfaces.Installer):
    """
    Base class for installer plugins with common functionality.
    
    Provides shared installer methods and utilities.
    """
    
    def get_all_names_answer(self, question: str) -> set[str]:
        """
        Get domain names with user interaction.
        
        Args:
            question: Question to ask user about domain selection
            
        Returns:
            Set of selected domain names
        """

# In pyproject.toml:
[project.entry-points."certbot.plugins"]
my-authenticator = "my_package.plugin:MyAuthenticator"
my-installer = "my_package.plugin:MyInstaller"

# In setup.py:
entry_points={
    'certbot.plugins': [
        'my-authenticator = my_package.plugin:MyAuthenticator',
        'my-installer = my_package.plugin:MyInstaller',
    ],
}

from certbot import interfaces, errors
from certbot.plugins.common import Plugin
from certbot import configuration
from typing import Optional, Iterable, Union

class CustomAuthenticator(Plugin, interfaces.Authenticator):
    """Custom authenticator plugin example."""
    
    description = "Custom DNS authenticator"
    name = "custom-dns"
    
    def __init__(self, config: Optional[configuration.NamespaceConfig], name: str):
        super().__init__(config, name)
        self.credentials = None
    
    @classmethod
    def add_parser_arguments(cls, add):
        add('credentials', help='Path to credentials file')
    
    def prepare(self):
        """Prepare the authenticator."""
        # Load credentials, validate configuration, etc.
        if not self.config.credentials:
            raise errors.PluginError("Credentials file not specified")
    
    def perform(self, achalls):
        """Perform DNS challenges."""
        responses = []
        for achall in achalls:
            # Implement DNS record creation logic
            self._create_dns_record(achall.domain, achall.validation)
            response = achall.response_and_validation(self.account_key)
            responses.append(response)
        return responses
    
    def cleanup(self, achalls):
        """Clean up DNS records."""
        for achall in achalls:
            self._delete_dns_record(achall.domain)
    
    def _create_dns_record(self, domain: str, validation: str):
        """Create DNS TXT record for validation."""
        # Implement DNS provider-specific logic
        pass
    
    def _delete_dns_record(self, domain: str):
        """Delete DNS TXT record."""
        # Implement DNS provider-specific cleanup
        pass

class CustomInstaller(Plugin, interfaces.Installer):
    """Custom installer plugin example."""
    
    description = "Custom web server installer"
    name = "custom-server"
    
    def get_all_names(self) -> Iterable[str]:
        """Get all domain names from server configuration."""
        # Parse server configuration files
        return ['example.com', 'www.example.com']
    
    def deploy_cert(self, domain: str, cert_path: str, key_path: str, 
                   chain_path: str, fullchain_path: str):
        """Deploy certificate to server."""
        # Update server configuration with certificate paths
        pass
    
    def enhance(self, domain: str, enhancement: str, 
               options: Optional[Union[list[str], str]] = None):
        """Apply enhancements like HTTPS redirect."""
        if enhancement == 'redirect':
            # Implement HTTPS redirect configuration
            pass
    
    def supported_enhancements(self) -> list[str]:
        """Return list of supported enhancements."""
        return ['redirect', 'hsts']
    
    def save(self, title: Optional[str] = None, temporary: bool = False):
        """Save server configuration."""
        # Write configuration files
        pass
    
    def config_test(self):
        """Test server configuration."""
        # Validate configuration syntax
        pass
    
    def restart(self):
        """Restart the server."""
        # Restart server process
        pass

class AccountStorage(metaclass=ABCMeta):
    """
    Interface for ACME account storage implementations.
    
    Provides methods for finding, loading, and saving ACME accounts.
    """
    
    def find_all(self) -> list[Account]:
        """
        Find all stored accounts.
        
        Returns:
            List of all Account objects found in storage
        """
    
    def load(self, account_id: str) -> Account:
        """
        Load an account by its identifier.
        
        Args:
            account_id: Unique identifier for the account
        
        Returns:
            Account object for the specified ID
            
        Raises:
            errors.AccountNotFound: If account cannot be found
            errors.AccountStorageError: If account cannot be loaded
        """
    
    def save(self, account: Account, client: ClientV2) -> None:
        """
        Save account to storage.
        
        Args:
            account: Account object to save
            client: ACME client associated with account
            
        Raises:
            errors.AccountStorageError: If account cannot be saved
        """

class RenewableCert(metaclass=ABCMeta):
    """
    Interface to a certificate lineage for renewal operations.
    
    Provides access to certificate paths and metadata.
    """
    
    @property
    def cert_path(self) -> str:
        """Path to the certificate file."""
    
    @property
    def key_path(self) -> str:
        """Path to the private key file."""
    
    @property
    def chain_path(self) -> str:
        """Path to the certificate chain file."""
    
    @property
    def fullchain_path(self) -> str:
        """Path to the full chain file (cert + chain)."""
    
    @property
    def lineagename(self) -> str:
        """Name given to the certificate lineage."""
    
    def names(self) -> list[str]:
        """
        Get subject names of this certificate.
        
        Returns:
            List of domain names in the certificate
            
        Raises:
            errors.CertStorageError: If certificate file cannot be read
        """