disallow non-UTF-8 symbols in path

Author: arjan-bal

grpc/src/byte_str.rs

@@ -21,111 +21,37 @@
  * IN THE SOFTWARE.
  *
  */
+
+use core::str;
 use std::ops::Deref;
 
 use bytes::Bytes;
 
 /// A cheaply cloneable and sliceable chunk of contiguous memory.
-///
-/// The bytes held by `ByteStr` are arbitrary and may not be valid UTF-8.
 #[derive(Debug, Default, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
 pub struct ByteStr {
+    // Invariant: bytes contains valid UTF-8
     bytes: Bytes,
 }
 
-impl ByteStr {
-    /// Strips a prefix, returning a new zero-copy ByteStr.
-    #[inline]
-    pub(crate) fn strip_prefix(&self, prefix: &[u8]) -> Option<ByteStr> {
-        if self.starts_with(prefix) {
-            Some(ByteStr {
-                bytes: self.bytes.slice(prefix.len()..),
-            })
-        } else {
-            None
-        }
-    }
-}
-
 impl Deref for ByteStr {
-    type Target = [u8];
+    type Target = str;
 
     #[inline]
-    fn deref(&self) -> &[u8] {
-        &self.bytes
+    fn deref(&self) -> &str {
+        let b: &[u8] = self.bytes.as_ref();
+        // The invariant of `bytes` is that it contains valid UTF-8 allows us
+        // to unwrap.
+        str::from_utf8(b).unwrap()
     }
 }
 
 impl From<String> for ByteStr {
     #[inline]
     fn from(src: String) -> ByteStr {
         ByteStr {
+            // Invariant: src is a String so contains valid UTF-8.
             bytes: Bytes::from(src),
         }
     }
 }
-
-impl<'a> TryFrom<&'a ByteStr> for &'a str {
-    type Error = std::str::Utf8Error;
-
-    #[inline]
-    fn try_from(value: &'a ByteStr) -> Result<Self, Self::Error> {
-        std::str::from_utf8(value)
-    }
-}
-
-impl TryFrom<ByteStr> for String {
-    type Error = std::str::Utf8Error;
-
-    #[inline]
-    fn try_from(value: ByteStr) -> Result<Self, Self::Error> {
-        let s = std::str::from_utf8(&value)?;
-        Ok(s.to_owned())
-    }
-}
-
-impl PartialEq<str> for ByteStr {
-    #[inline]
-    fn eq(&self, other: &str) -> bool {
-        self.bytes == other.as_bytes()
-    }
-}
-
-impl<'a> PartialEq<&'a str> for ByteStr {
-    #[inline]
-    fn eq(&self, other: &&'a str) -> bool {
-        self.bytes == other.as_bytes()
-    }
-}
-
-impl PartialEq<ByteStr> for str {
-    #[inline]
-    fn eq(&self, other: &ByteStr) -> bool {
-        self.as_bytes() == other.bytes
-    }
-}
-
-impl PartialEq<ByteStr> for &str {
-    #[inline]
-    fn eq(&self, other: &ByteStr) -> bool {
-        self.as_bytes() == other.bytes
-    }
-}
-
-impl FromIterator<u8> for ByteStr {
-    #[inline]
-    fn from_iter<T: IntoIterator<Item = u8>>(iter: T) -> Self {
-        ByteStr {
-            bytes: Bytes::from_iter(iter),
-        }
-    }
-}
-
-impl From<&'static str> for ByteStr {
-    #[inline]
-    fn from(src: &'static str) -> ByteStr {
-        ByteStr {
-            bytes: Bytes::from_static(src.as_bytes()),
-        }
-    }
-}

grpc/src/client/channel.rs

@@ -36,7 +36,6 @@ use std::vec;
 use serde_json::json;
 use tokio::sync::mpsc;
 use tokio::sync::watch;
-use url::Url; // NOTE: http::Uri requires non-empty authority portion of URI
 
 use crate::StatusCodeError;
 use crate::StatusError;
@@ -248,10 +247,9 @@ impl PersistentChannel {
         options: ChannelOptions,
         credentials: Arc<dyn DynChannelCredentials>,
     ) -> Self {
-        // TODO(arjan-bal): Return errors here instead of panicking.
-        let target = Url::from_str(&target.into()).unwrap();
+        // TODO(nathanielford): Return errors here instead of panicking.
+        let target = Target::from_str(&target.into()).unwrap();
         let resolver_builder = global_registry().get(target.scheme()).unwrap();
-        let target = name_resolution::Target::from(target);
         let authority = options
             .channel_authority
             .clone()

grpc/src/client/load_balancing/graceful_switch.rs

@@ -452,10 +452,12 @@ mod test {
         let PickResult::Pick(pick) = pick else {
             panic!("unexpected pick result: {:?}", pick);
         };
-        let received_address = &pick.subchannel.address().address;
+        let received_address = &pick.subchannel.address().address.to_string();
+        // It's good practice to create the expected value once.
+        let expected_address = name.to_string();
 
         // Check for inequality and panic with a detailed message if they don't match.
-        assert_eq!(received_address, name);
+        assert_eq!(received_address, &expected_address);
     }
 
     fn move_subchannel_to_state(
@@ -654,7 +656,7 @@ mod test {
             .resolver_update(update.clone(), Some(&parsed_config2), &mut *tcc)
             .unwrap();
         let subchannel = verify_subchannel_creation_from_policy(&mut rx_events);
-        assert_eq!(subchannel.address().address, "127.0.0.1:1234");
+        assert_eq!(&*subchannel.address().address, "127.0.0.1:1234");
         assert_channel_empty(&mut rx_events);
     }
 

grpc/src/client/load_balancing/pick_first.rs

@@ -425,7 +425,7 @@ impl PickFirstPolicy {
         // Partition by family (Basic IPv6 detection via colon).
         let (ipv6, ipv4): (Vec<Address>, Vec<Address>) = tcp_addresses
             .into_iter()
-            .partition(|addr| addr.address.contains(&b':'));
+            .partition(|addr| addr.address.contains(':'));
 
         // Interleave the two lists so ipv6 and ipv4 addresses are alternated.
         let mut interleaved = Vec::with_capacity(ipv6.len() + ipv4.len() + unknown.len());
@@ -738,9 +738,9 @@ mod test {
     use std::time::Duration;
 
     use super::*;
-    use crate::client::load_balancing::test_utils::TestChannelController;
-    use crate::client::load_balancing::test_utils::TestEvent;
-    use crate::client::load_balancing::test_utils::TestWorkScheduler;
+    use crate::client::load_balancing::test_utils::{
+        TestChannelController, TestEvent, TestWorkScheduler,
+    };
 
     const DEFAULT_TEST_DURATION: Duration = Duration::from_secs(10);
 
@@ -928,7 +928,7 @@ mod test {
         let res = state.picker.pick(&RequestHeaders::default());
         match res {
             PickResult::Pick(pick) => {
-                assert_eq!(pick.subchannel.address().address, "addr1")
+                assert_eq!(pick.subchannel.address().address.to_string(), "addr1")
             }
             other => panic!("unexpected pick result {:?}", other),
         }
@@ -942,7 +942,7 @@ mod test {
 
         // Should connect to addr2.
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address, "addr2");
+        assert_eq!(addr.address.to_string(), "addr2");
 
         // Simulate addr2 succeeding.
         let sc2 = policy.subchannels[1].clone();
@@ -986,16 +986,25 @@ mod test {
 
         // Should create new subchannel for addr2 (was cleared by cleanup).
         let sc2 = expect_new_subchannel(&rx);
-        assert_eq!(sc2.address().address, "addr2");
+        assert_eq!(sc2.address().address.to_string(), "addr2");
         // Should create new subchannel for addr3 (was not in previous list).
         let sc3 = expect_new_subchannel(&rx);
-        assert_eq!(sc3.address().address, "addr3");
+        assert_eq!(sc3.address().address.to_string(), "addr3");
 
         // Should NOT have any more events (no Connect, no UpdatePicker),
         // because it stuck to the original selected subchannel.
         assert!(rx.try_recv().is_err(), "unexpected event");
 
-        assert_eq!(policy.selected.as_ref().unwrap().address().address, "addr1");
+        assert_eq!(
+            policy
+                .selected
+                .as_ref()
+                .unwrap()
+                .address()
+                .address
+                .to_string(),
+            "addr1"
+        );
     }
 
     // If all addresses fail during a connection pass, the LB should update to
@@ -1079,7 +1088,7 @@ mod test {
         let mut resulting_addrs = Vec::with_capacity(NUM_ADDRS);
         for _ in 0..NUM_ADDRS {
             let sc = expect_new_subchannel(&rx);
-            resulting_addrs.push(sc.address().address);
+            resulting_addrs.push(sc.address().address.to_string());
         }
 
         // Mock shuffler reverses endpoints: [EP3, EP2, EP1]
@@ -1148,9 +1157,9 @@ mod test {
 
         // Should only create subchannels for addr1 and addr2 (2 unique addrs).
         let sc1 = expect_new_subchannel(&rx);
-        assert_eq!(sc1.address().address, "addr1");
+        assert_eq!(sc1.address().address.to_string(), "addr1");
         let sc2 = expect_new_subchannel(&rx);
-        assert_eq!(sc2.address().address, "addr2");
+        assert_eq!(sc2.address().address.to_string(), "addr2");
 
         // Verify no 3rd subchannel was created.
         while let Ok(event) = rx.try_recv() {
@@ -1217,7 +1226,7 @@ mod test {
 
         // Expect Connect event for addr2 due to timer expiration.
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address, "addr2");
+        assert_eq!(addr.address.to_string(), "addr2");
     }
 
     // If all addresses fail during a connection pass, the LB should enter
@@ -1250,7 +1259,7 @@ mod test {
 
         // Should automatically call connect() again.
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address, "addr1");
+        assert_eq!(addr.address.to_string(), "addr1");
     }
 
     // If the LB is in steady state, and a new address becomes ready, it should
@@ -1265,7 +1274,7 @@ mod test {
 
         // Should failover to addr2: expect Connect(addr2).
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address, "addr2");
+        assert_eq!(addr.address.to_string(), "addr2");
 
         // While addr2 is connecting, simulate addr1 going IDLE (backoff over).
         policy.subchannel_update(
@@ -1300,7 +1309,7 @@ mod test {
         );
         expect_request_resolution(&rx);
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address, "addr1");
+        assert_eq!(addr.address.to_string(), "addr1");
 
         // Confirm LB is in steady state.
         assert!(policy.steady_state.is_some());
@@ -1317,7 +1326,7 @@ mod test {
 
         // Now it should automatically call connect() again.
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address, "addr1");
+        assert_eq!(addr.address.to_string(), "addr1");
 
         // Simulate addr1 successfully connecting and becoming READY.
         policy.subchannel_update(
@@ -1335,7 +1344,7 @@ mod test {
         let res = state.picker.pick(&RequestHeaders::default());
         match res {
             PickResult::Pick(pick) => {
-                assert_eq!(pick.subchannel.address().address, "addr1");
+                assert_eq!(pick.subchannel.address().address.to_string(), "addr1");
             }
             other => panic!("unexpected pick result {:?}", other),
         }
@@ -1354,7 +1363,7 @@ mod test {
 
         // Expect Connect(addr2).
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address, "addr2");
+        assert_eq!(addr.address.to_string(), "addr2");
 
         // Simulate addr1 backing off and transitioning to IDLE early
         // (while addr2 is still connecting).
@@ -1393,7 +1402,7 @@ mod test {
         // Expect an immediate Connect(addr1) event triggered by the exhaustion
         // loop sweeping up the early IDLE subchannel.
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address, "addr1");
+        assert_eq!(addr.address.to_string(), "addr1");
     }
 
     // This test is meant to validate that if a new address with different
@@ -1492,7 +1501,7 @@ mod test {
         policy.work(None, controller.as_mut());
 
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address, "addr2");
+        assert_eq!(addr.address.to_string(), "addr2");
 
         // 2. Simulate addr2 failing first while addr1 is still in flight.
         let sc2 = policy.subchannels[1].clone();
@@ -1574,7 +1583,7 @@ mod test {
         let res = state.picker.pick(&RequestHeaders::default());
         let sc1 = match res {
             PickResult::Pick(pick) => {
-                assert_eq!(pick.subchannel.address().address, "addr1");
+                assert_eq!(pick.subchannel.address().address.to_string(), "addr1");
                 pick.subchannel
             }
             other => panic!("unexpected pick result {:?}", other),
@@ -1610,7 +1619,7 @@ mod test {
 
         // 7. Verify that the policy initiates a reconnection to addr1.
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address, "addr1");
+        assert_eq!(addr.address.to_string(), "addr1");
 
         // And the picker goes to Connecting.
         let state = expect_picker_update(&rx);

grpc/src/client/load_balancing/round_robin.rs

@@ -1018,11 +1018,11 @@ mod test {
         let all_subchannels = verify_subchannel_creation(&mut rx_events, 2);
         let subchannel_one = all_subchannels
             .iter()
-            .find(|sc| sc.address().address == "subchannel_one")
+            .find(|sc| sc.address().address == "subchannel_one".to_string().into())
             .unwrap();
         let subchannel_two = all_subchannels
             .iter()
-            .find(|sc| sc.address().address == "subchannel_two")
+            .find(|sc| sc.address().address == "subchannel_two".to_string().into())
             .unwrap();
 
         move_subchannel_to_state(
@@ -1081,11 +1081,11 @@ mod test {
         let new_subchannels = verify_subchannel_creation(&mut rx_events, 2);
         let new_sc = new_subchannels
             .iter()
-            .find(|sc| sc.address().address == "new")
+            .find(|sc| sc.address().address == "new".to_string().into())
             .unwrap();
         let old_sc = new_subchannels
             .iter()
-            .find(|sc| sc.address().address == "subchannel_two")
+            .find(|sc| sc.address().address == "subchannel_two".to_string().into())
             .unwrap();
 
         move_subchannel_to_state(

grpc/src/client/name_resolution/dns/mod.rs

@@ -315,11 +315,8 @@ struct ParseResult {
 fn parse_endpoint_and_authority(target: &Target) -> Result<ParseResult, String> {
     // Parse the endpoint.
     let endpoint = target.path();
-    let endpoint = endpoint.strip_prefix(b"/").unwrap_or(endpoint);
-    let host_port = (&endpoint)
-        .try_into()
-        .map_err(|err| format!("target hostname contains invalid UTF-8 symbols: {err}"))?;
-    let parse_result = parse_host_port(host_port, DEFAULT_PORT)
+    let endpoint = endpoint.strip_prefix("/").unwrap_or(endpoint);
+    let parse_result = parse_host_port(endpoint, DEFAULT_PORT)
         .map_err(|err| format!("Failed to parse target {target}: {err}"))?;
     let endpoint = parse_result.ok_or("Received empty endpoint host.".to_string())?;