1#!/usr/bin/env python3 2# SPDX-License-Identifier: GPL-2.0 3 4'''Test possible UDP length overflow in udp_send_skb/udp_v6_send_skb.''' 5 6import errno 7import gzip 8import os 9import socket 10import struct 11import subprocess 12from contextlib import contextmanager 13 14from lib.py import ( 15 KsftNamedVariant, 16 KsftSkipEx, 17 NetNS, 18 NetNSEnter, 19 defer, 20 ip, 21 ksft_eq, 22 ksft_exit, 23 ksft_pr, 24 ksft_raises, 25 ksft_run, 26 ksft_true, 27 ksft_variants, 28) 29 30IP_MTU_DISCOVER = 10 31IP_PMTUDISC_PROBE = 3 32IPV6_MTU_DISCOVER = 23 33IPV6_PMTUDISC_DO = 2 34IPV6_PMTUDISC_PROBE = 3 35IPV6_TLV_JUMBO = 194 36 37 38def check_kernel_config(option: str) -> bool | None: 39 ''' 40 Check whether the option is enabled in the config of the running kernel. 41 Returns None if the config is not found; otherwise returns True/False 42 depending on the option value in the config. 43 ''' 44 45 for filename, method in [ 46 ('/proc/config.gz', gzip.open), 47 (f'/boot/config-{os.uname().release}', open), 48 ]: 49 try: 50 with method(filename, 'rt') as config: 51 for line in config: 52 if line.rstrip() == f'{option}=y': 53 return True 54 return False 55 except OSError: 56 continue 57 return None 58 59 60def assert_debug_kernel() -> None: 61 ''' 62 Skip the test if CONFIG_DEBUG_NET is not set in the kernel config. 63 ''' 64 65 res = check_kernel_config('CONFIG_DEBUG_NET') 66 if res is None: 67 ksft_pr("WARN: Can't read kernel config; assuming debug kernel, and running the test") 68 elif not res: 69 raise KsftSkipEx('CONFIG_DEBUG_NET is not set') 70 71 72def check_dmesg_clean(func: str) -> bool: 73 ''' 74 Check if the given function produced a WARN in dmesg. 75 ''' 76 77 with subprocess.Popen(['dmesg'], stdout=subprocess.PIPE) as dmesg: 78 res = subprocess.run(['grep', '-q', f'WARNING:.*{func}'], stdin=dmesg.stdout, check=False) 79 return res.returncode != 0 and dmesg.returncode == 0 80 81 82@contextmanager 83def dummy_netdev(ns: NetNS, mtu: int, ipv6: bool) -> None: 84 ''' 85 Create a dummy netdev inside the given namespace, and tune it for the test. 86 ''' 87 88 ip('link add dummy type dummy', ns=ns) 89 with defer(ip, 'link del dummy', ns=ns): 90 ip(f'link set dummy mtu {mtu}', ns=ns) 91 ip('link set dummy up', ns=ns) 92 flag = '-6' if ipv6 else '' 93 nodad = 'nodad' if ipv6 else '' 94 local = 'fd00::1/64' if ipv6 else '10.0.0.1/24' 95 remote = 'fd00::2' if ipv6 else '10.0.0.2' 96 ip(f'{flag} addr add {local} dev dummy {nodad}', ns=ns) 97 ip(f'{flag} neigh add {remote} lladdr 02:00:00:00:00:02 dev dummy nud permanent', ns=ns) 98 yield 99 100 101@ksft_variants([ 102 KsftNamedVariant( 103 'ipv6', 104 True, 105 socket.AF_INET6, 106 (socket.IPPROTO_IPV6, IPV6_MTU_DISCOVER, IPV6_PMTUDISC_DO), 107 'fd00::2', 108 'udp_v6_send_skb', 109 ), 110 KsftNamedVariant( 111 'ipv4', 112 False, 113 socket.AF_INET, 114 (socket.IPPROTO_IP, IP_MTU_DISCOVER, IP_PMTUDISC_PROBE), 115 '10.0.0.2', 116 'udp_send_skb', 117 ), 118]) 119def test_udp( 120 ipv6: bool, 121 af: socket.AddressFamily, 122 sockopts: tuple[int, int, int], 123 destip: str, 124 func: str 125) -> None: 126 ''' 127 Test that sending an oversized UDP packet over a UDP socket doesn't overflow 128 the 16-bit length field in the UDP header, which could happen on older 129 kernels in udp_send_skb/udp_v6_send_skb. 130 131 IPv4: The packet will be dropped with EMSGSIZE, but the overflow could 132 happen before it happens. The only way to test this is to check dmesg on 133 CONFIG_DEBUG_NET=y kernels that have udp_set_len_short with the warning. 134 135 IPv6: The packet will be dropped with EMSGSIZE on fixed kernels, and will be 136 sent corrupted on older kernels. Test both: sendto must return EMSGSIZE, and 137 dmesg must be clean of warnings on CONFIG_DEBUG_NET=y kernels. 138 ''' 139 140 if not ipv6: 141 assert_debug_kernel() 142 143 with ( 144 NetNS() as ns, 145 dummy_netdev(ns, 65556 + 20 * ipv6, ipv6), 146 NetNSEnter(ns), 147 socket.socket(af, socket.SOCK_DGRAM) as fd, 148 ): 149 fd.setsockopt(*sockopts) 150 with ksft_raises(OSError) as e: 151 fd.sendto(b' ' * 65528, (destip, 1234)) 152 # IPv6: EMSGSIZE happens on kernels with the fix. 153 # IPv4: EMSGSIZE happens on both fixed and unfixed kernels, after the 154 # WARN is printed - ignore it and rely on the dmesg check. 155 if e.exception is not None: 156 ksft_eq(e.exception.errno, errno.EMSGSIZE) 157 158 ksft_true(check_dmesg_clean(func), 'WARNING detected in dmesg') 159 160 161def test_ipv6_jumbo() -> None: 162 ''' 163 Test that sending UDP jumbograms over a raw IPv6 socket works, despite 164 having the fix for oversized UDP packets. sendto must not raise an OSError 165 exception (when raised, the test fails automatically). 166 ''' 167 168 with ( 169 NetNS() as ns, 170 dummy_netdev(ns, 65584, True), 171 NetNSEnter(ns), 172 socket.socket(socket.AF_INET6, socket.SOCK_RAW, socket.IPPROTO_UDP) as fd, 173 ): 174 hopopts = struct.pack('!BBBBI', 0, 0, IPV6_TLV_JUMBO, 4, 65544) 175 fd.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_HOPOPTS, hopopts) 176 fd.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_CHECKSUM, 6) 177 fd.setsockopt(socket.IPPROTO_IPV6, IPV6_MTU_DISCOVER, IPV6_PMTUDISC_PROBE) 178 udp = struct.pack('!HHHH', 1234, 1234, 0, 0) + b' ' * 65528 179 fd.sendto(udp, ('fd00::2', 0)) 180 181 182if __name__ == "__main__": 183 ksft_run([ 184 test_udp, 185 test_ipv6_jumbo, 186 ]) 187 ksft_exit() 188