#! /usr/bin/python
'''Python client to update your dynamic DNS entry.
Usage:
dyndns.py <userid> <password> [ip=<my-ip>] [wildcard=ON]
dyndns.py -f <filename>|-
dyndns.py -h|-v
The defaults are the server-detected IP address and no wildcards,
so the only mandatory arguments are <userid> and <password>.
The format of the configuration file is as follows:
<userid>
<password>
[ip=<aaa.bbb.ccc.ddd>]
[wildcard=ON]
[MX=...]
...
userid and password must be the first two lines of the file, the remaining
parameters depend on the respective dynamic DNS provider.
The file name "-" means to read the configuration form standard input.
This is useful if you have the configuration in an encrypted file, like this:
decrypt dyndns.conf | dyndns.py -f -
See also www.dyndns.org, www.yi.org, eyep.net, gnudip.cheapnet.net'''
import sys, os, time, string, re, urllib, httplib, base64
import nettools
def day():
'''returns the number of days since the Epoch (1970-01-01)'''
return int(time.time() / 24 / 3600)
################################################################
class DynamicDNS:
'''This is the abstract base class for all dynamic DNS clients.'''
version = 'dyndns.py 0.2/2000-04-14 <beat.bolli@earthling.net>'
server = '' # host [:port] to send request to
url = '' # base request URL
request = 'GET' # or POST
command_ip = '' # name of IP argument to server
command_host = '' # name of host name argument to server
commands = [] # list of other known commands
arguments = {} # hash of arguments and their values
str_success = '' # unique string indicating success
str_error = '' # unique string indicating error
policy = 0 # 0: update always
# -1: update when IP address changes only
# n: update when IP address changes,
# but at least every n days (n>0)
verbose = 1
def __init__(self):
self.name = self.__class__.__name__
self.ip = nettools.localaddr()
self.host = ''
self.cfg = []
try:
home = os.environ['HOME']
except KeyError:
home = '/etc'
if not os.path.isdir(home):
home = ''
self.cfg_file = home + '/.dyndns'
self.state = 3, 'not finished'
def read_config(self):
try:
f = open(self.cfg_file)
self.cfg = f.readlines()
f.close()
except:
print 'Can not read configuration file'
self.cfg = []
for i in range(len(self.cfg)):
c = self.cfg[i].split(':')
if len(c) >= 4 \
and c[0] == self.name and c[1] == self.host:
del self.cfg[i]
return c[2], int(c[3])
return '*', 0
def write_config(self):
c = ':'.join( (self.name, self.host, self.ip, str(day())) )
self.cfg.insert(0, c + '\n')
try:
f = open(self.cfg_file, 'w')
f.writelines(self.cfg)
f.close()
except:
print 'Can not write configuration file'
def update(self, userid, password, args):
self.userid = userid
self.password = password
# scan the arguments
for arg in args:
arg = arg.strip()
if '=' not in arg or arg[:1] == '#':
continue
key, val = arg.split('=', 1)
key, val = key.strip().lower(), val.strip()
if not val:
continue
if key == 'ip':
if self.command_ip:
self.arguments[self.command_ip] = self.ip = val
else:
print 'Ignoring IP address', val
elif key == 'host':
if self.command_host:
self.arguments[self.command_host] = self.host = val
else:
print 'Ignoring host name', val
elif key in self.commands:
self.arguments[key] = val
else:
print 'Ignoring argument', arg
# check the policy
last_ip, last_update = self.read_config()
if last_ip == self.ip: # policy applies only unless address has changed
if self.policy == -1 \
or self.policy > 0 \
and last_update + self.policy >= day():
if self.verbose:
print self.name, 'policy says update is not neccessary.'
self.state = 0, 'update not neccessary'
return
# talk to the server
errcode, errmsg, headers, reply = self.protocol()
if self.verbose:
print errcode, errmsg
# print headers
print reply
# check for success
if errcode == 200:
if re.search(self.str_success, reply):
self.state = 0, 'update successful'
self.write_config()
elif self.str_error and re.search(self.str_error, reply):
self.state = 1, 'error:\n\n' + reply
else:
self.state = 2, 'not OK'
else:
self.state = -errcode, errmsg
def protocol(self):
'''The default HTTP protocol'''
# allow some special treatment by the subclass
self.url_hook()
if self.url is None:
if self.verbose:
print 'Cancelled...'
return 0, 'Cancel', '', ''
# stick the arguments onto the URL
content = urllib.urlencode(self.arguments)
if self.request == 'GET':
self.url = self.url + content
content = None
# authenticate
auth = self.auth()
# are we going through a proxy?
try:
server = os.environ['http_proxy']
self.url = 'http://' + self.server + self.url
except KeyError:
server = self.server
if self.verbose:
print 'Server:', server
if auth:
print 'Authorization:', auth
print self.request, self.url
print
# send the request
h = httplib.HTTP(server)
h.putrequest(self.request, self.url)
h.putheader('Host', self.server)
h.putheader('Connection', 'close')
h.putheader('Accept', '*/*')
h.putheader('User-Agent', self.version)
if auth:
h.putheader('Authorization', auth)
if content:
h.putheader('Content-Type', 'application/x-www-form-urlencoded')
h.putheader('Content-Length', '%d' % len(content))
h.endheaders()
if content:
if self.verbose:
print content
h.send(content)
# return the results
errcode, errmsg, headers = h.getreply()
reply = h.getfile().read()
return errcode, errmsg, headers, reply
def url_hook(self):
pass
def auth_none(self):
return ''
def auth_basic(self):
a = base64.encodestring(self.userid + ':' + self.password)
return 'Basic ' + a.strip()
# def auth_xxx(sel): # to be defined...
auth = auth_basic # default authentication scheme
def __str__(self):
return self.name + ': %d (%s)' % self.state
def ok(self):
return self.state[0] == 0
################################################################
class dyndns_org(DynamicDNS):
command_ip = 'myip'
command_host = 'hostname'
commands = ['wildcard', 'mx', 'backmx']
str_success = 'good '
server = 'members.dyndns.org'
url = '/nic/update?'
policy = 20
################################################################
class yi_org(DynamicDNS):
command_ip = 'ipaddr'
str_success = 'STATUS:OK'
str_error = 'STATUS:ERROR'
server = 'www.yi.org'
url = '/bin/dyndns.fcgi?'
policy = 1
################################################################
class eyep_net(DynamicDNS):
command_ip = 'ip'
command_host = 'host'
server = 'eyep.net'
url = '/update-new.php3/'
str_success = 'RETURN4'
auth = DynamicDNS.auth_none
def url_hook(self):
self.url = self.url + self.userid + '/' + \
self.password + '/' + self.arguments['host'] + '/'
if self.arguments.has_key('ip'):
self.url = self.url + self.arguments['ip'] + '/'
self.arguments = {}
################################################################
class ez_ip_net(DynamicDNS):
command_ip = 'ipaddress'
commands = ['mx', 'wildcard', 'preferred']
arguments = {'mode': 'update'}
server = 'www.ez-ip.net'
url = '/members/update/?'
str_success = 'OK'
def url_hook(self):
if self.arguments.has_key('preferred'):
self.url = '/members/preferred/update/?'
del self.arguments['preferred']
################################################################
class gnudip(DynamicDNS):
commands = ['domain']
arguments = { 'updatehost': 'Go' }
server = 'www.cheapnet.net'
url = '/cgi-bin/gnudip2.cgi/'
request = 'POST'
def url_hook(self):
self.str_success = string.replace(self.ip, '.', r'\.')
self.arguments['username'] = self.userid
self.arguments['password'] = self.password
################################################################
class gnudip_MD5(DynamicDNS):
server = 'gnudip.cheapnet.net'
port = 3495
command_host = 'domain'
str_success = '0'
str_error = '1'
verbose = 1
def digest(self, s):
'''returns a hex representation of s's MD5 hash'''
import md5
d = md5.new(s).digest()
return string.join(map(
lambda c: string.zfill(hex(ord(c))[2:], 2), d
), '')
def protocol(self):
'''This server uses a propietary protocol.'''
import socket
# connect to the server
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect( (self.server, self.port) )
# read the server challenge (10 random characters)
challenge = s.recv(64)[:10]
if self.verbose:
print 'Challenge:', challenge
# calculate the digest of our password's digest plus the challenge
digest = self.digest(self.password)
secret = self.digest(digest + '.' + challenge)
# send the update command
cmd = string.join( (self.userid, secret, self.host, '0'), ':')
if self.verbose:
print 'Command:', cmd
s.send(cmd + '\n')
# read the reply
reply = str(s.recv(1024))
return 200, 'OK', '', reply
################################################################
class uptime_net(DynamicDNS):
command_host = 'hid'
server = 'data.uptimes.net'
url = '/server.html'
request = 'POST'
string_success = 'UP4: 000'
def procread(self, file):
return open('/proc/' + file).read().split()
def url_hook(self):
args = self.arguments
import os
# uptime and idle percentage
try:
up, idle = map(float, self.procread('uptime')[:2])
except IOError:
self.url = None
return
args['uptime'] = '%d' % (up / 60)
args['idle'] = '%d' % (100 * idle / up)
# load
try:
args['load'] = self.procread('loadavg')[0]
except IOError:
pass
# OS, OS level and CPU type
if hasattr(os, 'uname'):
u = os.uname()
args['os'] = u[0]
args['oslevel'] = u[2]
args['cpu'] = u[4]
################################################################
def update(DynamicDNSProvider, userid, password, *args):
p = DynamicDNSProvider() # class parameter!
p.update(userid, password, args)
print p
if not p.ok():
print 'Press <Enter> to continue'
sys.stdin.readline() # wait
print '-' * 64
def main():
argv = sys.argv[1:]
argc = len(argv)
if argc == 0 or argv[0] == '-h':
print __doc__
return
elif argv[0] == '-v':
print DynDNS.version
return
elif argv[0][:2] == '-f':
if argc == 1 and argv[0][2:]:
file = argv[0][2:]
elif argc == 2:
file = argv[1]
else:
print 'Missing filename'
return
if file == '-':
fp = sys.stdin
else:
try:
fp = open(file)
except IOError:
print 'Error opening', file
return
argv = fp.read().split()
fp.close()
elif argc < 2:
print 'Missing argument'
return
update(dyndns_org, *argv)
################################################################
update(dyndns_org, 'test', 'test', 'host=test.ath.cx') # DynDNS test account
print 'Done.'
Quelltext von py/dyndns.py
© 2005-2024 Beat "2b" Bolli
