Cleanup, implement zero padding IP octets, switch to fstrings

This commit is contained in:
meemo committed 2022-02-06 18:56:35 -05:00
1 parent 2c96403522
commit 0cf217b088
1 file changed
+63 -40
+63 -40
View File
@@ -12,20 +12,30 @@ import requests
import json import json
import sys import sys
def get_platform(): def get_platform():
platforms = { platforms = {
'linux1' : 'Linux', 'linux1': 'Linux',
'linux2' : 'Linux', 'linux2': 'Linux',
'darwin' : 'macOS', 'darwin': 'macOS',
'win32' : 'Windows' 'win32': 'Windows'
} }
if sys.platform not in platforms: if sys.platform not in platforms:
return sys.platform return sys.platform
return platforms[sys.platform] return platforms[sys.platform]
SUDOMEMODNS_VERSION = "1.2" SUDOMEMODNS_VERSION = "1.2"
# Adds preceding zeros to IP addresses
# e.g `1.1.1.1` becomes `001.001.001.001`
def format_ip(address):
octets = str(address).split(".")
return f"{int(octets[0]):02d}.{int(octets[1]):02d}.{int(octets[2]):02d}.{int(octets[3]):02d}"
def get_ip(): def get_ip():
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try: try:
@@ -38,25 +48,27 @@ def get_ip():
s.close() s.close()
return IP return IP
EPOCH = datetime(1970, 1, 1) EPOCH = datetime(1970, 1, 1)
SERIAL = int((datetime.utcnow() - EPOCH).total_seconds()) SERIAL = int((datetime.utcnow() - EPOCH).total_seconds())
MY_IP = get_ip() MY_IP = get_ip()
print("+===============================+") print("+===============================+")
print("| Sudomemo DNS Server |") print("| Sudomemo DNS Server |")
print("| Version " + SUDOMEMODNS_VERSION + " |") print(f"| Version {SUDOMEMODNS_VERSION} |")
print("+===============================+\n") print("+===============================+\n")
print("== Welcome to sudomemoDNS! ==") print("== Welcome to sudomemoDNS! ==")
print("This server will allow you to connect to Sudomemo when your Internet Service Provider does not work with custom DNS.\n") print(
"This server will allow you to connect to Sudomemo when your Internet Service Provider does not work with custom DNS.\n")
print("== How To Use ==") print("== How To Use ==")
print("First, make sure that your console is connected to the same network as this computer.\n") print("First, make sure that your console is connected to the same network as this computer.\n")
print("Then, put these settings in for DNS on your console:") print("Then, put these settings in for DNS on your console:")
print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~") print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~")
print("Primary DNS: ",MY_IP) print(f"Primary DNS: {MY_IP}")
print("Secondary DNS: 8.8.8.8") print("Secondary DNS: 008.008.008.008")
print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~\n") print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~\n")
print("== Getting Help ==") print("== Getting Help ==")
@@ -74,24 +86,32 @@ TYPE_LOOKUP = {
TXT: QTYPE.TXT, TXT: QTYPE.TXT,
} }
# Can't seem to turn off DNSLogger with a None type so let's just null it out with a dummy function # Can't seem to turn off DNSLogger with a None type so let's just null it out with a dummy function
class SudomemoDNSLogger(object): class SudomemoDNSLogger(object):
def log_recv(self, handler, data): def log_recv(self, handler, data):
pass pass
def log_send(self, handler, data): def log_send(self, handler, data):
pass pass
def log_request(self, handler, request): def log_request(self, handler, request):
print("[INFO] Received DNS request from console at " + handler.client_address[0]) print("[INFO] Received DNS request from console at " + handler.client_address[0])
def log_reply(self, handler, reply): def log_reply(self, handler, reply):
print("[INFO] Sent response to console at " + handler.client_address[0]) print("[INFO] Sent response to console at " + handler.client_address[0])
def log_error(self, handler, e): def log_error(self, handler, e):
print("[ERROR] Invalid DNS request from " + handler.client_address[0]) print("[ERROR] Invalid DNS request from " + handler.client_address[0])
def log_truncated(self, handler, reply): def log_truncated(self, handler, reply):
pass pass
def log_data(self, dnsobj): def log_data(self, dnsobj):
pass pass
class Record: class Record:
def __init__(self, rdata_type, *args, rtype=None, rname=None, ttl=None, **kwargs): def __init__(self, rdata_type, *args, rtype=None, rname=None, ttl=None, **kwargs):
if isinstance(rdata_type, RD): if isinstance(rdata_type, RD):
@@ -103,12 +123,12 @@ class Record:
if rdata_type == SOA and len(args) == 2: if rdata_type == SOA and len(args) == 2:
# add sensible times to SOA # add sensible times to SOA
args += (( args += ((
SERIAL, # serial number SERIAL, # serial number
60 * 60 * 1, # refresh 60 * 60 * 1, # refresh
60 * 60 * 3, # retry 60 * 60 * 3, # retry
60 * 60 * 24, # expire 60 * 60 * 24, # expire
60 * 60 * 1, # minimum 60 * 60 * 1, # minimum
),) ),)
rdata = rdata_type(*args) rdata = rdata_type(*args)
if rtype: if rtype:
@@ -144,26 +164,28 @@ class Record:
ZONES = {} ZONES = {}
try: try:
get_zones = requests.get("https://www.sudomemo.net/api/dns_zones.json", headers={'User-Agent': 'SudomemoDNS/' + SUDOMEMODNS_VERSION + ' (' + get_platform() + ')'}) get_zones = requests.get("https://www.sudomemo.net/api/dns_zones.json",
headers={'User-Agent': 'SudomemoDNS/' + SUDOMEMODNS_VERSION + ' (' + get_platform() + ')'})
except requests.exceptions.Timeout: except requests.exceptions.Timeout:
print("[ERROR] Unable to load DNS data: Connection to Sudomemo timed out. Are you connected to the Internet?") print("[ERROR] Unable to load DNS data: Connection to Sudomemo timed out. Are you connected to the Internet?")
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
print("[ERROR] Unable load DNS data.") print("[ERROR] Unable load DNS data.")
print("[ERROR] Exception: ",e) print("[ERROR] Exception: ", e)
sys.exit(1) sys.exit(1)
try: try:
zones = json.loads(get_zones.text) zones = json.loads(get_zones.text)
except ValueError as e: except ValueError as e:
print("[ERROR] Unable load DNS data: Invalid response from server. Check that you can visit sudomemo.net") print("[ERROR] Unable load DNS data: Invalid response from server. Check that you can visit sudomemo.net")
for zone in zones: for zone in zones:
if zone["type"] == "a": if zone["type"] == "a":
ZONES[zone["name"]] = [ Record(A, zone["value"]) ] ZONES[zone["name"]] = [Record(A, zone["value"])]
elif zone["type"] == "p": elif zone["type"] == "p":
ZONES[zone["name"]] = [ Record(A, socket.gethostbyname(zone["value"])) ] ZONES[zone["name"]] = [Record(A, socket.gethostbyname(zone["value"]))]
print("[INFO] DNS information loaded successfully.") print("[INFO] DNS information loaded successfully.")
class Resolver: class Resolver:
def __init__(self): def __init__(self):
self.zones = {DNSLabel(k): v for k, v in ZONES.items()} self.zones = {DNSLabel(k): v for k, v in ZONES.items()}
@@ -171,6 +193,7 @@ class Resolver:
def resolve(self, request, handler): def resolve(self, request, handler):
reply = request.reply() reply = request.reply()
zone = self.zones.get(request.q.qname) zone = self.zones.get(request.q.qname)
if zone is not None: if zone is not None:
for zone_records in zone: for zone_records in zone:
rr = zone_records.try_rr(request.q) rr = zone_records.try_rr(request.q)
@@ -196,26 +219,26 @@ dnsLogger = SudomemoDNSLogger()
print("[INFO] Detected operating system:", get_platform()) print("[INFO] Detected operating system:", get_platform())
if get_platform() == 'linux': if get_platform() == 'linux':
print("[INFO] Please note that you will have to run this as root or with permissions to bind to UDP port 53.") print("[INFO] Please note that you will have to run this as root or with permissions to bind to UDP port 53.")
print("[INFO] If you aren't seeing any requests, check that this is the case first with lsof -i:53 (requires lsof)") print("[INFO] If you aren't seeing any requests, check that this is the case first with lsof -i:53 (requires lsof)")
print("[INFO] To run as root, prefix the command with 'sudo'") print("[INFO] To run as root, prefix the command with 'sudo'")
elif get_platform() == 'macOS': elif get_platform() == 'macOS':
print("[INFO] Please note that you will have to run this as root or with permissions to bind to UDP port 53.") print("[INFO] Please note that you will have to run this as root or with permissions to bind to UDP port 53.")
print("[INFO] If you aren't seeing any requests, check that this is the case first with lsof -i:53 (requires lsof)") print("[INFO] If you aren't seeing any requests, check that this is the case first with lsof -i:53 (requires lsof)")
print("[INFO] To run as root, prefix the command with 'sudo'") print("[INFO] To run as root, prefix the command with 'sudo'")
elif get_platform() == 'Windows': elif get_platform() == 'Windows':
print("[INFO] Please note that you may have to allow this application through the firewall. If so, a popup will appear in a moment.") print("[INFO] Please note that you may have to allow this application through the firewall. If so, a popup will appear in a moment.")
print("[INFO] If you are not seeing any requests, make sure you have allowed this application through the firewall. If you have already done so, disregard this message.") print("[INFO] If you are not seeing any requests, make sure you have allowed this application through the firewall. If you have already done so, disregard this message.")
try: try:
servers = [ servers = [
DNSServer(resolver=resolver, port=53, address=MY_IP, tcp=True, logger=dnsLogger), DNSServer(resolver=resolver, port=53, address=MY_IP, tcp=True, logger=dnsLogger),
DNSServer(resolver=resolver, port=53, address=MY_IP, tcp=False, logger=dnsLogger), DNSServer(resolver=resolver, port=53, address=MY_IP, tcp=False, logger=dnsLogger)
] ]
except PermissionError:
print("[ERROR] Permission error: Check that you are running this as an administrator or root")
sys.exit(1)
except PermissionError:
print("[ERROR] Permission error: Check that you are running this as an administrator or root")
sys.exit(1)
print("[INFO] sudomemoDNS is ready. Now waiting for DNS requests from your console...") print("[INFO] sudomemoDNS is ready. Now waiting for DNS requests from your console...")