Function receive_tcp

dns/query.py:1146–1190

Read a DNS message from a TCP socket. *sock*, a ``socket``. *expiration*, a ``float`` or ``None``, the absolute time at which a timeout exception should be raised. If ``None``, no timeout will occur. *one_rr_per_rrset*, a ``bool``. If ``True``, put each RR into its own R

(
    sock: Any,
    expiration: float | None = None,
    one_rr_per_rrset: bool = False,
    keyring: Dict[dns.name.Name, dns.tsig.Key] | None = None,
    request_mac: bytes | None = b"",
    ignore_trailing: bool = False,
)

def receive_tcp(
    sock: Any,
    expiration: float | None = None,
    one_rr_per_rrset: bool = False,
    keyring: Dict[dns.name.Name, dns.tsig.Key] | None = None,
    request_mac: bytes | None = b"",
    ignore_trailing: bool = False,
) -> Tuple[dns.message.Message, float]:
    """Read a DNS message from a TCP socket.

    *sock*, a ``socket``.

    *expiration*, a ``float`` or ``None``, the absolute time at which
    a timeout exception should be raised.  If ``None``, no timeout will
    occur.

    *one_rr_per_rrset*, a ``bool``.  If ``True``, put each RR into its own
    RRset.

    *keyring*, a ``dict``, the keyring to use for TSIG.

    *request_mac*, a ``bytes`` or ``None``, the MAC of the request (for TSIG).

    *ignore_trailing*, a ``bool``.  If ``True``, ignore trailing
    junk at end of the received message.

    Raises if the message is malformed, if network errors occur, of if
    there is a timeout.

    Returns a ``(dns.message.Message, float)`` tuple of the received message
    and the received time.
    """

    ldata = _net_read(sock, 2, expiration)
    (l,) = struct.unpack("!H", ldata)
    wire = _net_read(sock, l, expiration)
    received_time = time.time()
    r = dns.message.from_wire(
        wire,
        keyring=keyring,
        request_mac=request_mac,
        one_rr_per_rrset=one_rr_per_rrset,
        ignore_trailing=ignore_trailing,
    )
    return (r, received_time)


def _connect(s, address, expiration):