revert url credential encoding (to be reintroduced as an option in future) (#1882)

davidhewitt · web-flow · commit 4d23017bf1d0 · 2025-11-04T10:48:10.000Z
diff --git a/src/url.rs b/src/url.rs
@@ -1,7 +1,7 @@
 use std::borrow::Cow;
 use std::collections::hash_map::DefaultHasher;
-use std::fmt;
-use std::fmt::Formatter;
+use std::fmt::{self, Display};
+use std::fmt::{Formatter, Write};
 use std::hash::{Hash, Hasher};
 use std::sync::OnceLock;
 
@@ -229,14 +229,38 @@ impl PyUrl {
         path: Option<&str>,
         query: Option<&str>,
         fragment: Option<&str>,
+        // encode_credentials: bool, // TODO: re-enable this
+    ) -> PyResult<Bound<'py, PyAny>> {
+        Self::build_inner(
+            cls, scheme, host, username, password, port, path, query, fragment, false,
+        )
+    }
+}
+
+impl PyUrl {
+    #[allow(clippy::too_many_arguments)]
+    fn build_inner<'py>(
+        cls: &Bound<'py, PyType>,
+        scheme: &str,
+        host: &str,
+        username: Option<&str>,
+        password: Option<&str>,
+        port: Option<u16>,
+        path: Option<&str>,
+        query: Option<&str>,
+        fragment: Option<&str>,
+        encode_credentials: bool,
     ) -> PyResult<Bound<'py, PyAny>> {
         let url_host = UrlHostParts {
             username: username.map(Into::into),
             password: password.map(Into::into),
             host: Some(host.into()),
             port,
         };
-        let mut url = format!("{scheme}://{url_host}");
+        let mut url = format!("{scheme}://");
+        url_host
+            .to_writer(&mut url, encode_credentials)
+            .expect("writing to string should not fail");
         if let Some(path) = path {
             url.push('/');
             url.push_str(path);
@@ -446,7 +470,30 @@ impl PyMultiHostUrl {
     #[classmethod]
     #[pyo3(signature=(*, scheme, hosts=None, path=None, query=None, fragment=None, host=None, username=None, password=None, port=None))]
     #[allow(clippy::too_many_arguments)]
-    pub fn build<'py>(
+    fn build<'py>(
+        cls: &Bound<'py, PyType>,
+        scheme: &str,
+        hosts: Option<Vec<UrlHostParts>>,
+        path: Option<&str>,
+        query: Option<&str>,
+        fragment: Option<&str>,
+        // convenience parameters to build with a single host
+        host: Option<&str>,
+        username: Option<&str>,
+        password: Option<&str>,
+        port: Option<u16>,
+        // encode_credentials: bool, // TODO: re-enable this
+    ) -> PyResult<Bound<'py, PyAny>> {
+        Self::build_inner(
+            cls, scheme, hosts, path, query, fragment, host, username, password, port,
+            false, // TODO: re-enable this
+        )
+    }
+}
+
+impl PyMultiHostUrl {
+    #[allow(clippy::too_many_arguments)]
+    fn build_inner<'py>(
         cls: &Bound<'py, PyType>,
         scheme: &str,
         hosts: Option<Vec<UrlHostParts>>,
@@ -458,39 +505,44 @@ impl PyMultiHostUrl {
         username: Option<&str>,
         password: Option<&str>,
         port: Option<u16>,
+        encode_credentials: bool,
     ) -> PyResult<Bound<'py, PyAny>> {
-        let mut url =
-            if hosts.is_some() && (host.is_some() || username.is_some() || password.is_some() || port.is_some()) {
-                return Err(PyValueError::new_err(
-                    "expected one of `hosts` or singular values to be set.",
-                ));
-            } else if let Some(hosts) = hosts {
-                // check all of host / user / password / port empty
-                // build multi-host url
-                let mut multi_url = format!("{scheme}://");
-                for (index, single_host) in hosts.iter().enumerate() {
-                    if single_host.is_empty() {
-                        return Err(PyValueError::new_err(
-                            "expected one of 'host', 'username', 'password' or 'port' to be set",
-                        ));
-                    }
-                    multi_url.push_str(&single_host.to_string());
-                    if index != hosts.len() - 1 {
-                        multi_url.push(',');
-                    }
+        let mut url = format!("{scheme}://");
+
+        if hosts.is_some() && (host.is_some() || username.is_some() || password.is_some() || port.is_some()) {
+            return Err(PyValueError::new_err(
+                "expected one of `hosts` or singular values to be set.",
+            ));
+        } else if let Some(hosts) = hosts {
+            // check all of host / user / password / port empty
+            // build multi-host url
+            let len = hosts.len();
+            for (index, single_host) in hosts.into_iter().enumerate() {
+                if single_host.is_empty() {
+                    return Err(PyValueError::new_err(
+                        "expected one of 'host', 'username', 'password' or 'port' to be set",
+                    ));
                 }
-                multi_url
-            } else if host.is_some() {
-                let url_host = UrlHostParts {
-                    username: username.map(Into::into),
-                    password: password.map(Into::into),
-                    host: host.map(Into::into),
-                    port,
-                };
-                format!("{scheme}://{url_host}")
-            } else {
-                return Err(PyValueError::new_err("expected either `host` or `hosts` to be set"));
+                single_host
+                    .to_writer(&mut url, encode_credentials)
+                    .expect("writing to string should not fail");
+                if index != len - 1 {
+                    url.push(',');
+                }
+            }
+        } else if host.is_some() {
+            let url_host = UrlHostParts {
+                username: username.map(Into::into),
+                password: password.map(Into::into),
+                host: host.map(Into::into),
+                port,
             };
+            url_host
+                .to_writer(&mut url, encode_credentials)
+                .expect("writing to string should not fail");
+        } else {
+            return Err(PyValueError::new_err("expected either `host` or `hosts` to be set"));
+        }
 
         if let Some(path) = path {
             url.push('/');
@@ -508,55 +560,65 @@ impl PyMultiHostUrl {
     }
 }
 
-pub struct UrlHostParts {
+struct UrlHostParts {
     username: Option<String>,
     password: Option<String>,
     host: Option<String>,
     port: Option<u16>,
 }
 
-impl UrlHostParts {
-    fn is_empty(&self) -> bool {
-        self.host.is_none() && self.password.is_none() && self.host.is_none() && self.port.is_none()
+struct MaybeEncoded<'a>(&'a str, bool);
+
+impl fmt::Display for MaybeEncoded<'_> {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        if self.1 {
+            write!(f, "{}", encode_userinfo_component(self.0))
+        } else {
+            write!(f, "{}", self.0)
+        }
     }
 }
 
-impl FromPyObject<'_> for UrlHostParts {
-    fn extract_bound(ob: &Bound<'_, PyAny>) -> PyResult<Self> {
-        let py = ob.py();
-        let dict = ob.downcast::<PyDict>()?;
-        Ok(UrlHostParts {
-            username: dict.get_as(intern!(py, "username"))?,
-            password: dict.get_as(intern!(py, "password"))?,
-            host: dict.get_as(intern!(py, "host"))?,
-            port: dict.get_as(intern!(py, "port"))?,
-        })
+impl UrlHostParts {
+    fn is_empty(&self) -> bool {
+        self.host.is_none() && self.password.is_none() && self.host.is_none() && self.port.is_none()
     }
-}
 
-impl fmt::Display for UrlHostParts {
-    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+    fn to_writer(&self, mut w: impl Write, encode_credentials: bool) -> fmt::Result {
         match (&self.username, &self.password) {
-            (Some(username), None) => write!(f, "{}@", encode_userinfo_component(username))?,
-            (None, Some(password)) => write!(f, ":{}@", encode_userinfo_component(password))?,
+            (Some(username), None) => write!(w, "{}@", MaybeEncoded(username, encode_credentials))?,
+            (None, Some(password)) => write!(w, ":{}@", MaybeEncoded(password, encode_credentials))?,
             (Some(username), Some(password)) => write!(
-                f,
+                w,
                 "{}:{}@",
-                encode_userinfo_component(username),
-                encode_userinfo_component(password)
+                MaybeEncoded(username, encode_credentials),
+                MaybeEncoded(password, encode_credentials)
             )?,
             (None, None) => {}
         }
         if let Some(host) = &self.host {
-            write!(f, "{host}")?;
+            write!(w, "{host}")?;
         }
         if let Some(port) = self.port {
-            write!(f, ":{port}")?;
+            write!(w, ":{port}")?;
         }
         Ok(())
     }
 }
 
+impl FromPyObject<'_> for UrlHostParts {
+    fn extract_bound(ob: &Bound<'_, PyAny>) -> PyResult<Self> {
+        let py = ob.py();
+        let dict = ob.downcast::<PyDict>()?;
+        Ok(UrlHostParts {
+            username: dict.get_as(intern!(py, "username"))?,
+            password: dict.get_as(intern!(py, "password"))?,
+            host: dict.get_as(intern!(py, "host"))?,
+            port: dict.get_as(intern!(py, "port"))?,
+        })
+    }
+}
+
 fn host_to_dict<'a>(py: Python<'a>, lib_url: &Url) -> PyResult<Bound<'a, PyDict>> {
     let dict = PyDict::new(py);
     dict.set_item("username", Some(lib_url.username()).filter(|s| !s.is_empty()))?;
@@ -630,14 +692,10 @@ const USERINFO_ENCODE_SET: &AsciiSet = &CONTROLS
     // we must also percent-encode '%'
     .add(b'%');
 
-fn encode_userinfo_component(value: &str) -> Cow<'_, str> {
-    let encoded = percent_encode(value.as_bytes(), USERINFO_ENCODE_SET).to_string();
-    if encoded == value {
-        Cow::Borrowed(value)
-    } else {
-        Cow::Owned(encoded)
-    }
+fn encode_userinfo_component(value: &str) -> impl Display + '_ {
+    percent_encode(value.as_bytes(), USERINFO_ENCODE_SET)
 }
+
 // based on https://github.com/servo/rust-url/blob/1c1e406874b3d2aa6f36c5d2f3a5c2ea74af9efb/url/src/parser.rs#L161-L167
 pub fn scheme_is_special(scheme: &str) -> bool {
     matches!(scheme, "http" | "https" | "ws" | "wss" | "ftp" | "file")
diff --git a/tests/validators/test_url.py b/tests/validators/test_url.py
@@ -1319,13 +1319,64 @@ def test_multi_url_build() -> None:
 
 
 @pytest.mark.parametrize('url_type', [Url, MultiHostUrl])
-def test_url_build_encodes_credentials(url_type: type[Union[Url, MultiHostUrl]]) -> None:
+@pytest.mark.parametrize(
+    'include_kwarg',
+    [
+        pytest.param(
+            True,
+            marks=pytest.mark.xfail(
+                reason='semantics of `encode_credentials` need to be fully decided, not enabled yet'
+            ),
+        ),
+        False,
+    ],
+)
+def test_url_build_not_encode_credentials(url_type: type[Union[Url, MultiHostUrl]], include_kwarg: bool) -> None:
+    kwargs = {}
+    if include_kwarg:
+        kwargs['encode_credentials'] = False
+    url = url_type.build(
+        scheme='postgresql',
+        username='user name',
+        password='p@ss/word?#__',
+        host='example.com',
+        port=5432,
+        **kwargs,
+    )
+    assert url == url_type('postgresql://user%20name:p@ss/word?#__@example.com:5432')
+    assert str(url) == 'postgresql://user%20name:p@ss/word?#__@example.com:5432'
+
+    # NB without encoding, the special characters can seriously affect the URL
+    # parts, probably not what users want.
+    #
+    # TODO: in v3, probably should set `encode_credentials=True` by default
+    #
+    # FIXME: I guess there are similar issues with query containing #, for
+    # example? Potentially for all of these cases we could just raise an error?
+
+    if url_type is Url:
+        assert url.host == 'ss'
+        assert url.username == 'user%20name'
+        assert url.password == 'p'
+        assert url.port is None
+    else:
+        assert url.hosts() == [{'username': 'user%20name', 'password': 'p', 'host': 'ss', 'port': None}]
+
+    assert url.path == '/word'
+    assert url.query == ''
+    assert url.fragment == '__@example.com:5432'
+
+
+@pytest.mark.xfail(reason='semantics of `encode_credentials` need to be fully decided, not enabled yet')
+@pytest.mark.parametrize('url_type', [Url, MultiHostUrl])
+def test_url_build_encode_credentials(url_type: type[Union[Url, MultiHostUrl]]) -> None:
     url = url_type.build(
         scheme='postgresql',
         username='user name',
         password='p@ss/word?#__',
         host='example.com',
         port=5432,
+        encode_credentials=True,
     )
     assert url == url_type('postgresql://user%20name:p%40ss%2Fword%3F%23__@example.com:5432')
     assert str(url) == 'postgresql://user%20name:p%40ss%2Fword%3F%23__@example.com:5432'
@@ -1338,12 +1389,47 @@ def test_url_build_encodes_credentials(url_type: type[Union[Url, MultiHostUrl]])
         ]
 
 
-def test_multi_url_build_hosts_encodes_credentials() -> None:
+@pytest.mark.parametrize(
+    'include_kwarg',
+    [
+        pytest.param(
+            True,
+            marks=pytest.mark.xfail(
+                reason='semantics of `encode_credentials` need to be fully decided, not enabled yet'
+            ),
+        ),
+        False,
+    ],
+)
+def test_multi_url_build_hosts_not_encode_credentials(include_kwarg: bool) -> None:
+    kwargs = {}
+    if include_kwarg:
+        kwargs['encode_credentials'] = False
+    hosts = [
+        {'host': 'example.com', 'password': 'p@ss/word?#__', 'username': 'user name', 'port': 5431},
+        {'host': 'example.org', 'password': 'p@%ss__', 'username': 'other', 'port': 5432},
+    ]
+    url = MultiHostUrl.build(scheme='postgresql', hosts=hosts, **kwargs)
+
+    # NB: see comment in `test_url_build_not_encode_credentials` about not
+    # encoding credentials leading to VERY broken results
+
+    assert str(url) == 'postgresql://user%20name:p@ss/word?#__@example.com:5431,other:p@%ss__@example.org:5432'
+    assert url.hosts() == [
+        {'username': 'user%20name', 'password': 'p', 'host': 'ss', 'port': None},
+    ]
+    assert url.path == '/word'
+    assert url.query == ''
+    assert url.fragment == '__@example.com:5431,other:p@%ss__@example.org:5432'
+
+
+@pytest.mark.xfail(reason='semantics of `encode_credentials` need to be fully decided, not enabled yet')
+def test_multi_url_build_hosts_encode_credentials() -> None:
     hosts = [
         {'host': 'example.com', 'password': 'p@ss/word?#__', 'username': 'user name', 'port': 5431},
         {'host': 'example.org', 'password': 'p@%ss__', 'username': 'other', 'port': 5432},
     ]
-    url = MultiHostUrl.build(scheme='postgresql', hosts=hosts)
+    url = MultiHostUrl.build(scheme='postgresql', hosts=hosts, encode_credentials=True)
     assert (
         str(url) == 'postgresql://user%20name:p%40ss%2Fword%3F%23__@example.com:5431,other:p%40%25ss__@example.org:5432'
     )
