Disconnect if client uses authentication, as authentication is not su…

…pported.

README.rst

@@ -33,6 +33,8 @@ Broker features:
 QoS level 1 and 2 messages are not yet supprted. A session state
 storage is required to do so, both in the client and the broker.
 
+Authentication is not supported.
+
 MQTT version 5.0 specification:
 https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html
 

mqttools/broker.py

@@ -4,6 +4,7 @@
 from collections import defaultdict
 
 from .common import ControlPacketType
+from .common import ConnectReasonCode
 from .common import SubackReasonCode
 from .common import UnsubackReasonCode
 from .common import PropertyIds
@@ -130,8 +131,12 @@ async def read_packet(self):
         return packet_type, flags, PayloadReader(data)
 
     def on_connect(self, payload):
-        client_id, clean_start, keep_alive_s, properties = unpack_connect(
-            payload)
+        (client_id,
+         clean_start,
+         keep_alive_s,
+         properties,
+         user_name,
+         password) = unpack_connect(payload)
         self._session, session_present = self._broker.get_session(
             client_id,
             clean_start)
@@ -141,16 +146,28 @@ def on_connect(self, payload):
                           len(self._session.subscriptions))
 
         self._session.client = self
+        reason = ConnectReasonCode.SUCCESS
+
+        if PropertyIds.AUTHENTICATION_METHOD in properties:
+            reason = ConnectReasonCode.BAD_AUTHENTICATION_METHOD
+
+        if (user_name is not None) or (password is not None):
+            reason = ConnectReasonCode.BAD_USER_NAME_OR_PASSWORD
+
         self._write_packet(pack_connack(
             session_present,
-            0,
+            reason,
             {
                 PropertyIds.MAXIMUM_QOS: 0,
                 PropertyIds.RETAIN_AVAILABLE: 0,
                 PropertyIds.WILDCARD_SUBSCRIPTION_AVAILABLE: 0,
                 PropertyIds.SHARED_SUBSCRIPTION_AVAILABLE: 0
             }))
 
+        # ToDo: Clean disconnect.
+        if reason != ConnectReasonCode.SUCCESS:
+            raise Exception('Connect failed.')
+
         self.log_info('Client connected.')
 
     def on_publish(self, payload):

mqttools/common.py

@@ -520,7 +520,17 @@ def unpack_connect(payload):
     if flags & WILL_FLAG:
         raise MalformedPacketError()
 
-    return client_id, clean_start, keep_alive_s, properties
+    if flags & USER_NAME_FLAG:
+        user_name = unpack_string(payload)
+    else:
+        user_name = None
+
+    if flags & PASSWORD_FLAG:
+        password = unpack_binary(payload)
+    else:
+        password = None
+
+    return client_id, clean_start, keep_alive_s, properties, user_name, password
 
 
 def pack_connack(session_present,
@@ -790,12 +800,19 @@ def format_properties(properties):
 
 
 def format_connect(payload):
-    client_id, clean_start, keep_alive_s, properties = unpack_connect(payload)
+    (client_id,
+     clean_start,
+     keep_alive_s,
+     properties,
+     user_name,
+     password) = unpack_connect(payload)
 
     return [
         f'  ClientId:   {client_id}',
         f'  CleanStart: {clean_start}',
-        f'  KeepAlive:  {keep_alive_s}'
+        f'  KeepAlive:  {keep_alive_s}',
+        f'  UserName:   {user_name}',
+        f'  Password:   {password}'
     ] + format_properties(properties)
 
 

tests/test_broker.py

@@ -7,17 +7,20 @@
 
 class BrokerTest(unittest.TestCase):
 
-    def test_multiple_subscribers(self):
-        asyncio.run(self.multiple_subscribers())
-
-    async def multiple_subscribers(self):
+    def create_broker(self):
         broker = mqttools.Broker('localhost', 0)
 
         async def broker_wrapper():
             with self.assertRaises(asyncio.CancelledError):
                 await broker.serve_forever()
 
-        broker_task = asyncio.create_task(broker_wrapper())
+        return broker, asyncio.create_task(broker_wrapper())
+
+    def test_multiple_subscribers(self):
+        asyncio.run(self.multiple_subscribers())
+
+    async def multiple_subscribers(self):
+        broker, broker_task = self.create_broker()
 
         async def tester():
             address = await broker.getsockname()
@@ -97,13 +100,7 @@ def test_unsubscribe(self):
         asyncio.run(self.unsubscribe())
 
     async def unsubscribe(self):
-        broker = mqttools.Broker('localhost', 0)
-
-        async def broker_wrapper():
-            with self.assertRaises(asyncio.CancelledError):
-                await broker.serve_forever()
-
-        broker_task = asyncio.create_task(broker_wrapper())
+        broker, broker_task = self.create_broker()
 
         async def tester():
             address = await broker.getsockname()
@@ -203,13 +200,7 @@ def test_resume_session(self):
         asyncio.run(self.resume_session())
 
     async def resume_session(self):
-        broker = mqttools.Broker('localhost', 0)
-
-        async def broker_wrapper():
-            with self.assertRaises(asyncio.CancelledError):
-                await broker.serve_forever()
-
-        broker_task = asyncio.create_task(broker_wrapper())
+        broker, broker_task = self.create_broker()
 
         async def tester():
             address = await broker.getsockname()
@@ -288,13 +279,7 @@ def test_ping(self):
         asyncio.run(self.ping())
 
     async def ping(self):
-        broker = mqttools.Broker('localhost', 0)
-
-        async def broker_wrapper():
-            with self.assertRaises(asyncio.CancelledError):
-                await broker.serve_forever()
-
-        broker_task = asyncio.create_task(broker_wrapper())
+        broker, broker_task = self.create_broker()
 
         async def tester():
             address = await broker.getsockname()
@@ -317,6 +302,58 @@ async def tester():
 
         await asyncio.wait_for(asyncio.gather(broker_task, tester()), 1)
 
+    def test_authentication_with_user_name_and_password(self):
+        asyncio.run(self.authentication_with_user_name_and_password())
+
+    async def authentication_with_user_name_and_password(self):
+        broker, broker_task = self.create_broker()
+
+        async def tester():
+            address = await broker.getsockname()
+
+            # Connect with user name and password, and get 0x86 in
+            # response.
+            reader_1, writer_1 = await asyncio.open_connection(*address)
+            connect = (
+                b'\x10\x1c\x00\x04MQTT\x05\xc0\x00\x00\x00\x00\x03su1\x00\x04user'
+                b'\x00\x04pass')
+            writer_1.write(connect)
+            connack = await reader_1.readexactly(13)
+            self.assertEqual(
+                connack,
+                b'\x20\x0b\x00\x86\x08\x24\x00\x25\x00\x28\x00\x2a\x00')
+
+            writer_1.close()
+            broker_task.cancel()
+
+        await asyncio.wait_for(asyncio.gather(broker_task, tester()), 1)
+
+    def test_authentication_method(self):
+        asyncio.run(self.authentication_method())
+
+    async def authentication_method(self):
+        broker, broker_task = self.create_broker()
+
+        async def tester():
+            address = await broker.getsockname()
+
+            # Connect with authentication method, and get 0x8c in
+            # response.
+            reader_1, writer_1 = await asyncio.open_connection(*address)
+            connect = (
+                b'\x10\x1b\x00\x04MQTT\x05\x00\x00\x00\x0b\x15\x00\x08bad-auth'
+                b'\x00\x03su1')
+            writer_1.write(connect)
+            connack = await reader_1.readexactly(13)
+            self.assertEqual(
+                connack,
+                b'\x20\x0b\x00\x8c\x08\x24\x00\x25\x00\x28\x00\x2a\x00')
+
+            writer_1.close()
+            broker_task.cancel()
+
+        await asyncio.wait_for(asyncio.gather(broker_task, tester()), 1)
+
 
 logging.basicConfig(level=logging.DEBUG)