ci: cram: add DHCPv4 Force option cram tests

Test the Force parameter of Device.DHCPv4.Server.Pool.{i}.Options.{i} to
force an option when ever it is not requested by the client

Closes: PCF-2462
Signed-off-by: Alexandre Fiset <alexandre.fiset@softathome.com>
This commit is contained in:
Alexandre Fiset 2026-03-31 16:28:02 +02:00 committed by FAOUZI YAHYA
parent 850b5641d1
commit 951c2d1d74
5 changed files with 299 additions and 1 deletions

1
.gitignore vendored
View File

@ -35,3 +35,4 @@ git-src
.cproject .cproject
.ccache .ccache
.vscode* .vscode*
__pycache__/

View File

@ -0,0 +1,54 @@
Create R alias:
$ alias R="${CRAM_REMOTE_COMMAND:-}"
Install scapy with pip:
$ pip install scapy >/dev/null 2>&1
Verify python can import scapy:
$ python3 -c "import scapy.all" >/dev/null 2>&1
Pick test constants:
$ TAG=42
$ VALUE_HEX=74657374
$ REQ_WITHOUT_TAG=1,3,6,15
$ REQ_WITH_TAG=1,3,6,15,42
$ PROBE_TIMEOUT=12
Create pool option with Force=true:
$ R "ba-cli 'Device.DHCPv4.Server.Pool.lan.Option.+{Alias=\"cpe-force-opt\",Enable=1,Tag=${TAG},Value=\"${VALUE_HEX}\",Force=1}'" | grep -Ev '^(>|$)'
Device\.DHCPv4\.Server\.Pool\.1\.Option\.[0-9]+\. (re)
Device\.DHCPv4\.Server\.Pool\.1\.Option\.[0-9]+\.Alias="cpe-force-opt" (re)
$ sleep 1
Force=true, tag 42 absent from DHCPREQUEST PRL -> tag must be present in final DHCPACK:
$ sudo -E python3 "$TESTDIR/dhcpv4_option_probe.py" --iface "$TESTBED_LAN_INTERFACE" --request-options "$REQ_WITHOUT_TAG" --timeout "$PROBE_TIMEOUT" 2>/dev/null
RECEIVED_TAGS=.*(=|,)42(,|$).* (re)
Set Force=false:
$ R "ba-cli 'Device.DHCPv4.Server.Pool.lan.Option.cpe-force-opt.Force=0'" | grep -Ev '^(>|$)'
Device\.DHCPv4\.Server\.Pool\.1\.Option\.[0-9]+\. (re)
Device\.DHCPv4\.Server\.Pool\.1\.Option\.[0-9]+\.Force=0 (re)
$ sleep 1
Force=false, tag 42 absent from DHCPREQUEST PRL -> tag must be absent from final DHCPACK:
$ sudo -E python3 "$TESTDIR/dhcpv4_option_probe.py" --iface "$TESTBED_LAN_INTERFACE" --request-options "$REQ_WITHOUT_TAG" --timeout "$PROBE_TIMEOUT" 2>/dev/null
RECEIVED_TAGS=(?!.*(=|,)42(,|$)).* (re)
Force=false, tag 42 present in DHCPREQUEST PRL -> tag must be present in final DHCPACK:
$ sudo -E python3 "$TESTDIR/dhcpv4_option_probe.py" --iface "$TESTBED_LAN_INTERFACE" --request-options "$REQ_WITH_TAG" --timeout "$PROBE_TIMEOUT" 2>/dev/null
RECEIVED_TAGS=.*(=|,)42(,|$).* (re)
Cleanup:
$ R "ba-cli 'Device.DHCPv4.Server.Pool.lan.Option.cpe-force-opt.-'" | grep -Ev '^(>|$)'
Device\.DHCPv4\.Server\.Pool\.1\.Option\.[0-9]+\. (re)

View File

@ -0,0 +1,76 @@
# DHCPv4 Option Probe Client (Python)
This directory contains the Python DHCPv4 probe client:
- `dhcpv4_option_probe.py`
## What the client checks
The client performs the DHCP exchange:
- `DHCPDISCOVER`
- `DHCPOFFER`
- `DHCPREQUEST`
- `DHCPACK`
Option checks are based on:
- options requested in `DHCPREQUEST` (`param_req_list`)
- options received in the final `DHCPACK`
The output is extracted from the final `DHCPACK` only.
## Requirements
- `python3`
- `scapy`
- root privileges (raw packet send/receive)
Install dependency:
```bash
pip install scapy
```
## Usage
```bash
sudo -E python3 dhcpv4_option_probe.py --iface <interface> --request-options "1,3,6,15,42"
```
Default output format:
```text
RECEIVED_TAGS=1,3,6,15,42
```
## Debug mode
Pass `--debug` to print the full scapy packet dump for each DHCP exchange to stderr:
```bash
sudo -E python3 dhcpv4_option_probe.py --iface <interface> --request-options "1,3,6,15,42" --debug
```
The debug output includes:
- `--- DHCPDISCOVER sent ---` — full packet before sending
- `--- DHCPOFFER received ---` — full packet from server
- `--- DHCPREQUEST sent ---` — full packet before sending
- `--- DHCPACK received ---` — full packet from server
- `DEBUG_DHCPREQUEST_PRL=<tags>` — the parameter request list sent in DHCPREQUEST (stdout)
Debug output goes to stderr so the `RECEIVED_TAGS=` line on stdout is unaffected.
## Real-environment recommendation (network delay tolerance)
Use a higher timeout when network delays are expected:
```bash
sudo -E python3 dhcpv4_option_probe.py --iface <interface> --request-options "1,3,6,15,42" --timeout 12
```
## Exit codes
- `0`: success (`DHCPACK` received and parsed)
- `2`: error (`DHCPOFFER`/`DHCPACK` missing, `DHCPNAK`, or invalid DHCP response)

View File

@ -0,0 +1,167 @@
#!/usr/bin/env python3
"""
Minimal DHCPv4 probe for Cram:
- Sends DHCPDISCOVER with configurable Parameter Request List (PRL)
- Sends DHCPREQUEST using the offered address and server identifier
- Prints received option tags from final DHCPACK as:
RECEIVED_TAGS=1,3,6,...
- Optional debug output can print the PRL sent in DHCPREQUEST
"""
import argparse
import random
import sys
from scapy.all import BOOTP, DHCP, Ether, IP, UDP, conf, srp1 # type: ignore
from scapy.layers.dhcp import DHCPRevOptions # type: ignore
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--iface", required=True, help="Interface used to send/receive DHCP")
parser.add_argument("--request-options", default="", help="CSV option tags in PRL, e.g. 1,3,6,15,42")
parser.add_argument("--timeout", type=int, default=8)
parser.add_argument("--debug", action="store_true")
return parser.parse_args()
def parse_prl(csv_text):
if not csv_text.strip():
return []
return [int(value.strip()) for value in csv_text.split(",") if value.strip()]
def extract_option_tags(dhcp_options):
tags = []
for opt in dhcp_options:
if not (isinstance(opt, tuple) and len(opt) >= 1):
continue
name = opt[0]
if isinstance(name, int):
tags.append(name)
continue
if isinstance(name, str):
code = DHCPRevOptions.get(name)
if isinstance(code, int):
tags.append(code)
elif isinstance(code, tuple) and len(code) >= 1 and isinstance(code[0], int):
tags.append(code[0])
return sorted(set(tags))
def get_option_value(dhcp_options, option_name):
for opt in dhcp_options:
if isinstance(opt, tuple) and len(opt) >= 2 and opt[0] == option_name:
return opt[1]
return None
def get_message_type(dhcp_options):
value = get_option_value(dhcp_options, "message-type")
if isinstance(value, int):
return value
if isinstance(value, str):
value_map = {
"discover": 1,
"offer": 2,
"request": 3,
"decline": 4,
"ack": 5,
"nak": 6,
"release": 7,
"inform": 8,
}
return value_map.get(value.lower())
return None
def main():
args = parse_args()
conf.checkIPaddr = False
xid = random.randint(1, 0xFFFFFFFF)
prl = parse_prl(args.request_options)
discover_packet = (
Ether(dst="ff:ff:ff:ff:ff:ff")
/ IP(src="0.0.0.0", dst="255.255.255.255")
/ UDP(sport=68, dport=67)
/ BOOTP(chaddr=b"\xaa\xbb\xcc\xdd\xee\xff", xid=xid, flags=0x8000)
/ DHCP(options=[
("message-type", "discover"),
("param_req_list", prl),
"end",
])
)
if args.debug:
print("--- DHCPDISCOVER sent ---", file=sys.stderr)
print(discover_packet.show(dump=True), file=sys.stderr)
offer = srp1(discover_packet, iface=args.iface, timeout=args.timeout, verbose=False)
if offer is None or not offer.haslayer(DHCP):
print("ERROR: no DHCPOFFER received", file=sys.stderr)
return 2
if args.debug:
print("--- DHCPOFFER received ---", file=sys.stderr)
print(offer.show(dump=True), file=sys.stderr)
if get_message_type(offer[DHCP].options) != 2:
print("ERROR: received DHCP response is not DHCPOFFER", file=sys.stderr)
return 2
offered_ip = offer[BOOTP].yiaddr
server_identifier = get_option_value(offer[DHCP].options, "server_id")
request_options = [
("message-type", "request"),
("requested_addr", offered_ip),
("param_req_list", prl),
]
if args.debug:
print("DEBUG_DHCPREQUEST_PRL=" + ",".join(str(tag) for tag in prl))
if server_identifier is not None:
request_options.append(("server_id", server_identifier))
request_options.append("end")
request_packet = (
Ether(dst="ff:ff:ff:ff:ff:ff")
/ IP(src="0.0.0.0", dst="255.255.255.255")
/ UDP(sport=68, dport=67)
/ BOOTP(chaddr=b"\xaa\xbb\xcc\xdd\xee\xff", xid=xid, flags=0x8000)
/ DHCP(options=request_options)
)
if args.debug:
print("--- DHCPREQUEST sent ---", file=sys.stderr)
print(request_packet.show(dump=True), file=sys.stderr)
final_reply = srp1(request_packet, iface=args.iface, timeout=args.timeout, verbose=False)
if final_reply is None or not final_reply.haslayer(DHCP):
print("ERROR: no DHCPACK received", file=sys.stderr)
return 2
if args.debug:
print("--- DHCPACK received ---", file=sys.stderr)
print(final_reply.show(dump=True), file=sys.stderr)
message_type = get_message_type(final_reply[DHCP].options)
if message_type == 6:
print("ERROR: DHCPNAK received", file=sys.stderr)
return 2
if message_type != 5:
print("ERROR: final DHCP response is not DHCPACK", file=sys.stderr)
return 2
tags = extract_option_tags(final_reply[DHCP].options)
print("RECEIVED_TAGS=" + ",".join(str(tag) for tag in tags))
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@ -224,9 +224,9 @@ Device.DHCPv4.Server.Pool.{i}.MinAddress
Device.DHCPv4.Server.Pool.{i}.Option.{i}. proto::dhcpv4-manager Device.DHCPv4.Server.Pool.{i}.Option.{i}. proto::dhcpv4-manager
Device.DHCPv4.Server.Pool.{i}.Option.{i}.Alias proto::dhcpv4-manager Device.DHCPv4.Server.Pool.{i}.Option.{i}.Alias proto::dhcpv4-manager
Device.DHCPv4.Server.Pool.{i}.Option.{i}.Enable proto::dhcpv4-manager Device.DHCPv4.Server.Pool.{i}.Option.{i}.Enable proto::dhcpv4-manager
Device.DHCPv4.Server.Pool.{i}.Option.{i}.Force proto::dhcpv4-manager
Device.DHCPv4.Server.Pool.{i}.Option.{i}.Tag proto::dhcpv4-manager Device.DHCPv4.Server.Pool.{i}.Option.{i}.Tag proto::dhcpv4-manager
Device.DHCPv4.Server.Pool.{i}.Option.{i}.Value proto::dhcpv4-manager Device.DHCPv4.Server.Pool.{i}.Option.{i}.Value proto::dhcpv4-manager
Device.DHCPv4.Server.Pool.{i}.Option.{i}.X_PRPLWARE-COM_Force proto::dhcpv4-manager
Device.DHCPv4.Server.Pool.{i}.OptionNumberOfEntries proto::dhcpv4-manager Device.DHCPv4.Server.Pool.{i}.OptionNumberOfEntries proto::dhcpv4-manager
Device.DHCPv4.Server.Pool.{i}.Order proto::dhcpv4-manager Device.DHCPv4.Server.Pool.{i}.Order proto::dhcpv4-manager
Device.DHCPv4.Server.Pool.{i}.ReservedAddresses proto::dhcpv4-manager Device.DHCPv4.Server.Pool.{i}.ReservedAddresses proto::dhcpv4-manager