liteeth/test/model/icmp.py

   1 import math
   2
   3 from liteeth.common import *
   4 from liteeth.test.common import *
   5
   6 from liteeth.test.model import ip
   7
   8 def print_icmp(s):
   9         print_with_prefix(s, "[ICMP]")
  10
  11 # ICMP model
  12 class ICMPPacket(Packet):
  13         def __init__(self, init=[]):
  14                 Packet.__init__(self, init)
  15
  16         def decode(self):
  17                 header = []
  18                 for byte in self[:icmp_header_len]:
  19                         header.append(self.pop(0))
  20                 for k, v in sorted(icmp_header.items()):
  21                         setattr(self, k, get_field_data(v, header))
  22
  23         def encode(self):
  24                 header = 0
  25                 for k, v in sorted(icmp_header.items()):
  26                         value = merge_bytes(split_bytes(getattr(self, k), math.ceil(v.width/8)), "little")
  27                         header += (value << v.offset+(v.byte*8))
  28                 for d in split_bytes(header, icmp_header_len):
  29                         self.insert(0, d)
  30
  31         def __repr__(self):
  32                 r = "--------\n"
  33                 for k in sorted(icmp_header.keys()):
  34                         r += k + " : 0x%x" %getattr(self,k) + "\n"
  35                 r += "payload: "
  36                 for d in self:
  37                         r += "%02x" %d
  38                 return r
  39
  40 class ICMP(Module):
  41         def  __init__(self, ip, ip_address, debug=False):
  42                 self.ip = ip
  43                 self.ip_address = ip_address
  44                 self.debug = debug
  45                 self.tx_packets = []
  46                 self.tx_packet = ICMPPacket()
  47                 self.rx_packet = ICMPPacket()
  48
  49                 self.ip.set_icmp_callback(self.callback)
  50
  51         def send(self, packet):
  52                 packet.encode()
  53                 if self.debug:
  54                         print_icmp(">>>>>>>>")
  55                         print_icmp(packet)
  56                 ip_packet = ip.IPPacket(packet)
  57                 ip_packet.version = 0x4
  58                 ip_packet.ihl = 0x5
  59                 ip_packet.total_length = len(packet) + ip_packet.ihl
  60                 ip_packet.identification = 0
  61                 ip_packet.flags = 0
  62                 ip_packet.fragment_offset = 0
  63                 ip_packet.ttl = 0x80
  64                 ip_packet.sender_ip = self.ip_address
  65                 ip_packet.target_ip = 0x12345678 # XXX
  66                 ip_packet.checksum = 0
  67                 ip_packet.protocol = icmp_protocol
  68                 self.ip.send(ip_packet)
  69
  70         def callback(self, packet):
  71                 packet = ICMPPacket(packet)
  72                 packet.decode()
  73                 if self.debug:
  74                         print_icmp("<<<<<<<<")
  75                         print_icmp(packet)
  76                 self.process(packet)
  77
  78         def process(self, packet):
  79                 pass
  80
  81 if __name__ == "__main__":
  82         from liteeth.test.model.dumps import *
  83         from liteeth.test.model.mac import *
  84         from liteeth.test.model.ip import *
  85         errors = 0
  86         # ICMP packet
  87         packet = MACPacket(ping_request)
  88         packet.decode_remove_header()
  89         #print(packet)
  90         packet = IPPacket(packet)
  91         packet.decode()
  92         #print(packet)
  93         packet = ICMPPacket(packet)
  94         packet.decode()
  95         #print(packet)
  96         errors += verify_packet(packet, ping_request_infos)
  97         packet.encode()
  98         packet.decode()
  99         #print(packet)
 100         errors += verify_packet(packet, ping_request_infos)
 101
 102         print("icmp errors " + str(errors))