test/test_pnat.py

   1 #!/usr/bin/env python3
   2 """Policy 1:1 NAT functional tests"""
   3
   4 import unittest
   5 from scapy.layers.inet import Ether, IP, UDP, ICMP
   6 from framework import VppTestCase, VppTestRunner
   7 from vpp_papi import VppEnum
   8
   9
  10 class TestPNAT(VppTestCase):
  11     """PNAT Test Case"""
  12
  13     maxDiff = None
  14
  15     @classmethod
  16     def setUpClass(cls):
  17         super(TestPNAT, cls).setUpClass()
  18         cls.create_pg_interfaces(range(2))
  19         cls.interfaces = list(cls.pg_interfaces)
  20
  21     @classmethod
  22     def tearDownClass(cls):
  23         super(TestPNAT, cls).tearDownClass()
  24
  25     def setUp(self):
  26         super(TestPNAT, self).setUp()
  27         for i in self.interfaces:
  28             i.admin_up()
  29             i.config_ip4()
  30             i.resolve_arp()
  31
  32     def tearDown(self):
  33         super(TestPNAT, self).tearDown()
  34         if not self.vpp_dead:
  35             for i in self.pg_interfaces:
  36                 i.unconfig_ip4()
  37                 i.admin_down()
  38
  39     def validate(self, rx, expected):
  40         self.assertEqual(rx, expected.__class__(expected))
  41
  42     def validate_bytes(self, rx, expected):
  43         self.assertEqual(rx, expected)
  44
  45     def ping_check(self):
  46         """Verify non matching traffic works."""
  47         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
  48
  49         icmpecho = IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) / ICMP()
  50         reply = IP(src=self.pg0.local_ip4, dst=self.pg0.remote_ip4) / ICMP(
  51             type="echo-reply"
  52         )
  53         rx = self.send_and_expect(self.pg0, p_ether / icmpecho * 1, self.pg0)
  54         for p in rx:
  55             reply[IP].id = p[IP].id
  56             self.validate(p[1], reply)
  57
  58     def test_pnat(self):
  59         """PNAT test"""
  60
  61         PNAT_IP4_INPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_INPUT
  62         PNAT_IP4_OUTPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_OUTPUT
  63
  64         tests = [
  65             {
  66                 "input": PNAT_IP4_INPUT,
  67                 "sw_if_index": self.pg0.sw_if_index,
  68                 "match": {
  69                     "mask": 0xA,
  70                     "dst": "10.10.10.10",
  71                     "proto": 17,
  72                     "dport": 6871,
  73                 },
  74                 "rewrite": {"mask": 0x2, "dst": self.pg1.remote_ip4},
  75                 "send": (
  76                     IP(src=self.pg0.remote_ip4, dst="10.10.10.10") / UDP(dport=6871)
  77                 ),
  78                 "reply": (
  79                     IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
  80                     / UDP(dport=6871)
  81                 ),
  82             },
  83             {
  84                 "input": PNAT_IP4_OUTPUT,
  85                 "sw_if_index": self.pg1.sw_if_index,
  86                 "match": {
  87                     "mask": 0x9,
  88                     "src": self.pg0.remote_ip4,
  89                     "proto": 17,
  90                     "dport": 6871,
  91                 },
  92                 "rewrite": {"mask": 0x1, "src": "11.11.11.11"},
  93                 "send": (
  94                     IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
  95                     / UDP(dport=6871)
  96                 ),
  97                 "reply": (
  98                     IP(src="11.11.11.11", dst=self.pg1.remote_ip4) / UDP(dport=6871)
  99                 ),
 100             },
 101             {
 102                 "input": PNAT_IP4_INPUT,
 103                 "sw_if_index": self.pg0.sw_if_index,
 104                 "match": {
 105                     "mask": 0xA,
 106                     "dst": "10.10.10.10",
 107                     "proto": 17,
 108                     "dport": 6871,
 109                 },
 110                 "rewrite": {"mask": 0xA, "dst": self.pg1.remote_ip4, "dport": 5555},
 111                 "send": (
 112                     IP(src=self.pg0.remote_ip4, dst="10.10.10.10")
 113                     / UDP(sport=65530, dport=6871)
 114                 ),
 115                 "reply": (
 116                     IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
 117                     / UDP(sport=65530, dport=5555)
 118                 ),
 119             },
 120             {
 121                 "input": PNAT_IP4_INPUT,
 122                 "sw_if_index": self.pg0.sw_if_index,
 123                 "match": {
 124                     "mask": 0xA,
 125                     "dst": self.pg1.remote_ip4,
 126                     "proto": 17,
 127                     "dport": 6871,
 128                 },
 129                 "rewrite": {"mask": 0x8, "dport": 5555},
 130                 "send": (
 131                     IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
 132                     / UDP(dport=6871, chksum=0)
 133                 ),
 134                 "reply": (
 135                     IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
 136                     / UDP(dport=5555, chksum=0)
 137                 ),
 138             },
 139             {
 140                 "input": PNAT_IP4_INPUT,
 141                 "sw_if_index": self.pg0.sw_if_index,
 142                 "match": {"mask": 0x2, "dst": self.pg1.remote_ip4, "proto": 1},
 143                 "rewrite": {"mask": 0x1, "src": "8.8.8.8"},
 144                 "send": (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / ICMP()),
 145                 "reply": IP(src="8.8.8.8", dst=self.pg1.remote_ip4) / ICMP(),
 146             },
 147         ]
 148
 149         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 150         for t in tests:
 151             rv = self.vapi.pnat_binding_add(match=t["match"], rewrite=t["rewrite"])
 152             self.vapi.pnat_binding_attach(
 153                 sw_if_index=t["sw_if_index"],
 154                 attachment=t["input"],
 155                 binding_index=rv.binding_index,
 156             )
 157
 158             reply = t["reply"]
 159             reply[IP].ttl -= 1
 160             rx = self.send_and_expect(self.pg0, p_ether / t["send"] * 1, self.pg1)
 161             for p in rx:
 162                 # p.show2()
 163                 self.validate(p[1], reply)
 164
 165             self.ping_check()
 166
 167             self.vapi.pnat_binding_detach(
 168                 sw_if_index=t["sw_if_index"],
 169                 attachment=t["input"],
 170                 binding_index=rv.binding_index,
 171             )
 172             self.vapi.pnat_binding_del(binding_index=rv.binding_index)
 173
 174     def test_pnat_show(self):
 175         """PNAT show tests"""
 176
 177         PNAT_IP4_INPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_INPUT
 178         PNAT_IP4_OUTPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_OUTPUT
 179
 180         tests = [
 181             {
 182                 "input": PNAT_IP4_INPUT,
 183                 "sw_if_index": self.pg0.sw_if_index,
 184                 "match": {
 185                     "mask": 0xA,
 186                     "dst": "10.10.10.10",
 187                     "proto": 17,
 188                     "dport": 6871,
 189                 },
 190                 "rewrite": {"mask": 0x2, "dst": self.pg1.remote_ip4},
 191                 "send": (
 192                     IP(src=self.pg0.remote_ip4, dst="10.10.10.10") / UDP(dport=6871)
 193                 ),
 194                 "reply": (
 195                     IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
 196                     / UDP(dport=6871)
 197                 ),
 198             },
 199             {
 200                 "input": PNAT_IP4_OUTPUT,
 201                 "sw_if_index": self.pg1.sw_if_index,
 202                 "match": {
 203                     "mask": 0x9,
 204                     "src": self.pg0.remote_ip4,
 205                     "proto": 17,
 206                     "dport": 6871,
 207                 },
 208                 "rewrite": {"mask": 0x1, "src": "11.11.11.11"},
 209                 "send": (
 210                     IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
 211                     / UDP(dport=6871)
 212                 ),
 213                 "reply": (
 214                     IP(src="11.11.11.11", dst=self.pg1.remote_ip4) / UDP(dport=6871)
 215                 ),
 216             },
 217         ]
 218         binding_index = []
 219         for t in tests:
 220             rv = self.vapi.pnat_binding_add(match=t["match"], rewrite=t["rewrite"])
 221             binding_index.append(rv.binding_index)
 222             self.vapi.pnat_binding_attach(
 223                 sw_if_index=t["sw_if_index"],
 224                 attachment=t["input"],
 225                 binding_index=rv.binding_index,
 226             )
 227
 228         rv, l = self.vapi.pnat_bindings_get()
 229         self.assertEqual(len(l), len(tests))
 230
 231         rv, l = self.vapi.pnat_interfaces_get()
 232         self.assertEqual(len(l), 2)
 233
 234         self.logger.info(self.vapi.cli("show pnat translations"))
 235         self.logger.info(self.vapi.cli("show pnat interfaces"))
 236
 237         for i, t in enumerate(tests):
 238             self.vapi.pnat_binding_detach(
 239                 sw_if_index=t["sw_if_index"],
 240                 attachment=t["input"],
 241                 binding_index=binding_index[i],
 242             )
 243             self.vapi.pnat_binding_del(binding_index=binding_index[i])
 244
 245
 246 if __name__ == "__main__":
 247     unittest.main(testRunner=VppTestRunner)