DNS dynamique avec l'api de Scaleway et de python

Posted on mer. 27 octobre 2021 in network

Cette année j'ai été obligé de changer de FAI, je suis passé chez Orange et j'ai (re)découvert les joies d'une ip dynamique...

Étant auto-hebergé, il m'a fallu trouver une solution.

Solutions

J'ai bien commencé à chercher des solutions de dynDNS mais je me suis vite rendu compte que ça ne me convenait pas (panne à répétition, temps de MAJ, etc ...).

Ensuite, je me suis dit que j'allais me monter un bind auto-hebergé, mais par manque de temps, j'ai mis cette solution de côté.

Et par le plus grand des hasards en mettant à jour une entrée DNS, je me suis rendu compte que Scaleway avait une API pour gérer ses services.

Dyndns.py

J'ai donc écrit ce petit script (qui est utilisable par tous) dont je vais détailler les différentes parties.

Il n'y a besoin que du token de l'api (disponible ici)

Fonction args

Elle parse les argument du scripts, et permets d'avoir un --help.

def get_args():
    """
    parse agrs
    """
    parser = argparse.ArgumentParser(description="Update dns zone with online.net API")
    group = parser.add_mutually_exclusive_group()
    group.add_argument("-v", "--verbose", action="store_true")
    group.add_argument("-q", "--quiet", action="store_true")
    parser.add_argument("-c", "--clean", help="clean old unused zones", action="store_true")
    parser.add_argument("-t", "--token", help="token's API (https://console.online.net/fr/api/access)")
    parser.add_argument("-u", "--url", help="url to get public ip, default=http://ifconfig.me", default="http://ifconfig.me")
    parser.add_argument("-r", "--records", help="records to update, comma separated list")
    parser.add_argument("domain", help="domain to update")
    return parser.parse_args()

Fonction clear

Efface les versions innutilisé:

def clear_versions(args, session):
    """
    clear inactive versions
    """
    all_versions = session.get("https://api.online.net/api/v1/domain/" + args.domain + "/version").json()
    versions_to_remove = [x for x in session.get("https://api.online.net/api/v1/domain/" + args.domain + "/version").json() if not x['active']]
    for version in versions_to_remove:
        if args.verbose:
            print("Deleting : " + version['name'])
        api_session.delete("https://api.online.net/api/v1/domain/" + args.domain + "/version/" + version["uuid_ref"])

Fonction update

Crée une nouvelle zone, la peuple (avec les entrées de la version active) et l'active.

def create_new_version(args, records, records_to_update):
    """
    Create new zone version with name YYYYMMDDhhmm
    """
    new_name = datetime.now().strftime("%Y%m%d%H%M")
    if args.verbose:
        print("Generating new zone called : " + new_name)
    url = "https://api.online.net/api/v1/domain/" + args.domain + "/version"
    data = {
            'name' : new_name
            }
    res = api_session.post(url, data)
    if "error" in res.json():
        exit(res.json()['error_description'])
    id_to_update = res.json()["uuid_ref"]
    url = "https://api.online.net/api/v1/domain/" + args.domain + "/version/" + id_to_update + "/zone"
    for record in records:
        if record['name'] in records_to_update:
            record['data'] = public_ip
        data = {
                'name' : record['name'],
                'type' : record['type'],
                'ttl' : record['ttl'],
                'data' : record['data'],
                'priority' : 0,
                }
        res = api_session.post(url, data)
        if "error" in res.json():
            exit(res['error_description'])
    url = "https://api.online.net/api/v1/domain/" + args.domain + "/version/" + id_to_update + "/enable"
    res = api_session.patch(url)
    try:
        res_json = res.json()
    except:
        res_json = {}
    if "error" in res_json:
        exit(res['error_description'])

Main

Appelé lors de l'exécution du scipt

if __name__ == "__main__":

    ## PARSE AGRS ##
    args = get_args()

    ## Check if Token is present
    if not args.token:
        exit("token is required")

    ## Get public IP ##
    try:
        public_ip = requests.get(args.url).text
    except:
        exit("Can't get public IP")
    if not public_ip:
        exit("Can't get public IP")
    if args.verbose:
        print("Public IP is : " + public_ip)

    ## Init api session ##
    api_session = requests.session()
    api_session.headers['Authorization'] = 'Bearer ' + args.token

    ## Check api connection ##
    try:
        response = api_session.get("https://api.online.net/api/v1/domain/list")
    except:
        exit("Can't connect to api")
    if "error" in response.json():
        exit(response.json()['error_description'])


    ## Get records and compare IP ##
    records = api_session.get("https://api.online.net/api/v1/domain/" + args.domain + "/zone").json()
    if "error" in records:
        exit(records['error_description'])

    records_to_update = args.records.split(',')

    old_ip = [x['data'] for x in records if x['name'] in records_to_update and not public_ip in x["data"]]

    if not old_ip:
        exit()

    else:
        create_new_version(args, records, records_to_update)

    ## Clear inactive versions ##
    if args.clean:
        clear_versions(args, api_session)

BONUS : acme.sh + api online

En me documentant sur l'api d'online, je me suis rendu compte qu'acme.sh permet d'utiliser l'api de Scaleway.

Jusqu'à présent je mettais à jour mes certificats à la main tous les trois mois (avec quelques raté à l'occasion...).

Il y a juste besoin du token dans une variable d'environement.

$ export ONLINE_API_KEY='xxx'
$ acme.sh --issue --dns dns_online -d 0w.tf -d "*.0w.tf" --reloadcmd  "nginx -s reload" --server letsencrypt --force

DNS API python