1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
#!/usr/bin/env python
from abc import abstractmethod, ABCMeta
from scapy.layers.l2 import Ether, Raw
from scapy.layers.inet import IP, UDP
class BridgeDomain(object):
""" Bridge domain abstraction """
__metaclass__ = ABCMeta
@property
def frame_pg0_to_pg1(self):
""" Ethernet frame sent from pg0 and expected to arrive at pg1 """
return (Ether(src='00:00:00:00:00:01', dst='00:00:00:00:00:02') /
IP(src='1.2.3.4', dst='4.3.2.1') /
UDP(sport=10000, dport=20000) /
Raw('\xa5' * 100))
@property
def frame_pg1_to_pg0(self):
""" Ethernet frame sent from pg1 and expected to arrive at pg0 """
return (Ether(src='00:00:00:00:00:02', dst='00:00:00:00:00:01') /
IP(src='4.3.2.1', dst='1.2.3.4') /
UDP(sport=20000, dport=10000) /
Raw('\xa5' * 100))
@abstractmethod
def encapsulate(self, pkt):
""" Encapsulate packet """
pass
@abstractmethod
def decapsulate(self, pkt):
""" Decapsulate packet """
pass
@abstractmethod
def check_encapsulation(self, pkt):
""" Verify the encapsulation """
pass
def test_decap(self):
""" Decapsulation test
Send encapsulated frames from pg0
Verify receipt of decapsulated frames on pg1
"""
encapsulated_pkt = self.encapsulate(self.frame_pg0_to_pg1)
self.pg0.add_stream([encapsulated_pkt, ])
self.pg1.enable_capture()
self.pg_start()
# Pick first received frame and check if it's the non-encapsulated frame
out = self.pg1.get_capture()
self.assertEqual(len(out), 1,
'Invalid number of packets on '
'output: {}'.format(len(out)))
pkt = out[0]
# TODO: add error messages
self.assertEqual(pkt[Ether].src, self.frame_pg0_to_pg1[Ether].src)
self.assertEqual(pkt[Ether].dst, self.frame_pg0_to_pg1[Ether].dst)
self.assertEqual(pkt[IP].src, self.frame_pg0_to_pg1[IP].src)
self.assertEqual(pkt[IP].dst, self.frame_pg0_to_pg1[IP].dst)
self.assertEqual(pkt[UDP].sport, self.frame_pg0_to_pg1[UDP].sport)
self.assertEqual(pkt[UDP].dport, self.frame_pg0_to_pg1[UDP].dport)
self.assertEqual(pkt[Raw], self.frame_pg0_to_pg1[Raw])
def test_encap(self):
""" Encapsulation test
Send frames from pg1
Verify receipt of encapsulated frames on pg0
"""
self.pg1.add_stream([self.frame_pg1_to_pg0])
self.pg0.enable_capture()
self.pg_start()
# Pick first received frame and check if it's corectly encapsulated.
out = self.pg0.get_capture()
self.assertEqual(len(out), 1,
'Invalid number of packets on '
'output: {}'.format(len(out)))
pkt = out[0]
self.check_encapsulation(pkt)
payload = self.decapsulate(pkt)
# TODO: add error messages
self.assertEqual(payload[Ether].src, self.frame_pg1_to_pg0[Ether].src)
self.assertEqual(payload[Ether].dst, self.frame_pg1_to_pg0[Ether].dst)
self.assertEqual(payload[IP].src, self.frame_pg1_to_pg0[IP].src)
self.assertEqual(payload[IP].dst, self.frame_pg1_to_pg0[IP].dst)
self.assertEqual(payload[UDP].sport, self.frame_pg1_to_pg0[UDP].sport)
self.assertEqual(payload[UDP].dport, self.frame_pg1_to_pg0[UDP].dport)
self.assertEqual(payload[Raw], self.frame_pg1_to_pg0[Raw])
|
