fix: Allow url-encoded username and passwords

amqtt/client.py

@@ -6,7 +6,7 @@
 import logging
 import ssl
 from typing import TYPE_CHECKING, Any, TypeAlias, cast
-from urllib.parse import urlparse, urlunparse
+from urllib.parse import unquote, urlparse, urlunparse
 
 import websockets
 from websockets import HeadersLike, InvalidHandshake, InvalidURI
@@ -424,10 +424,10 @@ async def _connect_coro(self) -> int:
         scheme = uri_attributes.scheme
         secure = scheme in ("mqtts", "wss")
         self.session.username = (
-            self.session.username or (str(uri_attributes.username) if uri_attributes.username else None)
+            self.session.username or (unquote(uri_attributes.username) if uri_attributes.username else None)
         )
         self.session.password = (
-            self.session.password or (str(uri_attributes.password) if uri_attributes.password else None)
+            self.session.password or (unquote(uri_attributes.password) if uri_attributes.password else None)
         )
         self.session.remote_address = str(uri_attributes.hostname) if uri_attributes.hostname else None
         self.session.remote_port = uri_attributes.port

tests/plugins/passwd

@@ -1,6 +1,7 @@
 # Test password file
 user:$6$1sSXVMBVKMF$uDStq59YfiuFxVF1Gi/.i/Li7Vwf5iTwg8LovLKmCvM5FsRNJM.OPWHhXwI2.4AscLZXSPQVxpIlX6laUl9570
 test_user:$6$.c9f9sAzs5YXX2de$GSdOi3iFwHJRCIJn1W63muDFQAL29yoFmU/TXcwDB42F2BZg3zcN5uKBM0.1PjwdMpWHRydbhXWSc3uWKSmKr.
+c@mpl3x:$6$BXHx0w6vwek1oWEb$sdfzb1ZWuxSB4CL06WzGFHCW6vDWTLwefDTr9fWLxztxpeofEAkgHj/YiUQZp1T/zwoIwyauYeAbLpkqYSwjj.
 # Password for these is "${USER}password"
 user1:$6$h.fV0zYsXI$8wKblqETpztRKcPD6OLWZc1mU4nW5yQ713R5ECs7EwJa7oas/yrhI2itUdhETI8BvmtfGy65ltAMap9gHkzdc1
 user2:$6$bUyF8v0mTo94$IJMa2BlCd6/mAM5N5s6heFB2ewC4j3CDkFb9mYIoarXg4OxiZVnLyh20IWHf6GY8i4sc5m2D9nWLrtbnIO5o8.

tests/test_client.py

@@ -22,6 +22,17 @@ async def test_connect_tcp(broker_fixture):
     assert client.session is not None
     await client.disconnect()
 
+
+@pytest.mark.asyncio
+async def test_connect_username_password(broker_fixture):
+    client = MQTTClient()
+    await client.connect("mqtt://c%40mpl3x:p%2F%2F%7C%5C%3A%24%24@127.0.0.1:1883/")
+    assert client.session is not None
+    assert client.session.username == "c@mpl3x"
+    assert client.session.password == "p//|\\:$$"
+    await client.disconnect()
+
+
 @pytest.mark.asyncio
 async def test_connect_tcp_secure(rsa_keys, broker_fixture):
     certfile, _ = rsa_keys