liteeth/test/model/ip.py

   1 import math, binascii
   2
   3 from liteeth.common import *
   4 from liteeth.mac.common import *
   5 from liteeth.test.common import *
   6
   7 from liteeth.test.model import mac
   8
   9 def print_ip(s):
  10         print_with_prefix(s, "[IP]")
  11
  12 preamble = split_bytes(eth_preamble, 8)
  13
  14 # IP model
  15 class IPPacket(Packet):
  16         def __init__(self, init=[]):
  17                 Packet.__init__(self, init)
  18
  19         def decode(self):
  20                 header = []
  21                 for byte in self[:ipv4_header_len]:
  22                         header.append(self.pop(0))
  23                 for k, v in sorted(ipv4_header.items()):
  24                         setattr(self, k, get_field_data(v, header))
  25
  26         def encode(self):
  27                 header = 0
  28                 for k, v in sorted(ipv4_header.items()):
  29                         value = merge_bytes(split_bytes(getattr(self, k), math.ceil(v.width/8)), "little")
  30                         header += (value << v.offset+(v.byte*8))
  31                 for d in split_bytes(header, ipv4_header_len):
  32                         self.insert(0, d)
  33
  34         def __repr__(self):
  35                 r = "--------\n"
  36                 for k in sorted(ipv4_header.keys()):
  37                         r += k + " : 0x%x" %getattr(self,k) + "\n"
  38                 r += "payload: "
  39                 for d in self:
  40                         r += "%02x" %d
  41                 return r
  42
  43 class IP(Module):
  44         def  __init__(self, mac, mac_address, ip_address, debug=False, loopback=False):
  45                 self.mac = mac
  46                 self.mac_address = mac_address
  47                 self.ip_address = ip_address
  48                 self.debug = debug
  49                 self.loopback = loopback
  50                 self.tx_packets = []
  51                 self.tx_packet = IPPacket()
  52                 self.rx_packet = IPPacket()
  53                 self.table = {}
  54                 self.request_pending = False
  55
  56                 self.mac.set_ip_callback(self.callback)
  57
  58         def send(self, packet):
  59                 packet.encode()
  60                 if self.debug:
  61                         print_ip(">>>>>>>>")
  62                         print_ip(packet)
  63                 mac_packet = mac.MACPacket(packet)
  64                 mac_packet.destination_mac_address = 0x12345678abcd # XXX
  65                 mac_packet.source_mac_address = self.mac_address
  66                 mac_packet.ethernet_type = ethernet_type_ip
  67                 self.mac.send(mac_packet)
  68
  69         def callback(self, packet):
  70                 packet = IPPacket(packet)
  71                 packet.decode()
  72                 if self.debug:
  73                         print_ip("<<<<<<<<")
  74                         print_ip(packet)
  75                 if self.loopback:
  76                         self.send(packet)
  77                 else:
  78                         self.process(packet)
  79
  80         def process(self, packet):
  81                 pass
  82
  83 if __name__ == "__main__":
  84         from liteeth.test.model.dumps import *
  85         from liteeth.test.model.mac import *
  86         errors = 0
  87         # ARP request
  88         packet = MACPacket(udp)
  89         packet.decode_remove_header()
  90         #print(packet)
  91         packet = IPPacket(packet)
  92         # check decoding
  93         packet.decode()
  94         print(packet)
  95         errors += verify_packet(packet, {})
  96         # check encoding
  97         packet.encode()
  98         packet.decode()
  99         #print(packet)
 100         errors += verify_packet(packet, {})
 101
 102         print("ip errors " + str(errors))