tests/topotests/bgp_auth/bgp_auth_common.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_bgp_auth.py
   6 # Part of NetDEF Topology Tests
   7 #
   8 # Copyright (c) 2020 by Volta Networks
   9 #
  10
  11 """
  12 test_bgp_auth.py: Test BGP Md5 Authentication
  13
  14                              +------+
  15                     +--------|      |--------+
  16                     | +------|  R1  |------+ |
  17                     | | -----|      |----+ | |
  18                     | | |    +------+    | | |
  19                     | | |                | | |
  20                    +------+            +------+
  21                    |      |------------|      |
  22                    |  R2  |------------|  R3  |
  23                    |      |------------|      |
  24                    +------+            +------+
  25
  26
  27 setup is 3 routers with 3 links between each each link in a different vrf
  28 Default, blue and red respectively
  29 Tests check various fiddling with passwords and checking that the peer
  30 establishment is as expected and passwords are not leaked across sockets
  31 for bgp instances
  32 """
  33 # pylint: disable=C0413
  34
  35 import json
  36 import os
  37 import platform
  38 import sys
  39 from time import sleep
  40
  41 from lib import common_config, topotest
  42 from lib.common_config import (
  43     save_initial_config_on_routers,
  44     reset_with_new_configs,
  45 )
  46 from lib.topogen import Topogen, TopoRouter, get_topogen
  47
  48 CWD = os.path.dirname(os.path.realpath(__file__))
  49
  50
  51 def vrf_str(vrf):
  52     if vrf == "":
  53         vrf_str = ""
  54     else:
  55         vrf_str = "vrf {}".format(vrf)
  56
  57     return vrf_str
  58
  59
  60 def peer_name(rtr, prefix, vrf):
  61     "generate VRF string for CLI"
  62     if vrf == "":
  63         vrf_str = ""
  64     else:
  65         vrf_str = "_" + vrf
  66
  67     if prefix == "yes":
  68         if rtr == "R2":
  69             return "TWO_GROUP" + vrf_str
  70         else:
  71             return "THREE_GROUP" + vrf_str
  72     else:
  73         if rtr == "R2":
  74             return "2.2.2.2"
  75         else:
  76             return "3.3.3.3"
  77
  78
  79 def print_diag(vrf):
  80     "print failure disagnostics"
  81
  82     tgen = get_topogen()
  83     router_list = tgen.routers()
  84     for rname, router in router_list.items():
  85         print(rname + ":")
  86         print(router.vtysh_cmd("show run"))
  87         print(router.vtysh_cmd("show ip route {}".format(vrf_str(vrf))))
  88         print(router.vtysh_cmd("show bgp {} neighbor".format(vrf_str(vrf))))
  89
  90
  91 @common_config.retry(retry_timeout=190)
  92 def _check_neigh_state(router, peer, state, vrf=""):
  93     "check BGP neighbor state on a router"
  94
  95     neigh_output = router.vtysh_cmd(
  96         "show bgp {} neighbors {} json".format(vrf_str(vrf), peer)
  97     )
  98
  99     peer_state = "Unknown"
 100     neigh_output_json = json.loads(neigh_output)
 101     if peer in neigh_output_json:
 102         peer_state = neigh_output_json[peer]["bgpState"]
 103         if peer_state == state:
 104             return True
 105     return "{} peer with {} expected state {} got {} ".format(
 106         router.name, peer, state, peer_state
 107     )
 108
 109
 110 def check_neigh_state(router, peer, state, vrf=""):
 111     "check BGP neighbor state on a router"
 112
 113     assertmsg = _check_neigh_state(router, peer, state, vrf)
 114     assert assertmsg is True, assertmsg
 115
 116
 117 def check_all_peers_established(vrf=""):
 118     "standard check for extablished peers per vrf"
 119
 120     tgen = get_topogen()
 121     r1 = tgen.gears["R1"]
 122     r2 = tgen.gears["R2"]
 123     r3 = tgen.gears["R3"]
 124     # do r1 last as he might be the dynamic one
 125     check_neigh_state(r2, "1.1.1.1", "Established", vrf)
 126     check_neigh_state(r2, "3.3.3.3", "Established", vrf)
 127     check_neigh_state(r3, "1.1.1.1", "Established", vrf)
 128     check_neigh_state(r3, "2.2.2.2", "Established", vrf)
 129     check_neigh_state(r1, "2.2.2.2", "Established", vrf)
 130     check_neigh_state(r1, "3.3.3.3", "Established", vrf)
 131
 132
 133 def check_vrf_peer_remove_passwords(vrf="", prefix="no"):
 134     "selectively remove passwords checking state"
 135
 136     tgen = get_topogen()
 137     r1 = tgen.gears["R1"]
 138     r2 = tgen.gears["R2"]
 139     r3 = tgen.gears["R3"]
 140
 141     check_all_peers_established(vrf)
 142
 143     r1.vtysh_cmd(
 144         "conf t\nrouter bgp 65001 {}\nno neighbor {} password".format(
 145             vrf_str(vrf), peer_name("R2", prefix, vrf)
 146         )
 147     )
 148
 149     check_neigh_state(r2, "1.1.1.1", "Connect", vrf)
 150     check_neigh_state(r2, "3.3.3.3", "Established", vrf)
 151     check_neigh_state(r3, "1.1.1.1", "Established", vrf)
 152     check_neigh_state(r3, "2.2.2.2", "Established", vrf)
 153     # don't check dynamic downed peers - they are removed
 154     if prefix == "no":
 155         check_neigh_state(r1, "2.2.2.2", "Connect", vrf)
 156     check_neigh_state(r1, "3.3.3.3", "Established", vrf)
 157
 158     r2.vtysh_cmd(
 159         "conf t\nrouter bgp 65002 {}\nno neighbor 1.1.1.1 password".format(vrf_str(vrf))
 160     )
 161     check_all_peers_established(vrf)
 162
 163     r1.vtysh_cmd(
 164         "conf t\nrouter bgp 65001 {}\nno neighbor {} password".format(
 165             vrf_str(vrf), peer_name("R3", prefix, vrf)
 166         )
 167     )
 168     check_neigh_state(r2, "1.1.1.1", "Established", vrf)
 169     check_neigh_state(r2, "3.3.3.3", "Established", vrf)
 170     check_neigh_state(r3, "1.1.1.1", "Connect", vrf)
 171     check_neigh_state(r3, "2.2.2.2", "Established", vrf)
 172     check_neigh_state(r1, "2.2.2.2", "Established", vrf)
 173     # don't check dynamic downed peers - they are removed
 174     if prefix == "no":
 175         check_neigh_state(r1, "3.3.3.3", "Connect", vrf)
 176
 177     r3.vtysh_cmd(
 178         "conf t\nrouter bgp 65003 {}\nno neighbor 1.1.1.1 password".format(vrf_str(vrf))
 179     )
 180     check_all_peers_established(vrf)
 181
 182     r2.vtysh_cmd(
 183         "conf t\nrouter bgp 65002 {}\nno neighbor 3.3.3.3 password".format(vrf_str(vrf))
 184     )
 185     check_neigh_state(r2, "1.1.1.1", "Established", vrf)
 186     check_neigh_state(r2, "3.3.3.3", "Connect", vrf)
 187     check_neigh_state(r3, "1.1.1.1", "Established", vrf)
 188     check_neigh_state(r3, "2.2.2.2", "Connect", vrf)
 189     check_neigh_state(r1, "2.2.2.2", "Established", vrf)
 190     check_neigh_state(r1, "3.3.3.3", "Established", vrf)
 191
 192     r3.vtysh_cmd(
 193         "conf t\nrouter bgp 65003 {}\nno neighbor 2.2.2.2 password".format(vrf_str(vrf))
 194     )
 195     check_all_peers_established(vrf)
 196
 197
 198 def check_vrf_peer_change_passwords(vrf="", prefix="no"):
 199     "selectively change passwords checking state"
 200
 201     tgen = get_topogen()
 202     r1 = tgen.gears["R1"]
 203     r2 = tgen.gears["R2"]
 204     r3 = tgen.gears["R3"]
 205     check_all_peers_established(vrf)
 206
 207     r1.vtysh_cmd(
 208         "conf t\nrouter bgp 65001 {}\nneighbor {} password change1".format(
 209             vrf_str(vrf), peer_name("R2", prefix, vrf)
 210         )
 211     )
 212     check_neigh_state(r2, "1.1.1.1", "Connect", vrf)
 213     check_neigh_state(r2, "3.3.3.3", "Established", vrf)
 214     check_neigh_state(r3, "1.1.1.1", "Established", vrf)
 215     check_neigh_state(r3, "2.2.2.2", "Established", vrf)
 216     # don't check dynamic downed peers - they are removed
 217     if prefix == "no":
 218         check_neigh_state(r1, "2.2.2.2", "Connect", vrf)
 219     check_neigh_state(r1, "3.3.3.3", "Established", vrf)
 220
 221     r2.vtysh_cmd(
 222         "conf t\nrouter bgp 65002 {}\nneighbor 1.1.1.1 password change1".format(
 223             vrf_str(vrf)
 224         )
 225     )
 226     check_all_peers_established(vrf)
 227
 228     r1.vtysh_cmd(
 229         "conf t\nrouter bgp 65001 {}\nneighbor {} password change2".format(
 230             vrf_str(vrf), peer_name("R3", prefix, vrf)
 231         )
 232     )
 233     check_neigh_state(r2, "1.1.1.1", "Established", vrf)
 234     check_neigh_state(r2, "3.3.3.3", "Established", vrf)
 235     check_neigh_state(r3, "1.1.1.1", "Connect", vrf)
 236     check_neigh_state(r3, "2.2.2.2", "Established", vrf)
 237     check_neigh_state(r1, "2.2.2.2", "Established", vrf)
 238     # don't check dynamic downed peers - they are removed
 239     if prefix == "no":
 240         check_neigh_state(r1, "3.3.3.3", "Connect", vrf)
 241
 242     r3.vtysh_cmd(
 243         "conf t\nrouter bgp 65003 {}\nneighbor 1.1.1.1 password change2".format(
 244             vrf_str(vrf)
 245         )
 246     )
 247     check_all_peers_established(vrf)
 248
 249     r2.vtysh_cmd(
 250         "conf t\nrouter bgp 65002 {}\nneighbor 3.3.3.3 password change3".format(
 251             vrf_str(vrf)
 252         )
 253     )
 254     check_neigh_state(r2, "1.1.1.1", "Established", vrf)
 255     check_neigh_state(r2, "3.3.3.3", "Connect", vrf)
 256     check_neigh_state(r3, "1.1.1.1", "Established", vrf)
 257     check_neigh_state(r3, "2.2.2.2", "Connect", vrf)
 258     check_neigh_state(r1, "2.2.2.2", "Established", vrf)
 259     check_neigh_state(r1, "3.3.3.3", "Established", vrf)
 260
 261     r3.vtysh_cmd(
 262         "conf t\nrouter bgp 65003 {}\nneighbor 2.2.2.2 password change3".format(
 263             vrf_str(vrf)
 264         )
 265     )
 266     check_all_peers_established(vrf)