Unit tests for SR and SR1, added hash functions for the CoAP packet.

Author: eHonnef

scapy/contrib/coap.py

@@ -278,6 +278,12 @@ def post_dissect(self, pay):
  self.content_format = k[1]
  return pay
 
+ def hashret(self):
+ return struct.pack('I', self.msg_id) + self.token
+
+ def answers(self, other):
+ return True
+
 
 bind_layers(UDP, CoAP, sport=5683)
 bind_layers(UDP, CoAP, dport=5683)

scapy/contrib/coap_socket.py

@@ -152,6 +152,20 @@ def send(self, x):
  self.impl.send(x.dst, x.dport, x[CoAP])
  return len(x)
 
+ @staticmethod
+ def select(sockets, remain=None):
+ # type: (list[SuperSocket], Optional[float]) -> list[SuperSocket]
+ """This function is called during sendrecv() routine to wait for
+ sockets to be ready to receive
+ """
+ obj_pipes = [x.impl.rx_queue for x in sockets if
+ isinstance(x, CoAPSocket) and not x.closed]
+
+ ready_pipes = select_objects(obj_pipes, 0)
+
+ return [x for x in sockets if isinstance(x, CoAPSocket) and
+ not x.closed and x.impl.rx_queue in ready_pipes]
+
  @staticmethod
  def make_coap_req_packet(method=GET, uri="", options=None, payload=b""):
  # type: (int, str, list[tuple], bytes) -> Packet

scapy/sendrecv.py

@@ -243,7 +243,8 @@ def _sndrcv_snd(self):
  # Populate the dictionary of _sndrcv_rcv
  # _sndrcv_rcv won't miss the answer of a packet that
  # has not been sent
- self.hsent.setdefault(p.hashret(), []).append(p)
+ h = p[self.pks.basecls]
+ self.hsent.setdefault(h.hashret(), []).append(p)
  # Send packet
  self.pks.send(p)
  time.sleep(self.inter)

test/contrib/coap_socket.uts

@@ -127,3 +127,26 @@ with CoAPSocket("127.0.0.1", 5683, lst_resources=lst_resources) as coap_server,
  # assert res.msg_id == req.msg_id - This assert doesn't make sense because it will send with another msg_id
  assert res.token == req.token
  assert res.payload.load == responses[1]
+
+= SR1
+
+with CoAPSocket("127.0.0.1", 5683, lst_resources=lst_resources) as coap_server, CoAPSocket("127.0.0.1", 5684) as coap_client:
+ req = CoAPSocket.make_coap_req_packet(uri="/dummy", payload=b"")
+ res = coap_client.sr1(IP(dst="127.0.0.1")/UDP(dport=5683)/req)
+ assert res.payload.load == responses[0]
+ assert res.type == ACK
+ assert res.code == CONTENT_205
+ assert res.msg_id == req.msg_id
+ assert res.token == req.token
+
+= SR
+
+with CoAPSocket("127.0.0.1", 5683, lst_resources=lst_resources) as coap_server, CoAPSocket("127.0.0.1", 5684) as coap_client:
+ pkt = CoAPSocket.make_coap_req_packet(uri="/dummy", payload=b"")
+ ans, _ = coap_client.sr(IP(dst="127.0.0.1")/UDP(dport=5683)/pkt)
+ for _, rcv in ans:
+ assert rcv.payload.load == responses[0]
+ assert rcv.type == ACK
+ assert rcv.code == CONTENT_205
+ assert rcv.msg_id == pkt.msg_id
+ assert rcv.token == pkt.token