X7ROOT File Manager
Current Path:
/opt/hc_python/lib/python3.12/site-packages/dns
opt
/
hc_python
/
lib
/
python3.12
/
site-packages
/
dns
/
ðŸ“
..
📄
__init__.py
(1.62 KB)
ðŸ“
__pycache__
📄
_asyncbackend.py
(2.34 KB)
📄
_asyncio_backend.py
(8.84 KB)
📄
_ddr.py
(5.12 KB)
📄
_features.py
(2.43 KB)
📄
_immutable_ctx.py
(2.4 KB)
📄
_trio_backend.py
(8.27 KB)
📄
asyncbackend.py
(2.73 KB)
📄
asyncquery.py
(30.1 KB)
📄
asyncresolver.py
(17.43 KB)
📄
dnssec.py
(40.74 KB)
ðŸ“
dnssecalgs
📄
dnssectypes.py
(1.76 KB)
📄
e164.py
(3.88 KB)
📄
edns.py
(16.69 KB)
📄
entropy.py
(4.14 KB)
📄
enum.py
(3.6 KB)
📄
exception.py
(5.81 KB)
📄
flags.py
(2.69 KB)
📄
grange.py
(2.09 KB)
📄
immutable.py
(1.97 KB)
📄
inet.py
(5.64 KB)
📄
ipv4.py
(2.49 KB)
📄
ipv6.py
(6.4 KB)
📄
message.py
(66.59 KB)
📄
name.py
(41.78 KB)
📄
namedict.py
(3.91 KB)
📄
nameserver.py
(9.88 KB)
📄
node.py
(12.37 KB)
📄
opcode.py
(2.67 KB)
📄
py.typed
(0 B)
📄
query.py
(54.98 KB)
ðŸ“
quic
📄
rcode.py
(4.06 KB)
📄
rdata.py
(30.29 KB)
📄
rdataclass.py
(2.91 KB)
📄
rdataset.py
(16.27 KB)
📄
rdatatype.py
(7.27 KB)
ðŸ“
rdtypes
📄
renderer.py
(10.99 KB)
📄
resolver.py
(72 KB)
📄
reversename.py
(3.74 KB)
📄
rrset.py
(8.96 KB)
📄
serial.py
(3.52 KB)
📄
set.py
(9 KB)
📄
tokenizer.py
(23.03 KB)
📄
transaction.py
(22.06 KB)
📄
tsig.py
(11.15 KB)
📄
tsigkeyring.py
(2.57 KB)
📄
ttl.py
(2.91 KB)
📄
update.py
(11.96 KB)
📄
version.py
(1.88 KB)
📄
versioned.py
(11.49 KB)
📄
win32util.py
(8.67 KB)
📄
wire.py
(2.76 KB)
📄
xfr.py
(12.96 KB)
📄
zone.py
(50.87 KB)
📄
zonefile.py
(27.27 KB)
📄
zonetypes.py
(690 B)
Editing: nameserver.py
from typing import Optional, Union from urllib.parse import urlparse import dns.asyncbackend import dns.asyncquery import dns.inet import dns.message import dns.query class Nameserver: def __init__(self): pass def __str__(self): raise NotImplementedError def kind(self) -> str: raise NotImplementedError def is_always_max_size(self) -> bool: raise NotImplementedError def answer_nameserver(self) -> str: raise NotImplementedError def answer_port(self) -> int: raise NotImplementedError def query( self, request: dns.message.QueryMessage, timeout: float, source: Optional[str], source_port: int, max_size: bool, one_rr_per_rrset: bool = False, ignore_trailing: bool = False, ) -> dns.message.Message: raise NotImplementedError async def async_query( self, request: dns.message.QueryMessage, timeout: float, source: Optional[str], source_port: int, max_size: bool, backend: dns.asyncbackend.Backend, one_rr_per_rrset: bool = False, ignore_trailing: bool = False, ) -> dns.message.Message: raise NotImplementedError class AddressAndPortNameserver(Nameserver): def __init__(self, address: str, port: int): super().__init__() self.address = address self.port = port def kind(self) -> str: raise NotImplementedError def is_always_max_size(self) -> bool: return False def __str__(self): ns_kind = self.kind() return f"{ns_kind}:{self.address}@{self.port}" def answer_nameserver(self) -> str: return self.address def answer_port(self) -> int: return self.port class Do53Nameserver(AddressAndPortNameserver): def __init__(self, address: str, port: int = 53): super().__init__(address, port) def kind(self): return "Do53" def query( self, request: dns.message.QueryMessage, timeout: float, source: Optional[str], source_port: int, max_size: bool, one_rr_per_rrset: bool = False, ignore_trailing: bool = False, ) -> dns.message.Message: if max_size: response = dns.query.tcp( request, self.address, timeout=timeout, port=self.port, source=source, source_port=source_port, one_rr_per_rrset=one_rr_per_rrset, ignore_trailing=ignore_trailing, ) else: response = dns.query.udp( request, self.address, timeout=timeout, port=self.port, source=source, source_port=source_port, raise_on_truncation=True, one_rr_per_rrset=one_rr_per_rrset, ignore_trailing=ignore_trailing, ignore_errors=True, ignore_unexpected=True, ) return response async def async_query( self, request: dns.message.QueryMessage, timeout: float, source: Optional[str], source_port: int, max_size: bool, backend: dns.asyncbackend.Backend, one_rr_per_rrset: bool = False, ignore_trailing: bool = False, ) -> dns.message.Message: if max_size: response = await dns.asyncquery.tcp( request, self.address, timeout=timeout, port=self.port, source=source, source_port=source_port, backend=backend, one_rr_per_rrset=one_rr_per_rrset, ignore_trailing=ignore_trailing, ) else: response = await dns.asyncquery.udp( request, self.address, timeout=timeout, port=self.port, source=source, source_port=source_port, raise_on_truncation=True, backend=backend, one_rr_per_rrset=one_rr_per_rrset, ignore_trailing=ignore_trailing, ignore_errors=True, ignore_unexpected=True, ) return response class DoHNameserver(Nameserver): def __init__( self, url: str, bootstrap_address: Optional[str] = None, verify: Union[bool, str] = True, want_get: bool = False, http_version: dns.query.HTTPVersion = dns.query.HTTPVersion.DEFAULT, ): super().__init__() self.url = url self.bootstrap_address = bootstrap_address self.verify = verify self.want_get = want_get self.http_version = http_version def kind(self): return "DoH" def is_always_max_size(self) -> bool: return True def __str__(self): return self.url def answer_nameserver(self) -> str: return self.url def answer_port(self) -> int: port = urlparse(self.url).port if port is None: port = 443 return port def query( self, request: dns.message.QueryMessage, timeout: float, source: Optional[str], source_port: int, max_size: bool = False, one_rr_per_rrset: bool = False, ignore_trailing: bool = False, ) -> dns.message.Message: return dns.query.https( request, self.url, timeout=timeout, source=source, source_port=source_port, bootstrap_address=self.bootstrap_address, one_rr_per_rrset=one_rr_per_rrset, ignore_trailing=ignore_trailing, verify=self.verify, post=(not self.want_get), http_version=self.http_version, ) async def async_query( self, request: dns.message.QueryMessage, timeout: float, source: Optional[str], source_port: int, max_size: bool, backend: dns.asyncbackend.Backend, one_rr_per_rrset: bool = False, ignore_trailing: bool = False, ) -> dns.message.Message: return await dns.asyncquery.https( request, self.url, timeout=timeout, source=source, source_port=source_port, bootstrap_address=self.bootstrap_address, one_rr_per_rrset=one_rr_per_rrset, ignore_trailing=ignore_trailing, verify=self.verify, post=(not self.want_get), http_version=self.http_version, ) class DoTNameserver(AddressAndPortNameserver): def __init__( self, address: str, port: int = 853, hostname: Optional[str] = None, verify: Union[bool, str] = True, ): super().__init__(address, port) self.hostname = hostname self.verify = verify def kind(self): return "DoT" def query( self, request: dns.message.QueryMessage, timeout: float, source: Optional[str], source_port: int, max_size: bool = False, one_rr_per_rrset: bool = False, ignore_trailing: bool = False, ) -> dns.message.Message: return dns.query.tls( request, self.address, port=self.port, timeout=timeout, one_rr_per_rrset=one_rr_per_rrset, ignore_trailing=ignore_trailing, server_hostname=self.hostname, verify=self.verify, ) async def async_query( self, request: dns.message.QueryMessage, timeout: float, source: Optional[str], source_port: int, max_size: bool, backend: dns.asyncbackend.Backend, one_rr_per_rrset: bool = False, ignore_trailing: bool = False, ) -> dns.message.Message: return await dns.asyncquery.tls( request, self.address, port=self.port, timeout=timeout, one_rr_per_rrset=one_rr_per_rrset, ignore_trailing=ignore_trailing, server_hostname=self.hostname, verify=self.verify, ) class DoQNameserver(AddressAndPortNameserver): def __init__( self, address: str, port: int = 853, verify: Union[bool, str] = True, server_hostname: Optional[str] = None, ): super().__init__(address, port) self.verify = verify self.server_hostname = server_hostname def kind(self): return "DoQ" def query( self, request: dns.message.QueryMessage, timeout: float, source: Optional[str], source_port: int, max_size: bool = False, one_rr_per_rrset: bool = False, ignore_trailing: bool = False, ) -> dns.message.Message: return dns.query.quic( request, self.address, port=self.port, timeout=timeout, one_rr_per_rrset=one_rr_per_rrset, ignore_trailing=ignore_trailing, verify=self.verify, server_hostname=self.server_hostname, ) async def async_query( self, request: dns.message.QueryMessage, timeout: float, source: Optional[str], source_port: int, max_size: bool, backend: dns.asyncbackend.Backend, one_rr_per_rrset: bool = False, ignore_trailing: bool = False, ) -> dns.message.Message: return await dns.asyncquery.quic( request, self.address, port=self.port, timeout=timeout, one_rr_per_rrset=one_rr_per_rrset, ignore_trailing=ignore_trailing, verify=self.verify, server_hostname=self.server_hostname, )
Upload File
Create Folder