Unquote username and password in DatabaseURL #247 (#248)

* Unquote username and password in DatabaseURL #247

* fix userinfo property to be the same as httpx

* encode in utf-8 instead of ascii

* improve test coverage on core

* explicit str type for _url

* fix black linting

Co-authored-by: Vadim Markovtsev <[email protected]>

Author: librehat

databases/core.py

@@ -5,7 +5,7 @@
 import sys
 import typing
 from types import TracebackType
-from urllib.parse import SplitResult, parse_qsl, urlsplit
+from urllib.parse import SplitResult, parse_qsl, urlsplit, unquote
 
 from sqlalchemy import text
 from sqlalchemy.sql import ClauseElement
@@ -389,7 +389,14 @@ def __bool__(self) -> bool:
 
 class DatabaseURL:
     def __init__(self, url: typing.Union[str, "DatabaseURL"]):
-        self._url = str(url)
+        if isinstance(url, DatabaseURL):
+            self._url: str = url._url
+        elif isinstance(url, str):
+            self._url = url
+        else:
+            raise TypeError(
+                f"Invalid type for DatabaseURL. Expected str or DatabaseURL, got {type(url)}"
+            )
 
     @property
     def components(self) -> SplitResult:
@@ -411,13 +418,26 @@ def driver(self) -> str:
             return ""
         return self.components.scheme.split("+", 1)[1]
 
+    @property
+    def userinfo(self) -> typing.Optional[bytes]:
+        if self.components.username:
+            info = self.components.username
+            if self.components.password:
+                info += ":" + self.components.password
+            return info.encode("utf-8")
+        return None
+
     @property
     def username(self) -> typing.Optional[str]:
-        return self.components.username
+        if self.components.username is None:
+            return None
+        return unquote(self.components.username)
 
     @property
     def password(self) -> typing.Optional[str]:
-        return self.components.password
+        if self.components.password is None:
+            return None
+        return unquote(self.components.password)
 
     @property
     def hostname(self) -> typing.Optional[str]:
@@ -436,7 +456,7 @@ def database(self) -> str:
         path = self.components.path
         if path.startswith("/"):
             path = path[1:]
-        return path
+        return unquote(path)
 
     @property
     def options(self) -> dict:
@@ -453,8 +473,8 @@ def replace(self, **kwargs: typing.Any) -> "DatabaseURL":
         ):
             hostname = kwargs.pop("hostname", self.hostname)
             port = kwargs.pop("port", self.port)
-            username = kwargs.pop("username", self.username)
-            password = kwargs.pop("password", self.password)
+            username = kwargs.pop("username", self.components.username)
+            password = kwargs.pop("password", self.components.password)
 
             netloc = hostname
             if port is not None:

tests/test_database_url.py

@@ -1,4 +1,6 @@
 from databases import DatabaseURL
+from urllib.parse import quote
+import pytest
 
 
 def test_database_url_repr():
@@ -11,6 +13,9 @@ def test_database_url_repr():
     u = DatabaseURL("postgresql://username:password@localhost/name")
     assert repr(u) == "DatabaseURL('postgresql://username:********@localhost/name')"
 
+    u = DatabaseURL(f"postgresql://username:{quote('[password')}@localhost/name")
+    assert repr(u) == "DatabaseURL('postgresql://username:********@localhost/name')"
+
 
 def test_database_url_properties():
     u = DatabaseURL("postgresql+asyncpg://username:password@localhost:123/mydatabase")
@@ -23,6 +28,27 @@ def test_database_url_properties():
     assert u.database == "mydatabase"
 
 
+def test_database_url_escape():
+    u = DatabaseURL(f"postgresql://username:{quote('[password')}@localhost/mydatabase")
+    assert u.username == "username"
+    assert u.password == "[password"
+    assert u.userinfo == f"username:{quote('[password')}".encode("utf-8")
+
+    u2 = DatabaseURL(u)
+    assert u2.password == "[password"
+
+    u3 = DatabaseURL(str(u))
+    assert u3.password == "[password"
+
+
+def test_database_url_constructor():
+    with pytest.raises(TypeError):
+        DatabaseURL(("postgresql", "username", "password", "localhost", "mydatabase"))
+
+    u = DatabaseURL("postgresql+asyncpg://username:password@localhost:123/mydatabase")
+    assert DatabaseURL(u) == u
+
+
 def test_database_url_options():
     u = DatabaseURL("postgresql://localhost/mydatabase?pool_size=20&ssl=true")
     assert u.options == {"pool_size": "20", "ssl": "true"}
@@ -46,6 +72,9 @@ def test_replace_database_url_components():
     assert new.port == 123
     assert str(new) == "postgresql://localhost:123/mydatabase"
 
+    assert u.username is None
+    assert u.userinfo is None
+
     u = DatabaseURL("sqlite:///mydatabase")
     assert u.database == "mydatabase"
     new = u.replace(database="test_" + u.database)