Python scapy.all 模块，Packet() 实例源码

def __repr__(self):
        if self.reserved != 0:
            return scapy_all.Packet.__repr__(self)

        # Known status
        if self.status in self.ERROR_CODES and self.additional_size == 0:
            return "<CIP_ResponseStatus  status={}>".format(self.ERROR_CODES[self.status])

        # Simple status
        if self.additional_size == 0:
            return "<CIP_ResponseStatus  status=%#x>" % self.status

        # Forward Open failure
        if self.status == 1 and self.additional == b"\x00\x01":
            return "<CIP_ResponseStatus  status=Connection failure>"
        return scapy_all.Packet.__repr__(self)

def get_write_state_pcap(udp_dest_port, out_dir, packet_size=256):

    class MemTest(Packet):
        name = "P4Bench Message for MemTest"
        fields_desc =  [
            XBitField("op", 0x1, 4),
            XBitField("index", 0x1, 12),
            XBitField("data", 0xf1f2f3f4, 32),
        ]

    pkt = add_eth_ip_udp_headers(udp_dest_port)

    pkt /= MemTest(op=2, index=0, data=0)

    pkt = add_padding(pkt, packet_size)
    wrpcap('%s/test.pcap' % out_dir, pkt)

def add_layers(nb_fields, nb_headers):
    class P4Bench(Packet):
        name = "P4Bench Message"
        fields_desc =  []
        for i in range(nb_fields):
            fields_desc.append(ShortField('field_%d' %i , 0))
    layers = ''
    for i in range(nb_headers):
        if i < (nb_headers - 1):
            layers = layers / P4Bench(field_0=1)
        else:
            layers = layers / P4Bench(field_0=0)
    return layers

def vary_header_field(nb_fields):
    class P4Bench(Packet):
        name = "P4Bench Message"
        fields_desc =  []
        for i in range(nb_fields):
            fields_desc.append(ShortField('field_%d' % i , i))
    return P4Bench()

def add_padding(pkt, packet_size):
    pad_len = packet_size - len(pkt)
    if pad_len < 0:
        print "Packet size [%d] is greater than expected %d" % (len(pkt), packet_size)
    else:
        pad = '\x00' * pad_len
        pkt = pkt/pad
    return pkt

def get_read_state_pcap(udp_dest_port, out_dir, packet_size=256):

    class MemTest(Packet):
        name = "P4Bench Message for MemTest"
        fields_desc =  [
            XBitField("op", 0x1, 4),
            XBitField("index", 0x1, 12),
            XBitField("data", 0xf1f2f3f4, 32),
        ]

    pkt = add_eth_ip_udp_headers(udp_dest_port)
    pkt /= MemTest(op=1, index=0)

    pkt = add_padding(pkt, packet_size)
    wrpcap('%s/test.pcap' % out_dir, pkt)