___init___

19class AcmeDnsAzureClient:
20    """Client for auto renewal of certificates. One of possible config params must be set.
21
22
23    Keyword arguments:
24
25    config_yaml -- Config based on schema as yaml string
26
27    config_env_var -- Env var name containing base64 encoded config based on schema as yaml
28
29    config_file -- Config path reference based on schema to yaml file
30
31    file_path_prefix -- Path prefix for creating working dir witin /tmp dir. (default acme_dns_azure)
32    """
33
34    def __init__(
35        self,
36        config_yaml: str = None,
37        config_env_var: str = None,
38        config_file: str = None,
39        file_path_prefix: str = "acme_dns_azure",
40    ) -> None:
41        self.ctx = Context()
42        self._work_dir = tempfile.TemporaryDirectory(prefix=file_path_prefix)
43        logger.info(
44            "Setting working directory for certicate renewal: %s", self._work_dir
45        )
46        self.ctx.work_dir = self._work_dir.name
47
48        if config_yaml is not None:
49            self.ctx.config = config.load(config_yaml)
50        elif config_env_var is not None:
51            self.ctx.config = config.load_from_base64_env_var(config_env_var)
52        elif config_file is not None:
53            self.ctx.config = config.load_from_file(config_file)
54        else:
55            raise ConfigurationError("No configuration source defined")
56
57        self.ctx.azure_credentials = DefaultAzureCredential()
58        self.ctx.keyvault = KeyVaultManager(self.ctx)
59        self.certbot = CertbotManager(self.ctx)
60
61    def __del__(self):
62        logger.info("Cleaning up directory %s", self.ctx.work_dir)
63
64    def issue_certificates(self) -> List[RotationResult]:
65        """Create/rotate all certificates based on initial client configuration."""
66        logger.info("Issuing certificates...")
67        return self.certbot.renew_certificates()

62@dataclass
63class RotationResult:
64    """Dataclass holding certificate rotation result information.
65
66    params:
67    certificate -- Rotation certificate reference
68    result -- Result of rotation action
69    message -- Message with additional information to result
70    """
71
72    certificate: RotationCertificate
73    result: CertbotResult
74    message: str = None

 7@dataclass
 8class DomainReference:
 9    """Dataclass holding Domain name - DNS zone record resource ID reference.
10
11    params:
12    dns_zone_resource_id -- resource ID of DNS Zone record
13    domain_name -- domain name
14    """
15
16    dns_zone_resource_id: str
17    domain_name: str

20@dataclass
21class RotationCertificate:
22    """Dataclass holding certificate rotation information.
23
24    params:
25    key_vault_cert_name -- Name of keyvault certificate to be created/updated
26    certbot_cert_name -- Certificate name of certbot
27    domains -- Domain references of this certificate
28    renew_before_expiry -- Number in days before expiration when this certificate will be renewed.
29    """
30
31    key_vault_cert_name: str
32    certbot_cert_name: str
33    domains: List[DomainReference]
34    renew_before_expiry: int = None

37class CertbotResult(Enum):
38    """Certbot renewal result."""
39
40    CREATED = 1
41    """
42    New certificate has been created.
43    """
44    RENEWED = 2
45    """
46    Existing certificate has been renewed.
47    """
48    STILL_VALID = 3
49    """
50    Existing certificate is still valid. No action taken.
51    """
52    FAILED = 4
53    """
54    Certbot creation or renewal of certificate has failed.
55    """
56    SKIPPED = 5
57    """
58    Existing certificate has been skipped due to mismatch of domain information of provided config.
59    """

 6def setup_custom_logger(name):
 7    """Setting up custom logger. Using INFO level as default when Log Level is not set via environment variable.
 8
 9    Environment variables:
10
11    ACME_DNS_AZURE_LOG_LEVEL -- Log level for all classes.
12
13    Allowed values:
14
15        "DEBUG"
16        "INFO"
17        "WARN"
18        "WARNING"
19        "ERROR"
20        "CRITICAL"
21        "FATAL"
22    """
23    log_level = logging.INFO
24    custom_level = os.environ.get("ACME_DNS_AZURE_LOG_LEVEL", None)
25    if custom_level and custom_level in [
26        "DEBUG",
27        "INFO",
28        "WARN",
29        "WARNING",
30        "ERROR",
31        "CRITICAL",
32        "FATAL",
33    ]:
34        log_level = logging.getLevelName(custom_level)
35        logging.info("Setting defined loglevel '%s'.", log_level)
36
37    formatter = logging.Formatter(
38        fmt="%(asctime)s - %(levelname)s - %(module)s - %(message)s"
39    )
40
41    handler = logging.StreamHandler()
42    handler.setFormatter(formatter)
43
44    logger = logging.getLogger(name)
45    logger.propagate = False
46    logger.setLevel(log_level)
47    logger.addHandler(handler)
48    return logger