From 5026178e21e38f2859fdb6d57b689ef5162711c7 Mon Sep 17 00:00:00 2001 From: Rekushi <170186554+RekuNote@users.noreply.github.com> Date: Tue, 15 Sep 2026 20:42:48 +0100 Subject: [PATCH] Add reximemoDNS.py file --- reximemoDNS.py | 244 ++++++++++++++++++++++++++++++++++++++++++++++ sudomemoDNS.py | 259 ------------------------------------------------- 2 files changed, 244 insertions(+), 259 deletions(-) create mode 100644 reximemoDNS.py delete mode 100644 sudomemoDNS.py diff --git a/reximemoDNS.py b/reximemoDNS.py new file mode 100644 index 0000000..2824c8d --- /dev/null +++ b/reximemoDNS.py @@ -0,0 +1,244 @@ +# reximemoDNS +# +# RexiMemo adaptation of the MIT-licensed sudomemoDNS project. +# See LICENSE and NOTICE.md for attribution. + +from datetime import datetime +from ipaddress import IPv4Address, ip_address +import os +from socket import AF_INET, SOCK_DGRAM, socket +from sys import platform +from time import sleep + +from dnslib import A, DNSLabel, DNSRecord, QTYPE, RCODE, RR +from dnslib.server import DNSServer + + +REXIMEMODNS_VERSION = "1.0.0" +DEFAULT_REXIMEMO_SERVER_IP = "141.147.76.115" +DEFAULT_UPSTREAM_DNS = "8.8.8.8" + +# Keep this list aligned with RexiMemo's LOCAL_DNS_HOSTS in network_services.py. +# conntest.nintendowifi.net is intentionally not intercepted. +REXIMEMO_HOSTS = ( + "nas.nintendowifi.net", + "flipnote.hatena.com", + "ugomemo.hatena.ne.jp", +) + + +def get_platform(): + platforms = { + "linux": "Linux", + "linux1": "Linux", + "linux2": "Linux", + "darwin": "macOS", + "win32": "Windows", + } + return platforms.get(platform, platform) + + +def format_ip(address): + """Format an IPv4 address the way Nintendo connection screens display it.""" + octets = str(address).split(".") + return ".".join(f"{int(octet):03d}" for octet in octets) + + +def _ipv4_env(name, default): + value = (os.environ.get(name) or default).strip() + try: + parsed = ip_address(value) + except ValueError as exc: + raise SystemExit(f"[ERROR] {name} must be a valid IPv4 address; got {value!r}") from exc + if not isinstance(parsed, IPv4Address): + raise SystemExit(f"[ERROR] {name} must be an IPv4 address; got {value!r}") + return str(parsed) + + +def get_ip(): + configured = (os.environ.get("REXIMEMODNS_BIND_IP") or "").strip() + if configured: + return _ipv4_env("REXIMEMODNS_BIND_IP", configured) + + probe = socket(AF_INET, SOCK_DGRAM) + try: + # The destination does not need to be reachable. Connecting a UDP socket + # is only used to discover which local IPv4 address the OS would route. + probe.connect(("10.255.255.255", 1)) + return probe.getsockname()[0] + except OSError: + return "127.0.0.1" + finally: + probe.close() + + +def _port_env(): + raw = (os.environ.get("REXIMEMODNS_PORT") or "53").strip() + try: + value = int(raw) + except ValueError as exc: + raise SystemExit(f"[ERROR] REXIMEMODNS_PORT must be a number; got {raw!r}") from exc + if not 1 <= value <= 65535: + raise SystemExit("[ERROR] REXIMEMODNS_PORT must be between 1 and 65535") + return value + + +MY_IP = get_ip() +REXIMEMO_SERVER_IP = _ipv4_env("REXIMEMODNS_SERVER_IP", DEFAULT_REXIMEMO_SERVER_IP) +UPSTREAM_DNS = _ipv4_env("REXIMEMODNS_UPSTREAM_DNS", DEFAULT_UPSTREAM_DNS) +DNS_PORT = _port_env() +SERIAL = int((datetime.utcnow() - datetime(1970, 1, 1)).total_seconds()) + + +def _print_banner(): + print("+===============================+") + print("| RexiMemo DNS Server |") + print(f"| Version {REXIMEMODNS_VERSION:<13}|") + print("+===============================+\n") + + print("== Welcome to reximemoDNS! ==") + print( + "This local DNS helper routes Flipnote Studio's Nintendo/Hatena service names " + "to RexiMemo while forwarding unrelated DNS normally.\n" + ) + + print("== How To Use ==") + 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("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~") + print(f"Primary DNS: {format_ip(MY_IP)}") + print(f"Secondary DNS: {format_ip(DEFAULT_UPSTREAM_DNS)}") + print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~\n") + + print("== RexiMemo Routing ==") + print(f"RexiMemo server: {REXIMEMO_SERVER_IP}") + for hostname in REXIMEMO_HOSTS: + print(f" {hostname} -> {REXIMEMO_SERVER_IP}") + print(f"Other DNS queries -> {UPSTREAM_DNS}\n") + + print("== Getting Help ==") + print("Join the RexiMemo Discord server: https://discord.gg/K5mKWvaruF\n") + + +class RexiMemoDNSLogger(object): + def log_recv(self, handler, data): + pass + + def log_send(self, handler, data): + pass + + def log_request(self, handler, request): + qname = str(request.q.qname).rstrip(".") + qtype = QTYPE.get(request.q.qtype, str(request.q.qtype)) + print( + "[INFO] DNS request from %s: %s %s" + % (handler.client_address[0], qtype, qname) + ) + + def log_reply(self, handler, reply): + answer_count = len(reply.rr) + print( + "[INFO] DNS response to %s: %d answer%s" + % (handler.client_address[0], answer_count, "" if answer_count == 1 else "s") + ) + + def log_error(self, handler, error): + print("[ERROR] Invalid DNS request from %s: %s" % (handler.client_address[0], error)) + + def log_truncated(self, handler, reply): + pass + + def log_data(self, dnsobj): + pass + + +class Resolver: + def __init__(self): + self.reximemo_names = {DNSLabel(name) for name in REXIMEMO_HOSTS} + + def _reximemo_reply(self, request): + reply = request.reply() + # Mirror RexiMemo's own DNS behavior: A/ANY gets the legacy-service IPv4 + # address. Other query types get an authoritative empty NOERROR response. + if request.q.qtype in (QTYPE.A, QTYPE.ANY): + reply.add_answer( + RR( + rname=request.q.qname, + rtype=QTYPE.A, + rclass=1, + ttl=0, + rdata=A(REXIMEMO_SERVER_IP), + ) + ) + return reply + + def _forward(self, request, handler): + try: + raw = request.send( + UPSTREAM_DNS, + 53, + tcp=(getattr(handler, "protocol", "udp") == "tcp"), + timeout=3, + ) + reply = DNSRecord.parse(raw) + # Never relay a mismatched DNS transaction as if it answered this query. + if reply.header.id != request.header.id: + raise ValueError("upstream DNS transaction ID mismatch") + return reply + except Exception as exc: + print(f"[ERROR] Upstream DNS lookup via {UPSTREAM_DNS} failed: {exc}") + reply = request.reply() + reply.header.rcode = RCODE.SERVFAIL + return reply + + def resolve(self, request, handler): + if request.q.qname in self.reximemo_names: + return self._reximemo_reply(request) + return self._forward(request, handler) + + +def main(): + _print_banner() + print("[INFO] Starting reximemoDNS...") + print("[INFO] Detected operating system:", get_platform()) + + current_platform = get_platform() + if current_platform in {"Linux", "macOS"}: + print("[INFO] DNS uses UDP/TCP port 53, so this normally must run as root.") + print("[INFO] If no requests appear, also check whether another service is already using port 53.") + elif current_platform == "Windows": + print("[INFO] Windows may ask you to allow reximemoDNS through the firewall.") + + resolver = Resolver() + logger = RexiMemoDNSLogger() + + try: + servers = [ + DNSServer(resolver=resolver, port=DNS_PORT, address=MY_IP, tcp=True, logger=logger), + DNSServer(resolver=resolver, port=DNS_PORT, address=MY_IP, tcp=False, logger=logger), + ] + except PermissionError: + print("[ERROR] Permission denied while opening DNS port. Run as Administrator/root.") + raise SystemExit(1) + except OSError as exc: + print(f"[ERROR] Could not bind DNS server to {MY_IP}:{DNS_PORT}: {exc}") + raise SystemExit(1) + + print(f"[INFO] reximemoDNS is ready on {MY_IP}:{DNS_PORT} (UDP/TCP).") + print("[INFO] Waiting for DNS requests from your console...") + + for server in servers: + server.start_thread() + + try: + while True: + sleep(0.1) + except KeyboardInterrupt: + print("\n[INFO] Stopping reximemoDNS...") + finally: + for server in servers: + server.stop() + + +if __name__ == "__main__": + main() diff --git a/sudomemoDNS.py b/sudomemoDNS.py deleted file mode 100644 index 8522546..0000000 --- a/sudomemoDNS.py +++ /dev/null @@ -1,259 +0,0 @@ -# sudomemoDNS - -from datetime import datetime -from json import loads -from socket import socket, AF_INET, SOCK_DGRAM, gethostbyname -from sys import platform -from time import sleep - -from dnslib import A, AAAA, CNAME, MX, NS, SOA, TXT -from dnslib import DNSLabel, QTYPE, RD, RR -from dnslib.server import DNSServer -from requests import get -from requests.exceptions import RequestException -from requests.exceptions import Timeout - - -def get_platform(): - platforms = { - 'linux1': 'Linux', - 'linux2': 'Linux', - 'darwin': 'macOS', - 'win32': 'Windows' - } - if platform not in platforms: - return platform - - return platforms[platform] - - -SUDOMEMODNS_VERSION = "1.2.1" - - -# 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]):03d}.{int(octets[1]):03d}.{int(octets[2]):03d}.{int(octets[3]):03d}" - - -def get_ip(): - s = socket(AF_INET, SOCK_DGRAM) - try: - # doesn't even have to be reachable - s.connect(('10.255.255.255', 1)) - IP = s.getsockname()[0] - except: - IP = '127.0.0.1' - finally: - s.close() - return IP - - -EPOCH = datetime(1970, 1, 1) -SERIAL = int((datetime.utcnow() - EPOCH).total_seconds()) -MY_IP = get_ip() - -print("+===============================+") -print("| Sudomemo DNS Server |") -print(f"| Version {SUDOMEMODNS_VERSION} |") -print("+===============================+\n") - -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("== How To Use ==") -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("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~") -print(f"Primary DNS: {format_ip(MY_IP)}") -print("Secondary DNS: 008.008.008.008") -print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~\n") - -print("== Getting Help ==") -print("Need help? Visit our Discord server or check out https://support.sudomemo.net\n") - -print("[INFO] Starting sudomemoDNS...") - -TYPE_LOOKUP = { - A: QTYPE.A, - AAAA: QTYPE.AAAA, - CNAME: QTYPE.CNAME, - MX: QTYPE.MX, - NS: QTYPE.NS, - SOA: QTYPE.SOA, - 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 - -class SudomemoDNSLogger(object): - def log_recv(self, handler, data): - pass - - def log_send(self, handler, data): - pass - - def log_request(self, handler, request): - print("[INFO] Received DNS request from console at " + handler.client_address[0]) - - def log_reply(self, handler, reply): - print("[INFO] Sent response to console at " + handler.client_address[0]) - - def log_error(self, handler, e): - print("[ERROR] Invalid DNS request from " + handler.client_address[0]) - - def log_truncated(self, handler, reply): - pass - - def log_data(self, dnsobj): - pass - - -class Record: - def __init__(self, rdata_type, *args, rtype=None, rname=None, ttl=None, **kwargs): - if isinstance(rdata_type, RD): - # actually an instance, not a type - self._rtype = TYPE_LOOKUP[rdata_type.__class__] - rdata = rdata_type - else: - self._rtype = TYPE_LOOKUP[rdata_type] - if rdata_type == SOA and len(args) == 2: - # add sensible times to SOA - args += (( - SERIAL, # serial number - 60 * 60 * 1, # refresh - 60 * 60 * 3, # retry - 60 * 60 * 24, # expire - 60 * 60 * 1, # minimum - ),) - rdata = rdata_type(*args) - - if rtype: - self._rtype = rtype - self._rname = rname - self.kwargs = dict( - rdata=rdata, - ttl=self.sensible_ttl() if ttl is None else ttl, - **kwargs, - ) - - def try_rr(self, q): - if q.qtype == QTYPE.ANY or q.qtype == self._rtype: - return self.as_rr(q.qname) - - def as_rr(self, alt_rname): - return RR(rname=self._rname or alt_rname, rtype=self._rtype, **self.kwargs) - - def sensible_ttl(self): - if self._rtype in (QTYPE.NS, QTYPE.SOA): - return 60 * 60 * 24 - else: - return 300 - - @property - def is_soa(self): - return self._rtype == QTYPE.SOA - - def __str__(self): - return '{} {}'.format(QTYPE[self._rtype], self.kwargs) - - -ZONES = {} - -try: - get_zones = get("https://www.sudomemo.net/api/dns_zones.json", - headers={'User-Agent': 'SudomemoDNS/' + SUDOMEMODNS_VERSION + ' (' + get_platform() + ')'}) -except Timeout: - print("[ERROR] Unable to load DNS data: Connection to Sudomemo timed out. Are you connected to the Internet?") -except RequestException as e: - print("[ERROR] Unable load DNS data.") - print("[ERROR] Exception: ", e) - exit(1) -try: - zones = loads(get_zones.text) -except ValueError as e: - print("[ERROR] Unable load DNS data: Invalid response from server. Check that you can visit sudomemo.net") - -for zone in zones: - if zone["type"] == "a": - ZONES[zone["name"]] = [Record(A, zone["value"])] - elif zone["type"] == "p": - ZONES[zone["name"]] = [Record(A, gethostbyname(zone["value"]))] - -print("[INFO] DNS information loaded successfully.") - - -class Resolver: - def __init__(self): - self.zones = {DNSLabel(k): v for k, v in ZONES.items()} - - def resolve(self, request, handler): - reply = request.reply() - zone = self.zones.get(request.q.qname) - - if zone is not None: - for zone_records in zone: - rr = zone_records.try_rr(request.q) - rr and reply.add_answer(rr) - else: - # no direct zone so look for an SOA record for a higher level zone - for zone_label, zone_records in self.zones.items(): - if request.q.qname.matchSuffix(zone_label): - try: - soa_record = next(r for r in zone_records if r.is_soa) - except StopIteration: - continue - else: - reply.add_answer(soa_record.as_rr(zone_label)) - break - - return reply - - -resolver = Resolver() -dnsLogger = SudomemoDNSLogger() - -print("[INFO] Detected operating system:", get_platform()) - -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] 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'") -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] 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'") -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] 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: - servers = [ - DNSServer(resolver=resolver, port=53, address=MY_IP, tcp=True, 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") - exit(1) - -print("[INFO] sudomemoDNS is ready. Now waiting for DNS requests from your console...") - -if __name__ == '__main__': - for s in servers: - s.start_thread() - - try: - while 1: - sleep(0.1) - except KeyboardInterrupt: - pass - finally: - for s in servers: - s.stop()