tests/topotests/bgp_ebgp_requires_policy/test_bgp_ebgp_requires_policy.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # bgp_ebgp_requires_policy.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2019 by
   8 # Donatas Abraitis <donatas.abraitis@gmail.com>
   9 #
  10 # Permission to use, copy, modify, and/or distribute this software
  11 # for any purpose with or without fee is hereby granted, provided
  12 # that the above copyright notice and this permission notice appear
  13 # in all copies.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  16 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  17 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  18 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  19 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  20 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  21 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  22 # OF THIS SOFTWARE.
  23 #
  24
  25 """
  26 bgp_ebgp_requires_policy.py:
  27
  28 Test if eBGP sender without a filter applied to the peer is allowed
  29 to send advertisements.
  30
  31 Scenario 1:
  32   r1 has a filter applied for outgoing direction,
  33   r2 receives 192.168.255.1/32.
  34
  35 Scenario 2:
  36   r3 hasn't a filter appied for outgoing direction,
  37   r4 does not receive 192.168.255.1/32.
  38
  39 Scenario 3:
  40   r5 and r6 establish iBGP session which in turn should ignore
  41   RFC8212. All routes for both directions MUST work.
  42 """
  43
  44 import os
  45 import sys
  46 import json
  47 import time
  48 import pytest
  49 import functools
  50
  51 CWD = os.path.dirname(os.path.realpath(__file__))
  52 sys.path.append(os.path.join(CWD, "../"))
  53
  54 # pylint: disable=C0413
  55 from lib import topotest
  56 from lib.topogen import Topogen, TopoRouter, get_topogen
  57 from lib.topolog import logger
  58 from mininet.topo import Topo
  59
  60 pytestmark = [pytest.mark.bgpd]
  61
  62
  63 class TemplateTopo(Topo):
  64     def build(self, *_args, **_opts):
  65         tgen = get_topogen(self)
  66
  67         for routern in range(1, 7):
  68             tgen.add_router("r{}".format(routern))
  69
  70         # Scenario 1.
  71         switch = tgen.add_switch("s1")
  72         switch.add_link(tgen.gears["r1"])
  73         switch.add_link(tgen.gears["r2"])
  74
  75         # Scenario 2.
  76         switch = tgen.add_switch("s2")
  77         switch.add_link(tgen.gears["r3"])
  78         switch.add_link(tgen.gears["r4"])
  79
  80         # Scenario 3.
  81         switch = tgen.add_switch("s3")
  82         switch.add_link(tgen.gears["r5"])
  83         switch.add_link(tgen.gears["r6"])
  84
  85
  86 def setup_module(mod):
  87     tgen = Topogen(TemplateTopo, mod.__name__)
  88     tgen.start_topology()
  89
  90     router_list = tgen.routers()
  91
  92     for i, (rname, router) in enumerate(router_list.items(), 1):
  93         router.load_config(
  94             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  95         )
  96         router.load_config(
  97             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  98         )
  99
 100     tgen.start_router()
 101
 102
 103 def teardown_module(mod):
 104     tgen = get_topogen()
 105     tgen.stop_topology()
 106
 107
 108 def test_ebgp_requires_policy():
 109     tgen = get_topogen()
 110
 111     if tgen.routers_have_failure():
 112         pytest.skip(tgen.errors)
 113
 114     def _bgp_converge(router):
 115         output = json.loads(
 116             tgen.gears[router].vtysh_cmd("show ip bgp neighbor 192.168.255.1 json")
 117         )
 118         expected = {"192.168.255.1": {"bgpState": "Established"}}
 119         return topotest.json_cmp(output, expected)
 120
 121     def _bgp_has_routes(router):
 122         output = json.loads(
 123             tgen.gears[router].vtysh_cmd(
 124                 "show ip bgp neighbor 192.168.255.1 routes json"
 125             )
 126         )
 127         expected = {"routes": {"172.16.255.254/32": [{"valid": True}]}}
 128         return topotest.json_cmp(output, expected)
 129
 130     def _bgp_advertised_routes(router):
 131         output = json.loads(
 132             tgen.gears[router].vtysh_cmd(
 133                 "show ip bgp neighbor 192.168.255.2 advertised-routes json"
 134             )
 135         )
 136         expected = {
 137             "advertisedRoutes": {},
 138             "totalPrefixCounter": 0,
 139             "filteredPrefixCounter": 0,
 140         }
 141         return topotest.json_cmp(output, expected)
 142
 143     # Scenario 1.
 144     logger.info("Scenario 1: r2 receives 192.168.255.1/32 from r1")
 145     test_func = functools.partial(_bgp_converge, "r2")
 146     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 147     assert success is True, "Failed bgp convergence (r2)"
 148
 149     test_func = functools.partial(_bgp_has_routes, "r2")
 150     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 151     assert success is True, "r2 does not receive 192.168.255.1/32"
 152
 153     # Scenario 2.
 154     logger.info("Scenario 2: r3 must not send 192.168.255.1/32 to r4")
 155     test_func = functools.partial(_bgp_converge, "r4")
 156     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 157     assert success is True, "Failed bgp convergence (r4)"
 158
 159     test_func = functools.partial(_bgp_advertised_routes, "r3")
 160     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 161     assert success is True, "r3 announced 192.168.255.1/32 to r4"
 162
 163     # Scenario 3.
 164     logger.info("Scenario 3: r6 receives 192.168.255.1/32 from r5 (iBGP)")
 165     test_func = functools.partial(_bgp_converge, "r6")
 166     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 167     assert success is True, "Failed bgp convergence (r6)"
 168
 169     test_func = functools.partial(_bgp_has_routes, "r6")
 170     success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
 171     assert success is True, "r6 does not receive 192.168.255.1/32"
 172
 173
 174 if __name__ == "__main__":
 175     args = ["-s"] + sys.argv[1:]
 176     sys.exit(pytest.main(args))