From 1cfd1559b2b94bb69adb775498a9fd03e11e7510 Mon Sep 17 00:00:00 2001 From: Fan Zhang Date: Thu, 4 Jul 2019 11:42:36 +0100 Subject: [PATCH] examples/ipsec-secgw: support header reconstruction This patch updates the ipsec-secgw application to support header reconstruction. In addition a series of tests have been added to prove the implementation's correctness. Signed-off-by: Fan Zhang Acked-by: Konstantin Ananyev Tested-by: Konstantin Ananyev --- doc/guides/rel_notes/release_19_08.rst | 6 + examples/ipsec-secgw/sa.c | 2 + examples/ipsec-secgw/test/run_test.sh | 3 +- .../test/tun_null_header_reconstruct.py | 479 ++++++++++++++++++ 4 files changed, 489 insertions(+), 1 deletion(-) create mode 100755 examples/ipsec-secgw/test/tun_null_header_reconstruct.py diff --git a/doc/guides/rel_notes/release_19_08.rst b/doc/guides/rel_notes/release_19_08.rst index 6fdb83718b..a01712c9c3 100644 --- a/doc/guides/rel_notes/release_19_08.rst +++ b/doc/guides/rel_notes/release_19_08.rst @@ -167,6 +167,12 @@ New Features handshake between two separate hosts and can share local memory for peer host to directly access. +* **Updated IPSec library Header Reconstruction.** + + Updated the IPSec library with ECN and DSCP field header reconstruction + feature followed by RFC4301. The IPSec-secgw sample application is also + updated to support this feature by default. + * **Updated telemetry library for global metrics support.** Updated ``librte_telemetry`` to fetch the global metrics from the diff --git a/examples/ipsec-secgw/sa.c b/examples/ipsec-secgw/sa.c index c672b4a601..1083915f9a 100644 --- a/examples/ipsec-secgw/sa.c +++ b/examples/ipsec-secgw/sa.c @@ -1063,6 +1063,8 @@ fill_ipsec_sa_prm(struct rte_ipsec_sa_prm *prm, const struct ipsec_sa *ss, prm->ipsec_xform.mode = (IS_TRANSPORT(ss->flags)) ? RTE_SECURITY_IPSEC_SA_MODE_TRANSPORT : RTE_SECURITY_IPSEC_SA_MODE_TUNNEL; + prm->ipsec_xform.options.ecn = 1; + prm->ipsec_xform.options.copy_dscp = 1; if (IS_IP4_TUNNEL(ss->flags)) { prm->ipsec_xform.tunnel.type = RTE_SECURITY_IPSEC_TUNNEL_IPV4; diff --git a/examples/ipsec-secgw/test/run_test.sh b/examples/ipsec-secgw/test/run_test.sh index b8c9fcda5d..8055a4c048 100755 --- a/examples/ipsec-secgw/test/run_test.sh +++ b/examples/ipsec-secgw/test/run_test.sh @@ -64,7 +64,8 @@ trs_3descbc_sha1_old \ trs_3descbc_sha1_esn \ trs_3descbc_sha1_esn_atom" -PKT_TESTS="trs_ipv6opts" +PKT_TESTS="trs_ipv6opts \ +tun_null_header_reconstruct" DIR=$(dirname $0) diff --git a/examples/ipsec-secgw/test/tun_null_header_reconstruct.py b/examples/ipsec-secgw/test/tun_null_header_reconstruct.py new file mode 100755 index 0000000000..d4f42dfc03 --- /dev/null +++ b/examples/ipsec-secgw/test/tun_null_header_reconstruct.py @@ -0,0 +1,479 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: BSD-3-Clause +# Copyright(c) 2019 Intel Corporation + +from scapy.all import * +import unittest +import pkttest + +#{ipv4{ipv4}} test +SRC_ADDR_IPV4_1 = "192.168.1.1" +DST_ADDR_IPV4_1 = "192.168.2.1" + +#{ipv6{ipv6}} test +SRC_ADDR_IPV6_1 = "1111:0000:0000:0000:0000:0000:0000:0001" +DST_ADDR_IPV6_1 = "2222:0000:0000:0000:0000:0000:0000:0001" + +#{ipv4{ipv6}} test +SRC_ADDR_IPV4_2 = "192.168.11.1" +DST_ADDR_IPV4_2 = "192.168.12.1" +SRC_ADDR_IPV6_2 = "1111:0000:0000:0000:0000:0000:0001:0001" +DST_ADDR_IPV6_2 = "2222:0000:0000:0000:0000:0000:0001:0001" + +#{ipv6{ipv4}} test +SRC_ADDR_IPV4_3 = "192.168.21.1" +DST_ADDR_IPV4_3 = "192.168.22.1" +SRC_ADDR_IPV6_3 = "1111:0000:0000:0000:0000:0001:0001:0001" +DST_ADDR_IPV6_3 = "2222:0000:0000:0000:0000:0001:0001:0001" + +def config(): + return """ +#outter-ipv4 inner-ipv4 tunnel mode test +sp ipv4 out esp protect 5 pri 1 \\ +src {0}/32 \\ +dst {1}/32 \\ +sport 0:65535 dport 0:65535 + +sp ipv4 in esp protect 6 pri 1 \\ +src {1}/32 \\ +dst {0}/32 \\ +sport 0:65535 dport 0:65535 + +sa out 5 cipher_algo null auth_algo null mode ipv4-tunnel \\ +src {0} dst {1} +sa in 6 cipher_algo null auth_algo null mode ipv4-tunnel \\ +src {1} dst {0} + +rt ipv4 dst {0}/32 port 1 +rt ipv4 dst {1}/32 port 0 + +#outter-ipv6 inner-ipv6 tunnel mode test +sp ipv6 out esp protect 7 pri 1 \\ +src {2}/128 \\ +dst {3}/128 \\ +sport 0:65535 dport 0:65535 + +sp ipv6 in esp protect 8 pri 1 \\ +src {3}/128 \\ +dst {2}/128 \\ +sport 0:65535 dport 0:65535 + +sa out 7 cipher_algo null auth_algo null mode ipv6-tunnel \\ +src {2} dst {3} +sa in 8 cipher_algo null auth_algo null mode ipv6-tunnel \\ +src {3} dst {2} + +rt ipv6 dst {2}/128 port 1 +rt ipv6 dst {3}/128 port 0 + +#outter-ipv4 inner-ipv6 tunnel mode test +sp ipv6 out esp protect 9 pri 1 \\ +src {4}/128 \\ +dst {5}/128 \\ +sport 0:65535 dport 0:65535 + +sp ipv6 in esp protect 10 pri 1 \\ +src {5}/128 \\ +dst {4}/128 \\ +sport 0:65535 dport 0:65535 + +sa out 9 cipher_algo null auth_algo null mode ipv4-tunnel \\ +src {6} dst {7} +sa in 10 cipher_algo null auth_algo null mode ipv4-tunnel \\ +src {7} dst {6} + +rt ipv6 dst {4}/128 port 1 +rt ipv4 dst {7}/32 port 0 + +#outter-ipv6 inner-ipv4 tunnel mode test +sp ipv4 out esp protect 11 pri 1 \\ +src {8}/32 \\ +dst {9}/32 \\ +sport 0:65535 dport 0:65535 + +sp ipv4 in esp protect 12 pri 1 \\ +src {9}/32 \\ +dst {8}/32 \\ +sport 0:65535 dport 0:65535 + +sa out 11 cipher_algo null auth_algo null mode ipv6-tunnel \\ +src {10} dst {11} +sa in 12 cipher_algo null auth_algo null mode ipv6-tunnel \\ +src {11} dst {10} + +rt ipv4 dst {8}/32 port 1 +rt ipv6 dst {11}/128 port 0 +""".format(SRC_ADDR_IPV4_1, DST_ADDR_IPV4_1, + SRC_ADDR_IPV6_1, DST_ADDR_IPV6_1, + SRC_ADDR_IPV6_2, DST_ADDR_IPV6_2, SRC_ADDR_IPV4_2, DST_ADDR_IPV4_2, + SRC_ADDR_IPV4_3, DST_ADDR_IPV4_3, SRC_ADDR_IPV6_3, DST_ADDR_IPV6_3) + +ECN_ECT0 = 0x02 +ECN_ECT1 = 0x01 +ECN_CE = 0x03 +DSCP_1 = 0x04 +DSCP_3F = 0xFC + +class TestTunnelHeaderReconstruct(unittest.TestCase): + def setUp(self): + self.px = pkttest.PacketXfer() + th = IP(src=DST_ADDR_IPV4_1, dst=SRC_ADDR_IPV4_1) + self.sa_ipv4v4 = SecurityAssociation(ESP, spi=6, tunnel_header = th) + + th = IPv6(src=DST_ADDR_IPV6_1, dst=SRC_ADDR_IPV6_1) + self.sa_ipv6v6 = SecurityAssociation(ESP, spi=8, tunnel_header = th) + + th = IP(src=DST_ADDR_IPV4_2, dst=SRC_ADDR_IPV4_2) + self.sa_ipv4v6 = SecurityAssociation(ESP, spi=10, tunnel_header = th) + + th = IPv6(src=DST_ADDR_IPV6_3, dst=SRC_ADDR_IPV6_3) + self.sa_ipv6v4 = SecurityAssociation(ESP, spi=12, tunnel_header = th) + + def gen_pkt_plain_ipv4(self, src, dst, tos): + pkt = IP(src=src, dst=dst, tos=tos) + pkt /= UDP(sport=123,dport=456)/Raw(load="abc") + return pkt + + def gen_pkt_plain_ipv6(self, src, dst, tc): + pkt = IPv6(src=src, dst=dst, tc=tc) + pkt /= UDP(sport=123,dport=456)/Raw(load="abc") + return pkt + + def gen_pkt_tun_ipv4v4(self, tos_outter, tos_inner): + pkt = self.gen_pkt_plain_ipv4(DST_ADDR_IPV4_1, SRC_ADDR_IPV4_1, + tos_inner) + pkt = self.sa_ipv4v4.encrypt(pkt) + self.assertEqual(pkt[IP].proto, socket.IPPROTO_ESP) + self.assertEqual(pkt[ESP].spi, 6) + pkt[IP].tos = tos_outter + return pkt + + def gen_pkt_tun_ipv6v6(self, tc_outter, tc_inner): + pkt = self.gen_pkt_plain_ipv6(DST_ADDR_IPV6_1, SRC_ADDR_IPV6_1, + tc_inner) + pkt = self.sa_ipv6v6.encrypt(pkt) + self.assertEqual(pkt[IPv6].nh, socket.IPPROTO_ESP) + self.assertEqual(pkt[ESP].spi, 8) + pkt[IPv6].tc = tc_outter + return pkt + + def gen_pkt_tun_ipv4v6(self, tos_outter, tc_inner): + pkt = self.gen_pkt_plain_ipv6(DST_ADDR_IPV6_2, SRC_ADDR_IPV6_2, + tc_inner) + pkt = self.sa_ipv4v6.encrypt(pkt) + self.assertEqual(pkt[IP].proto, socket.IPPROTO_ESP) + self.assertEqual(pkt[ESP].spi, 10) + pkt[IP].tos = tos_outter + return pkt + + def gen_pkt_tun_ipv6v4(self, tc_outter, tos_inner): + pkt = self.gen_pkt_plain_ipv4(DST_ADDR_IPV4_3, SRC_ADDR_IPV4_3, + tos_inner) + pkt = self.sa_ipv6v4.encrypt(pkt) + self.assertEqual(pkt[IPv6].nh, socket.IPPROTO_ESP) + self.assertEqual(pkt[ESP].spi, 12) + pkt[IPv6].tc = tc_outter + return pkt + +#RFC4301 5.1.2.1 & 5.1.2.2, outbound packets shall be copied ECN field + def test_outb_ipv4v4_ecn(self): + pkt = self.gen_pkt_plain_ipv4(SRC_ADDR_IPV4_1, DST_ADDR_IPV4_1, + ECN_ECT1) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 5) + self.assertEqual(resp[IP].tos, ECN_ECT1) + + pkt = self.gen_pkt_plain_ipv4(SRC_ADDR_IPV4_1, DST_ADDR_IPV4_1, + ECN_ECT0) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 5) + self.assertEqual(resp[IP].tos, ECN_ECT0) + + pkt = self.gen_pkt_plain_ipv4(SRC_ADDR_IPV4_1, DST_ADDR_IPV4_1, + ECN_CE) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 5) + self.assertEqual(resp[IP].tos, ECN_CE) + + def test_outb_ipv6v6_ecn(self): + pkt = self.gen_pkt_plain_ipv6(SRC_ADDR_IPV6_1, DST_ADDR_IPV6_1, + ECN_ECT1) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_ESP) + self.assertEqual(resp[IPv6].tc, ECN_ECT1) + + pkt = self.gen_pkt_plain_ipv6(SRC_ADDR_IPV6_1, DST_ADDR_IPV6_1, + ECN_ECT0) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 7) + self.assertEqual(resp[IPv6].tc, ECN_ECT0) + + pkt = self.gen_pkt_plain_ipv6(SRC_ADDR_IPV6_1, DST_ADDR_IPV6_1, + ECN_CE) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 7) + self.assertEqual(resp[IPv6].tc, ECN_CE) + + def test_outb_ipv4v6_ecn(self): + pkt = self.gen_pkt_plain_ipv6(SRC_ADDR_IPV6_2, DST_ADDR_IPV6_2, + ECN_ECT1) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_ESP) + self.assertEqual(resp[IP].tos, ECN_ECT1) + + pkt = self.gen_pkt_plain_ipv6(SRC_ADDR_IPV6_2, DST_ADDR_IPV6_2, + ECN_ECT0) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_ESP) + self.assertEqual(resp[IP].tos, ECN_ECT0) + + pkt = self.gen_pkt_plain_ipv6(SRC_ADDR_IPV6_2, DST_ADDR_IPV6_2, + ECN_CE) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_ESP) + self.assertEqual(resp[IP].tos, ECN_CE) + + def test_outb_ipv6v4_ecn(self): + pkt = self.gen_pkt_plain_ipv4(SRC_ADDR_IPV4_3, DST_ADDR_IPV4_3, + ECN_ECT1) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_ESP) + self.assertEqual(resp[IPv6].tc, ECN_ECT1) + + pkt = self.gen_pkt_plain_ipv4(SRC_ADDR_IPV4_3, DST_ADDR_IPV4_3, + ECN_ECT0) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_ESP) + self.assertEqual(resp[IPv6].tc, ECN_ECT0) + + pkt = self.gen_pkt_plain_ipv4(SRC_ADDR_IPV4_3, DST_ADDR_IPV4_3, + ECN_CE) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_ESP) + self.assertEqual(resp[IPv6].tc, ECN_CE) + +#RFC4301 5.1.2.1 & 5.1.2.2, if outbound packets ECN is CE (0x3), inbound packets +#ECN is overwritten to CE, otherwise no change + +#Outter header not CE, Inner header should be no change + def test_inb_ipv4v4_ecn_inner_no_change(self): + pkt = self.gen_pkt_tun_ipv4v4(ECN_ECT1, ECN_ECT0) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, ECN_ECT0) + + pkt = self.gen_pkt_tun_ipv4v4(ECN_ECT0, ECN_ECT1) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, ECN_ECT1) + + pkt = self.gen_pkt_tun_ipv4v4(ECN_ECT1, ECN_CE) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, ECN_CE) + + def test_inb_ipv6v6_ecn_inner_no_change(self): + pkt = self.gen_pkt_tun_ipv6v6(ECN_ECT1, ECN_ECT0) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, ECN_ECT0) + + pkt = self.gen_pkt_tun_ipv6v6(ECN_ECT0, ECN_ECT1) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, ECN_ECT1) + + pkt = self.gen_pkt_tun_ipv6v6(ECN_ECT1, ECN_CE) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, ECN_CE) + + def test_inb_ipv4v6_ecn_inner_no_change(self): + pkt = self.gen_pkt_tun_ipv4v6(ECN_ECT1, ECN_ECT0) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, ECN_ECT0) + + pkt = self.gen_pkt_tun_ipv4v6(ECN_ECT0, ECN_ECT1) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, ECN_ECT1) + + pkt = self.gen_pkt_tun_ipv4v6(ECN_ECT1, ECN_CE) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, ECN_CE) + + def test_inb_ipv6v4_ecn_inner_no_change(self): + pkt = self.gen_pkt_tun_ipv6v4(ECN_ECT1, ECN_ECT0) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, ECN_ECT0) + + pkt = self.gen_pkt_tun_ipv6v4(ECN_ECT0, ECN_ECT1) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, ECN_ECT1) + + pkt = self.gen_pkt_tun_ipv6v4(ECN_ECT1, ECN_CE) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, ECN_CE) + +#Outter header CE, Inner header should be changed to CE + def test_inb_ipv4v4_ecn_inner_change(self): + pkt = self.gen_pkt_tun_ipv4v4(ECN_CE, ECN_ECT0) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, ECN_CE) + + pkt = self.gen_pkt_tun_ipv4v4(ECN_CE, ECN_ECT1) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, ECN_CE) + + def test_inb_ipv6v6_ecn_inner_change(self): + pkt = self.gen_pkt_tun_ipv6v6(ECN_CE, ECN_ECT0) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, ECN_CE) + + pkt = self.gen_pkt_tun_ipv6v6(ECN_CE, ECN_ECT1) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, ECN_CE) + + def test_inb_ipv4v6_ecn_inner_change(self): + pkt = self.gen_pkt_tun_ipv4v6(ECN_CE, ECN_ECT0) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, ECN_CE) + + pkt = self.gen_pkt_tun_ipv4v6(ECN_CE, ECN_ECT1) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, ECN_CE) + + def test_inb_ipv6v4_ecn_inner_change(self): + pkt = self.gen_pkt_tun_ipv6v4(ECN_CE, ECN_ECT0) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, ECN_CE) + + pkt = self.gen_pkt_tun_ipv6v4(ECN_CE, ECN_ECT1) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, ECN_CE) + +#RFC4301 5.1.2.1.5 Outer DS field should be copied from Inner DS field + def test_outb_ipv4v4_dscp(self): + pkt = self.gen_pkt_plain_ipv4(SRC_ADDR_IPV4_1, DST_ADDR_IPV4_1, + DSCP_1) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 5) + self.assertEqual(resp[IP].tos, DSCP_1) + + pkt = self.gen_pkt_plain_ipv4(SRC_ADDR_IPV4_1, DST_ADDR_IPV4_1, + DSCP_3F) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 5) + self.assertEqual(resp[IP].tos, DSCP_3F) + + def test_outb_ipv6v6_dscp(self): + pkt = self.gen_pkt_plain_ipv6(SRC_ADDR_IPV6_1, DST_ADDR_IPV6_1, + DSCP_1) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 7) + self.assertEqual(resp[IPv6].tc, DSCP_1) + + pkt = self.gen_pkt_plain_ipv6(SRC_ADDR_IPV6_1, DST_ADDR_IPV6_1, + DSCP_3F) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 7) + self.assertEqual(resp[IPv6].tc, DSCP_3F) + + def test_outb_ipv4v6_dscp(self): + pkt = self.gen_pkt_plain_ipv6(SRC_ADDR_IPV6_2, DST_ADDR_IPV6_2, + DSCP_1) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 9) + self.assertEqual(resp[IP].tos, DSCP_1) + + pkt = self.gen_pkt_plain_ipv6(SRC_ADDR_IPV6_2, DST_ADDR_IPV6_2, + DSCP_3F) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 9) + self.assertEqual(resp[IP].tos, DSCP_3F) + + def test_outb_ipv6v4_dscp(self): + pkt = self.gen_pkt_plain_ipv4(SRC_ADDR_IPV4_3, DST_ADDR_IPV4_3, + DSCP_1) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 11) + self.assertEqual(resp[IPv6].tc, DSCP_1) + + pkt = self.gen_pkt_plain_ipv4(SRC_ADDR_IPV4_3, DST_ADDR_IPV4_3, + DSCP_3F) + resp = self.px.xfer_unprotected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_ESP) + self.assertEqual(resp[ESP].spi, 11) + self.assertEqual(resp[IPv6].tc, DSCP_3F) + +#RFC4301 5.1.2.1.5 Inner DS field should not be affected by Outer DS field + def test_inb_ipv4v4_dscp(self): + pkt = self.gen_pkt_tun_ipv4v4(DSCP_3F, DSCP_1) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, DSCP_1) + + pkt = self.gen_pkt_tun_ipv4v4(DSCP_1, DSCP_3F) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, DSCP_3F) + + def test_inb_ipv6v6_dscp(self): + pkt = self.gen_pkt_tun_ipv6v6(DSCP_3F, DSCP_1) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, DSCP_1) + + pkt = self.gen_pkt_tun_ipv6v6(DSCP_1, DSCP_3F) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, DSCP_3F) + + def test_inb_ipv4v6_dscp(self): + pkt = self.gen_pkt_tun_ipv4v6(DSCP_3F, DSCP_1) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, DSCP_1) + + pkt = self.gen_pkt_tun_ipv4v6(DSCP_1, DSCP_3F) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IPv6].nh, socket.IPPROTO_UDP) + self.assertEqual(resp[IPv6].tc, DSCP_3F) + + def test_inb_ipv6v4_dscp(self): + pkt = self.gen_pkt_tun_ipv6v4(DSCP_3F, DSCP_1) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, DSCP_1) + + pkt = self.gen_pkt_tun_ipv6v4(DSCP_1, DSCP_3F) + resp = self.px.xfer_protected(pkt) + self.assertEqual(resp[IP].proto, socket.IPPROTO_UDP) + self.assertEqual(resp[IP].tos, DSCP_3F) + +pkttest.pkttest() -- 2.20.1