Crontab-ready Python script to update AutoDNS / InternetX domains

autodns.py

# -*- coding: utf-8 -*-
"""Update script for AutoDNS / InternetX.

Requires:
- https://pypi.org/project/requests/
- https://pypi.org/project/click/
"""

import logging
from re import match
from sys import exit
from typing import Union
from xml.dom.minidom import getDOMImplementation, parseString

import click
from requests import get, post

GATEWAY = 'https://gateway.autodns.com'
"""Default AutoDNS gateway"""
GATEWAY_IP = ' http://whatismyip.akamai.com/'
"""Default gateway to obtain current IP"""

log = logging.getLogger(__name__)


@click.command(name='autodns', help=__doc__)
@click.option('--ip-file', '-i', required=True,
              metavar='LASTIP_FILE', help='File to store the new IP address')
@click.option('--username', '-u', required=True,
              metavar='USERNAME', help='AutoDNS username')
@click.option('--password', '-p', required=True,
              metavar='PASSWORD', help='AutoDNS password')
@click.option('--domain', '-d', required=True,
              metavar='DOMAIN', help='Zone domain')
def main(ip_file: str,
         username: str,
         password: str,
         domain: str) -> None:
    # Receive comparable IPs and exit if identical
    current_ip = _get_current_ip()
    last_ip = _get_last_ip(ip_file)
    if current_ip == last_ip:
        exit(0)

    log.info(f'Updating old IP "{last_ip}" to new IP "{current_ip}"')

    # Obtain current zone state
    xml, xml_string = _create_autodns_request(
        username, password, domain, '0205')
    response = post(GATEWAY, xml_string)
    if response.status_code != 200:
        log.error(f'Error on requesting current state:\n{response.text}')
        exit(1)
    xml = parseString(response.text)
    res_type = xml.getElementsByTagName('type')[0]
    res_text = res_type.firstChild.nodeValue
    if res_text == 'error':
        pretty_xml = xml.toprettyxml()
        log.error(f'Error on requesting current state:\n{pretty_xml}')
        exit(1)

    # Store intermediate result
    file_handle = open('status.xml', 'w')
    xml.writexml(file_handle, indent='  ', addindent='  ', newl='\n',
                 encoding='UTF-8')
    file_handle.close()

    _apply_current_ip_address(xml, current_ip)

    # Setup update request
    xml, xml_string = _create_autodns_request(
        username, password, domain, '0202',
        xml.getElementsByTagName('zone')[0])
    xml_string_updated = xml.toprettyxml()

    response = post(GATEWAY, xml_string_updated)
    if response.status_code != 200:
        log.error('Error on sending update request:\n{}'
                  .format(response.text))
        exit(1)
    xml = parseString(response.text)
    log.info(f'Update successful:\n{xml.toprettyxml()}')

    # Save new IP address to IP file
    with open(ip_file, 'w') as out:
        out.write(current_ip + '\n')


def _append_xml_element(
        xml: object, node: object, attr: object, val: object) -> None:
    el = xml.createElement(attr)
    el_ = xml.createTextNode(val)
    el.appendChild(el_)
    node.appendChild(el)


def _create_autodns_request(
        user: str, password: str, domain: str, code: str, zone: str = None
) -> Union[str, str]:
    impl = getDOMImplementation()
    xml = impl.createDocument(None, 'request', None)
    auth = xml.createElement('auth')
    _append_xml_element(xml, auth, 'user', user)
    _append_xml_element(xml, auth, 'password', password)
    _append_xml_element(xml, auth, 'context', '4')
    xml.documentElement.appendChild(auth)
    task = xml.createElement('task')
    _append_xml_element(xml, task, 'code', code)
    if not zone:
        zone = xml.createElement('zone')
    task.appendChild(zone)
    _append_xml_element(xml, zone, 'name', domain)
    xml.documentElement.appendChild(task)
    xml_string = xml.toprettyxml()
    return xml, xml_string


def _valid_ip(ip: str) -> bool:
    return match(
        r'^(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){3}'  # noqa: FS003 E501
        + r'([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])$', ip  # noqa: FS003 E501
    )


def _get_current_ip() -> str:
    response = get(GATEWAY_IP)
    if not _valid_ip(response.text):
        log.error(f'Could not obtain current IP from {GATEWAY_IP}')
        exit(1)
    return response.text


def _get_last_ip(infile: str) -> str:
    try:
        fh = open(infile)
    except FileNotFoundError:
        log.warn('Provided IP file does not exist!')
        return ''
    lines = fh.readlines()
    if not lines:
        log.warn('Provided IP file is empty!')
        return ''
    last_ip = ' '.join(lines).strip()
    if not _valid_ip(last_ip):
        log.warn('IP from IP file is not valid!')
        return ''
    return last_ip


def _rewrite_ip(el: object, xml: object, current_ip: str) -> None:
    for c in el.childNodes:
        el.removeChild(c)
    el_ = xml.createTextNode(current_ip)
    el.appendChild(el_)


def _apply_current_ip_address(xml: object, current_ip: str) -> None:
    # apply sublevel ip addresses
    zone = xml.getElementsByTagName('zone')[0]
    for el in zone.getElementsByTagName('main'):
        [_rewrite_ip(el2, xml, current_ip)
         for el2 in el.getElementsByTagName('value')]
    for el in zone.getElementsByTagName('rr'):
        type_el = el.getElementsByTagName('type')[0]
        type_str = type_el.firstChild.nodeValue
        if type_str != 'A':  # We only want to update A-records
            continue
        [_rewrite_ip(el2, xml, current_ip)
         for el2 in el.getElementsByTagName('value')]