initial commit

hetzner-api-dyndns.py

@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+
+import json
+import logging
+from argparse import ArgumentParser
+from copy import deepcopy
+
+import requests
+from rich.logging import RichHandler
+
+FORMAT = "%(message)s"
+logging.basicConfig(
+    level="INFO", format=FORMAT, datefmt="[%X]", handlers=[RichHandler()]
+)
+log = logging.getLogger("rich")
+
+
+def create_session(api_key):
+    s = requests.Session()
+    s.headers = {
+        "Content-Type": "application/json",
+        "Auth-API-Token": api_key,
+    }
+    return s
+
+
+def get_current_public_ip(version=4):
+    log.debug(f"Asking clara for our current ipv{version}")
+    response = requests.get(f"https://ip{version}.clerie.de/")
+    log.debug(f"Got {response.text} as answer")
+    response.raise_for_status()
+    return response.text
+
+
+def get_zone_matching_name(session, name):
+    response = session.get(url="https://dns.hetzner.com/api/v1/zones")
+    response.raise_for_status()
+    log.debug(response.json())
+    for z in response.json().get("zones", {}):
+        if z.get("name", None) == name:
+            id = z["id"]
+            log.debug(f"ID of the requested zone is {id}.")
+            break
+    return id
+
+
+def get_records_matching_name(session, zone, name):
+    response = session.get(
+        url="https://dns.hetzner.com/api/v1/records",
+        params={
+            "zone_id": zone,
+        },
+    )
+    response.raise_for_status()
+    record_objects = []
+    for r in response.json()["records"]:
+        if r.get("name", None) == name:
+            record_objects.append(r)
+    log.debug(record_objects)
+    return record_objects
+
+
+def update_records(session, records, ipv4, ipv6):
+    for record in records:
+        r = deepcopy(record)
+        log.debug("before modification:")
+        log.debug(record)
+        if record["type"] == "A":
+            r["value"] = ipv4
+            log.info("updating ipv4")
+        elif record["type"] == "AAAA":
+            r["value"] = ipv6
+            log.info("updating ipv6")
+        r.pop("id")
+        log.debug("after modification:")
+        log.debug(r)
+        response = session.put(
+            url=f"https://dns.hetzner.com/api/v1/records/{record['id']}",
+            data=json.dumps(r),
+        )
+        response.raise_for_status()
+        log.info(f"got {response.status_code}")
+
+
+def main():
+    parser = ArgumentParser()
+    parser.add_argument("--api_key", "-k")
+    parser.add_argument("--record", "-r")
+    parser.add_argument("--zone", "-z")
+    parser.add_argument("--verbose", help="increase output verbosity",
+                    action="store_true")
+    args = parser.parse_args()
+    if args.verbose:
+        log.setLevel(logging.DEBUG)
+    session = create_session(args.api_key)
+    log.info("Getting IPs")
+    ipv4 = get_current_public_ip(4)
+    log.info(f"Found ipv4: {ipv4}")
+    ipv6 = get_current_public_ip(6)
+    log.info(f"Found ipv6: {ipv6}")
+    zone_id = get_zone_matching_name(session, args.zone)
+    records = get_records_matching_name(session, zone_id, args.record)
+    log.info("gathered all data, let's try to use it")
+    update_records(session, records, ipv4, ipv6)
+    log.info("Dyndns entries updated if everything went smoothly.")
+
+
+if __name__ == "__main__":
+    main()