Source Code for Module paramiko.ber

  1  # Copyright (C) 2003-2007  Robey Pointer <robey@lag.net> 
  2  # 
  3  # This file is part of paramiko. 
  4  # 
  5  # Paramiko is free software; you can redistribute it and/or modify it under the 
  6  # terms of the GNU Lesser General Public License as published by the Free 
  7  # Software Foundation; either version 2.1 of the License, or (at your option) 
  8  # any later version. 
  9  # 
 10  # Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY 
 11  # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 
 12  # A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more 
 13  # details. 
 14  # 
 15  # You should have received a copy of the GNU Lesser General Public License 
 16  # along with Paramiko; if not, write to the Free Software Foundation, Inc., 
 17  # 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA. 
 18   
 19   
 20  import util 
 21   
 22   
 23 -class BERException (Exception): 
 24      pass 
 25   
 26   
 27 -class BER(object): 
 28      """ 
 29      Robey's tiny little attempt at a BER decoder. 
 30      """ 
 31   
 32 -    def __init__(self, content=''): 
 33          self.content = content 
 34          self.idx = 0 
 35   
 36 -    def __str__(self): 
 37          return self.content 
 38   
 39 -    def __repr__(self): 
 40          return 'BER(\'' + repr(self.content) + '\')' 
 41   
 42 -    def decode(self): 
 43          return self.decode_next() 
 44       
 45 -    def decode_next(self): 
 46          if self.idx >= len(self.content): 
 47              return None 
 48          ident = ord(self.content[self.idx]) 
 49          self.idx += 1 
 50          if (ident & 31) == 31: 
 51              # identifier > 30 
 52              ident = 0 
 53              while self.idx < len(self.content): 
 54                  t = ord(self.content[self.idx]) 
 55                  self.idx += 1 
 56                  ident = (ident << 7) | (t & 0x7f) 
 57                  if not (t & 0x80): 
 58                      break 
 59          if self.idx >= len(self.content): 
 60              return None 
 61          # now fetch length 
 62          size = ord(self.content[self.idx]) 
 63          self.idx += 1 
 64          if size & 0x80: 
 65              # more complimicated... 
 66              # FIXME: theoretically should handle indefinite-length (0x80) 
 67              t = size & 0x7f 
 68              if self.idx + t > len(self.content): 
 69                  return None 
 70              size = util.inflate_long(self.content[self.idx : self.idx + t], True) 
 71              self.idx += t 
 72          if self.idx + size > len(self.content): 
 73              # can't fit 
 74              return None 
 75          data = self.content[self.idx : self.idx + size] 
 76          self.idx += size 
 77          # now switch on id 
 78          if ident == 0x30: 
 79              # sequence 
 80              return self.decode_sequence(data) 
 81          elif ident == 2: 
 82              # int 
 83              return util.inflate_long(data) 
 84          else: 
 85              # 1: boolean (00 false, otherwise true) 
 86              raise BERException('Unknown ber encoding type %d (robey is lazy)' % ident) 
 87   
 88 -    def decode_sequence(data): 
 89          out = [] 
 90          b = BER(data) 
 91          while True: 
 92              x = b.decode_next() 
 93              if x is None: 
 94                  break 
 95              out.append(x) 
 96          return out 
 97      decode_sequence = staticmethod(decode_sequence) 
 98   
 99 -    def encode_tlv(self, ident, val): 
100          # no need to support ident > 31 here 
101          self.content += chr(ident) 
102          if len(val) > 0x7f: 
103              lenstr = util.deflate_long(len(val)) 
104              self.content += chr(0x80 + len(lenstr)) + lenstr 
105          else: 
106              self.content += chr(len(val)) 
107          self.content += val 
108   
109 -    def encode(self, x): 
110          if type(x) is bool: 
111              if x: 
112                  self.encode_tlv(1, '\xff') 
113              else: 
114                  self.encode_tlv(1, '\x00') 
115          elif (type(x) is int) or (type(x) is long): 
116              self.encode_tlv(2, util.deflate_long(x)) 
117          elif type(x) is str: 
118              self.encode_tlv(4, x) 
119          elif (type(x) is list) or (type(x) is tuple): 
120              self.encode_tlv(0x30, self.encode_sequence(x)) 
121          else: 
122              raise BERException('Unknown type for encoding: %s' % repr(type(x))) 
123   
124 -    def encode_sequence(data): 
125          b = BER() 
126          for item in data: 
127              b.encode(item) 
128          return str(b) 
129      encode_sequence = staticmethod(encode_sequence) 
130