Last active
December 9, 2021 23:58
-
-
Save terabyte128/240491b472109c6cebd0978e23fbc032 to your computer and use it in GitHub Desktop.
Update EdgeRouter local DNS entries from Traefik
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment