X-Git-Url: https://git.proxmox.com/?a=blobdiff_plain;f=src%2Fclient%2Fhttp_client.rs;h=76ab039158510829626b2508c9f35dae6e5dbc6f;hb=a941bbd0c9dfd2dd804780658f17c4f6e154efe2;hp=ff5a4ec1f80d6a6551cf86aec64fbce73376fd1f;hpb=8a1028e0e7ecdd63261889b49868187e29f24aba;p=proxmox-backup.git

diff --git a/src/client/http_client.rs b/src/client/http_client.rs
index ff5a4ec1..76ab0391 100644
--- a/src/client/http_client.rs
+++ b/src/client/http_client.rs
@@ -1,46 +1,143 @@
 use std::io::Write;
-use std::task::{Context, Poll};
+use std::sync::{Arc, Mutex, RwLock};
+use std::time::Duration;
 
-use chrono::Utc;
-use failure::*;
+use anyhow::{bail, format_err, Error};
 use futures::*;
 use http::Uri;
 use http::header::HeaderValue;
 use http::{Request, Response};
 use hyper::Body;
 use hyper::client::{Client, HttpConnector};
-use openssl::ssl::{SslConnector, SslMethod};
+use openssl::{ssl::{SslConnector, SslMethod}, x509::X509StoreContextRef};
 use serde_json::{json, Value};
 use percent_encoding::percent_encode;
 use xdg::BaseDirectories;
 
-use proxmox::tools::{
-    fs::{file_get_json, file_set_contents},
+use proxmox::{
+    api::error::HttpError,
+    sys::linux::tty,
+    tools::fs::{file_get_json, replace_file, CreateOptions},
 };
 
 use super::pipe_to_stream::PipeToSendStream;
-use crate::tools::async_io::EitherStream;
-use crate::tools::futures::{cancellable, Canceller};
-use crate::tools::{self, tty, BroadcastFuture, DEFAULT_ENCODE_SET};
+use crate::api2::types::{Authid, Userid};
+use crate::tools::{
+    self,
+    BroadcastFuture,
+    DEFAULT_ENCODE_SET,
+    http::HttpsConnector,
+};
+
+/// Timeout used for several HTTP operations that are expected to finish quickly but may block in
+/// certain error conditions. Keep it generous, to avoid false-positive under high load.
+const HTTP_TIMEOUT: Duration = Duration::from_secs(2 * 60);
 
 #[derive(Clone)]
 pub struct AuthInfo {
-    username: String,
-    ticket: String,
-    token: String,
+    pub auth_id: Authid,
+    pub ticket: String,
+    pub token: String,
+}
+
+pub struct HttpClientOptions {
+    prefix: Option<String>,
+    password: Option<String>,
+    fingerprint: Option<String>,
+    interactive: bool,
+    ticket_cache: bool,
+    fingerprint_cache: bool,
+    verify_cert: bool,
+}
+
+impl HttpClientOptions {
+
+    pub fn new_interactive(password: Option<String>, fingerprint: Option<String>) -> Self {
+        Self {
+            password,
+            fingerprint,
+            fingerprint_cache: true,
+            ticket_cache: true,
+            interactive: true,
+            prefix: Some("proxmox-backup".to_string()),
+            ..Self::default()
+        }
+    }
+
+    pub fn new_non_interactive(password: String, fingerprint: Option<String>) -> Self {
+        Self {
+            password: Some(password),
+            fingerprint,
+            ..Self::default()
+        }
+    }
+
+    pub fn prefix(mut self, prefix: Option<String>) -> Self {
+        self.prefix = prefix;
+        self
+    }
+
+    pub fn password(mut self, password: Option<String>) -> Self {
+        self.password = password;
+        self
+    }
+
+    pub fn fingerprint(mut self, fingerprint: Option<String>) -> Self {
+        self.fingerprint = fingerprint;
+        self
+    }
+
+    pub fn interactive(mut self, interactive: bool) -> Self {
+        self.interactive = interactive;
+        self
+    }
+
+    pub fn ticket_cache(mut self, ticket_cache: bool) -> Self {
+        self.ticket_cache = ticket_cache;
+        self
+    }
+
+    pub fn fingerprint_cache(mut self, fingerprint_cache: bool) -> Self {
+        self.fingerprint_cache = fingerprint_cache;
+        self
+    }
+
+    pub fn verify_cert(mut self, verify_cert: bool) -> Self {
+        self.verify_cert = verify_cert;
+        self
+    }
+}
+
+impl Default for HttpClientOptions {
+    fn default() -> Self {
+        Self {
+            prefix: None,
+            password: None,
+            fingerprint: None,
+            interactive: false,
+            ticket_cache: false,
+            fingerprint_cache: false,
+            verify_cert: true,
+        }
+    }
 }
 
 /// HTTP(S) API client
 pub struct HttpClient {
     client: Client<HttpsConnector>,
     server: String,
-    auth: BroadcastFuture<AuthInfo>,
+    port: u16,
+    fingerprint: Arc<Mutex<Option<String>>>,
+    first_auth: Option<BroadcastFuture<()>>,
+    auth: Arc<RwLock<AuthInfo>>,
+    ticket_abort: futures::future::AbortHandle,
+    _options: HttpClientOptions,
 }
 
 /// Delete stored ticket data (logout)
-pub fn delete_ticket_info(server: &str, username: &str) -> Result<(), Error> {
+pub fn delete_ticket_info(prefix: &str, server: &str, username: &Userid) -> Result<(), Error> {
 
-    let base = BaseDirectories::with_prefix("proxmox-backup")?;
+    let base = BaseDirectories::with_prefix(prefix)?;
 
     // usually /run/user/<uid>/...
     let path = base.place_runtime_file("tickets")?;
@@ -50,17 +147,78 @@ pub fn delete_ticket_info(server: &str, username: &str) -> Result<(), Error> {
     let mut data = file_get_json(&path, Some(json!({})))?;
 
     if let Some(map) = data[server].as_object_mut() {
-        map.remove(username);
+        map.remove(username.as_str());
     }
 
-    file_set_contents(path, data.to_string().as_bytes(), Some(mode))?;
+    replace_file(path, data.to_string().as_bytes(), CreateOptions::new().perm(mode))?;
+
+    Ok(())
+}
+
+fn store_fingerprint(prefix: &str, server: &str, fingerprint: &str) -> Result<(), Error> {
+
+    let base = BaseDirectories::with_prefix(prefix)?;
+
+    // usually ~/.config/<prefix>/fingerprints
+    let path = base.place_config_file("fingerprints")?;
+
+    let raw = match std::fs::read_to_string(&path) {
+        Ok(v) => v,
+        Err(err) => {
+            if err.kind() == std::io::ErrorKind::NotFound {
+                String::new()
+            } else {
+                bail!("unable to read fingerprints from {:?} - {}", path, err);
+            }
+        }
+    };
+
+    let mut result = String::new();
+
+    raw.split('\n').for_each(|line| {
+        let items: Vec<String> = line.split_whitespace().map(String::from).collect();
+        if items.len() == 2 {
+            if items[0] == server {
+                // found, add later with new fingerprint
+            } else {
+                result.push_str(line);
+                result.push('\n');
+            }
+        }
+    });
+
+    result.push_str(server);
+    result.push(' ');
+    result.push_str(fingerprint);
+    result.push('\n');
+
+    replace_file(path, result.as_bytes(), CreateOptions::new())?;
 
     Ok(())
 }
 
-fn store_ticket_info(server: &str, username: &str, ticket: &str, token: &str) -> Result<(), Error> {
+fn load_fingerprint(prefix: &str, server: &str) -> Option<String> {
+
+    let base = BaseDirectories::with_prefix(prefix).ok()?;
+
+    // usually ~/.config/<prefix>/fingerprints
+    let path = base.place_config_file("fingerprints").ok()?;
 
-    let base = BaseDirectories::with_prefix("proxmox-backup")?;
+    let raw = std::fs::read_to_string(&path).ok()?;
+
+    for line in raw.split('\n') {
+        let items: Vec<String> = line.split_whitespace().map(String::from).collect();
+        if items.len() == 2 && items[0] == server {
+            return Some(items[1].clone());
+        }
+    }
+
+    None
+}
+
+fn store_ticket_info(prefix: &str, server: &str, username: &str, ticket: &str, token: &str) -> Result<(), Error> {
+
+    let base = BaseDirectories::with_prefix(prefix)?;
 
     // usually /run/user/<uid>/...
     let path = base.place_runtime_file("tickets")?;
@@ -69,7 +227,7 @@ fn store_ticket_info(server: &str, username: &str, ticket: &str, token: &str) ->
 
     let mut data = file_get_json(&path, Some(json!({})))?;
 
-    let now = Utc::now().timestamp();
+    let now = proxmox::tools::time::epoch_i64();
 
     data[server][username] = json!({ "timestamp": now, "ticket": ticket, "token": token});
 
@@ -79,30 +237,30 @@ fn store_ticket_info(server: &str, username: &str, ticket: &str, token: &str) ->
 
     let empty = serde_json::map::Map::new();
     for (server, info) in data.as_object().unwrap_or(&empty) {
-        for (_user, uinfo) in info.as_object().unwrap_or(&empty) {
+        for (user, uinfo) in info.as_object().unwrap_or(&empty) {
             if let Some(timestamp) = uinfo["timestamp"].as_i64() {
                 let age = now - timestamp;
                 if age < ticket_lifetime {
-                    new_data[server][username] = uinfo.clone();
+                    new_data[server][user] = uinfo.clone();
                 }
             }
         }
     }
 
-    file_set_contents(path, new_data.to_string().as_bytes(), Some(mode))?;
+    replace_file(path, new_data.to_string().as_bytes(), CreateOptions::new().perm(mode))?;
 
     Ok(())
 }
 
-fn load_ticket_info(server: &str, username: &str) -> Option<(String, String)> {
-    let base = BaseDirectories::with_prefix("proxmox-backup").ok()?;
+fn load_ticket_info(prefix: &str, server: &str, userid: &Userid) -> Option<(String, String)> {
+    let base = BaseDirectories::with_prefix(prefix).ok()?;
 
     // usually /run/user/<uid>/...
     let path = base.place_runtime_file("tickets").ok()?;
     let data = file_get_json(&path, None).ok()?;
-    let now = Utc::now().timestamp();
+    let now = proxmox::tools::time::epoch_i64();
     let ticket_lifetime = tools::ticket::TICKET_LIFETIME - 60;
-    let uinfo = data[server][username].as_object()?;
+    let uinfo = data[server][userid.as_str()].as_object()?;
     let timestamp = uinfo["timestamp"].as_i64()?;
     let age = now - timestamp;
 
@@ -116,70 +274,246 @@ fn load_ticket_info(server: &str, username: &str) -> Option<(String, String)> {
 }
 
 impl HttpClient {
+    pub fn new(
+        server: &str,
+        port: u16,
+        auth_id: &Authid,
+        mut options: HttpClientOptions,
+    ) -> Result<Self, Error> {
 
-    pub fn new(server: &str, username: &str, password: Option<String>) -> Result<Self, Error> {
-        let client = Self::build_client();
+        let verified_fingerprint = Arc::new(Mutex::new(None));
+
+        let mut fingerprint = options.fingerprint.take();
+
+        if fingerprint.is_some() {
+            // do not store fingerprints passed via options in cache
+            options.fingerprint_cache = false;
+        } else if options.fingerprint_cache && options.prefix.is_some() {
+            fingerprint = load_fingerprint(options.prefix.as_ref().unwrap(), server);
+        }
+
+        let mut ssl_connector_builder = SslConnector::builder(SslMethod::tls()).unwrap();
+
+        if options.verify_cert {
+            let server = server.to_string();
+            let verified_fingerprint = verified_fingerprint.clone();
+            let interactive = options.interactive;
+            let fingerprint_cache = options.fingerprint_cache;
+            let prefix = options.prefix.clone();
+            ssl_connector_builder.set_verify_callback(openssl::ssl::SslVerifyMode::PEER, move |valid, ctx| {
+                let (valid, fingerprint) = Self::verify_callback(valid, ctx, fingerprint.clone(), interactive);
+                if valid {
+                    if let Some(fingerprint) = fingerprint {
+                        if fingerprint_cache && prefix.is_some() {
+                            if let Err(err) = store_fingerprint(
+                                prefix.as_ref().unwrap(), &server, &fingerprint) {
+                                eprintln!("{}", err);
+                            }
+                        }
+                        *verified_fingerprint.lock().unwrap() = Some(fingerprint);
+                    }
+                }
+                valid
+            });
+        } else {
+            ssl_connector_builder.set_verify(openssl::ssl::SslVerifyMode::NONE);
+        }
+
+        let mut httpc = HttpConnector::new();
+        httpc.set_nodelay(true); // important for h2 download performance!
+        httpc.enforce_http(false); // we want https...
+
+        httpc.set_connect_timeout(Some(std::time::Duration::new(10, 0)));
+        let https = HttpsConnector::with_connector(httpc, ssl_connector_builder.build());
+
+        let client = Client::builder()
+        //.http2_initial_stream_window_size( (1 << 31) - 2)
+        //.http2_initial_connection_window_size( (1 << 31) - 2)
+            .build::<_, Body>(https);
+
+        let password = options.password.take();
+        let use_ticket_cache = options.ticket_cache && options.prefix.is_some();
 
         let password = if let Some(password) = password {
             password
-        } else if let Some((ticket, _token)) = load_ticket_info(server, username) {
-            ticket
         } else {
-            Self::get_password(&username)?
+            let userid = if auth_id.is_token() {
+                bail!("API token secret must be provided!");
+            } else {
+                auth_id.user()
+            };
+            let mut ticket_info = None;
+            if use_ticket_cache {
+                ticket_info = load_ticket_info(options.prefix.as_ref().unwrap(), server, userid);
+            }
+            if let Some((ticket, _token)) = ticket_info {
+                ticket
+            } else {
+                Self::get_password(userid, options.interactive)?
+            }
+        };
+
+        let auth = Arc::new(RwLock::new(AuthInfo {
+            auth_id: auth_id.clone(),
+            ticket: password.clone(),
+            token: "".to_string(),
+        }));
+
+        let server2 = server.to_string();
+        let client2 = client.clone();
+        let auth2 = auth.clone();
+        let prefix2 = options.prefix.clone();
+
+        let renewal_future = async move {
+            loop {
+                tokio::time::sleep(Duration::new(60*15,  0)).await; // 15 minutes
+                let (auth_id, ticket) = {
+                    let authinfo = auth2.read().unwrap().clone();
+                    (authinfo.auth_id, authinfo.ticket)
+                };
+                match Self::credentials(client2.clone(), server2.clone(), port, auth_id.user().clone(), ticket).await {
+                    Ok(auth) => {
+                        if use_ticket_cache && prefix2.is_some() {
+                            let _ = store_ticket_info(prefix2.as_ref().unwrap(), &server2, &auth.auth_id.to_string(), &auth.ticket, &auth.token);
+                        }
+                        *auth2.write().unwrap() = auth;
+                    },
+                    Err(err) => {
+                        eprintln!("re-authentication failed: {}", err);
+                        return;
+                    }
+                }
+            }
         };
 
-        let login_future = Self::credentials(client.clone(), server.to_owned(), username.to_owned(), password);
+        let (renewal_future, ticket_abort) = futures::future::abortable(renewal_future);
+
+        let login_future = Self::credentials(
+            client.clone(),
+            server.to_owned(),
+            port,
+            auth_id.user().clone(),
+            password,
+        ).map_ok({
+            let server = server.to_string();
+            let prefix = options.prefix.clone();
+            let authinfo = auth.clone();
+
+            move |auth| {
+                if use_ticket_cache && prefix.is_some() {
+                    let _ = store_ticket_info(prefix.as_ref().unwrap(), &server, &auth.auth_id.to_string(), &auth.ticket, &auth.token);
+                }
+                *authinfo.write().unwrap() = auth;
+                tokio::spawn(renewal_future);
+            }
+        });
+
+        let first_auth = if auth_id.is_token() {
+            // TODO check access here?
+            None
+        } else {
+            Some(BroadcastFuture::new(Box::new(login_future)))
+        };
 
         Ok(Self {
             client,
             server: String::from(server),
-            auth: BroadcastFuture::new(Box::new(login_future)),
+            port,
+            fingerprint: verified_fingerprint,
+            auth,
+            ticket_abort,
+            first_auth,
+            _options: options,
         })
     }
 
     /// Login
     ///
-    /// Login is done on demand, so this is onyl required if you need
+    /// Login is done on demand, so this is only required if you need
     /// access to authentication data in 'AuthInfo'.
+    ///
+    /// Note: tickets a periodially re-newed, so one can use this
+    /// to query changed ticket.
     pub async fn login(&self) -> Result<AuthInfo, Error> {
-        self.auth.listen().await
+        if let Some(future) = &self.first_auth {
+            future.listen().await?;
+        }
+
+        let authinfo = self.auth.read().unwrap();
+        Ok(authinfo.clone())
     }
 
-    fn get_password(_username: &str) -> Result<String, Error> {
-        use std::env::VarError::*;
-        match std::env::var("PBS_PASSWORD") {
-            Ok(p) => return Ok(p),
-            Err(NotUnicode(_)) => bail!("PBS_PASSWORD contains bad characters"),
-            Err(NotPresent) => {
-                // Try another method
-            }
-        }
+    /// Returns the optional fingerprint passed to the new() constructor.
+    pub fn fingerprint(&self) -> Option<String> {
+        (*self.fingerprint.lock().unwrap()).clone()
+    }
 
+    fn get_password(username: &Userid, interactive: bool) -> Result<String, Error> {
         // If we're on a TTY, query the user for a password
-        if tty::stdin_isatty() {
-            return Ok(String::from_utf8(tty::read_password("Password: ")?)?);
+        if interactive && tty::stdin_isatty() {
+            let msg = format!("Password for \"{}\": ", username);
+            return Ok(String::from_utf8(tty::read_password(&msg)?)?);
         }
 
         bail!("no password input mechanism available");
     }
 
-    fn build_client() -> Client<HttpsConnector> {
+    fn verify_callback(
+        valid: bool, ctx:
+        &mut X509StoreContextRef,
+        expected_fingerprint: Option<String>,
+        interactive: bool,
+    ) -> (bool, Option<String>) {
+        if valid { return (true, None); }
 
-        let mut ssl_connector_builder = SslConnector::builder(SslMethod::tls()).unwrap();
+        let cert = match ctx.current_cert() {
+            Some(cert) => cert,
+            None => return (false, None),
+        };
 
-        ssl_connector_builder.set_verify(openssl::ssl::SslVerifyMode::NONE); // fixme!
+        let depth = ctx.error_depth();
+        if depth != 0 { return (false, None); }
 
-        let mut httpc = hyper::client::HttpConnector::new();
-        httpc.set_nodelay(true); // important for h2 download performance!
-        httpc.set_recv_buffer_size(Some(1024*1024)); //important for h2 download performance!
-        httpc.enforce_http(false); // we want https...
+        let fp = match cert.digest(openssl::hash::MessageDigest::sha256()) {
+            Ok(fp) => fp,
+            Err(_) => return (false, None), // should not happen
+        };
+        let fp_string = proxmox::tools::digest_to_hex(&fp);
+        let fp_string = fp_string.as_bytes().chunks(2).map(|v| std::str::from_utf8(v).unwrap())
+            .collect::<Vec<&str>>().join(":");
 
-        let https = HttpsConnector::with_connector(httpc, ssl_connector_builder.build());
+        if let Some(expected_fingerprint) = expected_fingerprint {
+            if expected_fingerprint.to_lowercase() == fp_string {
+                return (true, Some(fp_string));
+            } else {
+                return (false, None);
+            }
+        }
 
-        Client::builder()
-        //.http2_initial_stream_window_size( (1 << 31) - 2)
-        //.http2_initial_connection_window_size( (1 << 31) - 2)
-            .build::<_, Body>(https)
+        // If we're on a TTY, query the user
+        if interactive && tty::stdin_isatty() {
+            println!("fingerprint: {}", fp_string);
+            loop {
+                print!("Are you sure you want to continue connecting? (y/n): ");
+                let _ = std::io::stdout().flush();
+                use std::io::{BufRead, BufReader};
+                let mut line = String::new();
+                match BufReader::new(std::io::stdin()).read_line(&mut line) {
+                    Ok(_) => {
+                        let trimmed = line.trim();
+                        if trimmed == "y" || trimmed == "Y" {
+                            return (true, Some(fp_string));
+                        } else if trimmed == "n" || trimmed == "N" {
+                            return (false, None);
+                        } else {
+                            continue;
+                        }
+                    }
+                    Err(_) => return (false, None),
+                }
+            }
+        }
+        (false, None)
     }
 
     pub async fn request(&self, mut req: Request<Body>) -> Result<Value, Error> {
@@ -187,10 +521,14 @@ impl HttpClient {
         let client = self.client.clone();
 
         let auth =  self.login().await?;
-
-        let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
-        req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
-        req.headers_mut().insert("CSRFPreventionToken", HeaderValue::from_str(&auth.token).unwrap());
+        if auth.auth_id.is_token() {
+            let enc_api_token = format!("PBSAPIToken {}:{}", auth.auth_id, percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
+            req.headers_mut().insert("Authorization", HeaderValue::from_str(&enc_api_token).unwrap());
+        } else {
+            let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
+            req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
+            req.headers_mut().insert("CSRFPreventionToken", HeaderValue::from_str(&auth.token).unwrap());
+        }
 
         Self::api_request(client, req).await
     }
@@ -200,7 +538,7 @@ impl HttpClient {
         path: &str,
         data: Option<Value>,
     ) -> Result<Value, Error> {
-        let req = Self::request_builder(&self.server, "GET", path, data).unwrap();
+        let req = Self::request_builder(&self.server, self.port, "GET", path, data)?;
         self.request(req).await
     }
 
@@ -209,7 +547,7 @@ impl HttpClient {
         path: &str,
         data: Option<Value>,
     ) -> Result<Value, Error> {
-        let req = Self::request_builder(&self.server, "DELETE", path, data).unwrap();
+        let req = Self::request_builder(&self.server, self.port, "DELETE", path, data)?;
         self.request(req).await
     }
 
@@ -218,7 +556,16 @@ impl HttpClient {
         path: &str,
         data: Option<Value>,
     ) -> Result<Value, Error> {
-        let req = Self::request_builder(&self.server, "POST", path, data).unwrap();
+        let req = Self::request_builder(&self.server, self.port, "POST", path, data)?;
+        self.request(req).await
+    }
+
+    pub async fn put(
+        &mut self,
+        path: &str,
+        data: Option<Value>,
+    ) -> Result<Value, Error> {
+        let req = Self::request_builder(&self.server, self.port, "PUT", path, data)?;
         self.request(req).await
     }
 
@@ -226,8 +573,8 @@ impl HttpClient {
         &mut self,
         path: &str,
         output: &mut (dyn Write + Send),
-    ) ->  Result<(), Error> {
-        let mut req = Self::request_builder(&self.server, "GET", path, None).unwrap();
+    ) -> Result<(), Error> {
+        let mut req = Self::request_builder(&self.server, self.port, "GET", path, None)?;
 
         let client = self.client.clone();
 
@@ -236,7 +583,12 @@ impl HttpClient {
         let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
         req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
 
-        let resp = client.request(req).await?;
+        let resp = tokio::time::timeout(
+            HTTP_TIMEOUT,
+            client.request(req)
+        )
+            .await
+            .map_err(|_| format_err!("http download request timed out"))??;
         let status = resp.status();
         if !status.is_success() {
             HttpClient::api_response(resp)
@@ -263,7 +615,7 @@ impl HttpClient {
     ) -> Result<Value, Error> {
 
         let path = path.trim_matches('/');
-        let mut url = format!("https://{}:8007/{}", &self.server, path);
+        let mut url = format!("https://{}:{}/{}", &self.server, self.port, path);
 
         if let Some(data) = data {
             let query = tools::json_object_to_query(data).unwrap();
@@ -287,29 +639,36 @@ impl HttpClient {
         &self,
         mut req: Request<Body>,
         protocol_name: String,
-    ) -> Result<(H2Client, Canceller), Error> {
+    ) -> Result<(H2Client, futures::future::AbortHandle), Error> {
 
-        let auth = self.login().await?;
         let client = self.client.clone();
+        let auth =  self.login().await?;
+
+        if auth.auth_id.is_token() {
+            let enc_api_token = format!("PBSAPIToken {}:{}", auth.auth_id, percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
+            req.headers_mut().insert("Authorization", HeaderValue::from_str(&enc_api_token).unwrap());
+        } else {
+            let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
+            req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
+            req.headers_mut().insert("CSRFPreventionToken", HeaderValue::from_str(&auth.token).unwrap());
+        }
 
-        let enc_ticket = format!("PBSAuthCookie={}", percent_encode(auth.ticket.as_bytes(), DEFAULT_ENCODE_SET));
-        req.headers_mut().insert("Cookie", HeaderValue::from_str(&enc_ticket).unwrap());
         req.headers_mut().insert("UPGRADE", HeaderValue::from_str(&protocol_name).unwrap());
 
-        let resp = client.request(req).await?;
+        let resp = tokio::time::timeout(
+            HTTP_TIMEOUT,
+            client.request(req)
+        )
+            .await
+            .map_err(|_| format_err!("http upgrade request timed out"))??;
         let status = resp.status();
 
         if status != http::StatusCode::SWITCHING_PROTOCOLS {
-            Self::api_response(resp)
-                .map(|_| Err(format_err!("unknown error")))
-                .await?;
-            unreachable!();
+            Self::api_response(resp).await?;
+            bail!("unknown error");
         }
 
-        let upgraded = resp
-            .into_body()
-            .on_upgrade()
-            .await?;
+        let upgraded = hyper::upgrade::on(resp).await?;
 
         let max_window_size = (1 << 31) - 2;
 
@@ -321,9 +680,9 @@ impl HttpClient {
             .await?;
 
         let connection = connection
-            .map_err(|_| panic!("HTTP/2.0 connection failed"));
+            .map_err(|_| eprintln!("HTTP/2.0 connection failed"));
 
-        let (connection, canceller) = cancellable(connection)?;
+        let (connection, abort) = futures::future::abortable(connection);
         // A cancellable future returns an Option which is None when cancelled and
         // Some when it finished instead, since we don't care about the return type we
         // need to map it away:
@@ -334,26 +693,25 @@ impl HttpClient {
 
         // Wait until the `SendRequest` handle has available capacity.
         let c = h2.ready().await?;
-        Ok((H2Client::new(c), canceller))
+        Ok((H2Client::new(c), abort))
     }
 
     async fn credentials(
         client: Client<HttpsConnector>,
         server: String,
-        username: String,
+        port: u16,
+        username: Userid,
         password: String,
     ) -> Result<AuthInfo, Error> {
         let data = json!({ "username": username, "password": password });
-        let req = Self::request_builder(&server, "POST", "/api2/json/access/ticket", Some(data)).unwrap();
+        let req = Self::request_builder(&server, port, "POST", "/api2/json/access/ticket", Some(data))?;
         let cred = Self::api_request(client, req).await?;
         let auth = AuthInfo {
-            username: cred["data"]["username"].as_str().unwrap().to_owned(),
+            auth_id: cred["data"]["username"].as_str().unwrap().parse()?,
             ticket: cred["data"]["ticket"].as_str().unwrap().to_owned(),
             token: cred["data"]["CSRFPreventionToken"].as_str().unwrap().to_owned(),
         };
 
-        let _ = store_ticket_info(&server, &auth.username, &auth.ticket, &auth.token);
-
         Ok(auth)
     }
 
@@ -370,7 +728,7 @@ impl HttpClient {
                 Ok(value)
             }
         } else {
-            bail!("HTTP Error {}: {}", status, text);
+            Err(Error::from(HttpError::new(status, text)))
         }
     }
 
@@ -379,10 +737,14 @@ impl HttpClient {
         req: Request<Body>
     ) -> Result<Value, Error> {
 
-        client.request(req)
-            .map_err(Error::from)
-            .and_then(Self::api_response)
-            .await
+        Self::api_response(
+            tokio::time::timeout(
+                HTTP_TIMEOUT,
+                client.request(req)
+            )
+                .await
+                .map_err(|_| format_err!("http request timed out"))??
+        ).await
     }
 
     // Read-only access to server property
@@ -390,9 +752,13 @@ impl HttpClient {
         &self.server
     }
 
-    pub fn request_builder(server: &str, method: &str, path: &str, data: Option<Value>) -> Result<Request<Body>, Error> {
+    pub fn port(&self) -> u16 {
+        self.port
+    }
+
+    pub fn request_builder(server: &str, port: u16, method: &str, path: &str, data: Option<Value>) -> Result<Request<Body>, Error> {
         let path = path.trim_matches('/');
-        let url: Uri = format!("https://{}:8007/{}", server, path).parse()?;
+        let url: Uri = format!("https://{}:{}/{}", server, port, path).parse()?;
 
         if let Some(data) = data {
             if method == "POST" {
@@ -405,7 +771,7 @@ impl HttpClient {
                 return Ok(request);
             } else {
                 let query = tools::json_object_to_query(data)?;
-                let url: Uri = format!("https://{}:8007/{}?{}", server, path, query).parse()?;
+                let url: Uri = format!("https://{}:{}/{}?{}", server, port, path, query).parse()?;
                 let request = Request::builder()
                     .method(method)
                     .uri(url)
@@ -427,6 +793,12 @@ impl HttpClient {
     }
 }
 
+impl Drop for HttpClient {
+    fn drop(&mut self) {
+        self.ticket_abort.abort();
+    }
+}
+
 
 #[derive(Clone)]
 pub struct H2Client {
@@ -471,7 +843,7 @@ impl H2Client {
         path: &str,
         param: Option<Value>,
         mut output: W,
-    ) -> Result<W, Error> {
+    ) -> Result<(), Error> {
         let request = Self::request_builder("localhost", "GET", path, param, None).unwrap();
 
         let response_future = self.send_request(request, None).await?;
@@ -491,7 +863,7 @@ impl H2Client {
             output.write_all(&chunk)?;
         }
 
-        Ok(output)
+        Ok(())
     }
 
     pub async fn upload(
@@ -583,7 +955,7 @@ impl H2Client {
                 bail!("got result without data property");
             }
         } else {
-            bail!("HTTP Error {}: {}", status, text);
+            Err(Error::from(HttpError::new(status, text)))
         }
     }
 
@@ -624,61 +996,3 @@ impl H2Client {
         }
     }
 }
-
-#[derive(Clone)]
-pub struct HttpsConnector {
-    http: HttpConnector,
-    ssl_connector: std::sync::Arc<SslConnector>,
-}
-
-impl HttpsConnector {
-    pub fn with_connector(mut http: HttpConnector, ssl_connector: SslConnector) -> Self {
-        http.enforce_http(false);
-
-        Self {
-            http,
-            ssl_connector: std::sync::Arc::new(ssl_connector),
-        }
-    }
-}
-
-type MaybeTlsStream = EitherStream<
-    tokio::net::TcpStream,
-    tokio_openssl::SslStream<tokio::net::TcpStream>,
->;
-
-impl hyper::service::Service<Uri> for HttpsConnector {
-    type Response = MaybeTlsStream;
-    type Error = Error;
-    type Future = std::pin::Pin<Box<
-        dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static
-    >>;
-
-    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        // This connector is always ready, but others might not be.
-        Poll::Ready(Ok(()))
-    }
-
-    fn call(&mut self, dst: Uri) -> Self::Future {
-        let mut this = self.clone();
-        async move {
-            let is_https = dst
-                .scheme()
-                .ok_or_else(|| format_err!("missing URL scheme"))?
-                == "https";
-            let host = dst
-                .host()
-                .ok_or_else(|| format_err!("missing hostname in destination url?"))?
-                .to_string();
-
-            let config = this.ssl_connector.configure();
-            let conn = this.http.call(dst).await?;
-            if is_https {
-                let conn = tokio_openssl::connect(config?, &host, conn).await?;
-                Ok(MaybeTlsStream::Right(conn))
-            } else {
-                Ok(MaybeTlsStream::Left(conn))
-            }
-        }.boxed()
-    }
-}