resources/traffic_scripts/ipsec.py

   1 #!/usr/bin/env python
   2
   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:
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  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.
  15
  16 """Traffic script for IPsec verification."""
  17
  18 import sys
  19 import logging
  20
  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
  25
  26 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  27 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
  28
  29
  30 def check_ipv4(pkt_recv, dst_tun, src_ip, dst_ip, sa_in):
  31     """Check received IPv4 IPsec packet.
  32
  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
  39     :type dst_tun: str
  40     :type src_ip: str
  41     :type dst_ip: str
  42     :type sa_sa: scapy.layers.ipsec.SecurityAssociation
  43     :raises RuntimeError: If received packet is invalid.
  44     """
  45     if not pkt_recv.haslayer(IP):
  46         raise RuntimeError(
  47             'Not an IPv4 packet received: {0}'.format(pkt_recv.__repr__()))
  48
  49     if pkt_recv['IP'].dst != dst_tun:
  50         raise RuntimeError(
  51             'Received packet has invalid destination address: {0} '
  52             'should be: {1}'.format(pkt_recv['IP'].dst, dst_tun))
  53
  54     if not pkt_recv.haslayer(ESP):
  55         raise RuntimeError(
  56             'Not an ESP packet received: {0}'.format(pkt_recv.__repr__()))
  57
  58     ip_pkt = pkt_recv['IP']
  59     d_pkt = sa_in.decrypt(ip_pkt)
  60
  61     if d_pkt['IP'].dst != dst_ip:
  62         raise RuntimeError(
  63             'Decrypted packet has invalid destination address: {0} '
  64             'should be: {1}'.format(d_pkt['IP'].dst, dst_ip))
  65
  66     if d_pkt['IP'].src != src_ip:
  67         raise RuntimeError(
  68             'Decrypted packet has invalid source address: {0} should be: {1}'
  69             .format(d_pkt['IP'].src, src_ip))
  70
  71     if not d_pkt.haslayer(ICMP):
  72         raise RuntimeError(
  73             'Decrypted packet does not have ICMP layer: {0}'.format(
  74                 d_pkt.__repr__()))
  75
  76
  77 def check_ipv6(pkt_recv, dst_tun, src_ip, dst_ip, sa_in):
  78     """Check received IPv6 IPsec packet.
  79
  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
  86     :type dst_tun: str
  87     :type src_ip: str
  88     :type dst_ip: str
  89     :type sa_in: scapy.layers.ipsec.SecurityAssociation
  90     :raises RuntimeError: If received packet is invalid.
  91     """
  92     if not pkt_recv.haslayer(IPv6):
  93         raise RuntimeError(
  94             'Not an IPv6 packet received: {0}'.format(pkt_recv.__repr__()))
  95
  96     if pkt_recv['IPv6'].dst != dst_tun:
  97         raise RuntimeError(
  98             'Received packet has invalid destination address: {0} '
  99             'should be: {1}'.format(pkt_recv['IPv6'].dst, dst_tun))
 100
 101     if not pkt_recv.haslayer(ESP):
 102         raise RuntimeError(
 103             'Not an ESP packet received: {0}'.format(pkt_recv.__repr__()))
 104
 105     ip_pkt = pkt_recv['IPv6']
 106     d_pkt = sa_in.decrypt(ip_pkt)
 107
 108     if d_pkt['IPv6'].dst != dst_ip:
 109         raise RuntimeError(
 110             'Decrypted packet has invalid destination address {0}: '
 111             'should be: {1}'.format(d_pkt['IPv6'].dst, dst_ip))
 112
 113     if d_pkt['IPv6'].src != src_ip:
 114         raise RuntimeError(
 115             'Decrypted packet has invalid source address: {0} should be: {1}'
 116             .format(d_pkt['IPv6'].src, src_ip))
 117
 118     if not d_pkt.haslayer(ICMPv6EchoReply):
 119         raise RuntimeError(
 120             'Decrypted packet does not have ICMP layer: {0}'.format(
 121                 d_pkt.__repr__()))
 122
 123
 124 # pylint: disable=too-many-locals
 125 # pylint: disable=too-many-statements
 126 def main():
 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'])
 132
 133     rxq = RxQueue(args.get_arg('rx_if'))
 134     txq = TxQueue(args.get_arg('tx_if'))
 135
 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')
 148
 149     is_ipv4 = True
 150     if 6 == ip_address(unicode(src_ip)).version:
 151         is_ipv4 = False
 152
 153     tunnel_out = None
 154     tunnel_in = None
 155
 156     if src_tun and dst_tun:
 157         if is_ipv4:
 158             tunnel_out = IP(src=src_tun, dst=dst_tun)
 159             tunnel_in = IP(src=dst_tun, dst=src_tun)
 160         else:
 161             tunnel_out = IPv6(src=src_tun, dst=dst_tun)
 162             tunnel_in = IPv6(src=dst_tun, dst=src_tun)
 163     else:
 164         src_tun = src_ip
 165         dst_tun = dst_ip
 166
 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)
 170
 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)
 174
 175     sent_packets = []
 176
 177     ip_pkt = None
 178
 179     if is_ipv4:
 180         ip_pkt = IP(src=src_ip, dst=dst_ip) / \
 181                  ICMP()
 182         ip_pkt = IP(str(ip_pkt))
 183     else:
 184         ip_pkt = IPv6(src=src_ip, dst=dst_ip) / \
 185                  ICMPv6EchoRequest()
 186         ip_pkt = IPv6(str(ip_pkt))
 187
 188     e_pkt = sa_out.encrypt(ip_pkt)
 189     pkt_send = Ether(src=src_mac, dst=dst_mac) / \
 190                e_pkt
 191
 192     sent_packets.append(pkt_send)
 193     txq.send(pkt_send)
 194
 195     pkt_recv = rxq.recv(2, sent_packets)
 196
 197     if pkt_recv is None:
 198         raise RuntimeError('Rx timeout')
 199
 200     if is_ipv4:
 201         check_ipv4(pkt_recv, src_tun, dst_ip, src_ip, sa_in)
 202     else:
 203         check_ipv6(pkt_recv, src_tun, dst_ip, src_ip, sa_in)
 204
 205     sys.exit(0)
 206
 207 if __name__ == "__main__":
 208     main()