loopia-api-dyndns

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Dynamically updates the IPv4/IPv6 DNS records for domains registered with Loopia.
https://joelpurra.com/projects/loopia-api-dyndns/

Based on Loopia API's official sample script.
https://support.loopia.se/wiki/uppdatera-dynamisk-ip-adress-med-loopiaapi/
"""

import ipaddress
import os
import sys
import urllib.request
import xmlrpc.client


def print_error(msg):
    print(msg, file=sys.stderr)


class Config:
    try:
        username = os.environ["LOOPIA_API_USERNAME"]
        password = os.environ["LOOPIA_API_PASSWORD"]
        domain = os.environ["LOOPIA_DOMAIN"]
        subdomain = os.environ["LOOPIA_SUBDOMAIN"]
    except KeyError as exception:
        print_error(
            "Could not read configuration from environment variable: {exception}".format(
                exception=exception,
            )
        )
        quit(1)

    fqdn = domain if subdomain == "@" else subdomain + "." + domain

    check_ipv4_url = "https://api.ipify.org/"
    check_ipv6_url = "https://api6.ipify.org/"
    loopia_api_xml_rpc_url = "https://api.loopia.se/RPCSERV"


def api_error(exception):
    print_error(
        "Request to Loopia API failed: {exception}".format(
            exception=exception,
        )
    )
    quit(2)


def get_string_value_from_url(url, max=None):
    try:
        return urllib.request.urlopen(url).read(max).decode("utf-8").strip()
    except:
        print_error(
            "Could not get string from url: {url}".format(
                url=url,
            )
        )
        quit(5)


def get_public_ipv4():
    ipv4str_from_url = get_string_value_from_url(Config.check_ipv4_url, 15)
    try:
        return str(ipaddress.IPv4Address(ipv4str_from_url))
    except:
        print_error(
            "Could not parse IPv4 address: {ipv4str}".format(
                ipv4str_from_url=ipv4str_from_url,
            )
        )
        quit(7)


def get_public_ipv6():
    ipv6str_from_url = get_string_value_from_url(Config.check_ipv6_url, 45)
    try:
        return str(ipaddress.IPv6Address(ipv6str_from_url))
    except:
        print_error(
            "Could not parse IPv6 address: {ipv6str}".format(
                ipv6str_from_url=ipv6str_from_url,
            )
        )
        quit(8)


def get_records(recordType):
    try:
        zone_records = client.getZoneRecords(
            Config.username,
            Config.password,
            Config.domain,
            Config.subdomain,
        )
    except Exception as exception:
        api_error(exception)

    # Remove irrelevant records and return.
    return [d for d in zone_records if d["type"] == recordType]


def update_record(value, record):
    # Does the record need updating?
    if record["rdata"] != value:
        # Setting record_id updates an existing record.
        new_record = {
            "priority": record["priority"],
            "rdata": value,
            "record_id": record["record_id"],
            "ttl": record["ttl"],
            "type": record["type"]
        }

        try:
            status = client.updateZoneRecord(
                Config.username,
                Config.password,
                Config.domain,
                Config.subdomain,
                new_record,
            )
        except Exception as exception:
            api_error(exception)
    else:
        status = "No change"

    print(
        "{fqdn} {recordType}: {status} ({value})".format(
            fqdn=Config.fqdn,
            recordType=record["type"],
            status=status,
            value=value,
        )
    )


def check_and_update_record(recordType, value):
    """Fetch a single record and update it"""
    records = get_records(recordType)
    recordCount = len(records)

    if recordCount != 1:
        print_error(
            "Require exactly one {recordType} record for {fqdn}, but found {recordCount}.".format(
                fqdn=Config.fqdn,
                recordCount=recordCount,
                recordType=recordType,
            )
        )
        quit(4)

    update_record(value, records[0])


if __name__ == "__main__":
    # Build XML RPC client.
    client = xmlrpc.client.ServerProxy(
        uri=Config.loopia_api_xml_rpc_url,
        encoding="utf-8",
    )

    # Dynamic DNS updates.
    public_ipv4 = get_public_ipv4()
    check_and_update_record("A", public_ipv4)

    public_ipv6 = get_public_ipv6()
    check_and_update_record("AAAA", public_ipv6)