Update EdgeRouter local DNS entries from Traefik

update_dns.py

import os
import re
from typing import Set

import requests
from requests.sessions import RequestsCookieJar
import urllib3

"""
Poll Traefik for changes to Host configurations, and update EdgeRouter config
to point to the specified server.

This can be run in Docker using cron.
"""

# server to point EdgeRouter at
server_name = os.environ.get("SERVER_NAME", "")

# EdgeRouter hostname or IP
router_uri = os.environ.get("ROUTER_URI", "")
router_user = os.environ.get("ROUTER_USER", "")
router_pass = os.environ.get("ROUTER_PASS", "")

# Traefik API hostname or IP (can be a Docker container as long as they share a network)
traefik_uri = os.environ.get("TRAEFIK_URI", "")

matcher = re.compile(r"Host\(`(\w+)`\)")
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def get_new_hosts() -> Set[str]:
    rsp = requests.get(f"{traefik_uri}/api/http/routers")
    rsp.raise_for_status()
    routers = rsp.json()
    hosts = set()

    for router in routers:
        rule = router["rule"]
        match = matcher.match(rule)

        if match:
            hosts.add(match.group(1))

    return hosts


def get_auth_cookies() -> RequestsCookieJar:
    rsp = requests.post(
        router_uri,
        data={"username": router_user, "password": router_pass},
        allow_redirects=False,
        verify=False,
    )
    rsp.raise_for_status()

    return rsp.cookies


def get_existing_hosts(auth_cookies: RequestsCookieJar) -> Set[str]:
    rsp = requests.get(
        f"{router_uri}/api/edge/getcfg.json?node[]=system&node[]=static-host-mapping&node[]=host-name&node[]={server_name}",
        verify=False,
        cookies=auth_cookies,
    )
    rsp.raise_for_status()

    data = rsp.json()
    children = data["GETCFG"]["children"]

    alises = next(i for i in children if "alias" in i)

    return set(alises["alias"])


def set_new_hosts(
    auth_cookies: RequestsCookieJar,
    existing_hosts: Set[str],
    new_hosts: Set[str],
):
    req = {
        "DELETE": {
            "system": {
                "static-host-mapping": {
                    "host-name": {server_name: {"alias": list(existing_hosts)}}
                }
            }
        },
        "SET": {
            "system": {
                "static-host-mapping": {
                    "host-name": {server_name: {"alias": list(new_hosts)}}
                }
            }
        },
    }

    rsp = requests.post(
        f"{router_uri}/api/edge/batch.json",
        json=req,
        verify=False,
        cookies=auth_cookies,
        headers={
            "X-CSRF-TOKEN": auth_cookies.get("X-CSRF-TOKEN"),
        },
    )
    rsp.raise_for_status()


def main():
    new_hosts = get_new_hosts()
    auth_cookies = get_auth_cookies()
    existing_hosts = get_existing_hosts(auth_cookies)

    added = new_hosts - existing_hosts
    removed = existing_hosts - new_hosts

    if new_hosts == existing_hosts:
        print("no changes; exiting")
        return

    if added:
        print(f"adding hosts {added}")

    if removed:
        print(f"removing hosts {removed}")

    set_new_hosts(auth_cookies, existing_hosts, new_hosts)


if __name__ == "__main__":
    main()