ovhddns/main.py

#!/usr/bin/env python3

import itertools
import os
import sys
import logging

import yaml
import click
import ovh
import dns.resolver
import dns.rdatatype

from schema import And, Optional, Use, Schema

logger = logging.getLogger('ovhdns')

config_schema = Schema({
    'ovh': {
        'application_key': Use(str),
        'application_secret': Use(str),
        'consumer_key': Use(str),
        'endpoint': Use(str),
    },
    'zone': Use(str),
    'subdomain': Use(str),
    Optional('field_type', default='A'): And(Use(str), Use(str.upper), dns.rdatatype.RdataType.make),
    Optional('logging', default={'level': ':WARNING,ovhdns:INFO'}): {
        Optional('level', default=':WARNING,ovhdns:INFO'): Use(str),
    },
    Optional('ttl', default=None): Use(lambda i: int(i) if i is not None else None),
})

def get_opendns_ips():
    ips = []
    for addr, typ in itertools.product(
        ['resolver1.opendns.com', 'resolver2.opendns.com'],
        ['A', 'AAAA'],
    ):
        ips += [r.to_text() for r in dns.resolver.resolve(addr, typ)]
    return ips

def get_my_ips():
    resolver = dns.resolver.Resolver(configure=False)
    resolver.nameservers = get_opendns_ips()
    return [r.to_text() for r in resolver.resolve('myip.opendns.com')]


def get_records(client, config):
    zone = config['zone']
    records = client.get(f'/domain/zone/{zone}/record')
    records = [
        client.get(f'/domain/zone/{zone}/record/{id}')
        for id in records
    ]
    return records

def init_logging():
    logging.getLogger().addHandler(
        logging.StreamHandler(),
    )

def set_logging_level(level: str):
    levels = {
        'NOTSET': logging.NOTSET,
        'DEBUG': logging.DEBUG,
        'INFO': logging.INFO,
        'WARNING': logging.WARNING,
        'WARN': logging.WARNING,
        'ERROR': logging.ERROR,
        'CRITICAL': logging.CRITICAL,
    }

    log_levels = [x.split(':') for x in level.split(',')]
    for module, level in log_levels:
        module = module.strip() if module else None
        level = level.strip().upper()
        if level not in levels:
            raise ValueError(f'invalid log level, allowed values: {repr(set(levels.keys()))}')
        logging.getLogger(module).setLevel(levels[level])

def load_config(file):
    if file == '-':
        config = yaml.safe_load(sys.stdin)
    else:
        if os.stat(file).st_mode & 0o777 & ~0o600:
            raise Exception('refusing to load insecure configuration file, file must have permission 0o600')
        with open(file) as fp:
            config = yaml.safe_load(fp)
    return config_schema.validate(config)

@click.command()
@click.option('--config', '-c', envvar='OVHDNS_CONFIG', default='-', help='The configuration file')
@click.option('--level', '-l', envvar='OVHDNS_LEVEL', default=None, help='The log level for the application')
@click.option('--force-refresh', '-f', envvar='OVHDNS_FORCE_REFRESH', is_flag=True, help='Force zone refresh even if deemed not needed')
def main(config, level, force_refresh):
    init_logging()
    config = load_config(config)
    if level:
        config['logging']['level'] = level
    config = config_schema.validate(config)
    set_logging_level(config['logging']['level'])

    ovh_client = ovh.Client(
        **config['ovh'],
    )
    records = [
        r
        for r in get_records(ovh_client, {'zone': 'redxef.at'})
        if r['fieldType'] == config['field_type']
        and r['zone'] == config['zone']
        and r['subDomain'] == config['subdomain']
    ]
    logger.info('found matching records: %s', records)
    if len(records) > 1:
        logger.error('found more than one record, don\'t know which to pick')
        sys.exit(1)

    my_ips = get_my_ips()
    need_refresh = False
    if len(records) == 1:
        logger.info('record already present, updating if needed')
        if my_ips[0] == records[0]['target']:
            logger.info('ip is up-to-date, skipping update')
        else:
            need_refresh = True
            ovh_client.put(
                f'/domain/zone/{config["zone"]}/record/{records[0]["id"]}',
                target=my_ips[0],
                ttl=config['ttl'],
            )
            logger.info('updated record, new ip is %s', my_ips[0])

    else:
        logger.info('record not already present, creating new one')
        need_refresh = True
        ovh_client.post(
            f'/domain/zone/{config["zone"]}/record/',
            subDomain=config['subdomain'],
            target=my_ips[0],
            fieldType=config['field_type'],
            ttl=config['ttl'],
        )
        logger.info('created new record, ip is %s', my_ips[0])

    if need_refresh or force_refresh:
        ovh_client.post(f'/domain/zone/{config["zone"]}/refresh')
        logger.info('refreshed zone')
    else:
        logger.info('zone refresh skipped')
    sys.exit(0)

if __name__ == '__main__':
    main()