RIOT/tests/gnrc_dhcpv6_client_6lbr/tests-as-root/01-run.py

#!/usr/bin/env python3

# Copyright (C) 2018 Freie Universität Berlin
#
# This file is subject to the terms and conditions of the GNU Lesser
# General Public License v2.1. See the file LICENSE in the top level
# directory for more details.

import os
import pexpect
import random
import sys
import time

from scapy.all import AsyncSniffer, sendp, Ether, IPv6, UDP
from scapy.all import DHCP6_Solicit, DHCP6_Advertise, DHCP6_Request, DHCP6_Reply
from scapy.all import DHCP6OptClientId, DHCP6OptServerId, DHCP6OptIA_PD
from scapy.all import DUID_LL, DHCP6OptIAPrefix
from testrunner import run

try:
    from scapy.all import DHCP6OptMudUrl
    mud_option_loaded = True
except ImportError:
    from scapy.all import DHCP6OptUnknown
    DHCP6OptMudUrl = DHCP6OptUnknown
    mud_option_loaded = False

TIMEOUT = 1

MUD_OPTION_CODE = 112
MUD_TEST_URL = b'https://example.org'


def get_upstream_netif(child):
    child.sendline("ifconfig")
    child.sendline("help")  # workaround to spot end of ifconfig output
    candidate = None
    while True:     # Search for an interface that does _not_ contain 6LO flag
        if candidate is None:
            child.expect(r"Iface\s+([^\s]+)\s+")
            candidate = child.match.group(1)
        res = child.expect([r"\b6LO\b", r"Iface\s+([^\s]+)\s+", "Command",
                            pexpect.TIMEOUT], timeout=.2)
        if res > 0:
            break
        candidate = None
    # wait for a line in "help"
    child.expect("reboot")
    return candidate


def get_downstream_netif(child):
    child.sendline("ifconfig")
    child.sendline("help")  # workaround to spot end of ifconfig output
    candidate = None
    while True:     # Search for an interface that does _not_ contain 6LO flag
        if candidate is None:
            child.expect(r"Iface\s+([^\s]+)\s+")
        candidate = child.match.group(1)
        res = child.expect([r"\b6LO\b", r"Iface\s+([^\s]+)\s+", "Command",
                            pexpect.TIMEOUT], timeout=.2)
        if res == 0:
            break
        elif res == 1:
            candidate = child.match.group(1)
        elif res == 2:
            break
        else:
            candidate = None
    # wait for a line in "help"
    child.expect("reboot")
    return candidate


def get_hwaddrs(child, netif):
    hwaddrs = []
    child.sendline("ifconfig {}".format(netif))
    child.expect(r"HWaddr:\s+(([A-Fa-f0-9]{2}:?)+)\s")
    hwaddrs.append(child.match.group(1).lower())
    if len(hwaddrs[0]) == 5:    # short address
        res = child.expect([pexpect.TIMEOUT,
                            r"Long HWaddr:\s+(([A-Fa-f0-9]{2}:?)+)\s"])
        if res > 0:
            hwaddrs.append(child.match.group(1).lower())
    return hwaddrs


def start_sniffer(iface, count=None, stop_filter=None):
    sniffer = AsyncSniffer(
        iface=iface,
        filter="udp and dst port 547",
        count=count,
        stop_filter=stop_filter,
    )
    sniffer.start()
    return sniffer


def wait_for_dhcpv6_pkt(iface, sniffer=None, timeout=5):
    if sniffer is None:
        sniffer = start_sniffer(iface, count=1)
    sniffer.join(timeout=timeout)
    if sniffer.results is None:
        raise TimeoutError("Sniffing for DHCPv6 traffic timed out")
    return [p for p in sniffer.results
            # filter out packets only belonging to stop_filter if it existed
            if sniffer.kwargs.get("stop_filter") is None or
            sniffer.kwargs["stop_filter"](p)][-1]


def build_reply_headers(pkt):
    src_ether = pkt[Ether].src
    src_ip = pkt[IPv6].src
    sport = pkt[UDP].sport
    dport = pkt[UDP].dport
    return Ether(dst=src_ether) / IPv6(dst=src_ip) / \
        UDP(sport=dport, dport=sport)


def testfunc(child):
    iface = os.environ["TAP"]

    pkt = wait_for_dhcpv6_pkt(iface)
    # the packet was a solicit
    assert DHCP6_Solicit in pkt
    # check if the sender is the upstream interface of the node
    upstream_netif = get_upstream_netif(child)
    print(upstream_netif)
    upstream_hwaddrs = get_hwaddrs(child, upstream_netif)
    assert DHCP6OptClientId in pkt and DUID_LL in pkt[DHCP6OptClientId].duid
    assert pkt[DHCP6OptClientId].duid[DUID_LL].lladdr in upstream_hwaddrs
    # and it is asking for a prefix delegation
    assert DHCP6OptIA_PD in pkt

    # reply to solicit with advertise and a prefix provided
    trid = pkt[DHCP6_Solicit].trid
    srv_duid = "aa:bb:cc:dd:ee:ff"
    cli_id = DHCP6OptClientId(duid=pkt[DHCP6OptClientId].duid)
    srv_id = DHCP6OptServerId(duid=DUID_LL(lladdr=srv_duid))
    prefix = "2001:db8:{:x}:{:x}::".format(
            random.randint(0, 0xffff),
            random.randint(0, 0xffff)
        )
    ia_pd = DHCP6OptIA_PD(T1=12000, T2=13000, iaid=pkt[DHCP6OptIA_PD].iaid,
                          iapdopt=[
                              DHCP6OptIAPrefix(preflft=14000, validlft=15000,
                                               prefix=prefix, plen=64)])
    # start sniffer to catch incoming request
    sniffer = start_sniffer(iface,
                            stop_filter=lambda pkt: DHCP6_Request in pkt)
    sendp(build_reply_headers(pkt) / DHCP6_Advertise(trid=trid) /
          ia_pd / cli_id / srv_id, iface=iface, verbose=False)

    # wait for request
    pkt = wait_for_dhcpv6_pkt(iface, sniffer)
    # the packet was indeed a request
    assert DHCP6_Request in pkt
    # and from the client
    assert DHCP6OptClientId in pkt and DUID_LL in pkt[DHCP6OptClientId].duid
    assert pkt[DHCP6OptClientId].duid[DUID_LL].lladdr in upstream_hwaddrs
    # and it is trying to talk to this server
    assert DHCP6OptServerId in pkt and DUID_LL in pkt[DHCP6OptServerId].duid
    assert pkt[DHCP6OptServerId].duid[DUID_LL].lladdr == srv_duid
    # and is still asking for a prefix delegation
    assert DHCP6OptIA_PD in pkt

    assert DHCP6OptMudUrl in pkt
    mud_option = pkt[DHCP6OptMudUrl]
    assert mud_option.optlen == len(MUD_TEST_URL)

    if mud_option_loaded:
        assert mud_option.mudstring == MUD_TEST_URL
    else:
        assert mud_option.optcode == MUD_OPTION_CODE
        assert mud_option.data == MUD_TEST_URL

    # reply to request with reply and a prefix provided
    trid = pkt[DHCP6_Request].trid
    sendp(build_reply_headers(pkt) / DHCP6_Reply(trid=trid) /
          ia_pd / cli_id / srv_id, iface=iface, verbose=False)
    time.sleep(1)

    # check if global address was configured
    child.sendline("ifconfig {}".format(get_downstream_netif(child)))
    # remove one trailing ':' from prefix just to be safe ;-)
    child.expect(r"inet6 addr:\s+{}[0-9a-fA-F:]+\s"
                 .format(prefix[:-1]))
    print("SUCCESS")


if __name__ == "__main__":
    sys.exit(run(testfunc, timeout=TIMEOUT, echo=True))