topotests: add bgp_rpki_topo1

Add bgp_rpki_topo1 topotest to validate the RPKI feature. Use a RTR RPKI
server from the above link with a black cleaning.

Link: https://raw.githubusercontent.com/tmshlvck/pyrtr/90df586375396aae08b07069187308b5b7b8823b/pyrtr/__init__.py
Signed-off-by: Louis Scalbert <louis.scalbert@6wind.com>

tests/topotests/bgp_rpki_topo1/r1/bgpd.conf

@@ -0,0 +1,14 @@
+router bgp 65530
+ no bgp ebgp-requires-policy
+ no bgp network import-check
+ neighbor 192.0.2.2 remote-as 65002
+ neighbor 192.0.2.2 timers 1 3
+ neighbor 192.0.2.2 timers connect 1
+ neighbor 192.0.2.2 ebgp-multihop 3
+ neighbor 192.0.2.2 update-source 192.0.2.1
+ address-family ipv4 unicast
+  network 198.51.100.0/24
+  network 203.0.113.0/24
+  network 10.0.0.0/24
+ exit-address-family
+!

tests/topotests/bgp_rpki_topo1/r1/rtrd.py

@@ -0,0 +1,319 @@
+#!/usr/bin/python3
+# SPDX-License-Identifier: GPL-2.0-or-later
+
+# Copyright (C) 2023 Tomas Hlavacek (tmshlvck@gmail.com)
+
+from typing import List, Tuple, Callable, Type
+import socket
+import threading
+import socketserver
+import struct
+import ipaddress
+import csv
+import os
+import sys
+
+LISTEN_HOST, LISTEN_PORT = "0.0.0.0", 15432
+VRPS_FILE = os.path.join(sys.path[0], "vrps.csv")
+
+
+def dbg(m: str):
+    print(m)
+
+
+class RTRDatabase(object):
+    def __init__(self, vrps_file: str) -> None:
+        self.last_serial = 0
+        self.ann4 = []
+        self.ann6 = []
+        self.withdraw4 = []
+        self.withdraw6 = []
+
+        with open(vrps_file, "r") as fh:
+            for rasn, rnet, rmaxlen, _ in csv.reader(fh):
+                try:
+                    net = ipaddress.ip_network(rnet)
+                    asn = int(rasn[2:])
+                    maxlen = int(rmaxlen)
+                    if net.version == 6:
+                        self.ann6.append((asn, str(net), maxlen))
+                    elif net.version == 4:
+                        self.ann4.append((asn, str(net), maxlen))
+                    else:
+                        raise ValueError(f"Unknown AFI: {net.version}")
+                except Exception as e:
+                    dbg(
+                        f"VRPS load: ignoring {str((rasn, rnet,rmaxlen))} because {str(e)}"
+                    )
+
+    def get_serial(self) -> int:
+        return self.last_serial
+
+    def set_serial(self, serial: int) -> None:
+        self.last_serial = serial
+
+    def get_announcements4(self, serial: int = 0) -> List[Tuple[int, str, int]]:
+        if serial > self.last_serial:
+            return self.ann4
+        else:
+            return []
+
+    def get_withdrawals4(self, serial: int = 0) -> List[Tuple[int, str, int]]:
+        if serial > self.last_serial:
+            return self.withdraw4
+        else:
+            return []
+
+    def get_announcements6(self, serial: int = 0) -> List[Tuple[int, str, int]]:
+        if serial > self.last_serial:
+            return self.ann6
+        else:
+            return []
+
+    def get_withdrawals6(self, serial: int = 0) -> List[Tuple[int, str, int]]:
+        if serial > self.last_serial:
+            return self.withdraw6
+        else:
+            return []
+
+
+class RTRConnHandler(socketserver.BaseRequestHandler):
+    PROTO_VERSION = 0
+
+    def setup(self) -> None:
+        self.session_id = 2345
+        self.serial = 1024
+
+        dbg(f"New connection from: {str(self.client_address)} ")
+        # TODO: register for notifies
+
+    def finish(self) -> None:
+        pass
+        # TODO: de-register
+
+    HEADER_LEN = 8
+
+    def decode_header(self, buf: bytes) -> Tuple[int, int, int, int]:
+        # common header in all received packets
+        return struct.unpack("!BBHI", buf)
+        # reutnrs (proto_ver, pdu_type, sess_id, length)
+
+    SERNOTIFY_TYPE = 0
+    SERNOTIFY_LEN = 12
+
+    def send_sernotify(self, serial: int) -> None:
+        # serial notify PDU
+        dbg(f"<Serial Notify session_id={self.session_id} serial={serial}")
+        self.request.send(
+            struct.pack(
+                "!BBHII",
+                self.PROTO_VERSION,
+                self.SERNOTIFY_TYPE,
+                self.session_id,
+                self.SERNOTIFY_LEN,
+                serial,
+            )
+        )
+
+    CACHERESPONSE_TYPE = 3
+    CACHERESPONSE_LEN = 8
+
+    def send_cacheresponse(self) -> None:
+        # cache response PDU
+        dbg(f"<Cache response session_id={self.session_id}")
+        self.request.send(
+            struct.pack(
+                "!BBHI",
+                self.PROTO_VERSION,
+                self.CACHERESPONSE_TYPE,
+                self.session_id,
+                self.CACHERESPONSE_LEN,
+            )
+        )
+
+    FLAGS_ANNOUNCE = 1
+    FLAGS_WITHDRAW = 0
+
+    IPV4_TYPE = 4
+    IPV4_LEN = 20
+
+    def send_ipv4(self, ipnet: str, asn: int, maxlen: int, flags: int):
+        # IPv4 PDU
+        dbg(f"<IPv4 net={ipnet} asn={asn} maxlen={maxlen} flags={flags}")
+        ip = ipaddress.IPv4Network(ipnet)
+        self.request.send(
+            struct.pack(
+                "!BBHIBBBB4sI",
+                self.PROTO_VERSION,
+                self.IPV4_TYPE,
+                0,
+                self.IPV4_LEN,
+                flags,
+                ip.prefixlen,
+                maxlen,
+                0,
+                ip.network_address.packed,
+                asn,
+            )
+        )
+
+    def announce_ipv4(self, ipnet, asn, maxlen):
+        self.send_ipv4(ipnet, asn, maxlen, self.FLAGS_ANNOUNCE)
+
+    def withdraw_ipv4(self, ipnet, asn, maxlen):
+        self.send_ipv4(ipnet, asn, maxlen, self.FLAGS_WITHDRAW)
+
+    IPV6_TYPE = 6
+    IPV6_LEN = 32
+
+    def send_ipv6(self, ipnet: str, asn: int, maxlen: int, flags: int):
+        # IPv6 PDU
+        dbg(f"<IPv6 net={ipnet} asn={asn} maxlen={maxlen} flags={flags}")
+        ip = ipaddress.IPv6Network(ipnet)
+        self.request.send(
+            struct.pack(
+                "!BBHIBBBB16sI",
+                self.PROTO_VERSION,
+                self.IPV6_TYPE,
+                0,
+                self.IPV6_LEN,
+                flags,
+                ip.prefixlen,
+                maxlen,
+                0,
+                ip.network_address.packed,
+                asn,
+            )
+        )
+
+    def announce_ipv6(self, ipnet: str, asn: int, maxlen: int):
+        self.send_ipv6(ipnet, asn, maxlen, self.FLAGS_ANNOUNCE)
+
+    def withdraw_ipv6(self, ipnet: str, asn: int, maxlen: int):
+        self.send_ipv6(ipnet, asn, maxlen, self.FLAGS_WITHDRAW)
+
+    EOD_TYPE = 7
+    EOD_LEN = 12
+
+    def send_endofdata(self, serial: int):
+        # end of data PDU
+        dbg(f"<End of Data session_id={self.session_id} serial={serial}")
+        self.server.db.set_serial(serial)
+        self.request.send(
+            struct.pack(
+                "!BBHII",
+                self.PROTO_VERSION,
+                self.EOD_TYPE,
+                self.session_id,
+                self.EOD_LEN,
+                serial,
+            )
+        )
+
+    CACHERESET_TYPE = 8
+    CACHERESET_LEN = 8
+
+    def send_cachereset(self):
+        # cache reset PDU
+        dbg("<Cache Reset")
+        self.request.send(
+            struct.pack(
+                "!BBHI",
+                self.PROTO_VERSION,
+                self.CACHERESET_TYPE,
+                0,
+                self.CACHERESET_LEN,
+            )
+        )
+
+    SERIAL_QUERY_TYPE = 1
+    SERIAL_QUERY_LEN = 12
+
+    def handle_serial_query(self, buf: bytes, sess_id: int):
+        serial = struct.unpack("!I", buf)[0]
+        dbg(f">Serial query: {serial}")
+        if sess_id:
+            self.server.db.set_serial(serial)
+        else:
+            self.server.db.set_serial(0)
+        self.send_cacheresponse()
+
+        for asn, ipnet, maxlen in self.server.db.get_announcements4(serial):
+            self.announce_ipv4(ipnet, asn, maxlen)
+
+        for asn, ipnet, maxlen in self.server.db.get_withdrawals4(serial):
+            self.withdraw_ipv4(ipnet, asn, maxlen)
+
+        for asn, ipnet, maxlen in self.server.db.get_announcements6(serial):
+            self.announce_ipv6(ipnet, asn, maxlen)
+
+        for asn, ipnet, maxlen in self.server.db.get_withdrawals6(serial):
+            self.withdraw_ipv6(ipnet, asn, maxlen)
+
+        self.send_endofdata(self.serial)
+
+    RESET_TYPE = 2
+
+    def handle_reset(self):
+        dbg(">Reset")
+        self.session_id += 1
+        self.server.db.set_serial(0)
+        self.send_cacheresponse()
+
+        for asn, ipnet, maxlen in self.server.db.get_announcements4(self.serial):
+            self.announce_ipv4(ipnet, asn, maxlen)
+
+        for asn, ipnet, maxlen in self.server.db.get_announcements6(self.serial):
+            self.announce_ipv6(ipnet, asn, maxlen)
+
+        self.send_endofdata(self.serial)
+
+    ERROR_TYPE = 10
+
+    def handle_error(self, buf: bytes):
+        dbg(f">Error: {str(buf)}")
+
+    def handle(self):
+        while True:
+            b = self.request.recv(self.HEADER_LEN, socket.MSG_WAITALL)
+            if len(b) == 0:
+                break
+            proto_ver, pdu_type, sess_id, length = self.decode_header(b)
+            dbg(
+                f">Header proto_ver={proto_ver} pdu_type={pdu_type} sess_id={sess_id} length={length}"
+            )
+
+            if sess_id:
+                self.session_id = sess_id
+
+            if pdu_type == self.SERIAL_QUERY_TYPE:
+                b = self.request.recv(
+                    self.SERIAL_QUERY_LEN - self.HEADER_LEN, socket.MSG_WAITALL
+                )
+                self.handle_serial_query(b, sess_id)
+
+            elif pdu_type == self.RESET_TYPE:
+                self.handle_reset()
+
+            elif pdu_type == self.ERROR_TYPE:
+                b = self.request.recv(length - self.HEADER_LEN, socket.MSG_WAITALL)
+                self.handle_error(b)
+
+
+class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
+    def __init__(
+        self, bind: Tuple[str, int], handler: Type[RTRConnHandler], db: RTRDatabase
+    ) -> None:
+        super().__init__(bind, handler)
+        self.db = db
+
+
+def main():
+    db = RTRDatabase(VRPS_FILE)
+    server = ThreadedTCPServer((LISTEN_HOST, LISTEN_PORT), RTRConnHandler, db)
+    dbg(f"Server listening on {LISTEN_HOST} port {LISTEN_PORT}")
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()

tests/topotests/bgp_rpki_topo1/r1/staticd.conf

@@ -0,0 +1 @@
+ip route 192.0.2.2/32 192.168.1.2

tests/topotests/bgp_rpki_topo1/r1/vrps.csv

@@ -0,0 +1,3 @@
+ASN,IP Prefix,Max Length,Trust Anchor
+AS65530,198.51.100.0/24,24,private
+AS65530,203.0.113.0/24,24,private

tests/topotests/bgp_rpki_topo1/r1/zebra.conf

@@ -0,0 +1,6 @@
+interface lo
+ ip address 192.0.2.1/32
+!
+interface r1-eth0
+ ip address 192.168.1.1/24
+!

tests/topotests/bgp_rpki_topo1/r2/bgp_table_rmap_rpki_any.json

@@ -0,0 +1,37 @@
+{
+  "routerId": "192.0.2.2",
+  "defaultLocPrf": 100,
+  "localAS": 65002,
+  "routes": {
+    "198.51.100.0/24": [
+      {
+        "valid": true,
+        "bestpath": true,
+        "selectionReason": "First path received",
+        "pathFrom": "external",
+        "prefix": "198.51.100.0",
+        "prefixLen": 24,
+        "network": "198.51.100.0/24",
+        "metric": 0,
+        "weight": 0,
+        "path": "65530",
+        "origin": "IGP"
+      }
+    ],
+    "203.0.113.0/24": [
+      {
+        "valid": true,
+        "bestpath": true,
+        "selectionReason": "First path received",
+        "pathFrom": "external",
+        "prefix": "203.0.113.0",
+        "prefixLen": 24,
+        "network": "203.0.113.0/24",
+        "metric": 0,
+        "weight": 0,
+        "path": "65530",
+        "origin": "IGP"
+      }
+    ]
+  }
+}

tests/topotests/bgp_rpki_topo1/r2/bgp_table_rmap_rpki_notfound.json

@@ -0,0 +1,7 @@
+{
+  "routerId": "192.0.2.2",
+  "defaultLocPrf": 100,
+  "localAS": 65002,
+  "routes": {
+  }
+}

tests/topotests/bgp_rpki_topo1/r2/bgp_table_rmap_rpki_valid.json

@@ -0,0 +1 @@
+bgp_table_rpki_valid.json

tests/topotests/bgp_rpki_topo1/r2/bgp_table_rpki_any.json

@@ -0,0 +1,52 @@
+{
+  "routerId": "192.0.2.2",
+  "defaultLocPrf": 100,
+  "localAS": 65002,
+  "routes": {
+    "10.0.0.0/24": [
+      {
+        "valid": true,
+        "bestpath": true,
+        "selectionReason": "First path received",
+        "pathFrom": "external",
+        "prefix": "10.0.0.0",
+        "prefixLen": 24,
+        "network": "10.0.0.0/24",
+        "metric": 0,
+        "weight": 0,
+        "path": "65530",
+        "origin": "IGP"
+      }
+    ],
+    "198.51.100.0/24": [
+      {
+        "valid": true,
+        "bestpath": true,
+        "selectionReason": "First path received",
+        "pathFrom": "external",
+        "prefix": "198.51.100.0",
+        "prefixLen": 24,
+        "network": "198.51.100.0/24",
+        "metric": 0,
+        "weight": 0,
+        "path": "65530",
+        "origin": "IGP"
+      }
+    ],
+    "203.0.113.0/24": [
+      {
+        "valid": true,
+        "bestpath": true,
+        "selectionReason": "First path received",
+        "pathFrom": "external",
+        "prefix": "203.0.113.0",
+        "prefixLen": 24,
+        "network": "203.0.113.0/24",
+        "metric": 0,
+        "weight": 0,
+        "path": "65530",
+        "origin": "IGP"
+      }
+    ]
+  }
+}

tests/topotests/bgp_rpki_topo1/r2/bgp_table_rpki_notfound.json

@@ -0,0 +1,22 @@
+{
+  "routerId": "192.0.2.2",
+  "defaultLocPrf": 100,
+  "localAS": 65002,
+  "routes": {
+    "10.0.0.0/24": [
+      {
+        "valid": true,
+        "bestpath": true,
+        "selectionReason": "First path received",
+        "pathFrom": "external",
+        "prefix": "10.0.0.0",
+        "prefixLen": 24,
+        "network": "10.0.0.0/24",
+        "metric": 0,
+        "weight": 0,
+        "path": "65530",
+        "origin": "IGP"
+      }
+    ]
+  }
+}

tests/topotests/bgp_rpki_topo1/r2/bgp_table_rpki_valid.json

@@ -0,0 +1,35 @@
+{
+  "routerId": "192.0.2.2",
+  "defaultLocPrf": 100,
+  "localAS": 65002,
+  "routes": {
+    "198.51.100.0/24": [
+      {
+        "valid": true,
+        "bestpath": true,
+        "pathFrom": "external",
+        "prefix": "198.51.100.0",
+        "prefixLen": 24,
+        "network": "198.51.100.0/24",
+        "metric": 0,
+        "weight": 0,
+        "path": "65530",
+        "origin": "IGP"
+      }
+    ],
+    "203.0.113.0/24": [
+      {
+        "valid": true,
+        "bestpath": true,
+        "pathFrom": "external",
+        "prefix": "203.0.113.0",
+        "prefixLen": 24,
+        "network": "203.0.113.0/24",
+        "metric": 0,
+        "weight": 0,
+        "path": "65530",
+        "origin": "IGP"
+      }
+    ]
+  }
+}

tests/topotests/bgp_rpki_topo1/r2/bgpd.conf

@@ -0,0 +1,19 @@
+router bgp 65002
+ no bgp ebgp-requires-policy
+ neighbor 192.0.2.1 remote-as 65530
+ neighbor 192.0.2.1 timers connect 1
+ neighbor 192.0.2.1 ebgp-multihop 3
+ neighbor 192.0.2.1 update-source 192.0.2.2
+!
+router bgp 65002 vrf vrf10
+ no bgp ebgp-requires-policy
+ neighbor 192.0.2.3 remote-as 65530
+ neighbor 192.0.2.3 timers 1 3
+ neighbor 192.0.2.3 timers connect 1
+ neighbor 192.0.2.3 ebgp-multihop 3
+ neighbor 192.0.2.3 update-source 192.0.2.2
+!
+rpki
+ rpki retry_interval 5
+ rpki cache 192.0.2.1 15432 preference 1
+exit

tests/topotests/bgp_rpki_topo1/r2/rpki_prefix_table.json

@@ -0,0 +1,18 @@
+{
+  "prefixes":[
+    {
+      "prefix":"198.51.100.0",
+      "prefixLenMin":24,
+      "prefixLenMax":24,
+      "asn":65530
+    },
+    {
+      "prefix":"203.0.113.0",
+      "prefixLenMin":24,
+      "prefixLenMax":24,
+      "asn":65530
+    }
+  ],
+  "ipv4PrefixCount":2,
+  "ipv6PrefixCount":0
+}

tests/topotests/bgp_rpki_topo1/r2/staticd.conf

@@ -0,0 +1 @@
+ip route 192.0.2.1/32 192.168.1.1

tests/topotests/bgp_rpki_topo1/r2/zebra.conf

@@ -0,0 +1,9 @@
+interface lo
+ ip address 192.0.2.2/32
+!
+interface vrf10 vrf vrf10
+ ip address 192.0.2.2/32
+!
+interface r2-eth0
+ ip address 192.168.1.2/24
+!

tests/topotests/bgp_rpki_topo1/test_bgp_rpki_topo1.py

@@ -0,0 +1,264 @@
+#!/usr/bin/env python
+# SPDX-License-Identifier: ISC
+
+# Copyright 2023 6WIND S.A.
+
+import os
+import sys
+import json
+import pytest
+import functools
+
+CWD = os.path.dirname(os.path.realpath(__file__))
+sys.path.append(os.path.join(CWD, "../"))
+
+# pylint: disable=C0413
+from lib import topotest
+from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.common_config import step
+from lib.topolog import logger
+
+pytestmark = [pytest.mark.bgpd]
+
+
+def build_topo(tgen):
+    for routern in range(1, 3):
+        tgen.add_router("r{}".format(routern))
+
+    switch = tgen.add_switch("s1")
+    switch.add_link(tgen.gears["r1"])
+    switch.add_link(tgen.gears["r2"])
+
+
+def setup_module(mod):
+    tgen = Topogen(build_topo, mod.__name__)
+    tgen.start_topology()
+
+    router_list = tgen.routers()
+
+    for i, (rname, router) in enumerate(router_list.items(), 1):
+        router.load_config(
+            TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
+        )
+        router.load_config(
+            TopoRouter.RD_STATIC, os.path.join(CWD, "{}/staticd.conf".format(rname))
+        )
+        router.load_config(
+            TopoRouter.RD_BGP,
+            os.path.join(CWD, "{}/bgpd.conf".format(rname)),
+            " -M bgpd_rpki" if rname == "r2" else "",
+        )
+
+    tgen.start_router()
+
+    r1_path = os.path.join(CWD, "r1")
+
+    global rtrd_process
+
+    tgen.gears["r1"].cmd("chmod u+x {}/rtrd.py".format(r1_path))
+    rtrd_process = tgen.gears["r1"].popen("python3 {}/rtrd.py".format(r1_path))
+
+
+def teardown_module(mod):
+    tgen = get_topogen()
+
+    logger.info("r1: sending SIGTERM to rtrd RPKI server")
+    rtrd_process.kill()
+    tgen.stop_topology()
+
+
+def show_rpki_prefixes(rname, expected, vrf=None):
+    tgen = get_topogen()
+
+    if vrf:
+        cmd = "show rpki prefix-table vrf {} json".format(vrf)
+    else:
+        cmd = "show rpki prefix-table json"
+
+    output = json.loads(tgen.gears[rname].vtysh_cmd(cmd))
+
+    return topotest.json_cmp(output, expected)
+
+
+def show_bgp_ipv4_table_rpki(rname, rpki_state, expected, vrf=None):
+    tgen = get_topogen()
+
+    cmd = "show bgp"
+    if vrf:
+        cmd += " vrf {}".format(vrf)
+    cmd += " ipv4 unicast"
+    if rpki_state:
+        cmd += " rpki {}".format(rpki_state)
+    cmd += " json"
+
+    output = json.loads(tgen.gears[rname].vtysh_cmd(cmd))
+
+    expected_nb = len(expected.get("routes"))
+    output_nb = len(output.get("routes", {}))
+
+    if expected_nb != output_nb:
+        return {"error": "expected {} prefixes. Got {}".format(expected_nb, output_nb)}
+
+    return topotest.json_cmp(output, expected)
+
+
+def test_show_bgp_rpki_prefixes():
+    tgen = get_topogen()
+
+    if tgen.routers_have_failure():
+        pytest.skip(tgen.errors)
+
+    rname = "r2"
+
+    step("Check RPKI prefix table")
+
+    expected = open(os.path.join(CWD, "{}/rpki_prefix_table.json".format(rname))).read()
+    expected_json = json.loads(expected)
+    test_func = functools.partial(show_rpki_prefixes, rname, expected_json)
+    _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
+    assert result is None, "Failed to see RPKI prefixes on {}".format(rname)
+
+    for rpki_state in ["valid", "notfound", None]:
+        if rpki_state:
+            step("Check RPKI state of prefixes in BGP table: {}".format(rpki_state))
+        else:
+            step("Check prefixes in BGP table")
+        expected = open(
+            os.path.join(
+                CWD,
+                "{}/bgp_table_rpki_{}.json".format(
+                    rname, rpki_state if rpki_state else "any"
+                ),
+            )
+        ).read()
+        expected_json = json.loads(expected)
+        test_func = functools.partial(
+            show_bgp_ipv4_table_rpki, rname, rpki_state, expected_json
+        )
+        _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
+        assert result is None, "Unexpected prefixes RPKI state on {}".format(rname)
+
+
+def test_show_bgp_rpki_prefixes_no_rpki_cache():
+    tgen = get_topogen()
+
+    if tgen.routers_have_failure():
+        pytest.skip(tgen.errors)
+
+    def _show_rpki_no_connection(rname):
+        output = json.loads(
+            tgen.gears[rname].vtysh_cmd("show rpki cache-connection json")
+        )
+
+        return output == {"error": "No connection to RPKI cache server."}
+
+    step("Remove RPKI server from configuration")
+    rname = "r2"
+    tgen.gears[rname].vtysh_cmd(
+        """
+configure
+rpki
+ no rpki cache 192.0.2.1 15432 preference 1
+exit
+"""
+    )
+
+    step("Check RPKI connection state")
+
+    test_func = functools.partial(_show_rpki_no_connection, rname)
+    _, result = topotest.run_and_expect(test_func, True, count=60, wait=0.5)
+    assert result, "RPKI is still connected on {}".format(rname)
+
+
+def test_show_bgp_rpki_prefixes_reconnect():
+    tgen = get_topogen()
+
+    if tgen.routers_have_failure():
+        pytest.skip(tgen.errors)
+
+    step("Restore RPKI server configuration")
+
+    rname = "r2"
+    tgen.gears[rname].vtysh_cmd(
+        """
+configure
+rpki
+ rpki cache 192.0.2.1 15432 preference 1
+exit
+"""
+    )
+
+    step("Check RPKI prefix table")
+
+    expected = open(os.path.join(CWD, "{}/rpki_prefix_table.json".format(rname))).read()
+    expected_json = json.loads(expected)
+    test_func = functools.partial(show_rpki_prefixes, rname, expected_json)
+    _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
+    assert result is None, "Failed to see RPKI prefixes on {}".format(rname)
+
+    for rpki_state in ["valid", "notfound", None]:
+        if rpki_state:
+            step("Check RPKI state of prefixes in BGP table: {}".format(rpki_state))
+        else:
+            step("Check prefixes in BGP table")
+        expected = open(
+            os.path.join(
+                CWD,
+                "{}/bgp_table_rpki_{}.json".format(
+                    rname, rpki_state if rpki_state else "any"
+                ),
+            )
+        ).read()
+        expected_json = json.loads(expected)
+        _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
+        assert result is None, "Unexpected prefixes RPKI state on {}".format(rname)
+
+
+def test_show_bgp_rpki_route_map():
+    tgen = get_topogen()
+
+    if tgen.routers_have_failure():
+        pytest.skip(tgen.errors)
+
+    step("Apply RPKI valid route-map on neighbor")
+
+    rname = "r2"
+    tgen.gears[rname].vtysh_cmd(
+        """
+configure
+route-map RPKI permit 10
+ match rpki valid
+!
+router bgp 65002
+ address-family ipv4 unicast
+  neighbor 192.0.2.1 route-map RPKI in
+"""
+    )
+
+    for rpki_state in ["valid", "notfound", None]:
+        if rpki_state:
+            step("Check RPKI state of prefixes in BGP table: {}".format(rpki_state))
+        else:
+            step("Check prefixes in BGP table")
+        expected = open(
+            os.path.join(
+                CWD,
+                "{}/bgp_table_rmap_rpki_{}.json".format(
+                    rname, rpki_state if rpki_state else "any"
+                ),
+            )
+        ).read()
+        expected_json = json.loads(expected)
+        test_func = functools.partial(
+            show_bgp_ipv4_table_rpki,
+            rname,
+            rpki_state,
+            expected_json,
+        )
+        _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
+        assert result is None, "Unexpected prefixes RPKI state on {}".format(rname)
+
+
+if __name__ == "__main__":
+    args = ["-s"] + sys.argv[1:]
+    sys.exit(pytest.main(args))