acmebot with dns-01 zone delegation

acmebot_ex.py

#!/usr/bin/env python3

import sys

# Assuming https://github.com/plinss/acmebot.git is cloned to /original/acmebot
# Please also create a softlink from acmebot.py to acmebot, such as by doing
#   ln -s /original/acmebot/acmebot /original/acmebot/acmebot.py

sys.path.insert(0, "/original/acmebot")
from acmebot import AcmeManager

"""
Copyright (c) 2020 IT9 <[email protected]>

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
"""

"""
Support redirecting dns-01 challenges answering zone by CNAME

Example:
1. Set up a CNAME at _acme-challenge.example.com to somewhere._acme.foobar.com
2. Zone Key must be provided for the targeted zone, not the original one
```
   "zone_update_keys": { "_acme.foobar.com": "example.key" }
```
3. In case the internal/external view is difference, CNAME resolution could be forced by:
```
    "cname_override": {
        "_acme-challenge.example.com": "somewhere._acme.foobar.com"
    }
```

"""
class AcmeManagerEx(AcmeManager):
    def _handle_authorizations(self, order, fetch_only, domain_names):
        new_domain_names = collections.OrderedDict()
        for zone_name in domain_names:
            for domain_name in domain_names[zone_name]:
                new_zone_name = zone_name
                identifier = domain_name.replace('*.','')
                http_challenge_directory = self._http_challenge_directory(identifier, zone_name)
                if not http_challenge_directory:
                    rr = self._resolve_host_rr(f'_acme-challenge.{identifier}', zone_name)
                    new_zone_name = rr['zone_name']
                if not new_zone_name in new_domain_names:
                    new_domain_names[new_zone_name] = []
                new_domain_names[new_zone_name].append(domain_name)
        return super()._handle_authorizations(order, fetch_only, new_domain_names)

    def _set_dns_challenges(self, zone_name, zone_key, challenges):
        updates_per_zone = dict()
        for domain_name in challenges:
            rr = self._resolve_host_rr(f'_acme-challenge.{challenges[domain_name].identifier}', zone_name)
            challenge_host = rr['host']
            response = challenges[domain_name].response
            if not rr['zone_name'] in updates_per_zone:
                updates_per_zone[rr['zone_name']] = []
            updates_per_zone[rr['zone_name']].append(f'update add {challenge_host} 300 TXT "{response}"')

        for this_zone_name, updates in updates_per_zone.items():
            this_zone_key = self._zone_key(this_zone_name)
            if not self._update_zone(updates, this_zone_name, this_zone_key, 'Set DNS challenges'):
                return False
        return True

    def _remove_dns_challenges(self, zone_name, zone_key, challenges):
        updates_per_zone = dict()
        for domain_name in challenges:
            rr = self._resolve_host_rr(f'_acme-challenge.{challenges[domain_name].identifier}', zone_name)
            challenge_host = rr['host']
            response = challenges[domain_name].response
            if not rr['zone_name'] in updates_per_zone:
                updates_per_zone[rr['zone_name']] = []
            updates_per_zone[rr['zone_name']].append(f'update delete {challenge_host} 300 TXT "{response}"')

        for this_zone_name, updates in updates_per_zone.items():
            this_zone_key = self._zone_key(this_zone_name)
            if not self._update_zone(updates, this_zone_name, this_zone_key, 'Set DNS challenges'):
                return False
        return True

    def _lookup_dns_challenge(self, name_server, domain_name):
        rr = self._resolve_host_rr(f'_acme-challenge.{domain_name}', None)
        response, _ = self._dns_request(rr['host'], 'TXT', name_server)
        if (response):
            return [answer['data'][0].decode('ascii') for answer in response.answers]
        return []
    def _get_zone_name(self, domain_name):
        domain_name_parts = domain_name.split('.')
        for pos in range(len(domain_name_parts)):
            host = '.'.join(domain_name_parts[pos:])
            response, status = self._dns_request(host, 'SOA')
            if (response) and len(response.answers):
                return host
        self._warn('Unable to find zone name for ', domain_name, '\n')
        return None

    def _resolve_host_rr(self, host, zone_name):
        override = self._config('cname_override', host)
        if override:
            if isinstance(override, str):
                return {'host': override, 'zone_name': self._get_zone_name(override)}
            return override

        response, status = self._dns_request(host, 'CNAME')
        if response and len(response.answers):
            override = response.answers[0]['data']
            return {'host': override, 'zone_name': self._get_zone_name(override)}

        return {'host': host, 'zone_name': zone_name}

if __name__ == '__main__':      # called from the command line
    sys.exit(AcmeManagerEx.Run())