xref: /freebsd/sbin/dhclient/tests/functional.py (revision a143197801c680b958ab662c88cf086668a1958b)

#
# Copyright (c) 2026 The FreeBSD Foundation
#
# This software was developed by Mark Johnston under sponsorship from
# the FreeBSD Foundation.
#
# SPDX-License-Identifier: BSD-2-Clause
#

import subprocess

import pytest
import scapy.all as sp
from atf_python.sys.net.tools import ToolsHelper
from atf_python.sys.net.vnet import VnetTestTemplate


def mac_to_chaddr(mac_str):
    """Convert a MAC address string to a 16-byte BOOTP chaddr."""
    mac_bytes = bytes.fromhex(mac_str.replace(":", ""))
    return mac_bytes + b"\x00" * (16 - len(mac_bytes))


def get_dhcp_type(pkt):
    """Extract the DHCP message type from a packet."""
    if not pkt.haslayer(sp.DHCP):
        return None
    for opt in pkt[sp.DHCP].options:
        if isinstance(opt, tuple) and opt[0] == "message-type":
            return sp.DHCPTypes[opt[1]]
    return None


def build_reply(iface, server_ip, client_ip, client_mac, xid, msg_type,
                extra_opts=None, sname=b"", file=b""):
    """Build a DHCP OFFER or ACK packet."""
    options = [
        ("message-type", msg_type),
        ("server_id", server_ip),
        ("lease_time", 3600),
        ("subnet_mask", "255.255.255.0"),
        ("router", server_ip),
    ]
    if extra_opts:
        options.extend(extra_opts)
    options.append("end")

    return (
        sp.Ether(dst=client_mac, src=sp.get_if_hwaddr(iface))
        / sp.IP(src=server_ip, dst="255.255.255.255")
        / sp.UDP(sport=67, dport=68)
        / sp.BOOTP(op=2,
            xid=xid,
            yiaddr=client_ip,
            siaddr=server_ip,
            chaddr=mac_to_chaddr(client_mac),
            sname=sname,
            file=file,
        )
        / sp.DHCP(options=options)
    )


class TestDhclient(VnetTestTemplate):
    _SERVER_IP = "192.0.2.1"
    _CLIENT_IP = "192.0.2.10"

    TOPOLOGY = {
        "vnet1": {"ifaces": ["if1"]},
        "vnet2": {"ifaces": ["if1"]},
        "if1": {},
    }

    def vnet1_handler(self, vnet):
        """
        Run dhclient in vnet1 (the client side of the epair).  Send
        back the output of "ifconfig <iface>" once dhclient has finished
        its exchange.
        """
        ifname = vnet.iface_alias_map["if1"].name

        # Wait for the test body to be ready before starting dhclient.
        vnet.pipe.recv()

        try:
            result = subprocess.run(
                ["/sbin/dhclient", "-c", "/dev/null",
                 "-l", "dhclient.lease", ifname],
                capture_output=True, text=True, timeout=30,
            )
            stderr = result.stderr
        except subprocess.TimeoutExpired:
            stderr = "dhclient timed out"
        out = ToolsHelper.get_output("/sbin/ifconfig {}".format(ifname))
        lease = ""
        try:
            with open("dhclient.lease", "r") as f:
                lease = f.read()
        except FileNotFoundError:
            pass
        vnet.pipe.send({"ifconfig": out, "stderr": stderr, "lease": lease})

    def _setup_server(self):
        """Configure the server side of the epair and return its name."""
        ifname = self.vnet.iface_alias_map["if1"].name
        ToolsHelper.print_output(
            "/sbin/ifconfig {} inet {}/24 up".format(ifname, self._SERVER_IP)
        )
        return ifname

    def _do_dora(self, ifname, first_extra_opts=None, extra_opts=None,
                 sname=b"", file=b""):
        """
        Run a discover-offer-request-acknowledge exchange and
        return (discover_count, result).

        If first_extra_opts is set, it is used in the OFFER for the first
        DHCPDISCOVER only; subsequent DISCOVERs use extra_opts.
        """
        discover_count = 0
        done = False

        def handle_dhcp(pkt):
            nonlocal discover_count, done

            msg_type = get_dhcp_type(pkt)
            if msg_type is None:
                return

            client_mac = pkt[sp.Ether].src
            xid = pkt[sp.BOOTP].xid

            if msg_type == "discover":
                discover_count += 1
                opts = extra_opts
                if discover_count == 1 and first_extra_opts is not None:
                    opts = first_extra_opts
                reply = build_reply(
                    ifname, self._SERVER_IP, self._CLIENT_IP, client_mac,
                    xid, "offer", extra_opts=opts, sname=sname, file=file,
                )
                sp.sendp(reply, iface=ifname, verbose=0)

            elif msg_type == "request":
                reply = build_reply(
                    ifname, self._SERVER_IP, self._CLIENT_IP, client_mac,
                    xid, "ack", sname=sname, file=file,
                )
                sp.sendp(reply, iface=ifname, verbose=0)
                done = True

        sniffer = sp.AsyncSniffer(
            iface=ifname,
            filter="udp and dst port 67",
            prn=handle_dhcp,
            stop_filter=lambda _: done,
            started_callback=lambda: self.send_object(
                self.vnet_map["vnet1"].pipe, "go",
            ),
        )
        sniffer.start()
        sniffer.join(timeout=30)

        assert done, "DHCP exchange did not complete"
        result = self.wait_object(self.vnet_map["vnet1"].pipe, timeout=15)
        return discover_count, result

    @pytest.mark.require_user("root")
    def test_classless_routes_truncated(self):
        """
        Verify that dhclient rejects a DHCPOFFER containing a truncated
        classless static routes option.

        A malformed option 121 payload is constructed with a valid first
        entry (default route, width=0) followed by a second entry with
        width=24 that has only 1 byte of subnet data instead of the
        required 3 bytes.

        The test sends the bad offer first, then a valid offer, and
        verifies that dhclient rejects the bad offer and successfully
        acquires a lease from the good one.
        """
        ifname = self._setup_server()
        bad_routes = (
            b"\x00"              # width=0 (default route)
            b"\xc0\xa8\x01\x01"  # gateway: 192.168.1.1
            b"\x18"              # width=24 (/24 subnet)
            b"\x0a"              # only 1 of 3 required subnet octets
        )
        discover_count, result = self._do_dora(
            ifname, first_extra_opts=[(121, bad_routes)],
        )
        assert discover_count >= 2, \
            "Expected dhclient to reject the first (bad) offer and retry"
        assert "inet {}".format(self._CLIENT_IP) in result["ifconfig"]

    @pytest.mark.require_user("root")
    def test_filename_backslash(self):
        """
        Verify that backslashes in the BOOTP filename don't cause the
        lease to be rejected.  These can appear in Windows boot paths.
        """
        ifname = self._setup_server()
        discover_count, result = self._do_dora(
            ifname, file=b"\\windows\\boot\\pxeboot.com",
        )
        assert discover_count == 1, \
            "Lease with backslash in filename should be accepted"
        assert "inet {}".format(self._CLIENT_IP) in result["ifconfig"]

    @pytest.mark.require_user("root")
    def test_sname_valid(self):
        """
        Verify that a valid server name is preserved in the lease file.
        """
        ifname = self._setup_server()
        discover_count, result = self._do_dora(
            ifname, sname=b"dhcp.example.com",
        )
        assert discover_count == 1
        assert "inet {}".format(self._CLIENT_IP) in result["ifconfig"]
        assert "dhcp.example.com" in result["lease"]

    @pytest.mark.require_user("root")
    def test_sname_quote(self):
        """
        Verify that a double quote in the BOOTP server name causes the
        server name to be discarded but does not cause the lease to be
        rejected.  Values from the lease file may be handled by
        dhclient-script and thus are susceptible to shell injection.
        """
        ifname = self._setup_server()
        discover_count, result = self._do_dora(
            ifname, sname=b'evil"server',
        )
        assert discover_count == 1, \
            "Lease with quote in sname should be accepted"
        assert "inet {}".format(self._CLIENT_IP) in result["ifconfig"]
        assert 'evil"server' not in result["lease"], \
            "Server name with quote should not appear in the lease file"
        assert "server-name" not in result["lease"], \
            "Unsafe server name should be discarded entirely"