Refactor handling of idna domain names

README.md

@@ -12,14 +12,14 @@ Requirements
 ------------
 
   * Python (2.7+ and 3.5+ should work)
-  * cryptography>=0.6 (usually includes the optional idna module)
+  * cryptography>=0.6
+  * idna (usually included in cryptography)
 
 Optional requirements (to use specified features)
 ------------------------------------------------------
 
   * PyYAML: to parse YAML-formatted configuration files
   * dnspython: used by dns.* challenge handlers
-  * idna: to allow automatic conversion of unicode domain names to their IDNA2008 counterparts
   * cryptography>=2.1: for creating certificates with the OCSP must-staple flag (cert_must_staple)
   * cryptography>=2.6: for usage of Ed25519/Ed448 keys
 
@@ -121,4 +121,4 @@ Please keep the following in mind when using this software:
      * Create a dedicated user for acertmgr (e.g. acertmgr)
      * Run a acertmgr as that user (add acertmgr to that users cron!)
      * Access rights to read/write all files configured with the created user
-     * Run any programs/scripts defined on cert update as the created user (might need work-arounds with sudo or wrapper scripts)
+     * Run any programs/scripts defined on cert update as the created user (might need work-arounds with sudo or wrapper scripts)

acertmgr/__init__.py

@@ -28,14 +28,14 @@
 # @brief fetch new certificate from letsencrypt
 # @param settings the domain's configuration options
 def cert_get(settings):
-    log("Getting certificate for %s" % settings['domainlist'])
+    log("Getting certificate for %s" % settings['domainlist_human'])
 
     acme = authority(settings['authority'])
     acme.register_account()
 
     # create challenge handlers for this certificate
     challenge_handlers = dict()
-    for domain in settings['domainlist']:
+    for domain in settings['domainlist_idna']:
         # Create the challenge handler
         challenge_handlers[domain] = challenge_handler(settings['handlers'][domain])
 
@@ -53,13 +53,13 @@ def cert_get(settings):
         log('Loading CSR from {}'.format(csr_file))
         cr = tools.read_pem_file(csr_file, csr=True)
     else:
-        log('Generating CSR for {}'.format(settings['domainlist']))
+        log('Generating CSR for {}'.format(settings['domainlist_human']))
         must_staple = str(settings.get('cert_must_staple')).lower() == "true"
-        cr = tools.new_cert_request(settings['domainlist'], key, must_staple)
+        cr = tools.new_cert_request(settings['domainlist_idna'], key, must_staple)
         tools.write_pem_file(cr, csr_file)
 
     # request cert with csr
-    crt, ca = acme.get_crt_from_csr(cr, settings['domainlist'], challenge_handlers)
+    crt, ca = acme.get_crt_from_csr(cr, settings['domainlist_idna'], challenge_handlers)
 
     #  if resulting certificate is valid: store in final location
     if tools.is_cert_valid(crt, settings['ttl_days']):
@@ -123,7 +123,7 @@ def cert_revoke(cert, configs, fallback_authority, reason=None):
     domains = set(tools.get_cert_domains(cert))
     acmeconfig = None
     for config in configs:
-        if domains == set(config['domainlist']):
+        if domains == set(config['domainlist_idna']):
             acmeconfig = config['authority']
             break
     if not acmeconfig:
@@ -170,7 +170,7 @@ def main():
                             log("Failed to download issuer for cert file: {}. Cannot validate OCSP.".format(e2))
                             validate_ocsp = False
                 if not cert or ('force_renew' in runtimeconfig and all(
-                        d in config['domainlist'] for d in runtimeconfig['force_renew'])) \
+                        d in config['domainlist_idna'] for d in runtimeconfig['force_renew'])) \
                         or not tools.is_cert_valid(cert, config['ttl_days']) \
                         or (validate_ocsp and not tools.is_ocsp_valid(cert, issuer, config['validate_ocsp'])):
                     cert_get(config)

acertmgr/configuration.py

@@ -86,13 +86,11 @@ def parse_config_entry(entry, globalconfig, runtimeconfig):
 
     # Basic domain information
     domains, localconfig = entry
-    config['domainlist'] = domains.split(' ')
+    config['domainlist_human'] = domains.split(' ')
     config['id'] = hashlib.md5(domains.encode('utf-8')).hexdigest()
 
     # Convert unicode to IDNA domains
-    config['domaintranslation'] = idna_convert(config['domainlist'])
-    if len(config['domaintranslation']) > 0:
-        config['domainlist'] = [x for x, _ in config['domaintranslation']]
+    config['domainlist_idna'] = list(map(idna_convert, config['domainlist_human']))
 
     # Action config defaults
     config['defaults'] = globalconfig.get('defaults', {})
@@ -151,8 +149,7 @@ def parse_config_entry(entry, globalconfig, runtimeconfig):
     # Domain challenge handler configuration
     config['handlers'] = dict()
     handlerconfigs = [x for x in localconfig if 'mode' in x]
-    _domaintranslation_dict = {x: y for x, y in config.get('domaintranslation', [])}
-    for domain in config['domainlist']:
+    for domain_human, domain_idna in zip(config['domainlist_human'], config['domainlist_idna']):
         # Use global config as base handler config
         cfg = copy.deepcopy(globalconfig)
 
@@ -161,13 +158,11 @@ def parse_config_entry(entry, globalconfig, runtimeconfig):
         if len(genericfgs) > 0:
             cfg.update(genericfgs[0])
 
-        # Update handler config with more specific values (use original names for translated unicode domains)
-        _domain = _domaintranslation_dict.get(domain, domain)
-        specificcfgs = [x for x in handlerconfigs if 'domain' in x and x['domain'] == _domain]
+        specificcfgs = [x for x in handlerconfigs if 'domain' in x and x['domain'] == domain_human]
         if len(specificcfgs) > 0:
             cfg.update(specificcfgs[0])
 
-        config['handlers'][domain] = cfg
+        config['handlers'][domain_idna] = cfg
 
     return config
 
@@ -222,11 +217,7 @@ def load():
 
     # - force-rewew
     if args.force_renew:
-        domaintranslation = idna_convert(args.force_renew.split(' '))
-        if len(domaintranslation) > 0:
-            runtimeconfig['force_renew'] = [x for x, _ in domaintranslation]
-        else:
-            runtimeconfig['force_renew'] = args.force_renew.split(' ')
+        runtimeconfig['force_renew'] = list(map(idna_convert, args.force_renew.split(' ')))
 
     # - revoke
     if args.revoke:

acertmgr/tools.py

@@ -384,28 +384,15 @@ def target_is_current(target, file):
     crt_date = os.path.getmtime(file)
     return target_date >= crt_date
 
-
-# @brief convert domain list to idna representation (if applicable
-def idna_convert(domainlist):
-    if any(ord(c) >= 128 for c in ''.join(domainlist)):
-        try:
-            domaintranslation = list()
-            for domain in domainlist:
-                if any(ord(c) >= 128 for c in domain):
-                    # Translate IDNA domain name from a unicode domain (handle wildcards separately)
-                    if domain.startswith('*.'):
-                        idna_domain = "*.{}".format(domain[2:].encode('idna').decode('ascii'))
-                    else:
-                        idna_domain = domain.encode('idna').decode('ascii')
-                    result = idna_domain, domain
-                else:
-                    result = domain, domain
-                domaintranslation.append(result)
-            return domaintranslation
-        except Exception as e:
-            log("Unicode domain(s) found but IDNA names could not be translated due to error: {}".format(e), error=True)
-    return [(x, x) for x in domainlist]
-
+# @brief convert a domain to idna representation
+def idna_convert(domain):
+    try:
+        splitdomain = re.fullmatch('(\*\.)?(.*)', domain) # split wildcard off domain
+        idna_domain = (splitdomain.group(1) or '') + splitdomain.group(2).encode('idna').decode('ascii')
+        return idna_domain
+    except Exception as e:
+        log("Unicode domain found but IDNA names could not be translated due to error: {}".format(e), error=True)
+        raise
 
 # @brief validate the OCSP status for a given certificate by the given issuer
 def is_ocsp_valid(cert, issuer, hash_algo):