SoFunction
Updated on 2024-12-12

python's scapy library, example of sending/receiving packets to/from a network card.

Question:

When testing Transmit and receive streams are performed using instruments such as TestCenter and SmartBit. If they are still used for automated smoking, it will bring the problem of low efficiency and high cost.

Solution:

The use of a network card to send and receive streams has performance statistical drawbacks, but allows for verification of some basic functionality and is economical.

Using the scapy module.

1- Get the iface of the computer's NIC and pre-design which ifaces to use for sending and receiving streams;

2-conf.L2listen listens on each iface

to invoke the startup capture, or you can invoke the construct ping packet

4-sendp sends Layer 2 messages, send sends Layer 3 messages

5-sniff sniffs specified packets on iface, can have filter conditions

6 - Stop wireshark from capturing packets

7-close closes listening to iface.

Discussion:

No attempt was made to use sr1, srp for sending and receiving packets.

The whole process is relatively clear and the steps appear in pairs, making it easy to memorize.

When sniffing, it will drop the part of the message that appears before the iface, this problem may be caused by not performing a good listening and starting a packet capture.

There are no specific performance criteria stated for the NIC, so you may need to feel your way around, and if you find a NIC that is not suitable for testing, you need to switch to the meter immediately to test it.

#! usr/bin/env python
# -*- coding:utf-8 -*-
import os
import sys
import re
import struct
import string
from  import *
import subprocess

conf.use_pcap = True 

'''
cmd
python
from  import *
ls(Ether())
ls(IP())
ls(ICMP())
send(IP(dst='1.2.3.4')/ICMP())
sendp(Raw("zhongxing"), iface='eth15', loop=1, inter=0.2, verbose=False)
set up inter 参数来set up发送相邻两个包直接的时间间隔
set up timeout 参数来set up等待应答的超时时间
set up retry 参数来set up重试次数。
'''


print u"Implementing NIC Packet Sending."
target = []
for i in range(1,len()):
  m = [i].split('=')
  if m[0]=='-t':
    (m[1])
  if m[0]=='-ip':
    (m[1])
  if m[0]=='-mac':
    (m[1])
print 'test -- ',target
print

print u'Get the iface of the network card'
eth_local = []
a = repr().split('\n')
for x in a:
  b = []
  b = (' ')
  for y in b:
    if ('eth', y):
      eth_local.append(y)
print u'De-duplication'
c = []
(eth_local[0])
for i in range(0,len(eth_local),1):
  m = 0
  for j in range(0,len(c),1):
    if c[j] == eth_local[i]:
      m += 1
  if m==0:
    (eth_local[i])
print c #['eth15', 'eth21', 'eth17']

print u'Create Layer 2 message'
src_mac = '00:00:11:11:22:22'
dst_mac = '00:00:22:22:11:11'
dst_ip = '1.2.3.4'
src_ip = '5.6.7.8'
src_port = 1234
dst_port = 5678

##ls()
##ls(IP())
##IP().show()
##lsc()
pack_ip = IP(dst=dst_ip, src=src_ip, proto=1)
##ls(ICMP())
##ls(UDP())
pack_icmp = ICMP(type=8)
##ls(Ether())
pack_ether = Ether(dst=dst_mac, src=src_mac, type=0x0800)
info = Raw('zhongxing')
t = str(pack_ether/pack_ip/pack_icmp/info)
s = Ether(t)
print u'The message to be sent is:',
eth = c[1]
print u'The sent NIC iface is %s\n' % eth

print u'--------- start listening - send icmp - sniff icmp - close listening ----------'
print u'--------- started listening to -------------'
L2socket = conf.L2listen
listen_socket = L2socket(type=ETH_P_ALL, iface=eth)
print listen_socket
print conf.L2listen

#### startup grab bag
##cmd='C:\Program Files (x86)\Wireshark\'
##card_id = str(1)
##cap_file = str('H:\python\')
##args = [cmd,"-i "+card_id,"-w",cap_file]
##print "*DEBUG*",args
##p=(args)


print u'---------sendp()function call----------'
sendp(s,iface=eth, verbose=False)

##print u'---------srp() function call ----------'
The ##sr function is at the heart of Scapy, and returns two lists.
## The first list is the packets for which an answer was received and their corresponding answers.
## The second list is of packets for which no answer was received.
## Often, we need to call other functions to make the two return values more readable.
##help(srp)
##p = srp(s,iface=c[1], verbose=False)
##print ()

print u'--------- sniff, filter, save pcap, read pcap ----------'
##print sniff.__doc__
##pkts = sniff(iface = 'eth15',filter = 'icmp',count = 3, prn=lambda x: ())
ip = '172.10.0.1'
(["", ip]) # Offer to sniff
##Ether / IP / ICMP 172.10.1.124 > 172.10.0.1 echo-request 0 / Raw
##Ether / IP / ICMP 172.10.0.1 > 172.10.1.124 echo-reply 0 / Raw
##Ether / IP / ICMP 172.10.1.124 > 172.10.0.1 echo-request 0 / Raw
##listen_socket1 = L2socket(listen_socket)
##pkts = sniff(iface = eth,filter = 'icmp',count = 20, timeout = 10, L2socket=listen_socket)
pkts = sniff(iface = eth, filter = 'icmp', count = 20, timeout = 10) 
try:
  if 0 < len(pkts):
    print u'--------- sniffs the message ----------'
    ##pkts[0].show()
    wrpcap('',pkts)
    read_pkts = rdpcap('')
    print read_pkts[0]

    print u'---------------exportsbase64Data in coded format---------------'
    export_object(str(pkts[0]))

    print u'--------------- converted to base64 encoded format data ---------------'
    newPkt = import_object('eNprYAqN+Q8GGp/TOCfN5GBwZWDwc/nCwNAgOItrDRdjLxD/Z+gEQitpgwvijAIMjAxgoODmAYLO\
    /m7ebq6ubs7+ri6uAa5+YNrf2dHREaiEgbGQUQ8AnjEcMQ==')
    print newPkt
    s = Ether(newPkt)
    print u'The message to be sent is:',
    sendp(s,iface=eth, verbose=False)
  else:
    print u'--------- did not sniff the message ----------'
except:
  pass
finally:
  print u'--------- shut down listening to -------------'
  listen_socket.close()

Above this python's scapy library, the realization of the network card send and receive packets example is all I have shared with you, I hope to give you a reference, but also hope that you support me more.