fix: encode credentials in MultiHostUrl builder

Cargo.toml

@@ -46,6 +46,7 @@ num-traits = "0.2.19"
 uuid = "1.18.1"
 jiter = { version = "0.11.0", features = ["python"] }
 hex = "0.4.3"
+percent-encoding = "2.3.1"
 
 [lib]
 name = "_pydantic_core"

src/errors/validation_exception.rs

@@ -176,10 +176,7 @@ impl ValidationError {
                 use pyo3::exceptions::PyImportError;
                 match py.import("exceptiongroup") {
                     Ok(py_mod) => match py_mod.getattr("ExceptionGroup") {
-                        Ok(group_cls) => match group_cls.call1((title, user_py_errs)) {
-                            Ok(group_instance) => Some(group_instance),
-                            Err(_) => None,
-                        },
+                        Ok(group_cls) => group_cls.call1((title, user_py_errs)).ok(),
                         Err(_) => None,
                     },
                     Err(_) => return Some(PyImportError::new_err("validation_error_cause flag requires the exceptiongroup module backport to be installed when used on Python <3.11.")),

src/input/shared.rs

@@ -232,7 +232,7 @@ pub fn fraction_as_int<'py>(input: &Bound<'py, PyAny>) -> ValResult<EitherInt<'p
     #[cfg(Py_3_12)]
     let is_integer = input.call_method0("is_integer")?.extract::<bool>()?;
     #[cfg(not(Py_3_12))]
-    let is_integer = input.getattr("denominator")?.extract::<i64>().map_or(false, |d| d == 1);
+    let is_integer = input.getattr("denominator")?.extract::<i64>().is_ok_and(|d| d == 1);
 
     if is_integer {
         #[cfg(Py_3_11)]

src/url.rs

@@ -7,6 +7,7 @@ use std::sync::OnceLock;
 
 use idna::punycode::decode_to_string;
 use jiter::{PartialMode, StringCacheMode};
+use percent_encoding::{percent_encode, NON_ALPHANUMERIC};
 use pyo3::exceptions::PyValueError;
 use pyo3::pyclass::CompareOp;
 use pyo3::sync::OnceLockExt;
@@ -536,9 +537,14 @@ impl FromPyObject<'_> for UrlHostParts {
 impl fmt::Display for UrlHostParts {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         match (&self.username, &self.password) {
-            (Some(username), None) => write!(f, "{username}@")?,
-            (None, Some(password)) => write!(f, ":{password}@")?,
-            (Some(username), Some(password)) => write!(f, "{username}:{password}@")?,
+            (Some(username), None) => write!(f, "{}@", encode_userinfo_component(username))?,
+            (None, Some(password)) => write!(f, ":{}@", encode_userinfo_component(password))?,
+            (Some(username), Some(password)) => write!(
+                f,
+                "{}:{}@",
+                encode_userinfo_component(username),
+                encode_userinfo_component(password)
+            )?,
             (None, None) => {}
         }
         if let Some(host) = &self.host {
@@ -596,6 +602,14 @@ fn is_punnycode_domain(lib_url: &Url, domain: &str) -> bool {
     scheme_is_special(lib_url.scheme()) && domain.split('.').any(|part| part.starts_with(PUNYCODE_PREFIX))
 }
 
+fn encode_userinfo_component(value: &str) -> Cow<'_, str> {
+    let encoded = percent_encode(value.as_bytes(), NON_ALPHANUMERIC).to_string();
+    if encoded == value {
+        Cow::Borrowed(value)
+    } else {
+        Cow::Owned(encoded)
+    }
+}
 // 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")

tests/validators/test_url.py

@@ -1318,6 +1318,34 @@ def test_multi_url_build() -> None:
     assert str(url) == 'postgresql://testuser:[email protected]:5432/database?sslmode=require#test'
 
 
+def test_multi_url_build_encodes_credentials() -> None:
+    url = MultiHostUrl.build(
+        scheme='postgresql',
+        username='user name',
+        password='p@ss/word?#',
+        host='example.com',
+        port=5432,
+    )
+    assert url == MultiHostUrl('postgresql://user%20name:p%40ss%2Fword%3F%[email protected]:5432')
+    assert str(url) == 'postgresql://user%20name:p%40ss%2Fword%3F%[email protected]:5432'
+    assert url.hosts() == [
+        {'username': 'user%20name', 'password': 'p%40ss%2Fword%3F%23', 'host': 'example.com', 'port': 5432}
+    ]
+
+
+def test_multi_url_build_hosts_encodes_credentials() -> None:
+    hosts = [
+        {'host': 'example.com', 'password': 'p@ss/word?#', 'username': 'user name', 'port': 5431},
+        {'host': 'example.org', 'password': 'pa%ss', 'username': 'other', 'port': 5432},
+    ]
+    url = MultiHostUrl.build(scheme='postgresql', hosts=hosts)
+    assert str(url) == 'postgresql://user%20name:p%40ss%2Fword%3F%[email protected]:5431,other:pa%[email protected]:5432'
+    assert url.hosts() == [
+        {'username': 'user%20name', 'password': 'p%40ss%2Fword%3F%23', 'host': 'example.com', 'port': 5431},
+        {'username': 'other', 'password': 'pa%25ss', 'host': 'example.org', 'port': 5432},
+    ]
+
+
 @pytest.mark.parametrize('field', ['host', 'password', 'username', 'port'])
 def test_multi_url_build_hosts_set_with_single_value(field) -> None:
     """Hosts can't be provided with any single url values."""