Module ___init___

def setup_custom_logger(name):
    """Setting up custom logger. Using INFO level as default when Log Level is not set via environment variable.

    Environment variables:

    ACME_DNS_AZURE_LOG_LEVEL -- Log level for all classes.

    Allowed values:

        "DEBUG"
        "INFO"
        "WARN"
        "WARNING"
        "ERROR"
        "CRITICAL"
        "FATAL"
    """
    log_level = logging.INFO
    custom_level = os.environ.get("ACME_DNS_AZURE_LOG_LEVEL", None)
    if custom_level and custom_level in [
        "DEBUG",
        "INFO",
        "WARN",
        "WARNING",
        "ERROR",
        "CRITICAL",
        "FATAL",
    ]:
        log_level = logging.getLevelName(custom_level)
        logging.info("Setting defined loglevel '%s'.", log_level)

    formatter = logging.Formatter(
        fmt="%(asctime)s - %(levelname)s - %(module)s - %(message)s"
    )

    handler = logging.StreamHandler()
    handler.setFormatter(formatter)

    logger = logging.getLogger(name)
    logger.propagate = False
    logger.setLevel(log_level)
    logger.addHandler(handler)
    return logger

class AcmeDnsAzureClient:
    """Client for auto renewal of certificates. One of possible config params must be set.


    Keyword arguments:

    config_yaml -- Config based on schema as yaml string

    config_env_var -- Env var name containing base64 encoded config based on schema as yaml

    config_file -- Config path reference based on schema to yaml file

    file_path_prefix -- Path prefix for creating working dir witin /tmp dir. (default acme_dns_azure)
    """

    def __init__(
        self,
        config_yaml: str = None,
        config_env_var: str = None,
        config_file: str = None,
        file_path_prefix: str = "acme_dns_azure",
    ) -> None:
        self.ctx = Context()
        self._work_dir = tempfile.TemporaryDirectory(prefix=file_path_prefix)
        logger.info(
            "Setting working directory for certicate renewal: %s", self._work_dir
        )
        self.ctx.work_dir = self._work_dir.name

        if config_yaml is not None:
            self.ctx.config = config.load(config_yaml)
        elif config_env_var is not None:
            self.ctx.config = config.load_from_base64_env_var(config_env_var)
        elif config_file is not None:
            self.ctx.config = config.load_from_file(config_file)
        else:
            raise ConfigurationError("No configuration source defined")

        self.ctx.azure_credentials = DefaultAzureCredential()
        self.ctx.keyvault = KeyVaultManager(self.ctx)
        self.certbot = CertbotManager(self.ctx)

    def __del__(self):
        logger.info("Cleaning up directory %s", self.ctx.work_dir)

    def issue_certificates(self) -> List[RotationResult]:
        """Create/rotate all certificates based on initial client configuration."""
        logger.info("Issuing certificates...")
        return self.certbot.renew_certificates()

class CertbotResult(Enum):
    """Certbot renewal result."""

    CREATED = 1
    """
    New certificate has been created.
    """
    RENEWED = 2
    """
    Existing certificate has been renewed.
    """
    STILL_VALID = 3
    """
    Existing certificate is still valid. No action taken.
    """
    FAILED = 4
    """
    Certbot creation or renewal of certificate has failed.
    """
    SKIPPED = 5
    """
    Existing certificate has been skipped due to mismatch of domain information of provided config.
    """

@dataclass
class DomainReference:
    """Dataclass holding Domain name - DNS zone record resource ID reference.

    params:
    dns_zone_resource_id -- resource ID of DNS Zone record
    domain_name -- domain name
    """

    dns_zone_resource_id: str
    domain_name: str

@dataclass
class RotationCertificate:
    """Dataclass holding certificate rotation information.

    params:
    key_vault_cert_name -- Name of keyvault certificate to be created/updated
    certbot_cert_name -- Certificate name of certbot
    domains -- Domain references of this certificate
    renew_before_expiry -- Number in days before expiration when this certificate will be renewed.
    """

    key_vault_cert_name: str
    certbot_cert_name: str
    domains: List[DomainReference]
    renew_before_expiry: int = None

@dataclass
class RotationResult:
    """Dataclass holding certificate rotation result information.

    params:
    certificate -- Rotation certificate reference
    result -- Result of rotation action
    message -- Message with additional information to result
    """

    certificate: RotationCertificate
    result: CertbotResult
    message: str = None