Add support for http proxy

Currently only http proxy (no https) is supported.

core/Cargo.toml

@@ -12,8 +12,9 @@ base64 = "0.5.0"
 byteorder = "1.0"
 bytes = "0.4"
 error-chain = { version = "0.9.0", default_features = false }
-extprim = "1.5.1" 
+extprim = "1.5.1"
 futures = "0.1.8"
+httparse = "1.2.4"
 hyper = "0.11.2"
 lazy_static = "0.2.0"
 log = "0.3.5"

core/src/apresolve.rs

@@ -1,7 +1,7 @@
-const AP_FALLBACK: &'static str = "ap.spotify.com:80";
+const AP_FALLBACK: &'static str = "ap.spotify.com:443";
 const APRESOLVE_ENDPOINT: &'static str = "http://apresolve.spotify.com/";
 
-use futures::{Future, Stream};
+use futures::{future, Future, Stream};
 use hyper::{self, Client, Uri};
 use serde_json;
 use std::str::FromStr;
@@ -40,15 +40,24 @@ fn apresolve(handle: &Handle) -> Box<Future<Item = String, Error = Error>> {
     Box::new(ap)
 }
 
-pub(crate) fn apresolve_or_fallback<E>(handle: &Handle) -> Box<Future<Item = String, Error = E>>
+pub(crate) fn apresolve_or_fallback<E>(
+    handle: &Handle,
+    proxy: &Option<String>,
+) -> Box<Future<Item = String, Error = E>>
 where
     E: 'static,
 {
-    let ap = apresolve(handle).or_else(|e| {
-        warn!("Failed to resolve Access Point: {}", e.description());
-        warn!("Using fallback \"{}\"", AP_FALLBACK);
-        Ok(AP_FALLBACK.into())
-    });
-
-    Box::new(ap)
+    if proxy.is_some() {
+        // TODO: Use a proper proxy library and filter out a 443 proxy instead of relying on fallback.
+        //       The problem with current libraries (hyper-proxy, reqwest) is that they depend on TLS
+        //       and this is a dependency we might not want.
+        Box::new(future::result(Ok(AP_FALLBACK.into())))
+    } else {
+        let ap = apresolve(handle).or_else(|e| {
+            warn!("Failed to resolve Access Point: {}", e.description());
+            warn!("Using fallback \"{}\"", AP_FALLBACK);
+            Ok(AP_FALLBACK.into())
+        });
+        Box::new(ap)
+    }
 }

core/src/config.rs

@@ -8,6 +8,7 @@ use version;
 pub struct SessionConfig {
     pub user_agent: String,
     pub device_id: String,
+    pub proxy: Option<String>,
 }
 
 impl Default for SessionConfig {
@@ -16,6 +17,7 @@ impl Default for SessionConfig {
         SessionConfig {
             user_agent: version::version_string(),
             device_id: device_id,
+            proxy: None,
         }
     }
 }

core/src/connection/mod.rs

@@ -5,9 +5,11 @@ pub use self::codec::APCodec;
 pub use self::handshake::handshake;
 
 use futures::{Future, Sink, Stream};
+use hyper::Uri;
 use protobuf::{self, Message};
 use std::io;
 use std::net::ToSocketAddrs;
+use std::str::FromStr;
 use tokio_core::net::TcpStream;
 use tokio_core::reactor::Handle;
 use tokio_io::codec::Framed;
@@ -15,17 +17,42 @@ use tokio_io::codec::Framed;
 use authentication::Credentials;
 use version;
 
+use proxytunnel;
+
 pub type Transport = Framed<TcpStream, APCodec>;
 
 pub fn connect<A: ToSocketAddrs>(
     addr: A,
     handle: &Handle,
+    proxy: &Option<String>,
 ) -> Box<Future<Item = Transport, Error = io::Error>> {
-    let addr = addr.to_socket_addrs().unwrap().next().unwrap();
-    let socket = TcpStream::connect(&addr, handle);
-    let connection = socket.and_then(|socket| handshake(socket));
+    let (addr, connect_url) = match *proxy {
+        Some(ref url) => {
+            let url = Uri::from_str(url).expect("Malformed proxy address");
+            let host = url.host().expect("Malformed proxy address: no host");
+            let port = url.port().unwrap_or(3128);
 
-    Box::new(connection)
+            (
+                format!("{}:{}", host, port)
+                    .to_socket_addrs()
+                    .unwrap()
+                    .next()
+                    .unwrap(),
+                Some(addr.to_socket_addrs().unwrap().next().unwrap()),
+            )
+        }
+        None => (addr.to_socket_addrs().unwrap().next().unwrap(), None),
+    };
+
+    let socket = TcpStream::connect(&addr, handle);
+    if let Some(connect_url) = connect_url {
+        let connection =
+            socket.and_then(move |socket| proxytunnel::connect(socket, connect_url).and_then(handshake));
+        Box::new(connection)
+    } else {
+        let connection = socket.and_then(handshake);
+        Box::new(connection)
+    }
 }
 
 pub fn authenticate(

core/src/lib.rs

@@ -16,6 +16,7 @@ extern crate byteorder;
 extern crate bytes;
 extern crate crypto;
 extern crate extprim;
+extern crate httparse;
 extern crate hyper;
 extern crate num_bigint;
 extern crate num_integer;
@@ -44,6 +45,7 @@ mod connection;
 pub mod diffie_hellman;
 pub mod keymaster;
 pub mod mercury;
+mod proxytunnel;
 pub mod session;
 pub mod spotify_id;
 pub mod util;

core/src/proxytunnel.rs

@@ -0,0 +1,109 @@
+use futures::{Async, Future, Poll};
+use httparse;
+
+use std::io;
+use std::net::SocketAddr;
+use tokio_io::io::{read, write_all, Read, Window, WriteAll};
+use tokio_io::{AsyncRead, AsyncWrite};
+
+use std::error::Error;
+
+pub struct ProxyTunnel<T> {
+    state: ProxyState<T>,
+}
+
+enum ProxyState<T> {
+    ProxyConnect(WriteAll<T, Vec<u8>>),
+    ProxyResponse(Read<T, Window<Vec<u8>>>),
+}
+
+pub fn connect<T: AsyncRead + AsyncWrite>(connection: T, connect_url: SocketAddr) -> ProxyTunnel<T> {
+    let proxy = proxy_connect(connection, connect_url);
+    ProxyTunnel {
+        state: ProxyState::ProxyConnect(proxy),
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite> Future for ProxyTunnel<T> {
+    type Item = T;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, io::Error> {
+        use self::ProxyState::*;
+        loop {
+            self.state = match self.state {
+                ProxyConnect(ref mut write) => {
+                    let (connection, mut accumulator) = try_ready!(write.poll());
+
+                    let capacity = accumulator.capacity();
+                    accumulator.resize(capacity, 0);
+                    let window = Window::new(accumulator);
+
+                    let read = read(connection, window);
+                    ProxyResponse(read)
+                }
+
+                ProxyResponse(ref mut read_f) => {
+                    let (connection, mut window, bytes_read) = try_ready!(read_f.poll());
+
+                    if bytes_read == 0 {
+                        return Err(io::Error::new(io::ErrorKind::Other, "Early EOF from proxy"));
+                    }
+
+                    let data_end = window.start() + bytes_read;
+
+                    let buf = window.get_ref()[0..data_end].to_vec();
+                    let mut headers = [httparse::EMPTY_HEADER; 16];
+                    let mut response = httparse::Response::new(&mut headers);
+                    let status = match response.parse(&buf) {
+                        Ok(status) => status,
+                        Err(err) => return Err(io::Error::new(io::ErrorKind::Other, err.description())),
+                    };
+
+                    if status.is_complete() {
+                        if let Some(code) = response.code {
+                            if code == 200 {
+                                // Proxy says all is well
+                                return Ok(Async::Ready(connection));
+                            } else {
+                                let reason = response.reason.unwrap_or("no reason");
+                                let msg = format!("Proxy responded with {}: {}", code, reason);
+
+                                return Err(io::Error::new(io::ErrorKind::Other, msg));
+                            }
+                        } else {
+                            return Err(io::Error::new(
+                                io::ErrorKind::Other,
+                                "Malformed response from proxy",
+                            ));
+                        }
+                    } else {
+                        if data_end >= window.end() {
+                            // Allocate some more buffer space
+                            let newsize = data_end + 100;
+                            window.get_mut().resize(newsize, 0);
+                            window.set_end(newsize);
+                        }
+                        // We did not get a full header
+                        window.set_start(data_end);
+                        let read = read(connection, window);
+                        ProxyResponse(read)
+                    }
+                }
+            }
+        }
+    }
+}
+
+fn proxy_connect<T: AsyncWrite>(connection: T, connect_url: SocketAddr) -> WriteAll<T, Vec<u8>> {
+    // TODO: It would be better to use a non-resolved url here. This usually works,
+    //       but it may fail in some environments and it will leak DNS requests.
+    let buffer = format!(
+        "CONNECT {0}:{1} HTTP/1.1\r\n\
+         \r\n",
+        connect_url.ip(),
+        connect_url.port()
+    ).into_bytes();
+
+    write_all(connection, buffer)
+}

core/src/session.rs

@@ -50,12 +50,13 @@ impl Session {
         cache: Option<Cache>,
         handle: Handle,
     ) -> Box<Future<Item = Session, Error = io::Error>> {
-        let access_point = apresolve_or_fallback::<io::Error>(&handle);
+        let access_point = apresolve_or_fallback::<io::Error>(&handle, &config.proxy);
 
         let handle_ = handle.clone();
+        let proxy = config.proxy.clone();
         let connection = access_point.and_then(move |addr| {
             info!("Connecting to AP \"{}\"", addr);
-            connection::connect::<&str>(&addr, &handle_)
+            connection::connect::<&str>(&addr, &handle_, &proxy)
         });
 
         let device_id = config.device_id.clone();

src/main.rs

@@ -126,6 +126,7 @@ fn setup(args: &[String]) -> Setup {
         .optflag("v", "verbose", "Enable verbose output")
         .optopt("u", "username", "Username to sign in with", "USERNAME")
         .optopt("p", "password", "Password", "PASSWORD")
+        .optopt("", "proxy", "HTTP proxy to use when connecting", "PROXY")
         .optflag("", "disable-discovery", "Disable discovery mode")
         .optopt(
             "",
@@ -247,6 +248,7 @@ fn setup(args: &[String]) -> Setup {
         SessionConfig {
             user_agent: version::version_string(),
             device_id: device_id,
+            proxy: matches.opt_str("proxy").or(std::env::var("http_proxy").ok()),
         }
     };