mirror of
https://github.com/narrrl/proton-sdk-rs.git
synced 2026-11-03 16:41:26 +01:00
899 lines
33 KiB
Rust
899 lines
33 KiB
Rust
//! HTTP plumbing: header injection, the Proton response envelope, bearer-token
|
|
//! authentication and transparent 401 refresh.
|
|
//!
|
|
//! Mirrors `HttpApiCallBuilder`, `AuthorizationHandler` and `TokenCredential`
|
|
//! from the C# SDK, collapsed into a single reqwest-based client since Rust has
|
|
//! no `DelegatingHandler` pipeline.
|
|
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use bytes::Bytes;
|
|
use rand::RngExt;
|
|
use reqwest::{Method, StatusCode};
|
|
use serde::Serialize;
|
|
use serde::de::DeserializeOwned;
|
|
use tokio::sync::Mutex;
|
|
|
|
use crate::api::{ApiResponse, HumanVerificationCredential, ResponseCode};
|
|
use crate::config::{API_CONTENT_TYPE, ProtonClientConfiguration, RetryPolicy};
|
|
use crate::error::{ProtonApiError, ProtonError, Result};
|
|
use crate::ids::SessionId;
|
|
use crate::telemetry::{NoopTelemetry, Telemetry, TelemetryExt};
|
|
|
|
const SESSION_ID_HEADER: &str = "x-pm-uid";
|
|
const APP_VERSION_HEADER: &str = "x-pm-appversion";
|
|
const STORAGE_TOKEN_HEADER: &str = "pm-storage-token";
|
|
const HV_TOKEN_HEADER: &str = "x-pm-human-verification-token";
|
|
const HV_TOKEN_TYPE_HEADER: &str = "x-pm-human-verification-token-type";
|
|
|
|
/// The mutable authentication tokens for a session, shared between every
|
|
/// request and the refresh path.
|
|
#[derive(Debug, Clone)]
|
|
pub struct Tokens {
|
|
pub access_token: String,
|
|
pub refresh_token: String,
|
|
}
|
|
|
|
/// A reqwest-backed client bound to a single authenticated session.
|
|
///
|
|
/// Cloning is cheap (everything is reference-counted) and shares the same token
|
|
/// state, so a refresh triggered by one request is visible to all others.
|
|
#[derive(Clone)]
|
|
pub struct ApiHttpClient {
|
|
inner: Arc<Inner>,
|
|
/// Extra path segment prepended to every request path (after `base_url`,
|
|
/// before the per-call `path`). Mirrors C# `session.GetHttpClient(baseRoute)`
|
|
/// — the Drive client targets `…/drive/` while account/auth calls stay at the
|
|
/// root. Empty by default. Lives on the outer struct (not `Inner`) so clones
|
|
/// can carry different prefixes while sharing one token/telemetry store.
|
|
route_prefix: Arc<str>,
|
|
}
|
|
|
|
/// Callback invoked after a successful token refresh.
|
|
type TokensRefreshedCallback = Arc<dyn Fn(Tokens) + Send + Sync>;
|
|
|
|
struct Inner {
|
|
http: reqwest::Client,
|
|
base_url: String,
|
|
config: ProtonClientConfiguration,
|
|
session_id: SessionId,
|
|
tokens: Mutex<Tokens>,
|
|
/// Telemetry sink for per-request events. Interior-mutable because the
|
|
/// client is already shared (cloned into the Drive client) by the time a
|
|
/// caller attaches a sink via [`ApiHttpClient::set_telemetry`]. Defaults to
|
|
/// a no-op. `std::sync::Mutex` (not tokio's) — held only for the cheap
|
|
/// clone/replace, never across an await.
|
|
telemetry: std::sync::Mutex<Arc<dyn Telemetry>>,
|
|
on_tokens_refreshed: std::sync::Mutex<Option<TokensRefreshedCallback>>,
|
|
}
|
|
|
|
impl ApiHttpClient {
|
|
/// Build a client for an authenticated session.
|
|
pub fn new(
|
|
config: ProtonClientConfiguration,
|
|
session_id: SessionId,
|
|
tokens: Tokens,
|
|
) -> Result<Self> {
|
|
// `gzip` makes reqwest advertise Accept-Encoding and transparently decode
|
|
// the response. The Proton API honours it for the JSON envelope, which is
|
|
// the bulk of a metadata-heavy walk (link details, listings). Block bodies
|
|
// are already ciphertext and will not compress — the header costs nothing
|
|
// there, and storage responses are unaffected either way.
|
|
let http = reqwest::Client::builder()
|
|
.timeout(config.request_timeout)
|
|
.gzip(true)
|
|
.build()?;
|
|
|
|
let base_url = ensure_trailing_slash(&config.base_url);
|
|
|
|
Ok(Self {
|
|
inner: Arc::new(Inner {
|
|
http,
|
|
base_url,
|
|
config,
|
|
session_id,
|
|
tokens: Mutex::new(tokens),
|
|
telemetry: std::sync::Mutex::new(NoopTelemetry::shared()),
|
|
on_tokens_refreshed: std::sync::Mutex::new(None),
|
|
}),
|
|
route_prefix: Arc::from(""),
|
|
})
|
|
}
|
|
|
|
/// Derive a clone that prepends `route` to every request path, sharing this
|
|
/// client's token store, telemetry sink and connection pool. Mirrors C#
|
|
/// `session.GetHttpClient(baseRoute)`: the Drive client passes `"drive/"` so
|
|
/// its routes resolve under `…/drive/` while auth/account calls (and token
|
|
/// refresh) stay at the root. `route` should end in `/`.
|
|
pub fn with_base_route(&self, route: impl Into<Arc<str>>) -> Self {
|
|
Self {
|
|
inner: Arc::clone(&self.inner),
|
|
route_prefix: route.into(),
|
|
}
|
|
}
|
|
|
|
/// Snapshot the current tokens (e.g. to persist for a later `resume`).
|
|
pub async fn current_tokens(&self) -> Tokens {
|
|
self.inner.tokens.lock().await.clone()
|
|
}
|
|
|
|
/// Attach a telemetry sink to receive a per-request
|
|
/// [`TelemetryEvent`](crate::telemetry::TelemetryEvent) (operation
|
|
/// `http_request` for API calls, `storage_download` / `storage_upload` for
|
|
/// block storage; attributes carry the HTTP method and status). Replaces any
|
|
/// previous sink. Takes effect for every clone of this client, since they
|
|
/// share state.
|
|
pub fn set_telemetry(&self, telemetry: Arc<dyn Telemetry>) {
|
|
*self
|
|
.inner
|
|
.telemetry
|
|
.lock()
|
|
.expect("telemetry mutex poisoned") = telemetry;
|
|
}
|
|
|
|
/// Set a callback to be invoked whenever the session's tokens are refreshed.
|
|
/// Replaces any previous callback. Takes effect for every clone of this client.
|
|
pub fn set_on_tokens_refreshed(&self, callback: impl Fn(Tokens) + Send + Sync + 'static) {
|
|
*self
|
|
.inner
|
|
.on_tokens_refreshed
|
|
.lock()
|
|
.expect("on_tokens_refreshed mutex poisoned") = Some(Arc::new(callback));
|
|
}
|
|
|
|
/// Snapshot the current telemetry sink.
|
|
fn telemetry(&self) -> Arc<dyn Telemetry> {
|
|
self.inner
|
|
.telemetry
|
|
.lock()
|
|
.expect("telemetry mutex poisoned")
|
|
.clone()
|
|
}
|
|
|
|
/// `GET {url}` against block storage, returning the raw (still-encrypted)
|
|
/// blob bytes.
|
|
///
|
|
/// Block storage lives on a different host from the API: the URL is
|
|
/// absolute and authorization is a per-block `pm-storage-token` header
|
|
/// rather than the session bearer. Mirrors C# `StorageApiClient
|
|
/// .GetBlobStreamAsync`. A successful response is raw binary; an error
|
|
/// response is the usual JSON envelope.
|
|
///
|
|
/// Returns [`Bytes`] rather than `Vec<u8>` so the 4 MiB body is not copied
|
|
/// on its way to the decryptor — the reference-counted buffer reqwest
|
|
/// already assembled is handed straight through, including into the
|
|
/// blocking decrypt task.
|
|
pub async fn get_storage_blob(&self, url: &str, token: &str) -> Result<Bytes> {
|
|
let mut timer = self.telemetry().start("storage_download");
|
|
let response = send_retrying(&self.inner.config.retry_policy, || {
|
|
// Override the client-level (API) timeout: a 4 MiB block on a slow
|
|
// uplink legitimately outruns the 30s JSON budget.
|
|
let mut request = self
|
|
.inner
|
|
.http
|
|
.get(url)
|
|
.timeout(self.inner.config.storage_timeout)
|
|
.header(STORAGE_TOKEN_HEADER, token);
|
|
if !self.inner.config.user_agent.is_empty() {
|
|
request =
|
|
request.header(reqwest::header::USER_AGENT, &self.inner.config.user_agent);
|
|
}
|
|
request
|
|
})
|
|
.await?;
|
|
let status = response.status();
|
|
timer.attr("status", status.as_u16());
|
|
let bytes = response.bytes().await?;
|
|
|
|
// Success bodies are raw block bytes (not JSON); only error responses
|
|
// carry the envelope.
|
|
if let Ok(envelope) = serde_json::from_slice::<ApiResponse>(&bytes) {
|
|
if !envelope.is_success() {
|
|
return Err(api_error(status, &bytes));
|
|
}
|
|
} else if !status.is_success() {
|
|
return Err(api_error(status, &bytes));
|
|
}
|
|
|
|
timer.success();
|
|
Ok(bytes)
|
|
}
|
|
|
|
/// `POST {url}` a block blob to storage as `multipart/form-data`.
|
|
///
|
|
/// Mirrors C# `StorageApiClient.UploadBlobAsync`: a single `Block` part
|
|
/// (filename `blob`, `application/octet-stream`) on the storage host,
|
|
/// authorized by the per-block `pm-storage-token` header rather than the
|
|
/// session bearer. The response is the usual JSON envelope.
|
|
///
|
|
/// Takes [`Bytes`] rather than `Vec<u8>` because the multipart body has to be
|
|
/// rebuilt per attempt (a stream body can't be cloned) and a block is up to
|
|
/// 4 MiB: cloning `Bytes` bumps a refcount where cloning the `Vec` copied the
|
|
/// whole block on *every* attempt, first one included. The part is built with
|
|
/// an explicit length so the request still carries a `Content-Length`.
|
|
pub async fn post_storage_blob(&self, url: &str, token: &str, blob: Bytes) -> Result<()> {
|
|
// Validate the part once up front; the multipart body itself is rebuilt
|
|
// per attempt inside the retry closure.
|
|
reqwest::multipart::Part::bytes(Vec::new())
|
|
.mime_str("application/octet-stream")
|
|
.map_err(ProtonError::from)?;
|
|
|
|
let blob_len = blob.len() as u64;
|
|
let mut timer = self.telemetry().start("storage_upload");
|
|
let response = send_retrying(&self.inner.config.retry_policy, || {
|
|
let body = reqwest::Body::from(blob.clone());
|
|
let part = reqwest::multipart::Part::stream_with_length(body, blob_len)
|
|
.file_name("blob")
|
|
.mime_str("application/octet-stream")
|
|
.expect("octet-stream is a valid MIME type");
|
|
let form = reqwest::multipart::Form::new().part("Block", part);
|
|
|
|
let mut request = self
|
|
.inner
|
|
.http
|
|
.post(url)
|
|
.timeout(self.inner.config.storage_timeout)
|
|
.header(STORAGE_TOKEN_HEADER, token)
|
|
.multipart(form);
|
|
|
|
if !self.inner.config.user_agent.is_empty() {
|
|
request =
|
|
request.header(reqwest::header::USER_AGENT, &self.inner.config.user_agent);
|
|
}
|
|
request
|
|
})
|
|
.await?;
|
|
let status = response.status();
|
|
timer.attr("status", status.as_u16());
|
|
let bytes = response.bytes().await?;
|
|
|
|
if let Ok(envelope) = serde_json::from_slice::<ApiResponse>(&bytes) {
|
|
if !envelope.is_success() {
|
|
return Err(api_error(status, &bytes));
|
|
}
|
|
} else if !status.is_success() {
|
|
return Err(api_error(status, &bytes));
|
|
}
|
|
timer.success();
|
|
Ok(())
|
|
}
|
|
|
|
pub fn session_id(&self) -> &SessionId {
|
|
&self.inner.session_id
|
|
}
|
|
|
|
/// `GET {path}` returning a typed success body.
|
|
pub async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T> {
|
|
self.send::<(), T>(Method::GET, path, None).await
|
|
}
|
|
|
|
/// `POST {path}` with a JSON body, returning a typed success body.
|
|
pub async fn post<B: Serialize, T: DeserializeOwned>(&self, path: &str, body: &B) -> Result<T> {
|
|
self.send::<B, T>(Method::POST, path, Some(body)).await
|
|
}
|
|
|
|
/// `POST {path}` as multipart form data with a JSON `Metadata` part and
|
|
/// zero or more binary parts. Used by Drive's atomic small-upload endpoints.
|
|
pub async fn post_multipart<B: Serialize, T: DeserializeOwned>(
|
|
&self,
|
|
path: &str,
|
|
metadata: &B,
|
|
binary_parts: &[(String, Vec<u8>)],
|
|
) -> Result<T> {
|
|
let metadata = serde_json::to_vec(metadata)?;
|
|
let mut timer = self.telemetry().start("http_request");
|
|
timer.attr("method", "POST");
|
|
let rejected_token = self.inner.tokens.lock().await.access_token.clone();
|
|
let response = self
|
|
.send_multipart_with_token(path, &metadata, binary_parts, &rejected_token)
|
|
.await?;
|
|
let response = if response.status() == StatusCode::UNAUTHORIZED {
|
|
let bytes = response.bytes().await?;
|
|
if let Ok(envelope) = serde_json::from_slice::<ApiResponse>(&bytes)
|
|
&& matches!(
|
|
envelope.code,
|
|
ResponseCode::AccountDeleted | ResponseCode::AccountDisabled
|
|
)
|
|
{
|
|
return Err(api_error(StatusCode::UNAUTHORIZED, &bytes));
|
|
}
|
|
let token = self.refresh_access_token(&rejected_token).await?;
|
|
self.send_multipart_with_token(path, &metadata, binary_parts, &token)
|
|
.await?
|
|
} else {
|
|
response
|
|
};
|
|
timer.attr("status", response.status().as_u16());
|
|
let parsed = parse_response(response).await?;
|
|
timer.success();
|
|
Ok(parsed)
|
|
}
|
|
|
|
/// `PUT {path}` with a JSON body, returning a typed success body.
|
|
pub async fn put<B: Serialize, T: DeserializeOwned>(&self, path: &str, body: &B) -> Result<T> {
|
|
self.send::<B, T>(Method::PUT, path, Some(body)).await
|
|
}
|
|
|
|
/// `DELETE {path}` returning a typed success body.
|
|
pub async fn delete<T: DeserializeOwned>(&self, path: &str) -> Result<T> {
|
|
self.send::<(), T>(Method::DELETE, path, None).await
|
|
}
|
|
|
|
/// `DELETE {path}` with a JSON body, returning a typed success body.
|
|
///
|
|
/// A few Drive endpoints take their operand in the body of a `DELETE`
|
|
/// (C# `HttpApiCallBuilder.DeleteAsync(route, payload, ...)`), e.g. removing
|
|
/// photo tags.
|
|
pub async fn delete_with_body<B: Serialize, T: DeserializeOwned>(
|
|
&self,
|
|
path: &str,
|
|
body: &B,
|
|
) -> Result<T> {
|
|
self.send::<B, T>(Method::DELETE, path, Some(body)).await
|
|
}
|
|
|
|
async fn send<B: Serialize, T: DeserializeOwned>(
|
|
&self,
|
|
method: Method,
|
|
path: &str,
|
|
body: Option<&B>,
|
|
) -> Result<T> {
|
|
let mut timer = self.telemetry().start("http_request");
|
|
timer.attr("method", method.as_str());
|
|
|
|
let access_token = self.inner.tokens.lock().await.access_token.clone();
|
|
|
|
// An early `?` here records the op as a failure (OpTimer defaults to it).
|
|
let response = self
|
|
.send_with_token(method.clone(), path, body, &access_token)
|
|
.await?;
|
|
|
|
let response = if response.status() == StatusCode::UNAUTHORIZED {
|
|
self.handle_unauthorized(method, path, body, response, access_token)
|
|
.await?
|
|
} else {
|
|
response
|
|
};
|
|
|
|
timer.attr("status", response.status().as_u16());
|
|
let parsed = parse_response(response).await?;
|
|
timer.success();
|
|
Ok(parsed)
|
|
}
|
|
|
|
async fn handle_unauthorized<B: Serialize>(
|
|
&self,
|
|
method: Method,
|
|
path: &str,
|
|
body: Option<&B>,
|
|
response: reqwest::Response,
|
|
rejected_access_token: String,
|
|
) -> Result<reqwest::Response> {
|
|
// Don't bother refreshing for terminal account states.
|
|
let bytes = response.bytes().await?;
|
|
if let Ok(envelope) = serde_json::from_slice::<ApiResponse>(&bytes)
|
|
&& matches!(
|
|
envelope.code,
|
|
ResponseCode::AccountDeleted | ResponseCode::AccountDisabled
|
|
)
|
|
{
|
|
return Err(api_error(StatusCode::UNAUTHORIZED, &bytes));
|
|
}
|
|
|
|
let access_token = self.refresh_access_token(&rejected_access_token).await?;
|
|
self.send_with_token(method, path, body, &access_token)
|
|
.await
|
|
}
|
|
|
|
async fn send_with_token<B: Serialize>(
|
|
&self,
|
|
method: Method,
|
|
path: &str,
|
|
body: Option<&B>,
|
|
access_token: &str,
|
|
) -> Result<reqwest::Response> {
|
|
let url = format!(
|
|
"{}{}{}",
|
|
self.inner.base_url,
|
|
self.route_prefix,
|
|
path.trim_start_matches('/')
|
|
);
|
|
send_retrying(&self.inner.config.retry_policy, || {
|
|
let mut request = self
|
|
.inner
|
|
.http
|
|
.request(method.clone(), &url)
|
|
.header(SESSION_ID_HEADER, self.inner.session_id.as_str())
|
|
.header(APP_VERSION_HEADER, &self.inner.config.app_version)
|
|
.header(reqwest::header::ACCEPT, API_CONTENT_TYPE)
|
|
.bearer_auth(access_token);
|
|
|
|
if !self.inner.config.user_agent.is_empty() {
|
|
request =
|
|
request.header(reqwest::header::USER_AGENT, &self.inner.config.user_agent);
|
|
}
|
|
|
|
if let Some(body) = body {
|
|
request = request.json(body);
|
|
}
|
|
|
|
request
|
|
})
|
|
.await
|
|
}
|
|
|
|
async fn send_multipart_with_token(
|
|
&self,
|
|
path: &str,
|
|
metadata: &[u8],
|
|
binary_parts: &[(String, Vec<u8>)],
|
|
access_token: &str,
|
|
) -> Result<reqwest::Response> {
|
|
let url = format!(
|
|
"{}{}{}",
|
|
self.inner.base_url,
|
|
self.route_prefix,
|
|
path.trim_start_matches('/')
|
|
);
|
|
send_retrying(&self.inner.config.retry_policy, || {
|
|
let metadata_part = reqwest::multipart::Part::bytes(metadata.to_vec())
|
|
.file_name("Metadata")
|
|
.mime_str("application/json")
|
|
.expect("application/json is a valid MIME type");
|
|
let mut form = reqwest::multipart::Form::new().part("Metadata", metadata_part);
|
|
for (name, bytes) in binary_parts {
|
|
let part = reqwest::multipart::Part::bytes(bytes.clone())
|
|
.file_name(name.clone())
|
|
.mime_str("application/octet-stream")
|
|
.expect("octet-stream is a valid MIME type");
|
|
form = form.part(name.clone(), part);
|
|
}
|
|
let mut request = self
|
|
.inner
|
|
.http
|
|
.post(&url)
|
|
.timeout(self.inner.config.storage_timeout)
|
|
.header(SESSION_ID_HEADER, self.inner.session_id.as_str())
|
|
.header(APP_VERSION_HEADER, &self.inner.config.app_version)
|
|
.header(reqwest::header::ACCEPT, API_CONTENT_TYPE)
|
|
.bearer_auth(access_token)
|
|
.multipart(form);
|
|
if !self.inner.config.user_agent.is_empty() {
|
|
request =
|
|
request.header(reqwest::header::USER_AGENT, &self.inner.config.user_agent);
|
|
}
|
|
request
|
|
})
|
|
.await
|
|
}
|
|
|
|
/// Refresh the session tokens, deduplicating concurrent refreshes: if the
|
|
/// in-memory access token already differs from the rejected one, another
|
|
/// task refreshed first and we reuse its result.
|
|
async fn refresh_access_token(&self, rejected_access_token: &str) -> Result<String> {
|
|
let mut guard = self.inner.tokens.lock().await;
|
|
|
|
if guard.access_token != rejected_access_token {
|
|
return Ok(guard.access_token.clone());
|
|
}
|
|
|
|
let refreshed = self.request_refresh(&guard.refresh_token).await?;
|
|
*guard = refreshed.clone();
|
|
|
|
// Notify callback
|
|
if let Some(ref cb) = *self
|
|
.inner
|
|
.on_tokens_refreshed
|
|
.lock()
|
|
.expect("on_tokens_refreshed mutex poisoned")
|
|
{
|
|
cb(refreshed.clone());
|
|
}
|
|
|
|
Ok(refreshed.access_token)
|
|
}
|
|
|
|
async fn request_refresh(&self, refresh_token: &str) -> Result<Tokens> {
|
|
let url = format!("{}auth/v4/refresh", self.inner.base_url);
|
|
let body = SessionRefreshRequest {
|
|
response_type: "token",
|
|
grant_type: "refresh_token",
|
|
refresh_token,
|
|
redirect_uri: &self.inner.config.refresh_redirect_uri,
|
|
};
|
|
|
|
// The refresh call carries the session id but, deliberately, no bearer
|
|
// token (the access token is the thing being replaced).
|
|
let response = send_retrying(&self.inner.config.retry_policy, || {
|
|
self.inner
|
|
.http
|
|
.post(&url)
|
|
.header(SESSION_ID_HEADER, self.inner.session_id.as_str())
|
|
.header(APP_VERSION_HEADER, &self.inner.config.app_version)
|
|
.header(reqwest::header::ACCEPT, API_CONTENT_TYPE)
|
|
.json(&body)
|
|
})
|
|
.await?;
|
|
|
|
let refreshed: SessionRefreshResponse = parse_response(response).await?;
|
|
Ok(Tokens {
|
|
access_token: refreshed.access_token,
|
|
refresh_token: refreshed.refresh_token,
|
|
})
|
|
}
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
struct SessionRefreshRequest<'a> {
|
|
#[serde(rename = "ResponseType")]
|
|
response_type: &'a str,
|
|
#[serde(rename = "GrantType")]
|
|
grant_type: &'a str,
|
|
#[serde(rename = "RefreshToken")]
|
|
refresh_token: &'a str,
|
|
#[serde(rename = "RedirectURI")]
|
|
redirect_uri: &'a str,
|
|
}
|
|
|
|
#[derive(serde::Deserialize)]
|
|
struct SessionRefreshResponse {
|
|
#[serde(rename = "AccessToken")]
|
|
access_token: String,
|
|
#[serde(rename = "RefreshToken")]
|
|
refresh_token: String,
|
|
}
|
|
|
|
/// `GET {path}` without a session: no `x-pm-uid` and no bearer token.
|
|
///
|
|
/// The public-link flow needs this: `drive/urls/{token}/info` opens the SRP
|
|
/// handshake and is by definition callable by a visitor who has no Proton
|
|
/// session at all.
|
|
pub async fn get_unauthenticated<T: DeserializeOwned>(
|
|
config: &ProtonClientConfiguration,
|
|
path: &str,
|
|
) -> Result<T> {
|
|
let http = reqwest::Client::builder()
|
|
.timeout(config.request_timeout)
|
|
.gzip(true)
|
|
.build()?;
|
|
|
|
let base_url = ensure_trailing_slash(&config.base_url);
|
|
let url = format!("{}{}", base_url, path.trim_start_matches('/'));
|
|
|
|
let response = send_retrying(&config.retry_policy, || {
|
|
let mut request = http
|
|
.get(&url)
|
|
.header(APP_VERSION_HEADER, &config.app_version)
|
|
.header(reqwest::header::ACCEPT, API_CONTENT_TYPE);
|
|
if !config.user_agent.is_empty() {
|
|
request = request.header(reqwest::header::USER_AGENT, &config.user_agent);
|
|
}
|
|
request
|
|
})
|
|
.await?;
|
|
|
|
parse_response(response).await
|
|
}
|
|
|
|
/// `POST {path}` without a session: no `x-pm-uid` and no bearer token.
|
|
///
|
|
/// Used by the SRP login flow (`auth/v4/info`, `auth/v4`), which runs before a
|
|
/// session exists. Mirrors the C# SDK's `BeginAsync`, which issues these calls
|
|
/// on a session-less `HttpClient`.
|
|
pub async fn post_unauthenticated<B: Serialize, T: DeserializeOwned>(
|
|
config: &ProtonClientConfiguration,
|
|
path: &str,
|
|
body: &B,
|
|
) -> Result<T> {
|
|
post_unauthenticated_verified(config, path, body, None).await
|
|
}
|
|
|
|
/// [`post_unauthenticated`], optionally replaying a solved human-verification
|
|
/// challenge.
|
|
///
|
|
/// A login from an unfamiliar IP comes back `9001` with a challenge instead of a
|
|
/// session. Once the user has solved it, the *same* request is sent again with
|
|
/// the resulting token in the header pair below — the challenge is not a
|
|
/// separate handshake, it is a precondition attached to the original call, which
|
|
/// is why this takes the credential rather than exposing a "verify" endpoint.
|
|
pub async fn post_unauthenticated_verified<B: Serialize, T: DeserializeOwned>(
|
|
config: &ProtonClientConfiguration,
|
|
path: &str,
|
|
body: &B,
|
|
verification: Option<&HumanVerificationCredential>,
|
|
) -> Result<T> {
|
|
let http = reqwest::Client::builder()
|
|
.timeout(config.request_timeout)
|
|
.gzip(true)
|
|
.build()?;
|
|
|
|
let base_url = ensure_trailing_slash(&config.base_url);
|
|
let url = format!("{}{}", base_url, path.trim_start_matches('/'));
|
|
|
|
let response = send_retrying(&config.retry_policy, || {
|
|
let mut request = http
|
|
.post(&url)
|
|
.header(APP_VERSION_HEADER, &config.app_version)
|
|
.header(reqwest::header::ACCEPT, API_CONTENT_TYPE)
|
|
.json(body);
|
|
if !config.user_agent.is_empty() {
|
|
request = request.header(reqwest::header::USER_AGENT, &config.user_agent);
|
|
}
|
|
if let Some(hv) = verification {
|
|
request = request
|
|
.header(HV_TOKEN_HEADER, &hv.token)
|
|
.header(HV_TOKEN_TYPE_HEADER, &hv.method);
|
|
}
|
|
request
|
|
})
|
|
.await?;
|
|
|
|
parse_response(response).await
|
|
}
|
|
|
|
/// Read a response body, enforce the Proton success envelope, and deserialize
|
|
/// the typed success payload.
|
|
async fn parse_response<T: DeserializeOwned>(response: reqwest::Response) -> Result<T> {
|
|
let status = response.status();
|
|
let bytes = response.bytes().await?;
|
|
|
|
// Every Proton response embeds the envelope; a missing/non-success code or a
|
|
// non-2xx HTTP status is an API error. `MultipleResponses` (1001) is a batch
|
|
// multi-status, not a failure: the real per-item codes live in the body, so
|
|
// the caller (e.g. trash/restore/delete) inspects them itself.
|
|
if let Ok(envelope) = serde_json::from_slice::<ApiResponse>(&bytes) {
|
|
if !envelope.is_success() && envelope.code != ResponseCode::MultipleResponses {
|
|
return Err(api_error(status, &bytes));
|
|
}
|
|
} else if !status.is_success() {
|
|
return Err(api_error(status, &bytes));
|
|
}
|
|
|
|
Ok(serde_json::from_slice::<T>(&bytes)?)
|
|
}
|
|
|
|
fn api_error(status: StatusCode, bytes: &[u8]) -> ProtonError {
|
|
let envelope = serde_json::from_slice::<ApiResponse>(bytes).ok();
|
|
let code = envelope
|
|
.as_ref()
|
|
.map(|e| e.code)
|
|
.unwrap_or(ResponseCode::Unknown);
|
|
let details = envelope.as_ref().and_then(|e| e.details.clone());
|
|
let message = envelope.and_then(|e| e.error_message).unwrap_or_else(|| {
|
|
status
|
|
.canonical_reason()
|
|
.unwrap_or("unknown error")
|
|
.to_owned()
|
|
});
|
|
|
|
ProtonError::Api(ProtonApiError {
|
|
code,
|
|
http_status: status.as_u16(),
|
|
message,
|
|
details,
|
|
})
|
|
}
|
|
|
|
/// Send a request, transparently retrying retryable failures per `policy`.
|
|
///
|
|
/// `build` is called once per attempt to produce a fresh `RequestBuilder`
|
|
/// (reqwest builders are consumed by `send`, and streaming bodies like
|
|
/// `multipart` can't be cloned), so every retry resends the full request.
|
|
///
|
|
/// Retryable = HTTP 408/429/502/503/504 or a transient transport error
|
|
/// (timeout / connect). A `Retry-After` header (delta-seconds) is honoured;
|
|
/// otherwise the delay is exponential backoff with full jitter. Non-retryable
|
|
/// responses and errors — including ordinary 4xx and the 401 that drives token
|
|
/// refresh — pass straight through to the caller untouched.
|
|
async fn send_retrying<F>(policy: &RetryPolicy, build: F) -> Result<reqwest::Response>
|
|
where
|
|
F: Fn() -> reqwest::RequestBuilder,
|
|
{
|
|
let mut attempt: u32 = 0;
|
|
loop {
|
|
match build().send().await {
|
|
Ok(response) => {
|
|
if attempt < policy.max_retries && is_retryable_status(response.status()) {
|
|
let delay = retry_after(&response).unwrap_or_else(|| backoff(policy, attempt));
|
|
tokio::time::sleep(delay).await;
|
|
attempt += 1;
|
|
continue;
|
|
}
|
|
return Ok(response);
|
|
}
|
|
Err(err) => {
|
|
if attempt < policy.max_retries && is_retryable_error(&err) {
|
|
tokio::time::sleep(backoff(policy, attempt)).await;
|
|
attempt += 1;
|
|
continue;
|
|
}
|
|
return Err(err.into());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Status codes Proton (or an intermediary) returns for transient conditions:
|
|
/// request timeout, rate limit, and the gateway/unavailable family.
|
|
fn is_retryable_status(status: StatusCode) -> bool {
|
|
matches!(status.as_u16(), 408 | 429 | 502 | 503 | 504)
|
|
}
|
|
|
|
/// A transport error worth retrying: a timeout or a failure to connect. A
|
|
/// mid-body error (`is_body`) is not retried — the request may have been
|
|
/// applied server-side.
|
|
fn is_retryable_error(err: &reqwest::Error) -> bool {
|
|
err.is_timeout() || err.is_connect()
|
|
}
|
|
|
|
/// Parse a `Retry-After` header expressed as delta-seconds. The HTTP-date form
|
|
/// is not emitted by the Proton API, so it is ignored (falls back to backoff).
|
|
fn retry_after(response: &reqwest::Response) -> Option<Duration> {
|
|
let value = response
|
|
.headers()
|
|
.get(reqwest::header::RETRY_AFTER)?
|
|
.to_str()
|
|
.ok()?;
|
|
parse_retry_after_secs(value)
|
|
}
|
|
|
|
/// Parse a `Retry-After` delta-seconds value into a delay. Non-numeric values
|
|
/// (the HTTP-date form, which Proton does not emit) yield `None`.
|
|
fn parse_retry_after_secs(value: &str) -> Option<Duration> {
|
|
value.trim().parse().ok().map(Duration::from_secs)
|
|
}
|
|
|
|
/// Exponential backoff with full jitter: a uniformly random delay in
|
|
/// `[0, base_delay * 2^attempt]`, capped at `max_delay`.
|
|
fn backoff(policy: &RetryPolicy, attempt: u32) -> Duration {
|
|
let ceiling = policy
|
|
.base_delay
|
|
.saturating_mul(1u32.checked_shl(attempt).unwrap_or(u32::MAX))
|
|
.min(policy.max_delay);
|
|
let ceiling_ms = ceiling.as_millis() as u64;
|
|
if ceiling_ms == 0 {
|
|
return Duration::ZERO;
|
|
}
|
|
Duration::from_millis(rand::rng().random_range(0..=ceiling_ms))
|
|
}
|
|
|
|
fn ensure_trailing_slash(url: &str) -> String {
|
|
if url.ends_with('/') {
|
|
url.to_owned()
|
|
} else {
|
|
format!("{url}/")
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn retryable_statuses() {
|
|
for code in [408u16, 429, 502, 503, 504] {
|
|
assert!(is_retryable_status(StatusCode::from_u16(code).unwrap()));
|
|
}
|
|
for code in [200u16, 400, 401, 403, 404, 500] {
|
|
assert!(!is_retryable_status(StatusCode::from_u16(code).unwrap()));
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn retry_after_parses_seconds_only() {
|
|
assert_eq!(parse_retry_after_secs("5"), Some(Duration::from_secs(5)));
|
|
assert_eq!(
|
|
parse_retry_after_secs(" 12 "),
|
|
Some(Duration::from_secs(12))
|
|
);
|
|
assert_eq!(parse_retry_after_secs("0"), Some(Duration::ZERO));
|
|
// HTTP-date form is unsupported -> falls back to backoff.
|
|
assert_eq!(
|
|
parse_retry_after_secs("Wed, 21 Oct 2015 07:28:00 GMT"),
|
|
None
|
|
);
|
|
assert_eq!(parse_retry_after_secs(""), None);
|
|
}
|
|
|
|
#[test]
|
|
fn backoff_grows_then_caps_within_jitter_bounds() {
|
|
let policy = RetryPolicy {
|
|
max_retries: 5,
|
|
base_delay: Duration::from_millis(100),
|
|
max_delay: Duration::from_millis(1000),
|
|
};
|
|
// Full jitter: every sample stays within [0, ceiling] where the
|
|
// ceiling is base*2^attempt capped at max_delay.
|
|
for attempt in 0..8u32 {
|
|
let ceiling = Duration::from_millis(100u64.saturating_mul(1 << attempt.min(20)))
|
|
.min(policy.max_delay);
|
|
for _ in 0..64 {
|
|
assert!(backoff(&policy, attempt) <= ceiling);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn backoff_handles_large_attempt_without_overflow() {
|
|
let policy = RetryPolicy::default();
|
|
// attempt >= 32 would overflow a naive shift; must saturate to max_delay.
|
|
assert!(backoff(&policy, 64) <= policy.max_delay);
|
|
}
|
|
|
|
#[test]
|
|
fn disabled_policy_has_no_retries() {
|
|
assert_eq!(RetryPolicy::disabled().max_retries, 0);
|
|
}
|
|
|
|
/// A telemetry sink that records every event for assertions.
|
|
struct Capture(std::sync::Mutex<Vec<crate::telemetry::TelemetryEvent>>);
|
|
|
|
impl Telemetry for Capture {
|
|
fn record(&self, event: &crate::telemetry::TelemetryEvent) {
|
|
self.0.lock().unwrap().push(event.clone());
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn http_request_records_telemetry_event() {
|
|
use crate::telemetry::Outcome;
|
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
|
|
|
// One-shot loopback server: read the request, reply with the success
|
|
// envelope, then close.
|
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
|
let addr = listener.local_addr().unwrap();
|
|
let server = tokio::spawn(async move {
|
|
let (mut sock, _) = listener.accept().await.unwrap();
|
|
let mut buf = [0u8; 2048];
|
|
let _ = sock.read(&mut buf).await.unwrap();
|
|
let body = br#"{"Code":1000}"#;
|
|
let head = format!(
|
|
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
|
|
body.len()
|
|
);
|
|
sock.write_all(head.as_bytes()).await.unwrap();
|
|
sock.write_all(body).await.unwrap();
|
|
sock.flush().await.unwrap();
|
|
});
|
|
|
|
let config = ProtonClientConfiguration::new("test@1.0")
|
|
.with_base_url(format!("http://{addr}/"))
|
|
.with_retry_policy(RetryPolicy::disabled());
|
|
let client = ApiHttpClient::new(
|
|
config,
|
|
SessionId::from("test-session"),
|
|
Tokens {
|
|
access_token: "access".into(),
|
|
refresh_token: "refresh".into(),
|
|
},
|
|
)
|
|
.unwrap();
|
|
|
|
let capture = Arc::new(Capture(std::sync::Mutex::new(Vec::new())));
|
|
client.set_telemetry(capture.clone());
|
|
|
|
let _: ApiResponse = client.get("some/path").await.unwrap();
|
|
server.await.unwrap();
|
|
|
|
let events = capture.0.lock().unwrap();
|
|
assert_eq!(events.len(), 1, "exactly one http_request event");
|
|
let event = &events[0];
|
|
assert_eq!(event.operation, "http_request");
|
|
assert_eq!(event.outcome, Outcome::Success);
|
|
assert!(
|
|
event
|
|
.attributes
|
|
.iter()
|
|
.any(|(k, v)| *k == "method" && v == "GET")
|
|
);
|
|
assert!(
|
|
event
|
|
.attributes
|
|
.iter()
|
|
.any(|(k, v)| *k == "status" && v == "200")
|
|
);
|
|
}
|
|
}
|