Lima City API DynDNS

lima-dyndns.py

#!/usr/bin/env python3
import requests
import json
import sys
import datetime

# Updates Lima City DNS with v4 and v6
# Expects the file with all the secrets in /etc/.dyndns_secrets.json
#
# {
#     "api": "apikey",
#     "domain_id": "number",
#     "v4_subdomain_id": "number",
#     "v6_subdomain_id": "number",
#     "subdomain_name": "home.example.com"
# }


def set_v4_dns_entry(api_key, domain_id, subdomain_id, subdomain_name, ip_addr):
    headers = {'content-type': 'application/json'}
    url = f"https://www.lima-city.de/usercp/domains/{domain_id}/records/{subdomain_id}"
    session = requests.Session()
    session.auth = ("api", api_key)
    dns = {
        "nameserver_record": {
            "name": subdomain_name,
            "type": "A",
            "content": ip_addr,
            "ttl": 900,
            "priority": 0
        }
    }
    r = session.put(url=url, data=json.dumps(dns), headers=headers)
    if r.status_code == 200:
        return True
    else:
        log(ip_addr, f"failed - lima returned {r.status_code}")
        return False


def set_v6_dns_entry(api_key, domain_id, subdomain_id, subdomain_name, ip_addr):
    headers = {'content-type': 'application/json'}
    url = f"https://www.lima-city.de/usercp/domains/{domain_id}/records/{subdomain_id}"
    session = requests.Session()
    session.auth = ("api", api_key)
    dns = {
        "nameserver_record": {
            "name": subdomain_name,
            "type": "AAAA",
            "content": ip_addr,
            "ttl": 900,
            "priority": 0
        }
    }
    r = session.put(url=url, data=json.dumps(dns), headers=headers)
    if r.status_code == 200:
        return True
    else:
        log(ip_addr, f"failed - lima returned {r.status_code}")
        return False


def get_v4_ip():
    url = "https://ip4.seeip.org/json"
    r = requests.get(url=url)
    if r.status_code == 200:
        ip = r.json()["ip"]
        return ip
    else:
        log('???.???.???.???', f"failed - seeip.org returned {r.status_code}")
        return None


def get_v6_ip():
    url = "https://ip6.seeip.org/json"
    r = requests.get(url=url)
    if r.status_code == 200:
        ip = r.json()["ip"]
        return ip
    else:
        log('???.???.???.???', f"failed - seeip.org returned {r.status_code}")
        return None


def read_dns(api_key, domain_id, subdomain_id, ip_addr):
    url = f"https://www.lima-city.de/usercp/domains/{domain_id}/records.json"
    session = requests.Session()
    session.auth = ("api", api_key)
    r = session.get(url=url)
    items = r.json()["records"]
    for each in items:
        if each["id"] == int(subdomain_id):
            if each["content"] == ip_addr:
                log(ip_addr, "success")
                return True
            else:
                log(ip_addr, f"failed - IP still {each['content']}")
                return False


def log(ip_addr, status):
    print(f"{datetime.datetime.now()}  {ip_addr} - {status}\n")
    with open("/var/log/dyndns.log", "a") as fp:
        fp.write(f"{datetime.datetime.now()}  {ip_addr} - {status}\n")


def main():
    try:
        with open("/etc/.dyndns_secrets.json", "r") as fp:
            secrets = json.load(fp=fp)
    except FileNotFoundError:
        log("???.???.???.???", "Cannot open file secrets.json")
        sys.exit(1)

    v4ip = get_v4_ip()
    if v4ip is not None:
        if set_v4_dns_entry(secrets["api"],
                            secrets["domain_id"],
                            secrets["v4_subdomain_id"],
                            secrets["subdomain_name"],
                            v4ip):
            if not read_dns(secrets["api"],
                            secrets["domain_id"],
                            secrets["v4_subdomain_id"],
                            v4ip):
                sys.exit(1)

    v6ip = get_v6_ip()
    if v6ip is not None:
        if set_v6_dns_entry(secrets["api"],
                            secrets["domain_id"],
                            secrets["v6_subdomain_id"],
                            secrets["subdomain_name"],
                            v6ip):
            if not read_dns(secrets["api"],
                            secrets["domain_id"],
                            secrets["v6_subdomain_id"],
                            v6ip):
                sys.exit(1)
    sys.exit(0)


if __name__ == '__main__':
    main()