diff options
Diffstat (limited to 'test/test_pnat.py')
| -rw-r--r-- | test/test_pnat.py | 344 |
1 files changed, 245 insertions, 99 deletions
diff --git a/test/test_pnat.py b/test/test_pnat.py index d5b60050691..a7bd24b612c 100644 --- a/test/test_pnat.py +++ b/test/test_pnat.py @@ -3,12 +3,14 @@ import unittest from scapy.layers.inet import Ether, IP, UDP, ICMP -from framework import VppTestCase, VppTestRunner +from framework import VppTestCase +from asfframework import VppTestRunner from vpp_papi import VppEnum class TestPNAT(VppTestCase): - """ PNAT Test Case """ + """PNAT Test Case""" + maxDiff = None @classmethod @@ -42,147 +44,187 @@ class TestPNAT(VppTestCase): self.assertEqual(rx, expected) def ping_check(self): - """ Verify non matching traffic works. """ + """Verify non matching traffic works.""" p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) - icmpecho = (IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) / - ICMP()) - reply = (IP(src=self.pg0.local_ip4, dst=self.pg0.remote_ip4) / - ICMP(type='echo-reply')) - rx = self.send_and_expect(self.pg0, p_ether/icmpecho * 1, self.pg0) + icmpecho = IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) / ICMP() + reply = IP(src=self.pg0.local_ip4, dst=self.pg0.remote_ip4) / ICMP( + type="echo-reply" + ) + rx = self.send_and_expect(self.pg0, p_ether / icmpecho * 1, self.pg0) for p in rx: reply[IP].id = p[IP].id self.validate(p[1], reply) def test_pnat(self): - """ PNAT test """ + """PNAT test""" PNAT_IP4_INPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_INPUT - PNAT_IP4_OUTPUT = \ - VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_OUTPUT + PNAT_IP4_OUTPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_OUTPUT tests = [ { - 'input': PNAT_IP4_INPUT, - 'sw_if_index': self.pg0.sw_if_index, - 'match': {'mask': 0xa, 'dst': '10.10.10.10', 'proto': 17, - 'dport': 6871}, - 'rewrite': {'mask': 0x2, 'dst': self.pg1.remote_ip4}, - 'send': (IP(src=self.pg0.remote_ip4, dst='10.10.10.10') / - UDP(dport=6871)), - 'reply': (IP(src=self.pg0.remote_ip4, - dst=self.pg1.remote_ip4) / - UDP(dport=6871)) + "input": PNAT_IP4_INPUT, + "sw_if_index": self.pg0.sw_if_index, + "match": { + "mask": 0xA, + "dst": "10.10.10.10", + "proto": 17, + "dport": 6871, + }, + "rewrite": {"mask": 0x2, "dst": self.pg1.remote_ip4}, + "send": ( + IP(src=self.pg0.remote_ip4, dst="10.10.10.10") / UDP(dport=6871) + ), + "reply": ( + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / UDP(dport=6871) + ), }, { - 'input': PNAT_IP4_OUTPUT, - 'sw_if_index': self.pg1.sw_if_index, - 'match': {'mask': 0x9, 'src': self.pg0.remote_ip4, 'proto': 17, - 'dport': 6871}, - 'rewrite': {'mask': 0x1, 'src': '11.11.11.11'}, - 'send': (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - UDP(dport=6871)), - 'reply': (IP(src='11.11.11.11', dst=self.pg1.remote_ip4) / - UDP(dport=6871)) + "input": PNAT_IP4_OUTPUT, + "sw_if_index": self.pg1.sw_if_index, + "match": { + "mask": 0x9, + "src": self.pg0.remote_ip4, + "proto": 17, + "dport": 6871, + }, + "rewrite": {"mask": 0x1, "src": "11.11.11.11"}, + "send": ( + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / UDP(dport=6871) + ), + "reply": ( + IP(src="11.11.11.11", dst=self.pg1.remote_ip4) / UDP(dport=6871) + ), }, { - 'input': PNAT_IP4_INPUT, - 'sw_if_index': self.pg0.sw_if_index, - 'match': {'mask': 0xa, 'dst': '10.10.10.10', 'proto': 17, - 'dport': 6871}, - 'rewrite': {'mask': 0xa, 'dst': self.pg1.remote_ip4, - 'dport': 5555}, - 'send': (IP(src=self.pg0.remote_ip4, dst='10.10.10.10') / - UDP(sport=65530, dport=6871)), - 'reply': (IP(src=self.pg0.remote_ip4, - dst=self.pg1.remote_ip4) / - UDP(sport=65530, dport=5555)) + "input": PNAT_IP4_INPUT, + "sw_if_index": self.pg0.sw_if_index, + "match": { + "mask": 0xA, + "dst": "10.10.10.10", + "proto": 17, + "dport": 6871, + }, + "rewrite": {"mask": 0xA, "dst": self.pg1.remote_ip4, "dport": 5555}, + "send": ( + IP(src=self.pg0.remote_ip4, dst="10.10.10.10") + / UDP(sport=65530, dport=6871) + ), + "reply": ( + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / UDP(sport=65530, dport=5555) + ), }, { - 'input': PNAT_IP4_INPUT, - 'sw_if_index': self.pg0.sw_if_index, - 'match': {'mask': 0xa, 'dst': self.pg1.remote_ip4, 'proto': 17, - 'dport': 6871}, - 'rewrite': {'mask': 0x8, 'dport': 5555}, - 'send': (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - UDP(dport=6871, chksum=0)), - 'reply': (IP(src=self.pg0.remote_ip4, - dst=self.pg1.remote_ip4) / - UDP(dport=5555, chksum=0)) + "input": PNAT_IP4_INPUT, + "sw_if_index": self.pg0.sw_if_index, + "match": { + "mask": 0xA, + "dst": self.pg1.remote_ip4, + "proto": 17, + "dport": 6871, + }, + "rewrite": {"mask": 0x8, "dport": 5555}, + "send": ( + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / UDP(dport=6871, chksum=0) + ), + "reply": ( + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / UDP(dport=5555, chksum=0) + ), }, { - 'input': PNAT_IP4_INPUT, - 'sw_if_index': self.pg0.sw_if_index, - 'match': {'mask': 0x2, 'dst': self.pg1.remote_ip4, 'proto': 1}, - 'rewrite': {'mask': 0x1, 'src': '8.8.8.8'}, - 'send': (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - ICMP()), - 'reply': IP(src='8.8.8.8', dst=self.pg1.remote_ip4)/ICMP(), + "input": PNAT_IP4_INPUT, + "sw_if_index": self.pg0.sw_if_index, + "match": {"mask": 0x2, "dst": self.pg1.remote_ip4, "proto": 1}, + "rewrite": {"mask": 0x1, "src": "8.8.8.8"}, + "send": (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / ICMP()), + "reply": IP(src="8.8.8.8", dst=self.pg1.remote_ip4) / ICMP(), }, ] p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) for t in tests: - rv = self.vapi.pnat_binding_add(match=t['match'], - rewrite=t['rewrite']) - self.vapi.pnat_binding_attach(sw_if_index=t['sw_if_index'], - attachment=t['input'], - binding_index=rv.binding_index) + rv = self.vapi.pnat_binding_add(match=t["match"], rewrite=t["rewrite"]) + self.vapi.pnat_binding_attach( + sw_if_index=t["sw_if_index"], + attachment=t["input"], + binding_index=rv.binding_index, + ) - reply = t['reply'] + reply = t["reply"] reply[IP].ttl -= 1 - rx = self.send_and_expect(self.pg0, p_ether/t['send']*1, self.pg1) + rx = self.send_and_expect(self.pg0, p_ether / t["send"] * 1, self.pg1) for p in rx: # p.show2() self.validate(p[1], reply) self.ping_check() - self.vapi.pnat_binding_detach(sw_if_index=t['sw_if_index'], - attachment=t['input'], - binding_index=rv.binding_index) + self.vapi.pnat_binding_detach( + sw_if_index=t["sw_if_index"], + attachment=t["input"], + binding_index=rv.binding_index, + ) self.vapi.pnat_binding_del(binding_index=rv.binding_index) def test_pnat_show(self): - """ PNAT show tests """ + """PNAT show tests""" PNAT_IP4_INPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_INPUT - PNAT_IP4_OUTPUT = \ - VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_OUTPUT + PNAT_IP4_OUTPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_OUTPUT tests = [ { - 'input': PNAT_IP4_INPUT, - 'sw_if_index': self.pg0.sw_if_index, - 'match': {'mask': 0xa, 'dst': '10.10.10.10', 'proto': 17, - 'dport': 6871}, - 'rewrite': {'mask': 0x2, 'dst': self.pg1.remote_ip4}, - 'send': (IP(src=self.pg0.remote_ip4, dst='10.10.10.10') / - UDP(dport=6871)), - 'reply': (IP(src=self.pg0.remote_ip4, - dst=self.pg1.remote_ip4) / - UDP(dport=6871)) + "input": PNAT_IP4_INPUT, + "sw_if_index": self.pg0.sw_if_index, + "match": { + "mask": 0xA, + "dst": "10.10.10.10", + "proto": 17, + "dport": 6871, + }, + "rewrite": {"mask": 0x2, "dst": self.pg1.remote_ip4}, + "send": ( + IP(src=self.pg0.remote_ip4, dst="10.10.10.10") / UDP(dport=6871) + ), + "reply": ( + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / UDP(dport=6871) + ), }, { - 'input': PNAT_IP4_OUTPUT, - 'sw_if_index': self.pg1.sw_if_index, - 'match': {'mask': 0x9, 'src': self.pg0.remote_ip4, 'proto': 17, - 'dport': 6871}, - 'rewrite': {'mask': 0x1, 'src': '11.11.11.11'}, - 'send': (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / - UDP(dport=6871)), - 'reply': (IP(src='11.11.11.11', dst=self.pg1.remote_ip4) / - UDP(dport=6871)) + "input": PNAT_IP4_OUTPUT, + "sw_if_index": self.pg1.sw_if_index, + "match": { + "mask": 0x9, + "src": self.pg0.remote_ip4, + "proto": 17, + "dport": 6871, + }, + "rewrite": {"mask": 0x1, "src": "11.11.11.11"}, + "send": ( + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) + / UDP(dport=6871) + ), + "reply": ( + IP(src="11.11.11.11", dst=self.pg1.remote_ip4) / UDP(dport=6871) + ), }, ] binding_index = [] for t in tests: - rv = self.vapi.pnat_binding_add(match=t['match'], - rewrite=t['rewrite']) + rv = self.vapi.pnat_binding_add(match=t["match"], rewrite=t["rewrite"]) binding_index.append(rv.binding_index) - self.vapi.pnat_binding_attach(sw_if_index=t['sw_if_index'], - attachment=t['input'], - binding_index=rv.binding_index) + self.vapi.pnat_binding_attach( + sw_if_index=t["sw_if_index"], + attachment=t["input"], + binding_index=rv.binding_index, + ) rv, l = self.vapi.pnat_bindings_get() self.assertEqual(len(l), len(tests)) @@ -194,10 +236,114 @@ class TestPNAT(VppTestCase): self.logger.info(self.vapi.cli("show pnat interfaces")) for i, t in enumerate(tests): - self.vapi.pnat_binding_detach(sw_if_index=t['sw_if_index'], - attachment=t['input'], - binding_index=binding_index[i]) + self.vapi.pnat_binding_detach( + sw_if_index=t["sw_if_index"], + attachment=t["input"], + binding_index=binding_index[i], + ) self.vapi.pnat_binding_del(binding_index=binding_index[i]) -if __name__ == '__main__': + def test_pnat_wildcard_proto(self): + """ + PNAT test wildcard IP protocol, PNAT_PROTO for mask should be set by + handler + """ + + PNAT_IP4_INPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_INPUT + PNAT_IP4_OUTPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_OUTPUT + + tests = [ + { + "input": PNAT_IP4_INPUT, + "sw_if_index": self.pg0.sw_if_index, + "match": {"mask": 0x2, "dst": "10.10.10.10"}, + "rewrite": {"mask": 0x2, "dst": self.pg1.remote_ip4}, + "send": (IP(src=self.pg0.remote_ip4, dst="10.10.10.10")), + "reply": (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)), + }, + { + "input": PNAT_IP4_OUTPUT, + "sw_if_index": self.pg1.sw_if_index, + "match": {"mask": 0x1, "src": self.pg0.remote_ip4}, + "rewrite": {"mask": 0x1, "src": "11.11.11.11"}, + "send": (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)), + "reply": (IP(src="11.11.11.11", dst=self.pg1.remote_ip4)), + }, + ] + + p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + for t in tests: + rv = self.vapi.pnat_binding_add(match=t["match"], rewrite=t["rewrite"]) + self.vapi.pnat_binding_attach( + sw_if_index=t["sw_if_index"], + attachment=t["input"], + binding_index=rv.binding_index, + ) + + reply = t["reply"] + reply[IP].ttl -= 1 + rx = self.send_and_expect(self.pg0, p_ether / t["send"] * 1, self.pg1) + for p in rx: + self.validate(p[1], reply) + + self.ping_check() + + self.vapi.pnat_binding_detach( + sw_if_index=t["sw_if_index"], + attachment=t["input"], + binding_index=rv.binding_index, + ) + self.vapi.pnat_binding_del(binding_index=rv.binding_index) + + def test_pnat_wildcard_proto_v2(self): + """PNAT test wildcard IP protocol using pnat_binding_add_v2""" + + PNAT_IP4_INPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_INPUT + PNAT_IP4_OUTPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_OUTPUT + + tests = [ + { + "input": PNAT_IP4_INPUT, + "sw_if_index": self.pg0.sw_if_index, + "match": {"mask": 0x42, "dst": "10.10.10.10"}, + "rewrite": {"mask": 0x42, "dst": self.pg1.remote_ip4}, + "send": (IP(src=self.pg0.remote_ip4, dst="10.10.10.10")), + "reply": (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)), + }, + { + "input": PNAT_IP4_OUTPUT, + "sw_if_index": self.pg1.sw_if_index, + "match": {"mask": 0x41, "src": self.pg0.remote_ip4}, + "rewrite": {"mask": 0x41, "src": "11.11.11.11"}, + "send": (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)), + "reply": (IP(src="11.11.11.11", dst=self.pg1.remote_ip4)), + }, + ] + + p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + for t in tests: + rv = self.vapi.pnat_binding_add_v2(match=t["match"], rewrite=t["rewrite"]) + self.vapi.pnat_binding_attach( + sw_if_index=t["sw_if_index"], + attachment=t["input"], + binding_index=rv.binding_index, + ) + + reply = t["reply"] + reply[IP].ttl -= 1 + rx = self.send_and_expect(self.pg0, p_ether / t["send"] * 1, self.pg1) + for p in rx: + self.validate(p[1], reply) + + self.ping_check() + + self.vapi.pnat_binding_detach( + sw_if_index=t["sw_if_index"], + attachment=t["input"], + binding_index=rv.binding_index, + ) + self.vapi.pnat_binding_del(binding_index=rv.binding_index) + + +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner) |