(self, aBuffer)
| 128 | pass |
| 129 | |
| 130 | def decode(self, aBuffer): |
| 131 | i = ImpactPacket.IP(aBuffer) |
| 132 | self.set_decoded_protocol ( i ) |
| 133 | off = i.get_header_size() |
| 134 | end = i.get_ip_len() |
| 135 | # If ip_len == 0 we might be facing TCP segmentation offload, let's calculate the right len |
| 136 | if end == 0: |
| 137 | LOG.warning('IP len reported as 0, most probably because of TCP segmentation offload. Attempting to fix its size') |
| 138 | i.set_ip_len(len(aBuffer)) |
| 139 | end = i.get_ip_len() |
| 140 | |
| 141 | if i.get_ip_p() == ImpactPacket.UDP.protocol: |
| 142 | self.udp_decoder = UDPDecoder() |
| 143 | packet = self.udp_decoder.decode(aBuffer[off:end]) |
| 144 | elif i.get_ip_p() == ImpactPacket.TCP.protocol: |
| 145 | self.tcp_decoder = TCPDecoder() |
| 146 | packet = self.tcp_decoder.decode(aBuffer[off:end]) |
| 147 | elif i.get_ip_p() == ImpactPacket.ICMP.protocol: |
| 148 | self.icmp_decoder = ICMPDecoder() |
| 149 | packet = self.icmp_decoder.decode(aBuffer[off:end]) |
| 150 | elif i.get_ip_p() == ImpactPacket.IGMP.protocol: |
| 151 | self.igmp_decoder = IGMPDecoder() |
| 152 | packet = self.igmp_decoder.decode(aBuffer[off:end]) |
| 153 | else: |
| 154 | self.data_decoder = DataDecoder() |
| 155 | packet = self.data_decoder.decode(aBuffer[off:end]) |
| 156 | i.contains(packet) |
| 157 | return i |
| 158 | |
| 159 | class IP6MultiProtocolDecoder(Decoder): |
| 160 | def __init__(self, a_protocol_id): |
nothing calls this directly
no test coverage detected