From 25f0f6869347ef4518279667dab712e00a0d8175 Mon Sep 17 00:00:00 2001 From: Alessandro Gatti Date: Thu, 17 Sep 2026 20:50:57 +0200 Subject: [PATCH] udnspkt: Make the module work again. This commit updates the `udnspkt` module to make it work with current MicroPython versions. The code depended on two functions called `readbin` and `writebin` that operate on byte buffers, which are not present anymore in neither micropython-lib or micropython repositories. Those calls have been replaced with equivalent code that uses the `struct` module instead, or hardcoding a byte buffer with the static data being encoded. The module API was tweaked slightly, in which `udnspkt.parse_resp` now raises an exception if the DNS server on the other end refuses the request. The exception's payload is the raw response flags value from the server, in integer form. The sample code was also updated to make it work with CPython, to make sure the behaviour is consistent across interpreters too. And finally, the module is now 33 bytes shorter when compiled. Signed-off-by: Alessandro Gatti --- micropython/udnspkt/example_resolve.py | 20 +++--- micropython/udnspkt/manifest.py | 2 +- micropython/udnspkt/udnspkt.py | 85 +++++++++++--------------- 3 files changed, 44 insertions(+), 63 deletions(-) diff --git a/micropython/udnspkt/example_resolve.py b/micropython/udnspkt/example_resolve.py index d72c17a48..b52752b58 100644 --- a/micropython/udnspkt/example_resolve.py +++ b/micropython/udnspkt/example_resolve.py @@ -9,19 +9,17 @@ def resolve(domain, is_ipv6): - buf = io.BytesIO(48) - udnspkt.make_req(buf, "google.com", is_ipv6) - v = buf.getvalue() - print("query: ", v) - s.sendto(v, dns_addr) - + with io.BytesIO() as buf: + udnspkt.make_req(buf, domain, is_ipv6) + v = buf.getvalue() + print("query: ", v) + s.sendto(v, dns_addr) resp = s.recv(1024) print("resp:", resp) - buf = io.BytesIO(resp) - - addr = udnspkt.parse_resp(buf, is_ipv6) - print("bin addr:", addr) - print("addr:", socket.inet_ntop(socket.AF_INET6 if is_ipv6 else socket.AF_INET, addr)) + with io.BytesIO(resp) as buf: + addr = udnspkt.parse_resp(buf, is_ipv6) + print("bin addr:", addr) + print("addr:", socket.inet_ntop(socket.AF_INET6 if is_ipv6 else socket.AF_INET, addr)) resolve("google.com", False) diff --git a/micropython/udnspkt/manifest.py b/micropython/udnspkt/manifest.py index 2c2a78d2b..0b527ec27 100644 --- a/micropython/udnspkt/manifest.py +++ b/micropython/udnspkt/manifest.py @@ -1,4 +1,4 @@ -metadata(description="Make and parse DNS packets (Sans I/O approach).", version="0.1.0") +metadata(description="Make and parse DNS packets (Sans I/O approach).", version="0.2.0") # Originally written by Paul Sokolovsky. diff --git a/micropython/udnspkt/udnspkt.py b/micropython/udnspkt/udnspkt.py index f3b998a8a..b70b13c9f 100644 --- a/micropython/udnspkt/udnspkt.py +++ b/micropython/udnspkt/udnspkt.py @@ -1,67 +1,50 @@ +import struct + + def write_fqdn(buf, name): - parts = name.split(".") - for p in parts: - buf.writebin("B", len(p)) - buf.write(p) - buf.writebin("B", 0) + data = bytearray() + for part in name.split("."): + data.append(len(part)) + data.extend(part.encode("ascii")) + data.append(0) + buf.write(data) def skip_fqdn(buf): while True: - sz = buf.readbin("B") - if not sz: - break - if sz >= 0xC0: - buf.readbin("B") - break - buf.read(sz) + # Label size + size = (buf.read(1) or b"\x00")[0] + if not size: + # Truncated packet or last label + return + # Ignore compressed response pointer + if size >= 0xC0: + buf.read(1) + return + # Skip label + buf.read(size) def make_req(buf, fqdn, is_ipv6): - typ = 1 # A - if is_ipv6: - typ = 28 # AAAA - - buf.writebin(">H", 0) - buf.writebin(">H", 0x100) - # q count - buf.writebin(">H", 1) - buf.writebin(">H", 0) - # squashed together - buf.writebin(">I", 0) - + # As per RFC1035, §4.1.1 + # ID 0, Standard query, just one question entry + buf.write(b"\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00") + # §4.1.3 write_fqdn(buf, fqdn) - buf.writebin(">H", typ) - buf.writebin(">H", 1) # Class + buf.write(b"\x00\x1c\x00\x01" if is_ipv6 else b"\x00\x01\x00\x01") def parse_resp(buf, is_ipv6): - typ = 1 # A - if is_ipv6: - typ = 28 # AAAA - - buf.readbin(">H") # id - flags = buf.readbin(">H") - assert flags & 0x8000 - buf.readbin(">H") # qcnt - acnt = buf.readbin(">H") - buf.readbin(">H") # nscnt - buf.readbin(">H") # addcnt - + # As per RFC1035, §4.1.1 + _, flags, _, acnt, _, _ = struct.unpack(">6H", buf.read(6 * 2)) + # Bit 15 indicates it's a response, and bits 3..0 is the response code. + if flags & 0x800F != 0x8000: + raise ValueError(flags) skip_fqdn(buf) - buf.readbin(">H") - buf.readbin(">H") - - for i in range(acnt): - # print("Resp #%d" % i) - # v = read_fqdn(buf) - # print(v) + buf.read(4) + for _ in range(acnt): skip_fqdn(buf) - t = buf.readbin(">H") # Type - buf.readbin(">H") # Class - buf.readbin(">I") # TTL - rlen = buf.readbin(">H") + t, _, _, rlen = struct.unpack(">HHIH", buf.read(4 + (3 * 2))) rval = buf.read(rlen) - - if t == typ: + if t == (0x1C if is_ipv6 else 0x01): return rval