Fix credentials in server_url

asyncua/client/client.py

 import asyncio
 import logging
 from typing import List, Union, Coroutine, Optional
-from urllib.parse import urlparse
+from urllib.parse import urlparse, unquote
 
 from asyncua import ua
 from .ua_client import UaClient
@@ -29,6 +29,10 @@ class Client:
     use UaClient object, available as self.uaclient
     which offers the raw OPC-UA services interface.
     """
+
+    _username = None
+    _password = None
+
     def __init__(self, url: str, timeout: float = 4):
         """
         :param url: url of the server.
@@ -43,8 +47,15 @@ class Client:
         """
         self.server_url = urlparse(url)
         # take initial username and password from the url
-        self._username = self.server_url.username
-        self._password = self.server_url.password
+        userinfo, have_info, hostinfo = self.server_url.netloc.rpartition('@')
+        if have_info:
+            username, have_password, password = userinfo.partition(':')
+            self._username = unquote(username)
+            if have_password:
+                self._password = unquote(password)
+            # remove credentials from url, preventing them to be sent unencrypted in e.g. send_hello
+            self.server_url = self.server_url.__class__(self.server_url[0], hostinfo, *self.server_url[2:])
+
         self.name = "Pure Python Async. Client"
         self.description = self.name
         self.application_uri = "urn:freeopcua:client"

tests/test_password.py

+import pytest
+
+from asyncua import Client, Server, ua
+from asyncua.server.users import UserRole, User
+
+uri = "opc.tcp://127.0.0.1:48517/baz/server"
+uri_creds = "opc.tcp://foobar:hR%26yjjGhP%246%40nQ4e@127.0.0.1:48517/baz/server"
+uri_wrong_creds = "opc.tcp://foobar:wrong@127.0.0.1:48517/baz/server"
+
+
+class UserManager:
+    def get_user(self, iserver, username=None, password=None, certificate=None):
+        if username == "foobar" and password == "hR&yjjGhP$6@nQ4e":
+            return User(role=UserRole.User)
+        return None
+
+
+@pytest.fixture()
+async def srv_user():
+    srv = Server(user_manager=UserManager())
+    srv.set_endpoint(uri)
+    srv.set_security_IDs(["Username"])
+
+    await srv.init()
+    await srv.start()
+    yield srv
+    await srv.stop()
+
+
+async def test_creds(srv_user):
+    clt = Client(uri)
+    clt.set_user("foobar")
+    clt.set_password("hR&yjjGhP$6@nQ4e")
+    await clt.connect()
+    await clt.disconnect()
+
+
+async def test_wrong_creds(srv_user):
+    clt = Client(uri)
+    clt.set_user("foobar")
+    clt.set_password("wrong")
+    with pytest.raises(ua.uaerrors.BadUserAccessDenied):
+        await clt.connect()
+    await clt.disconnect()
+
+
+async def test_creds_in_uri(srv_user):
+    clt = Client(uri_creds)
+    # check if credentials got removed
+    assert clt.server_url.geturl() == uri
+    await clt.connect()
+    await clt.disconnect()
+
+
+async def test_wrong_creds_in_uri(srv_user):
+    clt = Client(uri_wrong_creds)
+    with pytest.raises(ua.uaerrors.BadUserAccessDenied):
+        await clt.connect()
+    await clt.disconnect()