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 # pylint: disable=no-name-in-module
  22 # pylint: disable=import-error
  23 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
  24
  25 from scapy.all import Ether
  26 from scapy.layers.inet import ICMP, IP
  27 from scapy.layers.inet6 import IPv6, ICMPv6ND_NS
  28 from scapy.layers.inet6 import ICMPv6EchoRequest, ICMPv6EchoReply
  29 from scapy.layers.ipsec import SecurityAssociation, ESP
  30 from ipaddress import ip_address
  31
  32 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  33 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
  34
  35
  36 def check_ipv4(pkt_recv, dst_tun, src_ip, dst_ip, sa_in):
  37     """Check received IPv4 IPsec packet.
  38
  39     :param pkt_recv: Received packet to verify.
  40     :param dst_tun: IPsec tunnel destination address.
  41     :param src_ip: Source address of original IPv4 packet.
  42     :param dst_ip: Destination address of original IPv4 packet.
  43     :param sa_in: IPsec SA for packet decryption.
  44     :type pkt_recv: scapy.Ether
  45     :type dst_tun: str
  46     :type src_ip: str
  47     :type dst_ip: str
  48     :type sa_in: scapy.layers.ipsec.SecurityAssociation
  49     :raises RuntimeError: If received packet is invalid.
  50     """
  51     if not pkt_recv.haslayer(IP):
  52         raise RuntimeError(
  53             'Not an IPv4 packet received: {0}'.format(pkt_recv.__repr__()))
  54
  55     if pkt_recv['IP'].dst != dst_tun:
  56         raise RuntimeError(
  57             'Received packet has invalid destination address: {0} '
  58             'should be: {1}'.format(pkt_recv['IP'].dst, dst_tun))
  59
  60     if not pkt_recv.haslayer(ESP):
  61         raise RuntimeError(
  62             'Not an ESP packet received: {0}'.format(pkt_recv.__repr__()))
  63
  64     ip_pkt = pkt_recv[IP]
  65     d_pkt = sa_in.decrypt(ip_pkt)
  66
  67     if d_pkt[IP].dst != dst_ip:
  68         raise RuntimeError(
  69             'Decrypted packet has invalid destination address: {0} '
  70             'should be: {1}'.format(d_pkt['IP'].dst, dst_ip))
  71
  72     if d_pkt[IP].src != src_ip:
  73         raise RuntimeError(
  74             'Decrypted packet has invalid source address: {0} should be: {1}'
  75             .format(d_pkt['IP'].src, src_ip))
  76
  77     if not d_pkt.haslayer(ICMP):
  78         raise RuntimeError(
  79             'Decrypted packet does not have ICMP layer: {0}'.format(
  80                 d_pkt.__repr__()))
  81
  82
  83 def check_ipv6(pkt_recv, dst_tun, src_ip, dst_ip, sa_in):
  84     """Check received IPv6 IPsec packet.
  85
  86     :param pkt_recv: Received packet to verify.
  87     :param dst_tun: IPsec tunnel destination address.
  88     :param src_ip: Source address of original IPv6 packet.
  89     :param dst_ip: Destination address of original IPv6 packet.
  90     :param sa_in: IPsec SA for packet decryption.
  91     :type pkt_recv: scapy.Ether
  92     :type dst_tun: str
  93     :type src_ip: str
  94     :type dst_ip: str
  95     :type sa_in: scapy.layers.ipsec.SecurityAssociation
  96     :raises RuntimeError: If received packet is invalid.
  97     """
  98     if not pkt_recv.haslayer(IPv6):
  99         raise RuntimeError(
 100             'Not an IPv6 packet received: {0}'.format(pkt_recv.__repr__()))
 101
 102     if pkt_recv[IPv6].dst != dst_tun:
 103         raise RuntimeError(
 104             'Received packet has invalid destination address: {0} '
 105             'should be: {1}'.format(pkt_recv['IPv6'].dst, dst_tun))
 106
 107     if not pkt_recv.haslayer(ESP):
 108         raise RuntimeError(
 109             'Not an ESP packet received: {0}'.format(pkt_recv.__repr__()))
 110
 111     ip_pkt = pkt_recv[IPv6]
 112     d_pkt = sa_in.decrypt(ip_pkt)
 113
 114     if d_pkt[IPv6].dst != dst_ip:
 115         raise RuntimeError(
 116             'Decrypted packet has invalid destination address {0}: '
 117             'should be: {1}'.format(d_pkt['IPv6'].dst, dst_ip))
 118
 119     if d_pkt[IPv6].src != src_ip:
 120         raise RuntimeError(
 121             'Decrypted packet has invalid source address: {0} should be: {1}'
 122             .format(d_pkt['IPv6'].src, src_ip))
 123
 124     if not d_pkt.haslayer(ICMPv6EchoReply):
 125         raise RuntimeError(
 126             'Decrypted packet does not have ICMP layer: {0}'.format(
 127                 d_pkt.__repr__()))
 128
 129
 130 # pylint: disable=too-many-locals
 131 # pylint: disable=too-many-statements
 132 def main():
 133     """Send and receive IPsec packet."""
 134     args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip',
 135                              'crypto_alg', 'crypto_key', 'integ_alg',
 136                              'integ_key', 'l_spi', 'r_spi'],
 137                             ['src_tun', 'dst_tun'])
 138
 139     rxq = RxQueue(args.get_arg('rx_if'))
 140     txq = TxQueue(args.get_arg('tx_if'))
 141
 142     src_mac = args.get_arg('src_mac')
 143     dst_mac = args.get_arg('dst_mac')
 144     src_ip = args.get_arg('src_ip')
 145     dst_ip = args.get_arg('dst_ip')
 146     crypto_alg = args.get_arg('crypto_alg')
 147     crypto_key = args.get_arg('crypto_key')
 148     integ_alg = args.get_arg('integ_alg')
 149     integ_key = args.get_arg('integ_key')
 150     l_spi = int(args.get_arg('l_spi'))
 151     r_spi = int(args.get_arg('r_spi'))
 152     src_tun = args.get_arg('src_tun')
 153     dst_tun = args.get_arg('dst_tun')
 154
 155     is_ipv4 = True
 156     if 6 == ip_address(unicode(src_ip)).version:
 157         is_ipv4 = False
 158
 159     tunnel_out = None
 160     tunnel_in = None
 161
 162     if src_tun and dst_tun:
 163         if is_ipv4:
 164             tunnel_out = IP(src=src_tun, dst=dst_tun)
 165             tunnel_in = IP(src=dst_tun, dst=src_tun)
 166         else:
 167             tunnel_out = IPv6(src=src_tun, dst=dst_tun)
 168             tunnel_in = IPv6(src=dst_tun, dst=src_tun)
 169     else:
 170         src_tun = src_ip
 171         dst_tun = dst_ip
 172
 173     sa_in = SecurityAssociation(ESP, spi=r_spi, crypt_algo=crypto_alg,
 174                                 crypt_key=crypto_key, auth_algo=integ_alg,
 175                                 auth_key=integ_key, tunnel_header=tunnel_in)
 176
 177     sa_out = SecurityAssociation(ESP, spi=l_spi, crypt_algo=crypto_alg,
 178                                  crypt_key=crypto_key, auth_algo=integ_alg,
 179                                  auth_key=integ_key, tunnel_header=tunnel_out)
 180
 181     sent_packets = []
 182
 183     if is_ipv4:
 184         ip_pkt = (IP(src=src_ip, dst=dst_ip) /
 185                   ICMP())
 186         ip_pkt = IP(str(ip_pkt))
 187     else:
 188         ip_pkt = (IPv6(src=src_ip, dst=dst_ip) /
 189                   ICMPv6EchoRequest())
 190         ip_pkt = IPv6(str(ip_pkt))
 191
 192     e_pkt = sa_out.encrypt(ip_pkt)
 193     pkt_send = (Ether(src=src_mac, dst=dst_mac) /
 194                 e_pkt)
 195
 196     sent_packets.append(pkt_send)
 197     txq.send(pkt_send)
 198
 199     while True:
 200         pkt_recv = rxq.recv(2, sent_packets)
 201
 202         if pkt_recv is None:
 203             raise RuntimeError('ESP packet Rx timeout')
 204
 205         if pkt_recv.haslayer(ICMPv6ND_NS):
 206             # read another packet in the queue if the current one is ICMPv6ND_NS
 207             continue
 208         else:
 209             # otherwise process the current packet
 210             break
 211
 212     if is_ipv4:
 213         check_ipv4(pkt_recv, src_tun, dst_ip, src_ip, sa_in)
 214     else:
 215         check_ipv6(pkt_recv, src_tun, dst_ip, src_ip, sa_in)
 216
 217     sys.exit(0)
 218
 219
 220 if __name__ == "__main__":
 221     main()