xref: /freebsd/sbin/dhclient/tests/functional.py (revision a143197801c680b958ab662c88cf086668a1958b)
1#
2# Copyright (c) 2026 The FreeBSD Foundation
3#
4# This software was developed by Mark Johnston under sponsorship from
5# the FreeBSD Foundation.
6#
7# SPDX-License-Identifier: BSD-2-Clause
8#
9
10import subprocess
11
12import pytest
13import scapy.all as sp
14from atf_python.sys.net.tools import ToolsHelper
15from atf_python.sys.net.vnet import VnetTestTemplate
16
17
18def mac_to_chaddr(mac_str):
19    """Convert a MAC address string to a 16-byte BOOTP chaddr."""
20    mac_bytes = bytes.fromhex(mac_str.replace(":", ""))
21    return mac_bytes + b"\x00" * (16 - len(mac_bytes))
22
23
24def get_dhcp_type(pkt):
25    """Extract the DHCP message type from a packet."""
26    if not pkt.haslayer(sp.DHCP):
27        return None
28    for opt in pkt[sp.DHCP].options:
29        if isinstance(opt, tuple) and opt[0] == "message-type":
30            return sp.DHCPTypes[opt[1]]
31    return None
32
33
34def build_reply(iface, server_ip, client_ip, client_mac, xid, msg_type,
35                extra_opts=None, sname=b"", file=b""):
36    """Build a DHCP OFFER or ACK packet."""
37    options = [
38        ("message-type", msg_type),
39        ("server_id", server_ip),
40        ("lease_time", 3600),
41        ("subnet_mask", "255.255.255.0"),
42        ("router", server_ip),
43    ]
44    if extra_opts:
45        options.extend(extra_opts)
46    options.append("end")
47
48    return (
49        sp.Ether(dst=client_mac, src=sp.get_if_hwaddr(iface))
50        / sp.IP(src=server_ip, dst="255.255.255.255")
51        / sp.UDP(sport=67, dport=68)
52        / sp.BOOTP(op=2,
53            xid=xid,
54            yiaddr=client_ip,
55            siaddr=server_ip,
56            chaddr=mac_to_chaddr(client_mac),
57            sname=sname,
58            file=file,
59        )
60        / sp.DHCP(options=options)
61    )
62
63
64class TestDhclient(VnetTestTemplate):
65    _SERVER_IP = "192.0.2.1"
66    _CLIENT_IP = "192.0.2.10"
67
68    TOPOLOGY = {
69        "vnet1": {"ifaces": ["if1"]},
70        "vnet2": {"ifaces": ["if1"]},
71        "if1": {},
72    }
73
74    def vnet1_handler(self, vnet):
75        """
76        Run dhclient in vnet1 (the client side of the epair).  Send
77        back the output of "ifconfig <iface>" once dhclient has finished
78        its exchange.
79        """
80        ifname = vnet.iface_alias_map["if1"].name
81
82        # Wait for the test body to be ready before starting dhclient.
83        vnet.pipe.recv()
84
85        try:
86            result = subprocess.run(
87                ["/sbin/dhclient", "-c", "/dev/null",
88                 "-l", "dhclient.lease", ifname],
89                capture_output=True, text=True, timeout=30,
90            )
91            stderr = result.stderr
92        except subprocess.TimeoutExpired:
93            stderr = "dhclient timed out"
94        out = ToolsHelper.get_output("/sbin/ifconfig {}".format(ifname))
95        lease = ""
96        try:
97            with open("dhclient.lease", "r") as f:
98                lease = f.read()
99        except FileNotFoundError:
100            pass
101        vnet.pipe.send({"ifconfig": out, "stderr": stderr, "lease": lease})
102
103    def _setup_server(self):
104        """Configure the server side of the epair and return its name."""
105        ifname = self.vnet.iface_alias_map["if1"].name
106        ToolsHelper.print_output(
107            "/sbin/ifconfig {} inet {}/24 up".format(ifname, self._SERVER_IP)
108        )
109        return ifname
110
111    def _do_dora(self, ifname, first_extra_opts=None, extra_opts=None,
112                 sname=b"", file=b""):
113        """
114        Run a discover-offer-request-acknowledge exchange and
115        return (discover_count, result).
116
117        If first_extra_opts is set, it is used in the OFFER for the first
118        DHCPDISCOVER only; subsequent DISCOVERs use extra_opts.
119        """
120        discover_count = 0
121        done = False
122
123        def handle_dhcp(pkt):
124            nonlocal discover_count, done
125
126            msg_type = get_dhcp_type(pkt)
127            if msg_type is None:
128                return
129
130            client_mac = pkt[sp.Ether].src
131            xid = pkt[sp.BOOTP].xid
132
133            if msg_type == "discover":
134                discover_count += 1
135                opts = extra_opts
136                if discover_count == 1 and first_extra_opts is not None:
137                    opts = first_extra_opts
138                reply = build_reply(
139                    ifname, self._SERVER_IP, self._CLIENT_IP, client_mac,
140                    xid, "offer", extra_opts=opts, sname=sname, file=file,
141                )
142                sp.sendp(reply, iface=ifname, verbose=0)
143
144            elif msg_type == "request":
145                reply = build_reply(
146                    ifname, self._SERVER_IP, self._CLIENT_IP, client_mac,
147                    xid, "ack", sname=sname, file=file,
148                )
149                sp.sendp(reply, iface=ifname, verbose=0)
150                done = True
151
152        sniffer = sp.AsyncSniffer(
153            iface=ifname,
154            filter="udp and dst port 67",
155            prn=handle_dhcp,
156            stop_filter=lambda _: done,
157            started_callback=lambda: self.send_object(
158                self.vnet_map["vnet1"].pipe, "go",
159            ),
160        )
161        sniffer.start()
162        sniffer.join(timeout=30)
163
164        assert done, "DHCP exchange did not complete"
165        result = self.wait_object(self.vnet_map["vnet1"].pipe, timeout=15)
166        return discover_count, result
167
168    @pytest.mark.require_user("root")
169    def test_classless_routes_truncated(self):
170        """
171        Verify that dhclient rejects a DHCPOFFER containing a truncated
172        classless static routes option.
173
174        A malformed option 121 payload is constructed with a valid first
175        entry (default route, width=0) followed by a second entry with
176        width=24 that has only 1 byte of subnet data instead of the
177        required 3 bytes.
178
179        The test sends the bad offer first, then a valid offer, and
180        verifies that dhclient rejects the bad offer and successfully
181        acquires a lease from the good one.
182        """
183        ifname = self._setup_server()
184        bad_routes = (
185            b"\x00"              # width=0 (default route)
186            b"\xc0\xa8\x01\x01"  # gateway: 192.168.1.1
187            b"\x18"              # width=24 (/24 subnet)
188            b"\x0a"              # only 1 of 3 required subnet octets
189        )
190        discover_count, result = self._do_dora(
191            ifname, first_extra_opts=[(121, bad_routes)],
192        )
193        assert discover_count >= 2, \
194            "Expected dhclient to reject the first (bad) offer and retry"
195        assert "inet {}".format(self._CLIENT_IP) in result["ifconfig"]
196
197    @pytest.mark.require_user("root")
198    def test_filename_backslash(self):
199        """
200        Verify that backslashes in the BOOTP filename don't cause the
201        lease to be rejected.  These can appear in Windows boot paths.
202        """
203        ifname = self._setup_server()
204        discover_count, result = self._do_dora(
205            ifname, file=b"\\windows\\boot\\pxeboot.com",
206        )
207        assert discover_count == 1, \
208            "Lease with backslash in filename should be accepted"
209        assert "inet {}".format(self._CLIENT_IP) in result["ifconfig"]
210
211    @pytest.mark.require_user("root")
212    def test_sname_valid(self):
213        """
214        Verify that a valid server name is preserved in the lease file.
215        """
216        ifname = self._setup_server()
217        discover_count, result = self._do_dora(
218            ifname, sname=b"dhcp.example.com",
219        )
220        assert discover_count == 1
221        assert "inet {}".format(self._CLIENT_IP) in result["ifconfig"]
222        assert "dhcp.example.com" in result["lease"]
223
224    @pytest.mark.require_user("root")
225    def test_sname_quote(self):
226        """
227        Verify that a double quote in the BOOTP server name causes the
228        server name to be discarded but does not cause the lease to be
229        rejected.  Values from the lease file may be handled by
230        dhclient-script and thus are susceptible to shell injection.
231        """
232        ifname = self._setup_server()
233        discover_count, result = self._do_dora(
234            ifname, sname=b'evil"server',
235        )
236        assert discover_count == 1, \
237            "Lease with quote in sname should be accepted"
238        assert "inet {}".format(self._CLIENT_IP) in result["ifconfig"]
239        assert 'evil"server' not in result["lease"], \
240            "Server name with quote should not appear in the lease file"
241        assert "server-name" not in result["lease"], \
242            "Unsafe server name should be discarded entirely"
243