|
1
|
|
|
"""Test Python-openflow network types.""" |
|
2
|
|
|
import unittest |
|
3
|
|
|
|
|
4
|
|
|
from pyof.foundation.basic_types import BinaryData |
|
5
|
|
|
from pyof.foundation.exceptions import UnpackException |
|
6
|
|
|
from pyof.foundation.network_types import ARP, VLAN, Ethernet, GenericTLV, IPv4 |
|
7
|
|
|
|
|
8
|
|
|
|
|
9
|
|
|
class TestARP(unittest.TestCase): |
|
10
|
|
|
"""Test ARP packets, without Ethernet headers.""" |
|
11
|
|
|
|
|
12
|
|
|
def test_arp_pack(self): |
|
13
|
|
|
"""Test pack method of ARP class.""" |
|
14
|
|
|
arp = ARP(oper=1, sha='00:15:af:d5:38:98', spa='172.16.0.10', |
|
15
|
|
|
tpa='172.16.10.20') |
|
16
|
|
|
packed = arp.pack() |
|
17
|
|
|
expected = b'\x00\x01\x08\x00\x06\x04\x00\x01\x00\x15\xaf\xd58\x98\xac' |
|
18
|
|
|
expected += b'\x10\x00\n\x00\x00\x00\x00\x00\x00\xac\x10\n\x14' |
|
19
|
|
|
self.assertEqual(packed, expected) |
|
20
|
|
|
|
|
21
|
|
|
def test_arp_unpack(self): |
|
22
|
|
|
"""Test unpack method of ARP class.""" |
|
23
|
|
|
raw = b'\x00\x01\x08\x00\x06\x04\x00\x02\x00\x1f:>\x9a\xcf\xac\x10\n' |
|
24
|
|
|
raw += b'\x14\x00\x15\xaf\xd58\x98\xac\x10\x00\n' |
|
25
|
|
|
expected = ARP(oper=2, sha='00:1f:3a:3e:9a:cf', spa='172.16.10.20', |
|
26
|
|
|
tha='00:15:af:d5:38:98', tpa='172.16.0.10') |
|
27
|
|
|
unpacked = ARP() |
|
28
|
|
|
unpacked.unpack(raw) |
|
29
|
|
|
self.assertEqual(unpacked, expected) |
|
30
|
|
|
|
|
31
|
|
|
def test_unpack_invalid_htype(self): |
|
32
|
|
|
"""Raise UnpackException when L2 protocol is not Ethernet.""" |
|
33
|
|
|
raw = b'\x01\x23\x08\x00\x06\x04\x00\x02\x00\x1f:>\x9a\xcf\xac\x10\n' |
|
34
|
|
|
raw += b'\x14\x00\x15\xaf\xd58\x98\xac\x10\x00\n' |
|
35
|
|
|
arp = ARP() |
|
36
|
|
|
with self.assertRaises(UnpackException): |
|
37
|
|
|
arp.unpack(raw) |
|
38
|
|
|
|
|
39
|
|
|
def test_unpack_invalid_ptype(self): |
|
40
|
|
|
"""Raise UnpackException when L3 protocol is not IPv4.""" |
|
41
|
|
|
raw = b'\x00\x01\x08\x90\x06\x04\x00\x02\x00\x1f:>\x9a\xcf\xac\x10\n' |
|
42
|
|
|
raw += b'\x14\x00\x15\xaf\xd58\x98\xac\x10\x00\n' |
|
43
|
|
|
arp = ARP() |
|
44
|
|
|
with self.assertRaises(UnpackException): |
|
45
|
|
|
arp.unpack(raw) |
|
46
|
|
|
|
|
47
|
|
|
|
|
48
|
|
|
class TestNetworkTypes(unittest.TestCase): |
|
49
|
|
|
"""Reproduce bugs found.""" |
|
50
|
|
|
|
|
51
|
|
|
def test_GenTLV_value_unpack(self): |
|
52
|
|
|
"""Value attribute should be the same after unpacking.""" |
|
53
|
|
|
value = BinaryData(b'test') |
|
54
|
|
|
tlv = GenericTLV(value=value) |
|
55
|
|
|
tlv_unpacked = GenericTLV() |
|
56
|
|
|
tlv_unpacked.unpack(tlv.pack()) |
|
57
|
|
|
self.assertEqual(tlv.value.value, tlv_unpacked.value.value) |
|
58
|
|
|
|
|
59
|
|
|
|
|
60
|
|
|
class TestEthernet(unittest.TestCase): |
|
61
|
|
|
"""Test Ethernet frames.""" |
|
62
|
|
|
|
|
63
|
|
|
def test_Ethernet_pack(self): |
|
64
|
|
|
"""Test pack method of Ethernet class without VLAN tag.""" |
|
65
|
|
|
ethernet = Ethernet(destination='00:1f:3a:3e:9a:cf', |
|
66
|
|
|
source='00:15:af:d5:38:98', ether_type=0x800, |
|
67
|
|
|
data=b'testdata') |
|
68
|
|
|
packed = ethernet.pack() |
|
69
|
|
|
expected = b'\x00\x1f:>\x9a\xcf\x00\x15\xaf\xd58\x98\x08\x00testdata' |
|
70
|
|
|
self.assertEqual(packed, expected) |
|
71
|
|
|
|
|
72
|
|
|
def test_Ethernet_unpack(self): |
|
73
|
|
|
"""Test pack method of Ethernet class without VLAN tag.""" |
|
74
|
|
|
raw = b'\x00\x15\xaf\xd58\x98\x00\x1f:>\x9a\xcf\x08\x00testdata' |
|
75
|
|
|
expected = Ethernet(destination='00:15:af:d5:38:98', |
|
76
|
|
|
source='00:1f:3a:3e:9a:cf', ether_type=0x800, |
|
77
|
|
|
data=b'testdata') |
|
78
|
|
|
expected.pack() |
|
79
|
|
|
unpacked = Ethernet() |
|
80
|
|
|
unpacked.unpack(raw) |
|
81
|
|
|
self.assertEqual(unpacked, expected) |
|
82
|
|
|
|
|
83
|
|
|
def test_Tagged_Ethernet_pack(self): |
|
84
|
|
|
"""Test pack method of Ethernet class including VLAN tag.""" |
|
85
|
|
|
ethernet = Ethernet(destination='00:1f:3a:3e:9a:cf', |
|
86
|
|
|
source='00:15:af:d5:38:98', vlan=VLAN(vid=200), |
|
87
|
|
|
ether_type=0x800, data=b'testdata') |
|
88
|
|
|
packed = ethernet.pack() |
|
89
|
|
|
expected = b'\x00\x1f:>\x9a\xcf\x00\x15\xaf\xd58' |
|
90
|
|
|
expected += b'\x98\x81\x00\x00\xc8\x08\x00testdata' |
|
91
|
|
|
self.assertEqual(packed, expected) |
|
92
|
|
|
|
|
93
|
|
|
def test_Tagged_Ethernet_unpack(self): |
|
94
|
|
|
"""Test pack method of Ethernet class including VLAN tag.""" |
|
95
|
|
|
raw = b'\x00\x15\xaf\xd58\x98\x00\x1f:>' |
|
96
|
|
|
raw += b'\x9a\xcf\x81\x00!^\x08\x00testdata' |
|
97
|
|
|
expected = Ethernet(destination='00:15:af:d5:38:98', |
|
98
|
|
|
source='00:1f:3a:3e:9a:cf', vlan=VLAN(pcp=1, |
|
99
|
|
|
vid=350), |
|
100
|
|
|
ether_type=0x800, data=b'testdata') |
|
101
|
|
|
expected.pack() |
|
102
|
|
|
unpacked = Ethernet() |
|
103
|
|
|
unpacked.unpack(raw) |
|
104
|
|
|
self.assertEqual(unpacked, expected) |
|
105
|
|
|
|
|
106
|
|
|
|
|
107
|
|
|
class TestVLAN(unittest.TestCase): |
|
108
|
|
|
"""Test VLAN headers.""" |
|
109
|
|
|
|
|
110
|
|
|
def test_VLAN_pack(self): |
|
111
|
|
|
"""Test pack method of VLAN class.""" |
|
112
|
|
|
vlan = VLAN(pcp=3, vid=20) |
|
113
|
|
|
packed = vlan.pack() |
|
114
|
|
|
expected = b'\x81\x00`\x14' |
|
115
|
|
|
self.assertEqual(packed, expected) |
|
116
|
|
|
|
|
117
|
|
|
def test_VLAN_unpack(self): |
|
118
|
|
|
"""Test unpack method of VLAN class.""" |
|
119
|
|
|
raw = b'\x81\x00\xa0{' |
|
120
|
|
|
expected = VLAN(pcp=5, vid=123) |
|
121
|
|
|
unpacked = VLAN() |
|
122
|
|
|
unpacked.unpack(raw) |
|
123
|
|
|
self.assertEqual(unpacked, expected) |
|
124
|
|
|
|
|
125
|
|
|
def test_unpack_wrong_tpid(self): |
|
126
|
|
|
"""Raise UnpackException if the tpid is not VLAN_TPID.""" |
|
127
|
|
|
raw = b'\x12\x34\xa0{' |
|
128
|
|
|
vlan = VLAN() |
|
129
|
|
|
with self.assertRaises(UnpackException): |
|
130
|
|
|
vlan.unpack(raw) |
|
131
|
|
|
|
|
132
|
|
|
|
|
133
|
|
|
class TestIPv4(unittest.TestCase): |
|
134
|
|
|
"""Test IPv4 packets.""" |
|
135
|
|
|
|
|
136
|
|
|
def test_IPv4_pack(self): |
|
137
|
|
|
"""Test pack/unpack of IPv4 class.""" |
|
138
|
|
|
packet = IPv4(dscp=10, ttl=64, protocol=17, source="192.168.0.10", |
|
139
|
|
|
destination="172.16.10.30", options=b'1000', |
|
140
|
|
|
data=b'testdata') |
|
141
|
|
|
packed = packet.pack() |
|
142
|
|
|
expected = b'F(\x00 \x00\x00\x00\x00@\x11\x02' |
|
143
|
|
|
expected += b'\xc5\xc0\xa8\x00\n\xac\x10\n\x1e1000testdata' |
|
144
|
|
|
self.assertEqual(packed, expected) |
|
145
|
|
|
|
|
146
|
|
|
def test_IPv4_unpack(self): |
|
147
|
|
|
"""Test unpack of IPv4 binary packet.""" |
|
148
|
|
|
raw = b'FP\x00$\x00\x00\x00\x00\x80\x06W' |
|
149
|
|
|
raw += b'\xf4\n\x9aN\x81\xc0\xa8\xc7\xcc1000somemoredata' |
|
150
|
|
|
expected = IPv4(dscp=20, ttl=128, protocol=6, source="10.154.78.129", |
|
151
|
|
|
destination="192.168.199.204", options=b'1000', |
|
152
|
|
|
data=b'somemoredata') |
|
153
|
|
|
expected.pack() |
|
154
|
|
|
unpacked = IPv4() |
|
155
|
|
|
unpacked.unpack(raw) |
|
156
|
|
|
self.assertEqual(unpacked, expected) |
|
157
|
|
|
|
|
158
|
|
|
def test_IPv4_size(self): |
|
159
|
|
|
"""Test Header size for IPv4 packet.""" |
|
160
|
|
|
packet = IPv4() |
|
161
|
|
|
packet.pack() |
|
162
|
|
|
self.assertEqual(20, packet.get_size()) |
|
163
|
|
|
self.assertEqual(20, packet.length) |
|
164
|
|
|
self.assertEqual(20, packet.ihl * 4) |
|
165
|
|
|
|
|
166
|
|
|
def test_IPv4_checksum(self): |
|
167
|
|
|
"""Test if the IPv4 checksum is being calculated correclty.""" |
|
168
|
|
|
packet = IPv4(dscp=10, ttl=64, protocol=17, source="192.168.0.10", |
|
169
|
|
|
destination="172.16.10.30", options=b'1000', |
|
170
|
|
|
data=b'testdata') |
|
171
|
|
|
packet.pack() |
|
172
|
|
|
self.assertEqual(packet.checksum, 709) |
|
173
|
|
|
|
kytos /
python-openflow
