Polina/robot-control-system
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Загрузить файлы в «venv/Lib/site-packages/dns»

Browse Source
This commit is contained in:
Polina
2026-07-02 18:06:30 +00:00
parent 84c061551c
commit 0dc649ddb0
5 changed files with 3890 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

953
venv/Lib/site-packages/dns/asyncquery.py Normal file
View File

@@ -0,0 +1,953 @@
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
# Copyright (C) 2003-2017 Nominum, Inc.
#
# Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose with or without fee is hereby granted,
# provided that the above copyright notice and this permission notice
# appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
"""Talk to a DNS server."""
import base64
import contextlib
import random
import socket
import struct
import time
import urllib.parse
from typing import Any, Dict, Optional, Tuple, cast
import dns.asyncbackend
import dns.exception
import dns.inet
import dns.message
import dns.name
import dns.quic
import dns.rdatatype
import dns.transaction
import dns.tsig
import dns.xfr
from dns._asyncbackend import NullContext
from dns.query import (
BadResponse,
HTTPVersion,
NoDOH,
NoDOQ,
UDPMode,
_check_status,
_compute_times,
_matches_destination,
_remaining,
have_doh,
make_ssl_context,
)
try:
import ssl
except ImportError:
import dns._no_ssl as ssl # type: ignore
if have_doh:
import httpx
# for brevity
_lltuple = dns.inet.low_level_address_tuple
def _source_tuple(af, address, port):
# Make a high level source tuple, or return None if address and port
# are both None
if address or port:
if address is None:
if af == socket.AF_INET:
address = "0.0.0.0"
elif af == socket.AF_INET6:
address = "::"
else:
raise NotImplementedError(f"unknown address family {af}")
return (address, port)
else:
return None
def _timeout(expiration, now=None):
if expiration is not None:
if not now:
now = time.time()
return max(expiration - now, 0)
else:
return None
async def send_udp(
sock: dns.asyncbackend.DatagramSocket,
what: dns.message.Message | bytes,
destination: Any,
expiration: float | None = None,
) -> Tuple[int, float]:
"""Send a DNS message to the specified UDP socket.
*sock*, a ``dns.asyncbackend.DatagramSocket``.
*what*, a ``bytes`` or ``dns.message.Message``, the message to send.
*destination*, a destination tuple appropriate for the address family
of the socket, specifying where to send the query.
*expiration*, a ``float`` or ``None``, the absolute time at which
a timeout exception should be raised. If ``None``, no timeout will
occur. The expiration value is meaningless for the asyncio backend, as
asyncio's transport sendto() never blocks.
Returns an ``(int, float)`` tuple of bytes sent and the sent time.
"""
if isinstance(what, dns.message.Message):
what = what.to_wire()
sent_time = time.time()
n = await sock.sendto(what, destination, _timeout(expiration, sent_time))
return (n, sent_time)
async def receive_udp(
sock: dns.asyncbackend.DatagramSocket,
destination: Any | None = None,
expiration: float | None = None,
ignore_unexpected: bool = False,
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,
raise_on_truncation: bool = False,
ignore_errors: bool = False,
query: dns.message.Message | None = None,
) -> Any:
"""Read a DNS message from a UDP socket.
*sock*, a ``dns.asyncbackend.DatagramSocket``.
See :py:func:`dns.query.receive_udp()` for the documentation of the other
parameters, and exceptions.
Returns a ``(dns.message.Message, float, tuple)`` tuple of the received message, the
received time, and the address where the message arrived from.
"""
wire = b""
while True:
(wire, from_address) = await sock.recvfrom(65535, _timeout(expiration))
if not _matches_destination(
sock.family, from_address, destination, ignore_unexpected
):
continue
received_time = time.time()
try:
r = dns.message.from_wire(
wire,
keyring=keyring,
request_mac=request_mac,
one_rr_per_rrset=one_rr_per_rrset,
ignore_trailing=ignore_trailing,
raise_on_truncation=raise_on_truncation,
)
except dns.message.Truncated as e:
# See the comment in query.py for details.
if (
ignore_errors
and query is not None
and not query.is_response(e.message())
):
continue
else:
raise
except Exception:
if ignore_errors:
continue
else:
raise
if ignore_errors and query is not None and not query.is_response(r):
continue
return (r, received_time, from_address)
async def udp(
q: dns.message.Message,
where: str,
timeout: float | None = None,
port: int = 53,
source: str | None = None,
source_port: int = 0,
ignore_unexpected: bool = False,
one_rr_per_rrset: bool = False,
ignore_trailing: bool = False,
raise_on_truncation: bool = False,
sock: dns.asyncbackend.DatagramSocket | None = None,
backend: dns.asyncbackend.Backend | None = None,
ignore_errors: bool = False,
) -> dns.message.Message:
"""Return the response obtained after sending a query via UDP.
*sock*, a ``dns.asyncbackend.DatagramSocket``, or ``None``,
the socket to use for the query. If ``None``, the default, a
socket is created. Note that if a socket is provided, the
*source*, *source_port*, and *backend* are ignored.
*backend*, a ``dns.asyncbackend.Backend``, or ``None``. If ``None``,
the default, then dnspython will use the default backend.
See :py:func:`dns.query.udp()` for the documentation of the other
parameters, exceptions, and return type of this method.
"""
wire = q.to_wire()
(begin_time, expiration) = _compute_times(timeout)
af = dns.inet.af_for_address(where)
destination = _lltuple((where, port), af)
if sock:
cm: contextlib.AbstractAsyncContextManager = NullContext(sock)
else:
if not backend:
backend = dns.asyncbackend.get_default_backend()
stuple = _source_tuple(af, source, source_port)
if backend.datagram_connection_required():
dtuple = (where, port)
else:
dtuple = None
cm = await backend.make_socket(af, socket.SOCK_DGRAM, 0, stuple, dtuple)
async with cm as s:
await send_udp(s, wire, destination, expiration) # pyright: ignore
(r, received_time, _) = await receive_udp(
s, # pyright: ignore
destination,
expiration,
ignore_unexpected,
one_rr_per_rrset,
q.keyring,
q.mac,
ignore_trailing,
raise_on_truncation,
ignore_errors,
q,
)
r.time = received_time - begin_time
# We don't need to check q.is_response() if we are in ignore_errors mode
# as receive_udp() will have checked it.
if not (ignore_errors or q.is_response(r)):
raise BadResponse
return r
async def udp_with_fallback(
q: dns.message.Message,
where: str,
timeout: float | None = None,
port: int = 53,
source: str | None = None,
source_port: int = 0,
ignore_unexpected: bool = False,
one_rr_per_rrset: bool = False,
ignore_trailing: bool = False,
udp_sock: dns.asyncbackend.DatagramSocket | None = None,
tcp_sock: dns.asyncbackend.StreamSocket | None = None,
backend: dns.asyncbackend.Backend | None = None,
ignore_errors: bool = False,
) -> Tuple[dns.message.Message, bool]:
"""Return the response to the query, trying UDP first and falling back
to TCP if UDP results in a truncated response.
*udp_sock*, a ``dns.asyncbackend.DatagramSocket``, or ``None``,
the socket to use for the UDP query. If ``None``, the default, a
socket is created. Note that if a socket is provided the *source*,
*source_port*, and *backend* are ignored for the UDP query.
*tcp_sock*, a ``dns.asyncbackend.StreamSocket``, or ``None``, the
socket to use for the TCP query. If ``None``, the default, a
socket is created. Note that if a socket is provided *where*,
*source*, *source_port*, and *backend* are ignored for the TCP query.
*backend*, a ``dns.asyncbackend.Backend``, or ``None``. If ``None``,
the default, then dnspython will use the default backend.
See :py:func:`dns.query.udp_with_fallback()` for the documentation
of the other parameters, exceptions, and return type of this
method.
"""
try:
response = await udp(
q,
where,
timeout,
port,
source,
source_port,
ignore_unexpected,
one_rr_per_rrset,
ignore_trailing,
True,
udp_sock,
backend,
ignore_errors,
)
return (response, False)
except dns.message.Truncated:
response = await tcp(
q,
where,
timeout,
port,
source,
source_port,
one_rr_per_rrset,
ignore_trailing,
tcp_sock,
backend,
)
return (response, True)
async def send_tcp(
sock: dns.asyncbackend.StreamSocket,
what: dns.message.Message | bytes,
expiration: float | None = None,
) -> Tuple[int, float]:
"""Send a DNS message to the specified TCP socket.
*sock*, a ``dns.asyncbackend.StreamSocket``.
See :py:func:`dns.query.send_tcp()` for the documentation of the other
parameters, exceptions, and return type of this method.
"""
if isinstance(what, dns.message.Message):
tcpmsg = what.to_wire(prepend_length=True)
else:
# copying the wire into tcpmsg is inefficient, but lets us
# avoid writev() or doing a short write that would get pushed
# onto the net
tcpmsg = len(what).to_bytes(2, "big") + what
sent_time = time.time()
await sock.sendall(tcpmsg, _timeout(expiration, sent_time))
return (len(tcpmsg), sent_time)
async def _read_exactly(sock, count, expiration):
"""Read the specified number of bytes from stream. Keep trying until we
either get the desired amount, or we hit EOF.
"""
s = b""
while count > 0:
n = await sock.recv(count, _timeout(expiration))
if n == b"":
raise EOFError("EOF")
count = count - len(n)
s = s + n
return s
async def receive_tcp(
sock: dns.asyncbackend.StreamSocket,
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 ``dns.asyncbackend.StreamSocket``.
See :py:func:`dns.query.receive_tcp()` for the documentation of the other
parameters, exceptions, and return type of this method.
"""
ldata = await _read_exactly(sock, 2, expiration)
(l,) = struct.unpack("!H", ldata)
wire = await _read_exactly(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)
async def tcp(
q: dns.message.Message,
where: str,
timeout: float | None = None,
port: int = 53,
source: str | None = None,
source_port: int = 0,
one_rr_per_rrset: bool = False,
ignore_trailing: bool = False,
sock: dns.asyncbackend.StreamSocket | None = None,
backend: dns.asyncbackend.Backend | None = None,
) -> dns.message.Message:
"""Return the response obtained after sending a query via TCP.
*sock*, a ``dns.asyncbacket.StreamSocket``, or ``None``, the
socket to use for the query. If ``None``, the default, a socket
is created. Note that if a socket is provided
*where*, *port*, *source*, *source_port*, and *backend* are ignored.
*backend*, a ``dns.asyncbackend.Backend``, or ``None``. If ``None``,
the default, then dnspython will use the default backend.
See :py:func:`dns.query.tcp()` for the documentation of the other
parameters, exceptions, and return type of this method.
"""
wire = q.to_wire()
(begin_time, expiration) = _compute_times(timeout)
if sock:
# Verify that the socket is connected, as if it's not connected,
# it's not writable, and the polling in send_tcp() will time out or
# hang forever.
await sock.getpeername()
cm: contextlib.AbstractAsyncContextManager = NullContext(sock)
else:
# These are simple (address, port) pairs, not family-dependent tuples
# you pass to low-level socket code.
af = dns.inet.af_for_address(where)
stuple = _source_tuple(af, source, source_port)
dtuple = (where, port)
if not backend:
backend = dns.asyncbackend.get_default_backend()
cm = await backend.make_socket(
af, socket.SOCK_STREAM, 0, stuple, dtuple, timeout
)
async with cm as s:
await send_tcp(s, wire, expiration) # pyright: ignore
(r, received_time) = await receive_tcp(
s, # pyright: ignore
expiration,
one_rr_per_rrset,
q.keyring,
q.mac,
ignore_trailing,
)
r.time = received_time - begin_time
if not q.is_response(r):
raise BadResponse
return r
async def tls(
q: dns.message.Message,
where: str,
timeout: float | None = None,
port: int = 853,
source: str | None = None,
source_port: int = 0,
one_rr_per_rrset: bool = False,
ignore_trailing: bool = False,
sock: dns.asyncbackend.StreamSocket | None = None,
backend: dns.asyncbackend.Backend | None = None,
ssl_context: ssl.SSLContext | None = None,
server_hostname: str | None = None,
verify: bool | str = True,
) -> dns.message.Message:
"""Return the response obtained after sending a query via TLS.
*sock*, an ``asyncbackend.StreamSocket``, or ``None``, the socket
to use for the query. If ``None``, the default, a socket is
created. Note that if a socket is provided, it must be a
connected SSL stream socket, and *where*, *port*,
*source*, *source_port*, *backend*, *ssl_context*, and *server_hostname*
are ignored.
*backend*, a ``dns.asyncbackend.Backend``, or ``None``. If ``None``,
the default, then dnspython will use the default backend.
See :py:func:`dns.query.tls()` for the documentation of the other
parameters, exceptions, and return type of this method.
"""
(begin_time, expiration) = _compute_times(timeout)
if sock:
cm: contextlib.AbstractAsyncContextManager = NullContext(sock)
else:
if ssl_context is None:
ssl_context = make_ssl_context(verify, server_hostname is not None, ["dot"])
af = dns.inet.af_for_address(where)
stuple = _source_tuple(af, source, source_port)
dtuple = (where, port)
if not backend:
backend = dns.asyncbackend.get_default_backend()
cm = await backend.make_socket(
af,
socket.SOCK_STREAM,
0,
stuple,
dtuple,
timeout,
ssl_context,
server_hostname,
)
async with cm as s:
timeout = _timeout(expiration)
response = await tcp(
q,
where,
timeout,
port,
source,
source_port,
one_rr_per_rrset,
ignore_trailing,
s,
backend,
)
end_time = time.time()
response.time = end_time - begin_time
return response
def _maybe_get_resolver(
resolver: Optional["dns.asyncresolver.Resolver"], # pyright: ignore
) -> "dns.asyncresolver.Resolver": # pyright: ignore
# We need a separate method for this to avoid overriding the global
# variable "dns" with the as-yet undefined local variable "dns"
# in https().
if resolver is None:
# pylint: disable=import-outside-toplevel,redefined-outer-name
import dns.asyncresolver
resolver = dns.asyncresolver.Resolver()
return resolver
async def https(
q: dns.message.Message,
where: str,
timeout: float | None = None,
port: int = 443,
source: str | None = None,
source_port: int = 0, # pylint: disable=W0613
one_rr_per_rrset: bool = False,
ignore_trailing: bool = False,
client: Optional["httpx.AsyncClient|dns.quic.AsyncQuicConnection"] = None,
path: str = "/dns-query",
post: bool = True,
verify: bool | str | ssl.SSLContext = True,
bootstrap_address: str | None = None,
resolver: Optional["dns.asyncresolver.Resolver"] = None, # pyright: ignore
family: int = socket.AF_UNSPEC,
http_version: HTTPVersion = HTTPVersion.DEFAULT,
) -> dns.message.Message:
"""Return the response obtained after sending a query via DNS-over-HTTPS.
*client*, a ``httpx.AsyncClient``. If provided, the client to use for
the query.
Unlike the other dnspython async functions, a backend cannot be provided
in this function because httpx always auto-detects the async backend.
See :py:func:`dns.query.https()` for the documentation of the other
parameters, exceptions, and return type of this method.
"""
try:
af = dns.inet.af_for_address(where)
except ValueError:
af = None
# we bind url and then override as pyright can't figure out all paths bind.
url = where
if af is not None and dns.inet.is_address(where):
if af == socket.AF_INET:
url = f"https://{where}:{port}{path}"
elif af == socket.AF_INET6:
url = f"https://[{where}]:{port}{path}"
extensions = {}
if bootstrap_address is None:
# pylint: disable=possibly-used-before-assignment
parsed = urllib.parse.urlparse(url)
if parsed.hostname is None:
raise ValueError("no hostname in URL")
if dns.inet.is_address(parsed.hostname):
bootstrap_address = parsed.hostname
extensions["sni_hostname"] = parsed.hostname
if parsed.port is not None:
port = parsed.port
if http_version == HTTPVersion.H3 or (
http_version == HTTPVersion.DEFAULT and not have_doh
):
if bootstrap_address is None:
resolver = _maybe_get_resolver(resolver)
assert parsed.hostname is not None # pyright: ignore
answers = await resolver.resolve_name( # pyright: ignore
parsed.hostname, family # pyright: ignore
)
bootstrap_address = random.choice(list(answers.addresses()))
if client and not isinstance(
client, dns.quic.AsyncQuicConnection
): # pyright: ignore
raise ValueError("client parameter must be a dns.quic.AsyncQuicConnection.")
assert client is None or isinstance(client, dns.quic.AsyncQuicConnection)
return await _http3(
q,
bootstrap_address,
url,
timeout,
port,
source,
source_port,
one_rr_per_rrset,
ignore_trailing,
verify=verify,
post=post,
connection=client,
)
if not have_doh:
raise NoDOH # pragma: no cover
# pylint: disable=possibly-used-before-assignment
if client and not isinstance(client, httpx.AsyncClient): # pyright: ignore
raise ValueError("client parameter must be an httpx.AsyncClient")
# pylint: enable=possibly-used-before-assignment
wire = q.to_wire()
headers = {"accept": "application/dns-message"}
h1 = http_version in (HTTPVersion.H1, HTTPVersion.DEFAULT)
h2 = http_version in (HTTPVersion.H2, HTTPVersion.DEFAULT)
backend = dns.asyncbackend.get_default_backend()
if source is None:
local_address = None
local_port = 0
else:
local_address = source
local_port = source_port
if client:
cm: contextlib.AbstractAsyncContextManager = NullContext(client)
else:
transport = backend.get_transport_class()(
local_address=local_address,
http1=h1,
http2=h2,
verify=verify,
local_port=local_port,
bootstrap_address=bootstrap_address,
resolver=resolver,
family=family,
)
cm = httpx.AsyncClient( # pyright: ignore
http1=h1, http2=h2, verify=verify, transport=transport # type: ignore
)
async with cm as the_client:
# see https://tools.ietf.org/html/rfc8484#section-4.1.1 for DoH
# GET and POST examples
if post:
headers.update(
{
"content-type": "application/dns-message",
"content-length": str(len(wire)),
}
)
response = await backend.wait_for(
the_client.post( # pyright: ignore
url,
headers=headers,
content=wire,
extensions=extensions,
),
timeout,
)
else:
wire = base64.urlsafe_b64encode(wire).rstrip(b"=")
twire = wire.decode() # httpx does a repr() if we give it bytes
response = await backend.wait_for(
the_client.get( # pyright: ignore
url,
headers=headers,
params={"dns": twire},
extensions=extensions,
),
timeout,
)
# see https://tools.ietf.org/html/rfc8484#section-4.2.1 for info about DoH
# status codes
if response.status_code < 200 or response.status_code > 299:
raise ValueError(
f"{where} responded with status code {response.status_code}"
f"\nResponse body: {response.content!r}"
)
r = dns.message.from_wire(
response.content,
keyring=q.keyring,
request_mac=q.request_mac,
one_rr_per_rrset=one_rr_per_rrset,
ignore_trailing=ignore_trailing,
)
r.time = response.elapsed.total_seconds()
if not q.is_response(r):
raise BadResponse
return r
async def _http3(
q: dns.message.Message,
where: str,
url: str,
timeout: float | None = None,
port: int = 443,
source: str | None = None,
source_port: int = 0,
one_rr_per_rrset: bool = False,
ignore_trailing: bool = False,
verify: bool | str | ssl.SSLContext = True,
backend: dns.asyncbackend.Backend | None = None,
post: bool = True,
connection: dns.quic.AsyncQuicConnection | None = None,
) -> dns.message.Message:
if not dns.quic.have_quic:
raise NoDOH("DNS-over-HTTP3 is not available.") # pragma: no cover
url_parts = urllib.parse.urlparse(url)
hostname = url_parts.hostname
assert hostname is not None
if url_parts.port is not None:
port = url_parts.port
q.id = 0
wire = q.to_wire()
the_connection: dns.quic.AsyncQuicConnection
if connection:
cfactory = dns.quic.null_factory
mfactory = dns.quic.null_factory
else:
(cfactory, mfactory) = dns.quic.factories_for_backend(backend)
async with cfactory() as context:
async with mfactory(
context, verify_mode=verify, server_name=hostname, h3=True
) as the_manager:
if connection:
the_connection = connection
else:
the_connection = the_manager.connect( # pyright: ignore
where, port, source, source_port
)
(start, expiration) = _compute_times(timeout)
stream = await the_connection.make_stream(timeout) # pyright: ignore
async with stream:
# note that send_h3() does not need await
stream.send_h3(url, wire, post)
wire = await stream.receive(_remaining(expiration))
_check_status(stream.headers(), where, wire)
finish = time.time()
r = dns.message.from_wire(
wire,
keyring=q.keyring,
request_mac=q.request_mac,
one_rr_per_rrset=one_rr_per_rrset,
ignore_trailing=ignore_trailing,
)
r.time = max(finish - start, 0.0)
if not q.is_response(r):
raise BadResponse
return r
async def quic(
q: dns.message.Message,
where: str,
timeout: float | None = None,
port: int = 853,
source: str | None = None,
source_port: int = 0,
one_rr_per_rrset: bool = False,
ignore_trailing: bool = False,
connection: dns.quic.AsyncQuicConnection | None = None,
verify: bool | str = True,
backend: dns.asyncbackend.Backend | None = None,
hostname: str | None = None,
server_hostname: str | None = None,
) -> dns.message.Message:
"""Return the response obtained after sending an asynchronous query via
DNS-over-QUIC.
*backend*, a ``dns.asyncbackend.Backend``, or ``None``. If ``None``,
the default, then dnspython will use the default backend.
See :py:func:`dns.query.quic()` for the documentation of the other
parameters, exceptions, and return type of this method.
"""
if not dns.quic.have_quic:
raise NoDOQ("DNS-over-QUIC is not available.") # pragma: no cover
if server_hostname is not None and hostname is None:
hostname = server_hostname
q.id = 0
wire = q.to_wire()
the_connection: dns.quic.AsyncQuicConnection
if connection:
cfactory = dns.quic.null_factory
mfactory = dns.quic.null_factory
the_connection = connection
else:
(cfactory, mfactory) = dns.quic.factories_for_backend(backend)
async with cfactory() as context:
async with mfactory(
context,
verify_mode=verify,
server_name=server_hostname,
) as the_manager:
if not connection:
the_connection = the_manager.connect( # pyright: ignore
where, port, source, source_port
)
(start, expiration) = _compute_times(timeout)
stream = await the_connection.make_stream(timeout) # pyright: ignore
async with stream:
await stream.send(wire, True)
wire = await stream.receive(_remaining(expiration))
finish = time.time()
r = dns.message.from_wire(
wire,
keyring=q.keyring,
request_mac=q.request_mac,
one_rr_per_rrset=one_rr_per_rrset,
ignore_trailing=ignore_trailing,
)
r.time = max(finish - start, 0.0)
if not q.is_response(r):
raise BadResponse
return r
async def _inbound_xfr(
txn_manager: dns.transaction.TransactionManager,
s: dns.asyncbackend.Socket,
query: dns.message.Message,
serial: int | None,
timeout: float | None,
expiration: float,
) -> Any:
"""Given a socket, does the zone transfer."""
rdtype = query.question[0].rdtype
is_ixfr = rdtype == dns.rdatatype.IXFR
origin = txn_manager.from_wire_origin()
wire = query.to_wire()
is_udp = s.type == socket.SOCK_DGRAM
if is_udp:
udp_sock = cast(dns.asyncbackend.DatagramSocket, s)
await udp_sock.sendto(wire, None, _timeout(expiration))
else:
tcp_sock = cast(dns.asyncbackend.StreamSocket, s)
tcpmsg = struct.pack("!H", len(wire)) + wire
await tcp_sock.sendall(tcpmsg, expiration)
with dns.xfr.Inbound(txn_manager, rdtype, serial, is_udp) as inbound:
done = False
tsig_ctx = None
r: dns.message.Message | None = None
while not done:
(_, mexpiration) = _compute_times(timeout)
if mexpiration is None or (
expiration is not None and mexpiration > expiration
):
mexpiration = expiration
if is_udp:
timeout = _timeout(mexpiration)
(rwire, _) = await udp_sock.recvfrom(65535, timeout) # pyright: ignore
else:
ldata = await _read_exactly(tcp_sock, 2, mexpiration) # pyright: ignore
(l,) = struct.unpack("!H", ldata)
rwire = await _read_exactly(tcp_sock, l, mexpiration) # pyright: ignore
r = dns.message.from_wire(
rwire,
keyring=query.keyring,
request_mac=query.mac,
xfr=True,
origin=origin,
tsig_ctx=tsig_ctx,
multi=(not is_udp),
one_rr_per_rrset=is_ixfr,
)
done = inbound.process_message(r)
yield r
tsig_ctx = r.tsig_ctx
if query.keyring and r is not None and not r.had_tsig:
raise dns.exception.FormError("missing TSIG")
async def inbound_xfr(
where: str,
txn_manager: dns.transaction.TransactionManager,
query: dns.message.Message | None = None,
port: int = 53,
timeout: float | None = None,
lifetime: float | None = None,
source: str | None = None,
source_port: int = 0,
udp_mode: UDPMode = UDPMode.NEVER,
backend: dns.asyncbackend.Backend | None = None,
) -> None:
"""Conduct an inbound transfer and apply it via a transaction from the
txn_manager.
*backend*, a ``dns.asyncbackend.Backend``, or ``None``. If ``None``,
the default, then dnspython will use the default backend.
See :py:func:`dns.query.inbound_xfr()` for the documentation of
the other parameters, exceptions, and return type of this method.
"""
if query is None:
(query, serial) = dns.xfr.make_query(txn_manager)
else:
serial = dns.xfr.extract_serial_from_query(query)
af = dns.inet.af_for_address(where)
stuple = _source_tuple(af, source, source_port)
dtuple = (where, port)
if not backend:
backend = dns.asyncbackend.get_default_backend()
(_, expiration) = _compute_times(lifetime)
if query.question[0].rdtype == dns.rdatatype.IXFR and udp_mode != UDPMode.NEVER:
s = await backend.make_socket(
af, socket.SOCK_DGRAM, 0, stuple, dtuple, _timeout(expiration)
)
async with s:
try:
async for _ in _inbound_xfr( # pyright: ignore
txn_manager,
s,
query,
serial,
timeout,
expiration, # pyright: ignore
):
pass
return
except dns.xfr.UseTCP:
if udp_mode == UDPMode.ONLY:
raise
s = await backend.make_socket(
af, socket.SOCK_STREAM, 0, stuple, dtuple, _timeout(expiration)
)
async with s:
async for _ in _inbound_xfr( # pyright: ignore
txn_manager, s, query, serial, timeout, expiration # pyright: ignore
):
pass

478
venv/Lib/site-packages/dns/asyncresolver.py Normal file
View File

@@ -0,0 +1,478 @@
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
# Copyright (C) 2003-2017 Nominum, Inc.
#
# Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose with or without fee is hereby granted,
# provided that the above copyright notice and this permission notice
# appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
"""Asynchronous DNS stub resolver."""
import socket
import time
from typing import Any, Dict, List
import dns._ddr
import dns.asyncbackend
import dns.asyncquery
import dns.exception
import dns.inet
import dns.name
import dns.nameserver
import dns.query
import dns.rdataclass
import dns.rdatatype
import dns.resolver # lgtm[py/import-and-import-from]
import dns.reversename
# import some resolver symbols for brevity
from dns.resolver import NXDOMAIN, NoAnswer, NoRootSOA, NotAbsolute
# for indentation purposes below
_udp = dns.asyncquery.udp
_tcp = dns.asyncquery.tcp
class Resolver(dns.resolver.BaseResolver):
"""Asynchronous DNS stub resolver."""
async def resolve(
self,
qname: dns.name.Name | str,
rdtype: dns.rdatatype.RdataType | str = dns.rdatatype.A,
rdclass: dns.rdataclass.RdataClass | str = dns.rdataclass.IN,
tcp: bool = False,
source: str | None = None,
raise_on_no_answer: bool = True,
source_port: int = 0,
lifetime: float | None = None,
search: bool | None = None,
backend: dns.asyncbackend.Backend | None = None,
) -> dns.resolver.Answer:
"""Query nameservers asynchronously to find the answer to the question.
*backend*, a ``dns.asyncbackend.Backend``, or ``None``. If ``None``,
the default, then dnspython will use the default backend.
See :py:func:`dns.resolver.Resolver.resolve()` for the
documentation of the other parameters, exceptions, and return
type of this method.
"""
resolution = dns.resolver._Resolution(
self, qname, rdtype, rdclass, tcp, raise_on_no_answer, search
)
if not backend:
backend = dns.asyncbackend.get_default_backend()
start = time.time()
while True:
(request, answer) = resolution.next_request()
# Note we need to say "if answer is not None" and not just
# "if answer" because answer implements __len__, and python
# will call that. We want to return if we have an answer
# object, including in cases where its length is 0.
if answer is not None:
# cache hit!
return answer
assert request is not None # needed for type checking
done = False
while not done:
(nameserver, tcp, backoff) = resolution.next_nameserver()
if backoff:
await backend.sleep(backoff)
timeout = self._compute_timeout(start, lifetime, resolution.errors)
try:
response = await nameserver.async_query(
request,
timeout=timeout,
source=source,
source_port=source_port,
max_size=tcp,
backend=backend,
)
except Exception as ex:
(_, done) = resolution.query_result(None, ex)
continue
(answer, done) = resolution.query_result(response, None)
# Note we need to say "if answer is not None" and not just
# "if answer" because answer implements __len__, and python
# will call that. We want to return if we have an answer
# object, including in cases where its length is 0.
if answer is not None:
return answer
async def resolve_address(
self, ipaddr: str, *args: Any, **kwargs: Any
) -> dns.resolver.Answer:
"""Use an asynchronous resolver to run a reverse query for PTR
records.
This utilizes the resolve() method to perform a PTR lookup on the
specified IP address.
*ipaddr*, a ``str``, the IPv4 or IPv6 address you want to get
the PTR record for.
All other arguments that can be passed to the resolve() function
except for rdtype and rdclass are also supported by this
function.
"""
# We make a modified kwargs for type checking happiness, as otherwise
# we get a legit warning about possibly having rdtype and rdclass
# in the kwargs more than once.
modified_kwargs: Dict[str, Any] = {}
modified_kwargs.update(kwargs)
modified_kwargs["rdtype"] = dns.rdatatype.PTR
modified_kwargs["rdclass"] = dns.rdataclass.IN
return await self.resolve(
dns.reversename.from_address(ipaddr), *args, **modified_kwargs
)
async def resolve_name(
self,
name: dns.name.Name | str,
family: int = socket.AF_UNSPEC,
**kwargs: Any,
) -> dns.resolver.HostAnswers:
"""Use an asynchronous resolver to query for address records.
This utilizes the resolve() method to perform A and/or AAAA lookups on
the specified name.
*qname*, a ``dns.name.Name`` or ``str``, the name to resolve.
*family*, an ``int``, the address family. If socket.AF_UNSPEC
(the default), both A and AAAA records will be retrieved.
All other arguments that can be passed to the resolve() function
except for rdtype and rdclass are also supported by this
function.
"""
# We make a modified kwargs for type checking happiness, as otherwise
# we get a legit warning about possibly having rdtype and rdclass
# in the kwargs more than once.
modified_kwargs: Dict[str, Any] = {}
modified_kwargs.update(kwargs)
modified_kwargs.pop("rdtype", None)
modified_kwargs["rdclass"] = dns.rdataclass.IN
if family == socket.AF_INET:
v4 = await self.resolve(name, dns.rdatatype.A, **modified_kwargs)
return dns.resolver.HostAnswers.make(v4=v4)
elif family == socket.AF_INET6:
v6 = await self.resolve(name, dns.rdatatype.AAAA, **modified_kwargs)
return dns.resolver.HostAnswers.make(v6=v6)
elif family != socket.AF_UNSPEC:
raise NotImplementedError(f"unknown address family {family}")
raise_on_no_answer = modified_kwargs.pop("raise_on_no_answer", True)
lifetime = modified_kwargs.pop("lifetime", None)
start = time.time()
v6 = await self.resolve(
name,
dns.rdatatype.AAAA,
raise_on_no_answer=False,
lifetime=self._compute_timeout(start, lifetime),
**modified_kwargs,
)
# Note that setting name ensures we query the same name
# for A as we did for AAAA. (This is just in case search lists
# are active by default in the resolver configuration and
# we might be talking to a server that says NXDOMAIN when it
# wants to say NOERROR no data.
name = v6.qname
v4 = await self.resolve(
name,
dns.rdatatype.A,
raise_on_no_answer=False,
lifetime=self._compute_timeout(start, lifetime),
**modified_kwargs,
)
answers = dns.resolver.HostAnswers.make(
v6=v6, v4=v4, add_empty=not raise_on_no_answer
)
if not answers:
raise NoAnswer(response=v6.response)
return answers
# pylint: disable=redefined-outer-name
async def canonical_name(self, name: dns.name.Name | str) -> dns.name.Name:
"""Determine the canonical name of *name*.
The canonical name is the name the resolver uses for queries
after all CNAME and DNAME renamings have been applied.
*name*, a ``dns.name.Name`` or ``str``, the query name.
This method can raise any exception that ``resolve()`` can
raise, other than ``dns.resolver.NoAnswer`` and
``dns.resolver.NXDOMAIN``.
Returns a ``dns.name.Name``.
"""
try:
answer = await self.resolve(name, raise_on_no_answer=False)
canonical_name = answer.canonical_name
except dns.resolver.NXDOMAIN as e:
canonical_name = e.canonical_name
return canonical_name
async def try_ddr(self, lifetime: float = 5.0) -> None:
"""Try to update the resolver's nameservers using Discovery of Designated
Resolvers (DDR). If successful, the resolver will subsequently use
DNS-over-HTTPS or DNS-over-TLS for future queries.
*lifetime*, a float, is the maximum time to spend attempting DDR. The default
is 5 seconds.
If the SVCB query is successful and results in a non-empty list of nameservers,
then the resolver's nameservers are set to the returned servers in priority
order.
The current implementation does not use any address hints from the SVCB record,
nor does it resolve addresses for the SCVB target name, rather it assumes that
the bootstrap nameserver will always be one of the addresses and uses it.
A future revision to the code may offer fuller support. The code verifies that
the bootstrap nameserver is in the Subject Alternative Name field of the
TLS certficate.
"""
try:
expiration = time.time() + lifetime
answer = await self.resolve(
dns._ddr._local_resolver_name, "svcb", lifetime=lifetime
)
timeout = dns.query._remaining(expiration)
nameservers = await dns._ddr._get_nameservers_async(answer, timeout)
if len(nameservers) > 0:
self.nameservers = nameservers
except Exception:
pass
default_resolver = None
def get_default_resolver() -> Resolver:
"""Get the default asynchronous resolver, initializing it if necessary."""
if default_resolver is None:
reset_default_resolver()
assert default_resolver is not None
return default_resolver
def reset_default_resolver() -> None:
"""Re-initialize default asynchronous resolver.
Note that the resolver configuration (i.e. /etc/resolv.conf on UNIX
systems) will be re-read immediately.
"""
global default_resolver
default_resolver = Resolver()
async def resolve(
qname: dns.name.Name | str,
rdtype: dns.rdatatype.RdataType | str = dns.rdatatype.A,
rdclass: dns.rdataclass.RdataClass | str = dns.rdataclass.IN,
tcp: bool = False,
source: str | None = None,
raise_on_no_answer: bool = True,
source_port: int = 0,
lifetime: float | None = None,
search: bool | None = None,
backend: dns.asyncbackend.Backend | None = None,
) -> dns.resolver.Answer:
"""Query nameservers asynchronously to find the answer to the question.
This is a convenience function that uses the default resolver
object to make the query.
See :py:func:`dns.asyncresolver.Resolver.resolve` for more
information on the parameters.
"""
return await get_default_resolver().resolve(
qname,
rdtype,
rdclass,
tcp,
source,
raise_on_no_answer,
source_port,
lifetime,
search,
backend,
)
async def resolve_address(
ipaddr: str, *args: Any, **kwargs: Any
) -> dns.resolver.Answer:
"""Use a resolver to run a reverse query for PTR records.
See :py:func:`dns.asyncresolver.Resolver.resolve_address` for more
information on the parameters.
"""
return await get_default_resolver().resolve_address(ipaddr, *args, **kwargs)
async def resolve_name(
name: dns.name.Name | str, family: int = socket.AF_UNSPEC, **kwargs: Any
) -> dns.resolver.HostAnswers:
"""Use a resolver to asynchronously query for address records.
See :py:func:`dns.asyncresolver.Resolver.resolve_name` for more
information on the parameters.
"""
return await get_default_resolver().resolve_name(name, family, **kwargs)
async def canonical_name(name: dns.name.Name | str) -> dns.name.Name:
"""Determine the canonical name of *name*.
See :py:func:`dns.resolver.Resolver.canonical_name` for more
information on the parameters and possible exceptions.
"""
return await get_default_resolver().canonical_name(name)
async def try_ddr(timeout: float = 5.0) -> None:
"""Try to update the default resolver's nameservers using Discovery of Designated
Resolvers (DDR). If successful, the resolver will subsequently use
DNS-over-HTTPS or DNS-over-TLS for future queries.
See :py:func:`dns.resolver.Resolver.try_ddr` for more information.
"""
return await get_default_resolver().try_ddr(timeout)
async def zone_for_name(
name: dns.name.Name | str,
rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN,
tcp: bool = False,
resolver: Resolver | None = None,
backend: dns.asyncbackend.Backend | None = None,
) -> dns.name.Name:
"""Find the name of the zone which contains the specified name.
See :py:func:`dns.resolver.Resolver.zone_for_name` for more
information on the parameters and possible exceptions.
"""
if isinstance(name, str):
name = dns.name.from_text(name, dns.name.root)
if resolver is None:
resolver = get_default_resolver()
if not name.is_absolute():
raise NotAbsolute(name)
while True:
try:
answer = await resolver.resolve(
name, dns.rdatatype.SOA, rdclass, tcp, backend=backend
)
assert answer.rrset is not None
if answer.rrset.name == name:
return name
# otherwise we were CNAMEd or DNAMEd and need to look higher
except (NXDOMAIN, NoAnswer):
pass
try:
name = name.parent()
except dns.name.NoParent: # pragma: no cover
raise NoRootSOA
async def make_resolver_at(
where: dns.name.Name | str,
port: int = 53,
family: int = socket.AF_UNSPEC,
resolver: Resolver | None = None,
) -> Resolver:
"""Make a stub resolver using the specified destination as the full resolver.
*where*, a ``dns.name.Name`` or ``str`` the domain name or IP address of the
full resolver.
*port*, an ``int``, the port to use. If not specified, the default is 53.
*family*, an ``int``, the address family to use. This parameter is used if
*where* is not an address. The default is ``socket.AF_UNSPEC`` in which case
the first address returned by ``resolve_name()`` will be used, otherwise the
first address of the specified family will be used.
*resolver*, a ``dns.asyncresolver.Resolver`` or ``None``, the resolver to use for
resolution of hostnames. If not specified, the default resolver will be used.
Returns a ``dns.resolver.Resolver`` or raises an exception.
"""
if resolver is None:
resolver = get_default_resolver()
nameservers: List[str | dns.nameserver.Nameserver] = []
if isinstance(where, str) and dns.inet.is_address(where):
nameservers.append(dns.nameserver.Do53Nameserver(where, port))
else:
answers = await resolver.resolve_name(where, family)
for address in answers.addresses():
nameservers.append(dns.nameserver.Do53Nameserver(address, port))
res = Resolver(configure=False)
res.nameservers = nameservers
return res
async def resolve_at(
where: dns.name.Name | str,
qname: dns.name.Name | str,
rdtype: dns.rdatatype.RdataType | str = dns.rdatatype.A,
rdclass: dns.rdataclass.RdataClass | str = dns.rdataclass.IN,
tcp: bool = False,
source: str | None = None,
raise_on_no_answer: bool = True,
source_port: int = 0,
lifetime: float | None = None,
search: bool | None = None,
backend: dns.asyncbackend.Backend | None = None,
port: int = 53,
family: int = socket.AF_UNSPEC,
resolver: Resolver | None = None,
) -> dns.resolver.Answer:
"""Query nameservers to find the answer to the question.
This is a convenience function that calls ``dns.asyncresolver.make_resolver_at()``
to make a resolver, and then uses it to resolve the query.
See ``dns.asyncresolver.Resolver.resolve`` for more information on the resolution
parameters, and ``dns.asyncresolver.make_resolver_at`` for information about the
resolver parameters *where*, *port*, *family*, and *resolver*.
If making more than one query, it is more efficient to call
``dns.asyncresolver.make_resolver_at()`` and then use that resolver for the queries
instead of calling ``resolve_at()`` multiple times.
"""
res = await make_resolver_at(where, port, family, resolver)
return await res.resolve(
qname,
rdtype,
rdclass,
tcp,
source,
raise_on_no_answer,
source_port,
lifetime,
search,
backend,
)

850
venv/Lib/site-packages/dns/btree.py Normal file
View File

@@ -0,0 +1,850 @@
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
"""
A BTree in the style of Cormen, Leiserson, and Rivest's "Algorithms" book, with
copy-on-write node updates, cursors, and optional space optimization for mostly-in-order
insertion.
"""
from collections.abc import MutableMapping, MutableSet
from typing import Any, Callable, Generic, Optional, Tuple, TypeVar, cast
DEFAULT_T = 127
KT = TypeVar("KT") # the type of a key in Element
class Element(Generic[KT]):
"""All items stored in the BTree are Elements."""
def key(self) -> KT:
"""The key for this element; the returned type must implement comparison."""
raise NotImplementedError # pragma: no cover
ET = TypeVar("ET", bound=Element) # the type of a value in a _KV
def _MIN(t: int) -> int:
"""The minimum number of keys in a non-root node for a BTree with the specified
``t``
"""
return t - 1
def _MAX(t: int) -> int:
"""The maximum number of keys in node for a BTree with the specified ``t``"""
return 2 * t - 1
class _Creator:
"""A _Creator class instance is used as a unique id for the BTree which created
a node.
We use a dedicated creator rather than just a BTree reference to avoid circularity
that would complicate GC.
"""
def __str__(self): # pragma: no cover
return f"{id(self):x}"
class _Node(Generic[KT, ET]):
"""A Node in the BTree.
A Node (leaf or internal) of the BTree.
"""
__slots__ = ["t", "creator", "is_leaf", "elts", "children"]
def __init__(self, t: int, creator: _Creator, is_leaf: bool):
assert t >= 3
self.t = t
self.creator = creator
self.is_leaf = is_leaf
self.elts: list[ET] = []
self.children: list[_Node[KT, ET]] = []
def is_maximal(self) -> bool:
"""Does this node have the maximal number of keys?"""
assert len(self.elts) <= _MAX(self.t)
return len(self.elts) == _MAX(self.t)
def is_minimal(self) -> bool:
"""Does this node have the minimal number of keys?"""
assert len(self.elts) >= _MIN(self.t)
return len(self.elts) == _MIN(self.t)
def search_in_node(self, key: KT) -> tuple[int, bool]:
"""Get the index of the ``Element`` matching ``key`` or the index of its
least successor.
Returns a tuple of the index and an ``equal`` boolean that is ``True`` iff.
the key was found.
"""
l = len(self.elts)
if l > 0 and key > self.elts[l - 1].key():
# This is optimizing near in-order insertion.
return l, False
l = 0
i = len(self.elts)
r = i - 1
equal = False
while l <= r:
m = (l + r) // 2
k = self.elts[m].key()
if key == k:
i = m
equal = True
break
elif key < k:
i = m
r = m - 1
else:
l = m + 1
return i, equal
def maybe_cow_child(self, index: int) -> "_Node[KT, ET]":
assert not self.is_leaf
child = self.children[index]
cloned = child.maybe_cow(self.creator)
if cloned:
self.children[index] = cloned
return cloned
else:
return child
def _get_node(self, key: KT) -> Tuple[Optional["_Node[KT, ET]"], int]:
"""Get the node associated with key and its index, doing
copy-on-write if we have to descend.
Returns a tuple of the node and the index, or the tuple ``(None, 0)``
if the key was not found.
"""
i, equal = self.search_in_node(key)
if equal:
return (self, i)
elif self.is_leaf:
return (None, 0)
else:
child = self.maybe_cow_child(i)
return child._get_node(key)
def get(self, key: KT) -> ET | None:
"""Get the element associated with *key* or return ``None``"""
i, equal = self.search_in_node(key)
if equal:
return self.elts[i]
elif self.is_leaf:
return None
else:
return self.children[i].get(key)
def optimize_in_order_insertion(self, index: int) -> None:
"""Try to minimize the number of Nodes in a BTree where the insertion
is done in-order or close to it, by stealing as much as we can from our
right sibling.
If we don't do this, then an in-order insertion will produce a BTree
where most of the nodes are minimal.
"""
if index == 0:
return
left = self.children[index - 1]
if len(left.elts) == _MAX(self.t):
return
left = self.maybe_cow_child(index - 1)
while len(left.elts) < _MAX(self.t):
if not left.try_right_steal(self, index - 1):
break
def insert_nonfull(self, element: ET, in_order: bool) -> ET | None:
assert not self.is_maximal()
while True:
key = element.key()
i, equal = self.search_in_node(key)
if equal:
# replace
old = self.elts[i]
self.elts[i] = element
return old
elif self.is_leaf:
self.elts.insert(i, element)
return None
else:
child = self.maybe_cow_child(i)
if child.is_maximal():
self.adopt(*child.split())
# Splitting might result in our target moving to us, so
# search again.
continue
oelt = child.insert_nonfull(element, in_order)
if in_order:
self.optimize_in_order_insertion(i)
return oelt
def split(self) -> tuple["_Node[KT, ET]", ET, "_Node[KT, ET]"]:
"""Split a maximal node into two minimal ones and a central element."""
assert self.is_maximal()
right = self.__class__(self.t, self.creator, self.is_leaf)
right.elts = list(self.elts[_MIN(self.t) + 1 :])
middle = self.elts[_MIN(self.t)]
self.elts = list(self.elts[: _MIN(self.t)])
if not self.is_leaf:
right.children = list(self.children[_MIN(self.t) + 1 :])
self.children = list(self.children[: _MIN(self.t) + 1])
return self, middle, right
def try_left_steal(self, parent: "_Node[KT, ET]", index: int) -> bool:
"""Try to steal from this Node's left sibling for balancing purposes.
Returns ``True`` if the theft was successful, or ``False`` if not.
"""
if index != 0:
left = parent.children[index - 1]
if not left.is_minimal():
left = parent.maybe_cow_child(index - 1)
elt = parent.elts[index - 1]
parent.elts[index - 1] = left.elts.pop()
self.elts.insert(0, elt)
if not left.is_leaf:
assert not self.is_leaf
child = left.children.pop()
self.children.insert(0, child)
return True
return False
def try_right_steal(self, parent: "_Node[KT, ET]", index: int) -> bool:
"""Try to steal from this Node's right sibling for balancing purposes.
Returns ``True`` if the theft was successful, or ``False`` if not.
"""
if index + 1 < len(parent.children):
right = parent.children[index + 1]
if not right.is_minimal():
right = parent.maybe_cow_child(index + 1)
elt = parent.elts[index]
parent.elts[index] = right.elts.pop(0)
self.elts.append(elt)
if not right.is_leaf:
assert not self.is_leaf
child = right.children.pop(0)
self.children.append(child)
return True
return False
def adopt(self, left: "_Node[KT, ET]", middle: ET, right: "_Node[KT, ET]") -> None:
"""Adopt left, middle, and right into our Node (which must not be maximal,
and which must not be a leaf). In the case were we are not the new root,
then the left child must already be in the Node."""
assert not self.is_maximal()
assert not self.is_leaf
key = middle.key()
i, equal = self.search_in_node(key)
assert not equal
self.elts.insert(i, middle)
if len(self.children) == 0:
# We are the new root
self.children = [left, right]
else:
assert self.children[i] == left
self.children.insert(i + 1, right)
def merge(self, parent: "_Node[KT, ET]", index: int) -> None:
"""Merge this node's parent and its right sibling into this node."""
right = parent.children.pop(index + 1)
self.elts.append(parent.elts.pop(index))
self.elts.extend(right.elts)
if not self.is_leaf:
self.children.extend(right.children)
def minimum(self) -> ET:
"""The least element in this subtree."""
if self.is_leaf:
return self.elts[0]
else:
return self.children[0].minimum()
def maximum(self) -> ET:
"""The greatest element in this subtree."""
if self.is_leaf:
return self.elts[-1]
else:
return self.children[-1].maximum()
def balance(self, parent: "_Node[KT, ET]", index: int) -> None:
"""This Node is minimal, and we want to make it non-minimal so we can delete.
We try to steal from our siblings, and if that doesn't work we will merge
with one of them."""
assert not parent.is_leaf
if self.try_left_steal(parent, index):
return
if self.try_right_steal(parent, index):
return
# Stealing didn't work, so both siblings must be minimal.
if index == 0:
# We are the left-most node so merge with our right sibling.
self.merge(parent, index)
else:
# Have our left sibling merge with us. This lets us only have "merge right"
# code.
left = parent.maybe_cow_child(index - 1)
left.merge(parent, index - 1)
def delete(
self, key: KT, parent: Optional["_Node[KT, ET]"], exact: ET | None
) -> ET | None:
"""Delete an element matching *key* if it exists. If *exact* is not ``None``
then it must be an exact match with that element. The Node must not be
minimal unless it is the root."""
assert parent is None or not self.is_minimal()
i, equal = self.search_in_node(key)
original_key = None
if equal:
# Note we use "is" here as we meant "exactly this object".
if exact is not None and self.elts[i] is not exact:
raise ValueError("exact delete did not match existing elt")
if self.is_leaf:
return self.elts.pop(i)
# Note we need to ensure exact is None going forward as we've
# already checked exactness and are about to change our target key
# to the least successor.
exact = None
original_key = key
least_successor = self.children[i + 1].minimum()
key = least_successor.key()
i = i + 1
if self.is_leaf:
# No match
if exact is not None:
raise ValueError("exact delete had no match")
return None
# recursively delete in the appropriate child
child = self.maybe_cow_child(i)
if child.is_minimal():
child.balance(self, i)
# Things may have moved.
i, equal = self.search_in_node(key)
assert not equal
child = self.children[i]
assert not child.is_minimal()
elt = child.delete(key, self, exact)
if original_key is not None:
node, i = self._get_node(original_key)
assert node is not None
assert elt is not None
oelt = node.elts[i]
node.elts[i] = elt
elt = oelt
return elt
def visit_in_order(self, visit: Callable[[ET], None]) -> None:
"""Call *visit* on all of the elements in order."""
for i, elt in enumerate(self.elts):
if not self.is_leaf:
self.children[i].visit_in_order(visit)
visit(elt)
if not self.is_leaf:
self.children[-1].visit_in_order(visit)
def _visit_preorder_by_node(self, visit: Callable[["_Node[KT, ET]"], None]) -> None:
"""Visit nodes in preorder. This method is only used for testing."""
visit(self)
if not self.is_leaf:
for child in self.children:
child._visit_preorder_by_node(visit)
def maybe_cow(self, creator: _Creator) -> Optional["_Node[KT, ET]"]:
"""Return a clone of this Node if it was not created by *creator*, or ``None``
otherwise (i.e. copy for copy-on-write if we haven't already copied it)."""
if self.creator is not creator:
return self.clone(creator)
else:
return None
def clone(self, creator: _Creator) -> "_Node[KT, ET]":
"""Make a shallow-copy duplicate of this node."""
cloned = self.__class__(self.t, creator, self.is_leaf)
cloned.elts.extend(self.elts)
if not self.is_leaf:
cloned.children.extend(self.children)
return cloned
def __str__(self): # pragma: no cover
if not self.is_leaf:
children = " " + " ".join([f"{id(c):x}" for c in self.children])
else:
children = ""
return f"{id(self):x} {self.creator} {self.elts}{children}"
class Cursor(Generic[KT, ET]):
"""A seekable cursor for a BTree.
If you are going to use a cursor on a mutable BTree, you should use it
in a ``with`` block so that any mutations of the BTree automatically park
the cursor.
"""
def __init__(self, btree: "BTree[KT, ET]"):
self.btree = btree
self.current_node: _Node | None = None
# The current index is the element index within the current node, or
# if there is no current node then it is 0 on the left boundary and 1
# on the right boundary.
self.current_index: int = 0
self.recurse = False
self.increasing = True
self.parents: list[tuple[_Node, int]] = []
self.parked = False
self.parking_key: KT | None = None
self.parking_key_read = False
def _seek_least(self) -> None:
# seek to the least value in the subtree beneath the current index of the
# current node
assert self.current_node is not None
while not self.current_node.is_leaf:
self.parents.append((self.current_node, self.current_index))
self.current_node = self.current_node.children[self.current_index]
assert self.current_node is not None
self.current_index = 0
def _seek_greatest(self) -> None:
# seek to the greatest value in the subtree beneath the current index of the
# current node
assert self.current_node is not None
while not self.current_node.is_leaf:
self.parents.append((self.current_node, self.current_index))
self.current_node = self.current_node.children[self.current_index]
assert self.current_node is not None
self.current_index = len(self.current_node.elts)
def park(self):
"""Park the cursor.
A cursor must be "parked" before mutating the BTree to avoid undefined behavior.
Cursors created in a ``with`` block register with their BTree and will park
automatically. Note that a parked cursor may not observe some changes made when
it is parked; for example a cursor being iterated with next() will not see items
inserted before its current position.
"""
if not self.parked:
self.parked = True
def _maybe_unpark(self):
if self.parked:
if self.parking_key is not None:
# remember our increasing hint, as seeking might change it
increasing = self.increasing
if self.parking_key_read:
# We've already returned the parking key, so we want to be before it
# if decreasing and after it if increasing.
before = not self.increasing
else:
# We haven't returned the parking key, so we've parked right
# after seeking or are on a boundary. Either way, the before
# hint we want is the value of self.increasing.
before = self.increasing
self.seek(self.parking_key, before)
self.increasing = increasing # might have been altered by seek()
self.parked = False
self.parking_key = None
def prev(self) -> ET | None:
"""Get the previous element, or return None if on the left boundary."""
self._maybe_unpark()
self.parking_key = None
if self.current_node is None:
# on a boundary
if self.current_index == 0:
# left boundary, there is no prev
return None
else:
assert self.current_index == 1
# right boundary; seek to the actual boundary
# so we can do a prev()
self.current_node = self.btree.root
self.current_index = len(self.btree.root.elts)
self._seek_greatest()
while True:
if self.recurse:
if not self.increasing:
# We only want to recurse if we are continuing in the decreasing
# direction.
self._seek_greatest()
self.recurse = False
self.increasing = False
self.current_index -= 1
if self.current_index >= 0:
elt = self.current_node.elts[self.current_index]
if not self.current_node.is_leaf:
self.recurse = True
self.parking_key = elt.key()
self.parking_key_read = True
return elt
else:
if len(self.parents) > 0:
self.current_node, self.current_index = self.parents.pop()
else:
self.current_node = None
self.current_index = 0
return None
def next(self) -> ET | None:
"""Get the next element, or return None if on the right boundary."""
self._maybe_unpark()
self.parking_key = None
if self.current_node is None:
# on a boundary
if self.current_index == 1:
# right boundary, there is no next
return None
else:
assert self.current_index == 0
# left boundary; seek to the actual boundary
# so we can do a next()
self.current_node = self.btree.root
self.current_index = 0
self._seek_least()
while True:
if self.recurse:
if self.increasing:
# We only want to recurse if we are continuing in the increasing
# direction.
self._seek_least()
self.recurse = False
self.increasing = True
if self.current_index < len(self.current_node.elts):
elt = self.current_node.elts[self.current_index]
self.current_index += 1
if not self.current_node.is_leaf:
self.recurse = True
self.parking_key = elt.key()
self.parking_key_read = True
return elt
else:
if len(self.parents) > 0:
self.current_node, self.current_index = self.parents.pop()
else:
self.current_node = None
self.current_index = 1
return None
def _adjust_for_before(self, before: bool, i: int) -> None:
if before:
self.current_index = i
else:
self.current_index = i + 1
def seek(self, key: KT, before: bool = True) -> None:
"""Seek to the specified key.
If *before* is ``True`` (the default) then the cursor is positioned just
before *key* if it exists, or before its least successor if it doesn't. A
subsequent next() will retrieve this value. If *before* is ``False``, then
the cursor is positioned just after *key* if it exists, or its greatest
precessessor if it doesn't. A subsequent prev() will return this value.
"""
self.current_node = self.btree.root
assert self.current_node is not None
self.recurse = False
self.parents = []
self.increasing = before
self.parked = False
self.parking_key = key
self.parking_key_read = False
while not self.current_node.is_leaf:
i, equal = self.current_node.search_in_node(key)
if equal:
self._adjust_for_before(before, i)
if before:
self._seek_greatest()
else:
self._seek_least()
return
self.parents.append((self.current_node, i))
self.current_node = self.current_node.children[i]
assert self.current_node is not None
i, equal = self.current_node.search_in_node(key)
if equal:
self._adjust_for_before(before, i)
else:
self.current_index = i
def seek_first(self) -> None:
"""Seek to the left boundary (i.e. just before the least element).
A subsequent next() will return the least element if the BTree isn't empty."""
self.current_node = None
self.current_index = 0
self.recurse = False
self.increasing = True
self.parents = []
self.parked = False
self.parking_key = None
def seek_last(self) -> None:
"""Seek to the right boundary (i.e. just after the greatest element).
A subsequent prev() will return the greatest element if the BTree isn't empty.
"""
self.current_node = None
self.current_index = 1
self.recurse = False
self.increasing = False
self.parents = []
self.parked = False
self.parking_key = None
def __enter__(self):
self.btree.register_cursor(self)
return self
def __exit__(self, exc_type, exc_value, traceback):
self.btree.deregister_cursor(self)
return False
class Immutable(Exception):
"""The BTree is immutable."""
class BTree(Generic[KT, ET]):
"""An in-memory BTree with copy-on-write and cursors."""
def __init__(self, *, t: int = DEFAULT_T, original: Optional["BTree"] = None):
"""Create a BTree.
If *original* is not ``None``, then the BTree is shallow-cloned from
*original* using copy-on-write. Otherwise a new BTree with the specified
*t* value is created.
The BTree is not thread-safe.
"""
# We don't use a reference to ourselves as a creator as we don't want
# to prevent GC of old btrees.
self.creator = _Creator()
self._immutable = False
self.t: int
self.root: _Node
self.size: int
self.cursors: set[Cursor] = set()
if original is not None:
if not original._immutable:
raise ValueError("original BTree is not immutable")
self.t = original.t
self.root = original.root
self.size = original.size
else:
if t < 3:
raise ValueError("t must be >= 3")
self.t = t
self.root = _Node(self.t, self.creator, True)
self.size = 0
def make_immutable(self):
"""Make the BTree immutable.
Attempts to alter the BTree after making it immutable will raise an
Immutable exception. This operation cannot be undone.
"""
if not self._immutable:
self._immutable = True
def _check_mutable_and_park(self) -> None:
if self._immutable:
raise Immutable
for cursor in self.cursors:
cursor.park()
# Note that we don't use insert() and delete() but rather insert_element() and
# delete_key() so that BTreeDict can be a proper MutableMapping and supply the
# rest of the standard mapping API.
def insert_element(self, elt: ET, in_order: bool = False) -> ET | None:
"""Insert the element into the BTree.
If *in_order* is ``True``, then extra work will be done to make left siblings
full, which optimizes storage space when the the elements are inserted in-order
or close to it.
Returns the previously existing element at the element's key or ``None``.
"""
self._check_mutable_and_park()
cloned = self.root.maybe_cow(self.creator)
if cloned:
self.root = cloned
if self.root.is_maximal():
old_root = self.root
self.root = _Node(self.t, self.creator, False)
self.root.adopt(*old_root.split())
oelt = self.root.insert_nonfull(elt, in_order)
if oelt is None:
# We did not replace, so something was added.
self.size += 1
return oelt
def get_element(self, key: KT) -> ET | None:
"""Get the element matching *key* from the BTree, or return ``None`` if it
does not exist.
"""
return self.root.get(key)
def _delete(self, key: KT, exact: ET | None) -> ET | None:
self._check_mutable_and_park()
cloned = self.root.maybe_cow(self.creator)
if cloned:
self.root = cloned
elt = self.root.delete(key, None, exact)
if elt is not None:
# We deleted something
self.size -= 1
if len(self.root.elts) == 0:
# The root is now empty. If there is a child, then collapse this root
# level and make the child the new root.
if not self.root.is_leaf:
assert len(self.root.children) == 1
self.root = self.root.children[0]
return elt
def delete_key(self, key: KT) -> ET | None:
"""Delete the element matching *key* from the BTree.
Returns the matching element or ``None`` if it does not exist.
"""
return self._delete(key, None)
def delete_exact(self, element: ET) -> ET | None:
"""Delete *element* from the BTree.
Returns the matching element or ``None`` if it was not in the BTree.
"""
delt = self._delete(element.key(), element)
assert delt is element
return delt
def __len__(self):
return self.size
def visit_in_order(self, visit: Callable[[ET], None]) -> None:
"""Call *visit*(element) on all elements in the tree in sorted order."""
self.root.visit_in_order(visit)
def _visit_preorder_by_node(self, visit: Callable[[_Node], None]) -> None:
self.root._visit_preorder_by_node(visit)
def cursor(self) -> Cursor[KT, ET]:
"""Create a cursor."""
return Cursor(self)
def register_cursor(self, cursor: Cursor) -> None:
"""Register a cursor for the automatic parking service."""
self.cursors.add(cursor)
def deregister_cursor(self, cursor: Cursor) -> None:
"""Deregister a cursor from the automatic parking service."""
self.cursors.discard(cursor)
def __copy__(self):
return self.__class__(original=self)
def __iter__(self):
with self.cursor() as cursor:
while True:
elt = cursor.next()
if elt is None:
break
yield elt.key()
VT = TypeVar("VT") # the type of a value in a BTreeDict
class KV(Element, Generic[KT, VT]):
"""The BTree element type used in a ``BTreeDict``."""
def __init__(self, key: KT, value: VT):
self._key = key
self._value = value
def key(self) -> KT:
return self._key
def value(self) -> VT:
return self._value
def __str__(self): # pragma: no cover
return f"KV({self._key}, {self._value})"
def __repr__(self): # pragma: no cover
return f"KV({self._key}, {self._value})"
class BTreeDict(Generic[KT, VT], BTree[KT, KV[KT, VT]], MutableMapping[KT, VT]):
"""A MutableMapping implemented with a BTree.
Unlike a normal Python dict, the BTreeDict may be mutated while iterating.
"""
def __init__(
self,
*,
t: int = DEFAULT_T,
original: BTree | None = None,
in_order: bool = False,
):
super().__init__(t=t, original=original)
self.in_order = in_order
def __getitem__(self, key: KT) -> VT:
elt = self.get_element(key)
if elt is None:
raise KeyError
else:
return cast(KV, elt).value()
def __setitem__(self, key: KT, value: VT) -> None:
elt = KV(key, value)
self.insert_element(elt, self.in_order)
def __delitem__(self, key: KT) -> None:
if self.delete_key(key) is None:
raise KeyError
class Member(Element, Generic[KT]):
"""The BTree element type used in a ``BTreeSet``."""
def __init__(self, key: KT):
self._key = key
def key(self) -> KT:
return self._key
class BTreeSet(BTree, Generic[KT], MutableSet[KT]):
"""A MutableSet implemented with a BTree.
Unlike a normal Python set, the BTreeSet may be mutated while iterating.
"""
def __init__(
self,
*,
t: int = DEFAULT_T,
original: BTree | None = None,
in_order: bool = False,
):
super().__init__(t=t, original=original)
self.in_order = in_order
def __contains__(self, key: Any) -> bool:
return self.get_element(key) is not None
def add(self, value: KT) -> None:
elt = Member(value)
self.insert_element(elt, self.in_order)
def discard(self, value: KT) -> None:
self.delete_key(value)

367
venv/Lib/site-packages/dns/btreezone.py Normal file
View File

@@ -0,0 +1,367 @@
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
# A derivative of a dnspython VersionedZone and related classes, using a BTreeDict and
# a separate per-version delegation index. These additions let us
#
# 1) Do efficient CoW versioning (useful for future online updates).
# 2) Maintain sort order
# 3) Allow delegations to be found easily
# 4) Handle glue
# 5) Add Node flags ORIGIN, DELEGATION, and GLUE whenever relevant. The ORIGIN
# flag is set at the origin node, the DELEGATION FLAG is set at delegation
# points, and the GLUE flag is set on nodes beneath delegation points.
import enum
from dataclasses import dataclass
from typing import Callable, MutableMapping, Tuple, cast
import dns.btree
import dns.immutable
import dns.name
import dns.node
import dns.rdataclass
import dns.rdataset
import dns.rdatatype
import dns.versioned
import dns.zone
class NodeFlags(enum.IntFlag):
ORIGIN = 0x01
DELEGATION = 0x02
GLUE = 0x04
class Node(dns.node.Node):
__slots__ = ["flags", "id"]
def __init__(self, flags: NodeFlags | None = None):
super().__init__()
if flags is None:
# We allow optional flags rather than a default
# as pyright doesn't like assigning a literal 0
# to flags.
flags = NodeFlags(0)
self.flags = flags
self.id = 0
def is_delegation(self):
return (self.flags & NodeFlags.DELEGATION) != 0
def is_glue(self):
return (self.flags & NodeFlags.GLUE) != 0
def is_origin(self):
return (self.flags & NodeFlags.ORIGIN) != 0
def is_origin_or_glue(self):
return (self.flags & (NodeFlags.ORIGIN | NodeFlags.GLUE)) != 0
@dns.immutable.immutable
class ImmutableNode(Node):
def __init__(self, node: Node):
super().__init__()
self.id = node.id
self.rdatasets = tuple( # type: ignore
[dns.rdataset.ImmutableRdataset(rds) for rds in node.rdatasets]
)
self.flags = node.flags
def find_rdataset(
self,
rdclass: dns.rdataclass.RdataClass,
rdtype: dns.rdatatype.RdataType,
covers: dns.rdatatype.RdataType = dns.rdatatype.NONE,
create: bool = False,
) -> dns.rdataset.Rdataset:
if create:
raise TypeError("immutable")
return super().find_rdataset(rdclass, rdtype, covers, False)
def get_rdataset(
self,
rdclass: dns.rdataclass.RdataClass,
rdtype: dns.rdatatype.RdataType,
covers: dns.rdatatype.RdataType = dns.rdatatype.NONE,
create: bool = False,
) -> dns.rdataset.Rdataset | None:
if create:
raise TypeError("immutable")
return super().get_rdataset(rdclass, rdtype, covers, False)
def delete_rdataset(
self,
rdclass: dns.rdataclass.RdataClass,
rdtype: dns.rdatatype.RdataType,
covers: dns.rdatatype.RdataType = dns.rdatatype.NONE,
) -> None:
raise TypeError("immutable")
def replace_rdataset(self, replacement: dns.rdataset.Rdataset) -> None:
raise TypeError("immutable")
def is_immutable(self) -> bool:
return True
class Delegations(dns.btree.BTreeSet[dns.name.Name]):
def get_delegation(self, name: dns.name.Name) -> Tuple[dns.name.Name | None, bool]:
"""Get the delegation applicable to *name*, if it exists.
If there delegation, then return a tuple consisting of the name of
the delegation point, and a boolean which is `True` if the name is a proper
subdomain of the delegation point, and `False` if it is equal to the delegation
point.
"""
cursor = self.cursor()
cursor.seek(name, before=False)
prev = cursor.prev()
if prev is None:
return None, False
cut = prev.key()
reln, _, _ = name.fullcompare(cut)
is_subdomain = reln == dns.name.NameRelation.SUBDOMAIN
if is_subdomain or reln == dns.name.NameRelation.EQUAL:
return cut, is_subdomain
else:
return None, False
def is_glue(self, name: dns.name.Name) -> bool:
"""Is *name* glue, i.e. is it beneath a delegation?"""
cursor = self.cursor()
cursor.seek(name, before=False)
cut, is_subdomain = self.get_delegation(name)
if cut is None:
return False
return is_subdomain
class WritableVersion(dns.zone.WritableVersion):
def __init__(self, zone: dns.zone.Zone, replacement: bool = False):
super().__init__(zone, True)
if not replacement:
assert isinstance(zone, dns.versioned.Zone)
version = zone._versions[-1]
self.nodes: dns.btree.BTreeDict[dns.name.Name, Node] = dns.btree.BTreeDict(
original=version.nodes # type: ignore
)
self.delegations = Delegations(original=version.delegations) # type: ignore
else:
self.delegations = Delegations()
def _is_origin(self, name: dns.name.Name) -> bool:
# Assumes name has already been validated (and thus adjusted to the right
# relativity too)
if self.zone.relativize:
return name == dns.name.empty
else:
return name == self.zone.origin
def _maybe_cow_with_name(
self, name: dns.name.Name
) -> Tuple[dns.node.Node, dns.name.Name]:
(node, name) = super()._maybe_cow_with_name(name)
node = cast(Node, node)
if self._is_origin(name):
node.flags |= NodeFlags.ORIGIN
elif self.delegations.is_glue(name):
node.flags |= NodeFlags.GLUE
return (node, name)
def update_glue_flag(self, name: dns.name.Name, is_glue: bool) -> None:
cursor = self.nodes.cursor() # type: ignore
cursor.seek(name, False)
updates = []
while True:
elt = cursor.next()
if elt is None:
break
ename = elt.key()
if not ename.is_subdomain(name):
break
node = cast(dns.node.Node, elt.value())
if ename not in self.changed:
new_node = self.zone.node_factory()
new_node.id = self.id # type: ignore
new_node.rdatasets.extend(node.rdatasets)
self.changed.add(ename)
node = new_node
assert isinstance(node, Node)
if is_glue:
node.flags |= NodeFlags.GLUE
else:
node.flags &= ~NodeFlags.GLUE
# We don't update node here as any insertion could disturb the
# btree and invalidate our cursor. We could use the cursor in a
# with block and avoid this, but it would do a lot of parking and
# unparking so the deferred update mode may still be better.
updates.append((ename, node))
for ename, node in updates:
self.nodes[ename] = node
def delete_node(self, name: dns.name.Name) -> None:
name = self._validate_name(name)
node = self.nodes.get(name)
if node is not None:
if node.is_delegation(): # type: ignore
self.delegations.discard(name)
self.update_glue_flag(name, False)
del self.nodes[name]
self.changed.add(name)
def put_rdataset(
self, name: dns.name.Name, rdataset: dns.rdataset.Rdataset
) -> None:
(node, name) = self._maybe_cow_with_name(name)
if (
rdataset.rdtype == dns.rdatatype.NS and not node.is_origin_or_glue() # type: ignore
):
node.flags |= NodeFlags.DELEGATION # type: ignore
if name not in self.delegations:
self.delegations.add(name)
self.update_glue_flag(name, True)
node.replace_rdataset(rdataset)
def delete_rdataset(
self,
name: dns.name.Name,
rdtype: dns.rdatatype.RdataType,
covers: dns.rdatatype.RdataType,
) -> None:
(node, name) = self._maybe_cow_with_name(name)
if rdtype == dns.rdatatype.NS and name in self.delegations: # type: ignore
node.flags &= ~NodeFlags.DELEGATION # type: ignore
self.delegations.discard(name) # type: ignore
self.update_glue_flag(name, False)
node.delete_rdataset(self.zone.rdclass, rdtype, covers)
if len(node) == 0:
del self.nodes[name]
@dataclass(frozen=True)
class Bounds:
name: dns.name.Name
left: dns.name.Name
right: dns.name.Name | None
closest_encloser: dns.name.Name
is_equal: bool
is_delegation: bool
def __str__(self):
if self.is_equal:
op = "="
else:
op = "<"
if self.is_delegation:
zonecut = " zonecut"
else:
zonecut = ""
return (
f"{self.left} {op} {self.name} < {self.right}{zonecut}; "
f"{self.closest_encloser}"
)
@dns.immutable.immutable
class ImmutableVersion(dns.zone.Version):
def __init__(self, version: dns.zone.Version):
if not isinstance(version, WritableVersion):
raise ValueError(
"a dns.btreezone.ImmutableVersion requires a "
"dns.btreezone.WritableVersion"
)
super().__init__(version.zone, True)
self.id = version.id
self.origin = version.origin
for name in version.changed:
node = version.nodes.get(name)
if node:
version.nodes[name] = ImmutableNode(node)
# the cast below is for mypy
self.nodes = cast(MutableMapping[dns.name.Name, dns.node.Node], version.nodes)
self.nodes.make_immutable() # type: ignore
self.delegations = version.delegations
self.delegations.make_immutable()
def bounds(self, name: dns.name.Name | str) -> Bounds:
"""Return the 'bounds' of *name* in its zone.
The bounds information is useful when making an authoritative response, as
it can be used to determine whether the query name is at or beneath a delegation
point. The other data in the ``Bounds`` object is useful for making on-the-fly
DNSSEC signatures.
The left bound of *name* is *name* itself if it is in the zone, or the greatest
predecessor which is in the zone.
The right bound of *name* is the least successor of *name*, or ``None`` if
no name in the zone is greater than *name*.
The closest encloser of *name* is *name* itself, if *name* is in the zone;
otherwise it is the name with the largest number of labels in common with
*name* that is in the zone, either explicitly or by the implied existence
of empty non-terminals.
The bounds *is_equal* field is ``True`` if and only if *name* is equal to
its left bound.
The bounds *is_delegation* field is ``True`` if and only if the left bound is a
delegation point.
"""
assert self.origin is not None
# validate the origin because we may need to relativize
origin = self.zone._validate_name(self.origin)
name = self.zone._validate_name(name)
cut, _ = self.delegations.get_delegation(name)
if cut is not None:
target = cut
is_delegation = True
else:
target = name
is_delegation = False
c = cast(dns.btree.BTreeDict, self.nodes).cursor()
c.seek(target, False)
left = c.prev()
assert left is not None
c.next() # skip over left
while True:
right = c.next()
if right is None or not right.value().is_glue():
break
left_comparison = left.key().fullcompare(name)
if right is not None:
right_key = right.key()
right_comparison = right_key.fullcompare(name)
else:
right_comparison = (
dns.name.NAMERELN_COMMONANCESTOR,
-1,
len(origin),
)
right_key = None
closest_encloser = dns.name.Name(
name[-max(left_comparison[2], right_comparison[2]) :]
)
return Bounds(
name,
left.key(),
right_key,
closest_encloser,
left_comparison[0] == dns.name.NameRelation.EQUAL,
is_delegation,
)
class Zone(dns.versioned.Zone):
node_factory: Callable[[], dns.node.Node] = Node
map_factory: Callable[[], MutableMapping[dns.name.Name, dns.node.Node]] = cast(
Callable[[], MutableMapping[dns.name.Name, dns.node.Node]],
dns.btree.BTreeDict[dns.name.Name, Node],
)
writable_version_factory: (
Callable[[dns.zone.Zone, bool], dns.zone.Version] | None
) = WritableVersion
immutable_version_factory: Callable[[dns.zone.Version], dns.zone.Version] | None = (
ImmutableVersion
)

1242
venv/Lib/site-packages/dns/dnssec.py Normal file
View File

File diff suppressed because it is too large Load Diff