path: root/tests/topotests/nhrp_redundancy/test_nhrp_redundancy.py

#!/usr/bin/env python
# SPDX-License-Identifier: GPL-2.0-or-later
#
# test_nhrp_redundancy.py
#
# Copyright 2024, LabN Consulting, L.L.C.
# Dave LeRoy
#

import os
import sys
import json
from time import sleep
from functools import partial
import pytest

# pylint: disable=C0413
# Import topogen and topotest helpers
from lib import topotest
from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger
from lib.common_config import (
    required_linux_kernel_version,
    shutdown_bringup_interface,
    retry,
)

"""
test_nhrp_redundancy.py: Test NHS redundancy for NHRP
"""

TOPOLOGY = """
+------------+                  +------------+                   +------------+               
|            |                  |            |                   |            |               
|            |                  |            |                   |            |               
|   NHS 1    |                  |   NHS 2    |                   |    NHS 3   |               
|            |                  |            |                   |            |               
+-----+------+                  +-----+------+                   +-----+------+               
      |.1                             |.2                              |.3                    
      |                               |                                |                      
      |                               |            192.168.1.0/24      |                      
------+-------------------------------+------------------+-------------+------                
                                                         |                                    
                                                         |.6                                  
         GRE P2MP between all NHS and NHC          +-----+------+                             
               172.16.1.x/32                       |            |                             
                                                   |            |                             
                                                   |   Router   |                             
                                                   |            |                             
                                                   +-----+------+                             
                                                         |                                    
                                                         |                                    
                               ---------+----------------+-------------+------                
                                        |          192.168.2.0/24      |                      
                                        |                              |                      
                       |                |.4                            |.5                    
+------------+         |        +-------+----+                  +------+-----+     |          
|            |         |        |            |                  |            |     |          
|            |         +--------+            |                  |            |     |          
|    Host    |.7       |        |    NHC 1   |                  |    NHC 2   +-----+5.5.5.0/24
|            +---------+        |            |                  |            |     |          
+------------+         |        +------------+                  +------------+     |          
                       |                                                           |          
                  4.4.4.0/24                                                                  
"""

# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, "../"))

# Required to instantiate the topology builder class.

pytestmark = [pytest.mark.nhrpd]


def build_topo(tgen):
    "Build function"

    # Create 7 routers
    for routern in range(1, 8):
        tgen.add_router("r{}".format(routern))

    # Interconnect routers 1, 2, 3, 6
    switch = tgen.add_switch("s1")
    switch.add_link(tgen.gears["r1"])
    switch.add_link(tgen.gears["r2"])
    switch.add_link(tgen.gears["r3"])
    switch.add_link(tgen.gears["r6"])

    # Interconnect routers 4, 5, 6
    switch = tgen.add_switch("s2")
    switch.add_link(tgen.gears["r4"])
    switch.add_link(tgen.gears["r5"])
    switch.add_link(tgen.gears["r6"])

    # Connect router 4, 7
    switch = tgen.add_switch("s3")
    switch.add_link(tgen.gears["r4"])
    switch.add_link(tgen.gears["r7"])

    # Connect router 5
    switch = tgen.add_switch("s4")
    switch.add_link(tgen.gears["r5"])


def _populate_iface():
    tgen = get_topogen()
    cmds_tot_hub = [
        "ip tunnel add {0}-gre0 mode gre ttl 64 key 42 dev {0}-eth0 local 192.168.1.{1} remote 0.0.0.0",
        "ip link set dev {0}-gre0 up",
        "echo 0 > /proc/sys/net/ipv4/ip_forward_use_pmtu",
        "echo 1 > /proc/sys/net/ipv6/conf/{0}-eth0/disable_ipv6",
        "echo 1 > /proc/sys/net/ipv6/conf/{0}-gre0/disable_ipv6",
        "iptables -A FORWARD -i {0}-gre0 -o {0}-gre0 -m hashlimit --hashlimit-upto 4/minute --hashlimit-burst 1 --hashlimit-mode srcip,dstip --hashlimit-srcmask 24 --hashlimit-dstmask 24 --hashlimit-name loglimit-0 -j NFLOG --nflog-group 1 --nflog-range 128",
    ]

    cmds_tot = [
        "ip tunnel add {0}-gre0 mode gre ttl 64 key 42 dev {0}-eth0 local 192.168.2.{1} remote 0.0.0.0",
        "ip link set dev {0}-gre0 up",
        "echo 0 > /proc/sys/net/ipv4/ip_forward_use_pmtu",
        "echo 1 > /proc/sys/net/ipv6/conf/{0}-eth0/disable_ipv6",
        "echo 1 > /proc/sys/net/ipv6/conf/{0}-gre0/disable_ipv6",
    ]

    for cmd in cmds_tot_hub:
        # Router 1
        input = cmd.format("r1", "1")
        logger.info("input: " + input)
        output = tgen.net["r1"].cmd(input)
        logger.info("output: " + output)

        # Router 2
        input = cmd.format("r2", "2")
        logger.info("input: " + input)
        output = tgen.net["r2"].cmd(input)
        logger.info("output: " + output)

        # Router 3
        input = cmd.format("r3", "3")
        logger.info("input: " + input)
        output = tgen.net["r3"].cmd(input)
        logger.info("output: " + output)

    for cmd in cmds_tot:
        input = cmd.format("r4", "4")
        logger.info("input: " + input)
        output = tgen.net["r4"].cmd(input)
        logger.info("output: " + output)

        input = cmd.format("r5", "5")
        logger.info("input: " + input)
        output = tgen.net["r5"].cmd(input)
        logger.info("output: " + output)


def _verify_iptables():
    tgen = get_topogen()
    # Verify iptables is installed. Required for shortcuts
    rc, _, _ = tgen.net["r1"].cmd_status("iptables")
    return False if rc == 127 else True


def setup_module(mod):
    logger.info("NHRP Redundant NHS:\n {}".format(TOPOLOGY))

    result = required_linux_kernel_version("5.0")
    if result is not True:
        pytest.skip("Kernel requirements are not met")

    tgen = Topogen(build_topo, mod.__name__)
    tgen.start_topology()

    # Starting Routers
    router_list = tgen.routers()
    _populate_iface()

    for rname, router in router_list.items():
        router.load_config(
            TopoRouter.RD_ZEBRA,
            os.path.join(CWD, "{}/zebra.conf".format(rname)),
        )
        if rname in ("r1", "r2", "r3", "r4", "r5"):
            router.load_config(
                TopoRouter.RD_NHRP, os.path.join(CWD, "{}/nhrpd.conf".format(rname))
            )

    # Initialize all routers.
    tgen.start_router()


def teardown_module(_mod):
    "Teardown the pytest environment"
    tgen = get_topogen()
    tgen.stop_topology()


def test_protocols_convergence():
    """
    Assert that all protocols have converged before checking for the NHRP
    statuses as they depend on it.
    """
    tgen = get_topogen()
    if tgen.routers_have_failure():
        pytest.skip(tgen.errors)

    logger.info("Checking NHRP cache and IPv4 routes for convergence")
    router_list = tgen.routers()

    # Check NHRP cache on servers and clients
    for rname, router in router_list.items():

        json_file = "{}/{}/nhrp_cache.json".format(CWD, router.name)
        if not os.path.isfile(json_file):
            logger.info("skipping file {}".format(json_file))
            continue

        expected = json.loads(open(json_file).read())
        test_func = partial(
            topotest.router_json_cmp, router, "show ip nhrp cache json", expected
        )
        _, result = topotest.run_and_expect(test_func, None, count=40, wait=0.5)

        output = router.vtysh_cmd("show ip nhrp cache")
        logger.info(output)

        assertmsg = '"{}" JSON output mismatches'.format(router.name)
        assert result is None, assertmsg

    # Check NHRP IPV4 routes on servers and clients
    for rname, router in router_list.items():

        json_file = "{}/{}/nhrp_route.json".format(CWD, router.name)
        if not os.path.isfile(json_file):
            logger.info("skipping file {}".format(json_file))
            continue

        expected = json.loads(open(json_file).read())
        test_func = partial(
            topotest.router_json_cmp, router, "show ip route nhrp json", expected
        )
        _, result = topotest.run_and_expect(test_func, None, count=40, wait=0.5)

        output = router.vtysh_cmd("show ip route nhrp")
        logger.info(output)

        assertmsg = '"{}" JSON output mismatches'.format(router.name)
        assert result is None, assertmsg

    # Test connectivity from 1 NHRP server to all clients
    pingrouter = tgen.gears["r1"]
    logger.info("Check Ping IPv4 from  R1 to R4 = 176.16.1.4)")
    output = pingrouter.run("ping 176.16.1.4 -f -c 1000")
    logger.info(output)
    if "1000 packets transmitted, 1000 received" not in output:
        assertmsg = "expected ping IPv4 from R1 to R4 should be ok"
        assert 0, assertmsg
    else:
        logger.info("Check Ping IPv4 from R1 to R4 OK")

    logger.info("Check Ping IPv4 from  R1 to R5 = 176.16.1.5)")
    output = pingrouter.run("ping 176.16.1.5 -f -c 1000")
    logger.info(output)
    if "1000 packets transmitted, 1000 received" not in output:
        assertmsg = "expected ping IPv4 from R1 to R5 should be ok"
        assert 0, assertmsg
    else:
        logger.info("Check Ping IPv4 from R1 to R5 OK")

    # Test connectivity from 1 NHRP client to all servers
    pingrouter = tgen.gears["r4"]
    logger.info("Check Ping IPv4 from  R4 to R1 = 176.16.1.1)")
    output = pingrouter.run("ping 176.16.1.1 -f -c 1000")
    logger.info(output)
    if "1000 packets transmitted, 1000 received" not in output:
        assertmsg = "expected ping IPv4 from R4 to R1 should be ok"
        assert 0, assertmsg
    else:
        logger.info("Check Ping IPv4 from R4 to R1 OK")

    logger.info("Check Ping IPv4 from  R4 to R2 = 176.16.1.2)")
    output = pingrouter.run("ping 176.16.1.2 -f -c 1000")
    logger.info(output)
    if "1000 packets transmitted, 1000 received" not in output:
        assertmsg = "expected ping IPv4 from R4 to R2 should be ok"
        assert 0, assertmsg
    else:
        logger.info("Check Ping IPv4 from R4 to R2 OK")

    logger.info("Check Ping IPv4 from  R4 to R3 = 176.16.1.3)")
    output = pingrouter.run("ping 176.16.1.3 -f -c 1000")
    logger.info(output)
    if "1000 packets transmitted, 1000 received" not in output:
        assertmsg = "expected ping IPv4 from R4 to R3 should be ok"
        assert 0, assertmsg
    else:
        logger.info("Check Ping IPv4 from R4 to R3 OK")


@retry(retry_timeout=30, initial_wait=5)
def verify_shortcut_path():
    """
    Verifying that traffic flows through shortcut path
    """
    tgen = get_topogen()
    pingrouter = tgen.gears["r7"]
    logger.info("Check Ping IPv4 from  R7 to R5 = 5.5.5.5")

    output = pingrouter.run("ping 5.5.5.5 -f -c 1000")
    logger.info(output)
    if "1000 packets transmitted, 1000 received" not in output:
        assertmsg = "expected ping IPv4 from R7 to R5 should be ok"
        assert 0, assertmsg
    else:
        logger.info("Check Ping IPv4 from R7 to R5 OK")


def test_redundancy_shortcut():
    """
    Assert that if shortcut created and then NHS goes down, there is no traffic disruption
    Stop traffic and verify next time traffic started, shortcut is initiated by backup NHS
    """
    tgen = get_topogen()
    if tgen.routers_have_failure():
        pytest.skip(tgen.errors)

    if not _verify_iptables():
        pytest.skip("iptables not installed")

    logger.info("Testing NHRP shortcuts with redundant servers")

    # Verify R4 nhrp routes before shortcut creation
    router = tgen.gears["r4"]
    json_file = "{}/{}/nhrp_route.json".format(CWD, router.name)
    assertmsg = "No nhrp_route file found"
    assert os.path.isfile(json_file), assertmsg

    expected = json.loads(open(json_file).read())
    test_func = partial(
        topotest.router_json_cmp, router, "show ip route nhrp json", expected
    )
    _, result = topotest.run_and_expect(test_func, None, count=40, wait=0.5)

    output = router.vtysh_cmd("show ip route nhrp")
    logger.info(output)

    assertmsg = '"{}" JSON output mismatches'.format(router.name)
    assert result is None, assertmsg

    # Initiate shortcut by pinging between clients
    pingrouter = tgen.gears["r7"]
    logger.info("Check Ping IPv4 from  R7 to R5 via shortcut = 5.5.5.5")

    output = pingrouter.run("ping 5.5.5.5 -f -c 1000")
    logger.info(output)
    if "1000 packets transmitted, 1000 received" not in output:
        assertmsg = "expected ping IPv4 from R7 to R5 via shortcut should be ok"
        assert 0, assertmsg
    else:
        logger.info("Check Ping IPv4 from R7 to R5 via shortcut OK")

    # Now check that NHRP shortcut route installed
    json_file = "{}/{}/nhrp_route_shortcut.json".format(CWD, router.name)
    assertmsg = "No nhrp_route file found"
    assert os.path.isfile(json_file), assertmsg

    expected = json.loads(open(json_file).read())
    test_func = partial(
        topotest.router_json_cmp, router, "show ip route nhrp json", expected
    )
    _, result = topotest.run_and_expect(test_func, None, count=40, wait=0.5)

    output = router.vtysh_cmd("show ip route nhrp")
    logger.info(output)

    assertmsg = '"{}" JSON output mismatches'.format(router.name)
    assert result is None, assertmsg

    # Bring down primary GRE interface and verify shortcut is not disturbed
    logger.info("Bringing down R1, primary NHRP server.")
    shutdown_bringup_interface(tgen, "r1", "r1-gre0", False)

    # Verify shortcut is still active
    pingrouter = tgen.gears["r7"]
    logger.info("Check Ping IPv4 from  R7 to R5 via shortcut = 5.5.5.5")

    output = pingrouter.run("ping 5.5.5.5 -f -c 1000")
    logger.info(output)
    if "1000 packets transmitted, 1000 received" not in output:
        assertmsg = "expected ping IPv4 from R7 to R5 via shortcut should be ok"
        assert 0, assertmsg
    else:
        logger.info("Check Ping IPv4 from R7 to R5 via shortcut OK")

    # Now verify shortcut is purged with lack of traffic
    json_file = "{}/{}/nhrp_route.json".format(CWD, router.name)
    assertmsg = "No nhrp_route file found"
    assert os.path.isfile(json_file), assertmsg

    expected = json.loads(open(json_file).read())
    test_func = partial(
        topotest.router_json_cmp, router, "show ip route nhrp json", expected
    )
    _, result = topotest.run_and_expect(test_func, None, count=40, wait=0.5)

    output = router.vtysh_cmd("show ip route nhrp")
    logger.info(output)

    assertmsg = '"{}" JSON output mismatches'.format(router.name)
    assert result is None, assertmsg


def test_memory_leak():
    "Run the memory leak test and report results."
    tgen = get_topogen()
    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))