tessl/pypi-twisted
An asynchronous networking framework written in Python
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
dns.mddocs/
DNS Implementation
DNS protocol implementation with client resolvers, server functionality, and caching capabilities for domain name resolution. Twisted Names provides comprehensive DNS support for both client and server applications.
Capabilities
DNS Client Resolution
DNS client functionality for resolving domain names to IP addresses and other record types.
class client.Resolver:
"""
DNS resolver for performing domain name lookups.
Attributes:
- servers: List of DNS server addresses
- timeout: Query timeout in seconds
- tries: Number of retry attempts
"""
servers = None
timeout = 10
tries = 3
def __init__(self, resolv=None, servers=None):
"""
Args:
resolv (str): Path to resolv.conf file
servers (list): List of DNS server addresses
"""
def lookupAddress(self, name, timeout=None):
"""
Look up A records for hostname.
Args:
name (str): Hostname to resolve
timeout (int): Query timeout override
Returns:
Deferred[list]: List of A record objects
"""
def lookupIPV6Address(self, name, timeout=None):
"""
Look up AAAA records for hostname.
Args:
name (str): Hostname to resolve
timeout (int): Query timeout override
Returns:
Deferred[list]: List of AAAA record objects
"""
def lookupMailExchange(self, name, timeout=None):
"""
Look up MX records for domain.
Args:
name (str): Domain name
timeout (int): Query timeout override
Returns:
Deferred[list]: List of MX record objects
"""
def lookupNameservers(self, name, timeout=None):
"""
Look up NS records for domain.
Args:
name (str): Domain name
timeout (int): Query timeout override
Returns:
Deferred[list]: List of NS record objects
"""
def lookupCanonicalName(self, name, timeout=None):
"""
Look up CNAME records for hostname.
Args:
name (str): Hostname
timeout (int): Query timeout override
Returns:
Deferred[list]: List of CNAME record objects
"""
def lookupPointer(self, name, timeout=None):
"""
Look up PTR records for reverse DNS.
Args:
name (str): IP address or reverse domain
timeout (int): Query timeout override
Returns:
Deferred[list]: List of PTR record objects
"""
def lookupText(self, name, timeout=None):
"""
Look up TXT records for domain.
Args:
name (str): Domain name
timeout (int): Query timeout override
Returns:
Deferred[list]: List of TXT record objects
"""
def lookupService(self, name, timeout=None):
"""
Look up SRV records for service.
Args:
name (str): Service name (_service._proto.domain)
timeout (int): Query timeout override
Returns:
Deferred[list]: List of SRV record objects
"""
def queryUDP(self, queries, timeout=None):
"""
Perform DNS query over UDP.
Args:
queries (list): List of Query objects
timeout (int): Query timeout override
Returns:
Deferred[Message]: DNS response message
"""
def queryTCP(self, queries, timeout=None):
"""
Perform DNS query over TCP.
Args:
queries (list): List of Query objects
timeout (int): Query timeout override
Returns:
Deferred[Message]: DNS response message
"""
class client.ThreadedResolver:
"""
DNS resolver that uses threads for blocking DNS calls.
"""
def __init__(self, reactor=None, getHostByName=None):
"""
Args:
reactor: Reactor to use
getHostByName: Function for host resolution
"""
def getHostByName(self, name, timeout=None):
"""
Resolve hostname to IP address.
Args:
name (str): Hostname
timeout (int): Resolution timeout
Returns:
Deferred[str]: IP address
"""DNS Client Usage Example:
from twisted.names import client
from twisted.internet import reactor, defer
# Create resolver
resolver = client.Resolver()
@defer.inlineCallbacks
def resolve_domains():
try:
# Look up A records
a_records = yield resolver.lookupAddress('example.com')
for record in a_records[0]:
print(f"A: {record.payload.dottedQuad()}")
# Look up MX records
mx_records = yield resolver.lookupMailExchange('example.com')
for record in mx_records[0]:
print(f"MX: {record.payload.exchange} (priority {record.payload.preference})")
# Look up TXT records
txt_records = yield resolver.lookupText('example.com')
for record in txt_records[0]:
print(f"TXT: {record.payload.data}")
# Reverse DNS lookup
ptr_records = yield resolver.lookupPointer('8.8.8.8.in-addr.arpa')
for record in ptr_records[0]:
print(f"PTR: {record.payload.name}")
except Exception as e:
print(f"DNS lookup failed: {e}")
reactor.stop()
resolve_domains()
reactor.run()DNS Protocol and Messages
Core DNS protocol implementation and message parsing.
class dns.DNSDatagramProtocol:
"""
DNS protocol over UDP.
Attributes:
- controller: DNS controller object
"""
controller = None
def __init__(self, controller):
"""
Args:
controller: DNS controller
"""
def query(self, address, queries, timeout=10, id=None):
"""
Send DNS query.
Args:
address: Server address tuple (host, port)
queries (list): List of Query objects
timeout (int): Query timeout
id (int): Query ID
Returns:
Deferred[Message]: DNS response
"""
class dns.DNSProtocol:
"""
DNS protocol over TCP.
"""
def __init__(self, controller):
"""
Args:
controller: DNS controller
"""
def query(self, queries, timeout=60):
"""
Send DNS query over TCP.
Args:
queries (list): List of Query objects
timeout (int): Query timeout
Returns:
Deferred[Message]: DNS response
"""
class dns.Message:
"""
DNS message representation.
Attributes:
- id: Message ID
- answer: Answer flag
- opCode: Operation code
- auth: Authoritative answer flag
- trunc: Truncated flag
- recDes: Recursion desired flag
- recAv: Recursion available flag
- rCode: Response code
- queries: List of queries
- answers: List of answer records
- authority: List of authority records
- additional: List of additional records
"""
id = 0
answer = 0
opCode = 0
auth = 0
trunc = 0
recDes = 0
recAv = 0
rCode = 0
queries = None
answers = None
authority = None
additional = None
def __init__(self, id=0, answer=0, opCode=0, recDes=0, recAv=0, auth=0, rCode=0, trunc=0, maxSize=512):
"""
Args:
id (int): Message ID
answer (int): Answer flag
opCode (int): Operation code
recDes (int): Recursion desired
recAv (int): Recursion available
auth (int): Authoritative answer
rCode (int): Response code
trunc (int): Truncated flag
maxSize (int): Maximum message size
"""
def addQuery(self, name, type=A, cls=IN):
"""
Add query to message.
Args:
name (str): Query name
type (int): Record type
cls (int): Record class
"""
def encode(self, strio):
"""
Encode message to wire format.
Args:
strio: String buffer to write to
"""
def decode(self, strio):
"""
Decode message from wire format.
Args:
strio: String buffer to read from
"""
class dns.Query:
"""
DNS query representation.
Attributes:
- name: Query name
- type: Record type
- cls: Record class
"""
name = None
type = None
cls = None
def __init__(self, name='', type=A, cls=IN):
"""
Args:
name (str): Query name
type (int): Record type (A, AAAA, MX, etc.)
cls (int): Record class (IN, etc.)
"""
class dns.RRHeader:
"""
DNS resource record header.
Attributes:
- name: Record name
- type: Record type
- cls: Record class
- ttl: Time to live
- payload: Record payload
"""
name = None
type = None
cls = None
ttl = None
payload = None
def __init__(self, name='', type=A, cls=IN, ttl=0, payload=None, auth=False):
"""
Args:
name (str): Record name
type (int): Record type
cls (int): Record class
ttl (int): Time to live in seconds
payload: Record data
auth (bool): Authoritative flag
"""DNS Record Types
Implementations of various DNS record types.
class dns.Record_A:
"""
A record (IPv4 address).
Attributes:
- address: IPv4 address (4 bytes)
"""
address = None
def __init__(self, address='0.0.0.0', ttl=None):
"""
Args:
address (str): IPv4 address
ttl (int): Time to live
"""
def dottedQuad(self):
"""
Get dotted quad representation.
Returns:
str: Dotted quad IPv4 address
"""
class dns.Record_AAAA:
"""
AAAA record (IPv6 address).
Attributes:
- address: IPv6 address (16 bytes)
"""
address = None
def __init__(self, address='::', ttl=None):
"""
Args:
address (str): IPv6 address
ttl (int): Time to live
"""
class dns.Record_CNAME:
"""
CNAME record (canonical name).
Attributes:
- name: Canonical name
"""
name = None
def __init__(self, name='', ttl=None):
"""
Args:
name (str): Canonical name
ttl (int): Time to live
"""
class dns.Record_MX:
"""
MX record (mail exchange).
Attributes:
- preference: Priority preference
- exchange: Mail server name
"""
preference = None
exchange = None
def __init__(self, preference=10, exchange='', ttl=None):
"""
Args:
preference (int): Priority (lower = higher priority)
exchange (str): Mail server hostname
ttl (int): Time to live
"""
class dns.Record_NS:
"""
NS record (name server).
Attributes:
- name: Name server hostname
"""
name = None
def __init__(self, name='', ttl=None):
"""
Args:
name (str): Name server hostname
ttl (int): Time to live
"""
class dns.Record_PTR:
"""
PTR record (pointer for reverse DNS).
Attributes:
- name: Pointed-to name
"""
name = None
def __init__(self, name='', ttl=None):
"""
Args:
name (str): Target hostname
ttl (int): Time to live
"""
class dns.Record_SOA:
"""
SOA record (start of authority).
Attributes:
- mname: Primary name server
- rname: Responsible person email
- serial: Serial number
- refresh: Refresh interval
- retry: Retry interval
- expire: Expiration time
- minimum: Minimum TTL
"""
mname = None
rname = None
serial = None
refresh = None
retry = None
expire = None
minimum = None
def __init__(self, mname='', rname='', serial=0, refresh=3600, retry=900, expire=604800, minimum=86400, ttl=None):
"""
Args:
mname (str): Primary name server
rname (str): Responsible person email
serial (int): Serial number
refresh (int): Refresh interval
retry (int): Retry interval
expire (int): Expiration time
minimum (int): Minimum TTL
ttl (int): Record TTL
"""
class dns.Record_SRV:
"""
SRV record (service location).
Attributes:
- priority: Service priority
- weight: Service weight
- port: Service port
- target: Service hostname
"""
priority = None
weight = None
port = None
target = None
def __init__(self, priority=0, weight=0, port=0, target='', ttl=None):
"""
Args:
priority (int): Service priority
weight (int): Service weight
port (int): Service port number
target (str): Service hostname
ttl (int): Time to live
"""
class dns.Record_TXT:
"""
TXT record (text data).
Attributes:
- data: Text data (list of strings)
"""
data = None
def __init__(self, *data, **kw):
"""
Args:
*data: Text strings
**kw: Additional parameters (ttl, etc.)
"""DNS Constants
DNS protocol constants and record type definitions.
# Record types
dns.A = 1 # IPv4 address
dns.NS = 2 # Name server
dns.CNAME = 5 # Canonical name
dns.SOA = 6 # Start of authority
dns.PTR = 12 # Pointer
dns.MX = 15 # Mail exchange
dns.TXT = 16 # Text
dns.AAAA = 28 # IPv6 address
dns.SRV = 33 # Service location
# Record classes
dns.IN = 1 # Internet class
dns.CS = 2 # CSNET class
dns.CH = 3 # CHAOS class
dns.HS = 4 # Hesiod class
# Response codes
dns.OK = 0 # No error
dns.EFORMAT = 1 # Format error
dns.ESERVER = 2 # Server failure
dns.ENAME = 3 # Name error
dns.ENOTIMP = 4 # Not implemented
dns.EREFUSED = 5 # Refused
# Operation codes
dns.OP_QUERY = 0 # Standard query
dns.OP_IQUERY = 1 # Inverse query
dns.OP_STATUS = 2 # Server statusDNS Server
DNS server implementation for authoritative and caching servers.
class server.DNSServerFactory:
"""
DNS server factory.
Attributes:
- authorities: List of authority resolvers
- caches: List of cache resolvers
- clients: List of client resolvers
"""
authorities = None
caches = None
clients = None
def __init__(self, authorities=None, caches=None, clients=None, verbose=0):
"""
Args:
authorities (list): Authority resolvers
caches (list): Cache resolvers
clients (list): Client resolvers
verbose (int): Verbosity level
"""
def buildProtocol(self, addr):
"""
Build DNS server protocol.
Args:
addr: Connection address
Returns:
DNSProtocol: DNS server protocol
"""DNS Caching
DNS caching resolver for improved performance.
class cache.CacheResolver:
"""
DNS caching resolver.
Attributes:
- cache: Cache storage
- reactor: Reactor for timeouts
"""
cache = None
reactor = None
def __init__(self, _cache=None, reactor=None):
"""
Args:
_cache: Cache storage object
reactor: Reactor instance
"""
def query(self, query, timeout=None):
"""
Perform cached DNS query.
Args:
query: DNS query object
timeout (int): Query timeout
Returns:
Deferred[Message]: DNS response (cached or fresh)
"""
def lookupAddress(self, name, timeout=None):
"""
Cached A record lookup.
Args:
name (str): Hostname
timeout (int): Query timeout
Returns:
Deferred[list]: A records
"""DNS Authority
Authoritative DNS server for specific domains.
class authority.FileAuthority:
"""
File-based DNS authority.
Loads DNS records from zone files.
"""
def __init__(self, file):
"""
Args:
file (str): Zone file path
"""
def _lookup(self, name, cls, type, timeout=None):
"""
Look up records in authority data.
Args:
name (str): Query name
cls (int): Record class
type (int): Record type
timeout (int): Query timeout
Returns:
Deferred: Query results
"""DNS Server Example:
from twisted.names import dns, server, authority, cache, client
from twisted.internet import reactor, endpoints
from twisted.application import service
# Create resolvers
authority_resolver = authority.FileAuthority('/etc/bind/db.example.com')
cache_resolver = cache.CacheResolver()
client_resolver = client.Resolver()
# Create DNS server factory
dns_factory = server.DNSServerFactory(
authorities=[authority_resolver],
caches=[cache_resolver],
clients=[client_resolver]
)
# Start DNS server
udp_endpoint = endpoints.UDP4ServerEndpoint(reactor, 53)
tcp_endpoint = endpoints.TCP4ServerEndpoint(reactor, 53)
udp_endpoint.listen(dns.DNSDatagramProtocol(dns_factory))
tcp_endpoint.listen(dns_factory)
print("DNS server listening on port 53")
reactor.run()Hosts File Resolver
DNS resolver that uses system hosts file.
class hosts.Resolver:
"""
DNS resolver using hosts file.
"""
def __init__(self, file='/etc/hosts'):
"""
Args:
file (str): Hosts file path
"""
def lookupAddress(self, name, timeout=None):
"""
Look up address in hosts file.
Args:
name (str): Hostname
timeout (int): Query timeout (ignored)
Returns:
Deferred[list]: A records from hosts file
"""Install with Tessl CLI
npx tessl i tessl/pypi-twisted