| # 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) |
