puppet-certbot/files/acme-cleanup-loopia.py

#!/usr/bin/env python3.8

import argparse
import configparser
import os
import sys
import time
import xmlrpc.client
import http.client

class ProxiedTransport(xmlrpc.client.Transport):

    def set_proxy(self, host, port=None, headers=None):
        self.proxy = host, port
        self.proxy_headers = headers

    def make_connection(self, host):
        connection = http.client.HTTPSConnection(*self.proxy)
        connection.set_tunnel(host, headers=self.proxy_headers)
        self._connection = host, connection
        return connection

def main():

    if all([x in os.environ for x in ['CERTBOT_DOMAIN', 'CERTBOT_VALIDATION']]):
        domain = os.environ['CERTBOT_DOMAIN']
        subdomain = '_acme-challenge'
        token = os.environ['CERTBOT_VALIDATION']
    else:
        parser = argparse.ArgumentParser(description='Update acme-record for subdomain')
        parser.add_argument('--domain', '-d', nargs=1, required=True, help='domain to update')
        parser.add_argument('--token', '-t', nargs=1, required=True, help='token to set as txt record')

        args = parser.parse_args()

        domain = args.domain[0]
        subdomain = '_acme-challenge'
        token = args.token[0]


    config = configparser.ConfigParser()
    config.read('/usr/local/etc/loopiaapi.ini')
    url = config.get('default', 'url')
    user = config.get('default', 'username')
    pwd = config.get('default', 'password')

    proxy = os.environ.get('http_proxy')
    if not proxy:
        proxy = os.environ.get('HTTP_PROXY')

    if proxy:
        transport = ProxiedTransport()
        proto, host, port = proxy.split(':')
        transport.set_proxy(host.strip('/'), int(port))
        client = xmlrpc.client.ServerProxy(uri = url, encoding='utf-8', transport=transport)
    else:
        client = xmlrpc.client.ServerProxy(uri = url, encoding='utf-8')

    parts=[]
    while domain:
        try:
            res = client.getSubdomains(user, pwd, domain)
        except xmlrpc.client.Fault as err:
            if err.faultCode != 404:
                raise err
        else:
            break
        part, domain = domain.split('.', maxsplit=1)
        parts.append(part)
        subdomain = '_acme-challenge.{}'.format('.'.join(parts))

    if not res:
        print("Failed to find domain in loopiadns")
        return 1

    if subdomain not in res:
        return 0

    res = client.getZoneRecords(user, pwd, domain, subdomain)

    ret = 0
    for rec in res:
        if rec['type'] == 'TXT':
            if rec['rdata'] == token:
                res = client.removeZoneRecord(user, pwd, domain, subdomain, rec['record_id'])
                if res != 'OK':
                    print('Failed to clean up record, loopia response: {}'.format(res))
                    ret = 1
                res = client.removeSubdomain(user, pwd, domain, subdomain)
                if res != 'OK':
                    print('Failed to clean up subdomain, loopia response: {}'.format(res))
                    ret = 1
                return ret
    return ret



if __name__ == '__main__':
    sys.exit(main())