third_party/dnspython/dns/rdata.py

   1 # Copyright (C) 2001-2007, 2009-2011 Nominum, Inc.
   2 #
   3 # Permission to use, copy, modify, and distribute this software and its
   4 # documentation for any purpose with or without fee is hereby granted,
   5 # provided that the above copyright notice and this permission notice
   6 # appear in all copies.
   7 #
   8 # THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
   9 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  10 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
  11 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
  12 # WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
  13 # ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
  14 # OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  15
  16 """DNS rdata.
  17
  18 @var _rdata_modules: A dictionary mapping a (rdclass, rdtype) tuple to
  19 the module which implements that type.
  20 @type _rdata_modules: dict
  21 @var _module_prefix: The prefix to use when forming modules names.  The
  22 default is 'dns.rdtypes'.  Changing this value will break the library.
  23 @type _module_prefix: string
  24 @var _hex_chunk: At most this many octets that will be represented in each
  25 chunk of hexstring that _hexify() produces before whitespace occurs.
  26 @type _hex_chunk: int"""
  27
  28 import cStringIO
  29
  30 import dns.exception
  31 import dns.name
  32 import dns.rdataclass
  33 import dns.rdatatype
  34 import dns.tokenizer
  35 import dns.wiredata
  36
  37 _hex_chunksize = 32
  38
  39 def _hexify(data, chunksize=None):
  40     """Convert a binary string into its hex encoding, broken up into chunks
  41     of I{chunksize} characters separated by a space.
  42
  43     @param data: the binary string
  44     @type data: string
  45     @param chunksize: the chunk size.  Default is L{dns.rdata._hex_chunksize}
  46     @rtype: string
  47     """
  48
  49     if chunksize is None:
  50         chunksize = _hex_chunksize
  51     hex = data.encode('hex_codec')
  52     l = len(hex)
  53     if l > chunksize:
  54         chunks = []
  55         i = 0
  56         while i < l:
  57             chunks.append(hex[i : i + chunksize])
  58             i += chunksize
  59         hex = ' '.join(chunks)
  60     return hex
  61
  62 _base64_chunksize = 32
  63
  64 def _base64ify(data, chunksize=None):
  65     """Convert a binary string into its base64 encoding, broken up into chunks
  66     of I{chunksize} characters separated by a space.
  67
  68     @param data: the binary string
  69     @type data: string
  70     @param chunksize: the chunk size.  Default is
  71     L{dns.rdata._base64_chunksize}
  72     @rtype: string
  73     """
  74
  75     if chunksize is None:
  76         chunksize = _base64_chunksize
  77     b64 = data.encode('base64_codec')
  78     b64 = b64.replace('\n', '')
  79     l = len(b64)
  80     if l > chunksize:
  81         chunks = []
  82         i = 0
  83         while i < l:
  84             chunks.append(b64[i : i + chunksize])
  85             i += chunksize
  86         b64 = ' '.join(chunks)
  87     return b64
  88
  89 __escaped = {
  90     '"' : True,
  91     '\\' : True,
  92     }
  93
  94 def _escapify(qstring):
  95     """Escape the characters in a quoted string which need it.
  96
  97     @param qstring: the string
  98     @type qstring: string
  99     @returns: the escaped string
 100     @rtype: string
 101     """
 102
 103     text = ''
 104     for c in qstring:
 105         if c in __escaped:
 106             text += '\\' + c
 107         elif ord(c) >= 0x20 and ord(c) < 0x7F:
 108             text += c
 109         else:
 110             text += '\\%03d' % ord(c)
 111     return text
 112
 113 def _truncate_bitmap(what):
 114     """Determine the index of greatest byte that isn't all zeros, and
 115     return the bitmap that contains all the bytes less than that index.
 116
 117     @param what: a string of octets representing a bitmap.
 118     @type what: string
 119     @rtype: string
 120     """
 121
 122     for i in xrange(len(what) - 1, -1, -1):
 123         if what[i] != '\x00':
 124             break
 125     return ''.join(what[0 : i + 1])
 126
 127 class Rdata(object):
 128     """Base class for all DNS rdata types.
 129     """
 130
 131     __slots__ = ['rdclass', 'rdtype']
 132
 133     def __init__(self, rdclass, rdtype):
 134         """Initialize an rdata.
 135         @param rdclass: The rdata class
 136         @type rdclass: int
 137         @param rdtype: The rdata type
 138         @type rdtype: int
 139         """
 140
 141         self.rdclass = rdclass
 142         self.rdtype = rdtype
 143
 144     def covers(self):
 145         """DNS SIG/RRSIG rdatas apply to a specific type; this type is
 146         returned by the covers() function.  If the rdata type is not
 147         SIG or RRSIG, dns.rdatatype.NONE is returned.  This is useful when
 148         creating rdatasets, allowing the rdataset to contain only RRSIGs
 149         of a particular type, e.g. RRSIG(NS).
 150         @rtype: int
 151         """
 152
 153         return dns.rdatatype.NONE
 154
 155     def extended_rdatatype(self):
 156         """Return a 32-bit type value, the least significant 16 bits of
 157         which are the ordinary DNS type, and the upper 16 bits of which are
 158         the "covered" type, if any.
 159         @rtype: int
 160         """
 161
 162         return self.covers() << 16 | self.rdtype
 163
 164     def to_text(self, origin=None, relativize=True, **kw):
 165         """Convert an rdata to text format.
 166         @rtype: string
 167         """
 168         raise NotImplementedError
 169
 170     def to_wire(self, file, compress = None, origin = None):
 171         """Convert an rdata to wire format.
 172         @rtype: string
 173         """
 174
 175         raise NotImplementedError
 176
 177     def to_digestable(self, origin = None):
 178         """Convert rdata to a format suitable for digesting in hashes.  This
 179         is also the DNSSEC canonical form."""
 180         f = cStringIO.StringIO()
 181         self.to_wire(f, None, origin)
 182         return f.getvalue()
 183
 184     def validate(self):
 185         """Check that the current contents of the rdata's fields are
 186         valid.  If you change an rdata by assigning to its fields,
 187         it is a good idea to call validate() when you are done making
 188         changes.
 189         """
 190         dns.rdata.from_text(self.rdclass, self.rdtype, self.to_text())
 191
 192     def __repr__(self):
 193         covers = self.covers()
 194         if covers == dns.rdatatype.NONE:
 195             ctext = ''
 196         else:
 197             ctext = '(' + dns.rdatatype.to_text(covers) + ')'
 198         return '<DNS ' + dns.rdataclass.to_text(self.rdclass) + ' ' + \
 199                dns.rdatatype.to_text(self.rdtype) + ctext + ' rdata: ' + \
 200                str(self) + '>'
 201
 202     def __str__(self):
 203         return self.to_text()
 204
 205     def _cmp(self, other):
 206         """Compare an rdata with another rdata of the same rdtype and
 207         rdclass.  Return < 0 if self < other in the DNSSEC ordering,
 208         0 if self == other, and > 0 if self > other.
 209         """
 210
 211         raise NotImplementedError
 212
 213     def __eq__(self, other):
 214         if not isinstance(other, Rdata):
 215             return False
 216         if self.rdclass != other.rdclass or \
 217            self.rdtype != other.rdtype:
 218             return False
 219         return self._cmp(other) == 0
 220
 221     def __ne__(self, other):
 222         if not isinstance(other, Rdata):
 223             return True
 224         if self.rdclass != other.rdclass or \
 225            self.rdtype != other.rdtype:
 226             return True
 227         return self._cmp(other) != 0
 228
 229     def __lt__(self, other):
 230         if not isinstance(other, Rdata) or \
 231                self.rdclass != other.rdclass or \
 232                self.rdtype != other.rdtype:
 233             return NotImplemented
 234         return self._cmp(other) < 0
 235
 236     def __le__(self, other):
 237         if not isinstance(other, Rdata) or \
 238                self.rdclass != other.rdclass or \
 239                self.rdtype != other.rdtype:
 240             return NotImplemented
 241         return self._cmp(other) <= 0
 242
 243     def __ge__(self, other):
 244         if not isinstance(other, Rdata) or \
 245                self.rdclass != other.rdclass or \
 246                self.rdtype != other.rdtype:
 247             return NotImplemented
 248         return self._cmp(other) >= 0
 249
 250     def __gt__(self, other):
 251         if not isinstance(other, Rdata) or \
 252                self.rdclass != other.rdclass or \
 253                self.rdtype != other.rdtype:
 254             return NotImplemented
 255         return self._cmp(other) > 0
 256
 257     def __hash__(self):
 258         return hash(self.to_digestable(dns.name.root))
 259
 260     def _wire_cmp(self, other):
 261         # A number of types compare rdata in wire form, so we provide
 262         # the method here instead of duplicating it.
 263         #
 264         # We specifiy an arbitrary origin of '.' when doing the
 265         # comparison, since the rdata may have relative names and we
 266         # can't convert a relative name to wire without an origin.
 267         b1 = cStringIO.StringIO()
 268         self.to_wire(b1, None, dns.name.root)
 269         b2 = cStringIO.StringIO()
 270         other.to_wire(b2, None, dns.name.root)
 271         return cmp(b1.getvalue(), b2.getvalue())
 272
 273     def from_text(cls, rdclass, rdtype, tok, origin = None, relativize = True):
 274         """Build an rdata object from text format.
 275
 276         @param rdclass: The rdata class
 277         @type rdclass: int
 278         @param rdtype: The rdata type
 279         @type rdtype: int
 280         @param tok: The tokenizer
 281         @type tok: dns.tokenizer.Tokenizer
 282         @param origin: The origin to use for relative names
 283         @type origin: dns.name.Name
 284         @param relativize: should names be relativized?
 285         @type relativize: bool
 286         @rtype: dns.rdata.Rdata instance
 287         """
 288
 289         raise NotImplementedError
 290
 291     from_text = classmethod(from_text)
 292
 293     def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
 294         """Build an rdata object from wire format
 295
 296         @param rdclass: The rdata class
 297         @type rdclass: int
 298         @param rdtype: The rdata type
 299         @type rdtype: int
 300         @param wire: The wire-format message
 301         @type wire: string
 302         @param current: The offet in wire of the beginning of the rdata.
 303         @type current: int
 304         @param rdlen: The length of the wire-format rdata
 305         @type rdlen: int
 306         @param origin: The origin to use for relative names
 307         @type origin: dns.name.Name
 308         @rtype: dns.rdata.Rdata instance
 309         """
 310
 311         raise NotImplementedError
 312
 313     from_wire = classmethod(from_wire)
 314
 315     def choose_relativity(self, origin = None, relativize = True):
 316         """Convert any domain names in the rdata to the specified
 317         relativization.
 318         """
 319
 320         pass
 321
 322
 323 class GenericRdata(Rdata):
 324     """Generate Rdata Class
 325
 326     This class is used for rdata types for which we have no better
 327     implementation.  It implements the DNS "unknown RRs" scheme.
 328     """
 329
 330     __slots__ = ['data']
 331
 332     def __init__(self, rdclass, rdtype, data):
 333         super(GenericRdata, self).__init__(rdclass, rdtype)
 334         self.data = data
 335
 336     def to_text(self, origin=None, relativize=True, **kw):
 337         return r'\# %d ' % len(self.data) + _hexify(self.data)
 338
 339     def from_text(cls, rdclass, rdtype, tok, origin = None, relativize = True):
 340         token = tok.get()
 341         if not token.is_identifier() or token.value != '\#':
 342             raise dns.exception.SyntaxError(r'generic rdata does not start with \#')
 343         length = tok.get_int()
 344         chunks = []
 345         while 1:
 346             token = tok.get()
 347             if token.is_eol_or_eof():
 348                 break
 349             chunks.append(token.value)
 350         hex = ''.join(chunks)
 351         data = hex.decode('hex_codec')
 352         if len(data) != length:
 353             raise dns.exception.SyntaxError('generic rdata hex data has wrong length')
 354         return cls(rdclass, rdtype, data)
 355
 356     from_text = classmethod(from_text)
 357
 358     def to_wire(self, file, compress = None, origin = None):
 359         file.write(self.data)
 360
 361     def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin = None):
 362         return cls(rdclass, rdtype, wire[current : current + rdlen])
 363
 364     from_wire = classmethod(from_wire)
 365
 366     def _cmp(self, other):
 367         return cmp(self.data, other.data)
 368
 369 _rdata_modules = {}
 370 _module_prefix = 'dns.rdtypes'
 371
 372 def get_rdata_class(rdclass, rdtype):
 373
 374     def import_module(name):
 375         mod = __import__(name)
 376         components = name.split('.')
 377         for comp in components[1:]:
 378             mod = getattr(mod, comp)
 379         return mod
 380
 381     mod = _rdata_modules.get((rdclass, rdtype))
 382     rdclass_text = dns.rdataclass.to_text(rdclass)
 383     rdtype_text = dns.rdatatype.to_text(rdtype)
 384     rdtype_text = rdtype_text.replace('-', '_')
 385     if not mod:
 386         mod = _rdata_modules.get((dns.rdatatype.ANY, rdtype))
 387         if not mod:
 388             try:
 389                 mod = import_module('.'.join([_module_prefix,
 390                                               rdclass_text, rdtype_text]))
 391                 _rdata_modules[(rdclass, rdtype)] = mod
 392             except ImportError:
 393                 try:
 394                     mod = import_module('.'.join([_module_prefix,
 395                                                   'ANY', rdtype_text]))
 396                     _rdata_modules[(dns.rdataclass.ANY, rdtype)] = mod
 397                 except ImportError:
 398                     mod = None
 399     if mod:
 400         cls = getattr(mod, rdtype_text)
 401     else:
 402         cls = GenericRdata
 403     return cls
 404
 405 def from_text(rdclass, rdtype, tok, origin = None, relativize = True):
 406     """Build an rdata object from text format.
 407
 408     This function attempts to dynamically load a class which
 409     implements the specified rdata class and type.  If there is no
 410     class-and-type-specific implementation, the GenericRdata class
 411     is used.
 412
 413     Once a class is chosen, its from_text() class method is called
 414     with the parameters to this function.
 415
 416     If I{tok} is a string, then a tokenizer is created and the string
 417     is used as its input.
 418
 419     @param rdclass: The rdata class
 420     @type rdclass: int
 421     @param rdtype: The rdata type
 422     @type rdtype: int
 423     @param tok: The tokenizer or input text
 424     @type tok: dns.tokenizer.Tokenizer or string
 425     @param origin: The origin to use for relative names
 426     @type origin: dns.name.Name
 427     @param relativize: Should names be relativized?
 428     @type relativize: bool
 429     @rtype: dns.rdata.Rdata instance"""
 430
 431     if isinstance(tok, str):
 432         tok = dns.tokenizer.Tokenizer(tok)
 433     cls = get_rdata_class(rdclass, rdtype)
 434     if cls != GenericRdata:
 435         # peek at first token
 436         token = tok.get()
 437         tok.unget(token)
 438         if token.is_identifier() and \
 439            token.value == r'\#':
 440             #
 441             # Known type using the generic syntax.  Extract the
 442             # wire form from the generic syntax, and then run
 443             # from_wire on it.
 444             #
 445             rdata = GenericRdata.from_text(rdclass, rdtype, tok, origin,
 446                                            relativize)
 447             return from_wire(rdclass, rdtype, rdata.data, 0, len(rdata.data),
 448                              origin)
 449     return cls.from_text(rdclass, rdtype, tok, origin, relativize)
 450
 451 def from_wire(rdclass, rdtype, wire, current, rdlen, origin = None):
 452     """Build an rdata object from wire format
 453
 454     This function attempts to dynamically load a class which
 455     implements the specified rdata class and type.  If there is no
 456     class-and-type-specific implementation, the GenericRdata class
 457     is used.
 458
 459     Once a class is chosen, its from_wire() class method is called
 460     with the parameters to this function.
 461
 462     @param rdclass: The rdata class
 463     @type rdclass: int
 464     @param rdtype: The rdata type
 465     @type rdtype: int
 466     @param wire: The wire-format message
 467     @type wire: string
 468     @param current: The offet in wire of the beginning of the rdata.
 469     @type current: int
 470     @param rdlen: The length of the wire-format rdata
 471     @type rdlen: int
 472     @param origin: The origin to use for relative names
 473     @type origin: dns.name.Name
 474     @rtype: dns.rdata.Rdata instance"""
 475
 476     wire = dns.wiredata.maybe_wrap(wire)
 477     cls = get_rdata_class(rdclass, rdtype)
 478     return cls.from_wire(rdclass, rdtype, wire, current, rdlen, origin)