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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
|
#!/usr/bin/env python
# SPDX-License-Identifier: ISC
#
# test_bgp_auth.py
# Part of NetDEF Topology Tests
#
# Copyright (c) 2020 by Volta Networks
#
"""
test_bgp_auth.py: Test BGP Md5 Authentication
+------+
+--------| |--------+
| +------| R1 |------+ |
| | -----| |----+ | |
| | | +------+ | | |
| | | | | |
+------+ +------+
| |------------| |
| R2 |------------| R3 |
| |------------| |
+------+ +------+
setup is 3 routers with 3 links between each each link in a different vrf
Default, blue and red respectively
Tests check various fiddling with passwords and checking that the peer
establishment is as expected and passwords are not leaked across sockets
for bgp instances
"""
# pylint: disable=C0413
import json
import os
import platform
import sys
from time import sleep
import pytest
from lib import common_config, topotest
from lib.common_config import (
save_initial_config_on_routers,
reset_with_new_configs,
)
from bgp_auth_common import (
check_vrf_peer_change_passwords,
check_all_peers_established,
check_vrf_peer_remove_passwords,
)
from lib.topogen import Topogen, TopoRouter, get_topogen
pytestmark = [pytest.mark.bgpd, pytest.mark.ospfd]
CWD = os.path.dirname(os.path.realpath(__file__))
def build_topo(tgen):
tgen.add_router("R1")
tgen.add_router("R2")
tgen.add_router("R3")
tgen.add_link(tgen.gears["R1"], tgen.gears["R2"])
tgen.add_link(tgen.gears["R1"], tgen.gears["R3"])
tgen.add_link(tgen.gears["R2"], tgen.gears["R3"])
tgen.add_link(tgen.gears["R1"], tgen.gears["R2"])
tgen.add_link(tgen.gears["R1"], tgen.gears["R3"])
tgen.add_link(tgen.gears["R2"], tgen.gears["R3"])
tgen.add_link(tgen.gears["R1"], tgen.gears["R2"])
tgen.add_link(tgen.gears["R1"], tgen.gears["R3"])
tgen.add_link(tgen.gears["R2"], tgen.gears["R3"])
def setup_module(mod):
"Sets up the pytest environment"
# This function initiates the topology build with Topogen...
tgen = Topogen(build_topo, mod.__name__)
# ... and here it calls Mininet initialization functions.
tgen.start_topology()
r1 = tgen.gears["R1"]
r2 = tgen.gears["R2"]
r3 = tgen.gears["R3"]
# blue vrf
r1.cmd_raises("ip link add blue type vrf table 1001")
r1.cmd_raises("ip link set up dev blue")
r2.cmd_raises("ip link add blue type vrf table 1001")
r2.cmd_raises("ip link set up dev blue")
r3.cmd_raises("ip link add blue type vrf table 1001")
r3.cmd_raises("ip link set up dev blue")
r1.cmd_raises("ip link add lo1 type dummy")
r1.cmd_raises("ip link set lo1 master blue")
r1.cmd_raises("ip link set up dev lo1")
r2.cmd_raises("ip link add lo1 type dummy")
r2.cmd_raises("ip link set up dev lo1")
r2.cmd_raises("ip link set lo1 master blue")
r3.cmd_raises("ip link add lo1 type dummy")
r3.cmd_raises("ip link set up dev lo1")
r3.cmd_raises("ip link set lo1 master blue")
r1.cmd_raises("ip link set R1-eth2 master blue")
r1.cmd_raises("ip link set R1-eth3 master blue")
r2.cmd_raises("ip link set R2-eth2 master blue")
r2.cmd_raises("ip link set R2-eth3 master blue")
r3.cmd_raises("ip link set R3-eth2 master blue")
r3.cmd_raises("ip link set R3-eth3 master blue")
r1.cmd_raises("ip link set up dev R1-eth2")
r1.cmd_raises("ip link set up dev R1-eth3")
r2.cmd_raises("ip link set up dev R2-eth2")
r2.cmd_raises("ip link set up dev R2-eth3")
r3.cmd_raises("ip link set up dev R3-eth2")
r3.cmd_raises("ip link set up dev R3-eth3")
# red vrf
r1.cmd_raises("ip link add red type vrf table 1002")
r1.cmd_raises("ip link set up dev red")
r2.cmd_raises("ip link add red type vrf table 1002")
r2.cmd_raises("ip link set up dev red")
r3.cmd_raises("ip link add red type vrf table 1002")
r3.cmd_raises("ip link set up dev red")
r1.cmd_raises("ip link add lo2 type dummy")
r1.cmd_raises("ip link set lo2 master red")
r1.cmd_raises("ip link set up dev lo2")
r2.cmd_raises("ip link add lo2 type dummy")
r2.cmd_raises("ip link set up dev lo2")
r2.cmd_raises("ip link set lo2 master red")
r3.cmd_raises("ip link add lo2 type dummy")
r3.cmd_raises("ip link set up dev lo2")
r3.cmd_raises("ip link set lo2 master red")
r1.cmd_raises("ip link set R1-eth4 master red")
r1.cmd_raises("ip link set R1-eth5 master red")
r2.cmd_raises("ip link set R2-eth4 master red")
r2.cmd_raises("ip link set R2-eth5 master red")
r3.cmd_raises("ip link set R3-eth4 master red")
r3.cmd_raises("ip link set R3-eth5 master red")
r1.cmd_raises("ip link set up dev R1-eth4")
r1.cmd_raises("ip link set up dev R1-eth5")
r2.cmd_raises("ip link set up dev R2-eth4")
r2.cmd_raises("ip link set up dev R2-eth5")
r3.cmd_raises("ip link set up dev R3-eth4")
r3.cmd_raises("ip link set up dev R3-eth5")
r1.cmd_raises("sysctl -w net.ipv4.tcp_l3mdev_accept=1")
r2.cmd_raises("sysctl -w net.ipv4.tcp_l3mdev_accept=1")
r3.cmd_raises("sysctl -w net.ipv4.tcp_l3mdev_accept=1")
# This is a sample of configuration loading.
router_list = tgen.routers()
# For all registered routers, load the zebra configuration file
for rname, router in router_list.items():
router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
router.load_config(TopoRouter.RD_OSPF, "")
router.load_config(TopoRouter.RD_BGP, "")
# After copying the configurations, this function loads configured daemons.
tgen.start_router()
# Save the initial router config. reset_config_on_routers will return to this config.
save_initial_config_on_routers(tgen)
def teardown_module(mod):
"Teardown the pytest environment"
tgen = get_topogen()
# This function tears down the whole topology.
tgen.stop_topology()
def test_multiple_vrf_peer_change_passwords(tgen):
"selectively change passwords checking state with multiple VRFs"
reset_with_new_configs(tgen, "bgpd_multi_vrf.conf", "ospfd_multi_vrf.conf")
check_vrf_peer_change_passwords("blue")
check_all_peers_established("red")
check_vrf_peer_change_passwords("red")
check_all_peers_established("blue")
def test_multiple_vrf_prefix_peer_established(tgen):
"default vrf 3 peers same password with multilpe VRFs and prefix config"
# only supported in kernel > 5.3
if topotest.version_cmp(platform.release(), "5.3") < 0:
return
reset_with_new_configs(tgen, "bgpd_multi_vrf_prefix.conf", "ospfd_multi_vrf.conf")
check_all_peers_established("blue")
check_all_peers_established("red")
def test_multiple_vrf_prefix_peer_remove_passwords(tgen):
"selectively remove passwords checking state with multiple vrfs and prefix config"
# only supported in kernel > 5.3
if topotest.version_cmp(platform.release(), "5.3") < 0:
return
reset_with_new_configs(tgen, "bgpd_multi_vrf_prefix.conf", "ospfd_multi_vrf.conf")
check_vrf_peer_remove_passwords(vrf="blue", prefix="yes")
check_all_peers_established("red")
check_vrf_peer_remove_passwords(vrf="red", prefix="yes")
check_all_peers_established("blue")
def test_multiple_vrf_prefix_peer_change_passwords(tgen):
"selectively change passwords checking state with multiple vrfs and prefix config"
# only supported in kernel > 5.3
if topotest.version_cmp(platform.release(), "5.3") < 0:
return
reset_with_new_configs(tgen, "bgpd_multi_vrf_prefix.conf", "ospfd_multi_vrf.conf")
check_vrf_peer_change_passwords(vrf="blue", prefix="yes")
check_all_peers_established("red")
check_vrf_peer_change_passwords(vrf="red", prefix="yes")
check_all_peers_established("blue")
def test_memory_leak(tgen):
"Run the memory leak test and report results."
if not tgen.is_memleak_enabled():
pytest.skip("Memory leak test/report is disabled")
tgen.report_memory_leaks()
if __name__ == "__main__":
args = ["-s"] + sys.argv[1:]
sys.exit(pytest.main(args))
|
