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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
#!/usr/bin/env python3
import unittest
from framework import VppTestCase, VppTestRunner, running_extended_tests
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
from scapy.contrib.geneve import GENEVE
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
from scapy.layers.vxlan import VXLAN
from scapy.compat import raw
class TestTracefilter(VppTestCase):
""" Packet Tracer Filter Test """
@classmethod
def setUpClass(cls):
super(TestTracefilter, cls).setUpClass()
@classmethod
def tearDownClass(cls):
super(TestTracefilter, cls).tearDownClass()
def setUp(self):
super(TestTracefilter, self).setUp()
self.create_pg_interfaces(range(1))
for i in self.pg_interfaces:
i.admin_up()
i.config_ip4()
def tearDown(self):
super(TestTracefilter, self).tearDown()
for i in self.pg_interfaces:
i.unconfig()
i.admin_down()
def cli(self, cmd):
r = self.vapi.cli_return_response(cmd)
if r.retval != 0:
if hasattr(r, 'reply'):
self.logger.info(cmd + " FAIL reply " + r.reply)
else:
self.logger.info(cmd + " FAIL retval " + str(r.retval))
return r
# check number of hits for classifier
def assert_hits(self, n):
r = self.cli("show classify table verbose 2")
self.assertTrue(r.retval == 0)
self.assertTrue(hasattr(r, 'reply'))
self.assertTrue(r.reply.find("hits %i" % n) != -1)
def test_mactime_unitTest(self):
""" Packet Tracer Filter Test """
cmds = ["loopback create",
"set int ip address loop0 192.168.1.1/24",
"set int state loop0 up",
"packet-generator new {\n"
" name classifyme\n"
" limit 100\n"
" size 300-300\n"
" interface loop0\n"
" node ethernet-input\n"
" data { \n"
" IP4: 1.2.3 -> 4.5.6\n"
" UDP: 192.168.1.10 - 192.168.1.20 -> 192.168.2.10\n"
" UDP: 1234 -> 2345\n"
" incrementing 286\n"
" }\n"
"}\n",
"classify filter trace mask l3 ip4 src"
" match l3 ip4 src 192.168.1.15",
"trace add pg-input 100 filter",
"pa en classifyme"]
for cmd in cmds:
self.cli(cmd)
# Check for 9 classifier hits, which is the right answer
self.assert_hits(9)
# cleanup
self.cli("pa de classifyme")
self.cli("classify filter trace del mask l3 ip4 src")
# install a classify rule, inject traffic and check for hits
def assert_classify(self, mask, match, packets, n=None):
r = self.cli(
"classify filter trace mask hex %s match hex %s" %
(mask, match))
self.assertTrue(r.retval == 0)
r = self.cli("trace add pg-input %i filter" % len(packets))
self.assertTrue(r.retval == 0)
self.pg0.add_stream(packets)
self.cli("pa en")
self.assert_hits(n if n is not None else len(packets))
self.cli("clear trace")
self.cli(
"classify filter trace del mask hex %s" %
(mask))
def test_encap(self):
""" Packet Tracer Filter Test with encap """
# the packet we are trying to match
p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
UDP() /
VXLAN() /
Ether() /
IP() /
UDP() /
GENEVE(vni=1234) /
Ether() /
IP(src='192.168.4.167') /
UDP() /
Raw('\xa5' * 100))
#
# compute filter mask & value
# we compute it by XOR'ing a template packet with a modified packet
# we need to set checksums to 0 to make sure scapy will not recompute
# them
#
tmpl = (Ether() /
IP(chksum=0) /
UDP(chksum=0) /
VXLAN() /
Ether() /
IP(chksum=0) /
UDP(chksum=0) /
GENEVE(vni=0) /
Ether() /
IP(src='0.0.0.0', chksum=0))
ori = raw(tmpl)
# the mask
tmpl[GENEVE].vni = 0xffffff
user = tmpl[GENEVE].payload
user[IP].src = '255.255.255.255'
new = raw(tmpl)
mask = "".join(("{:02x}".format(o ^ n) for o, n in zip(ori, new)))
# this does not match (wrong vni)
tmpl[GENEVE].vni = 1
user = tmpl[GENEVE].payload
user[IP].src = '192.168.4.167'
new = raw(tmpl)
match = "".join(("{:02x}".format(o ^ n) for o, n in zip(ori, new)))
self.assert_classify(mask, match, [p] * 11, 0)
# this must match
tmpl[GENEVE].vni = 1234
new = raw(tmpl)
match = "".join(("{:02x}".format(o ^ n) for o, n in zip(ori, new)))
self.assert_classify(mask, match, [p] * 17)
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
|
