[thread-cert] add case 8.1.6 using pktverify (#5898)

2026-07-30 15:47:46 +00:00 · 2020-12-14 23:24:43 -08:00
parent 09292f5d12
commit 69e97581e7
5 changed files with 269 additions and 12 deletions
@@ -0,0 +1,241 @@
+#!/usr/bin/env python3
+#
+#  Copyright (c) 2020, The OpenThread Authors.
+#  All rights reserved.
+#
+#  Redistribution and use in source and binary forms, with or without
+#  modification, are permitted provided that the following conditions are met:
+#  1. Redistributions of source code must retain the above copyright
+#     notice, this list of conditions and the following disclaimer.
+#  2. Redistributions in binary form must reproduce the above copyright
+#     notice, this list of conditions and the following disclaimer in the
+#     documentation and/or other materials provided with the distribution.
+#  3. Neither the name of the copyright holder nor the
+#     names of its contributors may be used to endorse or promote products
+#     derived from this software without specific prior written permission.
+#
+#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+#  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+#  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+#  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+#  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+#  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+#  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+#  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+#  POSSIBILITY OF SUCH DAMAGE.
+#
+
+import unittest
+
+import command
+from mesh_cop import MeshCopState
+import thread_cert
+from pktverify.consts import MLE_DISCOVERY_REQUEST, MLE_DISCOVERY_RESPONSE, HANDSHAKE_CLIENT_HELLO, HANDSHAKE_SERVER_HELLO, HANDSHAKE_SERVER_KEY_EXCHANGE, HANDSHAKE_SERVER_HELLO_DONE, HANDSHAKE_CLIENT_KEY_EXCHANGE, HANDSHAKE_HELLO_VERIFY_REQUEST, CONTENT_APPLICATION_DATA, NM_EXTENDED_PAN_ID_TLV, NM_NETWORK_NAME_TLV, NM_STEERING_DATA_TLV, NM_COMMISSIONER_UDP_PORT_TLV, NM_JOINER_UDP_PORT_TLV, NM_DISCOVERY_REQUEST_TLV, NM_DISCOVERY_RESPONSE_TLV, THREAD_DISCOVERY_TLV, CONTENT_CHANGE_CIPHER_SPEC, CONTENT_HANDSHAKE, CONTENT_ALERT
+from pktverify.packet_verifier import PacketVerifier
+
+COMMISSIONER = 1
+JOINER = 2
+PSKD = 'PSKD01'
+URL_1 = 'www.openthread.org'
+URL_2 = 'www.wrongurl.org'
+
+# Test Purpose and Description:
+# -----------------------------
+# The purpose of this test case is to verify the DTLS session between an on-mesh
+# Commissioner and a Joiner and ensure that the session does not stay open.
+#
+# Test Topology:
+# -------------
+# Commissioner
+#    |
+# Joiner
+#
+# DUT Types:
+# ----------
+#  Commissioner
+#  Joiner
+
+
+class Cert_8_1_06_Commissioning(thread_cert.TestCase):
+    SUPPORT_NCP = False
+
+    TOPOLOGY = {
+        COMMISSIONER: {
+            'name': 'COMMISSIONER',
+            'masterkey': '00112233445566778899aabbccddeeff',
+            'mode': 'rdn',
+            'panid': 0xface
+        },
+        JOINER: {
+            'masterkey': 'deadbeefdeadbeefdeadbeefdeadbeef',
+            'mode': 'rdn',
+            'router_selection_jitter': 1
+        },
+    }
+
+    def test(self):
+        self.nodes[COMMISSIONER].interface_up()
+        self.nodes[COMMISSIONER].thread_start()
+        self.simulator.go(5)
+        self.assertEqual(self.nodes[COMMISSIONER].get_state(), 'leader')
+        self.nodes[COMMISSIONER].commissioner_set_provisioning_url(URL_1)
+        self.nodes[COMMISSIONER].commissioner_start()
+        self.simulator.go(3)
+        self.nodes[COMMISSIONER].commissioner_add_joiner(self.nodes[JOINER].get_eui64(), PSKD)
+
+        self.nodes[JOINER].interface_up()
+        self.nodes[JOINER].joiner_start(PSKD, URL_2)
+        self.simulator.go(10)
+        self.simulator.read_cert_messages_in_commissioning_log([COMMISSIONER, JOINER])
+        self.assertEqual(
+            self.nodes[JOINER].get_masterkey(),
+            self.nodes[COMMISSIONER].get_masterkey(),
+        )
+        joiner_messages = self.simulator.get_messages_sent_by(JOINER)
+        commissioner_messages = self.simulator.get_messages_sent_by(COMMISSIONER)
+
+        # 3.7, 8, 9, 10
+        # - Joiner
+        command.check_joiner_commissioning_messages(joiner_messages.commissioning_messages, URL_2)
+        # - Commissioner
+        command.check_commissioner_commissioning_messages(commissioner_messages.commissioning_messages,
+                                                          MeshCopState.REJECT)
+        # As commissioner is also joiner router
+        command.check_joiner_router_commissioning_messages(commissioner_messages.commissioning_messages)
+        self.nodes[JOINER].thread_start()
+        self.simulator.go(5)
+        self.assertEqual(self.nodes[JOINER].get_state(), 'router')
+
+    def verify(self, pv):
+        pkts = pv.pkts
+        pv.summary.show()
+
+        COMMISSIONER = pv.vars['COMMISSIONER']
+        COMMISSIONER_VERSION = pv.vars['COMMISSIONER_VERSION']
+
+        _rs_pkt = pkts.filter_wpan_src64(COMMISSIONER).\
+            filter_mle_cmd(MLE_DISCOVERY_RESPONSE).\
+            filter(lambda p: {
+                              NM_EXTENDED_PAN_ID_TLV,
+                              NM_NETWORK_NAME_TLV,
+                              NM_STEERING_DATA_TLV,
+                              NM_COMMISSIONER_UDP_PORT_TLV,
+                              NM_JOINER_UDP_PORT_TLV,
+                              NM_DISCOVERY_RESPONSE_TLV
+                            } == set(p.thread_meshcop.tlv.type) and\
+                   p.thread_meshcop.tlv.discovery_rsp_ver ==
+                   COMMISSIONER_VERSION
+                  ).\
+            must_next()
+
+        # Step 3: Verify the following details occur in the exchange between
+        #         Joiner and the Commissioner
+        # 1. UDP port (Specified by the Commissioner: in Discovery Response)
+        #    is used as destination port for UDP datagrams from Joiner to
+        #    the Commissioner.
+
+        # 2. Joiner_1 sends an initial DTLS-ClientHello handshake record to the
+        #    Commissioner
+        pkts.filter_wpan_dst64(COMMISSIONER).\
+            filter(lambda p:
+                   p.dtls.handshake.type == [HANDSHAKE_CLIENT_HELLO] and\
+                   p.udp.srcport in _rs_pkt.thread_meshcop.tlv.udp_port and\
+                   p.udp.dstport in _rs_pkt.thread_meshcop.tlv.udp_port
+                   ).\
+            must_next()
+
+        # 3. The Commissioner receives the initial DTLS-ClientHello handshake
+        #    record and sends a DTLS-HelloVerifyRequest handshake record Joiner
+        _pkt = pkts.filter_wpan_src64(COMMISSIONER).\
+            filter(lambda p: p.dtls.handshake.type == [HANDSHAKE_HELLO_VERIFY_REQUEST]).\
+            must_next()
+        _pkt.must_verify(lambda p: p.dtls.handshake.cookie is not None)
+
+        # 4. Joiner receives the DTLS-HelloVerifyRequest handshake record and sends
+        #    a subsequent DTLS-ClientHello handshake record in one UDP datagram to the
+        #    Commissioner
+        #      Verify that both DTLS-HelloVerifyRequest and subsequent DTLS-ClientHello
+        #      contain the same cookie
+        pkts.filter_wpan_dst64(COMMISSIONER).\
+            filter(lambda p:
+                       p.dtls.handshake.type == [HANDSHAKE_CLIENT_HELLO] and\
+                       p.dtls.handshake.cookie == _pkt.dtls.handshake.cookie
+                       ).\
+            must_next()
+
+        # 5. Commissioner receives the subsequent DTLSClientHello handshake record
+        #    and then send, in order, DTLSServerHello, DTLS-ServerKeyExchange and
+        #    DTLSServerHelloDone handshake records to Joiner
+        pkts.filter_wpan_src64(COMMISSIONER).\
+            filter(lambda p:
+                      p.dtls.handshake.type == [HANDSHAKE_SERVER_HELLO,
+                                                HANDSHAKE_SERVER_KEY_EXCHANGE,
+                                                HANDSHAKE_SERVER_HELLO_DONE]
+                      ).\
+               must_next()
+
+        # 6. Joiner receives the DTLS-ServerHello, DTLSServerKeyExchange and
+        #    DTLS-ServerHelloDone handshake records and sends, in order,
+        #    a DTLS-ClientKeyExchange handshake record,
+        #    a DTLS-ChangeCipherSpec record and
+        #    an encrypted DTLS-Finished handshake record to the Commissioner.
+        pkts.filter_wpan_dst64(COMMISSIONER).\
+            filter(lambda p:
+                       p.dtls.handshake.type == [HANDSHAKE_CLIENT_KEY_EXCHANGE] and\
+                       {
+                        CONTENT_CHANGE_CIPHER_SPEC,
+                        CONTENT_HANDSHAKE
+                       } == set(p.dtls.record.content_type)
+                       ).\
+               must_next()
+
+        # 7. Commissioner receives the DTLS-ClientKeyExchange handshake record, the
+        #    DTLS-ChangeCipherSpec record and the encrypted DTLS-Finished handshake record,
+        #    and sends a DTLS-ChangeCipherSpec record and an encrypted DTLSFinished handshake
+        #    record in that order to Joiner
+        pkts.filter_wpan_src64(COMMISSIONER).\
+            filter(lambda p: {
+                              CONTENT_CHANGE_CIPHER_SPEC,
+                              CONTENT_HANDSHAKE
+                             } == set(p.dtls.record.content_type)
+                      ).\
+              must_next()
+
+        # 8. Joiner receives the DTLS-ChangeCipherSpec record and the encrypted DTLS-Finished
+        #    handshake record and sends a JOIN_FIN.req message in an encrypted DTLS-ApplicationData
+        #    record in a single UDP datagram to Commissioner.
+        pkts.filter_wpan_dst64(COMMISSIONER).\
+            filter(lambda p:
+                   [CONTENT_APPLICATION_DATA] == p.dtls.record.content_type
+                  ).\
+               must_next()
+
+        # 9. Commissioner receives the encrypted DTLS-ApplicationData record and sends a
+        #    JOIN_FIN.rsp message in an encrypted DTLS-ApplicationData record in a single
+        #    UDP datagram to Joiner
+        #    The JOIN_FIN.req message must contain a Provisioning URL TLV which the
+        #    Commissioner will not recognize.
+        pkts.filter_wpan_src64(COMMISSIONER).\
+            filter(lambda p:
+                   [CONTENT_APPLICATION_DATA] == p.dtls.record.content_type
+                  ).\
+               must_next()
+
+        # 10. Joiner receives the encrypted JOIN_ENT.ntf message and sends an encrypted
+        #     JOIN_ENT.ntf with Reject state to Commissioner
+
+        # Verify Step 7 - 10 in test()
+
+        # 11. Joiner sends an encrypted DTLS-Alert record with a code of 0 (close_notify)
+        #     to Commissioner
+        pkts.filter_wpan_dst64(COMMISSIONER).\
+            filter(lambda p:
+                   [CONTENT_ALERT] == p.dtls.record.content_type
+                  ).\
+               must_next()
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -643,28 +643,41 @@ def get_joiner_udp_port_in_discovery_response(command_msg):
    return udp_port_tlv.udp_port


-def check_joiner_commissioning_messages(commissioning_messages):
+def check_joiner_commissioning_messages(commissioning_messages, url=''):
    """Verify COAP messages sent by joiner while commissioning process.
    """
    print(commissioning_messages)
-    assert len(commissioning_messages) >= 2
+    assert len(commissioning_messages) >= 4
    join_fin_req = commissioning_messages[0]
    assert join_fin_req.type == mesh_cop.MeshCopMessageType.JOIN_FIN_REQ
-    assert_contains_tlv(join_fin_req.tlvs, CheckType.NOT_CONTAIN, mesh_cop.ProvisioningUrl)
-    join_ent_rsp = commissioning_messages[1]
+    if url:
+        provisioning_url = assert_contains_tlv(join_fin_req.tlvs, CheckType.CONTAIN, mesh_cop.ProvisioningUrl)
+        assert url == provisioning_url.url
+    else:
+        assert_contains_tlv(join_fin_req.tlvs, CheckType.NOT_CONTAIN, mesh_cop.ProvisioningUrl)
+
+    join_ent_rsp = commissioning_messages[3]
    assert join_ent_rsp.type == mesh_cop.MeshCopMessageType.JOIN_ENT_RSP


-def check_commissioner_commissioning_messages(commissioning_messages):
+def check_commissioner_commissioning_messages(commissioning_messages, state=mesh_cop.MeshCopState.ACCEPT):
    """Verify COAP messages sent by commissioner while commissioning process.
    """
-    assert any(msg.type == mesh_cop.MeshCopMessageType.JOIN_FIN_RSP for msg in commissioning_messages)
+    assert len(commissioning_messages) >= 2
+    join_fin_rsq = commissioning_messages[1]
+    assert join_fin_rsq.type == mesh_cop.MeshCopMessageType.JOIN_FIN_RSP
+    rsq_state = assert_contains_tlv(join_fin_rsq.tlvs, CheckType.CONTAIN, mesh_cop.State)
+    assert rsq_state.state == state


 def check_joiner_router_commissioning_messages(commissioning_messages):
    """Verify COAP messages sent by joiner router while commissioning process.
    """
-    assert any(msg.type == mesh_cop.MeshCopMessageType.JOIN_ENT_NTF for msg in commissioning_messages)
+    if len(commissioning_messages) >= 4:
+        join_ent_ntf = commissioning_messages[2]
+    else:
+        join_ent_ntf = commissioning_messages[0]
+    assert join_ent_ntf.type == mesh_cop.MeshCopMessageType.JOIN_ENT_NTF
    return None


@@ -551,7 +551,7 @@ class ProvisioningUrl(object):
 class ProvisioningUrlFactory:

    def parse(self, data, message_info):
-        url = data.decode('utf-8')
+        url = data.getvalue().decode('utf-8')
        return ProvisioningUrl(url)


@@ -570,7 +570,6 @@ class NodeImpl:
                data = [int(hex, 16) for hex in res.group(0)[1:-1].split(b' ') if hex and hex != b'..']
                payload += bytearray(data)
                log = log[res.end() - 1:]
-
        assert len(payload) == payload_len
        return (direction, type, payload)

@@ -639,6 +638,11 @@ class NodeImpl:
        self.send_command(cmd)
        self._expect('Done')

+    def commissioner_set_provisioning_url(self, provisioning_url=''):
+        cmd = 'commissioner provisioningurl %s' % provisioning_url
+        self.send_command(cmd)
+        self._expect('Done')
+
    def joiner_start(self, pskd='', provisioning_url=''):
        cmd = 'joiner start %s %s' % (pskd, provisioning_url)
        self.send_command(cmd)
@@ -85,9 +85,8 @@ class BaseSimulator(object):
                    type,
                    payload,
            ) in node.read_cert_messages_in_commissioning_log():
-                if direction == b'send':
-                    msg = self._payload_parse_factory.parse(type.decode("utf-8"), io.BytesIO(payload))
-                    self.commissioning_messages[nodeid].append(msg)
+                msg = self._payload_parse_factory.parse(type.decode("utf-8"), io.BytesIO(payload))
+                self.commissioning_messages[nodeid].append(msg)


 class RealTime(BaseSimulator):