-
Notifications
You must be signed in to change notification settings - Fork 0
/
updater.py
221 lines (197 loc) · 8.02 KB
/
updater.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
#!/usr/bin/env python
import requests
from requests.auth import HTTPBasicAuth
import time
import ctypes
import os
import logging
import argparse
import startup_utils
logging.basicConfig(
filename='{}/ddns-update-service.log'.format(os.path.dirname(__file__)),
level=logging.DEBUG,
format='%(asctime)s [ddns-update] %(levelname)-8s %(message)s'
)
_minute = 60
_hour = 3600
class UpdaterError(Exception):
pass
class InputError(UpdaterError):
def __init__(self, msg):
self.msg = msg
class BadError(UpdaterError):
def __init__(self, msg):
self.msg = msg
class Updater(object):
__auth = None
__hostname = None
__last_ip = None
__daemon = None
def __init__(self, _username, _password, _hostname, _daemon=False):
"""
No-IP.com DDNS updater
:param _username: Login name for NO-IP
:param _password: Login password for NO-IP
:param _hostname: Host name to update
:param _daemon: True for enable auto start-up. Default: False
"""
logging.info('')
logging.info('===============================================')
logging.info('Service Initializing')
logging.debug('username: {}'.format(_username))
self.__auth = HTTPBasicAuth(_username, _password)
logging.debug('hostname: {}'.format(_hostname))
self.__hostname = _hostname
logging.debug('daemon: {}'.format(_daemon))
self.__daemon = _daemon
if self.__daemon and not os.path.exists(startup_utils.get_path()):
if not os.path.exists(startup_utils.get_file_name()):
startup_utils.add_startup(arguments.username, arguments.password, arguments.hostname)
else:
logging.warning('Startup script need to be copied to startup folder')
ctypes.windll.user32.MessageBoxA(0, 'Startup script need to be copied to startup folder.',
'No-Ip Updater', 0)
def start(self):
"""
Start running the update process.
If in daemon mode, will loop forever and try to update every 2.5 hours.
"""
continue_running = True
logging.info('Starting service run...')
while continue_running:
continue_running = self._update() and self.__daemon
if continue_running:
Updater._start_hours_delay(2.5)
logging.info('Stopping service...')
def _update(self):
"""
Start update sequence
:return: True if successfully ran, else False
"""
logging.info('Getting current IP address')
# get the current IP and check if changed
new_ip = self._get_ip()
is_different = new_ip != self.__last_ip
logging.debug('if {} != {} : {}'.format(new_ip, self.__last_ip, is_different))
if is_different:
# IP changed so will send an update to NO-IP
try:
logging.info('IP was changed. Sending update to NO-IP')
self._send_update(new_ip)
logging.info('IP updated successfully')
return_value = True
except InputError as ex:
self._show_message(ex.msg)
logging.error('Input Error: {}'.format(ex.msg))
return_value = False
except BadError as ex:
log_msg = popup_msg = ''
if self.__daemon:
popup_msg = '\n\rRemoving script from start-up folder.'
log_msg = ' Removing script from start-up folder - {}'.format(startup_utils.get_path())
startup_utils.remove_startup()
self._show_message('Critical error.{}'.format(popup_msg))
logging.critical('{}.'.format(ex.msg, log_msg))
return_value = False
else:
# IP didn't change
logging.info('No change in IP. Skipping...')
return_value = True
return return_value
@staticmethod
def _get_ip():
"""
Gets the host IP address
:return: current IP address
"""
ip = None
while ip is None or ip == '':
try:
ip = requests.get('https://httpbin.org/ip').json()['origin']
except requests.ConnectionError:
logging.error('Connection to "httpbin.org" encountered an error. Retry in 5 minutes...')
Updater._start_minutes_delay(5)
except requests.Timeout:
logging.error('Connection to "httpbin.org" timed out. Retry in 1.5 minutes...')
Updater._start_minutes_delay(1.5)
logging.info('Current IP is: ' + ip)
return ip
def _send_update(self, new_ip):
"""
Send an update command to NO-IP
:param new_ip: the new IP of the host
"""
update_url = 'https://dynupdate.no-ip.com/nic/update'
payload = {'hostname': self.__hostname, 'myip': new_ip}
headers = {'user-agent': 'python update client Win10/ [email protected]'}
logging.debug(headers)
again = True
# start requesting until success or error
while again:
r = None
try:
r = requests.get(update_url, params=payload, auth=self.__auth, headers=headers)
except requests.ConnectionError:
logging.error('Connection to "no-ip.com" encountered an error. Retry in 5 minutes...')
self._start_minutes_delay(5)
again = True
except requests.Timeout:
logging.error('Connection to "no-ip.com" timed out. Retry in 1.5 minutes...')
self._start_minutes_delay(1.5)
again = True
if r is not None:
if 'nohost' in r.text:
raise InputError('No Host')
elif 'badauth' in r.text:
raise InputError('Bad Authentication')
elif 'badagent' in r.text:
raise BadError('Bad Agent')
elif 'abuse' in r.text:
raise BadError('Abuse')
elif '!donator' in r.text:
raise BadError('Not Donator')
elif '911' in r.text:
self._start_minutes_delay(35)
again = True
elif 'good ' + new_ip in r.text or 'nochg ' + new_ip in r.text:
again = False
self.__last_ip = new_ip
logging.debug('last ip updated to: {}'.format(new_ip))
@staticmethod
def _show_message(message):
"""
Shows a message window in case of an error
:param message: the message to show
"""
ctypes.windll.user32.MessageBoxA(0, message, 'No-Ip Updater', 0)
@staticmethod
def _start_minutes_delay(amount_of_minutes):
"""
Set up a delay in minutes
:param amount_of_minutes: how many minutes to hold
"""
time.sleep(_minute * amount_of_minutes)
@staticmethod
def _start_hours_delay(amount_of_hours):
"""
Set up a delay in hours
:param amount_of_hours: how many hours to hold
"""
time.sleep(_hour * amount_of_hours)
if __name__ == "__main__":
argument_parser = argparse.ArgumentParser()
argument_parser.add_argument('username',
type=str,
help='User name at no-ip.com')
argument_parser.add_argument('password',
type=str,
help='Password of the user')
argument_parser.add_argument('hostname',
type=str,
help='Host name URL')
argument_parser.add_argument('-s', '--startup',
action='store_true',
help='Configure to run on startup')
arguments = argument_parser.parse_args()
updater = Updater(arguments.username, arguments.password, arguments.hostname, arguments.startup)
updater.start()