Re: [PATCH net v3 3/4] selftests: net: Test UDP length overflow with PMTU discover and big MTU

Alice Mikityanska wrote:
> From: Alice Mikityanska <[email protected]>
> 
> Two previous commits fixed overflow of UDP length when setsockopt
> IP(V6)_MTU_DISCOVER is set to IPV6_PMTUDISC_DO or IP(V6)_PMTUDISC_PROBE,
> and a large packet is sent over a netdev with an unusually large MTU.
> 
> This commit adds the selftests that replicate the described steps to
> reproduce for IPv6 and IPv4, and also one more test that ensures that
> sending UDP jumbograms over a raw socket is still possible after the
> fix.
> 
> Signed-off-by: Alice Mikityanska <[email protected]>
> ---
>  tools/testing/selftests/net/Makefile         |   1 +
>  tools/testing/selftests/net/cork_fragsize.py | 130 +++++++++++++++++++
>  2 files changed, 131 insertions(+)
>  create mode 100755 tools/testing/selftests/net/cork_fragsize.py
> 
> diff --git a/tools/testing/selftests/net/Makefile b/tools/testing/selftests/net/Makefile
> index 0f5c178bc224..c6d332c90a54 100644
> --- a/tools/testing/selftests/net/Makefile
> +++ b/tools/testing/selftests/net/Makefile
> @@ -25,6 +25,7 @@ TEST_PROGS := \
>  	cmsg_so_mark.sh \
>  	cmsg_so_priority.sh \
>  	cmsg_time.sh \
> +	cork_fragsize.py \
>  	double_udp_encap.sh \
>  	drop_monitor_tests.sh \
>  	ecmp_rehash.sh \
> diff --git a/tools/testing/selftests/net/cork_fragsize.py b/tools/testing/selftests/net/cork_fragsize.py
> new file mode 100755
> index 000000000000..d89ef9892329
> --- /dev/null
> +++ b/tools/testing/selftests/net/cork_fragsize.py
> @@ -0,0 +1,130 @@
> +#!/usr/bin/env python3
> +# SPDX-License-Identifier: GPL-2.0
> +
> +# Test possible UDP length overflow in udp_send_skb/udp_v6_send_skb.
> +
> +from lib.py import ksft_run, ksft_exit, ksft_true, KsftSkipEx
> +from lib.py import ip, NetNS, NetNSEnter
> +import errno
> +import gzip
> +import os
> +import socket
> +import struct
> +import subprocess

Various pylint and ruff issues, such as ordering imports

> +
> +
> +IP_MTU_DISCOVER = 10
> +IP_PMTUDISC_PROBE = 3
> +IPV6_MTU_DISCOVER = 23
> +IPV6_PMTUDISC_DO = 2
> +IPV6_PMTUDISC_PROBE = 3
> +IPV6_TLV_JUMBO = 194
> +
> +
> +def check_kernel_config(option) -> bool | None:
> +    for filename, method in [
> +        ('/proc/config.gz', gzip.open),
> +        (f'/boot/config-{os.uname().release}', open),
> +    ]:
> +        try:
> +            with method(filename, 'rt') as config:
> +                for line in config:
> +                    if line.rstrip() == f'{option}=y':
> +                        return True
> +                return False
> +        except OSError:
> +            continue
> +
> +
> +def assert_debug_kernel() -> None:
> +    res = check_kernel_config('CONFIG_DEBUG_NET')
> +    if res is None:
> +        print("WARN: Can't read kernel config; assuming debug kernel, and running the test")
> +    elif not res:
> +        raise KsftSkipEx('CONFIG_DEBUG_NET is not set')
> +
> +
> +def check_dmesg_clean(func) -> bool:
> +    dmesg = subprocess.Popen(['dmesg'], stdout=subprocess.PIPE)
> +    result = subprocess.run(['grep', '-q', f'WARNING:.*{func}'], stdin=dmesg.stdout)
> +    dmesg.wait()
> +    return result.returncode != 0 and dmesg.returncode == 0
> +
> +
> +def ip_setup(ns: NetNS, mtu: int, ipv6: bool) -> None:
> +    ip('link add dummy type dummy', ns=ns)
> +    ip(f'link set dummy mtu {mtu}', ns=ns)
> +    ip('link set dummy up', ns=ns)
> +    flag = '-6' if ipv6 else ''
> +    nodad = 'nodad' if ipv6 else ''
> +    addr_local = 'fd00::1/64' if ipv6 else '10.0.0.1/24'
> +    addr_remote = 'fd00::2' if ipv6 else '10.0.0.2'
> +    ip(f'{flag} addr add {addr_local} dev dummy {nodad}', ns=ns)
> +    ip(f'{flag} neigh add {addr_remote} lladdr 02:00:00:00:00:02 dev dummy nud permanent', ns=ns)
> +
> +
> +def test_ipv6() -> None:
> +    assert_debug_kernel()
> +
> +    with NetNS() as ns:
> +        ip_setup(ns, 65576, True)
> +
> +        with NetNSEnter(ns):
> +            with socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) as fd:
> +                fd.setsockopt(socket.IPPROTO_IPV6, IPV6_MTU_DISCOVER, IPV6_PMTUDISC_DO)
> +                try:
> +                    fd.sendto(b' ' * 65528, ('fd00::2', 1234))
> +                except OSError as e:
> +                    # Ignore EMSGSIZE: it happens on kernels with the fix.
> +                    if e.errno != errno.EMSGSIZE:
> +                        raise
> +
> +        ip('link del dummy', ns=ns)
> +
> +    ksft_true(check_dmesg_clean('udp_v6_send_skb'), 'WARNING detected in dmesg')
> +
> +
> +def test_ipv4() -> None:
> +    assert_debug_kernel()
> +
> +    with NetNS() as ns:
> +        ip_setup(ns, 65556, False)
> +
> +        with NetNSEnter(ns):
> +            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as fd:
> +                fd.setsockopt(socket.IPPROTO_IP, IP_MTU_DISCOVER, IP_PMTUDISC_PROBE)
> +                try:
> +                    fd.sendto(b' ' * 65528, ('10.0.0.2', 1234))
> +                except OSError as e:
> +                    # Ignore EMSGSIZE: the check happens after the WARN is printed.
> +                    if e.errno != errno.EMSGSIZE:
> +                        raise
> +
> +        ip('link del dummy', ns=ns)
> +
> +    ksft_true(check_dmesg_clean('udp_send_skb'), 'WARNING detected in dmesg')

These two are identical apart from 7 constants.

Dedup and use @ksft_variants decorators?

> +def test_ipv6_jumbo() -> None:
> +    with NetNS() as ns:
> +        ip_setup(ns, 65584, True)
> +
> +        with NetNSEnter(ns):
> +            with socket.socket(socket.AF_INET6, socket.SOCK_RAW, socket.IPPROTO_UDP) as fd:
> +                hopopts = struct.pack('!BBBBI', 0, 0, IPV6_TLV_JUMBO, 4, 65544)
> +                fd.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_HOPOPTS, hopopts)
> +                fd.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_CHECKSUM, 6)
> +                fd.setsockopt(socket.IPPROTO_IPV6, IPV6_MTU_DISCOVER, IPV6_PMTUDISC_PROBE)
> +                udp = struct.pack('!HHHH', 1234, 1234, 0, 0) + b' ' * 65528
> +                fd.sendto(udp, ('fd00::2', 0))
> +
> +        ip('link del dummy', ns=ns)

Should this have some test at the end? check_dmesg_clean or perhaps
a socket receiving the data.

> +
> +
> +if __name__ == "__main__":
> +    ksft_run([
> +        test_ipv6,
> +        test_ipv4,
> +        test_ipv6_jumbo,
> +    ])
> +    ksft_exit()
> -- 
> 2.55.0
>