OVH Dynhost updater Python daemon

ovh-dynhost.service

[Unit]
Description=OVH Dynhost updater daemon
After=network-online.target
Wants=network-online.target

[Service]
ExecStart=/opt/ovh_dynhost.py --journald /etc/opt/ovh_dynhost.ini
Restart=on-abort

[Install]
WantedBy=multi-user.target

ovh_dynhost.ini

[dynhost-my-host1]
hostname=my.host.tld
username=host.tld-my
password=OhVeHach3

ovh_dynhost.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import argparse
import configparser
import logging
import requests
import socket
from time import sleep

def main():
  parser = argparse.ArgumentParser(formatter_class=argparse.RawTextHelpFormatter,
    description='Daemon updating dynhost',
    epilog='''\
Sample config.ini file:
[dynhost-my-host1]
hostname=my.host.tld
username=host.tld-my
password=OhVeHach3
    ''')
  parser.add_argument('config_file', metavar='config.ini', type=argparse.FileType('r'), help='configuration file to use')
  parser.add_argument('--single-update', '-s', action='store_true', help='run update and exit')
  parser.add_argument('--forced-ip', '-f', metavar='x.x.x.x', help='force IP when doing single update')
  parser.add_argument('--interval', '-i', default=5*60, metavar='sec', type=int, help='set the interval between 2 check & updates (defaults to 5 mins)')
  parser.add_argument('--verbose', '-v', action='count', default=0, help='print more informations, add multiple times to increase verbosity')
  parser.add_argument('--journald', '-j', action='store_true', help='use journald for logging')
  parser.add_argument('--version', '-V', action='version', help='print version number', version='{} version {}'.format(DynHost.program, DynHost.version))
  args = parser.parse_args()

  hosts = {}
  config = configparser.ConfigParser()
  with args.config_file as config_file:
    config.read_file(config_file)
  for section in config.sections():
    hosts[section] = {
      'hostname': config.get(section, 'hostname', raw=True),
      'username': config.get(section, 'username', raw=True),
      'password': config.get(section, 'password', raw=True),
    }

  daemon = DynHost(
    dynhosts=hosts,
    log_level=args.verbose,
    journald_identifier=parser.prog if args.journald else False
  )

  if args.single_update:
    daemon.update_all(my_ip=args.forced_ip)
  else:
    daemon.run(args.interval)


class DynHost:

  program = 'OVH DynHost'
  version = '1.2.0'

  get_ip_timeout = 0.5
  update_ip_timeout = 1.5
  update_endpoint = 'https://www.ovh.com/nic/update'
  
  # https://docs.ovh.com/gb/en/domains/hosting_dynhost/
  # https://docs.ovh.com/fr/domains/utilisation-dynhost/
  # https://gitlab.com/rubendibattista/ovh-dynhost
  # https://github.com/denouche/ovh-dyndns
  # https://github.com/janhenke/ovh-dynhost/blob/master/ovh_dynhost.py
  
  def __init__(self, dynhosts, log_level=0, journald_identifier=False):
    try:
      if not journald_identifier:
        raise Exception('Journald usage disabled, using StreamHandler')
      from systemd.journal import JournalHandler
      handler = JournalHandler(SYSLOG_IDENTIFIER=journald_identifier)
      handler.setFormatter(logging.Formatter(fmt='%(message)s'))
    except:
      handler = logging.StreamHandler()
      handler.setFormatter(logging.Formatter(fmt='[%(levelname)s]\t%(message)s'))
    self.logger = logging.getLogger(self.__class__.__name__)
    self.logger.addHandler(handler)
    logging.addLevelName(logging.WARNING, 'WARN')  # To have pretty print with \t

    # https://stackoverflow.com/questions/14097061/easier-way-to-enable-verbose-logging
    levels = [logging.WARNING, logging.INFO, logging.DEBUG]
    self.logger.setLevel(levels[ min(len(levels)-1, log_level) ])

    self.dynhosts = dynhosts

  # Daemon / continuous updates
  # http://zguide.zeromq.org/py:interrupt
  def run(self, interval=5*60):
    self.logger.warning('{} v{}: starting updates for {} host{} with interval of {} seconds...'.format(
      self.program,
      self.version,
      len(self.dynhosts),
      's' if len(self.dynhosts) > 1 else '',
      interval
    ))
    while True:
      try:
        self.update_all()
        self.logger.info('Sleeping for {} seconds'.format(interval))
        sleep(interval)
      except KeyboardInterrupt:
        self.logger.info('SIGINT received, exiting...')
        break

  def update_all(self, my_ip=None):
    # Get external IP if not given
    my_ip = my_ip if my_ip else self.get_ipv4()
    for identifier, dynhost in self.dynhosts.items():
      self._update_host(identifier, dynhost, my_ip=my_ip)

  # Can be used alone
  def _update_host(self, identifier, dynhost, my_ip):
    try:
      self.logger.info('Starting update of dynhost {} ({})'.format(identifier, dynhost['hostname']))

      update_arguments = self._get_update_arguments(identifier, dynhost, my_ip)
      if update_arguments:
        # Do update
        self._do_update(identifier, dynhost, update_arguments)
    except KeyboardInterrupt:
      raise
    except Exception as e:
      self.logger.error('Error while updating dynhost: {}'.format(e))

  def _get_update_arguments(self, identifier, dynhost, my_ip):
    update_arguments = {'system': 'dyndns', 'hostname': dynhost['hostname']}

    if my_ip:
      # Get current IP to compare
      current_dynhost_ip = self.get_current_dynhost_ip(dynhost['hostname'])
      if current_dynhost_ip:
        # Compare IPs
        if current_dynhost_ip != my_ip:
          self.logger.warning('IP change detected, updating dynhost record from {} to {}'.format(current_dynhost_ip, my_ip))
          update_arguments['myip'] = my_ip
        else:
          self.logger.info('No IP change detected, skipping dynhost record update (current IP: {})'.format(my_ip))
          return False
      else:
        self.logger.warning('Current dynhost record not retrieved, updating with IP {} anyway'.format(my_ip))
    else:
      self.logger.warning('Public IP not retrieved, updating dynhost without IP')

    return update_arguments

  
  def _do_update(self, identifier, dynhost, update_arguments):
    update = requests.get(
      self.update_endpoint,
      params=update_arguments,
      auth=(dynhost['username'], dynhost['password']),
      timeout=self.update_ip_timeout
    )
    self.logger.debug('Request: {}'.format(update.request.url))
    if update.ok and update.text.startswith('good '):
      self.logger.debug('Update successful')
    elif update.ok and update.text.startswith('nochg'):
      self.logger.warning('Request done but update wasn\'t required')
    else:
      extra = ': {}'.format(update.text.strip()) if update.ok else ''
      self.logger.error('Update request failed ({}/{}){}'.format(update.status_code, update.reason, extra))

  def get_ipv4(self):
    try:
      return self.get_ipv4_from_lb()
    except:
      return self.get_ipv4_from_websites()

  def get_current_dynhost_ip(self, hostname):
    current_ip = socket.gethostbyname(hostname)
    if self.is_valid_ipv4_address(current_ip):
      self.logger.debug('Current dynhost record for {}: {}'.format(hostname, current_ip))
      return current_ip
    else:
      self.logger.debug('Current dynhost record: {}'.format(current_ip))
      return False

  # https://www.orange-sans-guigne.com/osg-forum/viewtopic.php?id=565
  # https://github.com/fccagou/pylivebox
  # https://github.com/rene-d/sysbus
  # https://github.com/NextDom/plugin-livebox

  # r = requests.post('http://192.168.1.1/ws', 
  #   headers={
  #     'Content-Type': 'application/x-sah-ws-4-call+json',
  #     'X-Sah-Request-Type': 'idle',
  #     'X-Requested-With': 'XMLHttpRequest'
  #   },
  #   data='{"service":"NMC","method":"get","parameters":{}}'
  # )
  # r = requests.post('http://192.168.1.1/ws', 
  #   headers={
  #     'Content-Type': 'application/x-sah-ws-4-call+json',
  #     'X-Sah-Request-Type': 'idle',
  #     'X-Requested-With': 'XMLHttpRequest'
  #   },
  #   data='{"service":"DeviceInfo","method":"get","parameters":{}}'
  # )
  # r = requests.post('http://192.168.1.1/ws', headers={'Content-Type': 'application/x-sah-ws-call+json'}, 
  #   data='{"service":"HTTPService","method":"getCurrentUser","parameters":{}}'
  # )

  def get_ipv4_from_lb(self):
    wan_status = requests.post(
      'http://192.168.1.1/ws',
      timeout=self.get_ip_timeout,
      headers={'Content-Type': 'application/x-sah-ws-call+json'},
      json={'service': 'NMC', 'method': 'getWANStatus', 'parameters': {}}
    )
    ip = wan_status.json()['data']['IPAddress']
    if self.is_valid_ipv4_address(ip):
      self.logger.debug('Public IP retrieved from Livebox: {}'.format(ip))
      return ip
    else:
      raise Exception('Failed to fetch valid IP from Livebox')


  # https://stackoverflow.com/questions/33046733/force-requests-to-use-ipv4-ipv6

  # https://ipecho.net/
  # https://ipecho.net/developers.html
  # https://ipecho.net/plain

  # https://www.ipify.org/
  # https://github.com/rdegges/ipify-api
  # https://api.ipify.org/ (v4/v6)

  # https://www.myip.com/api-docs/
  # https://api.myip.com/ (v4/v6 ?)

  # https://seeip.org/
  # https://www.statdns.com/tools/
  # https://github.com/fcambus/telize
  # https://ip4.seeip.org

  # https://whatismyipaddress.com/api
  # https://ipv4bot.whatismyipaddress.com/

  # https://ifconfig.co/
  # https://github.com/mpolden/echoip
  # https://ifconfig.co/ip (v4/v6)

  # https://major.io/icanhazip-com-faq/
  # https://github.com/major/icanhaz
  # https://ipv4.icanhazip.com/

  def get_ipv4_from_websites(self, start_index=0):
    websites_endpoint = [
      'https://ipv4.icanhazip.com/',
      'https://ipecho.net/plain',
      'https://ip4.seeip.org',
      'https://ipv4bot.whatismyipaddress.com/',
    ]
    for website_endpoint in websites_endpoint[start_index:]:
      try:
        ip = requests.get(website_endpoint, timeout=self.get_ip_timeout).text.strip()
        if self.is_valid_ipv4_address(ip):
          self.logger.debug('Public IP retrieved from webservice ({}): {}'.format(website_endpoint, ip))
          return ip
      except:
        pass
    return False


  # https://stackoverflow.com/questions/319279/how-to-validate-ip-address-in-python
  # https://docs.python.org/3/library/ipaddress.html
  @staticmethod
  def is_valid_ipv4_address(address):
    try:
      socket.inet_pton(socket.AF_INET, address)
    except AttributeError:  # no inet_pton here, sorry
      try:
        socket.inet_aton(address)
      except socket.error:
        return False
      return address.count('.') == 3
    except socket.error:  # not a valid address
      return False

    return True


if __name__ == '__main__':
  main()

'''
Installation:

chmod +x /opt/ovh_dynhost.py

# https://www.freedesktop.org/software/systemd/man/systemd.unit.html
# https://www.freedesktop.org/software/systemd/man/systemd.service.html
# https://serverfault.com/questions/812584/in-systemd-whats-the-difference-between-after-and-requires
# https://medium.com/@trstringer/logging-to-systemd-in-python-45150662440a
# https://www.freedesktop.org/software/systemd/python-systemd/journal.html#journalhandler-class
cat <<EOF > /etc/systemd/system/ovh-dynhost.service
[Unit]
Description=OVH Dynhost updater daemon
After=network-online.target
Wants=network-online.target

[Service]
ExecStart=/opt/ovh_dynhost.py --journald /etc/opt/ovh_dynhost.ini
Restart=on-abort

[Install]
WantedBy=multi-user.target
EOF

cat <<EOF > /etc/opt/ovh_dynhost.ini
[dynhost-my-host1]
hostname=my.host.tld
username=host.tld-my
password=OhVeHach3
EOF

chmod u=rw,go= /etc/opt/ovh_dynhost.ini

systemctl daemon-reload
systemctl enable ovh-dynhost
systemctl restart ovh-dynhost
systemctl status ovh-dynhost
'''