dyn-gandi/dyn_gandi/main.py

214 lines
6.5 KiB
Python
Raw Normal View History

2023-08-03 21:14:04 +00:00
import os
import re
import json
import argparse
2023-07-25 22:49:47 +00:00
from datetime import datetime
2023-08-13 11:58:30 +00:00
from typing import Callable, Any
2023-07-25 22:49:47 +00:00
import requests
2023-08-03 21:14:04 +00:00
2023-08-15 19:45:47 +00:00
def loadSettings(path: str | None) -> dict:
'''
Load the settings file.
`path` Path to a settings file.
2023-10-13 11:53:54 +00:00
By default the program will look for `$HOME/.config/dyn-gandi/config.json` and then `/etc/dyn-gandi.json`.
'''
2023-08-15 19:45:47 +00:00
if path is None:
path = f"{os.path.expanduser('~')}/.config/dyn-gandi/config.json"
2023-08-13 11:58:30 +00:00
2023-08-15 19:45:47 +00:00
if not os.path.exists(path):
path = '/etc/dyn-gandi.json'
2023-08-13 11:58:30 +00:00
2023-10-13 11:53:54 +00:00
if not os.path.exists(path):
print(
f"[ERROR] No config file found at '~/.config/dyn-gandi/config.json' or '/etc/dyn-gandi.json'")
quit()
2023-08-13 11:58:30 +00:00
if not os.path.exists(path):
print(f"[ERROR] '{path}' does not exist.")
2023-08-03 21:14:04 +00:00
quit()
2023-08-13 11:58:30 +00:00
# TODO check integrity of the config file
2023-08-13 11:58:30 +00:00
with open(path, 'r') as file:
# Load as dictionary
2023-08-13 11:58:30 +00:00
config = json.load(file)
# API Key entries
2023-08-13 11:58:30 +00:00
keys = config['api'].keys()
# This will build up a copy of the API section with the API keys instead of filenames.
2023-08-13 11:58:30 +00:00
mask = dict()
# If the API key is a file the use the contents as a new key.
2023-08-13 11:58:30 +00:00
for key in keys:
api_key = key
if os.path.exists(key):
with open(key, 'r') as file:
api_key = file.readline().strip()
mask[key] = api_key
config['api'] = {mask[key]: value for key,
value in config['api'].items()}
2023-08-13 11:58:30 +00:00
return config
2023-07-25 22:49:47 +00:00
def log(logline: str, path: str = './log.txt', stdout: bool = False) -> None:
'''
Appends `logline` at the end of the file file with a timestamp.
`logline` The line to log.
'path' Path to file where to log.
`stdout` Wether to print to stdout instead of a file.
'''
line = f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}]{logline}"
if stdout:
print(line)
else:
with open(path, 'a') as file:
file.write(line + '\n')
2023-08-03 21:14:04 +00:00
2023-07-25 22:49:47 +00:00
2023-08-13 11:58:30 +00:00
def getCurrentIP(resolvers: list[str], log: Callable[[str], Any]) -> str | None:
'''
Go through resolvers until one returns an IP.
`resolvers` A list of IP resolvers.
`log` A function which takes a string and logs it.
'''
2023-07-25 22:49:47 +00:00
for resolver in resolvers:
2023-07-26 20:31:05 +00:00
response = requests.get(resolver)
if response.ok:
2023-07-25 22:49:47 +00:00
current_ip = response.text.strip()
# It suffices to check whether the search is not None since the regex matches from beginning to end.
2023-08-03 21:14:04 +00:00
is_ipv4 = re.search(
'^((25[0-5]|(2[0-4]|1\d|[1-9]|)\d)\.?\\b){4}$', current_ip) is not None
2023-07-25 22:49:47 +00:00
if is_ipv4:
log(f"[OK][{resolver}] Current IP: '{current_ip}'")
2023-08-03 21:14:04 +00:00
return current_ip # Break if we have found our IP
2023-07-25 22:49:47 +00:00
else:
log(f"[ERROR][{resolver}] '{current_ip}' is not IPv4")
else:
log(f"[ERROR][{resolver}][{response.status_code}] {response.text}")
2023-07-25 22:49:47 +00:00
return None
2023-08-03 21:14:04 +00:00
2023-08-13 11:58:30 +00:00
def checkRecords(api: dict[str, dict[str, list[str]]], current_ip: str, log: Callable[[str], Any]) -> dict[str, dict[str, list[str]]]:
'''
Return records that need to be renewed.
`api` The API section of the settings file.
`current_ip` The current IP address.
`log` A function that takes in a string and logs it.
'''
# TODO: Detect TTL change
2023-07-25 22:49:47 +00:00
records_to_renew = api
for key, domains in api.items():
2023-08-03 21:14:04 +00:00
headers = {'authorization': f"Apikey {key}"}
2023-08-13 11:58:30 +00:00
for domain, _ in domains.items():
2023-08-03 21:14:04 +00:00
response = requests.get(
f"https://api.gandi.net/v5/livedns/domains/{domain}/records?rrset_type=A", headers=headers)
2023-07-25 22:49:47 +00:00
if response.status_code == 200:
records_to_renew[key][domain] = list(set(records_to_renew[key][domain]) - set(record['rrset_name'] for record in (
filter(lambda x: current_ip in x['rrset_values'], response.json()))))
if (records := records_to_renew[key][domain]) != []:
log(f"[INFO][{','.join(records)}] Records need to be renewed")
2023-07-25 22:49:47 +00:00
else:
2023-07-26 20:31:05 +00:00
log(f"[ERROR][{domain}][{response.status_code}] {response.json()}")
2023-07-25 22:49:47 +00:00
return records_to_renew
2023-08-03 21:14:04 +00:00
def renewRecords(api: dict[str, dict[str, list[str]]], current_ip: str, log: Callable[[str], Any], ttl: int = 300) -> None:
'''
Update records and create them, if they don't exist.
`api` The API section of the settings file.
`current_ip` The current IP address.
`log` A function that takes a string and logs it.
`ttl` How long a record should be cached.
'''
2023-07-25 22:49:47 +00:00
payload = json.dumps({
'rrset_values': [current_ip],
'rrset_ttl': ttl
2023-07-25 22:49:47 +00:00
})
for api_key, domains in api.items():
2023-07-25 22:49:47 +00:00
headers = {
'authorization': f"Apikey {api_key}",
2023-07-25 22:49:47 +00:00
'content-type': 'application/json',
}
for domain, records in domains.items():
for record in records:
2023-08-03 21:14:04 +00:00
response = requests.put(
f"https://api.gandi.net/v5/livedns/domains/{domain}/records/{record}/A", data=payload, headers=headers)
2023-07-25 22:49:47 +00:00
if response.status_code == 201:
log(f"[OK][{domain}][{record}] Renewed with IP '{current_ip}'")
else:
2023-07-26 20:31:05 +00:00
log(f"[ERROR][{domain}][{record}][{response.status_code}] {response.json()}")
2023-07-25 22:49:47 +00:00
2023-08-03 21:14:04 +00:00
def parseArgs(args: list[str]) -> argparse.Namespace:
'''
Parse command-line arguments.
`args` A list of user-supplied command line arguments.
'''
parser = argparse.ArgumentParser(
description='A program that renews A records via the Gandi API v5.')
parser.add_argument('--config', type=str, default=None,
help='Path to a config file.')
parser.add_argument('--dry-run', action='store_true',
help='Just perform a dry-run.')
return parser.parse_args(args)
2023-08-15 19:45:47 +00:00
2023-07-25 22:49:47 +00:00
def main():
# Parse user-supplied command-line arguments.
args = parseArgs(os.sys.argv[1:])
2023-08-15 19:45:47 +00:00
# Read in the settings file.
config = loadSettings(args.config)
2023-08-15 19:45:47 +00:00
def logger(x: str) -> Any: return log(x, config['log_path'], args.dry_run)
2023-07-25 22:49:47 +00:00
current_ip = getCurrentIP(config['resolvers'], logger)
2023-08-13 11:58:30 +00:00
if current_ip is not None:
records_to_renew = checkRecords(config['api'], current_ip, logger)
# Do not perform an actual record change during a dry-run.
if not args.dry_run:
renewRecords(records_to_renew, current_ip, logger, config['ttl'])
2023-08-03 21:14:04 +00:00
2023-07-25 22:49:47 +00:00
if __name__ == '__main__':
2023-08-03 21:14:04 +00:00
main()