tests: fix assert_nothing_captured
[vpp.git] / test / test_ipsec_default.py
1 import socket
2 import unittest
3
4 from util import ppp
5 from framework import VppTestRunner
6 from template_ipsec import IPSecIPv4Fwd
7
8 """
9 When an IPSec SPD is configured on an interface, any inbound packets
10 not matching inbound policies, or outbound packets not matching outbound
11 policies, must be dropped by default as per RFC4301.
12
13 This test uses simple IPv4 forwarding on interfaces with IPSec enabled
14 to check if packets with no matching rules are dropped by default.
15
16 The basic setup is a single SPD bound to two interfaces, pg0 and pg1.
17
18                     ┌────┐        ┌────┐
19                     │SPD1│        │SPD1│
20                     ├────┤ ─────> ├────┤
21                     │PG0 │        │PG1 │
22                     └────┘        └────┘
23
24 First, both inbound and outbound BYPASS policies are configured allowing
25 traffic to pass from pg0 -> pg1.
26
27 Packets are captured and verified at pg1.
28
29 Then either the inbound or outbound policies are removed and we verify
30 packets are dropped as expected.
31
32 """
33
34
35 class IPSecInboundDefaultDrop(IPSecIPv4Fwd):
36     """ IPSec: inbound packets drop by default with no matching rule """
37
38     def test_ipsec_inbound_default_drop(self):
39         # configure two interfaces and bind the same SPD to both
40         self.create_interfaces(2)
41         self.spd_create_and_intf_add(1, self.pg_interfaces)
42         pkt_count = 5
43
44         # catch-all inbound BYPASS policy, all interfaces
45         inbound_policy = self.spd_add_rem_policy(
46             1, None, None, socket.IPPROTO_UDP, is_out=0, priority=10,
47             policy_type="bypass", all_ips=True)
48
49         # outbound BYPASS policy allowing traffic from pg0->pg1
50         outbound_policy = self.spd_add_rem_policy(
51             1, self.pg0, self.pg1, socket.IPPROTO_UDP,
52             is_out=1, priority=10, policy_type="bypass")
53
54         # create a packet stream pg0->pg1 + add to pg0
55         packets0 = self.create_stream(self.pg0, self.pg1, pkt_count)
56         self.pg0.add_stream(packets0)
57
58         # with inbound BYPASS rule at pg0, we expect to see forwarded
59         # packets on pg1
60         self.pg_interfaces[1].enable_capture()
61         self.pg_start()
62         cap1 = self.pg1.get_capture()
63         for packet in cap1:
64             try:
65                 self.logger.debug(ppp("SPD - Got packet:", packet))
66             except Exception:
67                 self.logger.error(
68                     ppp("Unexpected or invalid packet:", packet))
69                 raise
70         self.logger.debug("SPD: Num packets: %s", len(cap1.res))
71         # verify captures on pg1
72         self.verify_capture(self.pg0, self.pg1, cap1)
73         # verify policies matched correct number of times
74         self.verify_policy_match(pkt_count, inbound_policy)
75         self.verify_policy_match(pkt_count, outbound_policy)
76
77         # remove inbound catch-all BYPASS rule, traffic should now be dropped
78         self.spd_add_rem_policy(  # inbound, all interfaces
79             1, None, None, socket.IPPROTO_UDP, is_out=0, priority=10,
80             policy_type="bypass", all_ips=True, remove=True)
81
82         # create another packet stream pg0->pg1 + add to pg0
83         packets1 = self.create_stream(self.pg0, self.pg1, pkt_count)
84         self.pg0.add_stream(packets1)
85         self.pg_interfaces[1].enable_capture()
86         self.pg_start()
87         # confirm traffic has now been dropped
88         self.pg1.assert_nothing_captured(remark="inbound pkts with no matching"
89                                          "rules NOT dropped by default")
90         # both policies should not have matched any further packets
91         # since we've dropped at input stage
92         self.verify_policy_match(pkt_count, outbound_policy)
93         self.verify_policy_match(pkt_count, inbound_policy)
94
95
96 class IPSecOutboundDefaultDrop(IPSecIPv4Fwd):
97     """ IPSec: outbound packets drop by default with no matching rule """
98
99     def test_ipsec_inbound_default_drop(self):
100         # configure two interfaces and bind the same SPD to both
101         self.create_interfaces(2)
102         self.spd_create_and_intf_add(1, self.pg_interfaces)
103         pkt_count = 5
104
105         # catch-all inbound BYPASS policy, all interfaces
106         inbound_policy = self.spd_add_rem_policy(
107             1, None, None, socket.IPPROTO_UDP, is_out=0, priority=10,
108             policy_type="bypass", all_ips=True)
109
110         # outbound BYPASS policy allowing traffic from pg0->pg1
111         outbound_policy = self.spd_add_rem_policy(
112             1, self.pg0, self.pg1, socket.IPPROTO_UDP,
113             is_out=1, priority=10, policy_type="bypass")
114
115         # create a packet stream pg0->pg1 + add to pg0
116         packets0 = self.create_stream(self.pg0, self.pg1, pkt_count)
117         self.pg0.add_stream(packets0)
118
119         # with outbound BYPASS rule allowing pg0->pg1, we expect to see
120         # forwarded packets on pg1
121         self.pg_interfaces[1].enable_capture()
122         self.pg_start()
123         cap1 = self.pg1.get_capture()
124         for packet in cap1:
125             try:
126                 self.logger.debug(ppp("SPD - Got packet:", packet))
127             except Exception:
128                 self.logger.error(
129                     ppp("Unexpected or invalid packet:", packet))
130                 raise
131         self.logger.debug("SPD: Num packets: %s", len(cap1.res))
132         # verify captures on pg1
133         self.verify_capture(self.pg0, self.pg1, cap1)
134         # verify policies matched correct number of times
135         self.verify_policy_match(pkt_count, inbound_policy)
136         self.verify_policy_match(pkt_count, outbound_policy)
137
138         # remove outbound rule
139         self.spd_add_rem_policy(
140             1, self.pg0, self.pg1, socket.IPPROTO_UDP,
141             is_out=1, priority=10, policy_type="bypass",
142             remove=True)
143
144         # create another packet stream pg0->pg1 + add to pg0
145         packets1 = self.create_stream(self.pg0, self.pg1, pkt_count)
146         self.pg0.add_stream(packets1)
147         self.pg_interfaces[1].enable_capture()
148         self.pg_start()
149         # confirm traffic was dropped and not forwarded
150         self.pg1.assert_nothing_captured(
151             remark="outbound pkts with no matching rules NOT dropped "
152             "by default")
153         # inbound rule should have matched twice the # of pkts now
154         self.verify_policy_match(pkt_count*2, inbound_policy)
155         # as dropped at outbound, outbound policy is the same
156         self.verify_policy_match(pkt_count, outbound_policy)
157
158
159 if __name__ == '__main__':
160     unittest.main(testRunner=VppTestRunner)