dehydrated hook for IONOS dns-01 challenge

hook.py

#!/usr/bin/env python3

import requests
import sys
import logging
import os

#
# START of config-section - modify your needs!
#
API_KEY = '<prefix>.<key>'
DEBUG_LOG_ENABLED = False
#
# END of config section
#

API_URL = 'https://api.hosting.ionos.com/dns/v1'
HEADERS = {
  'accept': 'application/json',
  'X-API-Key': API_KEY,
  'Content-Type': 'application/json'
}

logging.basicConfig(stream=sys.stdout, level=logging.DEBUG if DEBUG_LOG_ENABLED else logging.INFO)
LOGGER = logging.getLogger('dehydrated/hook.py')

if (DEBUG_LOG_ENABLED):
    import http
    http.client.HTTPConnection.debuglevel = 1
    requests_log = logging.getLogger("requests.packages.urllib3")
    requests_log.setLevel(logging.DEBUG)
    requests_log.propagate = True

def extract_root_domain(domain):
    return '.'.join(domain.split('.')[-2:])

def find_zone(root):
    response = requests.get(f'{API_URL}/zones', headers = HEADERS)
    if not response.ok:
        LOGGER.error('Failed to read zones from IONOS API')
        return
    return next(filter(lambda zone: zone['name'] == root, response.json()), None)

def deploy_challenge(domain, token_filename, token_value):
    LOGGER.info(f'Creating ACME challenge DNS records for {domain}')

    root = extract_root_domain(domain)
    zone = find_zone(root)

    if not zone:
        LOGGER.error(f'No zone for {root} found')
        return

    txt_records = [
      {
      'name': f'_acme-challenge.{domain}',
      'type': 'txt',
      'content': token_value,
      'ttl': 3600,
      'prio': 0,
      'disabled': False
      }
    ]

    response = requests.post(f'{API_URL}/zones/{zone["id"]}/records', headers = HEADERS, json = txt_records)
    if not response.ok:
        LOGGER.error('Failed to set TXT records via IONOS API')
        return
    return

def clean_challenge(domain, token_filename, token_value):
    root = extract_root_domain(domain)
    LOGGER.info(f'Deleting all ACME challenge DNS records for {root}')

    zone = find_zone(root)

    if not zone:
        LOGGER.error(f'No zone for {root} found')
        return

    response = requests.get(f'{API_URL}/zones/{zone["id"]}', headers = HEADERS)
    if not response.ok:
        LOGGER.error('Failed to get records from zone via IONOS API')
        return

    records = filter(lambda record: record['name'].startswith('_acme-challenge'), response.json()['records'])
    for record in records:
        response = requests.delete(f'{API_URL}/zones/{zone["id"]}/records/{record["id"]}', headers = HEADERS)
    return

def deploy_cert(domain, privkey_pem, cert_pem, fullchain_pem, chain_pem, timestamp):
    LOGGER.info('Restarting NGINX')
    os.system('systemctl reload nginx')
    return

commands = {
  'deploy_challenge': deploy_challenge,
  'deploy_cert': deploy_cert,
  'clean_challenge': clean_challenge
}


command = sys.argv[1]
arguments = sys.argv[2:]

if command in commands:
    LOGGER.info('Executing hook %s(%s)', command, ', '.join(arguments))
    commands[command](*arguments)
else:
    LOGGER.debug('Unsupported hook %s', command)