Method unpack_message

mitmproxy/proxy/layers/dns.py:112–141

(self, data: bytes, from_client: bool)

        )

    def unpack_message(self, data: bytes, from_client: bool) -> List[dns.DNSMessage]:
        msgs: List[dns.DNSMessage] = []

        buf = self.req_buf if from_client else self.resp_buf

        if self.context.client.transport_protocol == "udp":
            msgs.append(dns.DNSMessage.unpack(data, timestamp=time.time()))
        elif self.context.client.transport_protocol == "tcp":
            buf.extend(data)
            size = len(buf)
            offset = 0

            while True:
                if size - offset < _LENGTH_LABEL.size:
                    break
                (expected_size,) = _LENGTH_LABEL.unpack_from(buf, offset)
                offset += _LENGTH_LABEL.size
                if expected_size == 0:
                    raise struct.error("Message length field cannot be zero")

                if size - offset < expected_size:
                    offset -= _LENGTH_LABEL.size
                    break

                data = bytes(buf[offset : expected_size + offset])
                offset += expected_size
                msgs.append(dns.DNSMessage.unpack(data, timestamp=time.time()))

            del buf[:offset]
        return msgs

    @expect(events.Start)
    def state_start(self, _) -> layer.CommandGenerator[None]: