import proxmox-client crate

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>

Cargo.toml

@@ -5,6 +5,7 @@ members = [
     "proxmox-async",
     "proxmox-auth-api",
     "proxmox-borrow",
+    "proxmox-client",
     "proxmox-compression",
     "proxmox-http",
     "proxmox-http-error",
@@ -93,6 +94,7 @@ proxmox-http-error = { version = "0.1.0", path = "proxmox-http-error" }
 proxmox-human-byte = { version = "0.1.0", path = "proxmox-human-byte" }
 proxmox-io = { version = "1.0.0", path = "proxmox-io" }
 proxmox-lang = { version = "1.1", path = "proxmox-lang" }
+proxmox-login = { version = "0.1.0", path = "proxmox-login" }
 proxmox-rest-server = { version = "0.4.0", path = "proxmox-rest-server" }
 proxmox-router = { version = "2.0.0", path = "proxmox-router" }
 proxmox-schema = { version = "2.0.0", path = "proxmox-schema" }

proxmox-client/Cargo.toml

@@ -0,0 +1,38 @@
+[package]
+name = "proxmox-client"
+version = "0.1.0"
+description = "Base client for proxmox APIs for handling login and ticket renewal"
+authors.workspace = true
+license.workspace = true
+edition.workspace = true
+exclude.workspace = true
+repository.workspace = true
+
+[dependencies]
+anyhow.workspace = true
+base64.workspace = true
+http.workspace = true
+once_cell.workspace = true
+percent-encoding.workspace = true
+regex.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+serde_plain.workspace = true
+
+# wasm-incompatible dependencies must stay optional
+log = { workspace = true, optional = true }
+openssl = { workspace = true, optional = true }
+
+proxmox-login = { workspace = true, features = [ "http" ] }
+webauthn-rs = { workspace = true, optional = true }
+
+proxmox-http = { workspace = true, optional = true, features = [ "client" ] }
+hyper = { workspace = true, optional = true }
+
+proxmox-section-config.workspace = true
+proxmox-schema = { workspace = true, features = [ "api-macro" ] }
+
+[features]
+default = []
+hyper-client = [ "dep:openssl", "dep:hyper", "dep:proxmox-http", "dep:log" ]
+webauthn = [ "dep:webauthn-rs", "proxmox-login/webauthn" ]

proxmox-client/debian/changelog

@@ -0,0 +1,5 @@
+rust-proxmox-client (0.1.0-1) bookworm; urgency=medium
+
+  * initial release
+
+ -- Proxmox Support Team <support@proxmox.com>  Tue, 01 Aug 2023 15:46:54 +0200

proxmox-client/debian/control

@@ -0,0 +1,98 @@
+Source: rust-proxmox-client
+Section: rust
+Priority: optional
+Build-Depends: debhelper (>= 12),
+ dh-cargo (>= 25),
+ cargo:native <!nocheck>,
+ rustc:native <!nocheck>,
+ libstd-rust-dev <!nocheck>,
+ librust-anyhow-1+default-dev <!nocheck>,
+ librust-base64-0.13+default-dev <!nocheck>,
+ librust-http-0.2+default-dev <!nocheck>,
+ librust-once-cell-1+default-dev (>= 1.3.1-~~) <!nocheck>,
+ librust-percent-encoding-2+default-dev (>= 2.1-~~) <!nocheck>,
+ librust-proxmox-login-0.1+default-dev <!nocheck>,
+ librust-proxmox-login-0.1+http-dev <!nocheck>,
+ librust-proxmox-schema-2+api-macro-dev <!nocheck>,
+ librust-proxmox-schema-2+default-dev <!nocheck>,
+ librust-proxmox-section-config-2+default-dev <!nocheck>,
+ librust-regex-1+default-dev (>= 1.5-~~) <!nocheck>,
+ librust-serde-1+default-dev <!nocheck>,
+ librust-serde-json-1+default-dev <!nocheck>,
+ librust-serde-plain-1+default-dev <!nocheck>
+Maintainer: Proxmox Support Team <support@proxmox.com>
+Standards-Version: 4.6.1
+Vcs-Git: git://git.proxmox.com/git/proxmox.git
+Vcs-Browser: https://git.proxmox.com/?p=proxmox.git
+X-Cargo-Crate: proxmox-client
+Rules-Requires-Root: no
+
+Package: librust-proxmox-client-dev
+Architecture: any
+Multi-Arch: same
+Depends:
+ ${misc:Depends},
+ librust-anyhow-1+default-dev,
+ librust-base64-0.13+default-dev,
+ librust-http-0.2+default-dev,
+ librust-once-cell-1+default-dev (>= 1.3.1-~~),
+ librust-percent-encoding-2+default-dev (>= 2.1-~~),
+ librust-proxmox-login-0.1+default-dev,
+ librust-proxmox-login-0.1+http-dev,
+ librust-proxmox-schema-2+api-macro-dev,
+ librust-proxmox-schema-2+default-dev,
+ librust-proxmox-section-config-2+default-dev,
+ librust-regex-1+default-dev (>= 1.5-~~),
+ librust-serde-1+default-dev,
+ librust-serde-json-1+default-dev,
+ librust-serde-plain-1+default-dev
+Suggests:
+ librust-proxmox-client+hyper-client-dev (= ${binary:Version}),
+ librust-proxmox-client+webauthn-dev (= ${binary:Version})
+Provides:
+ librust-proxmox-client+default-dev (= ${binary:Version}),
+ librust-proxmox-client-0-dev (= ${binary:Version}),
+ librust-proxmox-client-0+default-dev (= ${binary:Version}),
+ librust-proxmox-client-0.1-dev (= ${binary:Version}),
+ librust-proxmox-client-0.1+default-dev (= ${binary:Version}),
+ librust-proxmox-client-0.1.0-dev (= ${binary:Version}),
+ librust-proxmox-client-0.1.0+default-dev (= ${binary:Version})
+Description: Base client for proxmox APIs for handling login and ticket renewal - Rust source code
+ This package contains the source for the Rust proxmox-client crate, packaged by
+ debcargo for use with cargo and dh-cargo.
+
+Package: librust-proxmox-client+hyper-client-dev
+Architecture: any
+Multi-Arch: same
+Depends:
+ ${misc:Depends},
+ librust-proxmox-client-dev (= ${binary:Version}),
+ librust-hyper-0.14+default-dev (>= 0.14.5-~~),
+ librust-log-0.4+default-dev (>= 0.4.17-~~),
+ librust-openssl-0.10+default-dev,
+ librust-proxmox-http-0.9+client-dev,
+ librust-proxmox-http-0.9+default-dev
+Provides:
+ librust-proxmox-client-0+hyper-client-dev (= ${binary:Version}),
+ librust-proxmox-client-0.1+hyper-client-dev (= ${binary:Version}),
+ librust-proxmox-client-0.1.0+hyper-client-dev (= ${binary:Version})
+Description: Base client for proxmox APIs for handling login and ticket renewal - feature "hyper-client"
+ This metapackage enables feature "hyper-client" for the Rust proxmox-client
+ crate, by pulling in any additional dependencies needed by that feature.
+
+Package: librust-proxmox-client+webauthn-dev
+Architecture: any
+Multi-Arch: same
+Depends:
+ ${misc:Depends},
+ librust-proxmox-client-dev (= ${binary:Version}),
+ librust-proxmox-login-0.1+http-dev,
+ librust-proxmox-login-0.1+webauthn-dev,
+ librust-webauthn-rs-0.3+default-dev
+Provides:
+ librust-proxmox-client-0+webauthn-dev (= ${binary:Version}),
+ librust-proxmox-client-0.1+webauthn-dev (= ${binary:Version}),
+ librust-proxmox-client-0.1.0+webauthn-dev (= ${binary:Version})
+Description: Base client for proxmox APIs for handling login and ticket renewal - feature "webauthn"
+ This metapackage enables feature "webauthn" for the Rust proxmox-client crate,
+ by pulling in any additional dependencies needed by that feature.

proxmox-client/debian/copyright

@@ -0,0 +1,18 @@
+Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/
+
+Files:
+ *
+Copyright: 2019 - 2023 Proxmox Server Solutions GmbH <support@proxmox.com>
+License: AGPL-3.0-or-later
+ This program is free software: you can redistribute it and/or modify it under
+ the terms of the GNU Affero General Public License as published by the Free
+ Software Foundation, either version 3 of the License, or (at your option) any
+ later version.
+ .
+ This program is distributed in the hope that it will be useful, but WITHOUT
+ ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
+ details.
+ .
+ You should have received a copy of the GNU Affero General Public License along
+ with this program. If not, see <https://www.gnu.org/licenses/>.

proxmox-client/debian/debcargo.toml

@@ -0,0 +1,7 @@
+overlay = "."
+crate_src_path = ".."
+maintainer = "Proxmox Support Team <support@proxmox.com>"
+
+[source]
+vcs_git = "git://git.proxmox.com/git/proxmox.git"
+vcs_browser = "https://git.proxmox.com/?p=proxmox.git"

proxmox-client/src/auth.rs

@@ -0,0 +1,59 @@
+use crate::Authentication;
+
+/// How the client is logged in to the remote.
+pub enum AuthenticationKind {
+    /// With an API Ticket.
+    Ticket(Authentication),
+
+    /// With a token.
+    Token(Token),
+}
+
+impl AuthenticationKind {
+    pub fn set_auth_headers(&self, request: http::request::Builder) -> http::request::Builder {
+        match self {
+            AuthenticationKind::Ticket(auth) => auth.set_auth_headers(request),
+            AuthenticationKind::Token(auth) => auth.set_auth_headers(request),
+        }
+    }
+
+    pub fn userid(&self) -> &str {
+        match self {
+            AuthenticationKind::Ticket(auth) => &auth.userid,
+            AuthenticationKind::Token(auth) => &auth.userid,
+        }
+    }
+}
+
+impl From<Authentication> for AuthenticationKind {
+    fn from(auth: Authentication) -> Self {
+        Self::Ticket(auth)
+    }
+}
+
+impl From<Token> for AuthenticationKind {
+    fn from(auth: Token) -> Self {
+        Self::Token(auth)
+    }
+}
+
+/// Data used to log in with a token.
+pub struct Token {
+    /// The userid.
+    pub userid: String,
+
+    /// The api token name (usually the product abbreviation).
+    pub prefix: String,
+
+    /// The api token's value.
+    pub value: String,
+}
+
+impl Token {
+    pub fn set_auth_headers(&self, request: http::request::Builder) -> http::request::Builder {
+        request.header(
+            http::header::AUTHORIZATION,
+            format!("{}={}={}", self.prefix, self.userid, self.value),
+        )
+    }
+}

proxmox-client/src/client.rs

@@ -0,0 +1,887 @@
+use std::collections::HashMap;
+use std::fmt;
+use std::future::Future;
+use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering};
+use std::sync::Arc;
+use std::sync::Mutex as StdMutex;
+
+use http::request::Request;
+use http::response::Response;
+use http::uri::PathAndQuery;
+use http::{StatusCode, Uri};
+use serde_json::Value;
+
+use proxmox_login::{Login, TicketResult};
+
+use crate::auth::AuthenticationKind;
+use crate::{Authentication, Environment, Error, Token};
+
+/// HTTP client backend trait.
+///
+/// An async [`Client`] requires some kind of async HTTP client implementation.
+pub trait HttpClient: Send + Sync {
+    type Error: Error;
+    type Request: Future<Output = Result<Response<Vec<u8>>, Self::Error>> + Send;
+
+    fn request(&self, request: Request<Vec<u8>>) -> Self::Request;
+}
+
+/// In a cluster we may be able to connect to a different node if one connection fails.
+struct ApiUrls {
+    /// This is the list of cluster node URls.
+    urls: Vec<Uri>,
+
+    /// This is the current "good" URL. If we fail to connect here, we'll walk around the `urls`
+    /// vec once before failing completely.
+    /// Once a "good" URL is reached, we update this.
+    current: AtomicUsize,
+
+    /// Since another thread might be doing the same thing simultaneously, let's use this to keep
+    /// track of when some thread has updated `current`. If we see a `generation` bump while
+    /// probing URLs, we'll retry the new `current`.
+    generation: AtomicU32,
+}
+
+impl ApiUrls {
+    fn new(uri: Uri) -> Self {
+        Self {
+            urls: vec![uri],
+            current: AtomicUsize::new(0),
+            generation: AtomicU32::new(0),
+        }
+    }
+
+    fn index(&self) -> usize {
+        self.current.load(Ordering::Relaxed)
+    }
+
+    fn generation(&self) -> u32 {
+        self.generation.load(Ordering::Acquire)
+    }
+}
+
+/// Proxmox VE high level API client.
+pub struct Client<C, E: Environment> {
+    env: E,
+    api_urls: ApiUrls,
+    auth: StdMutex<Option<Arc<AuthenticationKind>>>,
+    client: C,
+    pve_compat: bool,
+}
+
+impl<C, E> Client<C, E>
+where
+    E: Environment,
+{
+    /// Get a reference to the current authentication information.
+    pub fn authentication(&self) -> Option<Arc<AuthenticationKind>> {
+        self.auth.lock().unwrap().clone()
+    }
+
+    pub fn use_api_token(&self, token: Token) {
+        *self.auth.lock().unwrap() = Some(Arc::new(token.into()));
+    }
+}
+
+fn to_request<E: Error>(request: proxmox_login::Request) -> Result<http::Request<Vec<u8>>, E> {
+    http::Request::builder()
+        .method(http::Method::POST)
+        .uri(request.url)
+        .header(http::header::CONTENT_TYPE, request.content_type)
+        .header(
+            http::header::CONTENT_LENGTH,
+            request.content_length.to_string(),
+        )
+        .body(request.body.into_bytes())
+        .map_err(E::internal)
+}
+
+impl<C, E: Environment> Client<C, E> {
+    /// Enable Proxmox VE login API compatibility. This is required to support TFA authentication
+    /// on Proxmox VE APIs which require the `new-format` option.
+    pub fn set_pve_compatibility(&mut self, compatibility: bool) {
+        self.pve_compat = compatibility;
+    }
+}
+
+impl<C, E> Client<C, E>
+where
+    E: Environment,
+    C: HttpClient,
+    E::Error: From<C::Error>,
+{
+    /// Instantiate a client for an API with a given environment and HTTP client instance.
+    pub fn with_client(api_url: Uri, environment: E, client: C) -> Self {
+        Self {
+            env: environment,
+            api_urls: ApiUrls::new(api_url),
+            auth: StdMutex::new(None),
+            client,
+            pve_compat: false,
+        }
+    }
+
+    pub async fn login_auth(&self) -> Result<Arc<AuthenticationKind>, E::Error> {
+        self.login().await?;
+        self.auth
+            .lock()
+            .unwrap()
+            .clone()
+            .ok_or_else(|| E::Error::internal("login failed to set authentication information"))
+    }
+
+    /// If currently logged in, this will fill in the auth cookie and CSRFPreventionToken header
+    /// and return `Ok(request)`, otherwise it'll return `Err(request)` with the request
+    /// unmodified.
+    pub fn try_set_auth_headers(
+        &self,
+        request: http::request::Builder,
+    ) -> Result<http::request::Builder, http::request::Builder> {
+        let auth = self.auth.lock().unwrap().clone();
+        match auth {
+            Some(auth) => Ok(auth.set_auth_headers(request)),
+            None => Err(request),
+        }
+    }
+
+    /// Convenience method to login and set the authentication headers for a request.
+    pub async fn set_auth_headers(
+        &self,
+        request: http::request::Builder,
+    ) -> Result<http::request::Builder, E::Error> {
+        Ok(self.login_auth().await?.set_auth_headers(request))
+    }
+
+    /// Ensure that we have a valid ticket.
+    ///
+    /// This will first attempt to load a ticket from the provided [`Environment`]. If successful,
+    /// its expiration time will be verified.
+    ///
+    /// If no valid ticket is available already, this will connect to the PVE API and perform
+    /// authentication.
+    pub async fn login(&self) -> Result<(), E::Error> {
+        let mut url_index = self.api_urls.index();
+        let current_url = &self.api_urls.urls[url_index];
+
+        let (userid, login) = self.need_login(current_url).await?;
+        let login = match login {
+            None => return Ok(()),
+            Some(login) => login,
+        };
+
+        let mut login = login.pve_compatibility(self.pve_compat);
+
+        let mut retry = None;
+        let generation = self.api_urls.generation();
+
+        // remember the finally successful address
+        let retry_success = |retry: &mut Option<usize>, url_index: usize| {
+            if retry.is_some() {
+                *retry = None;
+                if self.api_urls.generation() == generation {
+                    self.api_urls.current.store(url_index, Ordering::Relaxed);
+                    self.api_urls
+                        .generation
+                        .store(generation + 1, Ordering::Release);
+                }
+            }
+        };
+
+        // check whether we should be retrying a new address
+        let should_retry =
+            |retry: &mut Option<usize>, login: &mut Login, url_index: &mut usize| -> bool {
+                match *retry {
+                    Some(retry) => {
+                        if retry == *url_index {
+                            return false;
+                        }
+                    }
+                    None => *retry = Some(*url_index),
+                }
+
+                // if another thread successfully found a working URL already, use that as our last
+                // attempt:
+                if self.api_urls.generation() != generation {
+                    *url_index = self.api_urls.index();
+                    *retry = Some(*url_index);
+                    return true;
+                }
+
+                // otherwise cycle through the available addresses:
+                *url_index = (*url_index + 1) % self.api_urls.urls.len();
+                login.set_url(self.api_urls.urls[*url_index].to_string());
+
+                true
+            };
+
+        loop {
+            let current_url = &self.api_urls.urls[url_index];
+            let response = match self.client.request(to_request(login.request())?).await {
+                Ok(r) => {
+                    retry_success(&mut retry, url_index);
+                    r
+                }
+                Err(err) => {
+                    if should_retry(&mut retry, &mut login, &mut url_index) {
+                        continue;
+                    }
+                    return Err(err.into());
+                }
+            };
+
+            if !response.status().is_success() {
+                // FIXME: does `http` somehow expose the status string?
+                return Err(E::Error::api_error(
+                    response.status(),
+                    "authentication failed",
+                ));
+            }
+
+            let challenge = match login.response(response.body()).map_err(E::Error::bad_api)? {
+                TicketResult::Full(auth) => {
+                    return self.finish_auth(current_url, &userid, auth).await
+                }
+                TicketResult::TfaRequired(challenge) => challenge,
+            };
+
+            let response = self
+                .env
+                .query_second_factor_async(current_url, &userid, &challenge.challenge)
+                .await?;
+
+            let response = match self
+                .client
+                .request(to_request(challenge.respond_raw(&response))?)
+                .await
+            {
+                Ok(r) => {
+                    retry_success(&mut retry, url_index);
+                    r
+                }
+                Err(err) => {
+                    if should_retry(&mut retry, &mut login, &mut url_index) {
+                        continue;
+                    }
+                    return Err(err.into());
+                }
+            };
+
+            let status = response.status();
+            if !status.is_success() {
+                return Err(E::Error::api_error(status, "authentication failed"));
+            }
+
+            let auth = challenge
+                .response(response.body())
+                .map_err(E::Error::bad_api)?;
+
+            break self.finish_auth(current_url, &userid, auth).await;
+        }
+    }
+
+    /// Get the current username and, if required, a `Login` request.
+    async fn need_login(&self, current_url: &Uri) -> Result<(String, Option<Login>), E::Error> {
+        use proxmox_login::ticket::Validity;
+
+        let (userid, auth) = self.current_auth().await?;
+
+        let authkind = match auth {
+            None => {
+                let password = self.env.query_password_async(current_url, &userid).await?;
+                return Ok((
+                    userid.clone(),
+                    Some(Login::new(current_url.to_string(), userid, password)),
+                ));
+            }
+            Some(authkind) => authkind,
+        };
+
+        let auth = match &*authkind {
+            AuthenticationKind::Token(_) => return Ok((userid, None)),
+            AuthenticationKind::Ticket(auth) => auth,
+        };
+
+        Ok(match auth.ticket.validity() {
+            Validity::Valid => {
+                *self.auth.lock().unwrap() = Some(authkind);
+                (userid, None)
+            }
+            Validity::Refresh => (
+                userid,
+                Some(
+                    Login::renew(current_url.to_string(), auth.ticket.to_string())
+                        .map_err(E::Error::custom)?,
+                ),
+            ),
+
+            Validity::Expired => {
+                let password = self.env.query_password_async(current_url, &userid).await?;
+                (
+                    userid.clone(),
+                    Some(Login::new(current_url.to_string(), userid, password)),
+                )
+            }
+        })
+    }
+
+    /// Store the authentication info in our `auth` field and notify the environment.
+    async fn finish_auth(
+        &self,
+        current_url: &Uri,
+        userid: &str,
+        auth: Authentication,
+    ) -> Result<(), E::Error> {
+        let auth_string = serde_json::to_string(&auth).map_err(E::Error::internal)?;
+        *self.auth.lock().unwrap() = Some(Arc::new(auth.into()));
+        self.env
+            .store_ticket_async(current_url, userid, auth_string.as_bytes())
+            .await
+    }
+
+    /// Get the currently used API url from our array of possible cluster nodes.
+    fn api_url(&self) -> &Uri {
+        &self.api_urls.urls[self.api_urls.index()]
+    }
+
+    /// Get the current user id and a reference to the current authentication method.
+    /// If not authenticated yet, authenticate.
+    ///
+    /// This may cause the environment to be queried for user ids/passwords/FIDO/...
+    async fn current_auth(&self) -> Result<(String, Option<Arc<AuthenticationKind>>), E::Error> {
+        let auth = self.auth.lock().unwrap().clone();
+
+        let userid;
+        let auth = match auth {
+            Some(auth) => {
+                userid = auth.userid().to_owned();
+                Some(auth)
+            }
+            None => {
+                userid = self.env.query_userid_async(self.api_url()).await?;
+                self.reload_existing_ticket(&userid).await?
+            }
+        };
+
+        Ok((userid, auth))
+    }
+
+    /// Attempt to load an existing ticket from the environment.
+    async fn reload_existing_ticket(
+        &self,
+        userid: &str,
+    ) -> Result<Option<Arc<AuthenticationKind>>, E::Error> {
+        let ticket = match self.env.load_ticket_async(self.api_url(), userid).await? {
+            Some(auth) => auth,
+            None => return Ok(None),
+        };
+
+        let auth: Authentication = serde_json::from_slice(&ticket)
+            .map_err(|err| E::Error::env(format!("bad ticket data: {err}")))?;
+
+        let auth = Arc::new(auth.into());
+        *self.auth.lock().unwrap() = Some(Arc::clone(&auth));
+        Ok(Some(auth))
+    }
+
+    /// Build a URI relative to the current API endpoint.
+    fn build_uri(&self, base_uri: Uri, path: &str) -> Result<Uri, E::Error> {
+        let parts = base_uri.into_parts();
+        let mut builder = http::uri::Builder::new();
+        if let Some(scheme) = parts.scheme {
+            builder = builder.scheme(scheme);
+        }
+        if let Some(authority) = parts.authority {
+            builder = builder.authority(authority)
+        }
+        builder
+            .path_and_query(path.parse::<PathAndQuery>().map_err(E::Error::internal)?)
+            .build()
+            .map_err(E::Error::internal)
+    }
+
+    /// Attempt to execute a request, while automatically trying to reach different cluster nodes
+    /// if we fail to connect to the current node.
+    ///
+    /// The `make_request` closure gets the base `Uri` and should use it to build a `Request`.
+    /// The `Request` is then attempted. If there's a connection issue, `make_request` will be
+    /// called again with another cluster node (if available).
+    /// Only if no node responds - or a legitimate HTTP error is produced - will the error be
+    /// returned.
+    async fn request_retry_loop<Fut>(
+        &self,
+        make_request: impl Fn(Uri) -> Fut,
+    ) -> Result<Response<Vec<u8>>, E::Error>
+    where
+        Fut: Future<Output = Result<Request<Vec<u8>>, E::Error>> + Send,
+    {
+        let generation = self.api_urls.generation();
+        let mut url_index = self.api_urls.index();
+        let mut retry = None;
+        loop {
+            let err = match self
+                .client
+                .request(make_request(self.api_urls.urls[url_index].clone()).await?)
+                .await
+            {
+                Ok(response) => {
+                    if retry.is_some() && self.api_urls.generation() == generation {
+                        self.api_urls.current.store(url_index, Ordering::Relaxed);
+                        self.api_urls
+                            .generation
+                            .store(generation + 1, Ordering::Release);
+                    }
+                    return Ok(response);
+                }
+                Err(err) => err,
+            };
+
+            match retry {
+                Some(retry) => {
+                    if retry == url_index {
+                        return Err(err.into());
+                    }
+                }
+                None => retry = Some(url_index),
+            }
+
+            // if another thread successfully found a working URL already, use that as our last
+            // attempt:
+            if self.api_urls.generation() != generation {
+                url_index = self.api_urls.index();
+                retry = Some(url_index);
+                continue;
+            }
+
+            url_index = (url_index + 1) % self.api_urls.urls.len();
+        }
+    }
+
+    /// Execute a `GET` request, possibly trying multiple cluster nodes.
+    pub async fn get<'a, R>(&'a self, uri: &str) -> Result<ApiResponse<R>, E::Error>
+    where
+        R: serde::de::DeserializeOwned,
+    {
+        self.login().await?;
+
+        let response = self
+            .request_retry_loop(|base_uri| async {
+                self.set_auth_headers(Request::get(self.build_uri(base_uri, uri)?))
+                    .await?
+                    .body(Vec::new())
+                    .map_err(Error::internal)
+            })
+            .await?;
+
+        Self::handle_response(response)
+    }
+
+    /// Execute a `GET` request with the given body, possibly trying multiple cluster nodes.
+    pub async fn get_with_body<'a, B, R>(
+        &'a self,
+        uri: &str,
+        body: &'a B,
+    ) -> Result<ApiResponse<R>, E::Error>
+    where
+        B: serde::Serialize,
+        R: serde::de::DeserializeOwned,
+    {
+        let auth = self.login_auth().await?;
+        self.json_request(&auth, http::Method::GET, uri, body).await
+    }
+
+    /// Execute a `PUT` request with the given body, possibly trying multiple cluster nodes.
+    pub async fn put<'a, B, R>(&'a self, uri: &str, body: &'a B) -> Result<ApiResponse<R>, E::Error>
+    where
+        B: serde::Serialize,
+        R: serde::de::DeserializeOwned,
+    {
+        let auth = self.login_auth().await?;
+        self.json_request(&auth, http::Method::PUT, uri, body).await
+    }
+
+    /// Execute a `POST` request with the given body, possibly trying multiple cluster nodes.
+    pub async fn post<'a, B, R>(
+        &'a self,
+        uri: &str,
+        body: &'a B,
+    ) -> Result<ApiResponse<R>, E::Error>
+    where
+        B: serde::Serialize,
+        R: serde::de::DeserializeOwned,
+    {
+        let auth = self.login_auth().await?;
+        self.json_request(&auth, http::Method::POST, uri, body)
+            .await
+    }
+
+    /// Execute a `DELETE` request, possibly trying multiple cluster nodes.
+    pub async fn delete<'a, R>(&'a self, uri: &str) -> Result<ApiResponse<R>, E::Error>
+    where
+        R: serde::de::DeserializeOwned,
+    {
+        self.login().await?;
+
+        let response = self
+            .request_retry_loop(|base_uri| async {
+                self.set_auth_headers(Request::delete(self.build_uri(base_uri, uri)?))
+                    .await?
+                    .body(Vec::new())
+                    .map_err(Error::internal)
+            })
+            .await?;
+
+        Self::handle_response(response)
+    }
+
+    /// Execute a `DELETE` request with the given body, possibly trying multiple cluster nodes.
+    pub async fn delete_with_body<'a, B, R>(
+        &'a self,
+        uri: &str,
+        body: &'a B,
+    ) -> Result<ApiResponse<R>, E::Error>
+    where
+        B: serde::Serialize,
+        R: serde::de::DeserializeOwned,
+    {
+        let auth = self.login_auth().await?;
+        self.json_request(&auth, http::Method::DELETE, uri, body)
+            .await
+    }
+
+    /// Helper method for a JSON request with a JSON body `B`, yielding a JSON result type `R`.
+    pub(crate) async fn json_request<'a, B, R>(
+        &'a self,
+        auth: &'a AuthenticationKind,
+        method: http::Method,
+        uri: &str,
+        body: &'a B,
+    ) -> Result<ApiResponse<R>, E::Error>
+    where
+        B: serde::Serialize,
+        R: serde::de::DeserializeOwned,
+    {
+        let body = serde_json::to_vec(&body).map_err(E::Error::internal)?;
+        let content_length = body.len();
+        self.json_request_bytes(auth, method, uri, body, content_length)
+            .await
+    }
+
+    /// Helper method for a request with a byte body, yieldinig a JSON result of type `R`.
+    async fn json_request_bytes<'a, R>(
+        &'a self,
+        auth: &AuthenticationKind,
+        method: http::Method,
+        uri: &str,
+        body: Vec<u8>,
+        content_length: usize,
+    ) -> Result<ApiResponse<R>, E::Error>
+    where
+        R: serde::de::DeserializeOwned,
+    {
+        let response = self
+            .run_json_request_with_body(auth, method, uri, body, content_length)
+            .await?;
+        Self::handle_response(response)
+    }
+
+    async fn run_json_request_with_body<'a>(
+        &'a self,
+        auth: &'a AuthenticationKind,
+        method: http::Method,
+        uri: &str,
+        body: Vec<u8>,
+        content_length: usize,
+    ) -> Result<Response<Vec<u8>>, E::Error> {
+        self.request_retry_loop(|base_uri| async {
+            let request = Request::builder()
+                .method(method.clone())
+                .uri(self.build_uri(base_uri, uri)?)
+                .header(http::header::CONTENT_TYPE, "application/json")
+                .header(http::header::CONTENT_LENGTH, content_length.to_string());
+
+            auth.set_auth_headers(request)
+                .body(body.clone())
+                .map_err(Error::internal)
+        })
+        .await
+    }
+
+    /// Check the status code, deserialize the json/extjs `RawApiResponse` and check for error
+    /// messages inside.
+    /// On success, deserialize the expected result type.
+    fn handle_response<R>(response: Response<Vec<u8>>) -> Result<ApiResponse<R>, E::Error>
+    where
+        R: serde::de::DeserializeOwned,
+    {
+        if response.status() == StatusCode::UNAUTHORIZED {
+            return Err(E::Error::unauthorized());
+        }
+
+        if !response.status().is_success() {
+            // FIXME: Decode json errors...
+            //match serde_json::from_slice(&body)
+            //    Ok(value) =>
+            //        if value["error"]
+            let (response, body) = response.into_parts();
+            let body = String::from_utf8(body).map_err(Error::bad_api)?;
+            return Err(E::Error::api_error(response.status, body));
+        }
+
+        let data: RawApiResponse<R> =
+            serde_json::from_slice(&response.into_body()).map_err(Error::bad_api)?;
+
+        data.check()
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
+pub struct NoData;
+
+impl std::error::Error for NoData {}
+impl fmt::Display for NoData {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.write_str("api returned no data")
+    }
+}
+
+pub struct ApiResponse<T> {
+    pub data: Option<T>,
+    pub attribs: HashMap<String, Value>,
+}
+
+impl<T> ApiResponse<T> {
+    pub fn into_data_or_err(mut self) -> Result<T, NoData> {
+        self.data.take().ok_or(NoData)
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
+pub struct UnexpectedData;
+
+impl std::error::Error for UnexpectedData {}
+impl fmt::Display for UnexpectedData {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.write_str("api returned unexpected data")
+    }
+}
+
+impl ApiResponse<()> {
+    pub fn nodata(self) -> Result<(), UnexpectedData> {
+        if self.data.is_some() {
+            Err(UnexpectedData)
+        } else {
+            Ok(())
+        }
+    }
+}
+
+#[derive(serde::Deserialize)]
+struct RawApiResponse<T> {
+    #[serde(default, deserialize_with = "proxmox_login::parse::deserialize_u16")]
+    pub status: Option<u16>,
+    pub message: Option<String>,
+    #[serde(default, deserialize_with = "proxmox_login::parse::deserialize_bool")]
+    pub success: Option<bool>,
+    pub data: Option<T>,
+
+    #[serde(default)]
+    pub errors: HashMap<String, String>,
+
+    #[serde(default, flatten)]
+    pub attribs: HashMap<String, Value>,
+}
+
+impl<T> RawApiResponse<T> {
+    pub fn check<E: Error>(mut self) -> Result<ApiResponse<T>, E> {
+        if !self.success.unwrap_or(false) {
+            let status = http::StatusCode::from_u16(self.status.unwrap_or(400))
+                .unwrap_or(http::StatusCode::BAD_REQUEST);
+            let mut message = self
+                .message
+                .take()
+                .unwrap_or_else(|| "no message provided".to_string());
+            for (param, error) in self.errors {
+                use std::fmt::Write;
+                let _ = write!(message, "\n{param}: {error}");
+            }
+
+            return Err(E::api_error(status, message));
+        }
+
+        Ok(ApiResponse {
+            data: self.data,
+            attribs: self.attribs,
+        })
+    }
+}
+
+#[cfg(feature = "hyper-client")]
+pub type HyperClient<E> = Client<Arc<proxmox_http::client::Client>, E>;
+
+#[cfg(feature = "hyper-client")]
+impl<C, E> Client<C, E>
+where
+    E: Environment,
+    E::Error: From<anyhow::Error>,
+{
+    /// Create a new client instance which will connect to the provided endpoint.
+    pub fn new(api_url: Uri, environment: E) -> HyperClient<E> {
+        Client::with_client(
+            api_url,
+            environment,
+            Arc::new(proxmox_http::client::Client::new()),
+        )
+    }
+}
+
+#[cfg(feature = "hyper-client")]
+mod hyper_client_extras {
+    use std::future::Future;
+    use std::sync::Arc;
+
+    use anyhow::format_err;
+    use http::request::Request;
+    use http::response::Response;
+    use http::Uri;
+    use openssl::hash::MessageDigest;
+    use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};
+    use openssl::x509::{self, X509};
+
+    use proxmox_http::client::Client as ProxmoxClient;
+
+    use super::{Client, HyperClient};
+    use crate::Environment;
+
+    #[derive(Default)]
+    pub enum TlsOptions {
+        /// Default TLS verification.
+        #[default]
+        Verify,
+
+        /// Insecure: ignore invalid certificates.
+        Insecure,
+
+        /// Expect a specific certificate fingerprint.
+        Fingerprint(Vec<u8>),
+
+        /// Verify with a specific PEM formatted CA.
+        CaCert(X509),
+
+        /// Use a callback for certificate verification.
+        Callback(Box<dyn Fn(bool, &mut x509::X509StoreContextRef) -> bool + Send + Sync + 'static>),
+    }
+
+    fn fp_string(fp: &[u8]) -> String {
+        use std::fmt::Write as _;
+
+        let mut out = String::new();
+        for b in fp {
+            if !out.is_empty() {
+                out.push(':');
+            }
+            let _ = write!(out, "{b:02x}");
+        }
+        out
+    }
+
+    fn verify_fingerprint(chain: &x509::X509StoreContextRef, expected_fingerprint: &[u8]) -> bool {
+        let Some(cert) = chain.current_cert() else {
+            log::error!("no certificate in chain?");
+            return false;
+        };
+
+        let fp = match cert.digest(MessageDigest::sha256()) {
+            Err(err) => {
+                log::error!("error calculating certificate fingerprint: {err}");
+                return false;
+            }
+            Ok(fp) => fp,
+        };
+
+        if expected_fingerprint != fp.as_ref() {
+            log::error!("bad fingerprint: {}", fp_string(&fp));
+            log::error!("expected fingerprint: {}", fp_string(&expected_fingerprint));
+            return false;
+        }
+
+        true
+    }
+
+    impl<C, E> Client<C, E>
+    where
+        E: Environment,
+        E::Error: From<anyhow::Error>,
+    {
+        /// Create a new client instance which will connect to the provided endpoint.
+        pub fn with_options(
+            api_url: Uri,
+            environment: E,
+            tls_options: TlsOptions,
+            http_options: proxmox_http::HttpOptions,
+        ) -> Result<HyperClient<E>, E::Error> {
+            let mut connector = SslConnector::builder(SslMethod::tls_client())
+                .map_err(|err| format_err!("failed to create ssl connector builder: {err}"))?;
+
+            match tls_options {
+                TlsOptions::Verify => (),
+                TlsOptions::Insecure => connector.set_verify(SslVerifyMode::NONE),
+                TlsOptions::Fingerprint(expected_fingerprint) => {
+                    connector.set_verify_callback(SslVerifyMode::PEER, move |valid, chain| {
+                        if valid {
+                            return true;
+                        }
+                        verify_fingerprint(chain, &expected_fingerprint)
+                    });
+                }
+                TlsOptions::Callback(cb) => {
+                    connector.set_verify_callback(SslVerifyMode::PEER, move |valid, chain| {
+                        cb(valid, chain)
+                    });
+                }
+                TlsOptions::CaCert(ca) => {
+                    let mut store =
+                        openssl::x509::store::X509StoreBuilder::new().map_err(|err| {
+                            format_err!("failed to create certificate store builder: {err}")
+                        })?;
+                    store
+                        .add_cert(ca)
+                        .map_err(|err| format_err!("failed to build certificate store: {err}"))?;
+                    connector.set_cert_store(store.build());
+                }
+            }
+
+            let client = ProxmoxClient::with_ssl_connector(connector.build(), http_options);
+
+            Ok(Client::with_client(api_url, environment, Arc::new(client)))
+        }
+    }
+
+    impl super::HttpClient for Arc<proxmox_http::client::Client> {
+        type Error = anyhow::Error;
+        #[allow(clippy::type_complexity)]
+        type Request =
+            std::pin::Pin<Box<dyn Future<Output = Result<Response<Vec<u8>>, Self::Error>> + Send>>;
+
+        fn request(&self, request: Request<Vec<u8>>) -> Self::Request {
+            let (parts, body) = request.into_parts();
+            let request = Request::<hyper::Body>::from_parts(parts, body.into());
+            let this = Arc::clone(self);
+            Box::pin(async move {
+                use hyper::body::HttpBody;
+
+                let (response, mut body) = (*this).request(request).await?.into_parts();
+
+                let mut data = Vec::<u8>::new();
+                while let Some(more) = body.data().await {
+                    let more = more?;
+                    data.extend(&more[..]);
+                }
+
+                Ok::<_, anyhow::Error>(Response::from_parts(response, data))
+            })
+        }
+    }
+}
+
+#[cfg(feature = "hyper-client")]
+pub use hyper_client_extras::TlsOptions;

proxmox-client/src/environment.rs

@@ -0,0 +1,140 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::time::Duration;
+
+use http::Uri;
+
+use proxmox_login::tfa::TfaChallenge;
+
+use crate::Error;
+
+/// Provide input from the environment for storing/loading tickets or tokens and querying the user
+/// for passwords or 2nd factors.
+pub trait Environment: Send + Sync {
+    type Error: Error;
+
+    /// Store a ticket belonging to a user of an API.
+    ///
+    /// This is only used if `store_ticket_async` is not overwritten and may be left unimplemented
+    /// in async code. By default it will just return an error.
+    ///
+    /// [`store_ticket_async`]: Environment::store_ticket_async
+    fn store_ticket(&self, api_url: &Uri, userid: &str, ticket: &[u8]) -> Result<(), Self::Error> {
+        let _ = (api_url, userid, ticket);
+        Err(Self::Error::custom(
+            "missing store_ticket(_async) implementation",
+        ))
+    }
+
+    /// Load a user's cached ticket for an API url.
+    ///
+    /// This is only used if [`load_ticket_async`] is not overwritten and may be left unimplemented
+    /// in async code. By default it will just return an error.
+    ///
+    /// [`load_ticket_async`]: Environment::load_ticket_async
+    fn load_ticket(&self, api_url: &Uri, userid: &str) -> Result<Option<Vec<u8>>, Self::Error> {
+        let _ = (api_url, userid);
+        Err(Self::Error::custom(
+            "missing load_ticket(_async) implementation",
+        ))
+    }
+
+    /// Query for a userid (name and realm).
+    ///
+    /// This is only used if [`query_userid_async`] is not overwritten and may be left
+    /// unimplemented in async code. By default it will just return an error.
+    ///
+    /// [`query_userid_async`]: Environment::query_userid_async
+    fn query_userid(&self, api_url: &Uri) -> Result<String, Self::Error> {
+        let _ = api_url;
+        Err(Self::Error::custom(
+            "missing query_userid(_async) implementation",
+        ))
+    }
+
+    /// Query for a password.
+    ///
+    /// This is only used if [`query_password_async`] is not overwritten and may be left
+    /// unimplemented in async code. By default it will just return an error.
+    ///
+    /// [`query_password_async`]: Environment::query_password_async
+    fn query_password(&self, api_url: &Uri, userid: &str) -> Result<String, Self::Error> {
+        let _ = (api_url, userid);
+        Err(Self::Error::custom(
+            "missing query_password(_async) implementation",
+        ))
+    }
+
+    /// Query for a second factor. The default implementation is to not support 2nd factors.
+    ///
+    /// This is only used if [`query_second_factor_async`] is not overwritten and may be left
+    /// unimplemented in async code. By default it will just return an error.
+    ///
+    /// [`query_second_factor_async`]: Environment::query_second_factor_async
+    fn query_second_factor(
+        &self,
+        api_url: &Uri,
+        userid: &str,
+        challenge: &TfaChallenge,
+    ) -> Result<String, Self::Error> {
+        let _ = (api_url, userid, challenge);
+        Err(Self::Error::second_factor_not_supported())
+    }
+
+    /// The client code uses async rust and it is fine to implement this instead of `store_ticket`.
+    fn store_ticket_async<'a>(
+        &'a self,
+        api_url: &'a Uri,
+        userid: &'a str,
+        ticket: &'a [u8],
+    ) -> Pin<Box<dyn Future<Output = Result<(), Self::Error>> + Send + 'a>> {
+        Box::pin(async move { self.store_ticket(api_url, userid, ticket) })
+    }
+
+    #[allow(clippy::type_complexity)]
+    fn load_ticket_async<'a>(
+        &'a self,
+        api_url: &'a Uri,
+        userid: &'a str,
+    ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>, Self::Error>> + Send + 'a>> {
+        Box::pin(async move { self.load_ticket(api_url, userid) })
+    }
+
+    fn query_userid_async<'a>(
+        &'a self,
+        api_url: &'a Uri,
+    ) -> Pin<Box<dyn Future<Output = Result<String, Self::Error>> + Send + 'a>> {
+        Box::pin(async move { self.query_userid(api_url) })
+    }
+
+    fn query_password_async<'a>(
+        &'a self,
+        api_url: &'a Uri,
+        userid: &'a str,
+    ) -> Pin<Box<dyn Future<Output = Result<String, Self::Error>> + Send + 'a>> {
+        Box::pin(async move { self.query_password(api_url, userid) })
+    }
+
+    fn query_second_factor_async<'a>(
+        &'a self,
+        api_url: &'a Uri,
+        userid: &'a str,
+        challenge: &'a TfaChallenge,
+    ) -> Pin<Box<dyn Future<Output = Result<String, Self::Error>> + Send + 'a>> {
+        Box::pin(async move { self.query_second_factor(api_url, userid, challenge) })
+    }
+
+    /// In order to allow the polling based task API to function, we need a way to sleep in async
+    /// context.
+    /// This will likely be removed when the streaming tasks API is available.
+    ///
+    /// # Panics
+    ///
+    /// The default implementation simply panics.
+    fn sleep(
+        time: Duration,
+    ) -> Result<Pin<Box<dyn Future<Output = ()> + Send + 'static>>, Self::Error> {
+        let _ = time;
+        Err(Self::Error::sleep_not_supported())
+    }
+}

proxmox-client/src/error.rs

@@ -0,0 +1,61 @@
+use std::any::Any;
+use std::fmt::{self, Display};
+
+/// For error types provided by the user of this crate.
+pub trait Error: Sized + Display + fmt::Debug + Any + Send + Sync + 'static {
+    /// An arbitrary error message.
+    fn custom<T: Display>(msg: T) -> Self;
+
+    /// Successfully queried the status of a task, and the task has failed.
+    fn task_failed<T: Display>(msg: T) -> Self {
+        Self::custom(format!("task failed: {msg}"))
+    }
+
+    /// An API call returned an error status.
+    fn api_error<T: Display>(status: http::StatusCode, msg: T) -> Self {
+        Self::custom(format!("api error (status = {status}): {msg}"))
+    }
+
+    /// The API behaved unexpectedly.
+    fn bad_api<T: Display>(msg: T) -> Self {
+        Self::custom(msg)
+    }
+
+    /// The environment returned an error or bad data.
+    fn env<T: Display>(msg: T) -> Self {
+        Self::custom(msg)
+    }
+
+    /// A second factor was required, but the [`Environment`](crate::Environment) did not provide
+    /// an implementation to get it.
+    fn second_factor_not_supported() -> Self {
+        Self::custom("not supported")
+    }
+
+    /// There was an error building an [`http::Uri`].
+    fn uri(err: http::Error) -> Self {
+        Self::custom(err)
+    }
+
+    /// A generic internal error such as a serde_json serialization error.
+    fn internal<T: Display>(err: T) -> Self {
+        Self::custom(err)
+    }
+
+    /// An API call which requires authorization was attempted without logging in first.
+    fn unauthorized() -> Self {
+        Self::custom("unauthorized")
+    }
+
+    /// An extended client call required the ability to "pause" while polling API endpoints.
+    /// (Mostly to wait for "tasks" to finish.), and no implementation for this was provided.
+    fn sleep_not_supported() -> Self {
+        Self::custom("no async 'sleep' implementation available")
+    }
+}
+
+impl Error for anyhow::Error {
+    fn custom<T: Display>(msg: T) -> Self {
+        anyhow::format_err!("{msg}")
+    }
+}

proxmox-client/src/lib.rs

@@ -0,0 +1,17 @@
+mod environment;
+mod error;
+
+pub use environment::Environment;
+pub use error::Error;
+
+pub use proxmox_login::tfa::TfaChallenge;
+pub use proxmox_login::{Authentication, Ticket};
+
+pub(crate) mod auth;
+pub use auth::Token;
+
+mod client;
+pub use client::{ApiResponse, Client, HttpClient};
+
+#[cfg(feature = "hyper-client")]
+pub use client::{HyperClient, TlsOptions};