Source code for asyncdnspy.tcp_client

#!/usr/bin/env python

import socket

[docs]class TCPClient(object): """TCP client for connecting to a DNS server over TCP.""" def __init__(self, address: str, port: int, blocking: bool = False, timeout: int = 10): """ Inits TCPClient instance. >>> from asyncdnspy.tcp_client import TCPClient >>> tcp_client = TCPClient('8.8.8.8', 53) or >>> tcp_client = TCPClient('8.8.8.8', 53, True) :param address: DNS server ip address. :type address: str :param port: Port of DNS server (53). :type port: int :param blocking: Make socket blocking or nonblocking. :type blocking: bool :param timeout: Timeout if socket is nonblocking. :type timeout: int :rtype: TCPClient instance. """ self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.client_socket.setblocking(blocking) if not blocking: self.client_socket.settimeout(timeout) self.address = address self.port = port
[docs] def connect(self): """ Connects to given address and port. >>> from asyncdnspy.tcp_client import TCPClient >>> tcp_client = TCPClient('8.8.8.8', 53) >>> tcp_client.connect() """ self.client_socket.connect((self.address, self.port))
[docs] def send(self, data: bytes, flags: int = 0): """ Send bytes over UDP. >>> from asyncdnspy.tcp_client import TCPClient >>> from asyncdnspy.dns_raw_message import DNSRawMessage >>> from asyncdnspy.dnspy_enum import RecordType >>> tcp_client = TCPClient('8.8.8.8', 53) >>> tcp_client.connect() >>> dns_raw_message = DNSRawMessage() >>> data = dns_raw_message.query('google.com', RecordType.a, socket_type=SocketType.tcp) >>> result = tcp_client.send(data) :param data: Bytes created by DNSRawMessage. :type data: bytes :param flags: Flags to send data. :type flags: int :rtype: int """ return self.client_socket.send(data, flags)
[docs] def receive(self, bufsize: int = 1024, flags: int = 0): """ Receives result of send data. >>> from asyncdnspy.tcp_client import TCPClient >>> from asyncdnspy.dns_raw_message import DNSRawMessage >>> from asyncdnspy.dnspy_enum import RecordType >>> tcp_client = TCPClient('8.8.8.8', 53) >>> tcp_client.connect() >>> dns_raw_message = DNSRawMessage() >>> data = dns_raw_message.query('google.com', RecordType.a, socket_type=SocketType.tcp) >>> result = tcp_client.send(data) >>> response = tcp_client.receive() :param bufsize: Buffer size. :type bufsize: int :param flags: Flags to send data. :type flags: int :rtype: bytes """ return self.client_socket.recv(bufsize, flags)
[docs] def close(self): """ Closes the socket connection. >>> from asyncdnspy.tcp_client import TCPClient >>> tcp_client = TCPClient('8.8.8.8', 53) >>> tcp_client.connect() >>> tcp_client.close() """ self.client_socket.close()