A Dynamic DNS service using the GoDaddy API, set up to run on Heroku

godaddy.py

import json

import treq
from twisted.web.http_headers import Headers
from twisted.logger import Logger


class GodaddyAPICallFailed(Exception):
    pass


class GodaddyDNSAPIClient(object):
    log = Logger()
    API_BASE_URL = 'https://api.godaddy.com/v1/domains'

    def __init__(self, api_key, api_secret):
        self._auth_header_value = 'sso-key {key}:{secret}'.format(
            key=api_key, secret=api_secret)
        headers = Headers()
        headers.addRawHeader('Authorization', self._auth_header_value)
        headers.addRawHeader('Accept', 'application/json')
        headers.addRawHeader('Content-Type', 'application/json')
        self._base_headers = headers


    def setARecord(self, domain, hostname, ipv4_address):
        url = '/'.join([self.API_BASE_URL, domain, 'records', 'A', hostname])
        payload = json.dumps([{
            'data': ipv4_address, 'name': hostname, 'ttl': 3600, 'type': 'A'
        }])
        headers = self._base_headers.copy()

        self.log.info(
            u"Updating {domain!r} with IPv4 address {ipv4!r}",
            domain='.'.join([hostname, domain]),
            ipv4=ipv4_address)

        d = treq.put(url, headers=headers, data=payload)
        d.addCallback(self._cbCheckStatus, 'A')
        d.addCallback(self._cbLogSuccess, 'A')
        return d

    def setAAAARecord(self, domain, hostname, ipv6_address):
        url = '/'.join([
            self.API_BASE_URL, domain, 'records', 'AAAA', hostname])
        payload = json.dumps([{
            'data': ipv6_address,
            'name': hostname,
            'ttl': 3600,
            'type': 'AAAA'
        }])
        headers = self._base_headers.copy()

        self.log.info(
            u"Updating {domain!r} with IPv6 address {ipv6!r}",
            domain='.'.join([hostname, domain]),
            ipv6=ipv6_address)

        d = treq.put(url, headers=headers, data=payload)
        d.addCallback(self._cbCheckStatus, 'AAAA')
        d.addCallback(self._cbLogSuccess, 'AAAA')
        return d

    def _cbCheckStatus(self, response, recordType):
        if response.code != 200:
            fmt = "Status code {code} on DNS {rtype} record update attempt"
            raise GodaddyAPICallFailed(fmt.format(
                code=response.code,
                rtype=recordType,
            ))
        return response.content()

    def _cbLogSuccess(self, content, recordType):
        fmt = (
            u"Successfully updated DNS {rtype} record, response from API: "
            u"{content!r}"
        )
        self.log.info(fmt, rtype=recordType, content=content)
        return content

miscserver.py

import re
import os

from twisted.logger import Logger
from twisted.internet import defer
from klein import Klein

from godaddy import GodaddyDNSAPIClient


ALLOWED_DDNS_DOMAINS = {
    'mydomain.net': frozenset(['mysubdomain']),
}
GODADDY_API_KEY = os.environ['GODADDY_API_KEY']
GODADDY_API_SECRET = os.environ['GODADDY_API_SECRET']

godaddy = GodaddyDNSAPIClient(GODADDY_API_KEY, GODADDY_API_SECRET)
log = Logger()
app = Klein()


class BadRequest(Exception):
    pass


class InternalServerError(Exception):
    pass


@app.handle_errors(BadRequest)
def handle_forbidden(request, failure):
    request.setResponseCode(400)
    return '400 Bad Request'


@app.handle_errors(InternalServerError)
def handle_internal_server_error(request, failure):
    request.setResponseCode(500)
    return '500 Internal Server Error'


@app.route('/')
def home(request):
    return 'Hello, world!'


@app.route('/ddns')
def update_ddns(request):
    requested_ipv4_address = request.args.get('ipv4', [''])[0]
    if not is_valid_ipv4_address(requested_ipv4_address):
        raise BadRequest('Invalid IPv4 address {addr!r}'.format(
            addr=requested_ipv4_address
        ))
    ipv4_address = requested_ipv4_address

    requested_ipv6_address = request.args.get('ipv6', [''])[0]
    if requested_ipv6_address.strip():
        try:
            ipv6_address = expand_ipv6_address(requested_ipv6_address)
        except ValueError as e:
            raise BadRequest('Invalid IPv6 address {0!r}'.format(e))
    else:
        ipv6_address = None

    requested_domain = request.args.get('domain', [''])[0]
    hostname, _, domain = requested_domain.partition('.')
    if hostname not in ALLOWED_DDNS_DOMAINS.get(domain, []):
        raise BadRequest('Invalid domain requested: {domain!r}'.format(
            domain=requested_domain
        ))

    dfds = [godaddy.setARecord(domain, hostname, ipv4_address)]
    if ipv6_address is not None:
        dfds.append(
            godaddy.setAAAARecord(domain, hostname, ipv6_address)
        )

    def fail_with_internal_server_error(failure):
        log.failure("Error communicating with GoDaddy API", failure=failure)
        raise InternalServerError("Failed to update DNS record")

    def succeed_if_good(ignored):
        return 'Success'

    for d in dfds:
        d.addErrback(fail_with_internal_server_error)
    # TODO: Fix this, it probably doesn't fail right because of
    #       how DeferredList works

    updateDfd = defer.gatherResults(dfds, consumeErrors=True)
    updateDfd.addCallback(succeed_if_good)
    return updateDfd


def is_valid_ipv4_address(address):
    return (
        re.match(r'\d+\.\d+\.\d+\.\d+$', address) and
        all(0 <= n < 256 for n in map(int, address.split('.')))
    )


def expand_ipv6_address(address):
    given = address
    address = address.lower().strip()
    if not re.match(r'^[a-f0-9:]+$', address):
        raise ValueError(
            'Invalid IPv6 address {0!r}: bad characters'.format(given))
    front, sep, back = address.partition('::')
    frontparts = front.split(':')
    if sep:
        backparts = back.split(':')
    else:
        backparts = []
    if not all(frontparts + backparts):
        raise ValueError(
            'Invalid IPv6 address {0!r}: empty parts'.format(given))
    nexplicitparts = len(frontparts) + len(backparts)
    if (sep and nexplicitparts > 7) or (not sep and nexplicitparts != 8):
        fmt = 'Invalid IPv6 address {0!r}: wrong number of segments'
        raise ValueError(fmt.format(given))
    filler = ['0' for _ in range(8 - nexplicitparts)]
    parts = frontparts + filler + backparts
    if not all(re.match('^[a-f0-9]{1,4}$', part) for part in parts):
        raise ValueError(
            'Invalid IPv6 address {0!r}: bad part(s)'.format(given))
    parts = [part.rjust(4, '0') for part in parts]
    return ':'.join(parts)


resource = app.resource

Procfile

web: twistd -n web -p $PORT --class=miscserver.resource

requirements.txt

attrs==16.2.0
cffi==1.9.1
constantly==15.1.0
cryptography==1.5.3
enum34==1.1.6
idna==2.1
incremental==16.10.1
ipaddress==1.0.17
klein==15.3.1
pyasn1==0.1.9
pyasn1-modules==0.0.8
pycparser==2.17
pyOpenSSL==16.2.0
requests==2.11.1
service-identity==16.0.0
six==1.10.0
treq==15.1.0
Twisted==16.5.0
Werkzeug==0.11.11
zope.interface==4.3.2