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")], "mtu": 1492}, 47 } 48 49 def vnet2_handler(self, vnet): 50 ToolsHelper.print_output("/sbin/pfctl -e") 51 ToolsHelper.pf_rules([ 52 "set reassemble yes", 53 "set state-policy if-bound", 54 "block", 55 "pass inet proto icmp icmp-type echoreq", 56 ]) 57 58 ToolsHelper.print_output("/sbin/sysctl net.inet.ip.forwarding=1") 59 ToolsHelper.print_output("/sbin/pfctl -x loud") 60 61 def vnet3_handler(self, vnet): 62 # Import in the correct vnet, so at to not confuse Scapy 63 import scapy.all as sp 64 65 ifname = vnet.iface_alias_map["if2"].name 66 ToolsHelper.print_output("/sbin/route add default 198.51.100.1") 67 ToolsHelper.print_output("/sbin/ifconfig %s inet alias 198.51.100.3/24" % ifname) 68 69 def checkfn(packet): 70 icmp = packet.getlayer(sp.ICMP) 71 if not icmp: 72 return False 73 74 if icmp.type != 3: 75 return False 76 77 packet.show() 78 return True 79 80 sp.sniff(iface=ifname, stop_filter=checkfn) 81 vnet.pipe.send("Got ICMP destination unreachable packet") 82 83 @pytest.mark.require_user("root") 84 @pytest.mark.require_progs(["scapy"]) 85 def test_inner_match(self): 86 vnet = self.vnet_map["vnet1"] 87 dst_vnet = self.vnet_map["vnet3"] 88 sendif = vnet.iface_alias_map["if1"] 89 90 our_mac = sendif.ether 91 dst_mac = sendif.epairb.ether 92 93 # Import in the correct vnet, so at to not confuse Scapy 94 import scapy.all as sp 95 96 ToolsHelper.print_output("/sbin/route add default 192.0.2.1") 97 98 # Sanity check 99 check("/sbin/ping -c 1 192.0.2.1") 100 check("/sbin/ping -c 1 198.51.100.1") 101 check("/sbin/ping -c 2 198.51.100.3") 102 103 # Establish a state 104 pkt = sp.Ether(src=our_mac, dst=dst_mac) \ 105 / sp.IP(src="192.0.2.2", dst="198.51.100.2") \ 106 / sp.ICMP(type='echo-request') \ 107 / "PAYLOAD" 108 sp.sendp(pkt, sendif.name, verbose=False) 109 110 # Now try to pass an ICMP error message piggy-backing on that state, but 111 # use a different source address 112 pkt = sp.Ether(src=our_mac, dst=dst_mac) \ 113 / sp.IP(src="192.0.2.2", dst="198.51.100.3") \ 114 / sp.ICMP(type='dest-unreach') \ 115 / sp.IP(src="198.51.100.2", dst="192.0.2.2") \ 116 / sp.ICMP(type='echo-reply') 117 sp.sendp(pkt, sendif.name, verbose=False) 118 119 try: 120 rcvd = self.wait_object(dst_vnet.pipe, timeout=1) 121 if rcvd: 122 raise Exception(rcvd) 123 except TimeoutError as e: 124 # We expect the timeout here. It means we didn't get the destination 125 # unreachable packet in vnet3 126 pass 127 128 def check_icmp_echo(self, sp, payload_size): 129 packet = sp.IP(dst="198.51.100.2", flags="DF") \ 130 / sp.ICMP(type='echo-request') \ 131 / sp.raw(bytes.fromhex('f0') * payload_size) 132 133 p = sp.sr1(packet, timeout=3) 134 p.show() 135 136 ip = p.getlayer(sp.IP) 137 icmp = p.getlayer(sp.ICMP) 138 assert ip 139 assert icmp 140 141 if payload_size > 1464: 142 # Expect ICMP destination unreachable, fragmentation needed 143 assert ip.src == "192.0.2.1" 144 assert ip.dst == "192.0.2.2" 145 assert icmp.type == 3 # dest-unreach 146 assert icmp.code == 4 147 assert icmp.nexthopmtu == 1492 148 else: 149 # Expect echo reply 150 assert ip.src == "198.51.100.2" 151 assert ip.dst == "192.0.2.2" 152 assert icmp.type == 0 # "echo-reply" 153 assert icmp.code == 0 154 155 return 156 157 @pytest.mark.require_user("root") 158 @pytest.mark.require_progs(["scapy"]) 159 def test_fragmentation_needed(self): 160 ToolsHelper.print_output("/sbin/route add default 192.0.2.1") 161 162 ToolsHelper.print_output("/sbin/ping -c 1 198.51.100.2") 163 ToolsHelper.print_output("/sbin/ping -c 1 -D -s 1472 198.51.100.2") 164 165 # Import in the correct vnet, so at to not confuse Scapy 166 import scapy.all as sp 167 168 self.check_icmp_echo(sp, 128) 169 self.check_icmp_echo(sp, 1464) 170 self.check_icmp_echo(sp, 1468) 171 172 @pytest.mark.require_user("root") 173 @pytest.mark.require_progs(["scapy"]) 174 def test_truncated_opts(self): 175 ToolsHelper.print_output("/sbin/route add default 192.0.2.1") 176 177 # Import in the correct vnet, so at to not confuse Scapy 178 import scapy.all as sp 179 180 packet = sp.IP(dst="198.51.100.2", flags="DF") \ 181 / sp.ICMP(type='dest-unreach', length=108) \ 182 / sp.IP(src="198.51.100.2", dst="192.0.2.2", len=1000, \ 183 ihl=(120 >> 2), options=[ \ 184 sp.IPOption_Security(length=100)]) 185 packet.show() 186 sp.sr1(packet, timeout=3) 187 188class TestICMP_NAT(VnetTestTemplate): 189 REQUIRED_MODULES = [ "pf" ] 190 TOPOLOGY = { 191 "vnet1": {"ifaces": ["if1"]}, 192 "vnet2": {"ifaces": ["if1", "if2"]}, 193 "vnet3": {"ifaces": ["if2"]}, 194 "if1": {"prefixes4": [("192.0.2.2/24", "192.0.2.1/24")]}, 195 "if2": {"prefixes4": [("198.51.100.1/24", "198.51.100.2/24")]}, 196 } 197 198 def vnet2_handler(self, vnet): 199 ifname = vnet.iface_alias_map["if1"].name 200 if2name = vnet.iface_alias_map["if2"].name 201 202 ToolsHelper.print_output("/sbin/pfctl -e") 203 ToolsHelper.pf_rules([ 204 "set reassemble yes", 205 "set state-policy if-bound", 206 "nat on %s inet from 192.0.2.0/24 to any -> (%s)" % (if2name, if2name), 207 "block", 208 "pass inet proto icmp icmp-type echoreq", 209 ]) 210 211 ToolsHelper.print_output("/sbin/sysctl net.inet.ip.forwarding=1") 212 ToolsHelper.print_output("/sbin/pfctl -x loud") 213 214 def vnet3_handler(self, vnet): 215 ifname = vnet.iface_alias_map["if2"].name 216 ToolsHelper.print_output("/sbin/ifconfig %s inet alias 198.51.100.3/24" % ifname) 217 218 @pytest.mark.require_user("root") 219 @pytest.mark.require_progs(["scapy"]) 220 def test_id_conflict(self): 221 """ 222 Test ICMP echo requests with the same ID from different clients. 223 Windows does this, and it can confuse pf. 224 """ 225 ifname = self.vnet.iface_alias_map["if1"].name 226 ToolsHelper.print_output("/sbin/route add default 192.0.2.1") 227 ToolsHelper.print_output("/sbin/ifconfig %s inet alias 192.0.2.3/24" % ifname) 228 229 ToolsHelper.print_output("/sbin/ping -c 1 192.0.2.1") 230 ToolsHelper.print_output("/sbin/ping -c 1 198.51.100.1") 231 ToolsHelper.print_output("/sbin/ping -c 1 198.51.100.2") 232 ToolsHelper.print_output("/sbin/ping -c 1 198.51.100.3") 233 234 # Import in the correct vnet, so at to not confuse Scapy 235 import scapy.all as sp 236 237 # Address one 238 packet = sp.IP(src="192.0.2.2", dst="198.51.100.2", flags="DF") \ 239 / sp.ICMP(type='echo-request', id=42) \ 240 / sp.raw(bytes.fromhex('f0') * 16) 241 242 p = sp.sr1(packet, timeout=3) 243 p.show() 244 ip = p.getlayer(sp.IP) 245 icmp = p.getlayer(sp.ICMP) 246 assert ip 247 assert icmp 248 assert ip.dst == "192.0.2.2" 249 assert icmp.id == 42 250 251 # Address one 252 packet = sp.IP(src="192.0.2.3", dst="198.51.100.2", flags="DF") \ 253 / sp.ICMP(type='echo-request', id=42) \ 254 / sp.raw(bytes.fromhex('f0') * 16) 255 256 p = sp.sr1(packet, timeout=3) 257 p.show() 258 ip = p.getlayer(sp.IP) 259 icmp = p.getlayer(sp.ICMP) 260 assert ip 261 assert icmp 262 assert ip.dst == "192.0.2.3" 263 assert icmp.id == 42 264