"""Formato de los mensajes de DNS (RFC 1035, sección 4), byte por byte y sin bibliotecas.""" import struct from dataclasses import dataclass, field TYPES = {"A": 1, "NS": 2, "CNAME": 5, "SOA": 6, "MX": 15, "TXT": 16, "AAAA": 28} TYPE_NAMES = {number: name for name, number in TYPES.items()} CLASS_IN = 1 # Bits del segundo campo del encabezado. FLAG_QR = 0x8000 # 1 si es respuesta FLAG_AA = 0x0400 # respuesta autoritativa FLAG_TC = 0x0200 # truncada: pregunte por TCP FLAG_RD = 0x0100 # el cliente pide recursión FLAG_RA = 0x0080 # el servidor ofrece recursión NOERROR, FORMERR, SERVFAIL, NXDOMAIN, NOTIMP, REFUSED = range(6) RCODE_NAMES = ["NOERROR", "FORMERR", "SERVFAIL", "NXDOMAIN", "NOTIMP", "REFUSED"] MAX_POINTER_JUMPS = 32 class WireError(ValueError): """El mensaje no respeta el formato. Nunca confiamos en lo que llega de la red.""" @dataclass class Record: name: str rtype: int ttl: int data: object # depende del tipo; ver encode_rdata def __str__(self): return f"{self.name} {self.ttl} IN {TYPE_NAMES.get(self.rtype, self.rtype)} {self.data}" @dataclass class Message: msg_id: int = 0 flags: int = 0 question: list = field(default_factory=list) # lista de (nombre, tipo) answer: list = field(default_factory=list) authority: list = field(default_factory=list) additional: list = field(default_factory=list) @property def rcode(self): return self.flags & 0x000F def normalize(name): """Los nombres no distinguen mayúsculas; los guardamos en minúsculas y con punto final.""" name = name.lower() return name if name.endswith(".") else name + "." def encode_name(name, buffer, offsets): """Agrega un nombre a buffer. Si ya apareció un sufijo, usa un puntero a él.""" labels = [label for label in name.split(".") if label] for index, label in enumerate(labels): suffix = ".".join(labels[index:]).lower() if suffix in offsets: buffer += struct.pack("!H", 0xC000 | offsets[suffix]) return if len(buffer) < 0x4000: offsets[suffix] = len(buffer) raw = label.encode("ascii") if not 1 <= len(raw) <= 63: raise WireError(f"etiqueta de largo inválido: {label!r}") buffer.append(len(raw)) buffer += raw buffer.append(0) def decode_name(data, offset): """Lee un nombre, siguiendo punteros. Devuelve el nombre y dónde sigue el mensaje.""" labels = [] end = None jumps = 0 while True: if offset >= len(data): raise WireError("nombre cortado") size = data[offset] if size & 0xC0 == 0xC0: if offset + 1 >= len(data): raise WireError("puntero cortado") if end is None: end = offset + 2 target = ((size & 0x3F) << 8) | data[offset + 1] # Un puntero solo puede ir hacia atrás. Eso, y el límite, evitan los ciclos. jumps += 1 if target >= offset or jumps > MAX_POINTER_JUMPS: raise WireError("puntero inválido") offset = target elif size & 0xC0: raise WireError("etiqueta de tipo desconocido") elif size == 0: if end is None: end = offset + 1 return (".".join(labels) + ".") if labels else ".", end else: if offset + 1 + size > len(data): raise WireError("etiqueta cortada") labels.append(data[offset + 1:offset + 1 + size].decode("ascii", "replace")) offset += 1 + size def encode_ipv4(text): parts = [int(part) for part in text.split(".")] if len(parts) != 4 or not all(0 <= part <= 255 for part in parts): raise WireError(f"dirección inválida: {text}") return bytes(parts) def encode_rdata(record, buffer, offsets): """Agrega los datos del registro. Deja el RDLENGTH correcto antes de ellos.""" length_at = len(buffer) buffer += b"\0\0" kind = record.rtype if kind == TYPES["A"]: buffer += encode_ipv4(record.data) elif kind == TYPES["AAAA"]: import ipaddress buffer += ipaddress.IPv6Address(record.data).packed elif kind in (TYPES["NS"], TYPES["CNAME"]): encode_name(record.data, buffer, offsets) elif kind == TYPES["MX"]: preference, host = record.data buffer += struct.pack("!H", preference) encode_name(host, buffer, offsets) elif kind == TYPES["SOA"]: mname, rname, serial, refresh, retry, expire, minimum = record.data encode_name(mname, buffer, offsets) encode_name(rname, buffer, offsets) buffer += struct.pack("!5I", serial, refresh, retry, expire, minimum) elif kind == TYPES["TXT"]: # Un TXT es una o más cadenas de hasta 255 bytes, cada una con su largo. raw = record.data.encode("utf-8") chunks = [raw[start:start + 255] for start in range(0, len(raw), 255)] or [b""] for chunk in chunks: buffer.append(len(chunk)) buffer += chunk else: raise WireError(f"tipo no soportado: {kind}") struct.pack_into("!H", buffer, length_at, len(buffer) - length_at - 2) def encode(message): """Convierte un mensaje a bytes, con todos los campos en orden de red.""" buffer = bytearray(struct.pack( "!6H", message.msg_id, message.flags, len(message.question), len(message.answer), len(message.authority), len(message.additional))) offsets = {} for name, rtype in message.question: encode_name(name, buffer, offsets) buffer += struct.pack("!2H", rtype, CLASS_IN) for record in message.answer + message.authority + message.additional: encode_name(record.name, buffer, offsets) buffer += struct.pack("!HHI", record.rtype, CLASS_IN, record.ttl) encode_rdata(record, buffer, offsets) return bytes(buffer) def decode_rdata(data, offset, length, rtype): end = offset + length if rtype == TYPES["A"] and length == 4: return ".".join(str(byte) for byte in data[offset:end]) if rtype == TYPES["AAAA"] and length == 16: import ipaddress return str(ipaddress.IPv6Address(data[offset:end])) if rtype in (TYPES["NS"], TYPES["CNAME"]): return decode_name(data, offset)[0] if rtype == TYPES["MX"] and length >= 3: return struct.unpack_from("!H", data, offset)[0], decode_name(data, offset + 2)[0] if rtype == TYPES["SOA"]: mname, cursor = decode_name(data, offset) rname, cursor = decode_name(data, cursor) if cursor + 20 > end: raise WireError("SOA cortado") return (mname, rname, *struct.unpack_from("!5I", data, cursor)) if rtype == TYPES["TXT"]: chunks, cursor = [], offset while cursor < end: size = data[cursor] chunks.append(data[cursor + 1:cursor + 1 + size]) cursor += 1 + size return b"".join(chunks).decode("utf-8", "replace") # Tipos que no entendemos (por ejemplo OPT, de EDNS): se conservan como bytes. return bytes(data[offset:end]) def decode_record(data, offset): name, offset = decode_name(data, offset) if offset + 10 > len(data): raise WireError("registro cortado") rtype, _rclass, ttl, length = struct.unpack_from("!HHIH", data, offset) offset += 10 if offset + length > len(data): raise WireError("datos del registro cortados") return Record(normalize(name), rtype, ttl, decode_rdata(data, offset, length, rtype)), offset + length def decode(data): """Convierte bytes en un mensaje. Lanza WireError si algo no cuadra.""" if len(data) < 12: raise WireError("mensaje más corto que el encabezado") msg_id, flags, qd, an, ns, ar = struct.unpack_from("!6H", data, 0) message = Message(msg_id, flags) offset = 12 for _ in range(qd): name, offset = decode_name(data, offset) if offset + 4 > len(data): raise WireError("pregunta cortada") rtype, _rclass = struct.unpack_from("!2H", data, offset) offset += 4 message.question.append((name, rtype)) for section, count in ((message.answer, an), (message.authority, ns), (message.additional, ar)): for _ in range(count): record, offset = decode_record(data, offset) section.append(record) return message def in_bailiwick(name, zone): """¿El nombre cae dentro de la zona? Quien responde solo tiene autoridad sobre su zona.""" name, zone = normalize(name), normalize(zone) return zone == "." or name == zone or name.endswith("." + zone)