(self, data)
| 109 | raise NotImplementedError |
| 110 | |
| 111 | def decodePacket(self, data): |
| 112 | if len(data) < self.HEADER_STRUCT_SIZE: |
| 113 | raise Exception |
| 114 | |
| 115 | trn_id, code, question_count, answer_count, authority_count, additional_count = \ |
| 116 | struct.unpack(self.HEADER_STRUCT_FORMAT, data[:self.HEADER_STRUCT_SIZE]) |
| 117 | |
| 118 | is_response = bool((code >> 15) & 0x01) |
| 119 | opcode = (code >> 11) & 0x0F |
| 120 | flags = (code >> 4) & 0x7F |
| 121 | rcode = code & 0x0F |
| 122 | |
| 123 | if opcode == 0x0000 and is_response: |
| 124 | name_len = data[self.HEADER_STRUCT_SIZE] |
| 125 | # Constant 2 for the padding bytes before/after the Name and constant 8 for the Type, |
| 126 | # Class and TTL fields in the Answer section after the Name: |
| 127 | offset = self.HEADER_STRUCT_SIZE + 2 + name_len + 8 |
| 128 | record_count = (struct.unpack('>H', data[offset:offset+2])[0]) // 6 |
| 129 | |
| 130 | offset += 4 # Constant 4 for the Data Length and Flags field |
| 131 | ret = [] |
| 132 | for i in range(0, record_count): |
| 133 | ret.append('%d.%d.%d.%d' % struct.unpack('4B', (data[offset:offset + 4]))) |
| 134 | offset += 6 |
| 135 | return trn_id, ret |
| 136 | else: |
| 137 | return trn_id, None |
| 138 | |
| 139 | def prepareNameQuery(self, trn_id, name, is_broadcast = True): |
| 140 | header = struct.pack(self.HEADER_STRUCT_FORMAT, |
no test coverage detected