Dynamic DNS address updater for Porkbun

porkbun_ddns.py

import os.path
import yaml
import requests
import logging
from typing import Tuple

logging.basicConfig(level=logging.INFO)


class Porkbun():

    def __init__(self,
                 api_key: str,
                 api_secret: str,
                 api_url: str = 'https://api-ipv4.porkbun.com/api/json/v3'):
        self.api_key = api_key
        self.api_secret = api_secret
        self.api_auth = {
            'apikey': self.api_key,
            'secretapikey': self.api_secret,
        }
        self.api_url = api_url

    def getMyIP(self):
        ping = requests.post(self.api_url + '/ping', json=self.api_auth).json()
        return (ping["yourIp"])

    def getCurrentDNS(self, domain: str):
        sub_domain, root_domain = Porkbun.splitDomain(domain)
        records = requests.post(self.api_url + '/dns/retrieveByNameType/' +
                                root_domain + '/A/' + sub_domain,
                                json=self.api_auth).json()
        records = [x['content'] for x in records['records']]
        return records

    @staticmethod
    def splitDomain(domain: str) -> Tuple[str, str]:
        ''' Returns sub_domain, root_domain. '''
        dsplit = domain.split('.')
        return '.'.join(dsplit[:-2]), '.'.join(dsplit[-2:])

    def updateDNS(self, domain: str, newIP: str):
        current_record = self.getCurrentDNS(domain)

        sub_domain, root_domain = Porkbun.splitDomain(domain)
        if len(current_record) == 0:
            # create new record
            logging.info('Creating new record at %s with %s.' %
                         (domain, newIP))
            payload = {
                'name': sub_domain,
                'type': 'A',
                'content': newIP,
                'ttl': 60
            }
            payload.update(self.api_auth)
            return requests.post(self.api_url + '/dns/create/' + root_domain,
                                 json=payload).json()
        elif current_record[0] != newIP:
            # update existing record
            logging.info('Updating %s from %s => %s.' %
                         (domain, current_record[0], newIP))
            payload = {'content': newIP, 'ttl': 60}
            payload.update(self.api_auth)
            return requests.post(self.api_url + '/dns/editByNameType/' +
                                 root_domain + '/A/' + sub_domain,
                                 json=payload).json()
        else:
            # no change
            logging.info(
                '%s already matches specified IP (%s). No changes made.' %
                (domain, newIP))
            return None


if __name__ == '__main__':
    # Load config
    with open(os.path.join(os.path.dirname(__file__), 'config.yml'), 'r') as f:
        k = yaml.safe_load(f)

    porkbun = Porkbun(k['porkbun_apikey']['key'],
                      k['porkbun_apikey']['secret'])
    current_ip = porkbun.getMyIP()
    res = porkbun.updateDNS(k['domain'], current_ip)
    if res is None:
        pass
    elif res['status'] != 'SUCCESS':
        logging.error(res)
    else:
        logging.info(res)