'rdtypes',
'update',
'version',
+ 'wiredata',
'zone',
]
import dns.rrset
import dns.renderer
import dns.tsig
+import dns.wiredata
class ShortHeader(dns.exception.FormError):
"""Raised if the DNS packet passed to from_wire() is too short."""
def __init__(self, wire, message, question_only=False,
one_rr_per_rrset=False):
- self.wire = wire
+ self.wire = dns.wiredata.maybe_wrap(wire)
self.message = message
self.current = 0
self.updating = False
import encodings.idna
import dns.exception
+import dns.wiredata
NAMERELN_NONE = 0
NAMERELN_SUPERDOMAIN = 1
if not isinstance(message, str):
raise ValueError("input to from_wire() must be a byte string")
+ message = dns.wiredata.maybe_wrap(message)
labels = []
biggest_pointer = current
hops = 0
cused = 1
while count != 0:
if count < 64:
- labels.append(message[current : current + count])
+ labels.append(message[current : current + count].unwrap())
current += count
if hops == 0:
cused += count
import dns.rdataclass
import dns.rdatatype
import dns.tokenizer
+import dns.wiredata
_hex_chunksize = 32
@type origin: dns.name.Name
@rtype: dns.rdata.Rdata instance"""
+ wire = dns.wiredata.maybe_wrap(wire)
cls = get_rdata_class(rdclass, rdtype)
return cls.from_wire(rdclass, rdtype, wire, current, rdlen, origin)
file.write(self.certificate)
def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
- prefix = wire[current : current + 5]
+ prefix = wire[current : current + 5].unwrap()
current += 5
rdlen -= 5
if rdlen < 0:
raise dns.exception.FormError
(certificate_type, key_tag, algorithm) = struct.unpack("!HHB", prefix)
- certificate = wire[current : current + rdlen]
+ certificate = wire[current : current + rdlen].unwrap()
return cls(rdclass, rdtype, certificate_type, key_tag, algorithm,
certificate)
raise dns.exception.FormError
if not right == '' and not right.isdigit():
raise dns.exception.FormError
-
+
class GPOS(dns.rdata.Rdata):
"""GPOS record
@see: RFC 1712"""
__slots__ = ['latitude', 'longitude', 'altitude']
-
+
def __init__(self, rdclass, rdtype, latitude, longitude, altitude):
super(GPOS, self).__init__(rdclass, rdtype)
if isinstance(latitude, float) or \
def to_text(self, origin=None, relativize=True, **kw):
return '%s %s %s' % (self.latitude, self.longitude, self.altitude)
-
+
def from_text(cls, rdclass, rdtype, tok, origin = None, relativize = True):
latitude = tok.get_string()
longitude = tok.get_string()
altitude = tok.get_string()
tok.get_eol()
return cls(rdclass, rdtype, latitude, longitude, altitude)
-
+
from_text = classmethod(from_text)
def to_wire(self, file, compress = None, origin = None):
byte = chr(l)
file.write(byte)
file.write(self.altitude)
-
+
def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
l = ord(wire[current])
current += 1
rdlen -= 1
if l > rdlen:
raise dns.exception.FormError
- latitude = wire[current : current + l]
+ latitude = wire[current : current + l].unwrap()
current += l
rdlen -= l
l = ord(wire[current])
rdlen -= 1
if l > rdlen:
raise dns.exception.FormError
- longitude = wire[current : current + l]
+ longitude = wire[current : current + l].unwrap()
current += l
rdlen -= l
l = ord(wire[current])
rdlen -= 1
if l != rdlen:
raise dns.exception.FormError
- altitude = wire[current : current + l]
+ altitude = wire[current : current + l].unwrap()
return cls(rdclass, rdtype, latitude, longitude, altitude)
from_wire = classmethod(from_wire)
@see: RFC 1035"""
__slots__ = ['cpu', 'os']
-
+
def __init__(self, rdclass, rdtype, cpu, os):
super(HINFO, self).__init__(rdclass, rdtype)
self.cpu = cpu
def to_text(self, origin=None, relativize=True, **kw):
return '"%s" "%s"' % (dns.rdata._escapify(self.cpu),
dns.rdata._escapify(self.os))
-
+
def from_text(cls, rdclass, rdtype, tok, origin = None, relativize = True):
cpu = tok.get_string()
os = tok.get_string()
tok.get_eol()
return cls(rdclass, rdtype, cpu, os)
-
+
from_text = classmethod(from_text)
def to_wire(self, file, compress = None, origin = None):
byte = chr(l)
file.write(byte)
file.write(self.os)
-
+
def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
l = ord(wire[current])
current += 1
rdlen -= 1
if l > rdlen:
raise dns.exception.FormError
- cpu = wire[current : current + l]
+ cpu = wire[current : current + l].unwrap()
current += l
rdlen -= l
l = ord(wire[current])
rdlen -= 1
if l != rdlen:
raise dns.exception.FormError
- os = wire[current : current + l]
+ os = wire[current : current + l].unwrap()
return cls(rdclass, rdtype, cpu, os)
from_wire = classmethod(from_wire)
wire[current : current + 4])
current += 4
rdlen -= 4
- hit = wire[current : current + lh]
+ hit = wire[current : current + lh].unwrap()
current += lh
rdlen -= lh
- key = wire[current : current + lk]
+ key = wire[current : current + lk].unwrap()
current += lk
rdlen -= lk
servers = []
rdlen -= 1
if l > rdlen:
raise dns.exception.FormError
- address = wire[current : current + l]
+ address = wire[current : current + l].unwrap()
current += l
rdlen -= l
if rdlen > 0:
rdlen -= 1
if l != rdlen:
raise dns.exception.FormError
- subaddress = wire[current : current + l]
+ subaddress = wire[current : current + l].unwrap()
else:
subaddress = ''
return cls(rdclass, rdtype, address, subaddress)
rdlen -= 2
if rdlen < octets:
raise dns.exception.FormError("bad NSEC bitmap length")
- bitmap = wire[current : current + octets]
+ bitmap = wire[current : current + octets].unwrap()
current += octets
rdlen -= octets
windows.append((window, bitmap))
wire[current : current + 5])
current += 5
rdlen -= 5
- salt = wire[current : current + slen]
+ salt = wire[current : current + slen].unwrap()
current += slen
rdlen -= slen
(nlen, ) = struct.unpack('!B', wire[current])
current += 1
rdlen -= 1
- next = wire[current : current + nlen]
+ next = wire[current : current + nlen].unwrap()
current += nlen
rdlen -= nlen
windows = []
rdlen -= 2
if rdlen < octets:
raise dns.exception.FormError("bad NSEC3 bitmap length")
- bitmap = wire[current : current + octets]
+ bitmap = wire[current : current + octets].unwrap()
current += octets
rdlen -= octets
windows.append((window, bitmap))
wire[current : current + 5])
current += 5
rdlen -= 5
- salt = wire[current : current + slen]
+ salt = wire[current : current + slen].unwrap()
current += slen
rdlen -= slen
if rdlen != 0:
(next, cused) = dns.name.from_wire(wire[: current + rdlen], current)
current += cused
rdlen -= cused
- bitmap = wire[current : current + rdlen]
+ bitmap = wire[current : current + rdlen].unwrap()
if not origin is None:
next = next.relativize(origin)
return cls(rdclass, rdtype, next, bitmap)
__slots__ = ['mname', 'rname', 'serial', 'refresh', 'retry', 'expire',
'minimum']
-
+
def __init__(self, rdclass, rdtype, mname, rname, serial, refresh, retry,
expire, minimum):
super(SOA, self).__init__(rdclass, rdtype)
return '%s %s %d %d %d %d %d' % (
mname, rname, self.serial, self.refresh, self.retry,
self.expire, self.minimum )
-
+
def from_text(cls, rdclass, rdtype, tok, origin = None, relativize = True):
mname = tok.get_name()
rname = tok.get_name()
tok.get_eol()
return cls(rdclass, rdtype, mname, rname, serial, refresh, retry,
expire, minimum )
-
+
from_text = classmethod(from_text)
def to_wire(self, file, compress = None, origin = None):
@see: draft-ietf-secsh-dns-05.txt"""
__slots__ = ['algorithm', 'fp_type', 'fingerprint']
-
+
def __init__(self, rdclass, rdtype, algorithm, fp_type,
fingerprint):
super(SSHFP, self).__init__(rdclass, rdtype)
self.fp_type,
dns.rdata._hexify(self.fingerprint,
chunksize=128))
-
+
def from_text(cls, rdclass, rdtype, tok, origin = None, relativize = True):
algorithm = tok.get_uint8()
fp_type = tok.get_uint8()
fingerprint = fingerprint.decode('hex_codec')
tok.get_eol()
return cls(rdclass, rdtype, algorithm, fp_type, fingerprint)
-
+
from_text = classmethod(from_text)
def to_wire(self, file, compress = None, origin = None):
header = struct.pack("!BB", self.algorithm, self.fp_type)
file.write(header)
file.write(self.fingerprint)
-
+
def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
header = struct.unpack("!BB", wire[current : current + 2])
current += 2
rdlen -= 2
- fingerprint = wire[current : current + rdlen]
+ fingerprint = wire[current : current + rdlen].unwrap()
return cls(rdclass, rdtype, header[0], header[1], fingerprint)
from_wire = classmethod(from_wire)
@see: RFC 1183"""
__slots__ = ['address']
-
+
def __init__(self, rdclass, rdtype, address):
super(X25, self).__init__(rdclass, rdtype)
self.address = address
def to_text(self, origin=None, relativize=True, **kw):
return '"%s"' % dns.rdata._escapify(self.address)
-
+
def from_text(cls, rdclass, rdtype, tok, origin = None, relativize = True):
address = tok.get_string()
tok.get_eol()
return cls(rdclass, rdtype, address)
-
+
from_text = classmethod(from_text)
def to_wire(self, file, compress = None, origin = None):
byte = chr(l)
file.write(byte)
file.write(self.address)
-
+
def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
l = ord(wire[current])
current += 1
rdlen -= 1
if l != rdlen:
raise dns.exception.FormError
- address = wire[current : current + l]
+ address = wire[current : current + l].unwrap()
return cls(rdclass, rdtype, address)
from_wire = classmethod(from_wire)
rdlen -= 4
if rdlen < afdlen:
raise dns.exception.FormError
- address = wire[current : current + afdlen]
+ address = wire[current : current + afdlen].unwrap()
l = len(address)
if header[0] == 1:
if l < 4:
file.write(self.data)
def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
- data = wire[current : current + rdlen]
+ data = wire[current : current + rdlen].unwrap()
return cls(rdclass, rdtype, data)
from_wire = classmethod(from_wire)
rdlen -= cused
else:
raise dns.exception.FormError('invalid IPSECKEY gateway type')
- key = wire[current : current + rdlen]
+ key = wire[current : current + rdlen].unwrap()
return cls(rdclass, rdtype, header[0], gateway_type, header[2],
gateway, key)
__slots__ = ['order', 'preference', 'flags', 'service', 'regexp',
'replacement']
-
+
def __init__(self, rdclass, rdtype, order, preference, flags, service,
regexp, replacement):
super(NAPTR, self).__init__(rdclass, rdtype)
tok.get_eol()
return cls(rdclass, rdtype, order, preference, flags, service,
regexp, replacement)
-
+
from_text = classmethod(from_text)
def to_wire(self, file, compress = None, origin = None):
_write_string(file, self.service)
_write_string(file, self.regexp)
self.replacement.to_wire(file, compress, origin)
-
+
def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
(order, preference) = struct.unpack('!HH', wire[current : current + 4])
current += 4
rdlen -= 1
if l > rdlen or rdlen < 0:
raise dns.exception.FormError
- s = wire[current : current + l]
+ s = wire[current : current + l].unwrap()
current += l
rdlen -= l
strings.append(s)
def choose_relativity(self, origin = None, relativize = True):
self.replacement = self.replacement.choose_relativity(origin,
relativize)
-
+
def _cmp(self, other):
sp = struct.pack("!HH", self.order, self.preference)
op = struct.pack("!HH", other.order, other.preference)
file.write(self.address)
def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
- address = wire[current : current + rdlen]
+ address = wire[current : current + rdlen].unwrap()
return cls(rdclass, rdtype, address)
from_wire = classmethod(from_wire)
@see: RFC 2163"""
__slots__ = ['preference', 'map822', 'mapx400']
-
+
def __init__(self, rdclass, rdtype, preference, map822, mapx400):
super(PX, self).__init__(rdclass, rdtype)
self.preference = preference
map822 = self.map822.choose_relativity(origin, relativize)
mapx400 = self.mapx400.choose_relativity(origin, relativize)
return '%d %s %s' % (self.preference, map822, mapx400)
-
+
def from_text(cls, rdclass, rdtype, tok, origin = None, relativize = True):
preference = tok.get_uint16()
map822 = tok.get_name()
mapx400 = mapx400.choose_relativity(origin, relativize)
tok.get_eol()
return cls(rdclass, rdtype, preference, map822, mapx400)
-
+
from_text = classmethod(from_text)
def to_wire(self, file, compress = None, origin = None):
file.write(pref)
self.map822.to_wire(file, None, origin)
self.mapx400.to_wire(file, None, origin)
-
+
def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
(preference, ) = struct.unpack('!H', wire[current : current + 2])
current += 2
protocol, = struct.unpack('!B', wire[current + 4 : current + 5])
current += 5
rdlen -= 5
- bitmap = wire[current : current + rdlen]
+ bitmap = wire[current : current + rdlen].unwrap()
return cls(rdclass, rdtype, address, protocol, bitmap)
from_wire = classmethod(from_wire)
header = struct.unpack("!HBB", wire[current : current + 4])
current += 4
rdlen -= 4
- digest = wire[current : current + rdlen]
+ digest = wire[current : current + rdlen].unwrap()
return cls(rdclass, rdtype, header[0], header[1], header[2], digest)
from_wire = classmethod(from_wire)
header = struct.unpack('!HBB', wire[current : current + 4])
current += 4
rdlen -= 4
- key = wire[current : current + rdlen]
+ key = wire[current : current + rdlen].unwrap()
return cls(rdclass, rdtype, header[0], header[1], header[2],
key)
rdlen -= cused
if not origin is None:
signer = signer.relativize(origin)
- signature = wire[current : current + rdlen]
+ signature = wire[current : current + rdlen].unwrap()
return cls(rdclass, rdtype, header[0], header[1], header[2],
header[3], header[4], header[5], header[6], signer,
signature)
rdlen -= 1
if l > rdlen:
raise dns.exception.FormError
- s = wire[current : current + l]
+ s = wire[current : current + l].unwrap()
current += l
rdlen -= l
strings.append(s)
--- /dev/null
+# Copyright (C) 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 Wire Data Helper"""
+
+import sys
+
+import dns.exception
+
+class WireData(str):
+ # WireData is a string with stricter slicing
+ def __getitem__(self, key):
+ try:
+ return WireData(super(WireData, self).__getitem__(key))
+ except IndexError:
+ raise dns.exception.FormError
+ def __getslice__(self, i, j):
+ try:
+ if j == sys.maxint:
+ # handle the case where the right bound is unspecified
+ j = len(self)
+ if i < 0 or j < 0:
+ raise dns.exception.FormError
+ # If it's not an empty slice, access left and right bounds
+ # to make sure they're valid
+ if i != j:
+ super(WireData, self).__getitem__(i)
+ super(WireData, self).__getitem__(j - 1)
+ return WireData(super(WireData, self).__getslice__(i, j))
+ except IndexError:
+ raise dns.exception.FormError
+ def __iter__(self):
+ i = 0
+ while 1:
+ try:
+ yield self[i]
+ i += 1
+ except dns.exception.FormError:
+ raise StopIteration
+ def unwrap(self):
+ return str(self)
+
+def maybe_wrap(wire):
+ if not isinstance(wire, WireData):
+ return WireData(wire)
+ else:
+ return wire
projects / thirdparty / dnspython.git / commitdiff
