3 # Copyright (c) 2016 Cisco and/or its affiliates.
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at:
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
16 """Traffic script for IPsec verification."""
21 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
22 from scapy.all import Ether, IP, ICMP, IPv6, ICMPv6EchoRequest, ICMPv6EchoReply
23 from scapy.layers.ipsec import SecurityAssociation, ESP
24 from ipaddress import ip_address
26 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
27 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
30 def check_ipv4(pkt_recv, dst_tun, src_ip, dst_ip, sa_in):
31 """Check received IPv4 IPsec packet.
33 :param pkt_recv: Received packet to verify.
34 :param dst_tun: IPsec tunnel destination address.
35 :param src_ip: Source address of original IPv4 packet.
36 :param dst_ip: Destination address of original IPv4 packet.
37 :param sa_in: IPsec SA for packet decryption.
38 :type pkt_recv: scapy.Ether
42 :type sa_sa: scapy.layers.ipsec.SecurityAssociation
43 :raises RuntimeError: If received packet is invalid.
45 if not pkt_recv.haslayer(IP):
47 'Not an IPv4 packet received: {0}'.format(pkt_recv.__repr__()))
49 if pkt_recv['IP'].dst != dst_tun:
51 'Received packet has invalid destination address: {0} '
52 'should be: {1}'.format(pkt_recv['IP'].dst, dst_tun))
54 if not pkt_recv.haslayer(ESP):
56 'Not an ESP packet received: {0}'.format(pkt_recv.__repr__()))
58 ip_pkt = pkt_recv['IP']
59 d_pkt = sa_in.decrypt(ip_pkt)
61 if d_pkt['IP'].dst != dst_ip:
63 'Decrypted packet has invalid destination address: {0} '
64 'should be: {1}'.format(d_pkt['IP'].dst, dst_ip))
66 if d_pkt['IP'].src != src_ip:
68 'Decrypted packet has invalid source address: {0} should be: {1}'
69 .format(d_pkt['IP'].src, src_ip))
71 if not d_pkt.haslayer(ICMP):
73 'Decrypted packet does not have ICMP layer: {0}'.format(
77 def check_ipv6(pkt_recv, dst_tun, src_ip, dst_ip, sa_in):
78 """Check received IPv6 IPsec packet.
80 :param pkt_recv: Received packet to verify.
81 :param dst_tun: IPsec tunnel destination address.
82 :param src_ip: Source address of original IPv6 packet.
83 :param dst_ip: Destination address of original IPv6 packet.
84 :param sa_in: IPsec SA for packet decryption.
85 :type pkt_recv: scapy.Ether
89 :type sa_in: scapy.layers.ipsec.SecurityAssociation
90 :raises RuntimeError: If received packet is invalid.
92 if not pkt_recv.haslayer(IPv6):
94 'Not an IPv6 packet received: {0}'.format(pkt_recv.__repr__()))
96 if pkt_recv['IPv6'].dst != dst_tun:
98 'Received packet has invalid destination address: {0} '
99 'should be: {1}'.format(pkt_recv['IPv6'].dst, dst_tun))
101 if not pkt_recv.haslayer(ESP):
103 'Not an ESP packet received: {0}'.format(pkt_recv.__repr__()))
105 ip_pkt = pkt_recv['IPv6']
106 d_pkt = sa_in.decrypt(ip_pkt)
108 if d_pkt['IPv6'].dst != dst_ip:
110 'Decrypted packet has invalid destination address {0}: '
111 'should be: {1}'.format(d_pkt['IPv6'].dst, dst_ip))
113 if d_pkt['IPv6'].src != src_ip:
115 'Decrypted packet has invalid source address: {0} should be: {1}'
116 .format(d_pkt['IPv6'].src, src_ip))
118 if not d_pkt.haslayer(ICMPv6EchoReply):
120 'Decrypted packet does not have ICMP layer: {0}'.format(
124 # pylint: disable=too-many-locals
125 # pylint: disable=too-many-statements
127 """Send and receive IPsec packet."""
128 args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip',
129 'crypto_alg', 'crypto_key', 'integ_alg',
130 'integ_key', 'l_spi', 'r_spi'],
131 ['src_tun', 'dst_tun'])
133 rxq = RxQueue(args.get_arg('rx_if'))
134 txq = TxQueue(args.get_arg('tx_if'))
136 src_mac = args.get_arg('src_mac')
137 dst_mac = args.get_arg('dst_mac')
138 src_ip = args.get_arg('src_ip')
139 dst_ip = args.get_arg('dst_ip')
140 crypto_alg = args.get_arg('crypto_alg')
141 crypto_key = args.get_arg('crypto_key')
142 integ_alg = args.get_arg('integ_alg')
143 integ_key = args.get_arg('integ_key')
144 l_spi = int(args.get_arg('l_spi'))
145 r_spi = int(args.get_arg('r_spi'))
146 src_tun = args.get_arg('src_tun')
147 dst_tun = args.get_arg('dst_tun')
150 if 6 == ip_address(unicode(src_ip)).version:
156 if src_tun and dst_tun:
158 tunnel_out = IP(src=src_tun, dst=dst_tun)
159 tunnel_in = IP(src=dst_tun, dst=src_tun)
161 tunnel_out = IPv6(src=src_tun, dst=dst_tun)
162 tunnel_in = IPv6(src=dst_tun, dst=src_tun)
167 sa_in = SecurityAssociation(ESP, spi=r_spi, crypt_algo=crypto_alg,
168 crypt_key=crypto_key, auth_algo=integ_alg,
169 auth_key=integ_key, tunnel_header=tunnel_in)
171 sa_out = SecurityAssociation(ESP, spi=l_spi, crypt_algo=crypto_alg,
172 crypt_key=crypto_key, auth_algo=integ_alg,
173 auth_key=integ_key, tunnel_header=tunnel_out)
178 ip_pkt = IP(src=src_ip, dst=dst_ip) / \
180 ip_pkt = IP(str(ip_pkt))
182 ip_pkt = IPv6(src=src_ip, dst=dst_ip) / \
184 ip_pkt = IPv6(str(ip_pkt))
186 e_pkt = sa_out.encrypt(ip_pkt)
187 pkt_send = Ether(src=src_mac, dst=dst_mac) / \
190 sent_packets.append(pkt_send)
193 pkt_recv = rxq.recv(2, sent_packets)
196 raise RuntimeError('ESP packet Rx timeout')
199 check_ipv4(pkt_recv, src_tun, dst_ip, src_ip, sa_in)
201 check_ipv6(pkt_recv, src_tun, dst_ip, src_ip, sa_in)
205 if __name__ == "__main__":