#!/router/bin/python
from .stl_general_test import CStlGeneral_Test, CTRexScenario
from trex_stl_lib.api import *
import os, sys
import pprint
def ip2num (ip_str):
return struct.unpack('>L', socket.inet_pton(socket.AF_INET, ip_str))[0]
def num2ip (ip_num):
return socket.inet_ntoa(struct.pack('>L', ip_num))
def ip_add (ip_str, cnt):
return num2ip(ip2num(ip_str) + cnt)
class STLCapture_Test(CStlGeneral_Test):
"""Tests for capture packets"""
def setUp(self):
CStlGeneral_Test.setUp(self)
if not self.is_loopback:
self.skip('capture tests are skipped on a non-loopback machine')
assert 'bi' in CTRexScenario.stl_ports_map
self.c = CTRexScenario.stl_trex
self.tx_port, self.rx_port = CTRexScenario.stl_ports_map['bi'][0]
self.c.connect()
self.c.reset(ports = [self.tx_port, self.rx_port])
self.pkt = STLPktBuilder(pkt = Ether()/IP(src="16.0.0.1",dst="48.0.0.1")/UDP(dport=12,sport=1025)/IP()/'a_payload_example')
self.percentage = 5 if self.is_virt_nics else 50
@classmethod
def tearDownClass(cls):
if CTRexScenario.stl_init_error:
return
# connect back at end of tests
if not cls.is_connected():
CTRexScenario.stl_trex.connect()
# a simple capture test - inject packets and see the packets arrived the same
def test_basic_capture (self):
pkt_count = 100
try:
# move to service mode
self.c.set_service_mode(ports = self.rx_port)
# start a capture
rc = self.c.start_capture(rx_ports = [self.rx_port], limit = pkt_count)
# inject few packets with a VM
vm = STLScVmRaw( [STLVmFlowVar ( "ip_src", min_value="16.0.0.0", max_value="16.255.255.255", size=4, step = 7, op = "inc"),
STLVmWrFlowVar (fv_name="ip_src", pkt_offset= "IP.src"),
STLVmFixIpv4(offset = "IP")
]
);
pkt = STLPktBuilder(pkt = Ether()/IP(src="16.0.0.1",dst="48.0.0.1")/UDP(dport=12,sport=1025)/IP()/'a_payload_example',
vm = vm)
stream = STLStream(name = 'burst',
packet = pkt,
mode = STLTXSingleBurst(total_pkts = pkt_count,
percentage = self.percentage)
)
self.c.add_streams(ports = self.tx_port, streams = [stream])
self.c.start(ports = self.tx_port, force = True)
self.c.wait_on_traffic(ports = self.tx_port)
pkt_list = []
self.c.stop_capture(rc['id'], output = pkt_list)
assert (len(pkt_list) == pkt_count)
# generate all the values that should be
expected_src_ips = [ip_add('16.0.0.0', i * 7) for i in range(pkt_count)]
for i, pkt in enumerate(pkt_list):
pkt_scapy = Ether(pkt['binary'])
pkt_ts = pkt['ts']
assert('IP' in pkt_scapy)
assert(pkt_scapy['IP'].src in expected_src_ips)
# remove the match
del expected_src_ips[expected_src_ips.index(pkt_scapy['IP'].src)]
except STLError as e:
assert False , '{0}'.format(e)
finally:
self.c.remove_all_captures()
self.c.set_service_mode(ports = self.rx_port, enabled = False)
# in this test we apply captures under traffic multiple times
def test_stress_capture (self):
pkt_count = 100
try:
# move to service mode
self.c.set_service_mode(ports = self.rx_port)
# start heavy traffic
pkt = STLPktBuilder(pkt = Ether()/IP(src="16.0.0.1",dst="48.0.0.1")/UDP(dport=12,sport=1025)/IP()/'a_payload_example')
stream = STLStream(name = 'burst',
packet = pkt,
mode = STLTXCont(percentage = self.percentage)
)
self.c.add_streams(ports = self.tx_port, streams = [stream])
self.c.start(ports = self.tx_port, force = True)
captures = [{'capture_id': None, 'limit': 50}, {'capture_id': None, 'limit': 80}, {'capture_id': None, 'limit': 100}]
for i in range(0, 100):
# start a few captures
for capture in captures:
capture['capture_id'] = self.c.start_capture(rx_ports = [self.rx_port], limit = capture['limit'])['id']
# a little time to wait for captures to be full
wait_iterations = 0
while True:
server_captures = self.c.get_capture_status()
counts = ([c['count'] for c in server_captures.values()])
if {50, 80, 100} == set(counts):
break
time.sleep(0.1)
wait_iterations += 1
assert(wait_iterations <= 5)
for capture in captures:
capture_id = capture['capture_id']
# make sure the server registers us and we are full
assert(capture['capture_id'] in server_captures.keys())
assert(server_captures[capture_id]['count'] == capture['limit'])
# fetch packets
pkt_list = []
self.c.stop_capture(capture['capture_id'], pkt_list)
assert (len(pkt_list) == capture['limit'])
# a little sanity per packet
for pkt in pkt_list:
scapy_pkt = Ether(pkt['binary'])
assert(scapy_pkt['IP'].src == '16.0.0.1')
assert(scapy_pkt['IP'].dst == '48.0.0.1')
except STLError as e:
assert False , '{0}'.format(e)
finally:
self.c.remove_all_captures()
self.c.set_service_mode(ports = self.rx_port, enabled = False)
# in this test we capture and analyze the ARP request / response
def test_arp_capture (self):
if self.c.get_port_attr(self.tx_port)['layer_mode'] != 'IPv4':
return self.skip('skipping ARP capture test for non-ipv4 config on port {0}'.format(self.tx_port))
if self.c.get_port_attr(self.rx_port)['layer_mode'] != 'IPv4':
return self.skip('skipping ARP capture test for non-ipv4 config on port {0}'.format(self.rx_port))
try:
# move to service mode
self.c.set_service_mode(ports = [self.tx_port, self.rx_port])
# start a capture
capture_info = self.c.start_capture(rx_ports = [self.tx_port, self.rx_port], limit = 2)
# generate an ARP request
self.c.arp(ports = self.tx_port)
pkts = []
self.c.stop_capture(capture_info['id'], output = pkts)
assert len(pkts) == 2
# find the correct order
if pkts[0]['port'] == self.rx_port:
request = pkts[0]
response = pkts[1]
else:
request = pkts[1]
response = pkts[0]
assert request['port'] == self.rx_port
assert response['port'] == self.tx_port
arp_request, arp_response = Ether(request['binary']), Ether(response['binary'])
assert 'ARP' in arp_request
assert 'ARP' in arp_response
assert arp_request['ARP'].op == 1
assert arp_response['ARP'].op == 2
assert arp_request['ARP'].pdst == arp_response['ARP'].psrc
except STLError as e:
assert False , '{0}'.format(e)
finally:
self.c.remove_all_captures()
self.c.set_service_mode(ports = [self.tx_port, self.rx_port], enabled = False)
# test PING
def test_ping_capture (self):
if self.c.get_port_attr(self.tx_port)['layer_mode'] != 'IPv4':
return self.skip('skipping ARP capture test for non-ipv4 config on port {0}'.format(self.tx_port))
if self.c.get_port_attr(self.rx_port)['layer_mode'] != 'IPv4':
return self.skip('skipping ARP capture test for non-ipv4 config on port {0}'.format(self.rx_port))
try:
# move to service mode
self.c.set_service_mode(ports = [self.tx_port, self.rx_port])
# start a capture
capture_info = self.c.start_capture(rx_ports = [self.tx_port, self.rx_port], limit = 100)
# generate an ARP request
tx_ipv4 = self.c.get_port_attr(port = self.tx_port)['src_ipv4']
rx_ipv4 = self.c.get_port_attr(port = self.rx_port)['src_ipv4']
count = 50
self.c.ping_ip(src_port = self.tx_port, dst_ip = rx_ipv4, pkt_size = 1500, count = count, interval_sec = 0.01)
pkts = []
self.c.stop_capture(capture_info['id'], output = pkts)
req_pkts = [Ether(pkt['binary']) for pkt in pkts if pkt['port'] == self.rx_port]
res_pkts = [Ether(pkt['binary']) for pkt in pkts if pkt['port'] == self.tx_port]
assert len(req_pkts) == count
assert len(res_pkts) == count
for req_pkt in req_pkts:
assert 'ICMP' in req_pkt
assert req_pkt['IP'].src == tx_ipv4
assert req_pkt['IP'].dst == rx_ipv4
assert req_pkt['ICMP'].type == 8
assert len(req_pkt) == 1500
for res_pkt in res_pkts:
assert 'ICMP' in res_pkt
assert res_pkt['IP'].src == rx_ipv4
assert res_pkt['IP'].dst == tx_ipv4
assert res_pkt['ICMP'].type == 0
assert len(res_pkt) == 1500
except STLError as e:
assert False , '{0}'.format(e)
finally:
self.c.remove_all_captures()
self.c.set_service_mode(ports = [self.tx_port, self.rx_port], enabled = False)