ligang/vpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

VPP-1508: Use scapy.compat to manage packet level library differences.

Browse Source 
Change-Id: Icdf6abc9e53d33b26fd1d531c7dda6be0bb9cb55
Signed-off-by: Paul Vinciguerra <pvinci@vinciconsulting.com>
This commit is contained in:
Paul Vinciguerra
2019-03-10 10:04:23 -07:00
committed by Ole Trøan
parent 0f6602cb24
commit a7427ec6f8
27 changed files with 261 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/framework.py
+14 -6
View File
@@ -18,6 +18,8 @@ from threading import Thread, Event
from inspect import getdoc, isclass
from traceback import format_exception
from logging import FileHandler, DEBUG, Formatter
import scapy.compat
from scapy.packet import Raw
from hook import StepHook, PollHook, VppDiedError
from vpp_pg_interface import VppPGInterface
@@ -511,7 +513,11 @@ class VppTestCase(unittest.TestCase):
if hasattr(cls, 'vpp'):
if hasattr(cls, 'vapi'):
cls.logger.debug("Disconnecting class vapi client on %s",
cls.__name__)
cls.vapi.disconnect()
cls.logger.debug("Deleting class vapi attribute on %s",
cls.__name__)
del cls.vapi
cls.vpp.poll()
if cls.vpp.returncode is None:
@@ -519,6 +525,8 @@ class VppTestCase(unittest.TestCase):
cls.vpp.kill()
cls.logger.debug("Waiting for vpp to die")
cls.vpp.communicate()
cls.logger.debug("Deleting class vpp attribute on %s",
cls.__name__)
del cls.vpp
if cls.vpp_startup_failed:
@@ -560,7 +568,6 @@ class VppTestCase(unittest.TestCase):
def tearDown(self):
""" Show various debug prints after each test """
super(VppTestCase, self).tearDown()
self.logger.debug("--- tearDown() for %s.%s(%s) called ---" %
(self.__class__.__name__, self._testMethodName,
self._testMethodDoc))
@@ -848,14 +855,14 @@ class VppTestCase(unittest.TestCase):
def assert_packet_checksums_valid(self, packet,
ignore_zero_udp_checksums=True):
received = packet.__class__(str(packet))
received = packet.__class__(scapy.compat.raw(packet))
self.logger.debug(
ppp("Verifying packet checksums for packet:", received))
udp_layers = ['UDP', 'UDPerror']
checksum_fields = ['cksum', 'chksum']
checksums = []
counter = 0
temp = received.__class__(str(received))
temp = received.__class__(scapy.compat.raw(received))
while True:
layer = temp.getlayer(counter)
if layer:
@@ -872,7 +879,7 @@ class VppTestCase(unittest.TestCase):
counter = counter + 1
if 0 == len(checksums):
return
temp = temp.__class__(str(temp))
temp = temp.__class__(scapy.compat.raw(temp))
for layer, cf in checksums:
calc_sum = getattr(temp[layer], cf)
self.assert_equal(
@@ -889,9 +896,10 @@ class VppTestCase(unittest.TestCase):
received_packet_checksum = getattr(received_packet[layer], field_name)
if ignore_zero_checksum and 0 == received_packet_checksum:
return
recalculated = received_packet.__class__(str(received_packet))
recalculated = received_packet.__class__(
scapy.compat.raw(received_packet))
delattr(recalculated[layer], field_name)
recalculated = recalculated.__class__(str(recalculated))
recalculated = recalculated.__class__(scapy.compat.raw(recalculated))
self.assert_equal(received_packet_checksum,
getattr(recalculated[layer], field_name),
"packet checksum on layer: %s" % layer)
test/hook.py
+4 -1
View File
@@ -5,6 +5,9 @@ import traceback
from log import RED, single_line_delim, double_line_delim
import ipaddress
from subprocess import check_output, CalledProcessError
import scapy.compat
from util import check_core_path, get_core_path
@@ -31,7 +34,7 @@ class Hook(object):
return val
if len(val) == 6:
return '{!s} ({!s})'.format(val, ':'.join(['{:02x}'.format(
ord(x)) for x in val]))
scapy.compat.orb(x)) for x in val]))
try:
# we don't call test_type(val) because it is a packed value.
return '{!s} ({!s})'.format(val, str(
test/test_acl_plugin_l2l3.py
+6 -3
View File
@@ -28,6 +28,7 @@ from socket import inet_pton, AF_INET, AF_INET6
from random import choice, shuffle
from pprint import pprint
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP, ICMP, TCP
@@ -308,12 +309,14 @@ class TestACLpluginL2L3(VppTestCase):
# Scapy IPv6 stuff is too smart for its own good.
# So we do this and coerce the ICMP into unknown type
if packet.haslayer(UDP):
data = str(packet[UDP][Raw])
data = scapy.compat.raw(packet[UDP][Raw])
else:
if l3 == IP:
data = str(ICMP(str(packet[l3].payload))[Raw])
data = scapy.compat.raw(ICMP(
scapy.compat.raw(packet[l3].payload))[Raw])
else:
data = str(ICMPv6Unknown(str(packet[l3].payload)).msgbody)
data = scapy.compat.raw(ICMPv6Unknown(
scapy.compat.raw(packet[l3].payload)).msgbody)
udp_or_icmp = packet[l3].payload
payload_info = self.payload_to_info(data)
packet_index = payload_info.index
test/test_acl_plugin_macip.py
+29 -19
View File
@@ -3,12 +3,14 @@ from __future__ import print_function
"""ACL plugin - MACIP tests
"""
import binascii
import ipaddress
import random
from socket import inet_ntop, inet_pton, AF_INET, AF_INET6
from struct import pack, unpack
import re
import unittest
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
@@ -194,8 +196,10 @@ class MethodHolder(VppTestCase):
return acls
def create_rules(self, mac_type=EXACT_MAC, ip_type=EXACT_IP,
acl_count=1, rules_count=[1]):
acl_count=1, rules_count=None):
acls = []
if rules_count is None:
rules_count = [1]
src_mac = int("220000dead00", 16)
for acl in range(2, (acl_count+1) * 2):
rules = []
@@ -248,7 +252,7 @@ class MethodHolder(VppTestCase):
ip4[3] = 0
ip6[8] = random.randint(100, 200)
ip6[15] = 0
ip_pack = ''
ip_pack = b''
for j in range(0, len(ip)):
ip_pack += pack('<B', int(ip[j]))
@@ -256,8 +260,9 @@ class MethodHolder(VppTestCase):
'is_ipv6': is_ip6,
'src_ip_addr': ip_pack,
'src_ip_prefix_len': ip_len,
'src_mac': mac.replace(':', '').decode('hex'),
'src_mac_mask': mask.replace(':', '').decode('hex')})
'src_mac': binascii.unhexlify(mac.replace(':', '')),
'src_mac_mask': binascii.unhexlify(
mask.replace(':', ''))})
rules.append(rule)
if ip_type == self.WILD_IP:
break
@@ -275,7 +280,7 @@ class MethodHolder(VppTestCase):
def verify_macip_acls(self, acl_count, rules_count, expected_count=2):
reply = self.macip_acl_dump_debug()
for acl in range(2, (acl_count+1) * 2):
self.assertEqual(reply[acl - 2].count, rules_count[acl/2-1])
self.assertEqual(reply[acl - 2].count, rules_count[acl//2-1])
self.vapi.macip_acl_interface_get()
@@ -410,30 +415,32 @@ class MethodHolder(VppTestCase):
sub_ip[15] = random.randint(200, 255)
elif ip_type == self.SUBNET_IP:
if denyIP:
sub_ip[2] = str(int(sub_ip[2]) + 1)
sub_ip[2] = int(sub_ip[2]) + 1
sub_ip[14] = random.randint(100, 199)
sub_ip[15] = random.randint(200, 255)
src_ip6 = inet_ntop(AF_INET6, str(bytearray(sub_ip)))
packed_src_ip6 = b''.join(
[scapy.compat.chb(x) for x in sub_ip])
src_ip6 = inet_ntop(AF_INET6, packed_src_ip6)
packet /= IPv6(src=src_ip6, dst=dst_ip6)
else:
if ip_type != self.EXACT_IP:
sub_ip = ip_rule.split('.')
if ip_type == self.WILD_IP:
sub_ip[0] = str(random.randint(1, 49))
sub_ip[1] = str(random.randint(50, 99))
sub_ip[2] = str(random.randint(100, 199))
sub_ip[3] = str(random.randint(200, 255))
sub_ip[0] = random.randint(1, 49)
sub_ip[1] = random.randint(50, 99)
sub_ip[2] = random.randint(100, 199)
sub_ip[3] = random.randint(200, 255)
elif ip_type == self.SUBNET_IP:
if denyIP:
sub_ip[1] = str(int(sub_ip[1])+1)
sub_ip[2] = str(random.randint(100, 199))
sub_ip[3] = str(random.randint(200, 255))
src_ip4 = ".".join(sub_ip)
sub_ip[1] = int(sub_ip[1])+1
sub_ip[2] = random.randint(100, 199)
sub_ip[3] = random.randint(200, 255)
src_ip4 = '.'.join(['{!s}'.format(x) for x in sub_ip])
packet /= IP(src=src_ip4, dst=dst_ip4, frag=0, flags=0)
packet /= UDP(sport=src_port, dport=dst_port)/Raw(payload)
packet[Raw].load += " mac:"+src_mac
packet[Raw].load += b" mac:%s" % scapy.compat.raw(src_mac)
size = self.pg_if_packet_sizes[p % len(self.pg_if_packet_sizes)]
if isinstance(src_if, VppSubInterface):
@@ -485,7 +492,9 @@ class MethodHolder(VppTestCase):
sub_ip = list(unpack('<16B', inet_pton(AF_INET6, ip)))
for i in range(8, 16):
sub_ip[i] = 0
ip = inet_ntop(AF_INET6, str(bytearray(sub_ip)))
packed_ip = b''.join(
[scapy.compat.chb(x) for x in sub_ip])
ip = inet_ntop(AF_INET6, packed_ip)
else:
if ip_type == self.WILD_IP:
ip = "0.0.0.0"
@@ -540,8 +549,9 @@ class MethodHolder(VppTestCase):
'is_ipv6': is_ip6,
'src_ip_addr': ip_rule,
'src_ip_prefix_len': prefix_len,
'src_mac': mac_rule.replace(':', '').decode('hex'),
'src_mac_mask': mac_mask.replace(':', '').decode('hex')})
'src_mac': binascii.unhexlify(mac_rule.replace(':', '')),
'src_mac_mask': binascii.unhexlify(
mac_mask.replace(':', ''))})
macip_rules.append(macip_rule)
# deny all other packets
test/test_bfd.py
+16 -13
View File
@@ -12,6 +12,7 @@ from socket import AF_INET, AF_INET6, inet_ntop
from struct import pack, unpack
from six import moves
import scapy.compat
from scapy.layers.inet import UDP, IP
from scapy.layers.inet6 import IPv6
from scapy.layers.l2 import Ether
@@ -42,7 +43,8 @@ class AuthKeyFactory(object):
while conf_key_id in self._conf_key_ids:
conf_key_id = randint(0, 0xFFFFFFFF)
self._conf_key_ids[conf_key_id] = 1
key = str(bytearray([randint(0, 255) for _ in range(randint(1, 20))]))
key = scapy.compat.raw(
bytearray([randint(0, 255) for _ in range(randint(1, 20))]))
return VppBFDAuthKey(test=test, auth_type=auth_type,
conf_key_id=conf_key_id, key=key)
@@ -435,7 +437,8 @@ class BFDTestSession(object):
self.test.logger.debug("BFD: Creating packet")
self.fill_packet_fields(packet)
if self.sha1_key:
hash_material = str(packet[BFD])[:32] + self.sha1_key.key + \
hash_material = scapy.compat.raw(
packet[BFD])[:32] + self.sha1_key.key + \
"\0" * (20 - len(self.sha1_key.key))
self.test.logger.debug("BFD: Calculated SHA1 hash: %s" %
hashlib.sha1(hash_material).hexdigest())
@@ -492,7 +495,7 @@ class BFDTestSession(object):
# last 20 bytes represent the hash - so replace them with the key,
# pad the result with zeros and hash the result
hash_material = bfd.original[:-20] + self.sha1_key.key + \
"\0" * (20 - len(self.sha1_key.key))
b"\0" * (20 - len(self.sha1_key.key))
expected_hash = hashlib.sha1(hash_material).hexdigest()
self.test.assert_equal(binascii.hexlify(bfd.auth_key_hash),
expected_hash, "Auth key hash")
@@ -923,7 +926,7 @@ class BFD4TestCase(VppTestCase):
# halve required min rx
old_required_min_rx = self.vpp_session.required_min_rx
self.vpp_session.modify_parameters(
required_min_rx=0.5 * self.vpp_session.required_min_rx)
required_min_rx=self.vpp_session.required_min_rx // 2)
# now we wait 0.8*3*old-req-min-rx and the session should still be up
self.sleep(0.8 * self.vpp_session.detect_mult *
old_required_min_rx / USEC_IN_SEC,
@@ -1118,8 +1121,8 @@ class BFD4TestCase(VppTestCase):
udp_sport_rx += 1
# need to compare the hex payload here, otherwise BFD_vpp_echo
# gets in way
self.assertEqual(str(p[UDP].payload),
str(echo_packet[UDP].payload),
self.assertEqual(scapy.compat.raw(p[UDP].payload),
scapy.compat.raw(echo_packet[UDP].payload),
"Received packet is not the echo packet sent")
self.assert_equal(udp_sport_tx, udp_sport_rx, "UDP source port (== "
"ECHO packet identifier for test purposes)")
@@ -1612,8 +1615,8 @@ class BFD6TestCase(VppTestCase):
udp_sport_rx += 1
# need to compare the hex payload here, otherwise BFD_vpp_echo
# gets in way
self.assertEqual(str(p[UDP].payload),
str(echo_packet[UDP].payload),
self.assertEqual(scapy.compat.raw(p[UDP].payload),
scapy.compat.raw(echo_packet[UDP].payload),
"Received packet is not the echo packet sent")
self.assert_equal(udp_sport_tx, udp_sport_rx, "UDP source port (== "
"ECHO packet identifier for test purposes)")
@@ -2293,7 +2296,7 @@ class BFDCLITestCase(VppTestCase):
def cli_verify_no_response(self, cli):
""" execute a CLI, asserting that the response is empty """
self.assert_equal(self.vapi.cli(cli),
"",
b"",
"CLI command response")
def cli_verify_response(self, cli, expected):
@@ -2325,7 +2328,7 @@ class BFDCLITestCase(VppTestCase):
self.cli_verify_no_response(
"bfd key set conf-key-id %s type keyed-sha1 secret %s" %
(k.conf_key_id,
"".join("{:02x}".format(ord(c)) for c in k.key)))
"".join("{:02x}".format(scapy.compat.orb(c)) for c in k.key)))
self.assertTrue(k.query_vpp_config())
self.vpp_session = VppBFDUDPSession(
self, self.pg0, self.pg0.remote_ip4, sha1_key=k)
@@ -2342,7 +2345,7 @@ class BFDCLITestCase(VppTestCase):
self.cli_verify_response(
"bfd key set conf-key-id %s type keyed-sha1 secret %s" %
(k.conf_key_id,
"".join("{:02x}".format(ord(c)) for c in k2.key)),
"".join("{:02x}".format(scapy.compat.orb(c)) for c in k2.key)),
"bfd key set: `bfd_auth_set_key' API call failed, "
"rv=-103:BFD object in use")
# manipulating the session using old secret should still work
@@ -2361,7 +2364,7 @@ class BFDCLITestCase(VppTestCase):
self.cli_verify_no_response(
"bfd key set conf-key-id %s type meticulous-keyed-sha1 secret %s" %
(k.conf_key_id,
"".join("{:02x}".format(ord(c)) for c in k.key)))
"".join("{:02x}".format(scapy.compat.orb(c)) for c in k.key)))
self.assertTrue(k.query_vpp_config())
self.vpp_session = VppBFDUDPSession(self, self.pg0,
self.pg0.remote_ip6, af=AF_INET6,
@@ -2380,7 +2383,7 @@ class BFDCLITestCase(VppTestCase):
self.cli_verify_response(
"bfd key set conf-key-id %s type keyed-sha1 secret %s" %
(k.conf_key_id,
"".join("{:02x}".format(ord(c)) for c in k2.key)),
"".join("{:02x}".format(scapy.compat.orb(c)) for c in k2.key)),
"bfd key set: `bfd_auth_set_key' API call failed, "
"rv=-103:BFD object in use")
# manipulating the session using old secret should still work
test/test_bier.py
+24 -22
View File
@@ -12,6 +12,7 @@ from vpp_bier import BIER_HDR_PAYLOAD, VppBierImp, VppBierDispEntry, \
VppBierDispTable, VppBierTable, VppBierTableID, VppBierRoute
from vpp_udp_encap import VppUdpEncap
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
@@ -122,10 +123,11 @@ class TestBier(VppTestCase):
for pkt_size in pkt_sizes:
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
MPLS(label=77, ttl=255) /
BIER(length=hdr_len_id, BitString=chr(255)*n_bytes) /
BIER(length=hdr_len_id,
BitString=scapy.compat.chb(255)*n_bytes) /
IPv6(src=self.pg0.remote_ip6, dst=self.pg0.remote_ip6) /
UDP(sport=1234, dport=1234) /
Raw(chr(5) * pkt_size))
Raw(scapy.compat.chb(5) * pkt_size))
pkts = p
self.pg0.add_stream(pkts)
@@ -159,18 +161,18 @@ class TestBier(VppTestCase):
self.assertEqual(bier_hdr.Proto, 5)
# The bit-string should consist only of the BP given by i.
byte_array = ['\0'] * (n_bytes)
byte_val = chr(1 << (bp - 1) % 8)
byte_pos = n_bytes - (((bp - 1) / 8) + 1)
byte_array = [b'\0'] * (n_bytes)
byte_val = scapy.compat.chb(1 << (bp - 1) % 8)
byte_pos = n_bytes - (((bp - 1) // 8) + 1)
byte_array[byte_pos] = byte_val
bitstring = ''.join(byte_array)
bitstring = ''.join([scapy.compat.chb(x) for x in byte_array])
self.assertEqual(len(bitstring), len(bier_hdr.BitString))
self.assertEqual(bitstring, bier_hdr.BitString)
#
# cleanup. not strictly necessary, but it's much quicker this way
# becuase the bier_fib_dump and ip_fib_dump will be empty when the
# because the bier_fib_dump and ip_fib_dump will be empty when the
# auto-cleanup kicks in
#
for br in bier_routes:
@@ -222,7 +224,7 @@ class TestBier(VppTestCase):
MPLS(label=77, ttl=255) /
BIER(length=BIERLength.BIER_LEN_64,
entropy=ii,
BitString=chr(255)*16) /
BitString=scapy.compat.chb(255)*16) /
IPv6(src=self.pg0.remote_ip6,
dst=self.pg0.remote_ip6) /
UDP(sport=1234, dport=1234) /
@@ -334,7 +336,7 @@ class TestBier(VppTestCase):
#
# An imposition object with both bit-positions set
#
bi = VppBierImp(self, bti, 333, chr(0x3) * 32)
bi = VppBierImp(self, bti, 333, scapy.compat.chb(0x3) * 32)
bi.add_vpp_config()
#
@@ -446,7 +448,7 @@ class TestBier(VppTestCase):
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
MPLS(label=77, ttl=255) /
BIER(length=BIERLength.BIER_LEN_256,
BitString=chr(255)*32,
BitString=scapy.compat.chb(255)*32,
BFRID=99) /
IP(src="1.1.1.1", dst="232.1.1.1") /
UDP(sport=1234, dport=1234) /
@@ -460,7 +462,7 @@ class TestBier(VppTestCase):
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
MPLS(label=77, ttl=255) /
BIER(length=BIERLength.BIER_LEN_256,
BitString=chr(255)*32,
BitString=scapy.compat.chb(255)*32,
BFRID=77) /
IP(src="1.1.1.1", dst="232.1.1.1") /
UDP(sport=1234, dport=1234) /
@@ -486,7 +488,7 @@ class TestBier(VppTestCase):
# A multicast route to forward post BIER disposition that needs
# a check against sending back into the BIER core
#
bi = VppBierImp(self, bti, 333, chr(0x3) * 32)
bi = VppBierImp(self, bti, 333, scapy.compat.chb(0x3) * 32)
bi.add_vpp_config()
route_eg_232_1_1_2 = VppIpMRoute(
@@ -506,7 +508,7 @@ class TestBier(VppTestCase):
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
MPLS(label=77, ttl=255) /
BIER(length=BIERLength.BIER_LEN_256,
BitString=chr(255)*32,
BitString=scapy.compat.chb(255)*32,
BFRID=77) /
IP(src="1.1.1.1", dst="232.1.1.2") /
UDP(sport=1234, dport=1234) /
@@ -523,10 +525,10 @@ class TestBier(VppTestCase):
bt = VppBierTable(self, bti, 77)
bt.add_vpp_config()
lowest = ['\0'] * (n_bytes)
lowest[-1] = chr(1)
highest = ['\0'] * (n_bytes)
highest[0] = chr(128)
lowest = [b'\0'] * (n_bytes)
lowest[-1] = scapy.compat.chb(1)
highest = [b'\0'] * (n_bytes)
highest[0] = scapy.compat.chb(128)
#
# Impostion Sets bit strings
@@ -636,7 +638,7 @@ class TestBier(VppTestCase):
src=self.pg0.remote_mac) /
IP(src="1.1.1.1", dst="232.1.1.1") /
UDP(sport=1234, dport=1234) /
Raw(chr(5) * 32))
Raw(scapy.compat.chb(5) * 32))
rx = self.send_and_expect(self.pg0, p*65, self.pg1)
@@ -647,7 +649,7 @@ class TestBier(VppTestCase):
src=self.pg0.remote_mac) /
IP(src="1.1.1.1", dst="232.1.1.2") /
UDP(sport=1234, dport=1234) /
Raw(chr(5) * 512))
Raw(scapy.compat.chb(5) * 512))
rx = self.send_and_expect(self.pg0, p*65, self.pg1)
self.assertEqual(rx[0][IP].src, "1.1.1.1")
@@ -716,9 +718,9 @@ class TestBier(VppTestCase):
# only use the second, but creating 2 tests with a non-zero
# value index in the route add
#
bi = VppBierImp(self, bti, 333, chr(0xff) * 32)
bi = VppBierImp(self, bti, 333, scapy.compat.chb(0xff) * 32)
bi.add_vpp_config()
bi2 = VppBierImp(self, bti, 334, chr(0xff) * 32)
bi2 = VppBierImp(self, bti, 334, scapy.compat.chb(0xff) * 32)
bi2.add_vpp_config()
#
@@ -821,7 +823,7 @@ class TestBier(VppTestCase):
UDP(sport=333, dport=8138) /
BIFT(sd=1, set=0, bsl=2, ttl=255) /
BIER(length=BIERLength.BIER_LEN_256,
BitString=chr(255)*32,
BitString=scapy.compat.chb(255)*32,
BFRID=99) /
IP(src="1.1.1.1", dst="232.1.1.1") /
UDP(sport=1234, dport=1234) /
test/test_cdp.py
+4
View File
@@ -71,6 +71,10 @@ class TestCDP(VppTestCase):
super(TestCDP, cls).tearDownClass()
raise
@classmethod
def tearDownClass(cls):
super(TestCDP, cls).tearDownClass()
def test_enable_cdp(self):
self.logger.info(self.vapi.cli("cdp enable"))
ret = self.vapi.cli("show cdp")
test/test_dhcp.py
+3 -2
View File
@@ -8,6 +8,7 @@ from framework import VppTestCase, VppTestRunner, running_extended_tests
from vpp_neighbor import VppNeighbor
from vpp_ip_route import find_route, VppIpTable
from util import mk_ll_addr
import scapy.compat
from scapy.layers.l2 import Ether, getmacbyip, ARP
from scapy.layers.inet import IP, UDP, ICMP
from scapy.layers.inet6 import IPv6, in6_getnsmac
@@ -490,7 +491,7 @@ class TestDHCP(VppTestCase):
#
# 1. not our IP address = not checked by VPP? so offer is replayed
# to client
bad_ip = option_82[0:8] + chr(33) + option_82[9:]
bad_ip = option_82[0:8] + scapy.compat.chb(33) + option_82[9:]
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
@@ -504,7 +505,7 @@ class TestDHCP(VppTestCase):
"DHCP offer option 82 bad address")
# 2. Not a sw_if_index VPP knows
bad_if_index = option_82[0:2] + chr(33) + option_82[3:]
bad_if_index = option_82[0:2] + scapy.compat.chb(33) + option_82[3:]
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
test/test_gbp.py
+1 -1
View File
@@ -3207,7 +3207,7 @@ class TestGBP(VppTestCase):
vx_tun_l3.add_vpp_config()
#
# packets destined to unkown addresses in the BVI's subnet
# packets destined to unknown addresses in the BVI's subnet
# are ARP'd for
#
p4 = (Ether(src=self.pg0.remote_mac, dst=str(self.router_mac)) /
test/test_gre.py
+8 -5
View File
@@ -2,6 +2,7 @@
import unittest
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether, Dot1Q, GRE
from scapy.layers.inet import IP, UDP
@@ -146,7 +147,8 @@ class TestGRE(VppTestCase):
GRE() /
Ether(dst=RandMAC('*:*:*:*:*:*'),
src=RandMAC('*:*:*:*:*:*')) /
IP(src=str(RandIP()), dst=str(RandIP())) /
IP(src=scapy.compat.raw(RandIP()),
dst=scapy.compat.raw(RandIP())) /
UDP(sport=1234, dport=1234) /
Raw(payload))
info.data = p.copy()
@@ -165,7 +167,8 @@ class TestGRE(VppTestCase):
Ether(dst=RandMAC('*:*:*:*:*:*'),
src=RandMAC('*:*:*:*:*:*')) /
Dot1Q(vlan=vlan) /
IP(src=str(RandIP()), dst=str(RandIP())) /
IP(src=scapy.compat.raw(RandIP()),
dst=scapy.compat.raw(RandIP())) /
UDP(sport=1234, dport=1234) /
Raw(payload))
info.data = p.copy()
@@ -217,7 +220,7 @@ class TestGRE(VppTestCase):
self.assertEqual(rx_ip.src, tunnel_src)
self.assertEqual(rx_ip.dst, tunnel_dst)
rx_gre = GRE(str(rx_ip[IPv6].payload))
rx_gre = GRE(scapy.compat.raw(rx_ip[IPv6].payload))
rx_ip = rx_gre[IPv6]
self.assertEqual(rx_ip.src, tx_ip.src)
@@ -243,7 +246,7 @@ class TestGRE(VppTestCase):
self.assertEqual(rx_ip.src, tunnel_src)
self.assertEqual(rx_ip.dst, tunnel_dst)
rx_gre = GRE(str(rx_ip[IPv6].payload))
rx_gre = GRE(scapy.compat.raw(rx_ip[IPv6].payload))
tx_ip = tx[IP]
rx_ip = rx_gre[IP]
@@ -270,7 +273,7 @@ class TestGRE(VppTestCase):
self.assertEqual(rx_ip.src, tunnel_src)
self.assertEqual(rx_ip.dst, tunnel_dst)
rx_gre = GRE(str(rx_ip[IP].payload))
rx_gre = GRE(scapy.compat.raw(rx_ip[IP].payload))
rx_ip = rx_gre[IPv6]
tx_ip = tx[IPv6]
test/test_ip4.py
+3 -2
View File
@@ -4,6 +4,7 @@ import random
import socket
import unittest
import scapy.compat
from scapy.contrib.mpls import MPLS
from scapy.layers.inet import IP, UDP, TCP, ICMP, icmptypes, icmpcodes
from scapy.layers.l2 import Ether, Dot1Q, ARP
@@ -310,7 +311,7 @@ class TestIPv4FibCrud(VppTestCase):
dest_addr_len = 32
n_next_hop_addr = socket.inet_pton(socket.AF_INET, next_hop_addr)
for _ in range(count):
n_dest_addr = '{:08x}'.format(dest_addr).decode('hex')
n_dest_addr = binascii.unhexlify('{:08x}'.format(dest_addr))
self.vapi.ip_add_del_route(n_dest_addr, dest_addr_len,
n_next_hop_addr)
added_ips.append(socket.inet_ntoa(n_dest_addr))
@@ -325,7 +326,7 @@ class TestIPv4FibCrud(VppTestCase):
dest_addr_len = 32
n_next_hop_addr = socket.inet_pton(socket.AF_INET, next_hop_addr)
for _ in range(count):
n_dest_addr = '{:08x}'.format(dest_addr).decode('hex')
n_dest_addr = binascii.unhexlify('{:08x}'.format(dest_addr))
self.vapi.ip_add_del_route(n_dest_addr, dest_addr_len,
n_next_hop_addr, is_add=0)
removed_ips.append(socket.inet_ntoa(n_dest_addr))
test/test_ip4_vrf_multi_instance.py
+3 -1
View File
@@ -66,6 +66,7 @@ import unittest
import random
import socket
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP, ARP
@@ -344,7 +345,8 @@ class TestIp4VrfMultiInst(VppTestCase):
if found:
break
for host in pg_if.remote_hosts:
if str(addr) == str(host.ip4):
if scapy.compat.raw(addr) == \
scapy.compat.raw(host.ip4):
vrf_count += 1
found = True
break
test/test_ip6.py
+1
View File
@@ -4,6 +4,7 @@ import socket
import unittest
from parameterized import parameterized
import scapy.compat
import scapy.layers.inet6 as inet6
from scapy.contrib.mpls import MPLS
from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_RS, \
test/test_ipsec_nat.py
+4 -2
View File
@@ -2,9 +2,11 @@
import socket
import scapy.compat
from scapy.layers.l2 import Ether
from scapy.layers.inet import ICMP, IP, TCP, UDP
from scapy.layers.ipsec import SecurityAssociation, ESP
from util import ppp, ppc
from template_ipsec import TemplateIpsec
from vpp_ipsec import VppIpsecSA, VppIpsecSpd, VppIpsecSpdEntry,\
@@ -126,9 +128,9 @@ class IPSecNATTestCase(TemplateIpsec):
def verify_capture_encrypted(self, capture, sa):
for packet in capture:
try:
copy = packet.__class__(str(packet))
copy = packet.__class__(scapy.compat.raw(packet))
del copy[UDP].len
copy = packet.__class__(str(copy))
copy = packet.__class__(scapy.compat.raw(copy))
self.assert_equal(packet[UDP].len, copy[UDP].len,
"UDP header length")
self.assert_packet_checksums_valid(packet)
test/test_lb.py
+6 -4
View File
@@ -1,5 +1,6 @@
import socket
import scapy.compat
from scapy.layers.inet import IP, UDP
from scapy.layers.inet6 import IPv6
from scapy.layers.l2 import Ether, GRE
@@ -93,11 +94,12 @@ class TestLB(VppTestCase):
self.assertEqual(gre.proto, 0x0800 if isv4 else 0x86DD)
self.assertEqual(gre.flags, 0)
self.assertEqual(gre.version, 0)
inner = IPver(str(gre.payload))
inner = IPver(scapy.compat.raw(gre.payload))
payload_info = self.payload_to_info(inner[Raw])
self.info = self.packet_infos[payload_info.index]
self.assertEqual(payload_info.src, self.pg0.sw_if_index)
self.assertEqual(str(inner), str(self.info.data[IPver]))
self.assertEqual(scapy.compat.raw(inner),
scapy.compat.raw(self.info.data[IPver]))
def checkCapture(self, encap, isv4):
self.pg0.assert_nothing_captured()
@@ -135,7 +137,7 @@ class TestLB(VppTestCase):
)
self.assertEqual(ip.nh, 47)
# self.assertEqual(len(ip.options), 0)
gre = GRE(str(p[IPv6].payload))
gre = GRE(scapy.compat.raw(p[IPv6].payload))
self.checkInner(gre, isv4)
elif (encap == 'l3dsr'):
ip = p[IP]
@@ -174,7 +176,7 @@ class TestLB(VppTestCase):
)
self.assertEqual(ip.nh, 17)
self.assertGreaterEqual(ip.hlim, 63)
udp = UDP(str(p[IPv6].payload))
udp = UDP(scapy.compat.raw(p[IPv6].payload))
self.assertEqual(udp.dport, 3307)
load[asid] += 1
except:
test/test_map.py
+5 -3
View File
@@ -5,6 +5,8 @@ import unittest
from framework import VppTestCase, VppTestRunner
from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath
import scapy.compat
from scapy.layers.l2 import Ether, Raw
from scapy.layers.inet import IP, UDP, ICMP, TCP, fragment
from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded
@@ -187,7 +189,7 @@ class TestMAP(VppTestCase):
self.vapi.ppcli("map params pre-resolve del ip6-nh 4001::1")
def validate(self, rx, expected):
self.assertEqual(rx, expected.__class__(str(expected)))
self.assertEqual(rx, expected.__class__(scapy.compat.raw(expected)))
def payload(self, len):
return 'x' * len
@@ -369,7 +371,7 @@ class TestMAP(VppTestCase):
p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
dst="2001:db8:1f0::c0a8:1:f") / payload)
p6_translated.hlim -= 1
p6_translated['TCP'].options = [('MSS', 1300)]
p6_translated[TCP].options = [('MSS', 1300)]
rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
for p in rx:
self.validate(p[1], p6_translated)
@@ -383,7 +385,7 @@ class TestMAP(VppTestCase):
dst=self.pg0.remote_ip4) / payload)
p4_translated.id = 0
p4_translated.ttl -= 1
p4_translated['TCP'].options = [('MSS', 1300)]
p4_translated[TCP].options = [('MSS', 1300)]
rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
for p in rx:
self.validate(p[1], p4_translated)
test/test_mpls.py
+2 -1
View File
@@ -11,6 +11,7 @@ from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsRoute, \
VppMplsLabel, MplsLspMode, find_mpls_route
from vpp_mpls_tunnel_interface import VppMPLSTunnelInterface
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP, ICMP
@@ -1873,7 +1874,7 @@ class TestMPLSL2(VppTestCase):
verify_mpls_stack(self, rx, mpls_labels)
tx_eth = tx[Ether]
rx_eth = Ether(str(rx[MPLS].payload))
rx_eth = Ether(scapy.compat.raw(rx[MPLS].payload))
self.assertEqual(rx_eth.src, tx_eth.src)
self.assertEqual(rx_eth.dst, tx_eth.dst)
test/test_nat.py
+44 -41
View File
@@ -6,6 +6,8 @@ import struct
import random
from framework import VppTestCase, VppTestRunner, running_extended_tests
import scapy.compat
from scapy.layers.inet import IP, TCP, UDP, ICMP
from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply, \
@@ -320,7 +322,8 @@ class MethodHolder(VppTestCase):
pref_n[13] = ip4_n[1]
pref_n[14] = ip4_n[2]
pref_n[15] = ip4_n[3]
return socket.inet_ntop(socket.AF_INET6, ''.join(pref_n))
packed_pref_n = b''.join([scapy.compat.chb(x) for x in pref_n])
return socket.inet_ntop(socket.AF_INET6, packed_pref_n)
def extract_ip4(self, ip6, plen):
"""
@@ -689,8 +692,8 @@ class MethodHolder(VppTestCase):
p = (IP(src=src_if.remote_ip4, dst=dst) /
TCP(sport=sport, dport=dport) /
Raw(data))
p = p.__class__(str(p))
chksum = p['TCP'].chksum
p = p.__class__(scapy.compat.raw(p))
chksum = p[TCP].chksum
proto_header = TCP(sport=sport, dport=dport, chksum=chksum)
elif proto == IP_PROTOS.udp:
proto_header = UDP(sport=sport, dport=dport)
@@ -869,8 +872,8 @@ class MethodHolder(VppTestCase):
self.assertEqual(6, len(data))
for record in data:
# natEvent
self.assertIn(ord(record[230]), [4, 5])
if ord(record[230]) == 4:
self.assertIn(scapy.compat.orb(record[230]), [4, 5])
if scapy.compat.orb(record[230]) == 4:
nat44_ses_create_num += 1
else:
nat44_ses_delete_num += 1
@@ -882,16 +885,16 @@ class MethodHolder(VppTestCase):
# ingressVRFID
self.assertEqual(struct.pack("!I", 0), record[234])
# protocolIdentifier/sourceTransportPort/postNAPTSourceTransportPort
if IP_PROTOS.icmp == ord(record[4]):
if IP_PROTOS.icmp == scapy.compat.orb(record[4]):
self.assertEqual(struct.pack("!H", self.icmp_id_in), record[7])
self.assertEqual(struct.pack("!H", self.icmp_id_out),
record[227])
elif IP_PROTOS.tcp == ord(record[4]):
elif IP_PROTOS.tcp == scapy.compat.orb(record[4]):
self.assertEqual(struct.pack("!H", self.tcp_port_in),
record[7])
self.assertEqual(struct.pack("!H", self.tcp_port_out),
record[227])
elif IP_PROTOS.udp == ord(record[4]):
elif IP_PROTOS.udp == scapy.compat.orb(record[4]):
self.assertEqual(struct.pack("!H", self.udp_port_in),
record[7])
self.assertEqual(struct.pack("!H", self.udp_port_out),
@@ -910,7 +913,7 @@ class MethodHolder(VppTestCase):
self.assertEqual(1, len(data))
record = data[0]
# natEvent
self.assertEqual(ord(record[230]), 3)
self.assertEqual(scapy.compat.orb(record[230]), 3)
# natPoolID
self.assertEqual(struct.pack("!I", 0), record[283])
@@ -924,7 +927,7 @@ class MethodHolder(VppTestCase):
self.assertEqual(1, len(data))
record = data[0]
# natEvent
self.assertEqual(ord(record[230]), 13)
self.assertEqual(scapy.compat.orb(record[230]), 13)
# natQuotaExceededEvent
self.assertEqual(struct.pack("I", 1), record[466])
# maxSessionEntries
@@ -940,7 +943,7 @@ class MethodHolder(VppTestCase):
self.assertEqual(1, len(data))
record = data[0]
# natEvent
self.assertEqual(ord(record[230]), 13)
self.assertEqual(scapy.compat.orb(record[230]), 13)
# natQuotaExceededEvent
self.assertEqual(struct.pack("I", 2), record[466])
# maxBIBEntries
@@ -957,7 +960,7 @@ class MethodHolder(VppTestCase):
self.assertEqual(1, len(data))
record = data[0]
# natEvent
self.assertEqual(ord(record[230]), 13)
self.assertEqual(scapy.compat.orb(record[230]), 13)
# natQuotaExceededEvent
self.assertEqual(struct.pack("I", 5), record[466])
# maxFragmentsPendingReassembly
@@ -976,7 +979,7 @@ class MethodHolder(VppTestCase):
self.assertEqual(1, len(data))
record = data[0]
# natEvent
self.assertEqual(ord(record[230]), 13)
self.assertEqual(scapy.compat.orb(record[230]), 13)
# natQuotaExceededEvent
self.assertEqual(struct.pack("I", 5), record[466])
# maxFragmentsPendingReassembly
@@ -996,15 +999,15 @@ class MethodHolder(VppTestCase):
record = data[0]
# natEvent
if is_create:
self.assertEqual(ord(record[230]), 10)
self.assertEqual(scapy.compat.orb(record[230]), 10)
else:
self.assertEqual(ord(record[230]), 11)
self.assertEqual(scapy.compat.orb(record[230]), 11)
# sourceIPv6Address
self.assertEqual(src_addr, record[27])
# postNATSourceIPv4Address
self.assertEqual(self.nat_addr_n, record[225])
# protocolIdentifier
self.assertEqual(IP_PROTOS.tcp, ord(record[4]))
self.assertEqual(IP_PROTOS.tcp, scapy.compat.orb(record[4]))
# ingressVRFID
self.assertEqual(struct.pack("!I", 0), record[234])
# sourceTransportPort
@@ -1027,9 +1030,9 @@ class MethodHolder(VppTestCase):
record = data[0]
# natEvent
if is_create:
self.assertEqual(ord(record[230]), 6)
self.assertEqual(scapy.compat.orb(record[230]), 6)
else:
self.assertEqual(ord(record[230]), 7)
self.assertEqual(scapy.compat.orb(record[230]), 7)
# sourceIPv6Address
self.assertEqual(src_addr, record[27])
# destinationIPv6Address
@@ -1044,7 +1047,7 @@ class MethodHolder(VppTestCase):
self.assertEqual(socket.inet_pton(socket.AF_INET, dst_addr),
record[226])
# protocolIdentifier
self.assertEqual(IP_PROTOS.tcp, ord(record[4]))
self.assertEqual(IP_PROTOS.tcp, scapy.compat.orb(record[4]))
# ingressVRFID
self.assertEqual(struct.pack("!I", 0), record[234])
# sourceTransportPort
@@ -1076,7 +1079,7 @@ class MethodHolder(VppTestCase):
self.assertEqual(1, len(data))
record = data[0]
# natEvent
self.assertEqual(ord(record[230]), 13)
self.assertEqual(scapy.compat.orb(record[230]), 13)
# natQuotaExceededEvent
self.assertEqual(struct.pack("I", 3), record[466])
# maxEntriesPerUser
@@ -1167,9 +1170,9 @@ class MethodHolder(VppTestCase):
layer = self.proto2layer(proto)
if proto == IP_PROTOS.tcp:
data = "A" * 4 + "B" * 16 + "C" * 3
data = b"A" * 4 + b"B" * 16 + b"C" * 3
else:
data = "A" * 16 + "B" * 16 + "C" * 3
data = b"A" * 16 + b"B" * 16 + b"C" * 3
self.port_in = random.randint(1025, 65535)
reass = self.vapi.nat_reass_dump()
@@ -1248,9 +1251,9 @@ class MethodHolder(VppTestCase):
layer = self.proto2layer(proto)
if proto == IP_PROTOS.tcp:
data = "A" * 4 + "B" * 16 + "C" * 3
data = b"A" * 4 + b"B" * 16 + b"C" * 3
else:
data = "A" * 16 + "B" * 16 + "C" * 3
data = b"A" * 16 + b"B" * 16 + b"C" * 3
self.port_in = random.randint(1025, 65535)
for i in range(2):
@@ -1317,9 +1320,9 @@ class MethodHolder(VppTestCase):
layer = self.proto2layer(proto)
if proto == IP_PROTOS.tcp:
data = "A" * 4 + "B" * 16 + "C" * 3
data = b"A" * 4 + b"B" * 16 + b"C" * 3
else:
data = "A" * 16 + "B" * 16 + "C" * 3
data = b"A" * 16 + b"B" * 16 + b"C" * 3
# send packet from host to server
pkts = self.create_stream_frag(self.pg0,
@@ -1346,9 +1349,9 @@ class MethodHolder(VppTestCase):
layer = self.proto2layer(proto)
if proto == IP_PROTOS.tcp:
data = "A" * 4 + "B" * 16 + "C" * 3
data = b"A" * 4 + b"B" * 16 + b"C" * 3
else:
data = "A" * 16 + "B" * 16 + "C" * 3
data = b"A" * 16 + b"B" * 16 + b"C" * 3
self.port_in = random.randint(1025, 65535)
for i in range(2):
@@ -1422,9 +1425,9 @@ class MethodHolder(VppTestCase):
layer = self.proto2layer(proto)
if proto == IP_PROTOS.tcp:
data = "A" * 4 + "B" * 16 + "C" * 3
data = b"A" * 4 + b"B" * 16 + b"C" * 3
else:
data = "A" * 16 + "B" * 16 + "C" * 3
data = b"A" * 16 + b"B" * 16 + b"C" * 3
self.port_in = random.randint(1025, 65535)
for i in range(2):
@@ -3539,7 +3542,7 @@ class TestNAT44(MethodHolder):
is_inside=0)
self.vapi.nat44_forwarding_enable_disable(1)
data = "A" * 16 + "B" * 16 + "C" * 3
data = b"A" * 16 + b"B" * 16 + b"C" * 3
pkts = self.create_stream_frag(self.pg1,
self.pg0.remote_ip4,
4789,
@@ -3667,7 +3670,7 @@ class TestNAT44(MethodHolder):
self.vapi.nat_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
src_port=self.ipfix_src_port)
data = "A" * 4 + "B" * 16 + "C" * 3
data = b"A" * 4 + b"B" * 16 + b"C" * 3
self.tcp_port_in = random.randint(1025, 65535)
pkts = self.create_stream_frag(self.pg0,
self.pg1.remote_ip4,
@@ -8043,7 +8046,7 @@ class TestNAT64(MethodHolder):
reass_n_start = len(reass)
# in2out
data = 'a' * 200
data = b'a' * 200
pkts = self.create_stream_frag_ip6(self.pg0, self.pg1.remote_ip4,
self.tcp_port_in, 20, data)
self.pg0.add_stream(pkts)
@@ -8059,7 +8062,7 @@ class TestNAT64(MethodHolder):
self.assertEqual(data, p[Raw].load)
# out2in
data = "A" * 4 + "b" * 16 + "C" * 3
data = b"A" * 4 + b"b" * 16 + b"C" * 3
pkts = self.create_stream_frag(self.pg1,
self.nat_addr,
20,
@@ -8127,7 +8130,7 @@ class TestNAT64(MethodHolder):
self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
# in2out
data = 'a' * 200
data = b'a' * 200
pkts = self.create_stream_frag_ip6(self.pg0, self.pg1.remote_ip4,
self.tcp_port_in, 20, data)
pkts.reverse()
@@ -8144,7 +8147,7 @@ class TestNAT64(MethodHolder):
self.assertEqual(data, p[Raw].load)
# out2in
data = "A" * 4 + "B" * 16 + "C" * 3
data = b"A" * 4 + b"B" * 16 + b"C" * 3
pkts = self.create_stream_frag(self.pg1,
self.nat_addr,
20,
@@ -8283,7 +8286,7 @@ class TestNAT64(MethodHolder):
self.vapi.nat_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
src_port=self.ipfix_src_port)
data = 'a' * 200
data = b'a' * 200
pkts = self.create_stream_frag_ip6(self.pg0, self.pg1.remote_ip4,
self.tcp_port_in, 20, data)
pkts.reverse()
@@ -8358,9 +8361,9 @@ class TestNAT64(MethodHolder):
for p in capture:
if p.haslayer(Data):
data = ipfix.decode_data_set(p.getlayer(Set))
if ord(data[0][230]) == 10:
if scapy.compat.orb(data[0][230]) == 10:
self.verify_ipfix_bib(data, 1, self.pg0.remote_ip6n)
elif ord(data[0][230]) == 6:
elif scapy.compat.orb(data[0][230]) == 6:
self.verify_ipfix_nat64_ses(data,
1,
self.pg0.remote_ip6n,
@@ -8387,9 +8390,9 @@ class TestNAT64(MethodHolder):
self.ipfix_domain_id)
if p.haslayer(Data):
data = ipfix.decode_data_set(p.getlayer(Set))
if ord(data[0][230]) == 11:
if scapy.compat.orb(data[0][230]) == 11:
self.verify_ipfix_bib(data, 0, self.pg0.remote_ip6n)
elif ord(data[0][230]) == 7:
elif scapy.compat.orb(data[0][230]) == 7:
self.verify_ipfix_nat64_ses(data,
0,
self.pg0.remote_ip6n,
test/test_neighbor.py
+4 -2
View File
@@ -9,6 +9,7 @@ from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, \
VppIpTable, DpoProto
from vpp_papi import VppEnum
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether, ARP, Dot1Q
from scapy.layers.inet import IP, UDP
@@ -1178,8 +1179,9 @@ class ARPTestCase(VppTestCase):
#
# change the interface's MAC
#
mac = [chr(0x00), chr(0x00), chr(0x00),
chr(0x33), chr(0x33), chr(0x33)]
mac = [scapy.compat.chb(0x00), scapy.compat.chb(0x00),
scapy.compat.chb(0x00), scapy.compat.chb(0x33),
scapy.compat.chb(0x33), scapy.compat.chb(0x33)]
mac_string = ''.join(mac)
self.vapi.sw_interface_set_mac_address(self.pg1.sw_if_index,
test/test_punt.py
+3 -2
View File
@@ -11,6 +11,7 @@ import struct
from struct import unpack, unpack_from
from util import ppp, ppc
from re import compile
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP, ICMP
@@ -20,12 +21,12 @@ from framework import VppTestCase, VppTestRunner
# Format MAC Address
def get_mac_addr(bytes_addr):
return ':'.join('%02x' % ord(b) for b in bytes_addr)
return ':'.join('%02x' % scapy.compat.orb(b) for b in bytes_addr)
# Format IP Address
def ipv4(bytes_addr):
return '.'.join('%d' % ord(b) for b in bytes_addr)
return '.'.join('%d' % scapy.compat.orb(b) for b in bytes_addr)
# Unpack Ethernet Frame
test/test_qos.py
+31 -30
View File
@@ -9,6 +9,7 @@ from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsRoute, \
VppMplsLabel, VppMplsTable
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether, Dot1Q
from scapy.layers.inet import IP, UDP
@@ -50,10 +51,10 @@ class TestQOS(VppTestCase):
# for table 1 map the n=0xff possible values of input QoS mark,
# n to 1-n
#
output = [chr(0)] * 256
output = [scapy.compat.chb(0)] * 256
for i in range(0, 255):
output[i] = chr(255 - i)
os = ''.join(output)
output[i] = scapy.compat.chb(255 - i)
os = b''.join(output)
rows = [{'outputs': os},
{'outputs': os},
{'outputs': os},
@@ -64,8 +65,8 @@ class TestQOS(VppTestCase):
#
# For table 2 (and up) use the value n for everything
#
output = [chr(2)] * 256
os = ''.join(output)
output = [scapy.compat.chb(2)] * 256
os = b''.join(output)
rows = [{'outputs': os},
{'outputs': os},
{'outputs': os},
@@ -73,8 +74,8 @@ class TestQOS(VppTestCase):
self.vapi.qos_egress_map_update(2, rows)
output = [chr(3)] * 256
os = ''.join(output)
output = [scapy.compat.chb(3)] * 256
os = b''.join(output)
rows = [{'outputs': os},
{'outputs': os},
{'outputs': os},
@@ -82,8 +83,8 @@ class TestQOS(VppTestCase):
self.vapi.qos_egress_map_update(3, rows)
output = [chr(4)] * 256
os = ''.join(output)
output = [scapy.compat.chb(4)] * 256
os = b''.join(output)
rows = [{'outputs': os},
{'outputs': os},
{'outputs': os},
@@ -121,12 +122,12 @@ class TestQOS(VppTestCase):
p_v4 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, tos=1) /
UDP(sport=1234, dport=1234) /
Raw(chr(100) * 65))
Raw(scapy.compat.chb(100) * 65))
p_v6 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6,
tc=1) /
UDP(sport=1234, dport=1234) /
Raw(chr(100) * 65))
Raw(scapy.compat.chb(100) * 65))
#
# Since we have not yet enabled the recording of the input QoS
@@ -282,14 +283,14 @@ class TestQOS(VppTestCase):
from_ip = 6
from_mpls = 5
from_vlan = 4
output = [chr(from_ext)] * 256
os1 = ''.join(output)
output = [chr(from_vlan)] * 256
os2 = ''.join(output)
output = [chr(from_mpls)] * 256
os3 = ''.join(output)
output = [chr(from_ip)] * 256
os4 = ''.join(output)
output = [scapy.compat.chb(from_ext)] * 256
os1 = b''.join(output)
output = [scapy.compat.chb(from_vlan)] * 256
os2 = b''.join(output)
output = [scapy.compat.chb(from_mpls)] * 256
os3 = b''.join(output)
output = [scapy.compat.chb(from_ip)] * 256
os4 = b''.join(output)
rows = [{'outputs': os1},
{'outputs': os2},
{'outputs': os3},
@@ -333,11 +334,11 @@ class TestQOS(VppTestCase):
p_1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst="10.0.0.1", tos=1) /
UDP(sport=1234, dport=1234) /
Raw(chr(100) * 65))
Raw(scapy.compat.chb(100) * 65))
p_3 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst="10.0.0.3", tos=1) /
UDP(sport=1234, dport=1234) /
Raw(chr(100) * 65))
Raw(scapy.compat.chb(100) * 65))
rx = self.send_and_expect(self.pg0, p_1 * 65, self.pg1)
@@ -388,7 +389,7 @@ class TestQOS(VppTestCase):
MPLS(label=32, cos=3, ttl=2) /
IP(src=self.pg0.remote_ip4, dst="10.0.0.1", tos=1) /
UDP(sport=1234, dport=1234) /
Raw(chr(100) * 65))
Raw(scapy.compat.chb(100) * 65))
rx = self.send_and_expect(self.pg0, p_m1 * 65, self.pg1)
for p in rx:
@@ -414,7 +415,7 @@ class TestQOS(VppTestCase):
MPLS(label=33, ttl=2, cos=3) /
IP(src=self.pg0.remote_ip4, dst="10.0.0.4", tos=1) /
UDP(sport=1234, dport=1234) /
Raw(chr(100) * 65))
Raw(scapy.compat.chb(100) * 65))
rx = self.send_and_expect(self.pg0, p_m2 * 65, self.pg1)
@@ -446,10 +447,10 @@ class TestQOS(VppTestCase):
#
# QoS for all input values
#
output = [chr(0)] * 256
output = [scapy.compat.chb(0)] * 256
for i in range(0, 255):
output[i] = chr(255 - i)
os = ''.join(output)
output[i] = scapy.compat.chb(255 - i)
os = b''.join(output)
rows = [{'outputs': os},
{'outputs': os},
{'outputs': os},
@@ -515,12 +516,12 @@ class TestQOS(VppTestCase):
Dot1Q(vlan=11, prio=1) /
IP(src="1.1.1.1", dst="10.0.0.2", tos=1) /
UDP(sport=1234, dport=1234) /
Raw(chr(100) * 65))
Raw(scapy.compat.chb(100) * 65))
p_v2 = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
IP(src="1.1.1.1", dst="10.0.0.1", tos=1) /
UDP(sport=1234, dport=1234) /
Raw(chr(100) * 65))
Raw(scapy.compat.chb(100) * 65))
rx = self.send_and_expect(self.pg1, p_v2 * 65, self.pg0)
@@ -536,12 +537,12 @@ class TestQOS(VppTestCase):
Dot1Q(vlan=11, prio=2) /
IPv6(src="2001::1", dst="2001::2", tc=1) /
UDP(sport=1234, dport=1234) /
Raw(chr(100) * 65))
Raw(scapy.compat.chb(100) * 65))
p_v2 = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
IPv6(src="3001::1", dst="2001::1", tc=1) /
UDP(sport=1234, dport=1234) /
Raw(chr(100) * 65))
Raw(scapy.compat.chb(100) * 65))
rx = self.send_and_expect(self.pg1, p_v2 * 65, self.pg0)
test/test_reassembly.py
+8 -4
View File
@@ -5,6 +5,7 @@ import six
import unittest
from parameterized import parameterized
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether, GRE
from scapy.layers.inet import IP, UDP, ICMP
@@ -258,7 +259,8 @@ class TestIPv4Reassembly(TestIPReassemblyMixin, VppTestCase):
cls.pkt_infos = []
for index, info in six.iteritems(infos):
p = info.data
# cls.logger.debug(ppp("Packet:", p.__class__(str(p))))
# cls.logger.debug(ppp("Packet:",
# p.__class__(scapy.compat.raw(p))))
fragments_400 = fragment_rfc791(p, 400)
fragments_300 = fragment_rfc791(p, 300)
fragments_200 = [
@@ -589,7 +591,8 @@ class TestIPv6Reassembly(TestIPReassemblyMixin, VppTestCase):
cls.pkt_infos = []
for index, info in six.iteritems(infos):
p = info.data
# cls.logger.debug(ppp("Packet:", p.__class__(str(p))))
# cls.logger.debug(ppp("Packet:",
# p.__class__(scapy.compat.raw(p))))
fragments_400 = fragment_rfc8200(p, info.index, 400)
fragments_300 = fragment_rfc8200(p, info.index, 300)
cls.pkt_infos.append((index, fragments_400, fragments_300))
@@ -772,7 +775,7 @@ class TestIPv6Reassembly(TestIPReassemblyMixin, VppTestCase):
Raw())
self.extend_packet(p, 1000, self.padding)
fragments = fragment_rfc8200(p, 1, 500)
bad_fragment = p.__class__(str(fragments[1]))
bad_fragment = p.__class__(scapy.compat.raw(fragments[1]))
bad_fragment[IPv6ExtHdrFragment].nh = 59
bad_fragment[IPv6ExtHdrFragment].offset = 0
self.pg_enable_capture()
@@ -880,7 +883,8 @@ class TestIPv4ReassemblyLocalNode(VppTestCase):
cls.pkt_infos = []
for index, info in six.iteritems(infos):
p = info.data
# cls.logger.debug(ppp("Packet:", p.__class__(str(p))))
# cls.logger.debug(ppp("Packet:",
# p.__class__(scapy.compat.raw(p))))
fragments_300 = fragment_rfc791(p, 300)
cls.pkt_infos.append((index, fragments_300))
cls.fragments_300 = [x for (_, frags) in cls.pkt_infos for x in frags]
test/test_srv6.py
+8 -7
View File
@@ -9,6 +9,7 @@ from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto, VppIpTable
from vpp_srv6 import SRv6LocalSIDBehaviors, VppSRv6LocalSID, VppSRv6Policy, \
SRv6PolicyType, VppSRv6Steering, SRv6PolicySteeringTypes
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether, Dot1Q
from scapy.layers.inet6 import IPv6, UDP, IPv6ExtHdrSegmentRouting
@@ -1437,7 +1438,7 @@ class TestSRv6(VppTestCase):
tx_ip.chksum = None
# read back the pkt (with str()) to force computing these fields
# probably other ways to accomplish this are possible
tx_ip = IP(str(tx_ip))
tx_ip = IP(scapy.compat.raw(tx_ip))
self.assertEqual(rx_srh.payload, tx_ip)
@@ -1489,7 +1490,7 @@ class TestSRv6(VppTestCase):
self.assertEqual(rx_srh.nh, 59)
# the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
self.assertEqual(Ether(str(rx_srh.payload)), tx_ether)
self.assertEqual(Ether(scapy.compat.raw(rx_srh.payload)), tx_ether)
self.logger.debug("packet verification: SUCCESS")
@@ -1765,7 +1766,7 @@ class TestSRv6(VppTestCase):
tx_ip2.chksum = None
# read back the pkt (with str()) to force computing these fields
# probably other ways to accomplish this are possible
tx_ip2 = IP(str(tx_ip2))
tx_ip2 = IP(scapy.compat.raw(tx_ip2))
self.assertEqual(rx_ip, tx_ip2)
@@ -1791,7 +1792,7 @@ class TestSRv6(VppTestCase):
tx_ip = tx_pkt.getlayer(IPv6)
# we can't just get the 2nd Ether layer
# get the Raw content and dissect it as Ether
tx_eth1 = Ether(str(tx_pkt[Raw]))
tx_eth1 = Ether(scapy.compat.raw(tx_pkt[Raw]))
# verify if rx'ed packet has no SRH
self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
@@ -1837,7 +1838,7 @@ class TestSRv6(VppTestCase):
# read back the dumped packet (with str())
# to force computing these fields
# probably other ways are possible
p = Ether(str(p))
p = Ether(scapy.compat.raw(p))
payload_info.data = p.copy()
self.logger.debug(ppp("Created packet:", p))
pkts.append(p)
@@ -2087,7 +2088,7 @@ class TestSRv6(VppTestCase):
# take packet[Raw], convert it to an Ether layer
# and then extract Raw from it
payload_info = self.payload_to_info(
str(Ether(str(packet[Raw]))[Raw]))
Ether(scapy.compat.r(packet[Raw]))[Raw])
return payload_info
@@ -2101,7 +2102,7 @@ class TestSRv6(VppTestCase):
:param compare_func: function to compare in and out packet
"""
self.logger.info("Verifying capture on interface %s using function %s"
% (dst_if.name, compare_func.func_name))
% (dst_if.name, compare_func.__name__))
last_info = dict()
for i in self.pg_interfaces:
test/test_srv6_ad.py
+8 -7
View File
@@ -10,6 +10,7 @@ from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable
from vpp_srv6 import SRv6LocalSIDBehaviors, VppSRv6LocalSID, VppSRv6Policy, \
SRv6PolicyType, VppSRv6Steering, SRv6PolicySteeringTypes
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether, Dot1Q
from scapy.layers.inet6 import IPv6, UDP, IPv6ExtHdrSegmentRouting
@@ -363,7 +364,7 @@ class TestSRv6(VppTestCase):
tx_ip2.chksum = None
# read back the pkt (with str()) to force computing these fields
# probably other ways to accomplish this are possible
tx_ip2 = IP(str(tx_ip2))
tx_ip2 = IP(scapy.compat.raw(tx_ip2))
self.assertEqual(rx_ip, tx_ip2)
@@ -405,7 +406,7 @@ class TestSRv6(VppTestCase):
tx_ip.chksum = None
# -> read back the pkt (with str()) to force computing these fields
# probably other ways to accomplish this are possible
self.assertEqual(rx_srh.payload, IP(str(tx_ip)))
self.assertEqual(rx_srh.payload, IP(scapy.compat.raw(tx_ip)))
self.logger.debug("packet verification: SUCCESS")
@@ -495,7 +496,7 @@ class TestSRv6(VppTestCase):
tx_ip = tx_pkt.getlayer(IPv6)
# we can't just get the 2nd Ether layer
# get the Raw content and dissect it as Ether
tx_eth1 = Ether(str(tx_pkt[Raw]))
tx_eth1 = Ether(scapy.compat.raw(tx_pkt[Raw]))
# verify if rx'ed packet has no SRH
self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
@@ -535,7 +536,7 @@ class TestSRv6(VppTestCase):
# the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
tx_ether = tx_pkt.getlayer(Ether)
self.assertEqual(Ether(str(rx_srh.payload)), tx_ether)
self.assertEqual(Ether(scapy.compat.raw(rx_srh.payload)), tx_ether)
self.logger.debug("packet verification: SUCCESS")
@@ -573,7 +574,7 @@ class TestSRv6(VppTestCase):
# read back the dumped packet (with str())
# to force computing these fields
# probably other ways are possible
p = Ether(str(p))
p = Ether(scapy.compat.raw(p))
payload_info.data = p.copy()
self.logger.debug(ppp("Created packet:", p))
pkts.append(p)
@@ -736,7 +737,7 @@ class TestSRv6(VppTestCase):
# take packet[Raw], convert it to an Ether layer
# and then extract Raw from it
payload_info = self.payload_to_info(
Ether(str(packet[Raw]))[Raw])
Ether(scapy.compat.raw(packet[Raw]))[Raw])
return payload_info
@@ -750,7 +751,7 @@ class TestSRv6(VppTestCase):
:param compare_func: function to compare in and out packet
"""
self.logger.info("Verifying capture on interface %s using function %s"
% (dst_if.name, compare_func.func_name))
% (dst_if.name, compare_func.__name__))
last_info = dict()
for i in self.pg_interfaces:
test/test_srv6_as.py
+8 -7
View File
@@ -9,6 +9,7 @@ from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto, VppIpTable
from vpp_srv6 import SRv6LocalSIDBehaviors, VppSRv6LocalSID, VppSRv6Policy, \
SRv6PolicyType, VppSRv6Steering, SRv6PolicySteeringTypes
import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether, Dot1Q
from scapy.layers.inet6 import IPv6, UDP, IPv6ExtHdrSegmentRouting
@@ -488,7 +489,7 @@ class TestSRv6(VppTestCase):
tx_ip.chksum = None
# read back the pkt (with str()) to force computing these fields
# probably other ways to accomplish this are possible
tx_ip = IP(str(tx_ip))
tx_ip = IP(scapy.compat.raw(tx_ip))
self.assertEqual(payload, tx_ip)
@@ -537,7 +538,7 @@ class TestSRv6(VppTestCase):
payload = rx_ip.payload
# the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
self.assertEqual(Ether(str(payload)), tx_ether)
self.assertEqual(Ether(scapy.compat.raw(payload)), tx_ether)
self.logger.debug("packet verification: SUCCESS")
@@ -590,7 +591,7 @@ class TestSRv6(VppTestCase):
tx_ip2.chksum = None
# read back the pkt (with str()) to force computing these fields
# probably other ways to accomplish this are possible
tx_ip2 = IP(str(tx_ip2))
tx_ip2 = IP(scapy.compat.raw(tx_ip2))
self.assertEqual(rx_ip, tx_ip2)
@@ -609,7 +610,7 @@ class TestSRv6(VppTestCase):
tx_ip = tx_pkt.getlayer(IPv6)
# we can't just get the 2nd Ether layer
# get the Raw content and dissect it as Ether
tx_eth1 = Ether(str(tx_pkt[Raw]))
tx_eth1 = Ether(scapy.compat.raw(tx_pkt[Raw]))
# verify if rx'ed packet has no SRH
self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
@@ -655,7 +656,7 @@ class TestSRv6(VppTestCase):
# read back the dumped packet (with str())
# to force computing these fields
# probably other ways are possible
p = Ether(str(p))
p = Ether(scapy.compat.raw(p))
payload_info.data = p.copy()
self.logger.debug(ppp("Created packet:", p))
pkts.append(p)
@@ -817,7 +818,7 @@ class TestSRv6(VppTestCase):
# take packet[Raw], convert it to an Ether layer
# and then extract Raw from it
payload_info = self.payload_to_info(
Ether(str(packet[Raw]))[Raw])
Ether(scapy.compat.raw(packet[Raw]))[Raw])
return payload_info
@@ -831,7 +832,7 @@ class TestSRv6(VppTestCase):
:param compare_func: function to compare in and out packet
"""
self.logger.info("Verifying capture on interface %s using function %s"
% (dst_if.name, compare_func.func_name))
% (dst_if.name, compare_func.__name__))
last_info = dict()
for i in self.pg_interfaces:
test/util.py
+7 -6
View File
@@ -7,6 +7,7 @@ import six
import sys
import os.path
import scapy.compat
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP
from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment, IPv6ExtHdrRouting,\
@@ -282,7 +283,7 @@ def fragment_rfc791(packet, fragsize, _logger=None):
"""
logger = LoggerWrapper(_logger)
logger.debug(ppp("Fragmenting packet:", packet))
packet = packet.__class__(str(packet)) # recalculate all values
packet = packet.__class__(scapy.compat.raw(packet)) # recalc. all values
if len(packet[IP].options) > 0:
raise Exception("Not implemented")
if len(packet) <= fragsize:
@@ -290,7 +291,7 @@ def fragment_rfc791(packet, fragsize, _logger=None):
pre_ip_len = len(packet) - len(packet[IP])
ip_header_len = packet[IP].ihl * 4
hex_packet = str(packet)
hex_packet = scapy.compat.raw(packet)
hex_headers = hex_packet[:(pre_ip_len + ip_header_len)]
hex_payload = hex_packet[(pre_ip_len + ip_header_len):]
@@ -327,7 +328,7 @@ def fragment_rfc8200(packet, identification, fragsize, _logger=None):
:returns: list of fragments
"""
logger = LoggerWrapper(_logger)
packet = packet.__class__(str(packet)) # recalculate all values
packet = packet.__class__(scapy.compat.raw(packet)) # recalc. all values
if len(packet) <= fragsize:
return [packet]
logger.debug(ppp("Fragmenting packet:", packet))
@@ -378,7 +379,7 @@ def fragment_rfc8200(packet, identification, fragsize, _logger=None):
logger.debug(ppp("Per-fragment headers:", per_fragment_headers))
ext_and_upper_layer = packet.getlayer(last_per_fragment_hdr)[1]
hex_payload = str(ext_and_upper_layer)
hex_payload = scapy.compat.raw(ext_and_upper_layer)
logger.debug("Payload length is %s" % len(hex_payload))
logger.debug(ppp("Ext and upper layer:", ext_and_upper_layer))
@@ -407,7 +408,7 @@ def fragment_rfc8200(packet, identification, fragsize, _logger=None):
p[IPv6ExtHdrFragment].id = identification
p[IPv6ExtHdrFragment].offset = 0
p[IPv6ExtHdrFragment].m = 1
p = p.__class__(str(p))
p = p.__class__(scapy.compat.raw(p))
logger.debug(ppp("Fragment %s:" % len(pkts), p))
pkts.append(p)
offset = first_payload_len_nfb * 8
@@ -424,7 +425,7 @@ def fragment_rfc8200(packet, identification, fragsize, _logger=None):
p[IPv6ExtHdrFragment].id = identification
p[IPv6ExtHdrFragment].offset = offset / 8
p[IPv6ExtHdrFragment].m = 1
p = p.__class__(str(p))
p = p.__class__(scapy.compat.raw(p))
logger.debug(ppp("Fragment %s:" % len(pkts), p))
pkts.append(p)
offset = offset + l_nfb * 8
test/vpp_pg_interface.py
+7 -2
View File
@@ -3,6 +3,8 @@ import time
import socket
import struct
from traceback import format_exc, format_stack
import scapy.compat
from scapy.utils import wrpcap, rdpcap, PcapReader
from scapy.plist import PacketList
from vpp_interface import VppInterface
@@ -414,7 +416,7 @@ class VppPGInterface(VppInterface):
# Make Dot1AD packet content recognizable to scapy
if arp_reply.type == 0x88a8:
arp_reply.type = 0x8100
arp_reply = Ether(str(arp_reply))
arp_reply = Ether(scapy.compat.raw(arp_reply))
try:
if arp_reply[ARP].op == ARP.is_at:
self.test.logger.info("VPP %s MAC address is %s " %
@@ -460,8 +462,11 @@ class VppPGInterface(VppInterface):
ndp_reply = captured_packet.copy() # keep original for exception
# Make Dot1AD packet content recognizable to scapy
if ndp_reply.type == 0x88a8:
self._test.logger.info(
"Replacing EtherType: 0x88a8 with "
"0x8100 and regenerating Ethernet header. ")
ndp_reply.type = 0x8100
ndp_reply = Ether(str(ndp_reply))
ndp_reply = Ether(scapy.compat.raw(ndp_reply))
try:
ndp_na = ndp_reply[ICMPv6ND_NA]
opt = ndp_na[ICMPv6NDOptDstLLAddr]