test/template_ipsec.py

   1 import unittest
   2 import socket
   3
   4 from scapy.layers.inet import IP, ICMP, TCP
   5 from scapy.layers.ipsec import SecurityAssociation
   6 from scapy.layers.l2 import Ether, Raw
   7 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest
   8
   9 from framework import VppTestCase, VppTestRunner
  10 from util import ppp
  11
  12
  13 class IPsecIPv4Params(object):
  14     addr_type = socket.AF_INET
  15     addr_any = "0.0.0.0"
  16     addr_bcast = "255.255.255.255"
  17     addr_len = 32
  18     is_ipv6 = 0
  19     remote_tun_if_host = '1.1.1.1'
  20
  21     scapy_tun_sa_id = 10
  22     scapy_tun_spi = 1001
  23     vpp_tun_sa_id = 20
  24     vpp_tun_spi = 1000
  25
  26     scapy_tra_sa_id = 30
  27     scapy_tra_spi = 2001
  28     vpp_tra_sa_id = 40
  29     vpp_tra_spi = 2000
  30
  31     auth_algo_vpp_id = 2  # internal VPP enum value for SHA1_96
  32     auth_algo = 'HMAC-SHA1-96'  # scapy name
  33     auth_key = 'C91KUR9GYMm5GfkEvNjX'
  34
  35     crypt_algo_vpp_id = 1  # internal VPP enum value for AES_CBC_128
  36     crypt_algo = 'AES-CBC'  # scapy name
  37     crypt_key = 'JPjyOWBeVEQiMe7h'
  38
  39
  40 class IPsecIPv6Params(object):
  41     addr_type = socket.AF_INET6
  42     addr_any = "0::0"
  43     addr_bcast = "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"
  44     addr_len = 128
  45     is_ipv6 = 1
  46     remote_tun_if_host = '1111:1111:1111:1111:1111:1111:1111:1111'
  47
  48     scapy_tun_sa_id = 50
  49     scapy_tun_spi = 3001
  50     vpp_tun_sa_id = 60
  51     vpp_tun_spi = 3000
  52
  53     scapy_tra_sa_id = 70
  54     scapy_tra_spi = 4001
  55     vpp_tra_sa_id = 80
  56     vpp_tra_spi = 4000
  57
  58     auth_algo_vpp_id = 4  # internal VPP enum value for SHA_256_128
  59     auth_algo = 'SHA2-256-128'  # scapy name
  60     auth_key = 'C91KUR9GYMm5GfkEvNjX'
  61
  62     crypt_algo_vpp_id = 3  # internal VPP enum value for AES_CBC_256
  63     crypt_algo = 'AES-CBC'  # scapy name
  64     crypt_key = 'JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h'
  65
  66
  67 class TemplateIpsec(VppTestCase):
  68     """
  69     TRANSPORT MODE:
  70
  71      ------   encrypt   ---
  72     |tra_if| <-------> |VPP|
  73      ------   decrypt   ---
  74
  75     TUNNEL MODE:
  76
  77      ------   encrypt   ---   plain   ---
  78     |tun_if| <-------  |VPP| <------ |pg1|
  79      ------             ---           ---
  80
  81      ------   decrypt   ---   plain   ---
  82     |tun_if| ------->  |VPP| ------> |pg1|
  83      ------             ---           ---
  84     """
  85
  86     def ipsec_select_backend(self):
  87         """ empty method to be overloaded when necessary """
  88         pass
  89
  90     def setUp(self):
  91         super(TemplateIpsec, self).setUp()
  92
  93         self.ipv4_params = IPsecIPv4Params()
  94         self.ipv6_params = IPsecIPv6Params()
  95         self.params = {self.ipv4_params.addr_type: self.ipv4_params,
  96                        self.ipv6_params.addr_type: self.ipv6_params}
  97
  98         self.payload = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"\
  99                        "XXXXXXXXXXXXXXXXXXXXX"
 100
 101         self.tun_spd_id = 1
 102         self.tra_spd_id = 2
 103
 104         self.vpp_esp_protocol = 1
 105         self.vpp_ah_protocol = 0
 106
 107         self.create_pg_interfaces(range(3))
 108         self.interfaces = list(self.pg_interfaces)
 109         for i in self.interfaces:
 110             i.admin_up()
 111             i.config_ip4()
 112             i.resolve_arp()
 113             i.config_ip6()
 114             i.resolve_ndp()
 115         self.ipsec_select_backend()
 116
 117     def tearDown(self):
 118         super(TemplateIpsec, self).tearDown()
 119
 120         for i in self.interfaces:
 121             i.admin_down()
 122             i.unconfig_ip4()
 123             i.unconfig_ip6()
 124
 125         if not self.vpp_dead:
 126             self.vapi.cli("show hardware")
 127
 128     def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1):
 129         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 130                 sa.encrypt(IP(src=src, dst=dst) / ICMP() / self.payload)
 131                 for i in range(count)]
 132
 133     def gen_encrypt_pkts6(self, sa, sw_intf, src, dst, count=1):
 134         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 135                 sa.encrypt(IPv6(src=src, dst=dst) /
 136                            ICMPv6EchoRequest(id=0, seq=1, data=self.payload))
 137                 for i in range(count)]
 138
 139     def gen_pkts(self, sw_intf, src, dst, count=1):
 140         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 141                 IP(src=src, dst=dst) / ICMP() / self.payload
 142                 for i in range(count)]
 143
 144     def gen_pkts6(self, sw_intf, src, dst, count=1):
 145         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 146                 IPv6(src=src, dst=dst) /
 147                 ICMPv6EchoRequest(id=0, seq=1, data=self.payload)
 148                 for i in range(count)]
 149
 150     def configure_sa_tun(self, params):
 151         ip_class_by_addr_type = {socket.AF_INET: IP, socket.AF_INET6: IPv6}
 152         scapy_tun_sa = SecurityAssociation(
 153             self.encryption_type, spi=params.vpp_tun_spi,
 154             crypt_algo=params.crypt_algo, crypt_key=params.crypt_key,
 155             auth_algo=params.auth_algo, auth_key=params.auth_key,
 156             tunnel_header=ip_class_by_addr_type[params.addr_type](
 157                 src=self.tun_if.remote_addr[params.addr_type],
 158                 dst=self.tun_if.local_addr[params.addr_type]))
 159         vpp_tun_sa = SecurityAssociation(
 160             self.encryption_type, spi=params.scapy_tun_spi,
 161             crypt_algo=params.crypt_algo, crypt_key=params.crypt_key,
 162             auth_algo=params.auth_algo, auth_key=params.auth_key,
 163             tunnel_header=ip_class_by_addr_type[params.addr_type](
 164                 dst=self.tun_if.remote_addr[params.addr_type],
 165                 src=self.tun_if.local_addr[params.addr_type]))
 166         return vpp_tun_sa, scapy_tun_sa
 167
 168     def configure_sa_tra(self, params):
 169         params.scapy_tra_sa = SecurityAssociation(self.encryption_type,
 170                                                   spi=params.vpp_tra_spi,
 171                                                   crypt_algo=params.crypt_algo,
 172                                                   crypt_key=params.crypt_key,
 173                                                   auth_algo=params.auth_algo,
 174                                                   auth_key=params.auth_key)
 175         params.vpp_tra_sa = SecurityAssociation(self.encryption_type,
 176                                                 spi=params.scapy_tra_spi,
 177                                                 crypt_algo=params.crypt_algo,
 178                                                 crypt_key=params.crypt_key,
 179                                                 auth_algo=params.auth_algo,
 180                                                 auth_key=params.auth_key)
 181
 182
 183 class IpsecTcpTests(object):
 184     def test_tcp_checksum(self):
 185         """ verify checksum correctness for vpp generated packets """
 186         self.vapi.cli("test http server")
 187         p = self.params[socket.AF_INET]
 188         vpp_tun_sa, scapy_tun_sa = self.configure_sa_tun(p)
 189         send = (Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac) /
 190                 scapy_tun_sa.encrypt(IP(src=p.remote_tun_if_host,
 191                                         dst=self.tun_if.local_ip4) /
 192                                      TCP(flags='S', dport=80)))
 193         self.logger.debug(ppp("Sending packet:", send))
 194         recv = self.send_and_expect(self.tun_if, [send], self.tun_if)
 195         recv = recv[0]
 196         decrypted = vpp_tun_sa.decrypt(recv[IP])
 197         self.assert_packet_checksums_valid(decrypted)
 198
 199
 200 class IpsecTraTests(object):
 201     def test_tra_anti_replay(self, count=1):
 202         """ ipsec v4 transport anti-reply test """
 203         p = self.params[socket.AF_INET]
 204
 205         # fire in a packet with seq number 1
 206         pkt = (Ether(src=self.tra_if.remote_mac,
 207                      dst=self.tra_if.local_mac) /
 208                p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 209                                          dst=self.tra_if.local_ip4) /
 210                                       ICMP(),
 211                                       seq_num=1))
 212         recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 213
 214         # now move the window over to 235
 215         pkt = (Ether(src=self.tra_if.remote_mac,
 216                      dst=self.tra_if.local_mac) /
 217                p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 218                                          dst=self.tra_if.local_ip4) /
 219                                       ICMP(),
 220                                       seq_num=235))
 221         recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 222
 223         # the window size is 64 packets
 224         # in window are still accepted
 225         pkt = (Ether(src=self.tra_if.remote_mac,
 226                      dst=self.tra_if.local_mac) /
 227                p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 228                                          dst=self.tra_if.local_ip4) /
 229                                       ICMP(),
 230                                       seq_num=172))
 231         recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 232
 233         # out of window are dropped
 234         pkt = (Ether(src=self.tra_if.remote_mac,
 235                      dst=self.tra_if.local_mac) /
 236                p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 237                                          dst=self.tra_if.local_ip4) /
 238                                       ICMP(),
 239                                       seq_num=17))
 240         self.send_and_assert_no_replies(self.tra_if, pkt * 17)
 241
 242         self.assert_packet_counter_equal(
 243             '/err/%s/SA replayed packet' % self.tra4_decrypt_node_name, 17)
 244
 245         # a packet that does not decrypt does not move the window forward
 246         bogus_sa = SecurityAssociation(self.encryption_type,
 247                                        p.vpp_tra_spi)
 248         pkt = (Ether(src=self.tra_if.remote_mac,
 249                      dst=self.tra_if.local_mac) /
 250                bogus_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 251                                    dst=self.tra_if.local_ip4) /
 252                                 ICMP(),
 253                                 seq_num=350))
 254         self.send_and_assert_no_replies(self.tra_if, pkt * 17)
 255
 256         self.assert_packet_counter_equal(
 257             '/err/%s/Integrity check failed' % self.tra4_decrypt_node_name, 17)
 258
 259         # which we can determine since this packet is still in the window
 260         pkt = (Ether(src=self.tra_if.remote_mac,
 261                      dst=self.tra_if.local_mac) /
 262                p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 263                                          dst=self.tra_if.local_ip4) /
 264                                       ICMP(),
 265                                       seq_num=234))
 266         self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 267
 268         # move the security-associations seq number on to the last we used
 269         p.scapy_tra_sa.seq_num = 351
 270         p.vpp_tra_sa.seq_num = 351
 271
 272     def test_tra_basic(self, count=1):
 273         """ ipsec v4 transport basic test """
 274         self.vapi.cli("clear errors")
 275         try:
 276             p = self.params[socket.AF_INET]
 277             send_pkts = self.gen_encrypt_pkts(p.scapy_tra_sa, self.tra_if,
 278                                               src=self.tra_if.remote_ip4,
 279                                               dst=self.tra_if.local_ip4,
 280                                               count=count)
 281             recv_pkts = self.send_and_expect(self.tra_if, send_pkts,
 282                                              self.tra_if)
 283             for rx in recv_pkts:
 284                 try:
 285                     decrypted = p.vpp_tra_sa.decrypt(rx[IP])
 286                     self.assert_packet_checksums_valid(decrypted)
 287                 except:
 288                     self.logger.debug(ppp("Unexpected packet:", rx))
 289                     raise
 290         finally:
 291             self.logger.info(self.vapi.ppcli("show error"))
 292             self.logger.info(self.vapi.ppcli("show ipsec"))
 293
 294         self.assert_packet_counter_equal(self.tra4_encrypt_node_name, count)
 295         self.assert_packet_counter_equal(self.tra4_decrypt_node_name, count)
 296
 297     def test_tra_burst(self):
 298         """ ipsec v4 transport burst test """
 299         self.test_tra_basic(count=257)
 300
 301     def test_tra_basic6(self, count=1):
 302         """ ipsec v6 transport basic test """
 303         self.vapi.cli("clear errors")
 304         try:
 305             p = self.params[socket.AF_INET6]
 306             send_pkts = self.gen_encrypt_pkts6(p.scapy_tra_sa, self.tra_if,
 307                                                src=self.tra_if.remote_ip6,
 308                                                dst=self.tra_if.local_ip6,
 309                                                count=count)
 310             recv_pkts = self.send_and_expect(self.tra_if, send_pkts,
 311                                              self.tra_if)
 312             for rx in recv_pkts:
 313                 try:
 314                     decrypted = p.vpp_tra_sa.decrypt(rx[IPv6])
 315                     self.assert_packet_checksums_valid(decrypted)
 316                 except:
 317                     self.logger.debug(ppp("Unexpected packet:", rx))
 318                     raise
 319         finally:
 320             self.logger.info(self.vapi.ppcli("show error"))
 321             self.logger.info(self.vapi.ppcli("show ipsec"))
 322
 323         self.assert_packet_counter_equal(self.tra6_encrypt_node_name, count)
 324         self.assert_packet_counter_equal(self.tra6_decrypt_node_name, count)
 325
 326     def test_tra_burst6(self):
 327         """ ipsec v6 transport burst test """
 328         self.test_tra_basic6(count=257)
 329
 330
 331 class IpsecTun4Tests(object):
 332     def test_tun_basic44(self, count=1):
 333         """ ipsec 4o4 tunnel basic test """
 334         self.vapi.cli("clear errors")
 335         try:
 336             p = self.params[socket.AF_INET]
 337             vpp_tun_sa, scapy_tun_sa = self.configure_sa_tun(p)
 338             send_pkts = self.gen_encrypt_pkts(scapy_tun_sa, self.tun_if,
 339                                               src=p.remote_tun_if_host,
 340                                               dst=self.pg1.remote_ip4,
 341                                               count=count)
 342             recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
 343             for recv_pkt in recv_pkts:
 344                 self.assert_equal(recv_pkt[IP].src, p.remote_tun_if_host)
 345                 self.assert_equal(recv_pkt[IP].dst, self.pg1.remote_ip4)
 346                 self.assert_packet_checksums_valid(recv_pkt)
 347             send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
 348                                       dst=p.remote_tun_if_host, count=count)
 349             recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
 350             for recv_pkt in recv_pkts:
 351                 try:
 352                     decrypt_pkt = vpp_tun_sa.decrypt(recv_pkt[IP])
 353                     if not decrypt_pkt.haslayer(IP):
 354                         decrypt_pkt = IP(decrypt_pkt[Raw].load)
 355                     self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
 356                     self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
 357                     self.assert_packet_checksums_valid(decrypt_pkt)
 358                 except:
 359                     self.logger.debug(ppp("Unexpected packet:", recv_pkt))
 360                     try:
 361                         self.logger.debug(
 362                             ppp("Decrypted packet:", decrypt_pkt))
 363                     except:
 364                         pass
 365                     raise
 366         finally:
 367             self.logger.info(self.vapi.ppcli("show error"))
 368             self.logger.info(self.vapi.ppcli("show ipsec"))
 369
 370         self.assert_packet_counter_equal(self.tun4_encrypt_node_name, count)
 371         self.assert_packet_counter_equal(self.tun4_decrypt_node_name, count)
 372
 373     def test_tun_burst44(self):
 374         """ ipsec 4o4 tunnel burst test """
 375         self.test_tun_basic44(count=257)
 376
 377
 378 class IpsecTun6Tests(object):
 379     def test_tun_basic66(self, count=1):
 380         """ ipsec 6o6 tunnel basic test """
 381         self.vapi.cli("clear errors")
 382         try:
 383             p = self.params[socket.AF_INET6]
 384             vpp_tun_sa, scapy_tun_sa = self.configure_sa_tun(p)
 385             send_pkts = self.gen_encrypt_pkts6(scapy_tun_sa, self.tun_if,
 386                                                src=p.remote_tun_if_host,
 387                                                dst=self.pg1.remote_ip6,
 388                                                count=count)
 389             recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
 390             for recv_pkt in recv_pkts:
 391                 self.assert_equal(recv_pkt[IPv6].src, p.remote_tun_if_host)
 392                 self.assert_equal(recv_pkt[IPv6].dst, self.pg1.remote_ip6)
 393                 self.assert_packet_checksums_valid(recv_pkt)
 394             send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6,
 395                                        dst=p.remote_tun_if_host,
 396                                        count=count)
 397             recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
 398             for recv_pkt in recv_pkts:
 399                 try:
 400                     decrypt_pkt = vpp_tun_sa.decrypt(recv_pkt[IPv6])
 401                     if not decrypt_pkt.haslayer(IPv6):
 402                         decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
 403                     self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
 404                     self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
 405                     self.assert_packet_checksums_valid(decrypt_pkt)
 406                 except:
 407                     self.logger.debug(ppp("Unexpected packet:", recv_pkt))
 408                     try:
 409                         self.logger.debug(
 410                             ppp("Decrypted packet:", decrypt_pkt))
 411                     except:
 412                         pass
 413                     raise
 414         finally:
 415             self.logger.info(self.vapi.ppcli("show error"))
 416             self.logger.info(self.vapi.ppcli("show ipsec"))
 417
 418         self.assert_packet_counter_equal(self.tun6_encrypt_node_name, count)
 419         self.assert_packet_counter_equal(self.tun6_decrypt_node_name, count)
 420
 421     def test_tun_burst66(self):
 422         """ ipsec 6o6 tunnel burst test """
 423         self.test_tun_basic66(count=257)
 424
 425
 426 class IpsecTunTests(IpsecTun4Tests, IpsecTun6Tests):
 427     pass
 428
 429
 430 if __name__ == '__main__':
 431     unittest.main(testRunner=VppTestRunner)