# MQTT layer unit tests
|
# Copyright (C) Santiago Hernandez Ramos <shramos@protonmail.com>
|
#
|
# Type the following command to launch start the tests:
|
# $ test/run_tests -P "load_contrib('mqtt')" -t scapy/contrib/mqtt.uts
|
|
+ Syntax check
|
= Import the MQTT layer
|
from scapy.contrib.mqtt import *
|
|
|
+ MQTT protocol test
|
|
= MQTTPublish, packet instanciation
|
p = MQTT()/MQTTPublish(topic='test1',value='test2')
|
assert(p.type == 3)
|
assert(p.topic == b'test1')
|
assert(p.value == b'test2')
|
assert(p.len == None)
|
assert(p.length == None)
|
|
= Fixed header and MQTTPublish, packet dissection
|
s = b'0\n\x00\x04testtest'
|
publish = MQTT(s)
|
assert(publish.type == 3)
|
assert(publish.QOS == 0)
|
assert(publish.DUP == 0)
|
assert(publish.RETAIN == 0)
|
assert(publish.len == 10)
|
assert(publish[MQTTPublish].length == 4)
|
assert(publish[MQTTPublish].topic == b'test')
|
assert(publish[MQTTPublish].value == b'test')
|
|
|
= MQTTConnect, packet instanciation
|
c = MQTT()/MQTTConnect(clientIdlen=5, clientId='newid')
|
assert(c.type == 1)
|
assert(c.clientId == b'newid')
|
assert(c.clientIdlen == 5)
|
|
= MQTTConnect, packet dissection
|
s = b'\x10\x1f\x00\x06MQIsdp\x03\x02\x00<\x00\x11mosqpub/1440-kali'
|
connect = MQTT(s)
|
assert(connect.length == 6)
|
assert(connect.protoname == b'MQIsdp')
|
assert(connect.protolevel == 3)
|
assert(connect.usernameflag == 0)
|
assert(connect.passwordflag == 0)
|
assert(connect.willretainflag == 0)
|
assert(connect.willQOSflag == 0)
|
assert(connect.willflag == 0)
|
assert(connect.cleansess == 1)
|
assert(connect.reserved == 0)
|
assert(connect.klive == 60)
|
assert(connect.clientIdlen == 17)
|
assert(connect.clientId == b'mosqpub/1440-kali')
|
|
|
=MQTTConnack, packet instanciation
|
ck = MQTT()/MQTTConnack(sessPresentFlag=1,retcode=0)
|
assert(ck.type == 2)
|
assert(ck.sessPresentFlag == 1)
|
assert(ck.retcode == 0)
|
|
= MQTTConnack, packet dissection
|
s = b' \x02\x00\x00'
|
connack = MQTT(s)
|
assert(connack.sessPresentFlag == 0)
|
assert(connack.retcode == 0)
|
|
|
= MQTTSubscribe, packet instanciation
|
sb = MQTT()/MQTTSubscribe(msgid=1,topic='newtopic',QOS=0,length=0)
|
assert(sb.type == 8)
|
assert(sb.msgid == 1)
|
assert(sb.topic == b'newtopic')
|
assert(sb.length == 0)
|
assert(sb[MQTTSubscribe].QOS == 0)
|
|
= MQTTSubscribe, packet dissection
|
s = b'\x82\t\x00\x01\x00\x04test\x00'
|
subscribe = MQTT(s)
|
assert(subscribe.msgid == 1)
|
assert(subscribe.length == 4)
|
assert(subscribe.topic == b'test')
|
assert(subscribe.QOS == 1)
|
|
|
= MQTTSuback, packet instanciation
|
sk = MQTT()/MQTTSuback(msgid=1, retcode=0)
|
assert(sk.type == 9)
|
assert(sk.msgid == 1)
|
assert(sk.retcode == 0)
|
|
= MQTTSuback, packet dissection
|
s = b'\x90\x03\x00\x01\x00'
|
suback = MQTT(s)
|
assert(suback.msgid == 1)
|
assert(suback.retcode == 0)
|
|
|
= MQTTPubrec, packet instanciation
|
pc = MQTT()/MQTTPubrec(msgid=1)
|
assert(pc.type == 5)
|
assert(pc.msgid == 1)
|
|
= MQTTPubrec packet dissection
|
s = b'P\x02\x00\x01'
|
pubrec = MQTT(s)
|
assert(pubrec.msgid == 1)
|
