1# 2# SPDX-License-Identifier: BSD-2-Clause 3# 4# Copyright (c) 2024 Rubicon Communications, LLC (Netgate) 5# 6# Redistribution and use in source and binary forms, with or without 7# modification, are permitted provided that the following conditions 8# are met: 9# 1. Redistributions of source code must retain the above copyright 10# notice, this list of conditions and the following disclaimer. 11# 2. Redistributions in binary form must reproduce the above copyright 12# notice, this list of conditions and the following disclaimer in the 13# documentation and/or other materials provided with the distribution. 14# 15# THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND 16# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 17# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 18# ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE 19# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 20# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS 21# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) 22# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 23# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY 24# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF 25# SUCH DAMAGE. 26 27import pytest 28import subprocess 29import re 30from atf_python.sys.net.tools import ToolsHelper 31from atf_python.sys.net.vnet import VnetTestTemplate 32 33def check(cmd): 34 ps = subprocess.Popen(cmd, shell=True) 35 ret = ps.wait() 36 if ret != 0: 37 raise Exception("Command %s returned %d" % (cmd, ret)) 38 39class TestICMP(VnetTestTemplate): 40 REQUIRED_MODULES = [ "pf" ] 41 TOPOLOGY = { 42 "vnet1": {"ifaces": ["if1"]}, 43 "vnet2": {"ifaces": ["if1", "if2"]}, 44 "vnet3": {"ifaces": ["if2"]}, 45 "if1": {"prefixes4": [("192.0.2.2/24", "192.0.2.1/24")]}, 46 "if2": {"prefixes4": [("198.51.100.1/24", "198.51.100.2/24")]}, 47 } 48 49 def vnet2_handler(self, vnet): 50 ifname = vnet.iface_alias_map["if1"].name 51 if2name = vnet.iface_alias_map["if2"].name 52 53 ToolsHelper.print_output("/sbin/pfctl -e") 54 ToolsHelper.pf_rules([ 55 "set reassemble yes", 56 "set state-policy if-bound", 57 "block", 58 "pass inet proto icmp icmp-type echoreq", 59 ]) 60 61 ToolsHelper.print_output("/sbin/sysctl net.inet.ip.forwarding=1") 62 ToolsHelper.print_output("/sbin/pfctl -x loud") 63 64 ToolsHelper.print_output("/sbin/ifconfig %s mtu 1492" % if2name) 65 66 def vnet3_handler(self, vnet): 67 # Import in the correct vnet, so at to not confuse Scapy 68 import scapy.all as sp 69 70 ifname = vnet.iface_alias_map["if2"].name 71 ToolsHelper.print_output("/sbin/route add default 198.51.100.1") 72 ToolsHelper.print_output("/sbin/ifconfig %s inet alias 198.51.100.3/24" % ifname) 73 ToolsHelper.print_output("/sbin/ifconfig %s mtu 1492" % ifname) 74 75 def checkfn(packet): 76 icmp = packet.getlayer(sp.ICMP) 77 if not icmp: 78 return False 79 80 if icmp.type != 3: 81 return False 82 83 packet.show() 84 return True 85 86 sp.sniff(iface=ifname, stop_filter=checkfn) 87 vnet.pipe.send("Got ICMP destination unreachable packet") 88 89 @pytest.mark.require_user("root") 90 @pytest.mark.require_progs(["scapy"]) 91 def test_inner_match(self): 92 vnet = self.vnet_map["vnet1"] 93 dst_vnet = self.vnet_map["vnet3"] 94 sendif = vnet.iface_alias_map["if1"] 95 96 our_mac = sendif.ether 97 dst_mac = sendif.epairb.ether 98 99 # Import in the correct vnet, so at to not confuse Scapy 100 import scapy.all as sp 101 102 ToolsHelper.print_output("/sbin/route add default 192.0.2.1") 103 104 # Sanity check 105 check("/sbin/ping -c 1 192.0.2.1") 106 check("/sbin/ping -c 1 198.51.100.1") 107 check("/sbin/ping -c 2 198.51.100.3") 108 109 # Establish a state 110 pkt = sp.Ether(src=our_mac, dst=dst_mac) \ 111 / sp.IP(src="192.0.2.2", dst="198.51.100.2") \ 112 / sp.ICMP(type='echo-request') \ 113 / "PAYLOAD" 114 sp.sendp(pkt, sendif.name, verbose=False) 115 116 # Now try to pass an ICMP error message piggy-backing on that state, but 117 # use a different source address 118 pkt = sp.Ether(src=our_mac, dst=dst_mac) \ 119 / sp.IP(src="192.0.2.2", dst="198.51.100.3") \ 120 / sp.ICMP(type='dest-unreach') \ 121 / sp.IP(src="198.51.100.2", dst="192.0.2.2") \ 122 / sp.ICMP(type='echo-reply') 123 sp.sendp(pkt, sendif.name, verbose=False) 124 125 try: 126 rcvd = self.wait_object(dst_vnet.pipe, timeout=1) 127 if rcvd: 128 raise Exception(rcvd) 129 except TimeoutError as e: 130 # We expect the timeout here. It means we didn't get the destination 131 # unreachable packet in vnet3 132 pass 133 134 def check_icmp_echo(self, sp, payload_size): 135 packet = sp.IP(dst="198.51.100.2", flags="DF") \ 136 / sp.ICMP(type='echo-request') \ 137 / sp.raw(bytes.fromhex('f0') * payload_size) 138 139 p = sp.sr1(packet, timeout=3) 140 p.show() 141 142 ip = p.getlayer(sp.IP) 143 icmp = p.getlayer(sp.ICMP) 144 assert ip 145 assert icmp 146 147 if payload_size > 1464: 148 # Expect ICMP destination unreachable, fragmentation needed 149 assert ip.src == "192.0.2.1" 150 assert ip.dst == "192.0.2.2" 151 assert icmp.type == 3 # dest-unreach 152 assert icmp.code == 4 153 assert icmp.nexthopmtu == 1492 154 else: 155 # Expect echo reply 156 assert ip.src == "198.51.100.2" 157 assert ip.dst == "192.0.2.2" 158 assert icmp.type == 0 # "echo-reply" 159 assert icmp.code == 0 160 161 return 162 163 @pytest.mark.require_user("root") 164 @pytest.mark.require_progs(["scapy"]) 165 def test_fragmentation_needed(self): 166 ToolsHelper.print_output("/sbin/route add default 192.0.2.1") 167 168 ToolsHelper.print_output("/sbin/ping -c 1 198.51.100.2") 169 ToolsHelper.print_output("/sbin/ping -c 1 -D -s 1472 198.51.100.2") 170 171 # Import in the correct vnet, so at to not confuse Scapy 172 import scapy.all as sp 173 174 self.check_icmp_echo(sp, 128) 175 self.check_icmp_echo(sp, 1464) 176 self.check_icmp_echo(sp, 1468) 177 178 @pytest.mark.require_user("root") 179 @pytest.mark.require_progs(["scapy"]) 180 def test_truncated_opts(self): 181 ToolsHelper.print_output("/sbin/route add default 192.0.2.1") 182 183 # Import in the correct vnet, so at to not confuse Scapy 184 import scapy.all as sp 185 186 packet = sp.IP(dst="198.51.100.2", flags="DF") \ 187 / sp.ICMP(type='dest-unreach', length=108) \ 188 / sp.IP(src="198.51.100.2", dst="192.0.2.2", len=1000, \ 189 ihl=(120 >> 2), options=[ \ 190 sp.IPOption_Security(length=100)]) 191 packet.show() 192 sp.sr1(packet, timeout=3) 193 194class TestICMP_NAT(VnetTestTemplate): 195 REQUIRED_MODULES = [ "pf" ] 196 TOPOLOGY = { 197 "vnet1": {"ifaces": ["if1"]}, 198 "vnet2": {"ifaces": ["if1", "if2"]}, 199 "vnet3": {"ifaces": ["if2"]}, 200 "if1": {"prefixes4": [("192.0.2.2/24", "192.0.2.1/24")]}, 201 "if2": {"prefixes4": [("198.51.100.1/24", "198.51.100.2/24")]}, 202 } 203 204 def vnet2_handler(self, vnet): 205 ifname = vnet.iface_alias_map["if1"].name 206 if2name = vnet.iface_alias_map["if2"].name 207 208 ToolsHelper.print_output("/sbin/pfctl -e") 209 ToolsHelper.pf_rules([ 210 "set reassemble yes", 211 "set state-policy if-bound", 212 "nat on %s inet from 192.0.2.0/24 to any -> (%s)" % (if2name, if2name), 213 "block", 214 "pass inet proto icmp icmp-type echoreq", 215 ]) 216 217 ToolsHelper.print_output("/sbin/sysctl net.inet.ip.forwarding=1") 218 ToolsHelper.print_output("/sbin/pfctl -x loud") 219 220 def vnet3_handler(self, vnet): 221 ifname = vnet.iface_alias_map["if2"].name 222 ToolsHelper.print_output("/sbin/ifconfig %s inet alias 198.51.100.3/24" % ifname) 223 224 @pytest.mark.require_user("root") 225 @pytest.mark.require_progs(["scapy"]) 226 def test_id_conflict(self): 227 """ 228 Test ICMP echo requests with the same ID from different clients. 229 Windows does this, and it can confuse pf. 230 """ 231 ifname = self.vnet.iface_alias_map["if1"].name 232 ToolsHelper.print_output("/sbin/route add default 192.0.2.1") 233 ToolsHelper.print_output("/sbin/ifconfig %s inet alias 192.0.2.3/24" % ifname) 234 235 ToolsHelper.print_output("/sbin/ping -c 1 192.0.2.1") 236 ToolsHelper.print_output("/sbin/ping -c 1 198.51.100.1") 237 ToolsHelper.print_output("/sbin/ping -c 1 198.51.100.2") 238 ToolsHelper.print_output("/sbin/ping -c 1 198.51.100.3") 239 240 # Import in the correct vnet, so at to not confuse Scapy 241 import scapy.all as sp 242 243 # Address one 244 packet = sp.IP(src="192.0.2.2", dst="198.51.100.2", flags="DF") \ 245 / sp.ICMP(type='echo-request', id=42) \ 246 / sp.raw(bytes.fromhex('f0') * 16) 247 248 p = sp.sr1(packet, timeout=3) 249 p.show() 250 ip = p.getlayer(sp.IP) 251 icmp = p.getlayer(sp.ICMP) 252 assert ip 253 assert icmp 254 assert ip.dst == "192.0.2.2" 255 assert icmp.id == 42 256 257 # Address one 258 packet = sp.IP(src="192.0.2.3", dst="198.51.100.2", flags="DF") \ 259 / sp.ICMP(type='echo-request', id=42) \ 260 / sp.raw(bytes.fromhex('f0') * 16) 261 262 p = sp.sr1(packet, timeout=3) 263 p.show() 264 ip = p.getlayer(sp.IP) 265 icmp = p.getlayer(sp.ICMP) 266 assert ip 267 assert icmp 268 assert ip.dst == "192.0.2.3" 269 assert icmp.id == 42 270