From 841277660013bb2fed24db5ea28793365946c238 Mon Sep 17 00:00:00 2001 From: Sophie Schiller Date: Wed, 16 Apr 2025 21:53:09 +0200 Subject: [PATCH] initial commit --- hetzner-api-dyndns.py | 109 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 109 insertions(+) create mode 100644 hetzner-api-dyndns.py diff --git a/hetzner-api-dyndns.py b/hetzner-api-dyndns.py new file mode 100644 index 0000000..5924cf5 --- /dev/null +++ b/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()