1 # MQTT layer unit tests 2 # Copyright (C) Santiago Hernandez Ramos <shramos (a] protonmail.com> 3 # 4 # Type the following command to launch start the tests: 5 # $ test/run_tests -P "load_contrib('mqtt')" -t scapy/contrib/mqtt.uts 6 7 + Syntax check 8 = Import the MQTT layer 9 from scapy.contrib.mqtt import * 10 11 12 + MQTT protocol test 13 14 = MQTTPublish, packet instanciation 15 p = MQTT()/MQTTPublish(topic='test1',value='test2') 16 assert(p.type == 3) 17 assert(p.topic == b'test1') 18 assert(p.value == b'test2') 19 assert(p.len == None) 20 assert(p.length == None) 21 22 = Fixed header and MQTTPublish, packet dissection 23 s = b'0\n\x00\x04testtest' 24 publish = MQTT(s) 25 assert(publish.type == 3) 26 assert(publish.QOS == 0) 27 assert(publish.DUP == 0) 28 assert(publish.RETAIN == 0) 29 assert(publish.len == 10) 30 assert(publish[MQTTPublish].length == 4) 31 assert(publish[MQTTPublish].topic == b'test') 32 assert(publish[MQTTPublish].value == b'test') 33 34 35 = MQTTConnect, packet instanciation 36 c = MQTT()/MQTTConnect(clientIdlen=5, clientId='newid') 37 assert(c.type == 1) 38 assert(c.clientId == b'newid') 39 assert(c.clientIdlen == 5) 40 41 = MQTTConnect, packet dissection 42 s = b'\x10\x1f\x00\x06MQIsdp\x03\x02\x00<\x00\x11mosqpub/1440-kali' 43 connect = MQTT(s) 44 assert(connect.length == 6) 45 assert(connect.protoname == b'MQIsdp') 46 assert(connect.protolevel == 3) 47 assert(connect.usernameflag == 0) 48 assert(connect.passwordflag == 0) 49 assert(connect.willretainflag == 0) 50 assert(connect.willQOSflag == 0) 51 assert(connect.willflag == 0) 52 assert(connect.cleansess == 1) 53 assert(connect.reserved == 0) 54 assert(connect.klive == 60) 55 assert(connect.clientIdlen == 17) 56 assert(connect.clientId == b'mosqpub/1440-kali') 57 58 59 =MQTTConnack, packet instanciation 60 ck = MQTT()/MQTTConnack(sessPresentFlag=1,retcode=0) 61 assert(ck.type == 2) 62 assert(ck.sessPresentFlag == 1) 63 assert(ck.retcode == 0) 64 65 = MQTTConnack, packet dissection 66 s = b' \x02\x00\x00' 67 connack = MQTT(s) 68 assert(connack.sessPresentFlag == 0) 69 assert(connack.retcode == 0) 70 71 72 = MQTTSubscribe, packet instanciation 73 sb = MQTT()/MQTTSubscribe(msgid=1,topic='newtopic',QOS=0,length=0) 74 assert(sb.type == 8) 75 assert(sb.msgid == 1) 76 assert(sb.topic == b'newtopic') 77 assert(sb.length == 0) 78 assert(sb[MQTTSubscribe].QOS == 0) 79 80 = MQTTSubscribe, packet dissection 81 s = b'\x82\t\x00\x01\x00\x04test\x00' 82 subscribe = MQTT(s) 83 assert(subscribe.msgid == 1) 84 assert(subscribe.length == 4) 85 assert(subscribe.topic == b'test') 86 assert(subscribe.QOS == 1) 87 88 89 = MQTTSuback, packet instanciation 90 sk = MQTT()/MQTTSuback(msgid=1, retcode=0) 91 assert(sk.type == 9) 92 assert(sk.msgid == 1) 93 assert(sk.retcode == 0) 94 95 = MQTTSuback, packet dissection 96 s = b'\x90\x03\x00\x01\x00' 97 suback = MQTT(s) 98 assert(suback.msgid == 1) 99 assert(suback.retcode == 0) 100 101 102 = MQTTPubrec, packet instanciation 103 pc = MQTT()/MQTTPubrec(msgid=1) 104 assert(pc.type == 5) 105 assert(pc.msgid == 1) 106 107 = MQTTPubrec packet dissection 108 s = b'P\x02\x00\x01' 109 pubrec = MQTT(s) 110 assert(pubrec.msgid == 1) 111
