Source Code for Module dns.rdtypes.IN.IPSECKEY

  1  # Copyright (C) 2006, 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  import struct 
 17  import base64 
 18   
 19  import dns.exception 
 20  import dns.inet 
 21  import dns.name 
 22   
 23   
 24 -class IPSECKEY(dns.rdata.Rdata): 
 25   
 26      """IPSECKEY record 
 27   
 28      @ivar precedence: the precedence for this key data 
 29      @type precedence: int 
 30      @ivar gateway_type: the gateway type 
 31      @type gateway_type: int 
 32      @ivar algorithm: the algorithm to use 
 33      @type algorithm: int 
 34      @ivar gateway: the public key 
 35      @type gateway: None, IPv4 address, IPV6 address, or domain name 
 36      @ivar key: the public key 
 37      @type key: string 
 38      @see: RFC 4025""" 
 39   
 40      __slots__ = ['precedence', 'gateway_type', 'algorithm', 'gateway', 'key'] 
 41   
 42 -    def __init__(self, rdclass, rdtype, precedence, gateway_type, algorithm, 
 43                   gateway, key): 
 44          super(IPSECKEY, self).__init__(rdclass, rdtype) 
 45          if gateway_type == 0: 
 46              if gateway != '.' and gateway is not None: 
 47                  raise SyntaxError('invalid gateway for gateway type 0') 
 48              gateway = None 
 49          elif gateway_type == 1: 
 50              # check that it's OK 
 51              dns.inet.inet_pton(dns.inet.AF_INET, gateway) 
 52          elif gateway_type == 2: 
 53              # check that it's OK 
 54              dns.inet.inet_pton(dns.inet.AF_INET6, gateway) 
 55          elif gateway_type == 3: 
 56              pass 
 57          else: 
 58              raise SyntaxError( 
 59                  'invalid IPSECKEY gateway type: %d' % gateway_type) 
 60          self.precedence = precedence 
 61          self.gateway_type = gateway_type 
 62          self.algorithm = algorithm 
 63          self.gateway = gateway 
 64          self.key = key 
 65   
 66 -    def to_text(self, origin=None, relativize=True, **kw): 
 67          if self.gateway_type == 0: 
 68              gateway = '.' 
 69          elif self.gateway_type == 1: 
 70              gateway = self.gateway 
 71          elif self.gateway_type == 2: 
 72              gateway = self.gateway 
 73          elif self.gateway_type == 3: 
 74              gateway = str(self.gateway.choose_relativity(origin, relativize)) 
 75          else: 
 76              raise ValueError('invalid gateway type') 
 77          return '%d %d %d %s %s' % (self.precedence, self.gateway_type, 
 78                                     self.algorithm, gateway, 
 79                                     dns.rdata._base64ify(self.key)) 
 80   
 81      @classmethod 
 82 -    def from_text(cls, rdclass, rdtype, tok, origin=None, relativize=True): 
 83          precedence = tok.get_uint8() 
 84          gateway_type = tok.get_uint8() 
 85          algorithm = tok.get_uint8() 
 86          if gateway_type == 3: 
 87              gateway = tok.get_name().choose_relativity(origin, relativize) 
 88          else: 
 89              gateway = tok.get_string() 
 90          chunks = [] 
 91          while 1: 
 92              t = tok.get().unescape() 
 93              if t.is_eol_or_eof(): 
 94                  break 
 95              if not t.is_identifier(): 
 96                  raise dns.exception.SyntaxError 
 97              chunks.append(t.value.encode()) 
 98          b64 = b''.join(chunks) 
 99          key = base64.b64decode(b64) 
100          return cls(rdclass, rdtype, precedence, gateway_type, algorithm, 
101                     gateway, key) 
102   
103 -    def to_wire(self, file, compress=None, origin=None): 
104          header = struct.pack("!BBB", self.precedence, self.gateway_type, 
105                               self.algorithm) 
106          file.write(header) 
107          if self.gateway_type == 0: 
108              pass 
109          elif self.gateway_type == 1: 
110              file.write(dns.inet.inet_pton(dns.inet.AF_INET, self.gateway)) 
111          elif self.gateway_type == 2: 
112              file.write(dns.inet.inet_pton(dns.inet.AF_INET6, self.gateway)) 
113          elif self.gateway_type == 3: 
114              self.gateway.to_wire(file, None, origin) 
115          else: 
116              raise ValueError('invalid gateway type') 
117          file.write(self.key) 
118   
119      @classmethod 
120 -    def from_wire(cls, rdclass, rdtype, wire, current, rdlen, origin=None): 
121          if rdlen < 3: 
122              raise dns.exception.FormError 
123          header = struct.unpack('!BBB', wire[current: current + 3]) 
124          gateway_type = header[1] 
125          current += 3 
126          rdlen -= 3 
127          if gateway_type == 0: 
128              gateway = None 
129          elif gateway_type == 1: 
130              gateway = dns.inet.inet_ntop(dns.inet.AF_INET, 
131                                           wire[current: current + 4]) 
132              current += 4 
133              rdlen -= 4 
134          elif gateway_type == 2: 
135              gateway = dns.inet.inet_ntop(dns.inet.AF_INET6, 
136                                           wire[current: current + 16]) 
137              current += 16 
138              rdlen -= 16 
139          elif gateway_type == 3: 
140              (gateway, cused) = dns.name.from_wire(wire[: current + rdlen], 
141                                                    current) 
142              current += cused 
143              rdlen -= cused 
144          else: 
145              raise dns.exception.FormError('invalid IPSECKEY gateway type') 
146          key = wire[current: current + rdlen].unwrap() 
147          return cls(rdclass, rdtype, header[0], gateway_type, header[2], 
148                     gateway, key) 
149