scapy.contrib.coap_socket
- class scapy.contrib.coap_socket.CoAPResource(url: str, content_format: bytes = b'\x00', title: str = '', description: str = '', resource_type: str = '')[source]
Bases:
objectUser should implement this class if he wants an answering machine for the CoAPSocket
- Parameters:
url – the resource URL
content_format – the default content format, this can be overridden by specifying the CF in the method’s return value. RFC 7252 @ section-7.2.1
title – A human-readable title for this resource. RFC 5988 @ section 5.4.
description – One can think of this as describing verbs usable on a resource. RFC 6690 @ section-3.1
resource_type – One can think of this as a noun describing the resource. RFC 6690 @ section-3.2
- check_duplicated(message_id: int, token: int) bool[source]
Returns true if (message_id, token) duplicated.
- get(payload: bytes, options: list[tuple], token: int, sa_ll: tuple[str, int]) dict[source]
Implementation of the get method for this resource. User should return a dictionary containing, at least these keys:
type: one of the CoAP message type
code: one of the CoAP message response codes (RFC 7252 @ section-12.1.2)
- options: a list of tuples with the options for the response
(RFC 7252 @ section-5.10). Should have at least the pair CONTENT_FORMAT
payload: optional, byte encoded payload
token: the request token, in case you need to implement a delayed message
- sa_ll: the sender ip/port pair,
in case you need to implement a delayed message
RFC 7252 @ section-5.8.1
- get_CORE_string() str[source]
Will return a CORE formatted string as specified in RFC 6690 + RFC 7252 @ section-7.2.1
- put(payload: bytes, options: list[tuple], token: int, sa_ll: tuple[str, int]) dict[source]
Implementation of the put method for this resource. User should return a dictionary containing, at least these keys:
type: one of the CoAP message type
code: one of the CoAP message response codes (RFC 7252 @ section-12.1.2)
- options: a list of tuples with the options for the response
(RFC 7252 @ section-5.10). Should have at least the pair CONTENT_FORMAT
payload: optional, byte encoded payload
token: the request token, in case you need to implement a delayed message
- sa_ll: the sender ip/port pair,
in case you need to implement a delayed message
RFC 7252 @ section-5.8.3
- class scapy.contrib.coap_socket.CoAPSocket(ip='', port=5683, ack_timeout=500, retries=3, duplication_response_timeout=1.0, lst_resources=None, sock=None, close_on_timeout=False)[source]
Bases:
SuperSocketCoAP socket with client and server capabilities.
General and defaults timeouts for the protocol - RFC 7252 @ section-4.8.2
Client example: >>> with CoAPSocket(“127.0.0.1”, 1234) as coap_client: >>> req = CoAPSocket.make_coap_req_packet( >>> method=GET, uri=”endpoint-uri”, payload=b””) >>> coap_client.send(IP(dst=”192.168.1.1”) / UDP(dport=1234) / req) >>> # Careful, this will block until the coap_client receives something >>> res = coap_client.recv()
Server without specifying resources: >>> with CoAPSocket(“127.0.0.1”, 5683) as coap_server: >>> while True: >>> pkg = coap_server.recv() >>> handle_package(pkg)
Server with custom resources: >>> class DummyResource(CoAPResource): >>> def get(self, payload, options, token, sa_ll): >>> return {“type”: ACK, “code”: CONTENT_205, >>> “options”: [(CONTENT_FORMAT, CF_TEXT_PLAIN)], >>> “payload”: b’dummy response’} >>> >>> class DelayedResource(CoAPResource): >>> def __init__(self, url): >>> CoAPResource.__init__(self, url=url) >>> self.delayed_tokens = [] >>> def delayed_message(self): >>> token, address = self.delayed_tokens.pop(0) >>> pkt = CoAPSocket.make_delayed_resp_packet(token, >>> [(CONTENT_FORMAT, CF_TEXT_PLAIN)], b”delayed payload”) >>> self._send_separate_response(pkt, address) >>> def get(self, payload, options, token, sa_ll): >>> # We know that this can take a while, so we return an empty ACK now >>> # and wait for whatever resource to be available. >>> TimeoutScheduler.schedule(1, self.delayed_message) >>> self.delayed_tokens.append((token, sa_ll)) >>> 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”)] >>> with CoAPSocket(“127.0.0.1”, 5683, lst_resources=lst_resources) as coap_socket: >>> while True: >>> pkg = coap_socket.recv() >>> # You can handle the packages inside your resources, >>> # here will only be the “unhandled” ones.
- Parameters:
ip – ip address to bind udp socket to.
port – port to bind udp socket to.
ack_timeout – the time, in ms, that we should wait for the acknowledgment after sending a request.
retries – amount of retransmissions before giving up on the request.
duplication_response_timeout – Timeout, in fractions of seconds, that we will keep the response in case a response get lost.
lst_resources – optional, list of registered resources.
sock – optional, a socket instance to transmit, if None, a classic UDP socket will be open and bound to ip/port.
close_on_timeout – Will try to close the socket if the retries is exceeded
- static empty_ack_params() dict[source]
A dictionary containing the base parameters for the empty ACK response. Later, you should also add the request msg_id.
- Returns:
A dictionary containing the parameters necessary to build a CoAP package for an empty ACK response.
- static make_coap_req_packet(method: int = 1, uri: str = '', options: list[tuple] = None, payload: bytes = b'') Packet[source]
Create a CoAP request packet
- Parameters:
method – The target method, one of: GET, POST, PUT, DELETE
uri – The destination uri
options – The options, should be a list of tuples. You must specify here the payload type. Example: options = [(CONTENT_FORMAT, CF_APP_XML)]
payload – The payload to send, should be a byte array
- Returns:
The CoAP packet.
- static make_coap_resp_packet(coap_type: int, code: int, token: bytes, message_id: int, options: list[tuple] = None, payload: bytes = b'') Packet[source]
Create a CoAP response packet
- Parameters:
coap_type – Message type, one of: CON, NON, ACK, RST
code – Response code, one of: EMPTY_ACK, CONTENT_205, NOT_FOUND_404, NOT_ALLOWED_405, NOT_IMPLEMENTED_501
token – The token from the request
message_id – The message id from the request
options – The options, should be a list of tuples. You must specify here the payload type. If applicable. Example: options = [(CONTENT_FORMAT, CF_APP_XML)]
payload – The payload to send, should be a byte array.
- Returns:
The CoAP packet.
- static make_delayed_resp_packet(token: int | bytes, options: list[tuple], payload: bytes) Packet[source]
This will create a CoAP packet that contains all the correct parameters for the delayed response. The msg_id is not necessary to be specified, it will be random generated. After all, this is similar to a new request.
- Parameters:
token – The original request token
options – The options, should be a list of tuples. You must specify here the payload type. If applicable. Example: options = [(CONTENT_FORMAT, CF_APP_XML)]
payload – The payload to send, should be a byte array.
- Returns:
The CoAP packet.
- static select(sockets: list[SuperSocket], remain: float | None = None) list[SuperSocket][source]
This function is called during sendrecv() routine to wait for sockets to be ready to receive.
- send(x: Packet) int[source]
Send the packet using this socket. Should be a CoAP packet with IP and UDP data.
Example: >>> IP(dst=”192.168.1.1”) / UDP(dport=1234) / CoAP() >>> IP(dst=”192.168.1.1”) / UDP(dport=1234) / CoAPSocket.make_coap_req_packet()
- Parameters:
x – Concatenated packet with IP / UDP / CoAP
- Returns:
The length of x, which is the amount of bytes sent
- class scapy.contrib.coap_socket.CoAPSocketImpl(ip='', port=5683, ack_timeout=500, retries=3, duplication_response_timeout=1.0, lst_resources=None, sock=None, close_on_timeout=False)[source]
Bases:
objectImplementation of a CoAP socket with client and server capabilities.
- Parameters:
ip – ip address to bind udp socket to.
port – port to bind udp socket to.
ack_timeout – the time, in ms, that we should wait for the acknowledgment after sending a request.
retries – amount of retransmissions before giving up on the request.
duplication_response_timeout – Timeout, in fractions of seconds, that we will keep the response in case a response get lost.
lst_resources – optional, list of registered resources.
sock – optional, a socket instance to transmit, if None, a classic UDP socket will be open and bound to ip/port.
close_on_timeout – Will try to close the socket if the retries is exceeded
- class CoAPRequest(ip: str, port: int, max_retries: int, retry_timeout: float, pkt: CoAP, resource: CoAPResource | None = None)[source]
Bases:
objectClass to control a client request.
- Parameters:
ip – The remote server’s ip address.
port – The remote server’s port.
max_retries – Number of retransmissions before giving up.
retry_timeout – ACK timeout for retransmission.
pkt – The CoAP package to be sent.
- empty_ack_set() None[source]
Set the empty ack flag and will set the timeout. After the timeout, it will resend the request until should_give_up() is triggered.
- get_pkt_and_mark() Packet[source]
Returns the already sent packet for retransmission and sets a new timeout for retry. :return: A CoAP packet for retransmission.
- indexing() tuple[int, int][source]
Returns the indexing of this request. :return: A tuple containing the message_id and token of this request.
- class WellKnownResource(lst_resources: list[CoAPResource])[source]
Bases:
CoAPResourceThis is a default resource that will return information about all the registered resources in the server. Described at RFC 7252 @ section 7.2 and RFC 6690
- Parameters:
lst_resources – List of CoAPResource.
- static generate_msgId_token() tuple[int, bytes][source]
Will generate a pair of (msgId, token) with message id in the range of [0, 0xffff] and a random token with size from 1 to 8 bytes :return: msgId and token tuple
- static make_coap_req_packet(method: int = 1, uri: str = '', options: list[tuple] | None = None, payload: bytes = b'') Packet[source]
Check CoAPSocket for the documentation
- static make_coap_resp_packet(coap_type: int, code: int, token: bytes, message_id: int, options: list[tuple], payload: bytes) Packet[source]
Check CoAPSocket for the documentation