class Rdata:
"""Base class for all DNS rdata types."""
- __slots__ = ['rdclass', 'rdtype']
+ __slots__ = ['rdclass', 'rdtype', 'rdcomment']
def __init__(self, rdclass, rdtype):
"""Initialize an rdata.
object.__setattr__(self, 'rdclass', rdclass)
object.__setattr__(self, 'rdtype', rdtype)
+ object.__setattr__(self, 'rdcomment', None)
def __setattr__(self, name, value):
# Rdatas are immutable
def __setstate__(self, state):
for slot, val in state.items():
object.__setattr__(self, slot, val)
+ if not hasattr(self, 'rdcomment'):
+ # Pickled rdata from 2.0.x might not have a rdcomment, so add
+ # it if needed.
+ object.__setattr__(self, 'rdcomment', None)
def covers(self):
"""Return the type a Rdata covers.
# Ensure that all of the arguments correspond to valid fields.
# Don't allow rdclass or rdtype to be changed, though.
for key in kwargs:
+ if key == 'rdcomment':
+ continue
if key not in parameters:
raise AttributeError("'{}' object has no attribute '{}'"
.format(self.__class__.__name__, key))
# this validation can go away.
rd = self.__class__(*args)
dns.rdata.from_text(rd.rdclass, rd.rdtype, rd.to_text())
+ # The comment is not set in the constructor, so give it special
+ # handling.
+ rdcomment = kwargs.get('rdcomment', self.rdcomment)
+ if rdcomment is not None:
+ object.__setattr__(rd, 'rdcomment', rdcomment)
return rd
raise dns.exception.SyntaxError(
r'generic rdata does not start with \#')
length = tok.get_int()
- chunks = []
- while 1:
- token = tok.get()
- if token.is_eol_or_eof():
- break
- chunks.append(token.value.encode())
- hex = b''.join(chunks)
+ hex = tok.concatenate_remaining_identifiers().encode()
data = binascii.unhexlify(hex)
if len(data) != length:
raise dns.exception.SyntaxError(
rdclass = dns.rdataclass.RdataClass.make(rdclass)
rdtype = dns.rdatatype.RdataType.make(rdtype)
cls = get_rdata_class(rdclass, rdtype)
+ rdata = None
if cls != GenericRdata:
# peek at first token
token = tok.get()
# wire form from the generic syntax, and then run
# from_wire on it.
#
- rdata = GenericRdata.from_text(rdclass, rdtype, tok, origin,
- relativize, relativize_to)
- return from_wire(rdclass, rdtype, rdata.data, 0, len(rdata.data),
- origin)
- return cls.from_text(rdclass, rdtype, tok, origin, relativize,
- relativize_to)
+ grdata = GenericRdata.from_text(rdclass, rdtype, tok, origin,
+ relativize, relativize_to)
+ rdata = from_wire(rdclass, rdtype, grdata.data, 0, len(grdata.data),
+ origin)
+ if rdata is None:
+ rdata = cls.from_text(rdclass, rdtype, tok, origin, relativize,
+ relativize_to)
+ token = tok.get_eol_as_token()
+ if token.comment is not None:
+ object.__setattr__(rdata, 'rdcomment', token.comment)
+ return rdata
def from_wire_parser(rdclass, rdtype, parser, origin=None):
return not self.__eq__(other)
def to_text(self, name=None, origin=None, relativize=True,
- override_rdclass=None, **kw):
+ override_rdclass=None, want_comments=False, **kw):
"""Convert the rdataset into DNS master file format.
See ``dns.name.Name.choose_relativity`` for more information
*relativize*, a ``bool``. If ``True``, names will be relativized
to *origin*.
+
+ *want_comments*, a ``bool``. If ``True``, emit comments for rdata
+ which have them. The default is ``False``.
"""
if name is not None:
dns.rdatatype.to_text(self.rdtype)))
else:
for rd in self:
- s.write('%s%s%d %s %s %s\n' %
+ extra = ''
+ if want_comments:
+ if rd.rdcomment:
+ extra = f' ;{rd.rdcomment}'
+ s.write('%s%s%d %s %s %s%s\n' %
(ntext, pad, self.ttl, dns.rdataclass.to_text(rdclass),
dns.rdatatype.to_text(self.rdtype),
rd.to_text(origin=origin, relativize=relativize,
- **kw)))
+ **kw),
+ extra))
#
# We strip off the final \n for the caller's convenience in printing
#
latitude = tok.get_string()
longitude = tok.get_string()
altitude = tok.get_string()
- tok.get_eol()
return cls(rdclass, rdtype, latitude, longitude, altitude)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
relativize_to=None):
cpu = tok.get_string(max_length=255)
os = tok.get_string(max_length=255)
- tok.get_eol()
return cls(rdclass, rdtype, cpu, os)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
raise dns.exception.SyntaxError("HIT too long")
key = base64.b64decode(tok.get_string().encode())
servers = []
- while 1:
- token = tok.get()
- if token.is_eol_or_eof():
- break
+ for token in tok.get_remaining():
server = tok.as_name(token, origin, relativize, relativize_to)
servers.append(server)
return cls(rdclass, rdtype, hit, algorithm, key, servers)
def from_text(cls, rdclass, rdtype, tok, origin=None, relativize=True,
relativize_to=None):
address = tok.get_string()
- t = tok.get()
- if not t.is_eol_or_eof():
- tok.unget(t)
- subaddress = tok.get_string()
+ tokens = tok.get_remaining(max_tokens=1)
+ if len(tokens) >= 1:
+ subaddress = tokens[0].unescape().value
else:
- tok.unget(t)
subaddress = ''
- tok.get_eol()
return cls(rdclass, rdtype, address, subaddress)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
t = t[0: -1]
altitude = float(t) * 100.0 # m -> cm
- token = tok.get().unescape()
- if not token.is_eol_or_eof():
- value = token.value
+ tokens = tok.get_remaining(max_tokens=3)
+ if len(tokens) >= 1:
+ value = tokens[0].unescape().value
if value[-1] == 'm':
value = value[0: -1]
size = float(value) * 100.0 # m -> cm
- token = tok.get().unescape()
- if not token.is_eol_or_eof():
- value = token.value
+ if len(tokens) >= 2:
+ value = tokens[1].unescape().value
if value[-1] == 'm':
value = value[0: -1]
hprec = float(value) * 100.0 # m -> cm
- token = tok.get().unescape()
- if not token.is_eol_or_eof():
- value = token.value
+ if len(tokens) >= 3:
+ value = tokens[2].unescape().value
if value[-1] == 'm':
value = value[0: -1]
vprec = float(value) * 100.0 # m -> cm
- tok.get_eol()
# Try encoding these now so we raise if they are bad
_encode_size(size, "size")
salt = ''
else:
salt = binascii.unhexlify(salt.encode())
- tok.get_eol()
return cls(rdclass, rdtype, algorithm, flags, iterations, salt)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
relativize_to=None):
mbox = tok.get_name(origin, relativize, relativize_to)
txt = tok.get_name(origin, relativize, relativize_to)
- tok.get_eol()
return cls(rdclass, rdtype, mbox, txt)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
retry = tok.get_ttl()
expire = tok.get_ttl()
minimum = tok.get_ttl()
- tok.get_eol()
return cls(rdclass, rdtype, mname, rname, serial, refresh, retry,
expire, minimum)
target = tok.get().unescape()
if not (target.is_quoted_string() or target.is_identifier()):
raise dns.exception.SyntaxError("URI target must be a string")
- tok.get_eol()
return cls(rdclass, rdtype, priority, weight, target.value)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
def from_text(cls, rdclass, rdtype, tok, origin=None, relativize=True,
relativize_to=None):
address = tok.get_string()
- tok.get_eol()
return cls(rdclass, rdtype, address)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
relativize_to=None):
domain = tok.get_name(origin, relativize, relativize_to)
address = tok.get_uint16(base=8)
- tok.get_eol()
return cls(rdclass, rdtype, domain, address)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
def from_text(cls, rdclass, rdtype, tok, origin=None, relativize=True,
relativize_to=None):
address = tok.get_identifier()
- tok.get_eol()
return cls(rdclass, rdtype, address)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
def from_text(cls, rdclass, rdtype, tok, origin=None, relativize=True,
relativize_to=None):
address = tok.get_identifier()
- tok.get_eol()
return cls(rdclass, rdtype, address)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
def from_text(cls, rdclass, rdtype, tok, origin=None, relativize=True,
relativize_to=None):
items = []
- while True:
- token = tok.get().unescape()
- if token.is_eol_or_eof():
- break
- item = token.value
+ for token in tok.get_remaining():
+ item = token.unescape().value
if item[0] == '!':
negation = True
item = item[1:]
service = tok.get_string()
regexp = tok.get_string()
replacement = tok.get_name(origin, relativize, relativize_to)
- tok.get_eol()
return cls(rdclass, rdtype, order, preference, flags, service,
regexp, replacement)
def from_text(cls, rdclass, rdtype, tok, origin=None, relativize=True,
relativize_to=None):
address = tok.get_string()
- tok.get_eol()
if address[0:2] != '0x':
raise dns.exception.SyntaxError('string does not start with 0x')
address = address[2:].replace('.', '')
preference = tok.get_uint16()
map822 = tok.get_name(origin, relativize, relativize_to)
mapx400 = tok.get_name(origin, relativize, relativize_to)
- tok.get_eol()
return cls(rdclass, rdtype, preference, map822, mapx400)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
weight = tok.get_uint16()
port = tok.get_uint16()
target = tok.get_name(origin, relativize, relativize_to)
- tok.get_eol()
return cls(rdclass, rdtype, priority, weight, port, target)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
else:
protocol = socket.getprotobyname(protocol)
bitmap = bytearray()
- while 1:
- token = tok.get().unescape()
- if token.is_eol_or_eof():
- break
- if token.value.isdigit():
- serv = int(token.value)
+ for token in tok.get_remaining():
+ value = token.unescape().value
+ if value.isdigit():
+ serv = int(value)
else:
if protocol != _proto_udp and protocol != _proto_tcp:
raise NotImplementedError("protocol must be TCP or UDP")
protocol_text = "udp"
else:
protocol_text = "tcp"
- serv = socket.getservbyname(token.value, protocol_text)
+ serv = socket.getservbyname(value, protocol_text)
i = serv // 8
l = len(bitmap)
if l < i + 1:
def from_text(cls, rdclass, rdtype, tok, origin=None, relativize=True,
relativize_to=None):
text = tok.get_string()
- tok.get_eol()
if len(text) != cls.text_len:
raise dns.exception.SyntaxError(
'Input text must have %s characters' % cls.text_len)
relativize_to=None):
preference = tok.get_uint16()
exchange = tok.get_name(origin, relativize, relativize_to)
- tok.get_eol()
return cls(rdclass, rdtype, preference, exchange)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
def from_text(cls, rdclass, rdtype, tok, origin=None, relativize=True,
relativize_to=None):
target = tok.get_name(origin, relativize, relativize_to)
- tok.get_eol()
return cls(rdclass, rdtype, target)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
def from_text(cls, rdclass, rdtype, tok, origin=None, relativize=True,
relativize_to=None):
strings = []
- while 1:
- token = tok.get().unescape_to_bytes()
- if token.is_eol_or_eof():
- break
+ for token in tok.get_remaining():
+ token = token.unescape_to_bytes()
if not (token.is_quoted_string() or token.is_identifier()):
raise dns.exception.SyntaxError("expected a string")
if len(token.value) > 255:
def from_text(self, tok):
rdtypes = []
- while True:
- token = tok.get().unescape()
- if token.is_eol_or_eof():
- break
- rdtype = dns.rdatatype.from_text(token.value)
+ for token in tok.get_remaining():
+ rdtype = dns.rdatatype.from_text(token.unescape().value)
if rdtype == 0:
raise dns.exception.SyntaxError(f"{self.type_name} with bit 0")
rdtypes.append(rdtype)
has_escape: Does the token value contain escapes?
"""
- def __init__(self, ttype, value='', has_escape=False):
+ def __init__(self, ttype, value='', has_escape=False, comment=None):
"""Initialize a token instance."""
self.ttype = ttype
self.value = value
self.has_escape = has_escape
+ self.comment = comment
def is_eof(self):
return self.ttype == EOF
if self.multiline:
raise dns.exception.SyntaxError(
'unbalanced parentheses')
- return Token(EOF)
+ return Token(EOF, comment=token)
elif self.multiline:
self.skip_whitespace()
token = ''
continue
else:
- return Token(EOL, '\n')
+ return Token(EOL, '\n', comment=token)
else:
# This code exists in case we ever want a
# delimiter to be returned. It never produces
raise dns.exception.SyntaxError('expecting an identifier')
return token.value
+ def get_remaining(self, max_tokens=None):
+ """Return the remaining tokens on the line, until an EOL or EOF is seen.
+
+ max_tokens: If not None, stop after this number of tokens.
+
+ Returns a list of tokens.
+ """
+
+ tokens = []
+ while True:
+ token = self.get()
+ if token.is_eol_or_eof():
+ self.unget(token)
+ break
+ tokens.append(token)
+ if len(tokens) == max_tokens:
+ break
+ return tokens
+
def concatenate_remaining_identifiers(self):
"""Read the remaining tokens on the line, which should be identifiers.
while True:
token = self.get().unescape()
if token.is_eol_or_eof():
+ self.unget(token)
break
if not token.is_identifier():
raise dns.exception.SyntaxError
token = self.get()
return self.as_name(token, origin, relativize, relativize_to)
- def get_eol(self):
+ def get_eol_as_token(self):
"""Read the next token and raise an exception if it isn't EOL or
EOF.
raise dns.exception.SyntaxError(
'expected EOL or EOF, got %d "%s"' % (token.ttype,
token.value))
- return token.value
+ return token
+
+ def get_eol(self):
+ return self.get_eol_as_token().value
def get_ttl(self):
"""Read the next token and interpret it as a DNS TTL.
for rdata in rds:
yield (name, rds.ttl, rdata)
- def to_file(self, f, sorted=True, relativize=True, nl=None):
+ def to_file(self, f, sorted=True, relativize=True, nl=None,
+ want_comments=False):
"""Write a zone to a file.
*f*, a file or `str`. If *f* is a string, it is treated
*nl*, a ``str`` or None. The end of line string. If not
``None``, the output will use the platform's native
end-of-line marker (i.e. LF on POSIX, CRLF on Windows).
+
+ *want_comments*, a ``bool``. If ``True``, emit end-of-line comments
+ as part of writing the file. If ``False``, the default, do not
+ emit them.
"""
with contextlib.ExitStack() as stack:
names = self.keys()
for n in names:
l = self[n].to_text(n, origin=self.origin,
- relativize=relativize)
+ relativize=relativize,
+ want_comments=want_comments)
if isinstance(l, str):
l_b = l.encode(file_enc)
else:
f.write(l)
f.write(nl)
- def to_text(self, sorted=True, relativize=True, nl=None):
+ def to_text(self, sorted=True, relativize=True, nl=None,
+ want_comments=False):
"""Return a zone's text as though it were written to a file.
*sorted*, a ``bool``. If True, the default, then the file
``None``, the output will use the platform's native
end-of-line marker (i.e. LF on POSIX, CRLF on Windows).
+ *want_comments*, a ``bool``. If ``True``, emit end-of-line comments
+ as part of writing the file. If ``False``, the default, do not
+ emit them.
+
Returns a ``str``.
"""
temp_buffer = io.StringIO()
- self.to_file(temp_buffer, sorted, relativize, nl)
+ self.to_file(temp_buffer, sorted, relativize, nl, want_comments)
return_value = temp_buffer.getvalue()
temp_buffer.close()
return return_value
import dns.rrset
import dns.update
-def here(filename):
- return os.path.join(os.path.dirname(__file__), filename)
+from tests.util import here
query_text = """id 1234
opcode QUERY
import tests.stxt_module
import tests.ttxt_module
+from tests.util import here
class RdataTestCase(unittest.TestCase):
a1.replace(address="bogus")
self.assertRaises(dns.exception.SyntaxError, bad)
+ def test_replace_comment(self):
+ a1 = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.A,
+ "1.2.3.4 ;foo")
+ self.assertEqual(a1.rdcomment, "foo")
+ a2 = a1.replace(rdcomment="bar")
+ self.assertEqual(a1, a2)
+ self.assertEqual(a1.rdcomment, "foo")
+ self.assertEqual(a2.rdcomment, "bar")
+
def test_to_generic(self):
a = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.A, "1.2.3.4")
self.assertEqual(str(a.to_generic()), r'\# 4 01020304')
rdata = dns.rdata.from_wire('in', 'a', wire, 0, 4)
self.assertEqual(rdata, dns.rdata.from_text('in', 'a', '1.2.3.4'))
+ def test_unpickle(self):
+ expected_mx = dns.rdata.from_text('in', 'mx', '10 mx.example.')
+ with open(here('mx-2-0.pickle'), 'rb') as f:
+ mx = pickle.load(f)
+ self.assertEqual(mx, expected_mx)
+ self.assertIsNone(mx.rdcomment)
+
+
if __name__ == '__main__':
unittest.main()
import dns.zone
import dns.node
-def here(filename):
- return os.path.join(os.path.dirname(__file__), filename)
+from tests.util import here
example_text = """$TTL 3600
$ORIGIN example.
@ 300 ns ns2
"""
+example_comments_text = """$TTL 3600
+$ORIGIN example.
+@ soa foo bar (1 ; not kept
+2 3 4 5) ; kept
+@ ns ns1
+@ ns ns2
+ns1 a 10.0.0.1 ; comment1
+ns2 a 10.0.0.2 ; comment2
+"""
+
+example_comments_text_output = """@ 3600 IN SOA foo bar 1 2 3 4 5 ; kept
+@ 3600 IN NS ns1
+@ 3600 IN NS ns2
+ns1 3600 IN A 10.0.0.1 ; comment1
+ns2 3600 IN A 10.0.0.2 ; comment2
+"""
+
_keep_output = True
def _rdata_sort(a):
self.assertEqual(z._validate_name('foo.bar.example.'),
dns.name.from_text('foo.bar', None))
+ def testComments(self):
+ z = dns.zone.from_text(example_comments_text, 'example.',
+ relativize=True)
+ f = StringIO()
+ z.to_file(f, want_comments=True)
+ out = f.getvalue()
+ f.close()
+ self.assertEqual(out, example_comments_text_output)
+
if __name__ == '__main__':
unittest.main()
--- /dev/null
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+# Copyright (C) 2003-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.
+
+import os.path
+
+def here(filename):
+ return os.path.join(os.path.dirname(__file__), filename)
--- /dev/null
+import pickle
+import sys
+
+import dns.rdata
+import dns.version
+
+# Generate a pickled mx RR for the current dnspython version
+
+mx = dns.rdata.from_text('in', 'mx', '10 mx.example.')
+filename = f'pickled-{dns.version.MAJOR}-{dns.version.MINOR}.pickle'
+with open(filename, 'wb') as f:
+ pickle.dump(mx, f)
+with open(filename, 'rb') as f:
+ mx2 = pickle.load(f)
+if mx == mx2:
+ print('ok')
+else:
+ print('DIFFERENT!')
+ sys.exit(1)
projects / thirdparty / dnspython.git / commitdiff
