Created
April 2, 2025 08:51
-
-
Save YukariChiba/9fce6dce9744b253ed285f11e2aee1e9 to your computer and use it in GitHub Desktop.
ddns server
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| import argparse | |
| import datetime | |
| import sys | |
| import time | |
| import threading | |
| import traceback | |
| import socketserver | |
| import struct | |
| import hashlib | |
| import os.path | |
| import ipaddress | |
| import json | |
| import string | |
| try: | |
| import netifaces as ni | |
| except ImportError: | |
| print("Missing dependency netifaces: <https://pypi.python.org/pypi/netifaces>. Please install it with `pip`.") | |
| sys.exit(2) | |
| try: | |
| from dnslib import * | |
| except ImportError: | |
| print("Missing dependency dnslib: <https://pypi.python.org/pypi/dnslib>. Please install it with `pip`.") | |
| sys.exit(2) | |
| class DomainName(str): | |
| def __getattr__(self, item): | |
| return DomainName(item + '.' + self) | |
| D = DomainName("wh1.cn.strexp.net.") | |
| NSNAME = "ns.wh1.cn.strexp.net" | |
| TTL = 60 * 5 | |
| soa_record = SOA( | |
| mname=D, | |
| rname=D.admin, | |
| times=( | |
| 1145141919, # serial number | |
| 60 * 60 * 1, # refresh | |
| 60 * 60 * 3, # retry | |
| 60 * 60 * 24, # expire | |
| 60 * 60 * 1, # minimum | |
| ) | |
| ) | |
| ns_record = NS(NSNAME) | |
| def dns_response(data): | |
| request = DNSRecord.parse(data) | |
| reply = DNSRecord(DNSHeader(id=request.header.id, qr=1, aa=1, ra=0), q=request.q) | |
| qname = request.q.qname | |
| qn = str(qname) | |
| qtype = request.q.qtype | |
| qt = QTYPE[qtype] | |
| if qn == D and qt in ["AAAA"]: | |
| selfip = ni.ifaddresses('ppp0')[ni.AF_INET6][0]['addr'] | |
| reply.add_answer(RR(rname=qname, rtype=QTYPE.AAAA, rclass=1, ttl=TTL, rdata=AAAA(str(selfip)))) | |
| if qn == D and qt in ["NS"]: | |
| reply.add_answer(RR(rname=qn, rtype=QTYPE.NS, rclass=1, ttl=TTL, rdata=ns_record)) | |
| if qn == D and qt in ["SOA"]: | |
| reply.add_answer(RR(rname=qn, rtype=QTYPE.SOA, rclass=1, ttl=TTL, rdata=soa_record)) | |
| return reply.pack() | |
| return reply.pack() | |
| class BaseRequestHandler(socketserver.BaseRequestHandler): | |
| def get_data(self): | |
| raise NotImplementedError | |
| def send_data(self, data): | |
| raise NotImplementedError | |
| def handle(self): | |
| now = datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f') | |
| print("\n\n%s request %s (%s %s):" % (self.__class__.__name__[:3], now, self.client_address[0], | |
| self.client_address[1])) | |
| try: | |
| data = self.get_data() | |
| respdata = dns_response(data) | |
| self.send_data(respdata) | |
| except Exception: | |
| traceback.print_exc(file=sys.stderr) | |
| class UDPRequestHandler(BaseRequestHandler): | |
| def get_data(self): | |
| return self.request[0].strip() | |
| def send_data(self, data): | |
| return self.request[1].sendto(data, self.client_address) | |
| def main(): | |
| s = socketserver.ThreadingUDPServer(("127.0.0.1", 5353), UDPRequestHandler) | |
| thread = threading.Thread(target=s.serve_forever) # that thread will start one more thread for each request | |
| thread.daemon = True # exit the server thread when the main thread terminates | |
| thread.start() | |
| try: | |
| while 1: | |
| time.sleep(1) | |
| sys.stderr.flush() | |
| sys.stdout.flush() | |
| except KeyboardInterrupt: | |
| pass | |
| finally: | |
| s.shutdown() | |
| if __name__ == '__main__': | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment