admin/coapwithudp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

basic interactions work

Browse Source
This commit is contained in:
Md Imran Hossain
2026-09-28 23:34:04 +06:00
parent b8d3a44d22
commit 4060f64b98
8 changed files with 1034 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
coap-inspect.py
View File

@@ -4,6 +4,7 @@
./coap-inspect.py get "/hello?name=Imran"
./coap-inspect.py post /hello Imran
./coap-inspect.py get /health
./coap-inspect.py get /image --block 3 (one Block2 block of a stream)
No CoAP library: the packet is built and decoded right here, so every byte on
screen is accounted for. COAP_HOST / COAP_PORT retarget it (127.0.0.1:5683).
@@ -19,13 +20,15 @@ METHODS = {"get": 1, "post": 2, "put": 3, "delete": 4}
CODES = {
"0.00": "Empty", "0.01": "GET", "0.02": "POST", "0.03": "PUT", "0.04": "DELETE",
"2.01": "Created", "2.02": "Deleted", "2.03": "Valid", "2.04": "Changed", "2.05": "Content",
"4.00": "Bad Request", "4.04": "Not Found", "4.05": "Method Not Allowed",
"4.00": "Bad Request", "4.02": "Bad Option", "4.04": "Not Found", "4.05": "Method Not Allowed",
"5.00": "Internal Server Error",
}
OPTIONS = {3: "Uri-Host", 7: "Uri-Port", 11: "Uri-Path", 12: "Content-Format", 15: "Uri-Query", 17: "Accept"}
FORMATS = {0: "text/plain", 50: "application/json"}
OPTIONS = {3: "Uri-Host", 4: "ETag", 7: "Uri-Port", 11: "Uri-Path", 12: "Content-Format", 15: "Uri-Query",
17: "Accept", 23: "Block2", 27: "Block1", 28: "Size2", 60: "Size1"}
FORMATS = {0: "text/plain", 42: "application/octet-stream", 50: "application/json"}
TTY = sys.stdout.isatty()
DUMP_LIMIT = 128 # bytes shown in the hex and binary views
def bold(text):
@@ -46,7 +49,7 @@ def encode_option(delta, value):
return bytes([(d << 4) | l]) + d_ext + l_ext + value
def build_request(method, path, query, payload):
def build_request(method, path, query, payload, block=None):
token = os.urandom(4)
header = bytes([(1 << 6) | (0 << 4) | len(token), METHODS[method]]) # ver 1, CON, TKL, code
header += random.randrange(1 << 16).to_bytes(2, "big") + token # message id, token
@@ -54,6 +57,9 @@ def build_request(method, path, query, payload):
# Options must be in ascending number order; each stores the gap from the previous one.
options = [(11, s.encode()) for s in path.split("/") if s]
options += [(15, q.encode()) for q in query.split("&") if q]
if block is not None: # Block2: num << 4 | more << 3 | SZX, SZX 6 = 1024 B
value = block << 4 | 6
options.append((23, value.to_bytes((value.bit_length() + 7) // 8, "big")))
body, last = b"", 0
for number, value in options:
body += encode_option(number - last, value)
@@ -90,8 +96,12 @@ def decode(pkt):
if pkt[i] == 0xFF:
body = pkt[i + 1:]
rows.append((i, i + 1, "payload marker", "0xff -> the rest is the body"))
rows.append((i + 1, len(pkt), "payload",
f"{len(body)} bytes: {body.decode('utf-8', 'replace')}"))
try:
text = body.decode("utf-8")
shown = f"{len(body)} bytes: {text}" if text.isprintable() else None
except UnicodeDecodeError:
shown = None
rows.append((i + 1, len(pkt), "payload", shown or f"{len(body)} bytes of binary data"))
break
start = i
@@ -112,6 +122,13 @@ def decode(pkt):
if number in (12, 17):
n = int.from_bytes(value, "big") if value else 0
shown = f"{n} = {FORMATS.get(n, 'unknown format')}"
elif number in (23, 27):
n = int.from_bytes(value, "big") if value else 0
shown = f"num {n >> 4}, more {bool(n >> 3 & 1)}, size {1 << ((n & 7) + 4)}"
elif number in (28, 60):
shown = f"{int.from_bytes(value, 'big') if value else 0} B total"
elif number == 4:
shown = f"0x{value.hex()}"
else:
shown = f'"{value.decode("utf-8", "replace")}"'
rows.append((start, i, OPTIONS.get(number, f"option {number}"),
@@ -124,18 +141,22 @@ def show(title, pkt):
print(bold(f"\n{title} ({len(pkt)} bytes)"))
print(bold("\n hex + ascii"))
for off in range(0, len(pkt), 16):
chunk = pkt[off:off + 16]
for off in range(0, min(len(pkt), DUMP_LIMIT), 16):
chunk = pkt[off:min(off + 16, DUMP_LIMIT)]
hexes = " ".join(f"{b:02x}" for b in chunk)
text = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk)
print(f" {off:04x} {hexes:<47} |{text}|")
if len(pkt) > DUMP_LIMIT:
print(f" ... {len(pkt) - DUMP_LIMIT} more bytes not shown")
print(bold("\n binary"))
for off in range(0, len(pkt), 4):
chunk = pkt[off:off + 4]
for off in range(0, min(len(pkt), DUMP_LIMIT), 4):
chunk = pkt[off:min(off + 4, DUMP_LIMIT)]
bits = " ".join(f"{b:08b}" for b in chunk)
text = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk)
print(f" {off:04x} {bits:<35} |{text}|")
if len(pkt) > DUMP_LIMIT:
print(f" ... {len(pkt) - DUMP_LIMIT} more bytes not shown")
rows, meta = decode(pkt)
print(bold("\n fields"))
@@ -151,6 +172,15 @@ def show(title, pkt):
def main():
args = sys.argv[1:]
block = None
if "--block" in args:
i = args.index("--block")
try:
block = int(args[i + 1])
except (IndexError, ValueError):
print("--block needs a block number")
return 2
del args[i:i + 2]
if not args or args[0].lower() not in METHODS:
print(__doc__.strip())
return 2
@@ -162,7 +192,7 @@ def main():
host = os.environ.get("COAP_HOST", "127.0.0.1")
port = int(os.environ.get("COAP_PORT", "5683"))
request = build_request(method, path, query, payload)
request = build_request(method, path, query, payload, block)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # SOCK_DGRAM = UDP
sock.settimeout(3)