import socket
import unittest
import struct
-import StringIO
import random
from framework import VppTestCase, VppTestRunner, running_extended_tests
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, \
- ICMPv6ND_NS, ICMPv6ND_NA, ICMPv6NDOptDstLLAddr
+ ICMPv6ND_NS, ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, fragment6
from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6, IPv6ExtHdrFragment
from scapy.layers.l2 import Ether, ARP, GRE
from scapy.data import IP_PROTOS
from scapy.packet import bind_layers, Raw
-from scapy.all import fragment6
from util import ppp
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from time import sleep
from util import ip4_range
-from util import mactobinary
+from vpp_mac import mactobinary
from syslog_rfc5424_parser import SyslogMessage, ParseError
from syslog_rfc5424_parser.constants import SyslogFacility, SyslogSeverity
from vpp_papi_provider import SYSLOG_SEVERITY
+from io import BytesIO
class MethodHolder(VppTestCase):
for packet in capture:
try:
self.assertEqual(packet[IP].src, src_ip)
- self.assertTrue(packet.haslayer(ICMP))
+ self.assertEqual(packet.haslayer(ICMP), 1)
icmp = packet[ICMP]
self.assertEqual(icmp.type, icmp_type)
self.assertTrue(icmp.haslayer(IPerror))
for packet in capture:
try:
self.assertEqual(packet[IP].dst, in_if.remote_ip4)
- self.assertTrue(packet.haslayer(ICMP))
+ self.assertEqual(packet.haslayer(ICMP), 1)
icmp = packet[ICMP]
self.assertEqual(icmp.type, icmp_type)
self.assertTrue(icmp.haslayer(IPerror))
:returns: Reassembled IPv4 packet
"""
- buffer = StringIO.StringIO()
+ buffer = BytesIO()
for p in frags:
self.assertEqual(p[IP].src, src)
self.assertEqual(p[IP].dst, dst)
self.assert_ip_checksum_valid(p)
buffer.seek(p[IP].frag * 8)
- buffer.write(p[IP].payload)
+ buffer.write(bytes(p[IP].payload))
ip = IP(src=frags[0][IP].src, dst=frags[0][IP].dst,
proto=frags[0][IP].proto)
if ip.proto == IP_PROTOS.tcp:
:returns: Reassembled IPv6 packet
"""
- buffer = StringIO.StringIO()
+ buffer = BytesIO()
for p in frags:
self.assertEqual(p[IPv6].src, src)
self.assertEqual(p[IPv6].dst, dst)
buffer.seek(p[IPv6ExtHdrFragment].offset * 8)
- buffer.write(p[IPv6ExtHdrFragment].payload)
+ buffer.write(bytes(p[IPv6ExtHdrFragment].payload))
ip = IPv6(src=frags[0][IPv6].src, dst=frags[0][IPv6].dst,
nh=frags[0][IPv6ExtHdrFragment].nh)
if ip.nh == IP_PROTOS.tcp:
try:
self.assertEqual(packet[IP].src, nat_ip)
self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
- self.assertTrue(packet.haslayer(GRE))
+ self.assertEqual(packet.haslayer(GRE), 1)
self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
try:
self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
- self.assertTrue(packet.haslayer(GRE))
+ self.assertEqual(packet.haslayer(GRE), 1)
self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
try:
self.assertEqual(packet[IP].src, host_nat_ip)
self.assertEqual(packet[IP].dst, server.ip4)
- self.assertTrue(packet.haslayer(GRE))
+ self.assertEqual(packet.haslayer(GRE), 1)
self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
try:
self.assertEqual(packet[IP].src, server_nat_ip)
self.assertEqual(packet[IP].dst, host.ip4)
- self.assertTrue(packet.haslayer(GRE))
+ self.assertEqual(packet.haslayer(GRE), 1)
self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
try:
self.assertEqual(packet[IP].src, self.nat_addr)
self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
- self.assertTrue(packet.haslayer(GRE))
+ self.assertEqual(packet.haslayer(GRE), 1)
self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
try:
self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
- self.assertTrue(packet.haslayer(GRE))
+ self.assertEqual(packet.haslayer(GRE), 1)
self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
try:
self.assertEqual(packet[IP].src, self.nat_addr)
self.assertEqual(packet[IP].dst, server.ip4)
- self.assertTrue(packet.haslayer(GRE))
+ self.assertEqual(packet.haslayer(GRE), 1)
self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
try:
self.assertEqual(packet[IP].src, server_nat_ip)
self.assertEqual(packet[IP].dst, host.ip4)
- self.assertTrue(packet.haslayer(GRE))
+ self.assertEqual(packet.haslayer(GRE), 1)
self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
packet = capture[0]
try:
self.assertEqual(packet[IPv6].src, self.pg5.local_ip6)
- self.assertTrue(packet.haslayer(ICMPv6ND_NS))
+ self.assertEqual(packet.haslayer(ICMPv6ND_NS), 1)
tgt = packet[ICMPv6ND_NS].tgt
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
try:
self.assertEqual(packet[IPv6].src, self.pg5.local_ip6)
self.assertEqual(packet[IPv6].dst, self.pg5.remote_ip6)
- self.assertTrue(packet.haslayer(ICMPv6EchoReply))
+ self.assertEqual(packet.haslayer(ICMPv6EchoReply), 1)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
try:
self.assertEqual(packet[IP].src, self.nat_addr)
self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
- self.assertTrue(packet.haslayer(GRE))
+ self.assertEqual(packet.haslayer(GRE), 1)
self.assert_packet_checksums_valid(packet)
except:
self.logger.error(ppp("Unexpected or invalid packet:", packet))