108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import json
|
|
import logging
|
|
from argparse import ArgumentParser
|
|
from copy import deepcopy
|
|
|
|
import requests
|
|
|
|
FORMAT = "%(message)s"
|
|
logging.basicConfig(
|
|
level="INFO", format=FORMAT, datefmt="[%X]"
|
|
)
|
|
log = logging.getLogger()
|
|
|
|
|
|
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()
|