Add reximemoDNS.py file

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()

sudomemoDNS.py

@@ -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()