6 from scapy.layers.l2 import Ether
7 from scapy.layers.inet import ICMP, IP, TCP, UDP
8 from scapy.layers.ipsec import SecurityAssociation, ESP
10 from util import ppp, ppc
11 from template_ipsec import TemplateIpsec
12 from vpp_ipsec import VppIpsecSA, VppIpsecSpd, VppIpsecSpdEntry,\
14 from vpp_ip_route import VppIpRoute, VppRoutePath
15 from vpp_ip import DpoProto
16 from vpp_papi import VppEnum
19 class IPSecNATTestCase(TemplateIpsec):
24 public network | private network
25 --- encrypt --- plain ---
26 |pg0| <------- |VPP| <------ |pg1|
29 --- decrypt --- plain ---
30 |pg0| -------> |VPP| ------> |pg1|
43 super(IPSecNATTestCase, cls).setUpClass()
46 def tearDownClass(cls):
47 super(IPSecNATTestCase, cls).tearDownClass()
50 super(IPSecNATTestCase, self).setUp()
51 self.tun_if = self.pg0
53 self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
54 self.tun_spd.add_vpp_config()
55 VppIpsecSpdItfBinding(self, self.tun_spd,
56 self.tun_if).add_vpp_config()
59 self.config_esp_tun(p)
60 self.logger.info(self.vapi.ppcli("show ipsec all"))
62 d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4
63 VppIpRoute(self, p.remote_tun_if_host, p.addr_len,
64 [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
67 is_ip6=p.is_ipv6).add_vpp_config()
70 super(IPSecNATTestCase, self).tearDown()
72 def create_stream_plain(self, src_mac, dst_mac, src_ip, dst_ip):
75 Ether(src=src_mac, dst=dst_mac) /
76 IP(src=src_ip, dst=dst_ip) /
77 TCP(sport=self.tcp_port_in, dport=20),
79 Ether(src=src_mac, dst=dst_mac) /
80 IP(src=src_ip, dst=dst_ip) /
81 UDP(sport=self.udp_port_in, dport=20),
83 Ether(src=src_mac, dst=dst_mac) /
84 IP(src=src_ip, dst=dst_ip) /
85 ICMP(id=self.icmp_id_in, type='echo-request')
88 def create_stream_encrypted(self, src_mac, dst_mac, src_ip, dst_ip, sa):
91 Ether(src=src_mac, dst=dst_mac) /
92 sa.encrypt(IP(src=src_ip, dst=dst_ip) /
93 TCP(dport=self.tcp_port_out, sport=20)),
95 Ether(src=src_mac, dst=dst_mac) /
96 sa.encrypt(IP(src=src_ip, dst=dst_ip) /
97 UDP(dport=self.udp_port_out, sport=20)),
99 Ether(src=src_mac, dst=dst_mac) /
100 sa.encrypt(IP(src=src_ip, dst=dst_ip) /
101 ICMP(id=self.icmp_id_out, type='echo-request'))
104 def verify_capture_plain(self, capture):
105 for packet in capture:
107 self.assert_packet_checksums_valid(packet)
108 self.assert_equal(packet[IP].src, self.tun_if.remote_ip4,
109 "decrypted packet source address")
110 self.assert_equal(packet[IP].dst, self.pg1.remote_ip4,
111 "decrypted packet destination address")
112 if packet.haslayer(TCP):
114 packet.haslayer(UDP),
115 "unexpected UDP header in decrypted packet")
116 self.assert_equal(packet[TCP].dport, self.tcp_port_in,
117 "decrypted packet TCP destination port")
118 elif packet.haslayer(UDP):
119 if packet[UDP].payload:
121 packet[UDP][1].haslayer(UDP),
122 "unexpected UDP header in decrypted packet")
123 self.assert_equal(packet[UDP].dport, self.udp_port_in,
124 "decrypted packet UDP destination port")
127 packet.haslayer(UDP),
128 "unexpected UDP header in decrypted packet")
129 self.assert_equal(packet[ICMP].id, self.icmp_id_in,
130 "decrypted packet ICMP ID")
133 ppp("Unexpected or invalid plain packet:", packet))
136 def verify_capture_encrypted(self, capture, sa):
137 for packet in capture:
139 copy = packet.__class__(scapy.compat.raw(packet))
141 copy = packet.__class__(scapy.compat.raw(copy))
142 self.assert_equal(packet[UDP].len, copy[UDP].len,
144 self.assert_packet_checksums_valid(packet)
145 self.assertIn(ESP, packet[IP])
146 decrypt_pkt = sa.decrypt(packet[IP])
147 self.assert_packet_checksums_valid(decrypt_pkt)
148 self.assert_equal(decrypt_pkt[IP].src, self.pg1.remote_ip4,
149 "encrypted packet source address")
150 self.assert_equal(decrypt_pkt[IP].dst, self.tun_if.remote_ip4,
151 "encrypted packet destination address")
154 ppp("Unexpected or invalid encrypted packet:", packet))
157 def config_esp_tun(self, params):
158 addr_type = params.addr_type
159 scapy_tun_sa_id = params.scapy_tun_sa_id
160 scapy_tun_spi = params.scapy_tun_spi
161 vpp_tun_sa_id = params.vpp_tun_sa_id
162 vpp_tun_spi = params.vpp_tun_spi
163 auth_algo_vpp_id = params.auth_algo_vpp_id
164 auth_key = params.auth_key
165 crypt_algo_vpp_id = params.crypt_algo_vpp_id
166 crypt_key = params.crypt_key
167 addr_any = params.addr_any
168 addr_bcast = params.addr_bcast
169 flags = (VppEnum.vl_api_ipsec_sad_flags_t.
170 IPSEC_API_SAD_FLAG_UDP_ENCAP)
171 e = VppEnum.vl_api_ipsec_spd_action_t
173 VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi,
174 auth_algo_vpp_id, auth_key,
175 crypt_algo_vpp_id, crypt_key,
176 self.vpp_esp_protocol,
177 self.pg1.remote_addr[addr_type],
178 self.tun_if.remote_addr[addr_type],
179 flags=flags).add_vpp_config()
180 VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
181 auth_algo_vpp_id, auth_key,
182 crypt_algo_vpp_id, crypt_key,
183 self.vpp_esp_protocol,
184 self.tun_if.remote_addr[addr_type],
185 self.pg1.remote_addr[addr_type],
186 flags=flags).add_vpp_config()
188 VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
189 addr_any, addr_bcast,
190 addr_any, addr_bcast,
191 socket.IPPROTO_ESP).add_vpp_config()
192 VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
193 addr_any, addr_bcast,
194 addr_any, addr_bcast,
196 is_outbound=0).add_vpp_config()
197 VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
198 addr_any, addr_bcast,
199 addr_any, addr_bcast,
201 remote_port_start=4500,
202 remote_port_stop=4500).add_vpp_config()
203 VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
204 addr_any, addr_bcast,
205 addr_any, addr_bcast,
207 remote_port_start=4500,
208 remote_port_stop=4500,
209 is_outbound=0).add_vpp_config()
210 VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
211 self.tun_if.remote_addr[addr_type],
212 self.tun_if.remote_addr[addr_type],
213 self.pg1.remote_addr[addr_type],
214 self.pg1.remote_addr[addr_type],
216 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
217 is_outbound=0).add_vpp_config()
218 VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
219 self.pg1.remote_addr[addr_type],
220 self.pg1.remote_addr[addr_type],
221 self.tun_if.remote_addr[addr_type],
222 self.tun_if.remote_addr[addr_type],
223 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
224 priority=10).add_vpp_config()
226 def test_ipsec_nat_tun(self):
227 """ IPSec/NAT tunnel test case """
229 scapy_tun_sa = SecurityAssociation(ESP, spi=p.scapy_tun_spi,
230 crypt_algo=p.crypt_algo,
231 crypt_key=p.crypt_key,
232 auth_algo=p.auth_algo,
235 src=self.pg1.remote_ip4,
236 dst=self.tun_if.remote_ip4),
240 # in2out - from private network to public
241 pkts = self.create_stream_plain(
242 self.pg1.remote_mac, self.pg1.local_mac,
243 self.pg1.remote_ip4, self.tun_if.remote_ip4)
244 self.pg1.add_stream(pkts)
245 self.pg_enable_capture(self.pg_interfaces)
247 capture = self.tun_if.get_capture(len(pkts))
248 self.verify_capture_encrypted(capture, scapy_tun_sa)
250 vpp_tun_sa = SecurityAssociation(ESP,
252 crypt_algo=p.crypt_algo,
253 crypt_key=p.crypt_key,
254 auth_algo=p.auth_algo,
257 src=self.tun_if.remote_ip4,
258 dst=self.pg1.remote_ip4),
263 # out2in - from public network to private
264 pkts = self.create_stream_encrypted(
265 self.tun_if.remote_mac, self.tun_if.local_mac,
266 self.tun_if.remote_ip4, self.pg1.remote_ip4, vpp_tun_sa)
267 self.logger.info(ppc("Sending packets:", pkts))
268 self.tun_if.add_stream(pkts)
269 self.pg_enable_capture(self.pg_interfaces)
271 capture = self.pg1.get_capture(len(pkts))
272 self.verify_capture_plain(capture)