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
|
from scapy.layers.l2 import Ether, Dot1Q
from abc import abstractmethod, ABCMeta
from vpp_interface import VppInterface
from vpp_pg_interface import VppPGInterface
class VppSubInterface(VppPGInterface):
__metaclass__ = ABCMeta
@property
def parent(self):
"""Parent interface for this sub-interface"""
return self._parent
@property
def sub_id(self):
"""Sub-interface ID"""
return self._sub_id
def __init__(self, test, parent, sub_id):
self._test = test
self._parent = parent
self._parent.add_sub_if(self)
self._sub_id = sub_id
@abstractmethod
def create_arp_req(self):
pass
@abstractmethod
def create_ndp_req(self):
pass
def resolve_arp(self):
super(VppSubInterface, self).resolve_arp(self.parent)
def resolve_ndp(self):
super(VppSubInterface, self).resolve_ndp(self.parent)
@abstractmethod
def add_dot1_layer(self, pkt):
pass
def remove_vpp_config(self):
self.test.vapi.delete_subif(self._sw_if_index)
class VppDot1QSubint(VppSubInterface):
@property
def vlan(self):
"""VLAN tag"""
return self._vlan
def __init__(self, test, parent, sub_id, vlan=None):
if vlan is None:
vlan = sub_id
super(VppDot1QSubint, self).__init__(test, parent, sub_id)
self._vlan = vlan
r = self.test.vapi.create_vlan_subif(parent.sw_if_index, self.vlan)
self._sw_if_index = r.sw_if_index
VppInterface.post_init_setup(self)
def create_arp_req(self):
packet = VppPGInterface.create_arp_req(self)
return self.add_dot1_layer(packet)
def create_ndp_req(self):
packet = VppPGInterface.create_ndp_req(self)
return self.add_dot1_layer(packet)
def add_dot1_layer(self, packet):
payload = packet.payload
packet.remove_payload()
packet.add_payload(Dot1Q(vlan=self.sub_id) / payload)
return packet
def remove_dot1_layer(self, packet):
payload = packet.payload
self.test.instance().assertEqual(type(payload), Dot1Q)
self.test.instance().assertEqual(payload.vlan, self.vlan)
payload = payload.payload
packet.remove_payload()
packet.add_payload(payload)
return packet
class VppDot1ADSubint(VppSubInterface):
@property
def outer_vlan(self):
"""Outer VLAN tag"""
return self._outer_vlan
@property
def inner_vlan(self):
"""Inner VLAN tag"""
return self._inner_vlan
def __init__(self, test, parent, sub_id, outer_vlan, inner_vlan):
super(VppDot1ADSubint, self).__init__(test, parent, sub_id)
self.DOT1AD_TYPE = 0x88A8
self.DOT1Q_TYPE = 0x8100
self._outer_vlan = outer_vlan
self._inner_vlan = inner_vlan
r = self.test.vapi.create_subif(
parent.sw_if_index,
self.sub_id,
self.outer_vlan,
self.inner_vlan,
dot1ad=1,
two_tags=1,
exact_match=1)
self._sw_if_index = r.sw_if_index
VppInterface.post_init_setup(self)
def create_arp_req(self):
packet = VppPGInterface.create_arp_req(self)
return self.add_dot1_layer(packet)
def create_ndp_req(self):
packet = VppPGInterface.create_ndp_req(self)
return self.add_dot1_layer(packet)
def add_dot1_layer(self, packet):
payload = packet.payload
packet.remove_payload()
packet.add_payload(Dot1Q(vlan=self.outer_vlan) /
Dot1Q(vlan=self.inner_vlan) / payload)
packet.type = self.DOT1AD_TYPE
return packet
def remove_dot1_layer(self, packet):
self.test.instance().assertEqual(type(packet), Ether)
self.test.instance().assertEqual(packet.type, self.DOT1AD_TYPE)
packet.type = self.DOT1Q_TYPE
packet = Ether(str(packet))
payload = packet.payload
self.test.instance().assertEqual(type(payload), Dot1Q)
self.test.instance().assertEqual(payload.vlan, self.outer_vlan)
payload = payload.payload
self.test.instance().assertEqual(type(payload), Dot1Q)
self.test.instance().assertEqual(payload.vlan, self.inner_vlan)
payload = payload.payload
packet.remove_payload()
packet.add_payload(payload)
return packet
|
