| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924 |
- # Copyright (C) 2001-2007, 2009-2011 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 Names.
- @var root: The DNS root name.
- @type root: dns.name.Name object
- @var empty: The empty DNS name.
- @type empty: dns.name.Name object
- """
- from io import BytesIO
- import struct
- import sys
- import copy
- import encodings.idna
- try:
- import idna
- have_idna_2008 = True
- except ImportError:
- have_idna_2008 = False
- import dns.exception
- import dns.wiredata
- from ._compat import long, binary_type, text_type, unichr, maybe_decode
- try:
- maxint = sys.maxint
- except AttributeError:
- maxint = (1 << (8 * struct.calcsize("P"))) // 2 - 1
- NAMERELN_NONE = 0
- NAMERELN_SUPERDOMAIN = 1
- NAMERELN_SUBDOMAIN = 2
- NAMERELN_EQUAL = 3
- NAMERELN_COMMONANCESTOR = 4
- class EmptyLabel(dns.exception.SyntaxError):
- """A DNS label is empty."""
- class BadEscape(dns.exception.SyntaxError):
- """An escaped code in a text format of DNS name is invalid."""
- class BadPointer(dns.exception.FormError):
- """A DNS compression pointer points forward instead of backward."""
- class BadLabelType(dns.exception.FormError):
- """The label type in DNS name wire format is unknown."""
- class NeedAbsoluteNameOrOrigin(dns.exception.DNSException):
- """An attempt was made to convert a non-absolute name to
- wire when there was also a non-absolute (or missing) origin."""
- class NameTooLong(dns.exception.FormError):
- """A DNS name is > 255 octets long."""
- class LabelTooLong(dns.exception.SyntaxError):
- """A DNS label is > 63 octets long."""
- class AbsoluteConcatenation(dns.exception.DNSException):
- """An attempt was made to append anything other than the
- empty name to an absolute DNS name."""
- class NoParent(dns.exception.DNSException):
- """An attempt was made to get the parent of the root name
- or the empty name."""
- class NoIDNA2008(dns.exception.DNSException):
- """IDNA 2008 processing was requested but the idna module is not
- available."""
- class IDNAException(dns.exception.DNSException):
- """IDNA processing raised an exception."""
- supp_kwargs = set(['idna_exception'])
- fmt = "IDNA processing exception: {idna_exception}"
- class IDNACodec(object):
- """Abstract base class for IDNA encoder/decoders."""
- def __init__(self):
- pass
- def encode(self, label):
- raise NotImplementedError
- def decode(self, label):
- # We do not apply any IDNA policy on decode; we just
- downcased = label.lower()
- if downcased.startswith(b'xn--'):
- try:
- label = downcased[4:].decode('punycode')
- except Exception as e:
- raise IDNAException(idna_exception=e)
- else:
- label = maybe_decode(label)
- return _escapify(label, True)
- class IDNA2003Codec(IDNACodec):
- """IDNA 2003 encoder/decoder."""
- def __init__(self, strict_decode=False):
- """Initialize the IDNA 2003 encoder/decoder.
- @param strict_decode: If True, then IDNA2003 checking is done when
- decoding. This can cause failures if the name was encoded with
- IDNA2008. The default is False.
- @type strict_decode: bool
- """
- super(IDNA2003Codec, self).__init__()
- self.strict_decode = strict_decode
- def encode(self, label):
- if label == '':
- return b''
- try:
- return encodings.idna.ToASCII(label)
- except UnicodeError:
- raise LabelTooLong
- def decode(self, label):
- if not self.strict_decode:
- return super(IDNA2003Codec, self).decode(label)
- if label == b'':
- return u''
- try:
- return _escapify(encodings.idna.ToUnicode(label), True)
- except Exception as e:
- raise IDNAException(idna_exception=e)
- class IDNA2008Codec(IDNACodec):
- """IDNA 2008 encoder/decoder."""
- def __init__(self, uts_46=False, transitional=False,
- allow_pure_ascii=False, strict_decode=False):
- """Initialize the IDNA 2008 encoder/decoder.
- @param uts_46: If True, apply Unicode IDNA compatibility processing
- as described in Unicode Technical Standard #46
- (U{http://unicode.org/reports/tr46/}). This parameter is only
- meaningful if IDNA 2008 is in use. If False, do not apply
- the mapping. The default is False
- @type uts_46: bool
- @param transitional: If True, use the "transitional" mode described
- in Unicode Technical Standard #46. This parameter is only
- meaningful if IDNA 2008 is in use. The default is False.
- @type transitional: bool
- @param allow_pure_ascii: If True, then a label which
- consists of only ASCII characters is allowed. This is less strict
- than regular IDNA 2008, but is also necessary for mixed names,
- e.g. a name with starting with "_sip._tcp." and ending in an IDN
- suffixm which would otherwise be disallowed. The default is False
- @type allow_pure_ascii: bool
- @param strict_decode: If True, then IDNA2008 checking is done when
- decoding. This can cause failures if the name was encoded with
- IDNA2003. The default is False.
- @type strict_decode: bool
- """
- super(IDNA2008Codec, self).__init__()
- self.uts_46 = uts_46
- self.transitional = transitional
- self.allow_pure_ascii = allow_pure_ascii
- self.strict_decode = strict_decode
- def is_all_ascii(self, label):
- for c in label:
- if ord(c) > 0x7f:
- return False
- return True
- def encode(self, label):
- if label == '':
- return b''
- if self.allow_pure_ascii and self.is_all_ascii(label):
- return label.encode('ascii')
- if not have_idna_2008:
- raise NoIDNA2008
- try:
- if self.uts_46:
- label = idna.uts46_remap(label, False, self.transitional)
- return idna.alabel(label)
- except idna.IDNAError as e:
- raise IDNAException(idna_exception=e)
- def decode(self, label):
- if not self.strict_decode:
- return super(IDNA2008Codec, self).decode(label)
- if label == b'':
- return u''
- if not have_idna_2008:
- raise NoIDNA2008
- try:
- if self.uts_46:
- label = idna.uts46_remap(label, False, False)
- return _escapify(idna.ulabel(label), True)
- except idna.IDNAError as e:
- raise IDNAException(idna_exception=e)
- _escaped = bytearray(b'"().;\\@$')
- IDNA_2003_Practical = IDNA2003Codec(False)
- IDNA_2003_Strict = IDNA2003Codec(True)
- IDNA_2003 = IDNA_2003_Practical
- IDNA_2008_Practical = IDNA2008Codec(True, False, True, False)
- IDNA_2008_UTS_46 = IDNA2008Codec(True, False, False, False)
- IDNA_2008_Strict = IDNA2008Codec(False, False, False, True)
- IDNA_2008_Transitional = IDNA2008Codec(True, True, False, False)
- IDNA_2008 = IDNA_2008_Practical
- def _escapify(label, unicode_mode=False):
- """Escape the characters in label which need it.
- @param unicode_mode: escapify only special and whitespace (<= 0x20)
- characters
- @returns: the escaped string
- @rtype: string"""
- if not unicode_mode:
- text = ''
- if isinstance(label, text_type):
- label = label.encode()
- for c in bytearray(label):
- if c in _escaped:
- text += '\\' + chr(c)
- elif c > 0x20 and c < 0x7F:
- text += chr(c)
- else:
- text += '\\%03d' % c
- return text.encode()
- text = u''
- if isinstance(label, binary_type):
- label = label.decode()
- for c in label:
- if c > u'\x20' and c < u'\x7f':
- text += c
- else:
- if c >= u'\x7f':
- text += c
- else:
- text += u'\\%03d' % ord(c)
- return text
- def _validate_labels(labels):
- """Check for empty labels in the middle of a label sequence,
- labels that are too long, and for too many labels.
- @raises NameTooLong: the name as a whole is too long
- @raises EmptyLabel: a label is empty (i.e. the root label) and appears
- in a position other than the end of the label sequence"""
- l = len(labels)
- total = 0
- i = -1
- j = 0
- for label in labels:
- ll = len(label)
- total += ll + 1
- if ll > 63:
- raise LabelTooLong
- if i < 0 and label == b'':
- i = j
- j += 1
- if total > 255:
- raise NameTooLong
- if i >= 0 and i != l - 1:
- raise EmptyLabel
- def _ensure_bytes(label):
- if isinstance(label, binary_type):
- return label
- if isinstance(label, text_type):
- return label.encode()
- raise ValueError
- class Name(object):
- """A DNS name.
- The dns.name.Name class represents a DNS name as a tuple of labels.
- Instances of the class are immutable.
- @ivar labels: The tuple of labels in the name. Each label is a string of
- up to 63 octets."""
- __slots__ = ['labels']
- def __init__(self, labels):
- """Initialize a domain name from a list of labels.
- @param labels: the labels
- @type labels: any iterable whose values are strings
- """
- labels = [_ensure_bytes(x) for x in labels]
- super(Name, self).__setattr__('labels', tuple(labels))
- _validate_labels(self.labels)
- def __setattr__(self, name, value):
- raise TypeError("object doesn't support attribute assignment")
- def __copy__(self):
- return Name(self.labels)
- def __deepcopy__(self, memo):
- return Name(copy.deepcopy(self.labels, memo))
- def __getstate__(self):
- return {'labels': self.labels}
- def __setstate__(self, state):
- super(Name, self).__setattr__('labels', state['labels'])
- _validate_labels(self.labels)
- def is_absolute(self):
- """Is the most significant label of this name the root label?
- @rtype: bool
- """
- return len(self.labels) > 0 and self.labels[-1] == b''
- def is_wild(self):
- """Is this name wild? (I.e. Is the least significant label '*'?)
- @rtype: bool
- """
- return len(self.labels) > 0 and self.labels[0] == b'*'
- def __hash__(self):
- """Return a case-insensitive hash of the name.
- @rtype: int
- """
- h = long(0)
- for label in self.labels:
- for c in bytearray(label.lower()):
- h += (h << 3) + c
- return int(h % maxint)
- def fullcompare(self, other):
- """Compare two names, returning a 3-tuple (relation, order, nlabels).
- I{relation} describes the relation ship between the names,
- and is one of: dns.name.NAMERELN_NONE,
- dns.name.NAMERELN_SUPERDOMAIN, dns.name.NAMERELN_SUBDOMAIN,
- dns.name.NAMERELN_EQUAL, or dns.name.NAMERELN_COMMONANCESTOR
- I{order} is < 0 if self < other, > 0 if self > other, and ==
- 0 if self == other. A relative name is always less than an
- absolute name. If both names have the same relativity, then
- the DNSSEC order relation is used to order them.
- I{nlabels} is the number of significant labels that the two names
- have in common.
- """
- sabs = self.is_absolute()
- oabs = other.is_absolute()
- if sabs != oabs:
- if sabs:
- return (NAMERELN_NONE, 1, 0)
- else:
- return (NAMERELN_NONE, -1, 0)
- l1 = len(self.labels)
- l2 = len(other.labels)
- ldiff = l1 - l2
- if ldiff < 0:
- l = l1
- else:
- l = l2
- order = 0
- nlabels = 0
- namereln = NAMERELN_NONE
- while l > 0:
- l -= 1
- l1 -= 1
- l2 -= 1
- label1 = self.labels[l1].lower()
- label2 = other.labels[l2].lower()
- if label1 < label2:
- order = -1
- if nlabels > 0:
- namereln = NAMERELN_COMMONANCESTOR
- return (namereln, order, nlabels)
- elif label1 > label2:
- order = 1
- if nlabels > 0:
- namereln = NAMERELN_COMMONANCESTOR
- return (namereln, order, nlabels)
- nlabels += 1
- order = ldiff
- if ldiff < 0:
- namereln = NAMERELN_SUPERDOMAIN
- elif ldiff > 0:
- namereln = NAMERELN_SUBDOMAIN
- else:
- namereln = NAMERELN_EQUAL
- return (namereln, order, nlabels)
- def is_subdomain(self, other):
- """Is self a subdomain of other?
- The notion of subdomain includes equality.
- @rtype: bool
- """
- (nr, o, nl) = self.fullcompare(other)
- if nr == NAMERELN_SUBDOMAIN or nr == NAMERELN_EQUAL:
- return True
- return False
- def is_superdomain(self, other):
- """Is self a superdomain of other?
- The notion of subdomain includes equality.
- @rtype: bool
- """
- (nr, o, nl) = self.fullcompare(other)
- if nr == NAMERELN_SUPERDOMAIN or nr == NAMERELN_EQUAL:
- return True
- return False
- def canonicalize(self):
- """Return a name which is equal to the current name, but is in
- DNSSEC canonical form.
- @rtype: dns.name.Name object
- """
- return Name([x.lower() for x in self.labels])
- def __eq__(self, other):
- if isinstance(other, Name):
- return self.fullcompare(other)[1] == 0
- else:
- return False
- def __ne__(self, other):
- if isinstance(other, Name):
- return self.fullcompare(other)[1] != 0
- else:
- return True
- def __lt__(self, other):
- if isinstance(other, Name):
- return self.fullcompare(other)[1] < 0
- else:
- return NotImplemented
- def __le__(self, other):
- if isinstance(other, Name):
- return self.fullcompare(other)[1] <= 0
- else:
- return NotImplemented
- def __ge__(self, other):
- if isinstance(other, Name):
- return self.fullcompare(other)[1] >= 0
- else:
- return NotImplemented
- def __gt__(self, other):
- if isinstance(other, Name):
- return self.fullcompare(other)[1] > 0
- else:
- return NotImplemented
- def __repr__(self):
- return '<DNS name ' + self.__str__() + '>'
- def __str__(self):
- return self.to_text(False)
- def to_text(self, omit_final_dot=False):
- """Convert name to text format.
- @param omit_final_dot: If True, don't emit the final dot (denoting the
- root label) for absolute names. The default is False.
- @rtype: string
- """
- if len(self.labels) == 0:
- return maybe_decode(b'@')
- if len(self.labels) == 1 and self.labels[0] == b'':
- return maybe_decode(b'.')
- if omit_final_dot and self.is_absolute():
- l = self.labels[:-1]
- else:
- l = self.labels
- s = b'.'.join(map(_escapify, l))
- return maybe_decode(s)
- def to_unicode(self, omit_final_dot=False, idna_codec=None):
- """Convert name to Unicode text format.
- IDN ACE labels are converted to Unicode.
- @param omit_final_dot: If True, don't emit the final dot (denoting the
- root label) for absolute names. The default is False.
- @type omit_final_dot: bool
- @param idna_codec: IDNA encoder/decoder. If None, the
- IDNA_2003_Practical encoder/decoder is used. The IDNA_2003_Practical
- decoder does not impose any policy, it just decodes punycode, so if
- you don't want checking for compliance, you can use this decoder for
- IDNA2008 as well.
- @type idna_codec: dns.name.IDNA
- @rtype: string
- """
- if len(self.labels) == 0:
- return u'@'
- if len(self.labels) == 1 and self.labels[0] == b'':
- return u'.'
- if omit_final_dot and self.is_absolute():
- l = self.labels[:-1]
- else:
- l = self.labels
- if idna_codec is None:
- idna_codec = IDNA_2003_Practical
- return u'.'.join([idna_codec.decode(x) for x in l])
- def to_digestable(self, origin=None):
- """Convert name to a format suitable for digesting in hashes.
- The name is canonicalized and converted to uncompressed wire format.
- @param origin: If the name is relative and origin is not None, then
- origin will be appended to it.
- @type origin: dns.name.Name object
- @raises NeedAbsoluteNameOrOrigin: All names in wire format are
- absolute. If self is a relative name, then an origin must be supplied;
- if it is missing, then this exception is raised
- @rtype: string
- """
- if not self.is_absolute():
- if origin is None or not origin.is_absolute():
- raise NeedAbsoluteNameOrOrigin
- labels = list(self.labels)
- labels.extend(list(origin.labels))
- else:
- labels = self.labels
- dlabels = [struct.pack('!B%ds' % len(x), len(x), x.lower())
- for x in labels]
- return b''.join(dlabels)
- def to_wire(self, file=None, compress=None, origin=None):
- """Convert name to wire format, possibly compressing it.
- @param file: the file where the name is emitted (typically
- a BytesIO file). If None, a string containing the wire name
- will be returned.
- @type file: file or None
- @param compress: The compression table. If None (the default) names
- will not be compressed.
- @type compress: dict
- @param origin: If the name is relative and origin is not None, then
- origin will be appended to it.
- @type origin: dns.name.Name object
- @raises NeedAbsoluteNameOrOrigin: All names in wire format are
- absolute. If self is a relative name, then an origin must be supplied;
- if it is missing, then this exception is raised
- """
- if file is None:
- file = BytesIO()
- want_return = True
- else:
- want_return = False
- if not self.is_absolute():
- if origin is None or not origin.is_absolute():
- raise NeedAbsoluteNameOrOrigin
- labels = list(self.labels)
- labels.extend(list(origin.labels))
- else:
- labels = self.labels
- i = 0
- for label in labels:
- n = Name(labels[i:])
- i += 1
- if compress is not None:
- pos = compress.get(n)
- else:
- pos = None
- if pos is not None:
- value = 0xc000 + pos
- s = struct.pack('!H', value)
- file.write(s)
- break
- else:
- if compress is not None and len(n) > 1:
- pos = file.tell()
- if pos <= 0x3fff:
- compress[n] = pos
- l = len(label)
- file.write(struct.pack('!B', l))
- if l > 0:
- file.write(label)
- if want_return:
- return file.getvalue()
- def __len__(self):
- """The length of the name (in labels).
- @rtype: int
- """
- return len(self.labels)
- def __getitem__(self, index):
- return self.labels[index]
- def __add__(self, other):
- return self.concatenate(other)
- def __sub__(self, other):
- return self.relativize(other)
- def split(self, depth):
- """Split a name into a prefix and suffix at depth.
- @param depth: the number of labels in the suffix
- @type depth: int
- @raises ValueError: the depth was not >= 0 and <= the length of the
- name.
- @returns: the tuple (prefix, suffix)
- @rtype: tuple
- """
- l = len(self.labels)
- if depth == 0:
- return (self, dns.name.empty)
- elif depth == l:
- return (dns.name.empty, self)
- elif depth < 0 or depth > l:
- raise ValueError(
- 'depth must be >= 0 and <= the length of the name')
- return (Name(self[: -depth]), Name(self[-depth:]))
- def concatenate(self, other):
- """Return a new name which is the concatenation of self and other.
- @rtype: dns.name.Name object
- @raises AbsoluteConcatenation: self is absolute and other is
- not the empty name
- """
- if self.is_absolute() and len(other) > 0:
- raise AbsoluteConcatenation
- labels = list(self.labels)
- labels.extend(list(other.labels))
- return Name(labels)
- def relativize(self, origin):
- """If self is a subdomain of origin, return a new name which is self
- relative to origin. Otherwise return self.
- @rtype: dns.name.Name object
- """
- if origin is not None and self.is_subdomain(origin):
- return Name(self[: -len(origin)])
- else:
- return self
- def derelativize(self, origin):
- """If self is a relative name, return a new name which is the
- concatenation of self and origin. Otherwise return self.
- @rtype: dns.name.Name object
- """
- if not self.is_absolute():
- return self.concatenate(origin)
- else:
- return self
- def choose_relativity(self, origin=None, relativize=True):
- """Return a name with the relativity desired by the caller. If
- origin is None, then self is returned. Otherwise, if
- relativize is true the name is relativized, and if relativize is
- false the name is derelativized.
- @rtype: dns.name.Name object
- """
- if origin:
- if relativize:
- return self.relativize(origin)
- else:
- return self.derelativize(origin)
- else:
- return self
- def parent(self):
- """Return the parent of the name.
- @rtype: dns.name.Name object
- @raises NoParent: the name is either the root name or the empty name,
- and thus has no parent.
- """
- if self == root or self == empty:
- raise NoParent
- return Name(self.labels[1:])
- root = Name([b''])
- empty = Name([])
- def from_unicode(text, origin=root, idna_codec=None):
- """Convert unicode text into a Name object.
- Labels are encoded in IDN ACE form.
- @param text: The text to convert into a name.
- @type text: Unicode string
- @param origin: The origin to append to non-absolute names.
- @type origin: dns.name.Name
- @param idna_codec: IDNA encoder/decoder. If None, the default IDNA 2003
- encoder/decoder is used.
- @type idna_codec: dns.name.IDNA
- @rtype: dns.name.Name object
- """
- if not isinstance(text, text_type):
- raise ValueError("input to from_unicode() must be a unicode string")
- if not (origin is None or isinstance(origin, Name)):
- raise ValueError("origin must be a Name or None")
- labels = []
- label = u''
- escaping = False
- edigits = 0
- total = 0
- if idna_codec is None:
- idna_codec = IDNA_2003
- if text == u'@':
- text = u''
- if text:
- if text == u'.':
- return Name([b'']) # no Unicode "u" on this constant!
- for c in text:
- if escaping:
- if edigits == 0:
- if c.isdigit():
- total = int(c)
- edigits += 1
- else:
- label += c
- escaping = False
- else:
- if not c.isdigit():
- raise BadEscape
- total *= 10
- total += int(c)
- edigits += 1
- if edigits == 3:
- escaping = False
- label += unichr(total)
- elif c in [u'.', u'\u3002', u'\uff0e', u'\uff61']:
- if len(label) == 0:
- raise EmptyLabel
- labels.append(idna_codec.encode(label))
- label = u''
- elif c == u'\\':
- escaping = True
- edigits = 0
- total = 0
- else:
- label += c
- if escaping:
- raise BadEscape
- if len(label) > 0:
- labels.append(idna_codec.encode(label))
- else:
- labels.append(b'')
- if (len(labels) == 0 or labels[-1] != b'') and origin is not None:
- labels.extend(list(origin.labels))
- return Name(labels)
- def from_text(text, origin=root, idna_codec=None):
- """Convert text into a Name object.
- @param text: The text to convert into a name.
- @type text: string
- @param origin: The origin to append to non-absolute names.
- @type origin: dns.name.Name
- @param idna_codec: IDNA encoder/decoder. If None, the default IDNA 2003
- encoder/decoder is used.
- @type idna_codec: dns.name.IDNA
- @rtype: dns.name.Name object
- """
- if isinstance(text, text_type):
- return from_unicode(text, origin, idna_codec)
- if not isinstance(text, binary_type):
- raise ValueError("input to from_text() must be a string")
- if not (origin is None or isinstance(origin, Name)):
- raise ValueError("origin must be a Name or None")
- labels = []
- label = b''
- escaping = False
- edigits = 0
- total = 0
- if text == b'@':
- text = b''
- if text:
- if text == b'.':
- return Name([b''])
- for c in bytearray(text):
- byte_ = struct.pack('!B', c)
- if escaping:
- if edigits == 0:
- if byte_.isdigit():
- total = int(byte_)
- edigits += 1
- else:
- label += byte_
- escaping = False
- else:
- if not byte_.isdigit():
- raise BadEscape
- total *= 10
- total += int(byte_)
- edigits += 1
- if edigits == 3:
- escaping = False
- label += struct.pack('!B', total)
- elif byte_ == b'.':
- if len(label) == 0:
- raise EmptyLabel
- labels.append(label)
- label = b''
- elif byte_ == b'\\':
- escaping = True
- edigits = 0
- total = 0
- else:
- label += byte_
- if escaping:
- raise BadEscape
- if len(label) > 0:
- labels.append(label)
- else:
- labels.append(b'')
- if (len(labels) == 0 or labels[-1] != b'') and origin is not None:
- labels.extend(list(origin.labels))
- return Name(labels)
- def from_wire(message, current):
- """Convert possibly compressed wire format into a Name.
- @param message: the entire DNS message
- @type message: string
- @param current: the offset of the beginning of the name from the start
- of the message
- @type current: int
- @raises dns.name.BadPointer: a compression pointer did not point backwards
- in the message
- @raises dns.name.BadLabelType: an invalid label type was encountered.
- @returns: a tuple consisting of the name that was read and the number
- of bytes of the wire format message which were consumed reading it
- @rtype: (dns.name.Name object, int) tuple
- """
- if not isinstance(message, binary_type):
- raise ValueError("input to from_wire() must be a byte string")
- message = dns.wiredata.maybe_wrap(message)
- labels = []
- biggest_pointer = current
- hops = 0
- count = message[current]
- current += 1
- cused = 1
- while count != 0:
- if count < 64:
- labels.append(message[current: current + count].unwrap())
- current += count
- if hops == 0:
- cused += count
- elif count >= 192:
- current = (count & 0x3f) * 256 + message[current]
- if hops == 0:
- cused += 1
- if current >= biggest_pointer:
- raise BadPointer
- biggest_pointer = current
- hops += 1
- else:
- raise BadLabelType
- count = message[current]
- current += 1
- if hops == 0:
- cused += 1
- labels.append('')
- return (Name(labels), cused)
|