-#
-# The views and conclusions contained in the software and documentation are those
-# of the authors and should not be interpreted as representing official policies,
-# either expressed or implied, of the FreeBSD Project.
-
-import urllib.request, socket, sys, argparse
-
-# configuration variables
-server = 'ipv4.ns.ralfj.de'
-domains = ['domain.dyn.ralfj.de'] # list of domains to update
-password = 'yourpassword'
-# END of configuration variables
-
-# allow overwriting some values on the command-line
-parser = argparse.ArgumentParser(description='Update a domain managed by a dyn-nsupdate server')
-parser.add_argument("-s", "--server",
- dest="server", default=server,
- help="The dyn-nsupdate server")
-parser.add_argument("-p", "--password",
- dest="password", default=password,
- help="The password used to update the domains")
-parser.add_argument("-v", "--verbose",
- action="store_true", dest="verbose",
- help="Be more verbose")
-parser.add_argument("domains", metavar='DOMAIN', nargs='*', default=domains,
- help="The domains to update")
-args = parser.parse_args()
-
-def urlopen(url):
- return urllib.request.urlopen(url).read().decode('utf-8').strip()
-
-myip = urlopen('https://'+args.server+'/checkip')
-
-def update_domain(domain):
- '''Update the given domain, using the global server, user, password. Returns True on success, False on failure.'''
- global myip, args
- # check if the domain is already mapped to our current IP
+#==============================================================================
+
+import urllib.request, socket, sys, argparse, os, configparser, itertools, subprocess, re, ssl
+import dns, dns.resolver
+
+VERBOSE_CHANGE = 1
+VERBOSE_FULL = 2
+
+def sslContext(config):
+ if config['DEFAULT'].get('ssl_check_cert', 'yes').lower() in ('0', 'false', 'no'):
+ context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
+ return context
+ else:
+ return None
+
+def readConfig(fname, defSection = 'DEFAULT'):
+ config = configparser.ConfigParser()
+ with open(fname) as file:
+ stream = itertools.chain(("["+defSection+"]\n",), file)
+ config.read_file(stream)
+ return config
+
+def getConfigDir():
+ try:
+ from xdg import BaseDirectory
+ return os.path.join(BaseDirectory.xdg_config_home, "dyn-nsupdate")
+ except ImportError:
+ return os.path.expanduser("~/.config/dyn-nsupdate")
+
+def urlopen(url, config):
+ if sys.version_info >= (3, 4, 3):
+ return urllib.request.urlopen(url, context=sslContext(config)).read().decode('utf-8').strip('\n')
+ else:
+ return urllib.request.urlopen(url).read().decode('utf-8').strip('\n')
+
+def getMyIP(family, config, methods = {}, verbose = 0):
+ '''Returns our current IP address (<family> can be "IPv4" or "IPv6"), detected as given by the configuration.
+ Additional detection methods can be supplied via <methods>.'''
+ method = config[family]['method']
+ if method == 'none':
+ return None
+ elif method == 'remove':
+ return ""
+ elif method == 'web':
+ server = config[family].get('server', config['DEFAULT']['server'])
+ url = 'https://'+server+'/checkip'
+ try:
+ ip = urlopen(url, config)
+ except urllib.error.URLError:
+ raise Exception("Error fetching {}, make sure the URL is correct and the internet connection actually works.".format(url))
+ if verbose >= VERBOSE_FULL:
+ print("Server",server,"says my",family,"is",ip)
+ return ip
+ elif method in methods:
+ return methods[method]()
+ else:
+ raise Exception("Unsupported "+family+" detection method: "+method)
+
+def getMyIPv4(config, verbose = 0):
+ '''Returns our current IPv4 address, detected as given by the configuration'''
+ return getMyIP("IPv4", config, verbose=verbose)
+
+def getMyIPv6(config, verbose = 0):
+ '''Returns our current IPv6 address, detected as given by the configuration'''
+ def local():
+ device = config["IPv6"].get("device")
+ out = subprocess.check_output(["ip", "addr", "show"] + ([] if device is None else ["dev", device]))
+ for line in out.decode('utf-8').split('\n'):
+ m = re.search('inet6 ([a-fA-F0-9:]+)/64 ([a-zA-Z0-9 ]*)', line)
+ if m is not None:
+ ip = m.group(1)
+ flags = m.group(2).split()
+ if not 'temporary' in flags and not 'deprecated' in flags and not "link" in flags:
+ if verbose >= VERBOSE_FULL:
+ print("Local IPv6 detected to be",ip)
+ return ip
+ raise Exception("Unable to detect correct local IPv6 address")
+ return getMyIP("IPv6", config, methods={'local': local}, verbose=verbose)
+
+def getResolver(server):
+ '''Return a resovler with the given server (defined by DNS name)'''
+ addr = socket.getaddrinfo(server, None, family=socket.AF_INET)
+ addr = addr[0][4][0]
+ res = dns.resolver.Resolver()
+ res.nameservers = [addr]
+ return res
+
+def getCurIP(domain, rtype, res):
+ '''Return the current IP of the given <domain>. <rtype> can be A or AAAA.'''
projects / dyn-nsupdate.git / blobdiff