Загрузить файлы в «venv/Lib/site-packages/dns»
This commit is contained in:
169
venv/Lib/site-packages/dns/exception.py
Normal file
169
venv/Lib/site-packages/dns/exception.py
Normal file
@@ -0,0 +1,169 @@
|
|||||||
|
# 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.
|
||||||
|
|
||||||
|
"""Common DNS Exceptions.
|
||||||
|
|
||||||
|
Dnspython modules may also define their own exceptions, which will
|
||||||
|
always be subclasses of ``DNSException``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
from typing import Set
|
||||||
|
|
||||||
|
|
||||||
|
class DNSException(Exception):
|
||||||
|
"""Abstract base class shared by all dnspython exceptions.
|
||||||
|
|
||||||
|
It supports two basic modes of operation:
|
||||||
|
|
||||||
|
a) Old/compatible mode is used if ``__init__`` was called with
|
||||||
|
empty *kwargs*. In compatible mode all *args* are passed
|
||||||
|
to the standard Python Exception class as before and all *args* are
|
||||||
|
printed by the standard ``__str__`` implementation. Class variable
|
||||||
|
``msg`` (or doc string if ``msg`` is ``None``) is returned from ``str()``
|
||||||
|
if *args* is empty.
|
||||||
|
|
||||||
|
b) New/parametrized mode is used if ``__init__`` was called with
|
||||||
|
non-empty *kwargs*.
|
||||||
|
In the new mode *args* must be empty and all kwargs must match
|
||||||
|
those set in class variable ``supp_kwargs``. All kwargs are stored inside
|
||||||
|
``self.kwargs`` and used in a new ``__str__`` implementation to construct
|
||||||
|
a formatted message based on the ``fmt`` class variable, a ``string``.
|
||||||
|
|
||||||
|
In the simplest case it is enough to override the ``supp_kwargs``
|
||||||
|
and ``fmt`` class variables to get nice parametrized messages.
|
||||||
|
"""
|
||||||
|
|
||||||
|
msg: str | None = None # non-parametrized message
|
||||||
|
supp_kwargs: Set[str] = set() # accepted parameters for _fmt_kwargs (sanity check)
|
||||||
|
fmt: str | None = None # message parametrized with results from _fmt_kwargs
|
||||||
|
|
||||||
|
def __init__(self, *args, **kwargs):
|
||||||
|
self._check_params(*args, **kwargs)
|
||||||
|
if kwargs:
|
||||||
|
# This call to a virtual method from __init__ is ok in our usage
|
||||||
|
self.kwargs = self._check_kwargs(**kwargs) # lgtm[py/init-calls-subclass]
|
||||||
|
self.msg = str(self)
|
||||||
|
else:
|
||||||
|
self.kwargs = dict() # defined but empty for old mode exceptions
|
||||||
|
if self.msg is None:
|
||||||
|
# doc string is better implicit message than empty string
|
||||||
|
self.msg = self.__doc__
|
||||||
|
if args:
|
||||||
|
super().__init__(*args)
|
||||||
|
else:
|
||||||
|
super().__init__(self.msg)
|
||||||
|
|
||||||
|
def _check_params(self, *args, **kwargs):
|
||||||
|
"""Old exceptions supported only args and not kwargs.
|
||||||
|
|
||||||
|
For sanity we do not allow to mix old and new behavior."""
|
||||||
|
if args or kwargs:
|
||||||
|
assert bool(args) != bool(
|
||||||
|
kwargs
|
||||||
|
), "keyword arguments are mutually exclusive with positional args"
|
||||||
|
|
||||||
|
def _check_kwargs(self, **kwargs):
|
||||||
|
if kwargs:
|
||||||
|
assert (
|
||||||
|
set(kwargs.keys()) == self.supp_kwargs
|
||||||
|
), f"following set of keyword args is required: {self.supp_kwargs}"
|
||||||
|
return kwargs
|
||||||
|
|
||||||
|
def _fmt_kwargs(self, **kwargs):
|
||||||
|
"""Format kwargs before printing them.
|
||||||
|
|
||||||
|
Resulting dictionary has to have keys necessary for str.format call
|
||||||
|
on fmt class variable.
|
||||||
|
"""
|
||||||
|
fmtargs = {}
|
||||||
|
for kw, data in kwargs.items():
|
||||||
|
if isinstance(data, list | set):
|
||||||
|
# convert list of <someobj> to list of str(<someobj>)
|
||||||
|
fmtargs[kw] = list(map(str, data))
|
||||||
|
if len(fmtargs[kw]) == 1:
|
||||||
|
# remove list brackets [] from single-item lists
|
||||||
|
fmtargs[kw] = fmtargs[kw].pop()
|
||||||
|
else:
|
||||||
|
fmtargs[kw] = data
|
||||||
|
return fmtargs
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
if self.kwargs and self.fmt:
|
||||||
|
# provide custom message constructed from keyword arguments
|
||||||
|
fmtargs = self._fmt_kwargs(**self.kwargs)
|
||||||
|
return self.fmt.format(**fmtargs)
|
||||||
|
else:
|
||||||
|
# print *args directly in the same way as old DNSException
|
||||||
|
return super().__str__()
|
||||||
|
|
||||||
|
|
||||||
|
class FormError(DNSException):
|
||||||
|
"""DNS message is malformed."""
|
||||||
|
|
||||||
|
|
||||||
|
class SyntaxError(DNSException):
|
||||||
|
"""Text input is malformed."""
|
||||||
|
|
||||||
|
|
||||||
|
class UnexpectedEnd(SyntaxError):
|
||||||
|
"""Text input ended unexpectedly."""
|
||||||
|
|
||||||
|
|
||||||
|
class TooBig(DNSException):
|
||||||
|
"""The DNS message is too big."""
|
||||||
|
|
||||||
|
|
||||||
|
class Timeout(DNSException):
|
||||||
|
"""The DNS operation timed out."""
|
||||||
|
|
||||||
|
supp_kwargs = {"timeout"}
|
||||||
|
fmt = "The DNS operation timed out after {timeout:.3f} seconds"
|
||||||
|
|
||||||
|
# We do this as otherwise mypy complains about unexpected keyword argument
|
||||||
|
# idna_exception
|
||||||
|
def __init__(self, *args, **kwargs):
|
||||||
|
super().__init__(*args, **kwargs)
|
||||||
|
|
||||||
|
|
||||||
|
class UnsupportedAlgorithm(DNSException):
|
||||||
|
"""The DNSSEC algorithm is not supported."""
|
||||||
|
|
||||||
|
|
||||||
|
class AlgorithmKeyMismatch(UnsupportedAlgorithm):
|
||||||
|
"""The DNSSEC algorithm is not supported for the given key type."""
|
||||||
|
|
||||||
|
|
||||||
|
class ValidationFailure(DNSException):
|
||||||
|
"""The DNSSEC signature is invalid."""
|
||||||
|
|
||||||
|
|
||||||
|
class DeniedByPolicy(DNSException):
|
||||||
|
"""Denied by DNSSEC policy."""
|
||||||
|
|
||||||
|
|
||||||
|
class ExceptionWrapper:
|
||||||
|
def __init__(self, exception_class):
|
||||||
|
self.exception_class = exception_class
|
||||||
|
|
||||||
|
def __enter__(self):
|
||||||
|
return self
|
||||||
|
|
||||||
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||||
|
if exc_type is not None and not isinstance(exc_val, self.exception_class):
|
||||||
|
raise self.exception_class(str(exc_val)) from exc_val
|
||||||
|
return False
|
||||||
123
venv/Lib/site-packages/dns/flags.py
Normal file
123
venv/Lib/site-packages/dns/flags.py
Normal file
@@ -0,0 +1,123 @@
|
|||||||
|
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
|
||||||
|
|
||||||
|
# Copyright (C) 2001-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.
|
||||||
|
|
||||||
|
"""DNS Message Flags."""
|
||||||
|
|
||||||
|
import enum
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
# Standard DNS flags
|
||||||
|
|
||||||
|
|
||||||
|
class Flag(enum.IntFlag):
|
||||||
|
#: Query Response
|
||||||
|
QR = 0x8000
|
||||||
|
#: Authoritative Answer
|
||||||
|
AA = 0x0400
|
||||||
|
#: Truncated Response
|
||||||
|
TC = 0x0200
|
||||||
|
#: Recursion Desired
|
||||||
|
RD = 0x0100
|
||||||
|
#: Recursion Available
|
||||||
|
RA = 0x0080
|
||||||
|
#: Authentic Data
|
||||||
|
AD = 0x0020
|
||||||
|
#: Checking Disabled
|
||||||
|
CD = 0x0010
|
||||||
|
|
||||||
|
|
||||||
|
# EDNS flags
|
||||||
|
|
||||||
|
|
||||||
|
class EDNSFlag(enum.IntFlag):
|
||||||
|
#: DNSSEC answer OK
|
||||||
|
DO = 0x8000
|
||||||
|
|
||||||
|
|
||||||
|
def _from_text(text: str, enum_class: Any) -> int:
|
||||||
|
flags = 0
|
||||||
|
tokens = text.split()
|
||||||
|
for t in tokens:
|
||||||
|
flags |= enum_class[t.upper()]
|
||||||
|
return flags
|
||||||
|
|
||||||
|
|
||||||
|
def _to_text(flags: int, enum_class: Any) -> str:
|
||||||
|
text_flags = []
|
||||||
|
for k, v in enum_class.__members__.items():
|
||||||
|
if flags & v != 0:
|
||||||
|
text_flags.append(k)
|
||||||
|
return " ".join(text_flags)
|
||||||
|
|
||||||
|
|
||||||
|
def from_text(text: str) -> int:
|
||||||
|
"""Convert a space-separated list of flag text values into a flags
|
||||||
|
value.
|
||||||
|
|
||||||
|
Returns an ``int``
|
||||||
|
"""
|
||||||
|
|
||||||
|
return _from_text(text, Flag)
|
||||||
|
|
||||||
|
|
||||||
|
def to_text(flags: int) -> str:
|
||||||
|
"""Convert a flags value into a space-separated list of flag text
|
||||||
|
values.
|
||||||
|
|
||||||
|
Returns a ``str``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
return _to_text(flags, Flag)
|
||||||
|
|
||||||
|
|
||||||
|
def edns_from_text(text: str) -> int:
|
||||||
|
"""Convert a space-separated list of EDNS flag text values into a EDNS
|
||||||
|
flags value.
|
||||||
|
|
||||||
|
Returns an ``int``
|
||||||
|
"""
|
||||||
|
|
||||||
|
return _from_text(text, EDNSFlag)
|
||||||
|
|
||||||
|
|
||||||
|
def edns_to_text(flags: int) -> str:
|
||||||
|
"""Convert an EDNS flags value into a space-separated list of EDNS flag
|
||||||
|
text values.
|
||||||
|
|
||||||
|
Returns a ``str``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
return _to_text(flags, EDNSFlag)
|
||||||
|
|
||||||
|
|
||||||
|
### BEGIN generated Flag constants
|
||||||
|
|
||||||
|
QR = Flag.QR
|
||||||
|
AA = Flag.AA
|
||||||
|
TC = Flag.TC
|
||||||
|
RD = Flag.RD
|
||||||
|
RA = Flag.RA
|
||||||
|
AD = Flag.AD
|
||||||
|
CD = Flag.CD
|
||||||
|
|
||||||
|
### END generated Flag constants
|
||||||
|
|
||||||
|
### BEGIN generated EDNSFlag constants
|
||||||
|
|
||||||
|
DO = EDNSFlag.DO
|
||||||
|
|
||||||
|
### END generated EDNSFlag constants
|
||||||
72
venv/Lib/site-packages/dns/grange.py
Normal file
72
venv/Lib/site-packages/dns/grange.py
Normal file
@@ -0,0 +1,72 @@
|
|||||||
|
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
|
||||||
|
|
||||||
|
# Copyright (C) 2012-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.
|
||||||
|
|
||||||
|
"""DNS GENERATE range conversion."""
|
||||||
|
|
||||||
|
from typing import Tuple
|
||||||
|
|
||||||
|
import dns.exception
|
||||||
|
|
||||||
|
|
||||||
|
def from_text(text: str) -> Tuple[int, int, int]:
|
||||||
|
"""Convert the text form of a range in a ``$GENERATE`` statement to an
|
||||||
|
integer.
|
||||||
|
|
||||||
|
*text*, a ``str``, the textual range in ``$GENERATE`` form.
|
||||||
|
|
||||||
|
Returns a tuple of three ``int`` values ``(start, stop, step)``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
start = -1
|
||||||
|
stop = -1
|
||||||
|
step = 1
|
||||||
|
cur = ""
|
||||||
|
state = 0
|
||||||
|
# state 0 1 2
|
||||||
|
# x - y / z
|
||||||
|
|
||||||
|
if text and text[0] == "-":
|
||||||
|
raise dns.exception.SyntaxError("Start cannot be a negative number")
|
||||||
|
|
||||||
|
for c in text:
|
||||||
|
if c == "-" and state == 0:
|
||||||
|
start = int(cur)
|
||||||
|
cur = ""
|
||||||
|
state = 1
|
||||||
|
elif c == "/":
|
||||||
|
stop = int(cur)
|
||||||
|
cur = ""
|
||||||
|
state = 2
|
||||||
|
elif c.isdigit():
|
||||||
|
cur += c
|
||||||
|
else:
|
||||||
|
raise dns.exception.SyntaxError(f"Could not parse {c}")
|
||||||
|
|
||||||
|
if state == 0:
|
||||||
|
raise dns.exception.SyntaxError("no stop value specified")
|
||||||
|
elif state == 1:
|
||||||
|
stop = int(cur)
|
||||||
|
else:
|
||||||
|
assert state == 2
|
||||||
|
step = int(cur)
|
||||||
|
|
||||||
|
assert step >= 1
|
||||||
|
assert start >= 0
|
||||||
|
if start > stop:
|
||||||
|
raise dns.exception.SyntaxError("start must be <= stop")
|
||||||
|
|
||||||
|
return (start, stop, step)
|
||||||
68
venv/Lib/site-packages/dns/immutable.py
Normal file
68
venv/Lib/site-packages/dns/immutable.py
Normal file
@@ -0,0 +1,68 @@
|
|||||||
|
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
|
||||||
|
|
||||||
|
import collections.abc
|
||||||
|
from typing import Any, Callable
|
||||||
|
|
||||||
|
from dns._immutable_ctx import immutable
|
||||||
|
|
||||||
|
|
||||||
|
@immutable
|
||||||
|
class Dict(collections.abc.Mapping): # lgtm[py/missing-equals]
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
dictionary: Any,
|
||||||
|
no_copy: bool = False,
|
||||||
|
map_factory: Callable[[], collections.abc.MutableMapping] = dict,
|
||||||
|
):
|
||||||
|
"""Make an immutable dictionary from the specified dictionary.
|
||||||
|
|
||||||
|
If *no_copy* is `True`, then *dictionary* will be wrapped instead
|
||||||
|
of copied. Only set this if you are sure there will be no external
|
||||||
|
references to the dictionary.
|
||||||
|
"""
|
||||||
|
if no_copy and isinstance(dictionary, collections.abc.MutableMapping):
|
||||||
|
self._odict = dictionary
|
||||||
|
else:
|
||||||
|
self._odict = map_factory()
|
||||||
|
self._odict.update(dictionary)
|
||||||
|
self._hash = None
|
||||||
|
|
||||||
|
def __getitem__(self, key):
|
||||||
|
return self._odict.__getitem__(key)
|
||||||
|
|
||||||
|
def __hash__(self): # pylint: disable=invalid-hash-returned
|
||||||
|
if self._hash is None:
|
||||||
|
h = 0
|
||||||
|
for key in sorted(self._odict.keys()):
|
||||||
|
h ^= hash(key)
|
||||||
|
object.__setattr__(self, "_hash", h)
|
||||||
|
# this does return an int, but pylint doesn't figure that out
|
||||||
|
return self._hash
|
||||||
|
|
||||||
|
def __len__(self):
|
||||||
|
return len(self._odict)
|
||||||
|
|
||||||
|
def __iter__(self):
|
||||||
|
return iter(self._odict)
|
||||||
|
|
||||||
|
|
||||||
|
def constify(o: Any) -> Any:
|
||||||
|
"""
|
||||||
|
Convert mutable types to immutable types.
|
||||||
|
"""
|
||||||
|
if isinstance(o, bytearray):
|
||||||
|
return bytes(o)
|
||||||
|
if isinstance(o, tuple):
|
||||||
|
try:
|
||||||
|
hash(o)
|
||||||
|
return o
|
||||||
|
except Exception:
|
||||||
|
return tuple(constify(elt) for elt in o)
|
||||||
|
if isinstance(o, list):
|
||||||
|
return tuple(constify(elt) for elt in o)
|
||||||
|
if isinstance(o, dict):
|
||||||
|
cdict = dict()
|
||||||
|
for k, v in o.items():
|
||||||
|
cdict[k] = constify(v)
|
||||||
|
return Dict(cdict, True)
|
||||||
|
return o
|
||||||
195
venv/Lib/site-packages/dns/inet.py
Normal file
195
venv/Lib/site-packages/dns/inet.py
Normal file
@@ -0,0 +1,195 @@
|
|||||||
|
# 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.
|
||||||
|
|
||||||
|
"""Generic Internet address helper functions."""
|
||||||
|
|
||||||
|
import socket
|
||||||
|
from typing import Any, Tuple
|
||||||
|
|
||||||
|
import dns.ipv4
|
||||||
|
import dns.ipv6
|
||||||
|
|
||||||
|
# We assume that AF_INET and AF_INET6 are always defined. We keep
|
||||||
|
# these here for the benefit of any old code (unlikely though that
|
||||||
|
# is!).
|
||||||
|
AF_INET = socket.AF_INET
|
||||||
|
AF_INET6 = socket.AF_INET6
|
||||||
|
|
||||||
|
|
||||||
|
def inet_pton(family: int, text: str) -> bytes:
|
||||||
|
"""Convert the textual form of a network address into its binary form.
|
||||||
|
|
||||||
|
*family* is an ``int``, the address family.
|
||||||
|
|
||||||
|
*text* is a ``str``, the textual address.
|
||||||
|
|
||||||
|
Raises ``NotImplementedError`` if the address family specified is not
|
||||||
|
implemented.
|
||||||
|
|
||||||
|
Returns a ``bytes``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
if family == AF_INET:
|
||||||
|
return dns.ipv4.inet_aton(text)
|
||||||
|
elif family == AF_INET6:
|
||||||
|
return dns.ipv6.inet_aton(text, True)
|
||||||
|
else:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
|
||||||
|
def inet_ntop(family: int, address: bytes) -> str:
|
||||||
|
"""Convert the binary form of a network address into its textual form.
|
||||||
|
|
||||||
|
*family* is an ``int``, the address family.
|
||||||
|
|
||||||
|
*address* is a ``bytes``, the network address in binary form.
|
||||||
|
|
||||||
|
Raises ``NotImplementedError`` if the address family specified is not
|
||||||
|
implemented.
|
||||||
|
|
||||||
|
Returns a ``str``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
if family == AF_INET:
|
||||||
|
return dns.ipv4.inet_ntoa(address)
|
||||||
|
elif family == AF_INET6:
|
||||||
|
return dns.ipv6.inet_ntoa(address)
|
||||||
|
else:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
|
||||||
|
def af_for_address(text: str) -> int:
|
||||||
|
"""Determine the address family of a textual-form network address.
|
||||||
|
|
||||||
|
*text*, a ``str``, the textual address.
|
||||||
|
|
||||||
|
Raises ``ValueError`` if the address family cannot be determined
|
||||||
|
from the input.
|
||||||
|
|
||||||
|
Returns an ``int``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
try:
|
||||||
|
dns.ipv4.inet_aton(text)
|
||||||
|
return AF_INET
|
||||||
|
except Exception:
|
||||||
|
try:
|
||||||
|
dns.ipv6.inet_aton(text, True)
|
||||||
|
return AF_INET6
|
||||||
|
except Exception:
|
||||||
|
raise ValueError
|
||||||
|
|
||||||
|
|
||||||
|
def is_multicast(text: str) -> bool:
|
||||||
|
"""Is the textual-form network address a multicast address?
|
||||||
|
|
||||||
|
*text*, a ``str``, the textual address.
|
||||||
|
|
||||||
|
Raises ``ValueError`` if the address family cannot be determined
|
||||||
|
from the input.
|
||||||
|
|
||||||
|
Returns a ``bool``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
try:
|
||||||
|
first = dns.ipv4.inet_aton(text)[0]
|
||||||
|
return first >= 224 and first <= 239
|
||||||
|
except Exception:
|
||||||
|
try:
|
||||||
|
first = dns.ipv6.inet_aton(text, True)[0]
|
||||||
|
return first == 255
|
||||||
|
except Exception:
|
||||||
|
raise ValueError
|
||||||
|
|
||||||
|
|
||||||
|
def is_address(text: str) -> bool:
|
||||||
|
"""Is the specified string an IPv4 or IPv6 address?
|
||||||
|
|
||||||
|
*text*, a ``str``, the textual address.
|
||||||
|
|
||||||
|
Returns a ``bool``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
try:
|
||||||
|
dns.ipv4.inet_aton(text)
|
||||||
|
return True
|
||||||
|
except Exception:
|
||||||
|
try:
|
||||||
|
dns.ipv6.inet_aton(text, True)
|
||||||
|
return True
|
||||||
|
except Exception:
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def low_level_address_tuple(high_tuple: Tuple[str, int], af: int | None = None) -> Any:
|
||||||
|
"""Given a "high-level" address tuple, i.e.
|
||||||
|
an (address, port) return the appropriate "low-level" address tuple
|
||||||
|
suitable for use in socket calls.
|
||||||
|
|
||||||
|
If an *af* other than ``None`` is provided, it is assumed the
|
||||||
|
address in the high-level tuple is valid and has that af. If af
|
||||||
|
is ``None``, then af_for_address will be called.
|
||||||
|
"""
|
||||||
|
address, port = high_tuple
|
||||||
|
if af is None:
|
||||||
|
af = af_for_address(address)
|
||||||
|
if af == AF_INET:
|
||||||
|
return (address, port)
|
||||||
|
elif af == AF_INET6:
|
||||||
|
i = address.find("%")
|
||||||
|
if i < 0:
|
||||||
|
# no scope, shortcut!
|
||||||
|
return (address, port, 0, 0)
|
||||||
|
# try to avoid getaddrinfo()
|
||||||
|
addrpart = address[:i]
|
||||||
|
scope = address[i + 1 :]
|
||||||
|
if scope.isdigit():
|
||||||
|
return (addrpart, port, 0, int(scope))
|
||||||
|
try:
|
||||||
|
return (addrpart, port, 0, socket.if_nametoindex(scope))
|
||||||
|
except AttributeError: # pragma: no cover (we can't really test this)
|
||||||
|
ai_flags = socket.AI_NUMERICHOST
|
||||||
|
((*_, tup), *_) = socket.getaddrinfo(address, port, flags=ai_flags)
|
||||||
|
return tup
|
||||||
|
else:
|
||||||
|
raise NotImplementedError(f"unknown address family {af}")
|
||||||
|
|
||||||
|
|
||||||
|
def any_for_af(af):
|
||||||
|
"""Return the 'any' address for the specified address family."""
|
||||||
|
if af == socket.AF_INET:
|
||||||
|
return "0.0.0.0"
|
||||||
|
elif af == socket.AF_INET6:
|
||||||
|
return "::"
|
||||||
|
raise NotImplementedError(f"unknown address family {af}")
|
||||||
|
|
||||||
|
|
||||||
|
def canonicalize(text: str) -> str:
|
||||||
|
"""Verify that *address* is a valid text form IPv4 or IPv6 address and return its
|
||||||
|
canonical text form. IPv6 addresses with scopes are rejected.
|
||||||
|
|
||||||
|
*text*, a ``str``, the address in textual form.
|
||||||
|
|
||||||
|
Raises ``ValueError`` if the text is not valid.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
return dns.ipv6.canonicalize(text)
|
||||||
|
except Exception:
|
||||||
|
try:
|
||||||
|
return dns.ipv4.canonicalize(text)
|
||||||
|
except Exception:
|
||||||
|
raise ValueError
|
||||||
Reference in New Issue
Block a user