Add interoperability and stress testing for CoAPSocket with libcoap

AI-Assisted: yes (GitHub CoPilot Auto)

Author: Nils Weiss

.config/ci/install.sh

@@ -36,6 +36,7 @@ then
   sudo apt-get update
   sudo apt-get -qy install tshark net-tools || exit 1
   sudo apt-get -qy install can-utils || exit 1
+  sudo apt-get -qy install libcoap3-bin || exit 1
   sudo apt-get -qy install linux-modules-extra-$(uname -r) || exit 1
   sudo apt-get -qy install samba smbclient
   sudo bash $CUR/openldap/install.sh

scapy/contrib/coap_socket.py

@@ -79,7 +79,7 @@ class CoAPSocket(SuperSocket):
     >>>         return CoAPSocket.empty_ack_params()
     >>> # Doesn't matter if it starts with "/dummy" or "dummy",
     >>> # but it is an error if it is in the end
-    >>> lst_resources = [DummyResource("dummy"), DelayedResource("/delayed")].
+    >>> lst_resources = [DummyResource("dummy"), DelayedResource("/delayed")]
     >>> with CoAPSocket("127.0.0.1", 5683, lst_resources=lst_resources) as coap_socket:
     >>>     while True:
     >>>         pkg = coap_socket.recv()
@@ -263,7 +263,8 @@ def __init__(self,
         self.url = url
         if self.url[0] != "/":
             self.url = "/" + self.url
-        self.description = description  # interface description ("if" in CoRE link-format)
+        # interface description ("if" in CoRE link-format)
+        self.description = description
         self.content_format = content_format  # ct
         self.resource_type = resource_type  # rt
         self.title = title  # title

test/contrib/coap_socket.uts

@@ -8,6 +8,10 @@ from scapy.contrib.coap_socket import *
 
 = Redirect logging
 import logging
+import shutil
+import socket
+import subprocess
+import time
 from scapy.error import log_runtime
 
 from io import StringIO
@@ -17,6 +21,73 @@ handler = logging.StreamHandler(log_stream)
 log_runtime.addHandler(handler)
 log_coap_sock.addHandler(handler)
 
+coap_client_bin = shutil.which("coap-client")
+coap_server_bin = shutil.which("coap-server")
+libcoap_available = coap_client_bin is not None and coap_server_bin is not None
+
+
+def _pick_port():
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
+    s.bind(("127.0.0.1", 0))
+    p = s.getsockname()[1]
+    s.close()
+    return p
+
+
+def _run_coap_client(uri, timeout=5):
+    p = subprocess.Popen(
+        [coap_client_bin, "-m", "get", "-B", "2", uri],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    try:
+        out, err = p.communicate(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        p.kill()
+        out, err = p.communicate()
+    return p.returncode, out, err
+
+
+def _start_coap_server(port):
+    p = subprocess.Popen(
+        [coap_server_bin, "-A", "127.0.0.1", "-p", str(port), "-v", "0"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    time.sleep(0.4)
+    return p
+
+
+def _cleanup_process(p):
+    if p is None or p.poll() is not None:
+        return
+    p.terminate()
+    try:
+        p.wait(timeout=2)
+    except subprocess.TimeoutExpired:
+        p.kill()
+        p.wait(timeout=2)
+
+
+class InteropResource(CoAPResource):
+    def get(self, payload, options, token, sa_ll):
+        return {
+            "type": ACK,
+            "code": CONTENT_205,
+            "options": [(CONTENT_FORMAT, CF_TEXT_PLAIN)],
+            "payload": b"scapy-libcoap-interop",
+        }
+
+
+class StressResource(CoAPResource):
+    def get(self, payload, options, token, sa_ll):
+        return {
+            "type": ACK,
+            "code": CONTENT_205,
+            "options": [(CONTENT_FORMAT, CF_TEXT_PLAIN)],
+            "payload": b"scapy-stress",
+        }
+
 + Testing client -> server interactions
 
 = Setup dummy resources
@@ -150,3 +221,81 @@ with CoAPSocket("127.0.0.1", 5683, lst_resources=lst_resources) as coap_server,
         assert rcv.code == CONTENT_205
         assert rcv.msg_id == pkt.msg_id
         assert rcv.token == pkt.token
+
++ Testing interoperability with libcoap
+
+= libcoap client -> CoAPSocket server
+
+if not libcoap_available:
+    print("Skipping libcoap interoperability tests: coap-client/coap-server not found")
+    assert True
+else:
+    interop_port = _pick_port()
+    with CoAPSocket(
+        "127.0.0.1",
+        interop_port,
+        lst_resources=[InteropResource("/interop")]
+    ) as coap_server:
+        rc, out, err = _run_coap_client("coap://127.0.0.1:%d/interop" % interop_port)
+        assert rc == 0, "coap-client failed rc=%s stderr=%r" % (rc, err)
+        assert b"scapy-libcoap-interop" in out, "Unexpected payload: %r" % out
+
+= CoAPSocket client -> libcoap server
+
+if not libcoap_available:
+    print("Skipping libcoap interoperability tests: coap-client/coap-server not found")
+    assert True
+else:
+    libcoap_port = _pick_port()
+    libcoap_server = _start_coap_server(libcoap_port)
+    try:
+        with CoAPSocket("127.0.0.1", _pick_port()) as coap_client:
+            req = CoAPSocket.make_coap_req_packet(uri=".well-known/core", payload=b"")
+            coap_client.send(IP(dst="127.0.0.1")/UDP(dport=libcoap_port)/req)
+            res = coap_client.recv()
+            assert res is not None
+            assert res.code == CONTENT_205
+            assert b"</" in res.payload.load
+            assert res.token == req.token
+    finally:
+        _cleanup_process(libcoap_server)
+
++ Stress tests
+
+= Stress: libcoap client -> CoAPSocket server
+
+if not libcoap_available:
+    print("Skipping libcoap interoperability tests: coap-client/coap-server not found")
+    assert True
+else:
+    stress_port = _pick_port()
+    with CoAPSocket("127.0.0.1", stress_port, lst_resources=[StressResource("/stress")]):
+        for i in range(25):
+            rc, out, err = _run_coap_client("coap://127.0.0.1:%d/stress" % stress_port)
+            assert rc == 0, "coap-client stress failed #%d rc=%s stderr=%r" % (
+                i, rc, err
+            )
+            assert b"scapy-stress" in out
+
+= Stress: CoAPSocket client -> libcoap server
+
+if not libcoap_available:
+    print("Skipping libcoap interoperability tests: coap-client/coap-server not found")
+    assert True
+else:
+    stress_libcoap_port = _pick_port()
+    stress_libcoap_server = _start_coap_server(stress_libcoap_port)
+    try:
+        with CoAPSocket("127.0.0.1", _pick_port()) as coap_client:
+            for _ in range(25):
+                req = CoAPSocket.make_coap_req_packet(uri=".well-known/core", payload=b"")
+                coap_client.send(
+                    IP(dst="127.0.0.1") / UDP(dport=stress_libcoap_port) / req
+                )
+                res = coap_client.recv()
+                assert res is not None
+                assert res.code == CONTENT_205
+                assert b"</" in res.payload.load
+                assert res.token == req.token
+    finally:
+        _cleanup_process(stress_libcoap_server)