Merge pull request #674 from Johannesd3/proxy-support

[Tokio migration] Add back hyper-proxy

Cargo.lock

@@ -583,6 +583,21 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "futures"
+version = "0.3.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f55667319111d593ba876406af7c409c0ebb44dc4be6132a783ccf163ea14c1"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
 [[package]]
 name = "futures-channel"
 version = "0.3.13"
@@ -590,6 +605,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8c2dd2df839b57db9ab69c2c9d8f3e8c81984781937fe2807dc6dcf3b2ad2939"
 dependencies = [
  "futures-core",
+ "futures-sink",
 ]
 
 [[package]]
@@ -609,6 +625,12 @@ dependencies = [
  "futures-util",
 ]
 
+[[package]]
+name = "futures-io"
+version = "0.3.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d71c2c65c57704c32f5241c1223167c2c3294fd34ac020c807ddbe6db287ba59"
+
 [[package]]
 name = "futures-macro"
 version = "0.3.13"
@@ -639,10 +661,13 @@ version = "0.3.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1812c7ab8aedf8d6f2701a43e1243acdbcc2b36ab26e2ad421eb99ac963d96d1"
 dependencies = [
+ "futures-channel",
  "futures-core",
+ "futures-io",
  "futures-macro",
  "futures-sink",
  "futures-task",
+ "memchr",
  "pin-project-lite",
  "pin-utils",
  "proc-macro-hack",
@@ -853,30 +878,29 @@ dependencies = [
 ]
 
 [[package]]
-name = "h2"
+name = "headers"
-version = "0.3.0"
+version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6b67e66362108efccd8ac053abafc8b7a8d86a37e6e48fc4f6f7485eb5e9e6a5"
+checksum = "f0b7591fb62902706ae8e7aaff416b1b0fa2c0fd0878b46dc13baa3712d8a855"
 dependencies = [
+ "base64",
+ "bitflags",
  "bytes",
- "fnv",
+ "headers-core",
- "futures-core",
- "futures-sink",
- "futures-util",
  "http",
- "indexmap",
+ "mime",
- "slab",
+ "sha-1",
- "tokio",
+ "time 0.1.43",
- "tokio-util",
- "tracing",
- "tracing-futures",
 ]
 
 [[package]]
-name = "hashbrown"
+name = "headers-core"
-version = "0.9.1"
+version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d7afe4a420e3fe79967a00898cc1f4db7c8a49a9333a29f8a4bd76a253d5cd04"
+checksum = "e7f66481bfee273957b1f20485a4ff3362987f85b2c236580d81b4eb7a326429"
+dependencies = [
+ "http",
+]
 
 [[package]]
 name = "heck"
@@ -972,7 +996,6 @@ dependencies = [
  "futures-channel",
  "futures-core",
  "futures-util",
- "h2",
  "http",
  "http-body",
  "httparse",
@@ -986,6 +1009,21 @@ dependencies = [
  "want",
 ]
 
+[[package]]
+name = "hyper-proxy"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ca815a891b24fdfb243fa3239c86154392b0953ee584aa1a2a1f66d20cbe75cc"
+dependencies = [
+ "bytes",
+ "futures",
+ "headers",
+ "http",
+ "hyper",
+ "tokio",
+ "tower-service",
+]
+
 [[package]]
 name = "ident_case"
 version = "1.0.1"
@@ -1024,16 +1062,6 @@ dependencies = [
  "libc",
 ]
 
-[[package]]
-name = "indexmap"
-version = "1.6.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4fb1fa934250de4de8aef298d81c729a7d33d8c239daa3a7575e6b92bfc7313b"
-dependencies = [
- "autocfg",
- "hashbrown",
-]
-
 [[package]]
 name = "instant"
 version = "0.1.9"
@@ -1335,13 +1363,14 @@ dependencies = [
  "base64",
  "byteorder",
  "bytes",
- "cfg-if 1.0.0",
  "env_logger",
  "futures-core",
  "futures-util",
  "hmac",
+ "http",
  "httparse",
  "hyper",
+ "hyper-proxy",
  "librespot-protocol",
  "log",
  "num-bigint",
@@ -1472,6 +1501,12 @@ version = "2.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0ee1c47aaa256ecabcaea351eae4a9b01ef39ed810004e298d2511ed284b1525"
 
+[[package]]
+name = "mime"
+version = "0.3.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2a60c7ce501c71e03a9c9c0d35b861413ae925bd979cc7a4e30d060069aaac8d"
+
 [[package]]
 name = "miniz_oxide"
 version = "0.4.3"
@@ -2694,16 +2729,6 @@ dependencies = [
  "lazy_static",
 ]
 
-[[package]]
-name = "tracing-futures"
-version = "0.2.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "97d095ae15e245a057c8e8451bab9b3ee1e1f68e9ba2b4fbc18d0ac5237835f2"
-dependencies = [
- "pin-project",
- "tracing",
-]
-
 [[package]]
 name = "try-lock"
 version = "0.2.3"

Cargo.toml

@@ -59,7 +59,6 @@ sha-1 = "0.9"
 
 [features]
 apresolve = ["librespot-core/apresolve"]
-apresolve-http2 = ["librespot-core/apresolve-http2"]
 
 alsa-backend = ["librespot-playback/alsa-backend"]
 portaudio-backend = ["librespot-playback/portaudio-backend"]

core/Cargo.toml

@@ -17,12 +17,13 @@ aes = "0.6"
 base64 = "0.13"
 byteorder = "1.4"
 bytes = "1.0"
-cfg-if = "1"
 futures-core = { version = "0.3", default-features = false }
 futures-util = { version = "0.3", default-features = false, features = ["alloc", "bilock", "unstable", "sink"] }
 hmac = "0.10"
 httparse = "1.3"
+http = "0.2"
 hyper = { version = "0.14", optional = true, features = ["client", "tcp", "http1"] }
+hyper-proxy = { version = "0.9.1", optional = true, default-features = false }
 log = "0.4"
 num-bigint = "0.3"
 num-integer = "0.1"
@@ -51,5 +52,4 @@ env_logger = "*"
 tokio = {version = "1.0", features = ["macros"] }
 
 [features]
-apresolve = ["hyper"]
+apresolve = ["hyper", "hyper-proxy"]
-apresolve-http2 = ["apresolve", "hyper/http2"]

core/src/apresolve.rs

@@ -1,73 +1,68 @@
-const AP_FALLBACK: &str = "ap.spotify.com:443";
+use std::error::Error;
 
+use hyper::client::HttpConnector;
+use hyper::{Body, Client, Method, Request, Uri};
+use hyper_proxy::{Intercept, Proxy, ProxyConnector};
+use serde::Deserialize;
 use url::Url;
 
-cfg_if! {
+use super::AP_FALLBACK;
-    if #[cfg(feature = "apresolve")] {
-        const APRESOLVE_ENDPOINT: &str = "http://apresolve.spotify.com:80";
 
-        use std::error::Error;
+const APRESOLVE_ENDPOINT: &str = "http://apresolve.spotify.com:80";
 
-        use hyper::{Body, Client, Method, Request, Uri};
+#[derive(Clone, Debug, Deserialize)]
-        use serde::{Serialize, Deserialize};
+struct APResolveData {
-
+    ap_list: Vec<String>,
-        use crate::proxytunnel::ProxyTunnel;
+}
-
+
-        #[derive(Clone, Debug, Serialize, Deserialize)]
+async fn try_apresolve(
-        pub struct APResolveData {
+    proxy: Option<&Url>,
-            ap_list: Vec<String>,
+    ap_port: Option<u16>,
-        }
+) -> Result<String, Box<dyn Error>> {
-
+    let port = ap_port.unwrap_or(443);
-        async fn apresolve(proxy: &Option<Url>, ap_port: &Option<u16>) -> Result<String, Box<dyn Error>> {
+
-            let port = ap_port.unwrap_or(443);
+    let mut req = Request::new(Body::empty());
-
+    *req.method_mut() = Method::GET;
-            let req = Request::builder()
+    // panic safety: APRESOLVE_ENDPOINT above is valid url.
-                .method(Method::GET)
+    *req.uri_mut() = APRESOLVE_ENDPOINT.parse().expect("invalid AP resolve URL");
-                .uri(
+
-                    APRESOLVE_ENDPOINT
+    let response = if let Some(url) = proxy {
-                        .parse::<Uri>()
+        // Panic safety: all URLs are valid URIs
-                        .expect("invalid AP resolve URL"),
+        let uri = url.to_string().parse().unwrap();
-                )
+        let proxy = Proxy::new(Intercept::All, uri);
-                .body(Body::empty())?;
+        let connector = HttpConnector::new();
-
+        let proxy_connector = ProxyConnector::from_proxy_unsecured(connector, proxy);
-            let response = if let Some(url) = proxy {
+        Client::builder()
-                Client::builder()
+            .build(proxy_connector)
-                    .build(ProxyTunnel::new(&url.socket_addrs(|| None)?[..])?)
+            .request(req)
-                    .request(req)
+            .await?
-                    .await?
+    } else {
-            } else {
+        Client::new().request(req).await?
-                Client::new().request(req).await?
+    };
-            };
+
-
+    let body = hyper::body::to_bytes(response.into_body()).await?;
-            let body = hyper::body::to_bytes(response.into_body()).await?;
+    let data: APResolveData = serde_json::from_slice(body.as_ref())?;
-            let data: APResolveData = serde_json::from_slice(body.as_ref())?;
+
-
+    let ap = if ap_port.is_some() || proxy.is_some() {
-            let ap = if ap_port.is_some() || proxy.is_some() {
+        data.ap_list.into_iter().find_map(|ap| {
-                data.ap_list.into_iter().find_map(|ap| {
+            if ap.parse::<Uri>().ok()?.port()? == port {
-                    if ap.parse::<Uri>().ok()?.port()? == port {
+                Some(ap)
-                        Some(ap)
+            } else {
-                    } else {
+                None
-                        None
+            }
-                    }
+        })
-                })
+    } else {
-            } else {
+        data.ap_list.into_iter().next()
-                data.ap_list.into_iter().next()
+    }
-            }
+    .ok_or("empty AP List")?;
-            .ok_or("empty AP List")?;
+
-
+    Ok(ap)
-            Ok(ap)
+}
-        }
+
-
+pub async fn apresolve(proxy: Option<&Url>, ap_port: Option<u16>) -> String {
-        pub async fn apresolve_or_fallback(proxy: &Option<Url>, ap_port: &Option<u16>) -> String {
+    try_apresolve(proxy, ap_port).await.unwrap_or_else(|e| {
-            apresolve(proxy, ap_port).await.unwrap_or_else(|e| {
+        warn!("Failed to resolve Access Point: {}", e);
-                warn!("Failed to resolve Access Point: {}", e);
+        warn!("Using fallback \"{}\"", AP_FALLBACK);
-                warn!("Using fallback \"{}\"", AP_FALLBACK);
+        AP_FALLBACK.into()
-                AP_FALLBACK.into()
+    })
-            })
-        }
-    } else {
-        pub async fn apresolve_or_fallback(_: &Option<Url>, _: &Option<u16>) -> String {
-            AP_FALLBACK.to_string()
-        }
-    }
 }

core/src/connection/mod.rs

@@ -58,25 +58,11 @@ impl From<APLoginFailed> for AuthenticationError {
     }
 }
 
-pub async fn connect(addr: String, proxy: &Option<Url>) -> io::Result<Transport> {
+pub async fn connect(addr: String, proxy: Option<&Url>) -> io::Result<Transport> {
-    let socket = if let Some(proxy) = proxy {
+    let socket = if let Some(proxy_url) = proxy {
-        info!("Using proxy \"{}\"", proxy);
+        info!("Using proxy \"{}\"", proxy_url);
 
-        let mut split = addr.rsplit(':');
+        let socket_addr = proxy_url.socket_addrs(|| None).and_then(|addrs| {
-
-        let port = split
-            .next()
-            .unwrap() // will never panic, split iterator contains at least one element
-            .parse()
-            .map_err(|e| {
-                io::Error::new(io::ErrorKind::InvalidInput, format!("Invalid port: {}", e))
-            })?;
-
-        let host = split
-            .next()
-            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Missing port"))?;
-
-        let socket_addr = proxy.socket_addrs(|| None).and_then(|addrs| {
             addrs.into_iter().next().ok_or_else(|| {
                 io::Error::new(
                     io::ErrorKind::NotFound,
@@ -86,13 +72,34 @@ pub async fn connect(addr: String, proxy: &Option<Url>) -> io::Result<Transport>
         })?;
         let socket = TcpStream::connect(&socket_addr).await?;
 
-        proxytunnel::connect(socket, host, port).await?
+        let uri = addr.parse::<http::Uri>().map_err(|_| {
-    } else {
+            io::Error::new(
-        let socket_addr = addr.to_socket_addrs().and_then(|mut iter| {
+                io::ErrorKind::InvalidData,
-            iter.next().ok_or_else(|| {
+                "Can't parse access point address",
-                io::Error::new(io::ErrorKind::NotFound, "Can't resolve server address")
+            )
-            })
         })?;
+        let host = uri.host().ok_or_else(|| {
+            io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "The access point address contains no hostname",
+            )
+        })?;
+        let port = uri.port().ok_or_else(|| {
+            io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "The access point address contains no port",
+            )
+        })?;
+
+        proxytunnel::proxy_connect(socket, host, port.as_str()).await?
+    } else {
+        let socket_addr = addr.to_socket_addrs()?.next().ok_or_else(|| {
+            io::Error::new(
+                io::ErrorKind::NotFound,
+                "Can't resolve access point address",
+            )
+        })?;
+
         TcpStream::connect(&socket_addr).await?
     };
 

core/src/lib.rs

@@ -2,15 +2,12 @@
 
 #[macro_use]
 extern crate log;
-#[macro_use]
-extern crate cfg_if;
 
 use librespot_protocol as protocol;
 
 #[macro_use]
 mod component;
 
-mod apresolve;
 pub mod audio_key;
 pub mod authentication;
 pub mod cache;
@@ -25,3 +22,15 @@ pub mod session;
 pub mod spotify_id;
 pub mod util;
 pub mod version;
+
+const AP_FALLBACK: &str = "ap.spotify.com:443";
+
+#[cfg(feature = "apresolve")]
+mod apresolve;
+
+#[cfg(not(feature = "apresolve"))]
+mod apresolve {
+    pub async fn apresolve(_: Option<&url::Url>, _: Option<u16>) -> String {
+        return super::AP_FALLBACK.into();
+    }
+}

core/src/proxytunnel.rs

@@ -2,16 +2,16 @@ use std::io;
 
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
-pub async fn connect<T: AsyncRead + AsyncWrite + Unpin>(
+pub async fn proxy_connect<T: AsyncRead + AsyncWrite + Unpin>(
     mut proxy_connection: T,
     connect_host: &str,
-    connect_port: u16,
+    connect_port: &str,
 ) -> io::Result<T> {
     let mut buffer = Vec::new();
     buffer.extend_from_slice(b"CONNECT ");
     buffer.extend_from_slice(connect_host.as_bytes());
     buffer.push(b':');
-    buffer.extend_from_slice(connect_port.to_string().as_bytes());
+    buffer.extend_from_slice(connect_port.as_bytes());
     buffer.extend_from_slice(b" HTTP/1.1\r\n\r\n");
 
     proxy_connection.write_all(buffer.as_ref()).await?;
@@ -49,61 +49,7 @@ pub async fn connect<T: AsyncRead + AsyncWrite + Unpin>(
         }
 
         if offset >= buffer.len() {
-            buffer.resize(buffer.len() * 2, 0);
+            buffer.resize(buffer.len() + 100, 0);
-        }
-    }
-}
-
-cfg_if! {
-    if #[cfg(feature = "apresolve")] {
-        use std::future::Future;
-        use std::net::{SocketAddr, ToSocketAddrs};
-        use std::pin::Pin;
-        use std::task::Poll;
-
-        use hyper::service::Service;
-        use hyper::Uri;
-        use tokio::net::TcpStream;
-
-        #[derive(Clone)]
-        pub struct ProxyTunnel {
-            proxy_addr: SocketAddr,
-        }
-
-        impl ProxyTunnel {
-            pub fn new<T: ToSocketAddrs>(addr: T) -> io::Result<Self> {
-                let addr = addr.to_socket_addrs()?.next().ok_or_else(|| {
-                    io::Error::new(io::ErrorKind::InvalidInput, "No socket address given")
-                })?;
-                Ok(Self { proxy_addr: addr })
-            }
-        }
-
-        impl Service<Uri> for ProxyTunnel {
-            type Response = TcpStream;
-            type Error = io::Error;
-            type Future = Pin<Box<dyn Future<Output = io::Result<TcpStream>> + Send>>;
-
-            fn poll_ready(&mut self, _: &mut std::task::Context<'_>) -> Poll<io::Result<()>> {
-                Poll::Ready(Ok(()))
-            }
-
-            fn call(&mut self, url: Uri) -> Self::Future {
-                let proxy_addr = self.proxy_addr;
-                let fut = async move {
-                    let host = url
-                        .host()
-                        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Host is missing"))?;
-                    let port = url
-                        .port()
-                        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Port is missing"))?;
-
-                    let conn = TcpStream::connect(proxy_addr).await?;
-                    connect(conn, host, port.as_u16()).await
-                };
-
-                Box::pin(fut)
-            }
         }
     }
 }

core/src/session.rs

@@ -16,7 +16,7 @@ use thiserror::Error;
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::UnboundedReceiverStream;
 
-use crate::apresolve::apresolve_or_fallback;
+use crate::apresolve::apresolve;
 use crate::audio_key::AudioKeyManager;
 use crate::authentication::Credentials;
 use crate::cache::Cache;
@@ -67,10 +67,10 @@ impl Session {
         credentials: Credentials,
         cache: Option<Cache>,
     ) -> Result<Session, SessionError> {
-        let ap = apresolve_or_fallback(&config.proxy, &config.ap_port).await;
+        let ap = apresolve(config.proxy.as_ref(), config.ap_port).await;
 
         info!("Connecting to AP \"{}\"", ap);
-        let mut conn = connection::connect(ap, &config.proxy).await?;
+        let mut conn = connection::connect(ap, config.proxy.as_ref()).await?;
 
         let reusable_credentials =
             connection::authenticate(&mut conn, credentials, &config.device_id).await?;