-
Notifications
You must be signed in to change notification settings - Fork 1
/
main.py
110 lines (91 loc) · 3.2 KB
/
main.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
import logging
import os
import signal
import sys
import time
from configparser import ConfigParser
import dns.resolver
import requests
SITES_CONFIG = 'config/sites.cfg'
URI_UPDATE = 'http://www.ovh.com/nic/update?system=dyndns&hostname=%(hostname)s&myip=%(ip)s'
URI_GET_IP = 'https://ifconfig.me/ip'
DNS_RESOLVERS = ['8.8.8.8']
UPDATE_INTERVAL = 60 * 10 # Seconds
class Domain:
"""Represent a domain or sub-domain that can update DNS entries with DynHost (ovh)."""
def __init__(self, domain: str, user: str, password: str):
"""Instanciate the domain object."""
self.domain = domain
self.credentials = (user, password)
self.logger = logging.getLogger(self.domain)
self.old_ip = None
def is_update_to_date(self, current_ip) -> bool:
"""Indicates if the domain address is up to date."""
my_resolver = dns.resolver.Resolver()
my_resolver.nameservers = DNS_RESOLVERS
answers = my_resolver.resolve(
qname=self.domain,
rdtype='A',
)
self.old_ip = answers[0].address
return answers[0].address == current_ip
def update(self, current_ip: str) -> None:
"""Update DNS entry only if the DNS address ip is outdated."""
if self.is_update_to_date(current_ip):
return
data = {
'hostname': self.domain,
'ip': current_ip,
}
req = requests.get(
url=URI_UPDATE % data,
auth=self.credentials,
)
if req.status_code == 200:
self.logger.info('Successfully updated from %s to %s.' %
(self.old_ip, current_ip))
elif req.status_code == 401:
self.logger.error('Failed to update DNS. Bad credentials.')
else:
self.logger.error(
'Failed to update DNS. Unexpected error happened.')
def get_current_ip() -> str:
"""Returns the current device ip address."""
req = requests.get(URI_GET_IP)
return req.text
def signal_handler(signum: int, frame):
print(end="\r")
logging.info("stopped")
sys.exit(0)
def main():
"""Configure logs, and update the domains configured in SITES_CONFIG file."""
os.makedirs('logs', exist_ok=True)
logging.basicConfig(
format='[%(asctime)s][%(levelname)s][%(name)s] %(message)s',
datefmt='%m/%d/%Y %H:%M:%S',
level=logging.INFO,
handlers=[
logging.FileHandler('logs/lastest.log'),
logging.StreamHandler()
],
)
if not os.path.exists(SITES_CONFIG):
logging.error('sites.conf does not exists.')
sys.exit(1)
config = ConfigParser()
config.read(os.path.join(SITES_CONFIG))
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
logging.info("started")
while True:
current_ip = get_current_ip()
for section in config.sections():
domain = Domain(
domain=section,
user=config.get(section, 'user'),
password=config.get(section, 'password'),
)
domain.update(current_ip)
time.sleep(UPDATE_INTERVAL)
if __name__ == '__main__':
main()