resources/traffic_scripts/ipsec.py

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