client (version 0.0.1)
index
client.py

This package implements a basic asynchronous DNS client and server
with a feature to exfiltrate data through DNS.

 
Classes
       
asyncio.protocols.DatagramProtocol(asyncio.protocols.BaseProtocol)
DNSClientProtocol

 
class DNSClientProtocol(asyncio.protocols.DatagramProtocol)
    DNSClientProtocol(query_packet: bytes, response_future: _asyncio.Future)
 
Asyncio DatagramProtocol implementation for sending a DNS query and receiving a response.
Used internally by the `send_dns_udp_query` function.
 
 
Method resolution order:
DNSClientProtocol
asyncio.protocols.DatagramProtocol
asyncio.protocols.BaseProtocol
builtins.object

Methods defined here:
__init__(self, query_packet: bytes, response_future: _asyncio.Future)
Initialize the DNS client protocol.
 
Args:
    query_packet (bytes): The DNS query packet to send.
    response_future (Future): Future object used to deliver the response back to the caller.
connection_lost(self, exc)
Called when the connection is lost or closed.
 
Args:
    exc (Optional[Exception]): The reason for the closure, if any.
 
Ensures any pending exception is raised if the connection closes unexpectedly.
connection_made(self, transport)
Called when the connection is made.
 
Sends the DNS query packet immediately after the connection is established.
datagram_received(self, data: bytes, addr: Tuple[str, int])
Called when a datagram is received.
 
Args:
    data (bytes): The received DNS response.
    addr (Tuple[str, int]): The address of the sender.
 
Sets the response data in the future and closes the transport.
error_received(self, exc: Exception)
Called when a send or receive error occurs.
 
Args:
    exc (Exception): The error that occurred.
 
Sets the exception in the future and closes the transport.

Data descriptors defined here:
__dict__
dictionary for instance variables (if defined)
__weakref__
list of weak references to the object (if defined)

Methods inherited from asyncio.protocols.BaseProtocol:
pause_writing(self)
Called when the transport's buffer goes over the high-water mark.
 
Pause and resume calls are paired -- pause_writing() is called
once when the buffer goes strictly over the high-water mark
(even if subsequent writes increases the buffer size even
more), and eventually resume_writing() is called once when the
buffer size reaches the low-water mark.
 
Note that if the buffer size equals the high-water mark,
pause_writing() is not called -- it must go strictly over.
Conversely, resume_writing() is called when the buffer size is
equal or lower than the low-water mark.  These end conditions
are important to ensure that things go as expected when either
mark is zero.
 
NOTE: This is the only Protocol callback that is not called
through EventLoop.call_soon() -- if it were, it would have no
effect when it's most needed (when the app keeps writing
without yielding until pause_writing() is called).
resume_writing(self)
Called when the transport's buffer drains below the low-water mark.
 
See pause_writing() for details.

 
Functions
       
exit(status=None, /)
Exit the interpreter by raising SystemExit(status).
 
If the status is omitted or None, it defaults to zero (i.e., success).
If the status is an integer, it will be used as the system exit status.
If it is another kind of object, it will be printed and the system
exit status will be one (i.e., failure).
get_event_loop()
Return an asyncio event loop.
 
When called from a coroutine or a callback (e.g. scheduled with
call_soon or similar API), this function will always return the
running event loop.
 
If there is no running event loop set, the function will return
the result of `get_event_loop_policy().get_event_loop()` call.
main() -> int
Main function to parse arguments and run DNS resolution.
 
Returns:
    Exit code: 0 on success, 1 on usage error, 2 on DNS resolution failure.
pack(...)
pack(format, v1, v2, ...) -> bytes
 
Return a bytes object containing the values v1, v2, ... packed according
to the format string.  See help(struct) for more on format strings.
parse_dns_response(response_data: bytes) -> List[str]
Parse DNS response to extract IPv4 addresses from A records.
 
Args:
    response_data: Raw DNS response bytes.
 
Returns:
    List of IPv4 addresses as strings.
async resolve_all(domain_names: List[str], **kwargs) -> List[Tuple[str, List[str]]]
Resolve all domain names concurrently.
 
Args:
    domain_names: List of domain names.
 
Returns:
    List of tuples containing domain and result.
async resolve_domain(domain_name: str, dns_server_ip: str = '8.8.8.8', dns_server_port: int = 53) -> Tuple[str, List[str]]
Resolve a single domain asynchronously.
 
Args:
    domain_name: The domain name to resolve.
    dns_server_ip: DNS server IP address.
 
Returns:
    Tuple of domain name and list of IPs or error message.
async send_dns_tcp_query(domain_name: str, dns_server_ip: str = '8.8.8.8', dns_server_port: int = 53) -> List[str]
Send a DNS query using TCP.
 
Args:
    domain_name: The domain name to resolve.
    dns_server_ip: The DNS server IP address.
 
Returns:
    List of IPv4 addresses.
 
Raises:
    TimeoutError if no response is received.
async send_dns_udp_query(domain_name: str, dns_server_ip: str = '8.8.8.8', dns_server_port: int = 53, **kwargs) -> List[str]
Send a DNS query using fully asynchronous UDP sockets.
 
Args:
    domain_name (str): The domain name to resolve.
    dns_server_ip (str): The IP address of the DNS server.
    dns_server_port (int): The port number of the DNS server (default: 53).
 
Returns:
    List[str]: A list of resolved IPv4 addresses.
 
Raises:
    asyncio.TimeoutError: If no response is received within 2 seconds.
    Exception: For socket or protocol errors.
unpack(format, buffer, /)
Return a tuple containing values unpacked according to the format string.
 
The buffer's size in bytes must be calcsize(format).
 
See help(struct) for more on format strings.

 
Data
        AF_INET = <AddressFamily.AF_INET: 2>
List = typing.List
SOCK_DGRAM = <SocketKind.SOCK_DGRAM: 2>
SOCK_STREAM = <SocketKind.SOCK_STREAM: 1>
Tuple = typing.Tuple
__author_email__ = 'mauricelambert434@gmail.com'
__copyright__ = '\nPyAsyncDNS Copyright (C) 2025 Maurice Lambert...ome to redistribute it\nunder certain conditions.\n'
__description__ = '\nThis package implements a basic asynchronous DN...r\nwith a feature to exfiltrate data through DNS.\n'
__license__ = 'GPL-3.0 License'
__maintainer__ = 'Maurice Lambert'
__maintainer_email__ = 'mauricelambert434@gmail.com'
__url__ = 'https://github.com/mauricelambert/PyAsyncDNS'
argv = [r'C:\Program Files\Python310\lib\pydoc.py', '-w', r'client.py']
copyright = '\nPyAsyncDNS Copyright (C) 2025 Maurice Lambert...ome to redistribute it\nunder certain conditions.\n'
executable = r'C:\Program Files\Python310\python.exe'
license = 'GPL-3.0 License'
stderr = <_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>

 
Author
        Maurice Lambert