BuzzbeeSCD/clustri
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add PBS datastore management and harden macOS keychain storage

Browse Source 
Add Proxmox Backup Server datastore management: overview, datastore detail, download/prune/verify/GC dialogs, usePbs hook, backend commands, and tests.

Fix macOS keychain re-writes failing with 'item already exists': replace keyring 3 with keyring-core plus per-platform stores (macOS Keychain, Windows Credential Manager, Linux keyutils), recover by deleting the stale item and retrying once, and surface actionable messages for locked keychains.
This commit is contained in:
Matt
2026-08-13 01:02:33 +00:00
parent 5960102489
commit 039ac6f9d3
43 changed files with 4934 additions and 430 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/connection.rs
+390 -221
View File
@@ -1,4 +1,8 @@
use crate::error::Error;
use crate::keyring::{
delete_credential as keyring_delete_credential, describe_error as keyring_describe_error,
entry as keyring_entry, keyring, set_password as keyring_set_password,
};
use crate::proxmox::{
AddDiskConfig, AddNICConfig, Backup, BackupJob, BackupJobConfig, ClusterNode, ClusterStatus,
CreateSnapshotConfig, Disk, EditNICConfig, NetworkInterface, Node, RestoreConfig, Snapshot,
@@ -75,16 +79,57 @@ pub enum AuthMode {
Password,
}
/// The kind of Proxmox server a connection targets. The two platforms share
/// the JSON API shape and the ticket/token authentication model but use
/// different header names and login endpoints.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ServerType {
Pve,
Pbs,
}
impl ServerType {
/// Maps a connection's `serverType` string to a [`ServerType`]. Any value
/// other than `"pbs"` is treated as PVE.
pub(crate) fn from_config(s: &str) -> ServerType {
if s == "pbs" {
ServerType::Pbs
} else {
ServerType::Pve
}
}
/// The `Authorization` header value prefix for token authentication
/// (`PVEAPIToken` for VE, `PBSAPIToken` for PBS).
pub(crate) fn token_header_name(self) -> &'static str {
match self {
ServerType::Pve => "PVEAPIToken",
ServerType::Pbs => "PBSAPIToken",
}
}
/// The `Cookie` header value prefix for ticket authentication
/// (`PVEAuthCookie` for VE, `PBSAuthCookie` for PBS).
pub(crate) fn cookie_header_name(self) -> &'static str {
match self {
ServerType::Pve => "PVEAuthCookie",
ServerType::Pbs => "PBSAuthCookie",
}
}
}
/// Authentication material for a Proxmox API request.
///
/// Token mode uses `token` as a `PVEAPIToken` header; password mode uses
/// `ticket` as a `PVEAuthCookie` (plus `csrf_token` for non-GET requests).
/// Token mode uses `token` as a `PVEAPIToken`/`PBSAPIToken` header; password
/// mode uses `ticket` as a `PVEAuthCookie`/`PBSAuthCookie` (plus `csrf_token`
/// for non-GET requests). The header names are chosen from `server_type`.
#[derive(Debug, Clone)]
pub struct AuthContext {
pub mode: AuthMode,
pub token: Option<String>,
pub ticket: Option<String>,
pub csrf_token: Option<String>,
pub server_type: ServerType,
}
/// Builds the full Proxmox API URL for a request.
@@ -101,7 +146,7 @@ fn build_api_url(base_url: &str, path: &str) -> crate::Result<Url> {
/// Deserializes an API response payload into `T`, mapping a parse failure to a
/// [`crate::Error::SerializationError`] that names the endpoint so the
/// mismatch is easy to diagnose.
fn parse_api<T>(endpoint: &str, data: serde_json::Value) -> crate::Result<T>
pub(crate) fn parse_api<T>(endpoint: &str, data: serde_json::Value) -> crate::Result<T>
where
T: serde::de::DeserializeOwned,
{
@@ -160,29 +205,7 @@ pub async fn api_request(
}
let mut request = client.request(method, url);
match auth.mode {
AuthMode::Token => {
let token = auth
.token
.as_deref()
.ok_or_else(|| Error::InvalidCredentials("No API token configured".to_string()))?;
request = request.header("Authorization", format!("PVEAPIToken={}", token));
}
AuthMode::Password => {
let ticket = auth
.ticket
.as_deref()
.ok_or_else(|| Error::AuthError("Not logged in: no session ticket".to_string()))?;
request = request.header("Cookie", format!("PVEAuthCookie={}", ticket));
if needs_csrf {
let csrf = auth
.csrf_token
.as_deref()
.ok_or_else(|| Error::AuthError("Not logged in: no CSRF token".to_string()))?;
request = request.header("CSRFPreventionToken", csrf);
}
}
}
request = apply_auth_headers(request, auth, needs_csrf)?;
if let Some(fields) = form {
request = request.form(fields);
@@ -202,13 +225,7 @@ pub async fn api_request(
let status = response.status();
let text = response.text().await.map_err(Error::HttpError)?;
let body: serde_json::Value = serde_json::from_str(&text).unwrap_or_else(|_| {
if text.is_empty() {
serde_json::Value::Null
} else {
serde_json::Value::String(text)
}
});
let body = parse_body_text(&text);
if !status.is_success() {
let message = error_message_from_body(&body, status.as_u16());
@@ -218,6 +235,63 @@ pub async fn api_request(
Ok(body.get("data").cloned().unwrap_or(body))
}
/// Parses a raw response body into a JSON value, treating an empty body as
/// `Null` and a non-JSON body as a plain string. The standard
/// `{ "data": ... }` envelope is left intact; callers decide whether to
/// unwrap it.
fn parse_body_text(text: &str) -> serde_json::Value {
match serde_json::from_str(text) {
Ok(value) => value,
Err(_) if text.is_empty() => serde_json::Value::Null,
Err(_) => serde_json::Value::String(text.to_string()),
}
}
/// Applies the authentication headers from `auth` to a request builder.
///
/// Token mode sets an `Authorization` header whose value prefix depends on the
/// server type (`PVEAPIToken`/`PBSAPIToken`). Password mode sets a `Cookie`
/// header (`PVEAuthCookie`/`PBSAuthCookie`) and, for non-GET requests, the
/// `CSRFPreventionToken` header. Missing secrets surface as
/// [`Error::InvalidCredentials`]/[`Error::AuthError`], matching the behavior
/// of the pre-refactor inline injection in `api_request`.
fn apply_auth_headers(
mut request: reqwest::RequestBuilder,
auth: &AuthContext,
needs_csrf: bool,
) -> crate::Result<reqwest::RequestBuilder> {
match auth.mode {
AuthMode::Token => {
let token = auth
.token
.as_deref()
.ok_or_else(|| Error::InvalidCredentials("No API token configured".to_string()))?;
request = request.header(
"Authorization",
format!("{}={}", auth.server_type.token_header_name(), token),
);
}
AuthMode::Password => {
let ticket = auth
.ticket
.as_deref()
.ok_or_else(|| Error::AuthError("Not logged in: no session ticket".to_string()))?;
request = request.header(
"Cookie",
format!("{}={}", auth.server_type.cookie_header_name(), ticket),
);
if needs_csrf {
let csrf = auth
.csrf_token
.as_deref()
.ok_or_else(|| Error::AuthError("Not logged in: no CSRF token".to_string()))?;
request = request.header("CSRFPreventionToken", csrf);
}
}
}
Ok(request)
}
/// Builds the error message surfaced for a non-success API response.
///
/// The `errors` field is preferred over the generic `message` (Proxmox
@@ -267,9 +341,9 @@ fn error_message_from_body(body: &serde_json::Value, status: u16) -> String {
format!("Proxmox API error (HTTP {})", status)
}
struct Connection {
config: ConnectionConfig,
client: Client,
pub(crate) struct Connection {
pub(crate) config: ConnectionConfig,
pub(crate) client: Client,
ticket: Mutex<Option<String>>,
csrf_token: Mutex<Option<String>>,
current_endpoint_index: Mutex<usize>,
@@ -293,7 +367,8 @@ impl Connection {
/// Token mode reads the token from the connection config, falling back to
/// the keyring. Password mode uses the in-memory session, loading the
/// ticket/CSRF token from the keyring and caching them if absent.
fn auth_context(&self) -> crate::Result<AuthContext> {
pub(crate) fn auth_context(&self) -> crate::Result<AuthContext> {
let server_type = ServerType::from_config(&self.config.server_type);
if self.config.auth_mode == "token" {
let token = match self.config.primary.token.as_deref() {
Some(token) if !token.is_empty() => Some(token.to_string()),
@@ -304,6 +379,7 @@ impl Connection {
token,
ticket: None,
csrf_token: None,
server_type,
})
} else {
let ticket = self
@@ -315,6 +391,7 @@ impl Connection {
token: None,
ticket: Some(ticket),
csrf_token,
server_type,
})
}
}
@@ -326,7 +403,7 @@ impl Connection {
Ok(entry) => match entry.get_password() {
Ok(value) => Ok(Some(value)),
Err(keyring::Error::NoEntry) => Ok(None),
Err(e) => Err(Error::KeyringError(e.to_string())),
Err(e) => Err(Error::KeyringError(keyring_describe_error(&e))),
},
Err(_) => Ok(None),
}
@@ -357,7 +434,7 @@ impl Connection {
/// Returns the ordered, deduplicated candidate endpoint URLs for failover:
/// the primary first, followed by each configured fallback. Empty URLs are
/// dropped and duplicates are collapsed, preserving order.
fn endpoint_urls(&self) -> Vec<String> {
pub(crate) fn endpoint_urls(&self) -> Vec<String> {
let mut urls: Vec<String> = Vec::new();
let primary = self.config.primary.url.clone();
let fallbacks = self
@@ -376,7 +453,7 @@ impl Connection {
/// Remembers the endpoint that last served a request, so the next request
/// resumes rotation there instead of re-testing a down primary.
fn set_endpoint_index(&self, idx: usize) {
pub(crate) fn set_endpoint_index(&self, idx: usize) {
if let Ok(mut guard) = self.current_endpoint_index.lock() {
*guard = idx;
}
@@ -384,7 +461,7 @@ impl Connection {
/// Records the runtime status of the last request: `"connected"`,
/// `"failover"`, or `"failed"`.
fn set_runtime_status(&self, status: &str) {
pub(crate) fn set_runtime_status(&self, status: &str) {
if let Ok(mut guard) = self.runtime_status.lock() {
*guard = status.to_string();
}
@@ -406,7 +483,7 @@ impl Connection {
/// (connection refused, timeouts, DNS resolution) trigger failover to the
/// next candidate; authentication and API errors are returned immediately
/// without rotating.
async fn request(
pub(crate) async fn request(
&self,
method: Method,
path: &str,
@@ -439,15 +516,95 @@ impl Connection {
Err(last_transport_err
.unwrap_or_else(|| Error::ConnectionFailed("no endpoints available".to_string())))
}
}
fn keyring_service() -> &'static str {
"clustri"
}
/// Streams a GET response body (binary, no `{data}` envelope unwrap) to
/// `dest`.
///
/// Uses the same endpoint rotation and auth as `request()`: transport
/// failures rotate to the next candidate endpoint, while auth/API/other
/// errors propagate immediately. Returns the number of bytes written.
///
/// Used by the PBS backup-download endpoints, which stream raw archive
/// bytes rather than a JSON envelope.
pub(crate) async fn download_to_file(
&self,
path: &str,
query: &[(&str, String)],
dest: &std::path::Path,
) -> crate::Result<u64> {
let auth = self.auth_context()?;
let candidates = self.endpoint_urls();
let start = *self
.current_endpoint_index
.lock()
.unwrap_or_else(|e| e.into_inner());
let mut last_transport_err = None;
for offset in 0..candidates.len() {
let idx = (start + offset) % candidates.len();
let url = &candidates[idx];
match self
.download_from_endpoint(url, path, query, &auth, dest)
.await
{
Ok(bytes) => {
self.set_endpoint_index(idx);
self.set_runtime_status(if idx == 0 { "connected" } else { "failover" });
return Ok(bytes);
}
Err(Error::ConnectionFailed(message)) => {
last_transport_err = Some(Error::ConnectionFailed(message));
}
Err(error) => return Err(error),
}
}
self.set_runtime_status("failed");
Err(last_transport_err
.unwrap_or_else(|| Error::ConnectionFailed("no endpoints available".to_string())))
}
fn keyring_entry(connection_id: &str, field: &str) -> crate::Result<keyring::Entry> {
let key = format!("{}:{}", connection_id, field);
keyring::Entry::new(keyring_service(), &key).map_err(|e| Error::KeyringError(e.to_string()))
/// Performs a single raw GET download against `base_url` and writes the
/// response body to `dest`. Auth headers mirror `api_request` (a GET needs
/// no CSRF header, but the cookie/authorization header is always set); a
/// non-success status is surfaced as [`Error::ApiError`].
async fn download_from_endpoint(
&self,
base_url: &str,
path: &str,
query: &[(&str, String)],
auth: &AuthContext,
dest: &std::path::Path,
) -> crate::Result<u64> {
let mut url = build_api_url(base_url, path)?;
{
let mut pairs = url.query_pairs_mut();
for (key, value) in query {
pairs.append_pair(key, value);
}
}
let request = apply_auth_headers(self.client.get(url), auth, false)?;
let response = request.send().await.map_err(|e| {
if e.is_connect() || e.is_timeout() || e.is_request() {
// Transport-level failures are surfaced as `ConnectionFailed`
// so the caller can fail over to another endpoint.
Error::ConnectionFailed(format!("Cannot connect to server: {}", e))
} else {
Error::HttpError(e)
}
})?;
let status = response.status();
if !status.is_success() {
let text = response.text().await.map_err(Error::HttpError)?;
let body = parse_body_text(&text);
let message = error_message_from_body(&body, status.as_u16());
return Err(Error::ApiError(message));
}
let bytes = response.bytes().await.map_err(Error::HttpError)?;
tokio::fs::write(dest, &bytes).await?;
Ok(bytes.len() as u64)
}
}
pub struct ConnectionManager {
@@ -534,22 +691,10 @@ impl ConnectionManager {
pub async fn remove_connection(&mut self, id: &str, path: &Path) -> crate::Result<()> {
self.connections.remove(id);
// Clear stored credentials from keyring
let _ = keyring_entry(id, "ticket").and_then(|e| {
e.delete_credential()
.map_err(|e| Error::KeyringError(e.to_string()))
});
let _ = keyring_entry(id, "csrf_token").and_then(|e| {
e.delete_credential()
.map_err(|e| Error::KeyringError(e.to_string()))
});
let _ = keyring_entry(id, "password").and_then(|e| {
e.delete_credential()
.map_err(|e| Error::KeyringError(e.to_string()))
});
let _ = keyring_entry(id, "token").and_then(|e| {
e.delete_credential()
.map_err(|e| Error::KeyringError(e.to_string()))
});
let _ = keyring_delete_credential(id, "ticket");
let _ = keyring_delete_credential(id, "csrf_token");
let _ = keyring_delete_credential(id, "password");
let _ = keyring_delete_credential(id, "token");
if self.active_connection_id.as_deref() == Some(id) {
self.active_connection_id = None;
}
@@ -655,110 +800,116 @@ impl ConnectionManager {
// Same-cluster merge: when this connection belongs to the same cluster
// as an existing one, fold its endpoint and node list into the
// existing connection and drop it.
let cluster_id = {
// existing connection and drop it. PBS connections are single-host and
// never participate in cluster merging.
let (cluster_id, server_type) = {
let conn = self
.connections
.get(id)
.expect("connection existence was checked above");
conn.config.cluster_id.clone()
(
conn.config.cluster_id.clone(),
conn.config.server_type.clone(),
)
};
if let Some(cid) = cluster_id.filter(|cid| !cid.is_empty()) {
let other_id = self
.connections
.iter()
.find(|(other_id, conn)| {
other_id.as_str() != id
&& conn.config.cluster_id.as_deref() == Some(cid.as_str())
})
.map(|(other_id, _)| other_id.clone());
if server_type != "pbs" {
if let Some(cid) = cluster_id.filter(|cid| !cid.is_empty()) {
let other_id = self
.connections
.iter()
.find(|(other_id, conn)| {
other_id.as_str() != id
&& conn.config.cluster_id.as_deref() == Some(cid.as_str())
})
.map(|(other_id, _)| other_id.clone());
if let Some(other_id) = other_id {
let (this_primary_url, this_primary_node, this_nodes) = {
let conn = self
.connections
.get(id)
.expect("connection existence was checked above");
(
conn.config.primary.url.clone(),
conn.config.primary.node.clone(),
conn.config.nodes.clone(),
)
};
if let Some(other_id) = other_id {
let (this_primary_url, this_primary_node, this_nodes) = {
let conn = self
.connections
.get(id)
.expect("connection existence was checked above");
(
conn.config.primary.url.clone(),
conn.config.primary.node.clone(),
conn.config.nodes.clone(),
)
};
{
let other = self
.connections
.get_mut(&other_id)
.expect("merge target was located above");
// This connection's primary endpoint becomes a fallback on
// the surviving connection, deduplicated case-insensitively.
let url_known = other
.config
.fallbacks
.iter()
.any(|endpoint| endpoint.url.eq_ignore_ascii_case(&this_primary_url));
if !url_known {
other.config.fallbacks.push(EndpointConfig {
url: this_primary_url.clone(),
node: this_primary_node.clone(),
token: None,
});
}
// Adopt the merging connection's primary node only when the
// surviving connection has none yet.
if other
.config
.primary
.node
.as_deref()
.map_or(true, str::is_empty)
{
other.config.primary.node = this_primary_node;
}
// Merge the node lists (dedup by URL), then re-derive each
// node's primary marker against the surviving connection's
// primary URL.
let other_primary_url = other.config.primary.url.clone();
for node in this_nodes {
if !other
let other = self
.connections
.get_mut(&other_id)
.expect("merge target was located above");
// This connection's primary endpoint becomes a fallback on
// the surviving connection, deduplicated case-insensitively.
let url_known = other
.config
.nodes
.fallbacks
.iter()
.any(|existing| existing.url.eq_ignore_ascii_case(&node.url))
.any(|endpoint| endpoint.url.eq_ignore_ascii_case(&this_primary_url));
if !url_known {
other.config.fallbacks.push(EndpointConfig {
url: this_primary_url.clone(),
node: this_primary_node.clone(),
token: None,
});
}
// Adopt the merging connection's primary node only when the
// surviving connection has none yet.
if other
.config
.primary
.node
.as_deref()
.map_or(true, str::is_empty)
{
other.config.nodes.push(node);
other.config.primary.node = this_primary_node;
}
// Merge the node lists (dedup by URL), then re-derive each
// node's primary marker against the surviving connection's
// primary URL.
let other_primary_url = other.config.primary.url.clone();
for node in this_nodes {
if !other
.config
.nodes
.iter()
.any(|existing| existing.url.eq_ignore_ascii_case(&node.url))
{
other.config.nodes.push(node);
}
}
for node in &mut other.config.nodes {
node.is_primary = node.url.eq_ignore_ascii_case(&other_primary_url);
}
if other
.config
.primary
.node
.as_deref()
.map_or(true, str::is_empty)
{
if let Some(primary) =
other.config.nodes.iter().find(|node| node.is_primary)
{
other.config.primary.node = Some(primary.name.clone());
}
}
}
for node in &mut other.config.nodes {
node.is_primary = node.url.eq_ignore_ascii_case(&other_primary_url);
}
if other
.config
.primary
.node
.as_deref()
.map_or(true, str::is_empty)
{
if let Some(primary) =
other.config.nodes.iter().find(|node| node.is_primary)
{
other.config.primary.node = Some(primary.name.clone());
}
}
}
self.connections.remove(id);
if self.active_connection_id.as_deref() == Some(id) {
self.active_connection_id = Some(other_id.clone());
self.connections.remove(id);
if self.active_connection_id.as_deref() == Some(id) {
self.active_connection_id = Some(other_id.clone());
}
self.persist(path)?;
return Ok(ConnectResult {
connection_id: other_id.clone(),
merged_into: Some(other_id),
status: "connected".to_string(),
});
}
self.persist(path)?;
return Ok(ConnectResult {
connection_id: other_id.clone(),
merged_into: Some(other_id),
status: "connected".to_string(),
});
}
}
@@ -800,7 +951,11 @@ impl ConnectionManager {
let conn = self.connection(id)?;
conn.set_endpoint_index(0);
conn.set_runtime_status("connected");
self.discover_nodes(id).await?;
// PBS is single-host: there is no node list or cluster identity to
// discover, so the discovery pass is skipped entirely.
if !conn.config.is_pbs() {
self.discover_nodes(id).await?;
}
Ok(())
}
@@ -872,10 +1027,17 @@ impl ConnectionManager {
});
}
let transport_failed = match self.discover_nodes(connection_id).await {
Ok(_) => false,
Err(Error::ConnectionFailed(_)) => true,
Err(_) => false,
// PBS connections have no discoverable node list, so the status poll
// reports the snapshot and the current runtime status without any
// discovery request.
let transport_failed = if self.connection(connection_id)?.config.is_pbs() {
false
} else {
match self.discover_nodes(connection_id).await {
Ok(_) => false,
Err(Error::ConnectionFailed(_)) => true,
Err(_) => false,
}
};
let (primary_url, current_endpoint_url, nodes) = self.status_snapshot(connection_id)?;
@@ -1024,22 +1186,10 @@ impl ConnectionManager {
};
// Store credentials in keyring for later use
keyring_entry(&connection_id, "ticket").and_then(|e| {
e.set_password(&ticket)
.map_err(|e| Error::KeyringError(e.to_string()))
})?;
keyring_entry(&connection_id, "csrf_token").and_then(|e| {
e.set_password(&csrf_token)
.map_err(|e| Error::KeyringError(e.to_string()))
})?;
keyring_entry(&connection_id, "username").and_then(|e| {
e.set_password(username)
.map_err(|e| Error::KeyringError(e.to_string()))
})?;
keyring_entry(&connection_id, "password").and_then(|e| {
e.set_password(password)
.map_err(|e| Error::KeyringError(e.to_string()))
})?;
keyring_set_password(&connection_id, "ticket", &ticket)?;
keyring_set_password(&connection_id, "csrf_token", &csrf_token)?;
keyring_set_password(&connection_id, "username", username)?;
keyring_set_password(&connection_id, "password", password)?;
// Keep the in-memory session of an already-added connection in sync so
// requests made right after login have auth available.
@@ -1054,7 +1204,12 @@ impl ConnectionManager {
})
}
pub async fn login_with_token(&self, url: &str, token: &str) -> crate::Result<LoginResult> {
pub async fn login_with_token(
&self,
url: &str,
token: &str,
server_type: &str,
) -> crate::Result<LoginResult> {
if url.is_empty() {
return Err(Error::InvalidUrl("URL cannot be empty".to_string()));
}
@@ -1066,9 +1221,16 @@ impl ConnectionManager {
let client = build_client()?;
// Validate the token by making an authenticated request
let test_url = format!("{}/api2/json/cluster/status", url);
let auth_header = format!("PVEAPIToken={}", token);
// Validate the token by making an authenticated request. PVE exposes
// `/cluster/status`; PBS (which has no cluster concept) validates
// against `/version` instead. The header prefix follows the server
// type (`PVEAPIToken` vs `PBSAPIToken`).
let server_type = ServerType::from_config(server_type);
let test_url = match server_type {
ServerType::Pbs => format!("{}/api2/json/version", url),
ServerType::Pve => format!("{}/api2/json/cluster/status", url),
};
let auth_header = format!("{}={}", server_type.token_header_name(), token);
let response = client
.get(&test_url)
@@ -1097,10 +1259,7 @@ impl ConnectionManager {
};
// Store token in keyring for later use
keyring_entry(&connection_id, "token").and_then(|e| {
e.set_password(token)
.map_err(|e| Error::KeyringError(e.to_string()))
})?;
keyring_set_password(&connection_id, "token", token)?;
Ok(LoginResult {
connection_id,
@@ -1110,26 +1269,11 @@ impl ConnectionManager {
}
pub async fn logout(&self, connection_id: &str) -> crate::Result<()> {
let _ = keyring_entry(connection_id, "ticket").and_then(|e| {
e.delete_credential()
.map_err(|e| Error::KeyringError(e.to_string()))
});
let _ = keyring_entry(connection_id, "csrf_token").and_then(|e| {
e.delete_credential()
.map_err(|e| Error::KeyringError(e.to_string()))
});
let _ = keyring_entry(connection_id, "password").and_then(|e| {
e.delete_credential()
.map_err(|e| Error::KeyringError(e.to_string()))
});
let _ = keyring_entry(connection_id, "token").and_then(|e| {
e.delete_credential()
.map_err(|e| Error::KeyringError(e.to_string()))
});
let _ = keyring_entry(connection_id, "username").and_then(|e| {
e.delete_credential()
.map_err(|e| Error::KeyringError(e.to_string()))
});
let _ = keyring_delete_credential(connection_id, "ticket");
let _ = keyring_delete_credential(connection_id, "csrf_token");
let _ = keyring_delete_credential(connection_id, "password");
let _ = keyring_delete_credential(connection_id, "token");
let _ = keyring_delete_credential(connection_id, "username");
Ok(())
}
@@ -1144,7 +1288,7 @@ impl ConnectionManager {
match entry.get_password() {
Ok(ticket) => Ok(Some(ticket)),
Err(keyring::Error::NoEntry) => Ok(None),
Err(e) => Err(Error::KeyringError(e.to_string())),
Err(e) => Err(Error::KeyringError(keyring_describe_error(&e))),
}
}
@@ -1156,25 +1300,13 @@ impl ConnectionManager {
password: Option<&str>,
api_token: Option<&str>,
) -> crate::Result<()> {
keyring_entry(connection_id, "ticket").and_then(|e| {
e.set_password(ticket)
.map_err(|e| Error::KeyringError(e.to_string()))
})?;
keyring_entry(connection_id, "csrf_token").and_then(|e| {
e.set_password(csrf_token)
.map_err(|e| Error::KeyringError(e.to_string()))
})?;
keyring_set_password(connection_id, "ticket", ticket)?;
keyring_set_password(connection_id, "csrf_token", csrf_token)?;
if let Some(pw) = password {
keyring_entry(connection_id, "password").and_then(|e| {
e.set_password(pw)
.map_err(|e| Error::KeyringError(e.to_string()))
})?;
keyring_set_password(connection_id, "password", pw)?;
}
if let Some(tok) = api_token {
keyring_entry(connection_id, "token").and_then(|e| {
e.set_password(tok)
.map_err(|e| Error::KeyringError(e.to_string()))
})?;
keyring_set_password(connection_id, "token", tok)?;
}
Ok(())
}
@@ -1215,7 +1347,7 @@ impl ConnectionManager {
Err(keyring::Error::NoEntry) => Err(Error::AuthError(
"No stored credentials for re-authentication".to_string(),
)),
Err(e) => Err(Error::KeyringError(e.to_string())),
Err(e) => Err(Error::KeyringError(keyring_describe_error(&e))),
},
Err(_) => Err(Error::AuthError(
"No stored credentials for re-authentication".to_string(),
@@ -1244,7 +1376,7 @@ impl ConnectionManager {
/// Looks up a connection by id, mapping a missing id to
/// [`Error::ConnectionNotFound`].
fn connection(&self, id: &str) -> crate::Result<&Connection> {
pub(crate) fn connection(&self, id: &str) -> crate::Result<&Connection> {
self.connections
.get(id)
.ok_or_else(|| Error::ConnectionNotFound(id.to_string()))
@@ -1266,6 +1398,43 @@ impl ConnectionManager {
Ok(conn.config.clone())
}
/// Resolves the auth header to send alongside a WebSocket handshake for
/// `connection_id`, using the same secret resolution as `auth_context()`
/// (token from config/keyring; ticket from session/keyring).
///
/// Returns `Some((header_name, header_value))` when a secret is available
/// — `Authorization: {PVE|PBS}APIToken=...` for token mode, or
/// `Cookie: {PVE|PBS}AuthCookie=...` for password mode — and `None` when
/// no secret can be resolved.
pub fn auth_header_for(&self, connection_id: &str) -> crate::Result<Option<(String, String)>> {
let conn = self.connection(connection_id)?;
let auth = conn.auth_context()?;
match auth.mode {
AuthMode::Token => {
Ok(auth
.token
.as_deref()
.filter(|token| !token.is_empty())
.map(|token| {
(
"Authorization".to_string(),
format!("{}={}", auth.server_type.token_header_name(), token),
)
}))
}
AuthMode::Password => Ok(auth
.ticket
.as_deref()
.filter(|ticket| !ticket.is_empty())
.map(|ticket| {
(
"Cookie".to_string(),
format!("{}={}", auth.server_type.cookie_header_name(), ticket),
)
})),
}
}
/// Validates that `vm_type` is one of the Proxmox VM type path segments
/// (`qemu` for VMs, `lxc` for containers).
fn validate_vm_type(vm_type: &str) -> crate::Result<()> {
src-tauri/src/console_proxy.rs
+4 -2
View File
@@ -103,8 +103,10 @@ impl ConsoleProxyManager {
let ws_url = format!("{}{}", origin, path);
// Connect to Proxmox first so an unreachable server fails the command
// instead of leaving a dangling local listener.
let server = timeout(Duration::from_secs(20), connect_ws(&ws_url))
// instead of leaving a dangling local listener. The console websocket
// authenticates via the ticket in the URL query, so no auth header is
// sent.
let server = timeout(Duration::from_secs(20), connect_ws(&ws_url, None))
.await
.map_err(|_| Error::WebSocketError("Timed out opening console proxy".to_string()))?
.map_err(|e| Error::WebSocketError(format!("Cannot open console proxy: {}", e)))?;
src-tauri/src/error.rs
+3
View File
@@ -41,6 +41,9 @@ pub enum Error {
#[error("Tauri error: {0}")]
TauriError(#[from] tauri::Error),
#[error("I/O error: {0}")]
IoError(#[from] std::io::Error),
}
impl Serialize for Error {
src-tauri/src/keyring.rs
+219
View File
@@ -0,0 +1,219 @@
//! OS credential-store access for connection secrets, hardened against the
//! macOS Keychain's find-then-add behavior.
//!
//! Secrets live in the platform secure store: the macOS Keychain, the Windows
//! Credential Manager, or the Linux kernel keyutils (which needs no daemon and
//! therefore works headless). The store is selected once, on first use, via
//! [`keyring_core::set_default_store`].
//!
//! # macOS duplicate-item recovery
//!
//! The Keychain backend behind `keyring`/`security-framework` implements
//! "set" as *find then add*: the scoped lookup against the login keychain is
//! attempted first, and when it fails (for any reason) a new item is added.
//! If the login keychain is locked, the lookup fails, macOS prompts to unlock
//! it, and the subsequent add discovers the item created by an earlier login
//! still exists, returning `errSecDuplicateItem`. [`set_password`] detects
//! that OSStatus and recovers by deleting the stale item and retrying once,
//! so re-logins and ticket refreshes succeed after the keychain is unlocked.
//! Failures that persist are mapped to actionable messages via
//! [`describe_error`].
use std::sync::OnceLock;
pub use keyring_core as keyring;
/// The Keychain/Credential Manager service name under which all Clustri
/// secrets are stored. Each entry's account is `"{connection_id}:{field}"`.
pub const SERVICE: &str = "clustri";
/// Initializes the platform credential store, exactly once, before the first
/// entry is created. Subsequent calls return the cached outcome.
fn init_store() -> keyring::Result<()> {
static INIT: OnceLock<Result<(), String>> = OnceLock::new();
let cached = INIT.get_or_init(|| {
#[cfg(target_os = "macos")]
let store = apple_native_keyring_store::keychain::Store::new();
#[cfg(target_os = "windows")]
let store = windows_native_keyring_store::Store::new();
#[cfg(target_os = "linux")]
let store = linux_keyutils_keyring_store::Store::new();
#[cfg(any(
target_os = "macos",
target_os = "windows",
target_os = "linux"
))]
match store {
Ok(s) => {
keyring_core::set_default_store(s);
Ok(())
}
Err(e) => Err(e.to_string()),
}
#[cfg(not(any(
target_os = "macos",
target_os = "windows",
target_os = "linux"
)))]
Err("no keyring store is configured for this platform".to_string())
});
match cached {
Ok(()) => Ok(()),
Err(message) => Err(keyring::Error::Invalid(
"store".to_string(),
message.clone(),
)),
}
}
/// Returns the keyring entry for a connection's field, initializing the
/// platform store on first use. Entry construction failures are mapped to
/// [`crate::Error::KeyringError`] so callers that tolerate a missing store
/// (e.g. headless Linux) can treat them as `None`.
pub fn entry(connection_id: &str, field: &str) -> crate::Result<keyring::Entry> {
init_store().map_err(|e| crate::Error::KeyringError(describe_error(&e)))?;
let key = format!("{}:{}", connection_id, field);
keyring::Entry::new(SERVICE, &key).map_err(|e| crate::Error::KeyringError(describe_error(&e)))
}
/// Writes a secret for a connection field, recovering from the macOS
/// "item already exists" failure by deleting the stale item and retrying once.
pub fn set_password(connection_id: &str, field: &str, value: &str) -> crate::Result<()> {
let entry = entry(connection_id, field)?;
match entry.set_password(value) {
Ok(()) => Ok(()),
Err(keyring::Error::PlatformFailure(inner)) if is_duplicate_item(inner.as_ref()) => {
// The item exists but the pre-add lookup could not see it (the
// login keychain was locked during the lookup). Once the user has
// unlocked the keychain, deleting and re-adding succeeds.
let _ = entry.delete_credential();
entry
.set_password(value)
.map_err(|e| crate::Error::KeyringError(describe_error(&e)))
}
Err(e) => Err(crate::Error::KeyringError(describe_error(&e))),
}
}
/// Deletes a stored secret for a connection field. A missing entry is not an
/// error, so clearing credentials is idempotent.
pub fn delete_credential(connection_id: &str, field: &str) -> crate::Result<()> {
let entry = entry(connection_id, field)?;
match entry.delete_credential() {
Ok(()) => Ok(()),
Err(keyring::Error::NoEntry) => Ok(()),
Err(e) => Err(crate::Error::KeyringError(describe_error(&e))),
}
}
/// Maps a keyring error to a message with actionable guidance. On macOS the
/// underlying OSStatus is inspected for the known failure modes (duplicate
/// item, locked keychain, missing entitlement); everything else keeps the
/// platform detail so it can be reported or debugged.
pub fn describe_error(err: &keyring::Error) -> String {
match err {
keyring::Error::PlatformFailure(inner) => {
#[cfg(target_os = "macos")]
{
if let Some(sec_err) = inner.downcast_ref::<security_framework::base::Error>() {
return match sec_err.code() {
// errSecDuplicateItem: the item exists but could not be
// located for update (locked login keychain during the
// lookup, or a stray item in another keychain).
-25299 => format!(
"macOS Keychain: the stored item already exists and could not \
be replaced. Your login keychain is likely locked or its \
password is out of date. Open Keychain Access, unlock the \
'login' keychain (or use Edit > Change Password for Keychain \
'login'), then try again. If a duplicate '{SERVICE}' item is \
listed under the iCloud keychain, delete it there as well. \
({inner})"
),
// errSecAuthFailed: the keychain did not unlock.
-25293 => format!(
"macOS Keychain: the login keychain is locked or the password \
is incorrect. Unlock it in Keychain Access and try again. \
({inner})"
),
// errSecMissingEntitlement: unsigned app access denial.
-34018 => format!(
"macOS Keychain: access was denied because the app is not \
signed with keychain entitlements. ({inner})"
),
_ => format!("macOS Keychain error: {inner}"),
};
}
}
format!("Secure storage error: {inner}")
}
keyring::Error::NoStorageAccess(inner) => format!(
"Secure storage is locked or unavailable: {inner}. Unlock your login keychain \
(or keyring) and try again."
),
_ => err.to_string(),
}
}
/// True when a platform error is the macOS `errSecDuplicateItem` code, which
/// the login keychain reports when an add hits an item that already exists.
fn is_duplicate_item(err: &(dyn std::error::Error + Send + Sync)) -> bool {
#[cfg(target_os = "macos")]
{
if let Some(sec_err) = err.downcast_ref::<security_framework::base::Error>() {
return sec_err.code() == -25299;
}
}
let _ = err;
false
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn describe_error_explains_locked_storage() {
let err = keyring::Error::NoStorageAccess(Box::new(std::io::Error::other("locked")));
let msg = describe_error(&err);
assert!(msg.contains("locked or unavailable"), "{msg}");
assert!(msg.contains("Unlock"), "{msg}");
}
#[test]
fn describe_error_keeps_platform_detail() {
let err = keyring::Error::PlatformFailure(Box::new(std::io::Error::other("boom")));
let msg = describe_error(&err);
assert!(msg.contains("Secure storage error"), "{msg}");
assert!(msg.contains("boom"), "{msg}");
}
#[test]
fn describe_error_passes_other_variants_through() {
let err = keyring::Error::Invalid("service".to_string(), "cannot be empty".to_string());
assert_eq!(describe_error(&err), err.to_string());
}
#[test]
fn set_get_delete_round_trip() {
let id = uuid::Uuid::new_v4().to_string();
let field = "unit";
let value = "round-trip-secret";
set_password(&id, field, value).expect("set_password should succeed");
let entry = entry(&id, field).expect("entry should be constructible");
assert_eq!(
entry.get_password().expect("get_password should succeed"),
value
);
delete_credential(&id, field).expect("delete_credential should succeed");
match entry.get_password() {
Err(keyring::Error::NoEntry) => {}
other => panic!("expected NoEntry after delete, got {other:?}"),
}
}
#[test]
fn delete_credential_is_idempotent() {
let id = uuid::Uuid::new_v4().to_string();
delete_credential(&id, "unit").expect("deleting a missing entry should be a no-op");
}
}
src-tauri/src/lib.rs
+276 -4
View File
@@ -9,15 +9,20 @@ use tokio::sync::RwLock;
mod connection;
mod console_proxy;
mod error;
mod keyring;
mod pbs;
mod proxmox;
pub mod tls;
mod websocket;
pub use connection::{
api_request, derive_node_url, AuthContext, AuthMode, ConnectionManager, LoadResult,
api_request, derive_node_url, AuthContext, AuthMode, ConnectionManager, LoadResult, ServerType,
};
pub use console_proxy::ConsoleProxyInfo;
pub use error::Error;
pub use pbs::{
PbsBackupGroup, PbsDatastore, PbsJob, PbsNodeStatus, PbsSnapshot, PbsSnapshotFile, PbsVersion,
};
pub use proxmox::{
AddDiskConfig, AddNICConfig, BackupJobConfig, CreateSnapshotConfig, EditNICConfig,
RestoreConfig, UpdateVMConfig,
@@ -48,6 +53,23 @@ pub struct ConnectionConfig {
pub nodes: Vec<DiscoveredNode>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cluster_id: Option<String>,
#[serde(default = "default_server_type")]
pub server_type: String, // "pve" | "pbs"
}
/// Default server type for connections added without an explicit value. PVE is
/// the historical behavior and remains the default so older persisted configs
/// (which have no `serverType` field) keep working.
fn default_server_type() -> String {
"pve".to_string()
}
impl ConnectionConfig {
/// True when this connection targets a Proxmox Backup Server (PBS) instead
/// of a Proxmox VE cluster.
pub fn is_pbs(&self) -> bool {
self.server_type == "pbs"
}
}
/// A node discovered in the cluster connected through a
@@ -299,7 +321,14 @@ async fn get_tasks(
connection_id: String,
) -> Result<Vec<proxmox::Task>> {
let manager = state.connection_manager.read().await;
manager.get_tasks(&connection_id).await
// PBS exposes its task list under `/nodes/localhost/tasks` with a
// different payload shape than the PVE `/cluster/tasks` endpoint, so the
// server type selects the backing method.
if manager.is_pbs(&connection_id)? {
manager.pbs_get_tasks(&connection_id).await
} else {
manager.get_tasks(&connection_id).await
}
}
#[tauri::command]
@@ -638,9 +667,10 @@ async fn login_with_token(
state: tauri::State<'_, AppState>,
url: String,
token: String,
server_type: String,
) -> Result<LoginResult> {
let manager = state.connection_manager.read().await;
manager.login_with_token(&url, &token).await
manager.login_with_token(&url, &token, &server_type).await
}
#[tauri::command]
@@ -716,8 +746,18 @@ async fn connect_websocket(
url: String,
app_handle: tauri::AppHandle,
) -> Result<()> {
// Resolve the auth header server-side from the connection's stored
// credentials (token from config/keyring, ticket from session/keyring) so
// the frontend never has to hold secrets. A connection with no resolvable
// secret connects without an auth header.
let auth_header = {
let manager = state.connection_manager.read().await;
manager.auth_header_for(&connection_id).ok().flatten()
};
let mut ws_manager = state.ws_manager.write().await;
ws_manager.connect(connection_id, url, app_handle).await
ws_manager
.connect(connection_id, url, auth_header, app_handle)
.await
}
#[tauri::command]
@@ -847,6 +887,222 @@ async fn delete_backup(
manager.delete_backup(&connection_id, &volid).await
}
// Proxmox Backup Server (PBS) commands
#[tauri::command]
async fn get_pbs_datastores(
state: tauri::State<'_, AppState>,
connection_id: String,
) -> Result<Vec<pbs::PbsDatastore>> {
let manager = state.connection_manager.read().await;
manager.pbs_get_datastores(&connection_id).await
}
#[tauri::command]
async fn get_pbs_version(
state: tauri::State<'_, AppState>,
connection_id: String,
) -> Result<pbs::PbsVersion> {
let manager = state.connection_manager.read().await;
manager.pbs_get_version(&connection_id).await
}
#[tauri::command]
async fn get_pbs_node_status(
state: tauri::State<'_, AppState>,
connection_id: String,
) -> Result<pbs::PbsNodeStatus> {
let manager = state.connection_manager.read().await;
manager.pbs_get_node_status(&connection_id).await
}
#[tauri::command]
async fn get_pbs_groups(
state: tauri::State<'_, AppState>,
connection_id: String,
store: String,
) -> Result<Vec<pbs::PbsBackupGroup>> {
let manager = state.connection_manager.read().await;
manager.pbs_get_groups(&connection_id, &store).await
}
#[tauri::command]
async fn get_pbs_snapshots(
state: tauri::State<'_, AppState>,
connection_id: String,
store: String,
backup_id: String,
backup_type: String,
) -> Result<Vec<pbs::PbsSnapshot>> {
let manager = state.connection_manager.read().await;
manager
.pbs_get_snapshots(&connection_id, &store, &backup_id, &backup_type)
.await
}
#[tauri::command]
async fn get_pbs_snapshot_files(
state: tauri::State<'_, AppState>,
connection_id: String,
store: String,
backup_id: String,
backup_type: String,
backup_time: i64,
) -> Result<Vec<pbs::PbsSnapshotFile>> {
let manager = state.connection_manager.read().await;
manager
.pbs_get_snapshot_files(&connection_id, &store, &backup_id, &backup_type, backup_time)
.await
}
#[tauri::command]
//
// The parameter list is the Tauri invoke IPC contract with the frontend
// (`downloadPbsSnapshotFile` in src/lib/tauri.ts), so the args cannot be
// grouped without changing the frontend call site.
#[allow(clippy::too_many_arguments)]
async fn download_pbs_snapshot_file(
state: tauri::State<'_, AppState>,
connection_id: String,
store: String,
backup_id: String,
backup_type: String,
backup_time: i64,
file_name: String,
decoded: bool,
save_path: String,
) -> Result<String> {
let manager = state.connection_manager.read().await;
manager
.pbs_download_snapshot_file(
&connection_id,
&store,
&backup_id,
&backup_type,
backup_time,
&file_name,
decoded,
&save_path,
)
.await
}
#[tauri::command]
async fn delete_pbs_snapshot(
state: tauri::State<'_, AppState>,
connection_id: String,
store: String,
backup_id: String,
backup_type: String,
backup_time: i64,
) -> Result<()> {
let manager = state.connection_manager.read().await;
manager
.pbs_delete_snapshot(&connection_id, &store, &backup_id, &backup_type, backup_time)
.await
}
#[tauri::command]
async fn delete_pbs_group(
state: tauri::State<'_, AppState>,
connection_id: String,
store: String,
backup_id: String,
backup_type: String,
) -> Result<()> {
let manager = state.connection_manager.read().await;
manager
.pbs_delete_group(&connection_id, &store, &backup_id, &backup_type)
.await
}
#[tauri::command]
async fn run_pbs_verify(
state: tauri::State<'_, AppState>,
connection_id: String,
store: String,
) -> Result<String> {
let manager = state.connection_manager.read().await;
manager.pbs_run_verify(&connection_id, &store).await
}
#[tauri::command]
//
// The parameter list is the Tauri invoke IPC contract with the frontend
// (`runPbsPrune` in src/lib/tauri.ts), so the args cannot be grouped without
// changing the frontend call site.
#[allow(clippy::too_many_arguments)]
async fn run_pbs_prune(
state: tauri::State<'_, AppState>,
connection_id: String,
store: String,
keep_last: Option<u32>,
keep_daily: Option<u32>,
keep_weekly: Option<u32>,
keep_monthly: Option<u32>,
keep_yearly: Option<u32>,
dry_run: bool,
) -> Result<String> {
let manager = state.connection_manager.read().await;
manager
.pbs_run_prune(
&connection_id,
&store,
keep_last,
keep_daily,
keep_weekly,
keep_monthly,
keep_yearly,
dry_run,
)
.await
}
#[tauri::command]
async fn run_pbs_gc(
state: tauri::State<'_, AppState>,
connection_id: String,
store: String,
) -> Result<String> {
let manager = state.connection_manager.read().await;
manager.pbs_run_gc(&connection_id, &store).await
}
#[tauri::command]
async fn get_pbs_verify_jobs(
state: tauri::State<'_, AppState>,
connection_id: String,
store: Option<String>,
) -> Result<Vec<pbs::PbsJob>> {
let manager = state.connection_manager.read().await;
manager
.pbs_get_verify_jobs(&connection_id, store.as_deref())
.await
}
#[tauri::command]
async fn get_pbs_prune_jobs(
state: tauri::State<'_, AppState>,
connection_id: String,
store: Option<String>,
) -> Result<Vec<pbs::PbsJob>> {
let manager = state.connection_manager.read().await;
manager
.pbs_get_prune_jobs(&connection_id, store.as_deref())
.await
}
#[tauri::command]
async fn get_pbs_gc_jobs(
state: tauri::State<'_, AppState>,
connection_id: String,
store: Option<String>,
) -> Result<Vec<pbs::PbsJob>> {
let manager = state.connection_manager.read().await;
manager
.pbs_get_gc_jobs(&connection_id, store.as_deref())
.await
}
#[derive(Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TrayConnectionInfo {
@@ -975,8 +1231,24 @@ pub fn run() {
run_backup,
restore_backup,
delete_backup,
get_pbs_datastores,
get_pbs_version,
get_pbs_node_status,
get_pbs_groups,
get_pbs_snapshots,
get_pbs_snapshot_files,
download_pbs_snapshot_file,
delete_pbs_snapshot,
delete_pbs_group,
run_pbs_verify,
run_pbs_prune,
run_pbs_gc,
get_pbs_verify_jobs,
get_pbs_prune_jobs,
get_pbs_gc_jobs,
update_tray_menu,
])
.plugin(tauri_plugin_dialog::init())
.setup(|app| {
// Build the system tray menu
let show_hide = MenuItemBuilder::new("Show / Hide")
src-tauri/src/pbs.rs
+756
View File
@@ -0,0 +1,756 @@
//! Proxmox Backup Server (PBS) backend.
//!
//! PBS shares the JSON `{data}` envelope and the token/ticket authentication
//! model with Proxmox VE, but it is single-host and exposes its own endpoint
//! set. All datastore operations live here; the HTTP plumbing (auth headers,
//! endpoint rotation, envelope unwrapping, binary downloads) is shared from
//! `crate::connection`.
//!
//! The PBS API reports kebab-case JSON keys. The public structs serialize as
//! camelCase for the frontend, so the ones deserialized straight from a
//! response use `rename_all(serialize = "camelCase", deserialize =
//! "kebab-case")`; the datastore and gc-status structs are assembled from raw
//! kebab-case intermediate structs and only ever serialize to the frontend.
use crate::connection::{parse_api, ConnectionManager};
use crate::proxmox::Task;
use reqwest::Method;
use serde::{Deserialize, Serialize};
use std::path::Path;
// ---------------------------------------------------------------------------
// Public types (frontend-facing camelCase shapes)
// ---------------------------------------------------------------------------
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PbsDatastore {
pub store: String,
#[serde(default)]
pub comment: Option<String>,
#[serde(default)]
pub backend_type: Option<String>,
#[serde(default)]
pub mount_status: Option<String>,
#[serde(default)]
pub maintenance: Option<String>,
#[serde(default)]
pub total: Option<u64>,
#[serde(default)]
pub used: Option<u64>,
#[serde(default)]
pub avail: Option<u64>,
#[serde(default)]
pub error: Option<String>,
#[serde(default)]
pub estimated_full_date: Option<i64>,
#[serde(default)]
pub history: Option<Vec<f64>>,
#[serde(default)]
pub gc_status: Option<PbsGcStatus>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PbsGcStatus {
#[serde(default)]
pub disk_bytes: Option<u64>,
#[serde(default)]
pub disk_chunks: Option<u64>,
#[serde(default)]
pub index_data_bytes: Option<u64>,
#[serde(default)]
pub index_file_count: Option<u64>,
#[serde(default)]
pub pending_bytes: Option<u64>,
#[serde(default)]
pub pending_chunks: Option<u64>,
#[serde(default)]
pub removed_bad: Option<u64>,
#[serde(default)]
pub removed_bytes: Option<u64>,
#[serde(default)]
pub removed_chunks: Option<u64>,
#[serde(default)]
pub still_bad: Option<u64>,
#[serde(default)]
pub cache_hits: Option<u64>,
#[serde(default)]
pub cache_misses: Option<u64>,
#[serde(default)]
pub upid: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PbsVersion {
pub version: String,
pub release: String,
pub repoid: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all(serialize = "camelCase", deserialize = "kebab-case"))]
pub struct PbsNodeStatus {
#[serde(default)]
pub cpu: Option<f64>,
#[serde(default)]
pub loadavg: Option<Vec<f64>>,
#[serde(default)]
pub uptime: Option<u64>,
#[serde(default)]
pub memory: Option<PbsMem>,
#[serde(default)]
pub root: Option<PbsMem>,
#[serde(default)]
pub swap: Option<PbsMem>,
#[serde(default)]
pub cpuinfo: Option<PbsCpuInfo>,
#[serde(default)]
pub current_kernel: Option<PbsKernel>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PbsMem {
#[serde(default)]
pub free: Option<u64>,
#[serde(default)]
pub total: Option<u64>,
#[serde(default)]
pub used: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PbsCpuInfo {
#[serde(default)]
pub cpus: Option<u32>,
#[serde(default)]
pub model: Option<String>,
#[serde(default)]
pub sockets: Option<u32>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PbsKernel {
#[serde(default)]
pub machine: Option<String>,
#[serde(default)]
pub release: Option<String>,
#[serde(default)]
pub sysname: Option<String>,
#[serde(default)]
pub version: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all(serialize = "camelCase", deserialize = "kebab-case"))]
pub struct PbsBackupGroup {
pub backup_id: String,
pub backup_type: String,
#[serde(default)]
pub backup_count: Option<u32>,
#[serde(default)]
pub last_backup: Option<i64>,
#[serde(default)]
pub comment: Option<String>,
#[serde(default)]
pub files: Option<Vec<String>>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all(serialize = "camelCase", deserialize = "kebab-case"))]
pub struct PbsSnapshot {
pub backup_id: String,
pub backup_type: String,
pub backup_time: i64,
#[serde(default)]
pub size: Option<u64>,
#[serde(default)]
pub protected: Option<bool>,
#[serde(default)]
pub comment: Option<String>,
#[serde(default)]
pub files: Option<Vec<String>>,
#[serde(default)]
pub fingerprint: Option<String>,
#[serde(default)]
pub owner: Option<String>,
#[serde(default)]
pub verification: Option<PbsVerification>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PbsVerification {
#[serde(default)]
pub state: Option<String>,
#[serde(default)]
pub upid: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all(serialize = "camelCase", deserialize = "kebab-case"))]
pub struct PbsSnapshotFile {
pub filename: String,
#[serde(default)]
pub size: Option<u64>,
#[serde(default)]
pub crypt_mode: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all(serialize = "camelCase", deserialize = "kebab-case"))]
pub struct PbsJob {
pub id: String,
#[serde(default)]
pub store: Option<String>,
#[serde(default)]
pub schedule: Option<String>,
#[serde(default)]
pub comment: Option<String>,
#[serde(default)]
pub disable: Option<bool>,
#[serde(default)]
pub last_run_state: Option<String>,
#[serde(default)]
pub last_run_endtime: Option<i64>,
#[serde(default)]
pub next_run: Option<i64>,
#[serde(default)]
pub keep_last: Option<u32>,
#[serde(default)]
pub keep_daily: Option<u32>,
#[serde(default)]
pub keep_weekly: Option<u32>,
#[serde(default)]
pub keep_monthly: Option<u32>,
#[serde(default)]
pub keep_yearly: Option<u32>,
#[serde(default)]
pub ignore_verified: Option<bool>,
#[serde(default)]
pub max_depth: Option<u32>,
}
// ---------------------------------------------------------------------------
// Raw kebab-case response shapes (not serialized to the frontend)
// ---------------------------------------------------------------------------
/// Raw `/status/datastore-usage` entry. The usage endpoint reports kebab-case
/// keys and nests the cache stats inside `gc-status.cache-stats`, so the
/// public [`PbsDatastore`]/[`PbsGcStatus`] structs are built from this by
/// hand rather than deserialized directly.
#[derive(Debug, Deserialize)]
#[serde(rename_all = "kebab-case")]
struct PbsDatastoreUsageRaw {
store: String,
#[serde(default)]
backend_type: Option<String>,
#[serde(default)]
mount_status: Option<String>,
#[serde(default)]
avail: Option<u64>,
#[serde(default)]
total: Option<u64>,
#[serde(default)]
used: Option<u64>,
#[serde(default)]
error: Option<String>,
#[serde(default)]
estimated_full_date: Option<i64>,
#[serde(default)]
history: Option<Vec<f64>>,
#[serde(default)]
gc_status: Option<PbsGcStatusRaw>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "kebab-case")]
struct PbsGcStatusRaw {
#[serde(default)]
disk_bytes: Option<u64>,
#[serde(default)]
disk_chunks: Option<u64>,
#[serde(default)]
index_data_bytes: Option<u64>,
#[serde(default)]
index_file_count: Option<u64>,
#[serde(default)]
pending_bytes: Option<u64>,
#[serde(default)]
pending_chunks: Option<u64>,
#[serde(default)]
removed_bad: Option<u64>,
#[serde(default)]
removed_bytes: Option<u64>,
#[serde(default)]
removed_chunks: Option<u64>,
#[serde(default)]
still_bad: Option<u64>,
#[serde(default)]
cache_stats: Option<PbsCacheStatsRaw>,
#[serde(default)]
upid: Option<String>,
}
#[derive(Debug, Deserialize)]
struct PbsCacheStatsRaw {
#[serde(default)]
hits: Option<u64>,
#[serde(default)]
misses: Option<u64>,
}
/// Raw `/admin/datastore` entry carrying the static datastore config that the
/// usage endpoint does not report (comment, maintenance).
#[derive(Debug, Deserialize)]
#[serde(rename_all = "kebab-case")]
struct PbsDatastoreConfigRaw {
store: String,
#[serde(default)]
comment: Option<String>,
#[serde(default)]
backend_type: Option<String>,
#[serde(default)]
mount_status: Option<String>,
#[serde(default)]
maintenance: Option<String>,
}
/// Raw `/admin/gc` job entry. GC jobs carry no `id` — the store identifies the
/// job — so `id` falls back to `store` when mapping to [`PbsJob`].
#[derive(Debug, Deserialize)]
#[serde(rename_all = "kebab-case")]
struct PbsGcJobRaw {
store: String,
#[serde(default)]
schedule: Option<String>,
#[serde(default)]
comment: Option<String>,
#[serde(default)]
disable: Option<bool>,
#[serde(default)]
last_run_state: Option<String>,
#[serde(default)]
last_run_endtime: Option<i64>,
#[serde(default)]
next_run: Option<i64>,
#[serde(default)]
keep_last: Option<u32>,
#[serde(default)]
keep_daily: Option<u32>,
#[serde(default)]
keep_weekly: Option<u32>,
#[serde(default)]
keep_monthly: Option<u32>,
#[serde(default)]
keep_yearly: Option<u32>,
#[serde(default)]
ignore_verified: Option<bool>,
#[serde(default)]
max_depth: Option<u32>,
}
impl From<PbsDatastoreUsageRaw> for PbsDatastore {
fn from(raw: PbsDatastoreUsageRaw) -> Self {
PbsDatastore {
store: raw.store,
comment: None,
backend_type: raw.backend_type,
mount_status: raw.mount_status,
maintenance: None,
total: raw.total,
used: raw.used,
avail: raw.avail,
error: raw.error,
estimated_full_date: raw.estimated_full_date,
history: raw.history,
gc_status: raw.gc_status.map(Into::into),
}
}
}
impl From<PbsGcStatusRaw> for PbsGcStatus {
fn from(raw: PbsGcStatusRaw) -> Self {
PbsGcStatus {
disk_bytes: raw.disk_bytes,
disk_chunks: raw.disk_chunks,
index_data_bytes: raw.index_data_bytes,
index_file_count: raw.index_file_count,
pending_bytes: raw.pending_bytes,
pending_chunks: raw.pending_chunks,
removed_bad: raw.removed_bad,
removed_bytes: raw.removed_bytes,
removed_chunks: raw.removed_chunks,
still_bad: raw.still_bad,
cache_hits: raw.cache_stats.as_ref().and_then(|stats| stats.hits),
cache_misses: raw.cache_stats.as_ref().and_then(|stats| stats.misses),
upid: raw.upid,
}
}
}
impl From<PbsGcJobRaw> for PbsJob {
fn from(raw: PbsGcJobRaw) -> Self {
PbsJob {
id: raw.store.clone(),
store: Some(raw.store),
schedule: raw.schedule,
comment: raw.comment,
disable: raw.disable,
last_run_state: raw.last_run_state,
last_run_endtime: raw.last_run_endtime,
next_run: raw.next_run,
keep_last: raw.keep_last,
keep_daily: raw.keep_daily,
keep_weekly: raw.keep_weekly,
keep_monthly: raw.keep_monthly,
keep_yearly: raw.keep_yearly,
ignore_verified: raw.ignore_verified,
max_depth: raw.max_depth,
}
}
}
// ---------------------------------------------------------------------------
// PBS API methods
// ---------------------------------------------------------------------------
impl ConnectionManager {
/// True when the connection targets a PBS server rather than a PVE
/// cluster.
pub fn is_pbs(&self, connection_id: &str) -> crate::Result<bool> {
Ok(self.connection(connection_id)?.config.server_type == "pbs")
}
/// Lists the datastores with their live usage from `/status/datastore-usage`,
/// then merges the static datastore config (`/admin/datastore`: comment,
/// maintenance, mount status) onto each entry by `store`. A failure of the
/// usage call propagates; a failure of the config call degrades to the
/// usage-only list.
pub async fn pbs_get_datastores(&self, connection_id: &str) -> crate::Result<Vec<PbsDatastore>> {
let conn = self.connection(connection_id)?;
let data = conn
.request(Method::GET, "/status/datastore-usage", &[], None)
.await?;
let usage: Vec<PbsDatastoreUsageRaw> = parse_api("/status/datastore-usage", data)?;
let mut datastores: Vec<PbsDatastore> = usage.into_iter().map(Into::into).collect();
if let Ok(config_data) = conn.request(Method::GET, "/admin/datastore", &[], None).await {
if let Ok(configs) =
parse_api::<Vec<PbsDatastoreConfigRaw>>("/admin/datastore", config_data)
{
for config in configs {
if let Some(datastore) = datastores
.iter_mut()
.find(|datastore| datastore.store == config.store)
{
// Only overwrite with fields the config actually
// reports, so usage-derived values survive an omitted
// key.
if config.comment.is_some() {
datastore.comment = config.comment;
}
if config.backend_type.is_some() {
datastore.backend_type = config.backend_type;
}
if config.mount_status.is_some() {
datastore.mount_status = config.mount_status;
}
if config.maintenance.is_some() {
datastore.maintenance = config.maintenance;
}
}
}
}
}
Ok(datastores)
}
/// Fetches the server version information.
pub async fn pbs_get_version(&self, connection_id: &str) -> crate::Result<PbsVersion> {
let conn = self.connection(connection_id)?;
let data = conn.request(Method::GET, "/version", &[], None).await?;
parse_api("/version", data)
}
/// Fetches the resource usage of the local node. PBS is single-host, so the
/// `localhost` node is always the one being managed.
pub async fn pbs_get_node_status(&self, connection_id: &str) -> crate::Result<PbsNodeStatus> {
let conn = self.connection(connection_id)?;
let data = conn
.request(Method::GET, "/nodes/localhost/status", &[], None)
.await?;
parse_api("/nodes/localhost/status", data)
}
/// Lists the backup groups (per `backup-id`/`backup-type`) of a datastore.
pub async fn pbs_get_groups(
&self,
connection_id: &str,
store: &str,
) -> crate::Result<Vec<PbsBackupGroup>> {
let conn = self.connection(connection_id)?;
let path = format!("/admin/datastore/{}/groups", store);
let data = conn.request(Method::GET, &path, &[], None).await?;
parse_api(&path, data)
}
/// Lists the snapshots of one backup group.
pub async fn pbs_get_snapshots(
&self,
connection_id: &str,
store: &str,
backup_id: &str,
backup_type: &str,
) -> crate::Result<Vec<PbsSnapshot>> {
let conn = self.connection(connection_id)?;
let path = format!("/admin/datastore/{}/snapshots", store);
let query = [
("backup-id", backup_id.to_string()),
("backup-type", backup_type.to_string()),
];
let data = conn.request(Method::GET, &path, &query, None).await?;
parse_api(&path, data)
}
/// Lists the files of one snapshot.
pub async fn pbs_get_snapshot_files(
&self,
connection_id: &str,
store: &str,
backup_id: &str,
backup_type: &str,
backup_time: i64,
) -> crate::Result<Vec<PbsSnapshotFile>> {
let conn = self.connection(connection_id)?;
let path = format!("/admin/datastore/{}/files", store);
let query = [
("backup-id", backup_id.to_string()),
("backup-type", backup_type.to_string()),
("backup-time", backup_time.to_string()),
];
let data = conn.request(Method::GET, &path, &query, None).await?;
parse_api(&path, data)
}
/// Streams a snapshot file to `save_path` and returns the path. `decoded`
/// selects the `download-decoded` endpoint (raw plaintext archive bytes,
/// only available for unencrypted datastores) over the plain `download`
/// endpoint (raw archive bytes, possibly encrypted).
//
// The argument list mirrors the `download_pbs_snapshot_file` Tauri command
// (the invoke IPC contract), so it cannot be grouped without breaking the
// frontend call sites.
#[allow(clippy::too_many_arguments)]
pub async fn pbs_download_snapshot_file(
&self,
connection_id: &str,
store: &str,
backup_id: &str,
backup_type: &str,
backup_time: i64,
file_name: &str,
decoded: bool,
save_path: &str,
) -> crate::Result<String> {
let conn = self.connection(connection_id)?;
let endpoint = if decoded { "download-decoded" } else { "download" };
let path = format!("/admin/datastore/{}/{}", store, endpoint);
let query = [
("backup-id", backup_id.to_string()),
("backup-type", backup_type.to_string()),
("backup-time", backup_time.to_string()),
("file-name", file_name.to_string()),
];
conn.download_to_file(&path, &query, Path::new(save_path))
.await?;
Ok(save_path.to_string())
}
/// Deletes a single snapshot.
pub async fn pbs_delete_snapshot(
&self,
connection_id: &str,
store: &str,
backup_id: &str,
backup_type: &str,
backup_time: i64,
) -> crate::Result<()> {
let conn = self.connection(connection_id)?;
let path = format!("/admin/datastore/{}/snapshots", store);
let query = [
("backup-id", backup_id.to_string()),
("backup-type", backup_type.to_string()),
("backup-time", backup_time.to_string()),
];
conn.request(Method::DELETE, &path, &query, None).await?;
Ok(())
}
/// Deletes a whole backup group (all of its snapshots).
pub async fn pbs_delete_group(
&self,
connection_id: &str,
store: &str,
backup_id: &str,
backup_type: &str,
) -> crate::Result<()> {
let conn = self.connection(connection_id)?;
let path = format!("/admin/datastore/{}/groups", store);
let query = [
("backup-id", backup_id.to_string()),
("backup-type", backup_type.to_string()),
];
conn.request(Method::DELETE, &path, &query, None).await?;
Ok(())
}
/// Starts a verification task for a datastore and returns the UPID.
pub async fn pbs_run_verify(&self, connection_id: &str, store: &str) -> crate::Result<String> {
let conn = self.connection(connection_id)?;
let path = format!("/admin/datastore/{}/verify", store);
let form = [("store", store.to_string())];
let data = conn.request(Method::POST, &path, &[], Some(&form)).await?;
parse_api(&path, data)
}
/// Starts a prune task for a datastore and returns the UPID. Only the
/// provided keep-* retention fields are sent; `dry_run` marks the run as
/// a simulation.
//
// The argument list mirrors the `run_pbs_prune` Tauri command (the invoke
// IPC contract), so it cannot be grouped without breaking the frontend
// call sites.
#[allow(clippy::too_many_arguments)]
pub async fn pbs_run_prune(
&self,
connection_id: &str,
store: &str,
keep_last: Option<u32>,
keep_daily: Option<u32>,
keep_weekly: Option<u32>,
keep_monthly: Option<u32>,
keep_yearly: Option<u32>,
dry_run: bool,
) -> crate::Result<String> {
let conn = self.connection(connection_id)?;
let path = format!("/admin/datastore/{}/prune-datastore", store);
let mut form: Vec<(&str, String)> = vec![("store", store.to_string())];
if let Some(keep) = keep_last {
form.push(("keep-last", keep.to_string()));
}
if let Some(keep) = keep_daily {
form.push(("keep-daily", keep.to_string()));
}
if let Some(keep) = keep_weekly {
form.push(("keep-weekly", keep.to_string()));
}
if let Some(keep) = keep_monthly {
form.push(("keep-monthly", keep.to_string()));
}
if let Some(keep) = keep_yearly {
form.push(("keep-yearly", keep.to_string()));
}
if dry_run {
form.push(("dry-run", "1".to_string()));
}
let data = conn.request(Method::POST, &path, &[], Some(&form)).await?;
parse_api(&path, data)
}
/// Starts a garbage-collection task for a datastore and returns the UPID.
pub async fn pbs_run_gc(&self, connection_id: &str, store: &str) -> crate::Result<String> {
let conn = self.connection(connection_id)?;
let path = format!("/admin/datastore/{}/gc", store);
let form = [("store", store.to_string())];
let data = conn.request(Method::POST, &path, &[], Some(&form)).await?;
parse_api(&path, data)
}
/// Lists the verification jobs, optionally filtered by datastore.
pub async fn pbs_get_verify_jobs(
&self,
connection_id: &str,
store: Option<&str>,
) -> crate::Result<Vec<PbsJob>> {
let conn = self.connection(connection_id)?;
let query = store
.map(|store| vec![("store", store.to_string())])
.unwrap_or_default();
let data = conn.request(Method::GET, "/admin/verify", &query, None).await?;
parse_api("/admin/verify", data)
}
/// Lists the prune jobs, optionally filtered by datastore.
pub async fn pbs_get_prune_jobs(
&self,
connection_id: &str,
store: Option<&str>,
) -> crate::Result<Vec<PbsJob>> {
let conn = self.connection(connection_id)?;
let query = store
.map(|store| vec![("store", store.to_string())])
.unwrap_or_default();
let data = conn.request(Method::GET, "/admin/prune", &query, None).await?;
parse_api("/admin/prune", data)
}
/// Lists the garbage-collection jobs, optionally filtered by datastore.
/// GC jobs carry no `id` of their own, so the datastore name is used.
pub async fn pbs_get_gc_jobs(
&self,
connection_id: &str,
store: Option<&str>,
) -> crate::Result<Vec<PbsJob>> {
let conn = self.connection(connection_id)?;
let query = store
.map(|store| vec![("store", store.to_string())])
.unwrap_or_default();
let data = conn.request(Method::GET, "/admin/gc", &query, None).await?;
let raw: Vec<PbsGcJobRaw> = parse_api("/admin/gc", data)?;
Ok(raw.into_iter().map(PbsJob::from).collect())
}
/// Lists the tasks running on the local PBS node, mapping the kebab-case
/// `worker-type`/`worker-id` keys and the PBS task status codes onto the
/// shared [`Task`] shape (the same struct the PVE task list uses).
pub async fn pbs_get_tasks(&self, connection_id: &str) -> crate::Result<Vec<Task>> {
let conn = self.connection(connection_id)?;
let data = conn
.request(Method::GET, "/nodes/localhost/tasks", &[], None)
.await?;
let entries: Vec<serde_json::Value> = parse_api("/nodes/localhost/tasks", data)?;
let mut tasks = Vec::with_capacity(entries.len());
for entry in entries {
// PBS reports `running` while a task is active and `ok` /
// `warning` / `error` once it finished. `ok` is mapped onto the
// PVE-style `exitstatus` so the shared task row renders the same
// way for both platforms.
let (status, exitstatus) = match entry["status"].as_str() {
Some("running") => (Some("running".to_string()), None),
Some("ok") => (None, Some("OK".to_string())),
Some("warning") => (None, Some("WARNING".to_string())),
Some("error") => (None, Some("ERROR".to_string())),
_ => (None, None),
};
tasks.push(Task {
upid: entry["upid"].as_str().unwrap_or("").to_string(),
node: entry["node"].as_str().unwrap_or("").to_string(),
pid: entry["pid"].as_u64().unwrap_or(0) as u32,
pstart: entry["pstart"].as_u64().unwrap_or(0),
starttime: entry["starttime"].as_u64().unwrap_or(0),
endtime: entry["endtime"].as_u64(),
r#type: entry["worker-type"].as_str().unwrap_or("").to_string(),
id: entry["worker-id"].as_str().unwrap_or("").to_string(),
user: entry["user"].as_str().unwrap_or("").to_string(),
status,
exitstatus,
});
}
Ok(tasks)
}
}
src-tauri/src/websocket.rs
+20 -4
View File
@@ -17,13 +17,26 @@ use crate::error::Error;
/// rejects the self-signed certificates typical of home-lab Proxmox servers.
/// As with the HTTP transport, the application enforces trust at its own layer
/// (TOFU pinning in `tls.rs`), so the transport here accepts any certificate.
///
/// When `auth_header` is `Some((name, value))` the header is attached to the
/// handshake request (e.g. `("Cookie", "PVEAuthCookie=...")` or
/// `("Authorization", "PVEAPIToken=...")`). Unparseable header names/values
/// are silently ignored so a bad secret never breaks the connection outright.
pub async fn connect_ws(
url: &str,
auth_header: Option<(String, String)>,
) -> crate::Result<tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>>
{
let request = url
let mut request = url
.into_client_request()
.map_err(|e| Error::WebSocketError(e.to_string()))?;
if let Some((name, value)) = auth_header {
if let Ok(name) = http::HeaderName::from_bytes(name.as_bytes()) {
if let Ok(value) = http::HeaderValue::from_str(&value) {
request.headers_mut().insert(name, value);
}
}
}
let is_wss = request.uri().scheme_str() == Some("wss");
let connector = if is_wss {
@@ -90,11 +103,13 @@ impl WebSocketManager {
/// Connect to a Proxmox WebSocket URL for the given connection ID.
///
/// Messages are forwarded as Tauri events via `app_handle`. If a connection
/// already exists for this ID, it is disconnected first.
/// already exists for this ID, it is disconnected first. `auth_header`
/// (when present) is attached to every connection attempt's handshake.
pub async fn connect(
&mut self,
connection_id: String,
url: String,
auth_header: Option<(String, String)>,
app_handle: tauri::AppHandle,
) -> crate::Result<()> {
// Disconnect any existing connection for this ID
@@ -116,7 +131,7 @@ impl WebSocketManager {
_ = shutdown_rx.recv() => {
break;
}
result = connect_and_run(&cid, &ws_url, &app_handle) => {
result = connect_and_run(&cid, &ws_url, auth_header.clone(), &app_handle) => {
match result {
Ok(()) => {
reconnect_delay = Duration::from_secs(1);
@@ -170,9 +185,10 @@ impl WebSocketManager {
async fn connect_and_run(
connection_id: &str,
url: &str,
auth_header: Option<(String, String)>,
app_handle: &tauri::AppHandle,
) -> crate::Result<()> {
let ws_stream = connect_ws(url).await?;
let ws_stream = connect_ws(url, auth_header).await?;
let cid = connection_id.to_string();
let (mut write, mut read) = ws_stream.split();