Vydané zdroje
Stáhnout zdroje ZIP
CH-J Tools
Procházejte adresáře a soubory konkrétního vydání aplikace.
CH-J / Server Manager
872 souborů
core-20261030T220507Z-61e4558d889f
Přidáno: · Sestaveno:
1
//! Central SFTP Transfer Manager (port of `files/transferManager.js`).2
//!3
//! Every file is one job with its own SFTP channel, a bounded pipeline of positioned4
//! reads/writes and a confirmed offset that only advances over contiguous acknowledged chunks.5
//! Pause and cancel stop issuing requests; resume reopens the partial file, verifies the tail6
//! against the source and continues from that offset. Completed uploads are published by the7
//! shared remote helper (verified hash, original metadata, atomic rename, privileged when8
//! needed); completed downloads are hashed locally and renamed into place. Job state (never9
//! secrets) survives restarts in `transfers.json`, in the same format as 0.1.0, so transfers10
//! interrupted by the update can be resumed.12
use super::editor::{remote_checksum, Authorization};13
use super::pipeline::{pump, verify_tail, Consumer, LocalFile, Producer, PumpStop, TAIL_CHECK};14
use super::posix;15
use super::service::{file_error, normalize_remote_path, or_root, safe_entry_name, RemoteFileService};16
use crate::core::error::{CoreError, CoreResult};17
use crate::core::logging::Logger;18
use crate::core::ssh::session::SessionManager;19
use crate::core::ssh::sftp::{FileHandle, Sftp, SftpError, Stat};20
use futures::future::BoxFuture;21
use futures::{FutureExt, StreamExt};22
use parking_lot::Mutex;23
use rand::RngCore;24
use serde::{Deserialize, Deserializer, Serialize};25
use serde_json::{json, Map, Value};26
use sha2::{Digest, Sha256};27
use std::collections::{HashMap, HashSet};28
use std::path::{Path, PathBuf};29
use std::sync::atomic::{AtomicBool, Ordering};30
use std::sync::Arc;31
use std::time::Duration;32
use tokio::sync::watch;33
use zeroize::Zeroizing;35
const STATES: [&str; 11] = ["waiting", "connecting", "transferring", "paused", "interrupted", "reconnecting", "verifying", "finalizing", "completed", "failed", "cancelled"];36
const RUNNING: [&str; 4] = ["connecting", "transferring", "verifying", "finalizing"];37
const FINAL: [&str; 3] = ["completed", "failed", "cancelled"];38
const EMIT_MS: u64 = 250;39
const PERSIST_MS: u64 = 1000;40
const RETRY_DELAYS_MS: [u64; 5] = [2000, 4000, 8000, 16000, 32000];41
const MAX_TOTAL_RETRIES: u32 = 50;42
const MAX_HISTORY: usize = 500;43
const MAX_JOBS: usize = 20000;44
const MAX_PLAN_ENTRIES: usize = 10000;45
const MAX_PLAN_BYTES: u64 = 4 * 1024 * 1024 * 1024 * 1024;46
const PLAN_TTL_MS: i64 = 10 * 60 * 1000;47
const ORPHAN_AGE_MS: i64 = 24 * 3600 * 1000;48
const HASH_RATE: u64 = 10 * 1024 * 1024;49
const CONNECTION_CODES: [&str; 9] = ["SESSION_NOT_CONNECTED", "SFTP_OPEN_FAILED", "TRANSFER_STALLED", "REMOTE_COMMAND_TIMEOUT", "REMOTE_COMMAND_FAILED", "FILE_EXEC_FAILED", "ECONNRESET", "ETIMEDOUT", "EPIPE"];50
const NO_HELPER_CODES: [&str; 2] = ["FILE_DEPENDENCY_UNAVAILABLE", "FEATURE_PLATFORM_UNSUPPORTED"];51
const AUTH_CODES: [&str; 4] = ["SUDO_PASSWORD_REQUIRED", "SUDO_PASSWORD_INCORRECT", "FILE_PERMISSION_DENIED", "FILE_METADATA_PERMISSION"];53
fn now() -> i64 {54
chrono::Utc::now().timestamp_millis()55
}57
fn hex_id(bytes: usize) -> String {58
let mut buffer = vec![0u8; bytes];59
rand::rngs::OsRng.fill_bytes(&mut buffer);60
hex::encode(buffer)61
}63
fn part_name(name: &str, id: &str) -> String {64
format!(".{}.chj-part-{id}", name.chars().take(120).collect::<String>())65
}67
fn hash_timeout(size: u64) -> u64 {68
120_000 + (size as f64 / HASH_RATE as f64 * 1000.0).ceil() as u6469
}71
// --- lenient JSON numbers (0.1.0 wrote JavaScript numbers) ------------------------------------------73
fn number(value: &Value) -> Option<f64> {74
match value {75
Value::Number(n) => n.as_f64(),76
Value::String(s) => s.parse().ok(),77
_ => None,78
}79
}81
fn de_u64<'de, D: Deserializer<'de>>(d: D) -> Result<u64, D::Error> {82
Ok(number(&Value::deserialize(d)?).filter(|n| n.is_finite() && *n >= 0.0).map(|n| n as u64).unwrap_or(0))83
}84
fn de_i64<'de, D: Deserializer<'de>>(d: D) -> Result<i64, D::Error> {85
Ok(number(&Value::deserialize(d)?).filter(|n| n.is_finite()).map(|n| n as i64).unwrap_or(0))86
}87
fn de_opt_i64<'de, D: Deserializer<'de>>(d: D) -> Result<Option<i64>, D::Error> {88
Ok(number(&Value::deserialize(d)?).filter(|n| n.is_finite()).map(|n| n as i64))89
}90
fn de_u32<'de, D: Deserializer<'de>>(d: D) -> Result<u32, D::Error> {91
Ok(number(&Value::deserialize(d)?).filter(|n| n.is_finite() && *n >= 0.0).map(|n| n as u32).unwrap_or(0))92
}93
/// Keeps JSON `null` as `Some(Null)` (0.1.0 distinguishes `undefined` from `null`).94
fn de_present<'de, D: Deserializer<'de>>(d: D) -> Result<Option<Value>, D::Error> {95
Ok(Some(Value::deserialize(d)?))96
}97
fn de_opt_string<'de, D: Deserializer<'de>>(d: D) -> Result<Option<String>, D::Error> {98
Ok(match Value::deserialize(d)? {99
Value::String(s) => Some(s),100
_ => None,101
})102
}104
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]105
#[serde(rename_all = "camelCase")]106
pub struct SessionIdentity {107
#[serde(default)]108
pub profile_id: Value,109
#[serde(default)]110
pub host: Value,111
#[serde(default)]112
pub port: Value,113
#[serde(default)]114
pub username: Value,115
#[serde(default)]116
pub jump: Vec<Value>,117
}119
impl SessionIdentity {120
fn key(&self) -> String {121
let profile = self.profile_id.as_str().filter(|id| !id.starts_with("adhoc:")).unwrap_or("");122
serde_json::to_string(&json!([profile, self.host, self.port, self.username, self.jump])).unwrap_or_default()123
}124
fn of_status(status: &Value) -> SessionIdentity {125
SessionIdentity {126
profile_id: status["profileId"].clone(), host: status["host"].clone(), port: status["port"].clone(), username: status["username"].clone(),127
jump: status["jump"].as_array().map(|hops| hops.iter().map(|hop| json!({ "host": hop["host"], "port": hop["port"], "username": hop["username"] })).collect()).unwrap_or_default(),128
}129
}130
}132
fn matching_session(sessions: &[Value], identity: &SessionIdentity, preferred: &str) -> Option<String> {133
sessions.iter().find(|s| s["state"] == "connected" && s["sessionId"] == preferred && SessionIdentity::of_status(s) == *identity)134
.and_then(|s| s["sessionId"].as_str().map(str::to_string))135
}137
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]138
#[serde(rename_all = "camelCase")]139
pub struct LocalIdentity {140
#[serde(default, deserialize_with = "de_u64")]141
pub size: u64,142
#[serde(default, deserialize_with = "de_i64")]143
pub mtime_ms: i64,144
#[serde(default, deserialize_with = "de_u64")]145
pub ino: u64,146
#[serde(default, deserialize_with = "de_u64")]147
pub dev: u64,148
}150
impl LocalIdentity {151
pub fn of(meta: &std::fs::Metadata) -> LocalIdentity {152
let mtime_ms = meta153
.modified()154
.ok()155
.map(|t| match t.duration_since(std::time::UNIX_EPOCH) {156
Ok(d) => d.as_millis() as i64,157
Err(e) => -(e.duration().as_millis() as i64),158
})159
.unwrap_or(0);160
#[cfg(unix)]161
let (ino, dev) = {162
use std::os::unix::fs::MetadataExt;163
(meta.ino(), meta.dev())164
};165
// Stable Rust exposes no file index on Windows; size + mtime identify the source.166
#[cfg(not(unix))]167
let (ino, dev) = (0, 0);168
LocalIdentity { size: meta.len(), mtime_ms, ino, dev }169
}171
fn same(&self, other: &LocalIdentity) -> bool {172
let loose = cfg!(not(unix)) || self.ino == 0 || other.ino == 0;173
self.size == other.size && self.mtime_ms == other.mtime_ms && (loose || (self.ino == other.ino && self.dev == other.dev))174
}175
}177
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]178
pub struct RemoteIdentity {179
#[serde(default, deserialize_with = "de_u64")]180
pub size: u64,181
#[serde(default, deserialize_with = "de_u64")]182
pub mtime: u64,183
}185
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]186
pub struct JobError {187
#[serde(default)]188
pub code: String,189
#[serde(default)]190
pub message: String,191
}193
#[derive(Debug, Clone, Serialize, Deserialize)]194
#[serde(rename_all = "camelCase")]195
pub struct Job {196
pub id: String,197
#[serde(default)]198
pub owner: String,199
#[serde(default)]200
pub batch_id: String,201
#[serde(default)]202
pub direction: String,203
#[serde(default)]204
pub identity: SessionIdentity,205
#[serde(default)]206
pub session_id: Value,207
#[serde(default, deserialize_with = "de_u64")]208
pub order: u64,209
#[serde(default)]210
pub name: String,211
#[serde(default)]212
pub local_path: String,213
#[serde(default)]214
pub remote_path: String,215
#[serde(default, deserialize_with = "de_u64")]216
pub size: u64,217
#[serde(default, deserialize_with = "de_u64")]218
pub transferred: u64,219
#[serde(default, deserialize_with = "de_u64")]220
pub moved_bytes: u64,221
#[serde(default, deserialize_with = "de_u64")]222
pub active_ms: u64,223
pub state: String,224
#[serde(default)]225
pub mode: String,226
#[serde(default)]227
pub overwrite: bool,228
#[serde(default)]229
pub local_identity: Option<LocalIdentity>,230
#[serde(default)]231
pub remote_identity: Option<RemoteIdentity>,232
#[serde(default)]233
pub remote_mode: Option<Value>,234
#[serde(default)]235
pub finalize: Value,236
#[serde(default, deserialize_with = "de_u32")]237
pub attempts: u32,238
#[serde(default, deserialize_with = "de_u32")]239
pub total_retries: u32,240
#[serde(default)]241
pub error: Option<JobError>,242
#[serde(default, deserialize_with = "de_i64")]243
pub created_at: i64,244
#[serde(default, deserialize_with = "de_i64")]245
pub updated_at: i64,246
#[serde(default, deserialize_with = "de_opt_i64", skip_serializing_if = "Option::is_none")]247
pub started_at: Option<i64>,248
#[serde(default, deserialize_with = "de_opt_i64", skip_serializing_if = "Option::is_none")]249
pub finished_at: Option<i64>,250
#[serde(default, deserialize_with = "de_opt_i64")]251
pub next_retry_at: Option<i64>,252
#[serde(default, deserialize_with = "de_opt_string", skip_serializing_if = "Option::is_none")]253
pub resolved_path: Option<String>,254
#[serde(default, deserialize_with = "de_opt_string", skip_serializing_if = "Option::is_none")]255
pub part_path: Option<String>,256
/// `None` = not prepared yet (0.1.0 `undefined`), `Some(Null)` = target did not exist.257
#[serde(default, deserialize_with = "de_present", skip_serializing_if = "Option::is_none")]258
pub target_baseline: Option<Value>,259
#[serde(default, skip_serializing_if = "Option::is_none")]260
pub helper: Option<bool>,261
#[serde(default, skip_serializing_if = "Option::is_none")]262
pub owner_uid: Option<Value>,263
#[serde(default, deserialize_with = "de_opt_string", skip_serializing_if = "Option::is_none")]264
pub hash: Option<String>,265
#[serde(default, skip_serializing_if = "Option::is_none")]266
pub verification: Option<Value>,267
#[serde(default, skip_serializing_if = "Option::is_none")]268
pub staged: Option<Value>,269
#[serde(default, deserialize_with = "de_opt_string", skip_serializing_if = "Option::is_none")]270
pub local_part: Option<String>,271
#[serde(default)]272
pub auth_required: bool,273
#[serde(default)]274
pub resumable: bool,275
#[serde(default)]276
pub finalize_with_sudo: bool,277
#[serde(default)]278
pub cleanup_pending: bool,279
/// Unknown fields written by other versions are kept unchanged.280
#[serde(flatten)]281
pub extra: Map<String, Value>,282
}284
impl Job {285
fn finalize_state(&self) -> &str {286
self.finalize["state"].as_str().unwrap_or("pending")287
}288
fn session(&self) -> String {289
self.session_id.as_str().unwrap_or("").to_string()290
}291
}293
#[derive(Debug, Clone, Serialize, Deserialize)]294
#[serde(rename_all = "camelCase")]295
pub struct Settings {296
pub max_concurrent: u32,297
pub auto_retry: bool,298
pub verify: bool,299
}301
impl Default for Settings {302
fn default() -> Self {303
Settings { max_concurrent: 2, auto_retry: true, verify: true }304
}305
}307
#[derive(Debug, Clone)]308
struct PlanFile {309
name: String,310
local_path: String,311
remote_path: String,312
size: u64,313
local_identity: Option<LocalIdentity>,314
remote_identity: Option<RemoteIdentity>,315
mode: Option<u32>,316
conflict: bool,317
overwrite_confirmed: bool,318
}320
#[derive(Debug, Clone)]321
struct Plan {322
id: String,323
owner: String,324
direction: &'static str,325
session_id: String,326
identity: SessionIdentity,327
directory: String,328
local_directory: String,329
files: Vec<PlanFile>,330
directories: Vec<String>,331
skipped: Vec<Value>,332
bytes: u64,333
created_at: i64,334
mode: String,335
privilege: Option<Value>,336
/// Server-to-server: the source session and its identity. The plan's session is the destination.337
relay: Option<(String, SessionIdentity)>,338
/// Source folders of a move, removed when empty after the batch (a job exists only per file).339
source_directories: Vec<String>,340
}342
struct Runtime {343
running: bool,344
stop: watch::Sender<String>,345
generation: u64,346
speed: f64,347
samples: u32,348
last_bytes: u64,349
last_time: i64,350
active_since: Option<i64>,351
progress_since_resume: u64,352
retry_task: Option<tokio::task::JoinHandle<()>>,353
}355
impl Runtime {356
fn new(transferred: u64) -> Self {357
Runtime { running: false, stop: watch::channel(String::new()).0, generation: 0, speed: 0.0, samples: 0, last_bytes: transferred, last_time: now(), active_since: None, progress_since_resume: 0, retry_task: None }358
}359
}361
struct BatchMove {362
/// `None` = the folders are on this computer.363
session: Option<String>,364
directories: Vec<String>,365
}367
#[derive(Default)]368
struct Inner {369
batch_moves: HashMap<String, BatchMove>,370
jobs: HashMap<String, Job>,371
runtime: HashMap<String, Runtime>,372
secrets: HashMap<String, Zeroizing<String>>,373
plans: HashMap<String, Plan>,374
settings: Settings,375
order: u64,376
changed: HashSet<String>,377
pending_owners: HashSet<String>,378
full_owners: HashSet<String>,379
emit_scheduled: bool,380
cleaned_identities: HashSet<String>,381
closed: bool,382
}384
/// Stop requested (pause/cancel/shutdown/lost) or a real failure.385
enum RunError {386
Stop(String),387
Error(CoreError),388
}390
impl From<CoreError> for RunError {391
fn from(error: CoreError) -> Self {392
RunError::Error(error)393
}394
}396
type RunResult<T> = Result<T, RunError>;398
/// 0.1.0 `errorCode` for Core errors.399
fn error_code(error: &CoreError) -> String {400
if !error.code.is_empty() {401
return error.code.clone();402
}403
let lower = error.message.to_lowercase();404
if ["no response", "not connected", "connection lost", "connection closed", "channel", "econnreset", "socket"].iter().any(|p| lower.contains(p)) {405
return "SESSION_NOT_CONNECTED".into();406
}407
"TRANSFER_FAILED".into()408
}410
fn sftp_core(error: SftpError) -> CoreError {411
CoreError::new(error.transfer_code(), error.message())412
}414
fn io_core(error: std::io::Error) -> CoreError {415
let code = match error.kind() {416
std::io::ErrorKind::NotFound => "ENOENT",417
std::io::ErrorKind::PermissionDenied => "EACCES",418
std::io::ErrorKind::StorageFull => "ENOSPC",419
std::io::ErrorKind::AlreadyExists => "EEXIST",420
_ => "EIO",421
};422
CoreError::new(code, error.to_string())423
}425
fn percent(transferred: u64, size: u64) -> Value {426
if size == 0 {427
return Value::Null;428
}429
json!(((transferred.min(size) as f64 / size as f64) * 1000.0).floor().min(1000.0) / 10.0)430
}432
/// Aggregate view: percentages are byte based.433
pub fn summarize(jobs: &[Value]) -> Value {434
let counted: Vec<&Value> = jobs.iter().filter(|j| j["state"] != "cancelled").collect();435
let sizes_known = counted.iter().all(|j| j["size"].is_number());436
let total: f64 = counted.iter().map(|j| j["size"].as_f64().unwrap_or(0.0)).sum();437
let transferred: f64 = counted.iter().map(|j| j["transferred"].as_f64().unwrap_or(0.0).min(j["size"].as_f64().unwrap_or(f64::INFINITY))).sum();438
let completed = counted.iter().filter(|j| j["state"] == "completed").count();439
let active: Vec<&Value> = jobs.iter().filter(|j| RUNNING.contains(&j["state"].as_str().unwrap_or(""))).collect();440
let speed: f64 = active.iter().map(|j| j["speed"].as_f64().unwrap_or(0.0)).sum();441
let remaining: f64 = counted.iter().filter(|j| !FINAL.contains(&j["state"].as_str().unwrap_or(""))).map(|j| (j["size"].as_f64().unwrap_or(0.0) - j["transferred"].as_f64().unwrap_or(0.0)).max(0.0)).sum();442
let eta_reliable = !active.is_empty() && active.iter().all(|j| j["state"] != "transferring" || !j["eta"].is_null());443
let mut overall: Option<f64> = None;444
if sizes_known && total > 0.0 {445
overall = Some((transferred / total * 1000.0).floor() / 10.0);446
} else if sizes_known && !counted.is_empty() && completed == counted.len() {447
overall = Some(100.0);448
}449
if overall.is_some_and(|o| o >= 100.0) && completed != counted.len() {450
overall = Some(99.9);451
}452
let count = |state: &str| jobs.iter().filter(|j| j["state"] == state).count();453
json!({454
"total": jobs.len(),455
"completed": completed,456
"active": active.len(),457
"waiting": count("waiting") + count("reconnecting"),458
"paused": count("paused"),459
"interrupted": count("interrupted"),460
"failed": count("failed"),461
"cancelled": jobs.len() - counted.len(),462
"totalBytes": total as u64,463
"transferredBytes": transferred as u64,464
"sizesKnown": sizes_known,465
"percent": overall,466
"speed": speed,467
"eta": if speed > 0.0 && eta_reliable && remaining > 0.0 { json!((remaining / speed).ceil() as u64) } else if remaining == 0.0 && completed == counted.len() { json!(0) } else { Value::Null }468
})469
}471
/// Native dialogs for one planning call (shown by the command layer for the calling module).472
pub type PickPaths = Box<dyn FnOnce() -> BoxFuture<'static, Option<Vec<PathBuf>>> + Send>;473
pub type PickSavePath = Box<dyn FnOnce(String) -> BoxFuture<'static, Option<PathBuf>> + Send>;474
pub type PickDirectory = Box<dyn FnOnce() -> BoxFuture<'static, Option<PathBuf>> + Send>;476
pub type UpdateEmitter = Arc<dyn Fn(&str, Value) + Send + Sync>;478
pub struct TransferManager {479
sessions: Arc<SessionManager>,480
files: Arc<RemoteFileService>,481
logger: Arc<Logger>,482
file: Option<PathBuf>,483
inner: Mutex<Inner>,484
emitter: UpdateEmitter,485
persist_pending: AtomicBool,486
cleanup_scheduled: AtomicBool,487
scheduling: AtomicBool,488
}490
impl TransferManager {491
pub fn new(sessions: Arc<SessionManager>, files: Arc<RemoteFileService>, storage_root: Option<&Path>, logger: Arc<Logger>, emitter: UpdateEmitter) -> Arc<Self> {492
let manager = Arc::new(TransferManager {493
sessions,494
files,495
logger,496
file: storage_root.map(|r| r.join("transfers.json")),497
inner: Mutex::new(Inner::default()),498
emitter,499
persist_pending: AtomicBool::new(false),500
cleanup_scheduled: AtomicBool::new(false),501
scheduling: AtomicBool::new(false),502
});503
manager.load();504
manager505
}507
/// Subscribes to session state changes (must run inside the async runtime).508
pub fn start_session_listener(self: &Arc<Self>) {509
let mut events = self.sessions.state_events.subscribe();510
let weak = Arc::downgrade(self);511
tokio::spawn(async move {512
loop {513
match events.recv().await {514
Ok(status) => match weak.upgrade() {515
Some(manager) => manager.session_state(&status),516
None => break,517
},518
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,519
Err(_) => break,520
}521
}522
});523
}525
// ---- public API (owner = module id) ----------------------------------------------------------527
pub fn list(self: &Arc<Self>, owner: &str) -> Value {528
let jobs = {529
let inner = self.inner.lock();530
Self::owned(&inner, owner).iter().map(|j| Self::public(&inner, j)).collect::<Vec<_>>()531
};532
self.cleanup_orphans_later();533
let settings = self.inner.lock().settings.clone();534
json!({ "jobs": jobs, "summary": summarize(&jobs), "settings": settings })535
}537
pub fn update_settings(self: &Arc<Self>, patch: &Value) -> CoreResult<Value> {538
let settings = {539
let mut inner = self.inner.lock();540
if let Some(value) = patch.get("maxConcurrent").filter(|v| !v.is_null()) {541
match number(value) {542
Some(n) if n.fract() == 0.0 && (1.0..=8.0).contains(&n) => inner.settings.max_concurrent = n as u32,543
_ => return Err(file_error("TRANSFER_INVALID_SETTINGS", "Concurrent transfers must be between 1 and 8.")),544
}545
}546
if let Some(value) = patch.get("autoRetry").filter(|v| !v.is_null()) {547
inner.settings.auto_retry = *value == true;548
}549
if let Some(value) = patch.get("verify").filter(|v| !v.is_null()) {550
inner.settings.verify = *value == true;551
}552
inner.settings.clone()553
};554
self.persist(true, false);555
self.schedule();556
Ok(serde_json::to_value(settings).unwrap_or_default())557
}559
pub fn identity(&self, session_id: &str) -> CoreResult<SessionIdentity> {560
let session = self561
.sessions562
.list()563
.into_iter()564
.find(|s| s["sessionId"] == session_id && s["state"] == "connected")565
.ok_or_else(|| file_error("SESSION_NOT_CONNECTED", "The SSH session is not connected."))?;566
Ok(SessionIdentity::of_status(&session))567
}569
pub async fn plan_upload(self: &Arc<Self>, owner: &str, session_id: &str, directory: &Value, pick: PickPaths) -> CoreResult<Value> {570
let directory = normalize_remote_path(&or_root(directory))?;571
let identity = self.identity(session_id)?;572
let mut local_paths: Vec<PathBuf> = Vec::new();573
for path in pick().await.unwrap_or_default() {574
if !local_paths.contains(&path) {575
local_paths.push(path);576
}577
}578
if local_paths.is_empty() {579
return Ok(json!({ "canceled": true }));580
}581
let mut plan = Plan {582
id: hex_id(16),583
owner: owner.into(),584
direction: "upload",585
session_id: session_id.into(),586
identity,587
directory: directory.clone(),588
local_directory: String::new(),589
files: Vec::new(),590
directories: Vec::new(),591
skipped: Vec::new(),592
bytes: 0,593
created_at: now(),594
mode: String::new(),595
privilege: None,596
relay: None,597
source_directories: Vec::new(),598
};599
let walked = {600
let directory = directory.clone();601
tokio::task::spawn_blocking(move || -> CoreResult<(Vec<PlanFile>, Vec<String>, Vec<String>, Vec<Value>, u64)> {602
let mut walk = LocalWalk::default();603
for local in &local_paths {604
let name = safe_entry_name(&local.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default())?;605
walk.visit(local, &posix::join(&directory, &name), 0)?;606
}607
Ok((walk.files, walk.directories, walk.source_directories, walk.skipped, walk.bytes))608
})609
.await610
.map_err(|_| CoreError::plain("Local folder scan failed."))??611
};612
(plan.files, plan.directories, plan.source_directories, plan.skipped, plan.bytes) = walked;613
let sftp = self.files.sftp(session_id).await?;614
plan.mode = directory_mode(&sftp, &directory).await?.into();615
remote_conflicts(&sftp, &mut plan).await;616
drop(sftp);617
if plan.mode == "privileged" {618
let privilege = self.files.privileges(session_id).await?;619
if privilege["mode"] == "none" {620
let code = if privilege["sudoInstalled"] == true { "SUDO_NOT_PERMITTED" } else { "SUDO_UNAVAILABLE" };621
return Err(file_error(code, "The destination is not writable and the SSH account cannot use sudo."));622
}623
plan.privilege = Some(privilege);624
}625
self.store_plan(plan)626
}628
pub async fn plan_download(self: &Arc<Self>, owner: &str, session_id: &str, paths: &Value, options: &Value, pick_file: PickSavePath, pick_directory: PickDirectory) -> CoreResult<Value> {629
let identity = self.identity(session_id)?;630
let values: Vec<Value> = match paths {631
Value::Array(list) => list.clone(),632
other => vec![other.clone()],633
};634
let mut targets: Vec<String> = Vec::new();635
for value in values.into_iter().filter(|v| !v.is_null()) {636
let target = normalize_remote_path(&value)?;637
if !targets.contains(&target) {638
targets.push(target);639
}640
}641
if targets.is_empty() || targets.len() > 100 {642
return Err(file_error("TRANSFER_INVALID_SELECTION", "Select between 1 and 100 remote entries."));643
}644
if targets.iter().any(|t| t == "/") {645
return Err(file_error("TRANSFER_INVALID_SELECTION", "Downloading the entire remote root is not allowed in one batch."));646
}647
let sudo = options["sudo"] == true;648
let mut plan = Plan {649
id: hex_id(16),650
owner: owner.into(),651
direction: "download",652
session_id: session_id.into(),653
identity,654
directory: String::new(),655
local_directory: String::new(),656
files: Vec::new(),657
directories: Vec::new(),658
skipped: Vec::new(),659
bytes: 0,660
created_at: now(),661
mode: if sudo { "privileged-allowed".into() } else { "direct".into() },662
privilege: None,663
relay: None,664
source_directories: Vec::new(),665
};666
let sftp = self.files.sftp(session_id).await?;667
let first = if targets.len() == 1 { Some(self.remote_stat(&sftp, &targets[0], options, session_id).await?) } else { None };668
if let Some(info) = first.filter(|i| i.kind == "file") {669
let Some(chosen) = pick_file(posix::basename(&targets[0])).await else { return Ok(json!({ "canceled": true })) };670
plan.local_directory = chosen.parent().map(|p| p.to_string_lossy().to_string()).unwrap_or_default();671
add_download(&mut plan, &targets[0], &chosen.to_string_lossy(), &info);672
plan.files[0].overwrite_confirmed = options["noSaveDialog"] != true; // The native save dialog already confirmed replacing.673
} else {674
let Some(chosen) = pick_directory().await else { return Ok(json!({ "canceled": true })) };675
plan.local_directory = chosen.to_string_lossy().to_string();676
for target in &targets {677
let local = chosen.join(safe_entry_name(&posix::basename(target))?);678
self.walk_remote(&sftp, session_id, &mut plan, target, &local.to_string_lossy(), options, true, 0).await?;679
}680
}681
drop(sftp);682
for file in plan.files.iter_mut() {683
if !file.overwrite_confirmed && std::fs::symlink_metadata(&file.local_path).is_ok() {684
file.conflict = true;685
}686
}687
self.store_plan(plan)688
}690
/// Server-to-server plan. The data streams through this computer: nothing is stored on its disk.691
pub async fn plan_relay(self: &Arc<Self>, owner: &str, source_session: &str, paths: &Value, dest_session: &str, directory: &Value) -> CoreResult<Value> {692
let directory = normalize_remote_path(&or_root(directory))?;693
let identity = self.identity(dest_session)?;694
let source_identity = self.identity(source_session)?;695
let mut targets: Vec<String> = Vec::new();696
for value in paths.as_array().cloned().unwrap_or_default() {697
let target = normalize_remote_path(&value)?;698
if !targets.contains(&target) {699
targets.push(target);700
}701
}702
if targets.is_empty() || targets.len() > 100 || targets.iter().any(|t| t == "/") {703
return Err(file_error("TRANSFER_INVALID_SELECTION", "Select between 1 and 100 remote entries, not the remote root."));704
}705
if source_identity == identity {706
for target in &targets {707
if posix::dirname(target) == directory {708
return Err(file_error("TRANSFER_SAME_PATH", "The source and the destination are the same folder."));709
}710
if directory == *target || directory.starts_with(&format!("{target}/")) {711
return Err(file_error("TRANSFER_INTO_ITSELF", "A folder cannot be copied into itself."));712
}713
}714
}715
let mut plan = Plan {716
id: hex_id(16),717
owner: owner.into(),718
direction: "relay",719
session_id: dest_session.into(),720
identity,721
directory: directory.clone(),722
local_directory: String::new(),723
files: Vec::new(),724
directories: Vec::new(),725
skipped: Vec::new(),726
bytes: 0,727
created_at: now(),728
mode: String::new(),729
privilege: None,730
relay: Some((source_session.to_string(), source_identity)),731
source_directories: Vec::new(),732
};733
let source = self.files.sftp(source_session).await?;734
for target in &targets {735
let destination = posix::join(&directory, &safe_entry_name(&posix::basename(target))?);736
walk_relay(&source, &mut plan, target, &destination, 0).await?;737
}738
drop(source);739
let sftp = self.files.sftp(dest_session).await?;740
plan.mode = directory_mode(&sftp, &directory).await?.into();741
remote_conflicts(&sftp, &mut plan).await;742
drop(sftp);743
if plan.mode == "privileged" {744
let privilege = self.files.privileges(dest_session).await?;745
if privilege["mode"] == "none" {746
let code = if privilege["sudoInstalled"] == true { "SUDO_NOT_PERMITTED" } else { "SUDO_UNAVAILABLE" };747
return Err(file_error(code, "The destination is not writable and the SSH account cannot use sudo."));748
}749
plan.privilege = Some(privilege);750
}751
self.store_plan(plan)752
}754
pub async fn start(self: &Arc<Self>, owner: &str, plan_id: &Value, options: &Value) -> CoreResult<Value> {755
let plan = {756
let inner = self.inner.lock();757
inner.plans.get(plan_id.as_str().unwrap_or("")).filter(|p| p.owner == owner && now() - p.created_at <= PLAN_TTL_MS).cloned()758
}759
.ok_or_else(|| file_error("TRANSFER_PLAN_EXPIRED", "The transfer plan expired; select the files again."))?;760
if options["sessionId"].as_str().is_some_and(|id| id != plan.session_id)761
|| self.identity(&plan.session_id)? != plan.identity762
{763
return Err(file_error("TRANSFER_PLAN_SERVER_CHANGED", "The transfer plan belongs to a different server or connection."));764
}765
let mut job_extra = Map::new();766
if let Some((source, source_identity)) = &plan.relay {767
if self.identity(source)? != *source_identity {768
return Err(file_error("TRANSFER_PLAN_SERVER_CHANGED", "The transfer plan belongs to a different server or connection."));769
}770
job_extra.insert("relaySource".into(), json!({ "sessionId": source, "identity": source_identity }));771
}772
job_extra.insert("hostKey".into(), json!(self.sessions.trusted_host_key(&plan.session_id)));773
if let Some((source, _)) = &plan.relay {774
job_extra.insert("sourceHostKey".into(), json!(self.sessions.trusted_host_key(source)));775
}776
if options["move"] == true {777
job_extra.insert("moveSource".into(), json!(true));778
job_extra.insert("moveRoot".into(), json!(options["sourceDirectory"].as_str().unwrap_or("")));779
}780
let conflict = options["conflict"].as_str().filter(|c| ["overwrite", "skip"].contains(c)).unwrap_or("fail");781
let conflicts = plan.files.iter().filter(|f| f.conflict).count();782
if conflicts > 0 && conflict == "fail" {783
return Err(file_error("FILE_EXISTS", &format!("{conflicts} destination file(s) already exist.")));784
}785
if self.inner.lock().jobs.len() + plan.files.len() > MAX_JOBS {786
return Err(file_error("TRANSFER_QUEUE_FULL", "Remove finished transfers before starting more."));787
}788
let password = options["sudoPassword"].as_str().unwrap_or("").to_string();789
let privileged = plan.mode == "privileged" || (plan.direction == "download" && (plan.mode == "privileged-allowed" || options["sudo"] == true));790
if plan.mode == "privileged" {791
// Verifies the sudo authorization once before any data moves (and caches it on request).792
let auth = Authorization { sudo: true, sudo_password: (!password.is_empty()).then(|| password.clone()), remember_sudo: options["rememberSudo"] == true, ..Default::default() };793
self.files.helper(&plan.session_id, json!({ "operation": "stat", "path": plan.directory }), &auth).await?;794
}795
self.create_directories(&plan, privileged, &password).await?;796
let mut created = 0;797
{798
let mut inner = self.inner.lock();799
for file in &plan.files {800
if file.conflict && conflict == "skip" {801
continue;802
}803
inner.order += 1;804
let order = inner.order;805
let job = Job {806
id: hex_id(16),807
owner: owner.into(),808
batch_id: plan.id.clone(),809
direction: plan.direction.into(),810
identity: plan.identity.clone(),811
session_id: json!(plan.session_id),812
order,813
name: file.name.clone(),814
local_path: file.local_path.clone(),815
remote_path: file.remote_path.clone(),816
size: file.size,817
transferred: 0,818
moved_bytes: 0,819
active_ms: 0,820
state: "waiting".into(),821
mode: if plan.direction != "download" { plan.mode.clone() } else if privileged { "privileged-allowed".into() } else { "direct".into() },822
overwrite: conflict == "overwrite" || file.overwrite_confirmed,823
local_identity: file.local_identity.clone(),824
remote_identity: file.remote_identity.clone(),825
remote_mode: file.mode.map(Value::from),826
finalize: json!({ "state": "pending" }),827
attempts: 0,828
total_retries: 0,829
error: None,830
created_at: now(),831
updated_at: now(),832
started_at: None,833
finished_at: None,834
next_retry_at: None,835
resolved_path: None,836
part_path: None,837
target_baseline: None,838
helper: None,839
owner_uid: None,840
hash: None,841
verification: None,842
staged: None,843
local_part: None,844
auth_required: false,845
resumable: false,846
finalize_with_sudo: false,847
cleanup_pending: false,848
extra: job_extra.clone(),849
};850
if !password.is_empty() && privileged {851
inner.secrets.insert(job.id.clone(), Zeroizing::new(password.clone()));852
}853
let id = job.id.clone();854
inner.jobs.insert(id.clone(), job);855
created += 1;856
Self::mark(&mut inner, &id, owner);857
}858
inner.plans.remove(&plan.id);859
if options["move"] == true && !plan.source_directories.is_empty() {860
let session = plan.relay.as_ref().map(|(source, _)| source.clone()).or_else(|| (plan.direction == "download").then(|| plan.session_id.clone()));861
inner.batch_moves.insert(plan.id.clone(), BatchMove { session, directories: plan.source_directories.clone() });862
}863
Self::trim_history(&mut inner);864
}865
self.finish_batch_move(&plan.id).await;866
self.emit_soon();867
self.persist(true, false);868
self.logger.info("Transfer batch queued.", json!({ "direction": plan.direction, "files": created, "bytes": plan.bytes, "mode": plan.mode, "sessionId": plan.session_id }));869
self.schedule();870
Ok(json!({ "batchId": plan.id, "jobs": created, "skipped": plan.files.len() - created }))871
}873
pub fn pause(self: &Arc<Self>, owner: &str, ids: &Value) -> Value {874
let mut count = 0;875
for id in self.select(owner, ids) {876
let state = self.job_state(&id);877
if ["waiting", "reconnecting", "interrupted"].contains(&state.as_str()) {878
self.clear_retry(&id);879
self.set(&id, "paused", None);880
count += 1;881
} else if RUNNING.contains(&state.as_str()) && state != "finalizing" {882
self.stop(&id, "pause");883
count += 1;884
}885
}886
self.persist(true, false);887
json!({ "paused": count })888
}890
pub fn resume(self: &Arc<Self>, owner: &str, ids: &Value, options: &Value) -> Value {891
let mut count = 0;892
for id in self.select(owner, ids) {893
let resumable = {894
let inner = self.inner.lock();895
inner.jobs.get(&id).is_some_and(|j| ["paused", "interrupted", "reconnecting"].contains(&j.state.as_str()) || (j.state == "failed" && j.resumable))896
};897
if !resumable {898
continue;899
}900
if !self.bind_session(&id, options) { continue; }901
self.authorize(&id, options);902
{903
let mut inner = self.inner.lock();904
if let Some(job) = inner.jobs.get_mut(&id) {905
if options["overwrite"] == true && job.error.as_ref().is_some_and(|e| e.code == "FILE_EXISTS" || e.code == "FILE_CONFLICT") {906
job.overwrite = true;907
job.target_baseline = None;908
job.finalize = json!({ "state": "pending" });909
}910
job.attempts = 0;911
job.auth_required = false;912
}913
}914
self.clear_retry(&id);915
self.set(&id, "waiting", None);916
count += 1;917
}918
self.persist(true, false);919
self.schedule();920
json!({ "resumed": count })921
}923
/// Restart from offset zero (used when the partial data cannot be safely continued).924
pub fn retry(self: &Arc<Self>, owner: &str, ids: &Value, options: &Value) -> Value {925
let mut count = 0;926
for id in self.select(owner, ids) {927
let local_part = {928
let inner = self.inner.lock();929
match inner.jobs.get(&id) {930
Some(j) if ["failed", "interrupted", "paused", "cancelled"].contains(&j.state.as_str()) => Some(j.local_part.clone()),931
_ => None,932
}933
};934
let Some(local_part) = local_part else { continue };935
if !self.bind_session(&id, options) { continue; }936
// The deterministic partial file (same job id) is truncated and rewritten from offset zero.937
if let Some(part) = local_part {938
let _ = std::fs::remove_file(part);939
}940
self.authorize(&id, options);941
{942
let mut inner = self.inner.lock();943
if let Some(job) = inner.jobs.get_mut(&id) {944
job.transferred = 0;945
job.staged = None;946
job.target_baseline = None;947
job.hash = None;948
job.verification = None;949
job.finalize = json!({ "state": "pending" });950
job.attempts = 0;951
job.auth_required = false;952
job.error = None;953
job.resumable = false;954
job.finalize_with_sudo = false;955
if job.direction == "upload" {956
if let Ok(meta) = std::fs::metadata(&job.local_path) {957
job.local_identity = Some(LocalIdentity::of(&meta));958
job.size = meta.len();959
}960
} else {961
job.remote_identity = None;962
}963
if options["overwrite"] == true {964
job.overwrite = true;965
}966
}967
}968
self.set(&id, "waiting", None);969
count += 1;970
}971
self.persist(true, false);972
self.schedule();973
json!({ "retried": count })974
}976
pub async fn cancel(self: &Arc<Self>, owner: &str, ids: &Value) -> Value {977
let mut count = 0;978
for id in self.select(owner, ids) {979
let state = self.job_state(&id);980
if FINAL.contains(&state.as_str()) {981
continue;982
}983
count += 1;984
if RUNNING.contains(&state.as_str()) {985
if state == "finalizing" {986
continue; // The remote replacement must finish or fail on its own.987
}988
self.stop(&id, "cancel");989
continue;990
}991
self.clear_retry(&id);992
self.finish_cancel(&id).await;993
}994
self.persist(true, false);995
json!({ "cancelled": count })996
}998
pub fn remove(self: &Arc<Self>, owner: &str, ids: &Value) -> Value {999
let mut count = 0;1000
{1001
let mut inner = self.inner.lock();1002
let selected = Self::select_ids(&inner, owner, ids);1003
for id in selected {1004
let removable = inner.jobs.get(&id).is_some_and(|j| FINAL.contains(&j.state.as_str()) && !j.cleanup_pending);1005
if !removable {1006
continue;1007
}1008
inner.jobs.remove(&id);1009
inner.secrets.remove(&id);1010
inner.runtime.remove(&id);1011
inner.changed.insert(id);1012
count += 1;1013
}1014
inner.pending_owners.insert(owner.to_string());1015
inner.full_owners.insert(owner.to_string());1016
}1017
self.persist(true, false);1018
self.emit_soon();1019
json!({ "removed": count })1020
}1022
pub fn shutdown(self: &Arc<Self>) {1023
let ids: Vec<(String, String)> = {1024
let inner = self.inner.lock();1025
if inner.closed {1026
return;1027
}1028
inner.jobs.values().map(|j| (j.id.clone(), j.state.clone())).collect()1029
};1030
for (id, state) in ids {1031
self.clear_retry(&id);1032
if RUNNING.contains(&state.as_str()) {1033
self.stop(&id, "shutdown");1034
}1035
}1036
// Running jobs are recorded as interrupted so the next start offers to resume them.1037
{1038
let mut inner = self.inner.lock();1039
for job in inner.jobs.values_mut() {1040
if RUNNING.contains(&job.state.as_str()) || job.state == "waiting" || job.state == "reconnecting" {1041
if job.state == "finalizing" || job.finalize_state() == "running" {1042
job.finalize["state"] = json!("unknown");1043
}1044
job.state = "interrupted".into();1045
job.error = Some(JobError { code: "APP_RESTARTED".into(), message: "The application was closed during the transfer.".into() });1046
}1047
}1048
}1049
self.persist(true, true);1050
let mut inner = self.inner.lock();1051
inner.closed = true; // Late completions of stopped jobs must not overwrite the saved state.1052
inner.secrets.clear();1053
}1055
// ---- planning helpers ----------------------------------------------------------------------1057
fn store_plan(&self, plan: Plan) -> CoreResult<Value> {1058
if plan.files.len() + plan.directories.len() > MAX_PLAN_ENTRIES {1059
return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", &format!("A transfer can contain at most {MAX_PLAN_ENTRIES} entries.")));1060
}1061
if plan.bytes > MAX_PLAN_BYTES {1062
return Err(file_error("TRANSFER_TOO_LARGE", "The selection exceeds the transfer size limit."));1063
}1064
let conflicts: Vec<&PlanFile> = plan.files.iter().filter(|f| f.conflict).collect();1065
let result = json!({1066
"canceled": false,1067
"planId": plan.id,1068
"direction": plan.direction,1069
"files": plan.files.len(),1070
"directories": plan.directories.len(),1071
"bytes": plan.bytes,1072
"mode": plan.mode,1073
"requiresSudo": plan.mode == "privileged" && plan.privilege.as_ref().is_some_and(|p| p["mode"] != "root"),1074
"sudoPasswordRequired": plan.mode == "privileged" && plan.privilege.as_ref().is_some_and(|p| p["mode"] == "sudo-password" && p["sudoCached"] != true),1075
"conflicts": conflicts.iter().take(50).map(|f| if plan.direction != "download" { f.remote_path.clone() } else { f.local_path.clone() }).collect::<Vec<_>>(),1076
"conflictCount": conflicts.len(),1077
"skipped": plan.skipped.iter().take(50).cloned().collect::<Vec<_>>(),1078
"skippedCount": plan.skipped.len(),1079
"destination": if plan.direction != "download" { plan.directory.clone() } else { plan.local_directory.clone() }1080
});1081
let mut inner = self.inner.lock();1082
let cutoff = now() - PLAN_TTL_MS;1083
inner.plans.retain(|_, p| p.created_at >= cutoff);1084
inner.plans.insert(plan.id.clone(), plan);1085
Ok(result)1086
}1088
async fn remote_stat(&self, sftp: &Sftp, target: &str, options: &Value, session_id: &str) -> CoreResult<RemoteInfo> {1089
match sftp.stat(target).await {1090
Ok(stat) => Ok(RemoteInfo { kind: if stat.is_dir() { "directory" } else { "file" }, size: stat.size, mtime: stat.mtime as u64, mode: stat.mode }),1091
Err(error) => {1092
let code = error.transfer_code();1093
if code != "FILE_PERMISSION_DENIED" || options["sudo"] != true {1094
return Err(file_error(code, &format!("{target} is not accessible.")));1095
}1096
let value = self.files.helper(session_id, json!({ "operation": "stat", "path": target }), &Authorization::from_options(options).sudo()).await?;1097
if value["exists"] != true {1098
return Err(file_error("FILE_NOT_FOUND", &format!("{target} does not exist.")));1099
}1100
let mtime_ns = number(&value["baseline"]["mtimeNs"]).unwrap_or(0.0);1101
Ok(RemoteInfo {1102
kind: if value["type"] == "directory" { "directory" } else { "file" },1103
size: number(&value["baseline"]["size"]).unwrap_or(0.0) as u64,1104
mtime: (mtime_ns / 1e9).trunc() as u64,1105
mode: number(&value["baseline"]["mode"]).unwrap_or(0.0) as u32,1106
})1107
}1108
}1109
}1111
#[allow(clippy::too_many_arguments)]1112
fn walk_remote<'a>(&'a self, sftp: &'a Sftp, session_id: &'a str, plan: &'a mut Plan, remote: &'a str, local: &'a str, options: &'a Value, top: bool, depth: u32) -> BoxFuture<'a, CoreResult<()>> {1113
async move {1114
if plan.files.len() + plan.directories.len() > MAX_PLAN_ENTRIES {1115
return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", &format!("A transfer can contain at most {MAX_PLAN_ENTRIES} entries.")));1116
}1117
let info = if top {1118
self.remote_stat(sftp, remote, options, session_id).await?1119
} else {1120
let stat = sftp.lstat(remote).await.map_err(|e| file_error(e.transfer_code(), &format!("{remote} is not accessible.")))?;1121
if stat.is_symlink() {1122
plan.skipped.push(json!({ "path": remote, "reason": "symlink" }));1123
return Ok(());1124
}1125
RemoteInfo { kind: if stat.is_dir() { "directory" } else if !stat.is_file() { "other" } else { "file" }, size: stat.size, mtime: stat.mtime as u64, mode: stat.mode }1126
};1127
if info.kind == "other" {1128
plan.skipped.push(json!({ "path": remote, "reason": "type" }));1129
return Ok(());1130
}1131
if info.kind != "directory" {1132
add_download(plan, remote, local, &info);1133
return Ok(());1134
}1135
if depth > 64 {1136
return Err(file_error("TRANSFER_TOO_DEEP", "The selected directory tree is too deep."));1137
}1138
plan.directories.push(local.to_string());1139
plan.source_directories.push(remote.to_string());1140
match sftp.read_dir(remote).await {1141
Ok(entries) => {1142
for entry in entries {1143
if entry.name == "." || entry.name == ".." {1144
continue;1145
}1146
let Ok(name) = safe_entry_name(&entry.name) else {1147
plan.skipped.push(json!({ "path": posix::join(remote, &entry.name), "reason": "name" }));1148
continue;1149
};1150
let child_local = Path::new(local).join(&name).to_string_lossy().to_string();1151
self.walk_remote(sftp, session_id, plan, &posix::join(remote, &name), &child_local, options, false, depth + 1).await?;1152
}1153
}1154
Err(error) => {1155
if error.transfer_code() != "FILE_PERMISSION_DENIED" || options["sudo"] != true {1156
return Err(file_error(error.transfer_code(), &format!("Directory {remote} cannot be listed.")));1157
}1158
let listed = self.files.helper(session_id, json!({ "operation": "listdir", "path": remote }), &Authorization::from_options(options).sudo()).await?;1159
for entry in listed["entries"].as_array().cloned().unwrap_or_default() {1160
let raw = entry["name"].as_str().unwrap_or("").to_string();1161
let Ok(name) = safe_entry_name(&raw) else {1162
plan.skipped.push(json!({ "path": posix::join(remote, &raw), "reason": "name" }));1163
continue;1164
};1165
let child = posix::join(remote, &name);1166
let child_local = Path::new(local).join(&name).to_string_lossy().to_string();1167
match entry["type"].as_str() {1168
Some(kind @ ("symlink" | "other")) => plan.skipped.push(json!({ "path": child, "reason": kind })),1169
Some("directory") => self.walk_remote(sftp, session_id, plan, &child, &child_local, options, true, depth + 1).await?,1170
_ => add_download(1171
plan,1172
&child,1173
&child_local,1174
&RemoteInfo { kind: "file", size: number(&entry["size"]).unwrap_or(0.0) as u64, mtime: number(&entry["mtime"]).unwrap_or(0.0) as u64, mode: number(&entry["mode"]).unwrap_or(0.0) as u32 },1175
),1176
}1177
}1178
}1179
}1180
Ok(())1181
}1182
.boxed()1183
}1185
async fn create_directories(&self, plan: &Plan, privileged: bool, password: &str) -> CoreResult<()> {1186
if plan.direction == "download" {1187
for directory in &plan.directories {1188
std::fs::create_dir_all(directory).map_err(io_core)?;1189
#[cfg(unix)]1190
{1191
use std::os::unix::fs::PermissionsExt;1192
let _ = std::fs::set_permissions(directory, std::fs::Permissions::from_mode(0o700));1193
}1194
}1195
return Ok(());1196
}1197
if plan.directories.is_empty() {1198
return Ok(());1199
}1200
if privileged {1201
let auth = Authorization { sudo: true, sudo_password: (!password.is_empty()).then(|| password.to_string()), ..Default::default() };1202
for directory in &plan.directories {1203
if let Err(error) = self.files.helper(&plan.session_id, json!({ "operation": "mkdir", "path": directory, "parents": true }), &auth).await {1204
if error.code != "FILE_EXISTS" {1205
return Err(error);1206
}1207
}1208
}1209
return Ok(());1210
}1211
let sftp = self.files.sftp(&plan.session_id).await?;1212
for directory in &plan.directories {1213
if let Err(error) = sftp.mkdir(directory, 0o755).await {1214
if !sftp.stat(directory).await.map(|s| s.is_dir()).unwrap_or(false) {1215
return Err(file_error(error.transfer_code(), &format!("Directory {directory} could not be created.")));1216
}1217
}1218
}1219
Ok(())1220
}1222
// ---- scheduling ------------------------------------------------------------------------------1224
fn schedule(self: &Arc<Self>) {1225
if self.scheduling.swap(true, Ordering::SeqCst) {1226
return;1227
}1228
let manager = self.clone();1229
tokio::spawn(async move {1230
tokio::task::yield_now().await;1231
manager.scheduling.store(false, Ordering::SeqCst);1232
let start: Vec<String> = {1233
let mut inner = manager.inner.lock();1234
if inner.closed {1235
return;1236
}1237
let mut running = inner.runtime.values().filter(|r| r.running).count();1238
let mut waiting: Vec<(u64, String)> = inner.jobs.values().filter(|j| j.state == "waiting").map(|j| (j.order, j.id.clone())).collect();1239
waiting.sort();1240
let max = inner.settings.max_concurrent as usize;1241
let mut start = Vec::new();1242
for (_, id) in waiting {1243
if running >= max {1244
break;1245
}1246
if inner.runtime.get(&id).is_some_and(|r| r.running) {1247
continue;1248
}1249
running += 1;1250
let transferred = inner.jobs[&id].transferred;1251
inner.runtime.entry(id.clone()).or_insert_with(|| Runtime::new(transferred)).running = true;1252
start.push(id);1253
}1254
start1255
};1256
for id in start {1257
let manager = manager.clone();1258
tokio::spawn(async move { manager.run(id).await });1259
}1260
});1261
}1263
async fn run(self: Arc<Self>, id: String) {1264
let stop_rx = {1265
let mut inner = self.inner.lock();1266
let transferred = inner.jobs.get(&id).map(|j| j.transferred).unwrap_or(0);1267
let rt = inner.runtime.entry(id.clone()).or_insert_with(|| Runtime::new(transferred));1268
rt.running = true;1269
rt.stop = watch::channel(String::new()).0;1270
rt.generation += 1;1271
rt.speed = 0.0;1272
rt.samples = 0;1273
rt.last_bytes = transferred;1274
rt.last_time = now();1275
rt.progress_since_resume = 0;1276
let rx = rt.stop.subscribe();1277
if let Some(job) = inner.jobs.get_mut(&id) {1278
job.error = None;1279
job.resumable = false;1280
}1281
rx1282
};1283
self.set(&id, "connecting", None);1284
let outcome: RunResult<()> = async {1285
self.check_host_keys(&id)?;1286
let session_id = self.session_for(&id)?;1287
self.update(&id, |j| j.session_id = json!(session_id));1288
let direction = self.with_job(&id, |j| j.direction.clone()).unwrap_or_default();1289
if direction == "upload" {1290
self.run_upload(&id, &session_id, stop_rx.clone()).await1291
} else if direction == "relay" {1292
self.run_relay(&id, &session_id, stop_rx.clone()).await1293
} else {1294
self.run_download(&id, &session_id, stop_rx.clone()).await1295
}1296
}1297
.await;1298
match outcome {1299
Ok(()) => {1300
self.finish_move(&id).await;1301
if let Some(batch) = self.with_job(&id, |j| j.batch_id.clone()) {1302
self.finish_batch_move(&batch).await;1303
}1304
}1305
Err(error) => self.failure(&id, error, &stop_rx).await,1306
}1307
if let Some(rt) = self.inner.lock().runtime.get_mut(&id) {1308
rt.running = false;1309
}1310
self.persist(true, false);1311
self.schedule();1312
}1314
async fn failure(self: &Arc<Self>, id: &str, error: RunError, stop: &watch::Receiver<String>) {1315
let requested = stop.borrow().clone();1316
let error = match error {1317
RunError::Stop(kind) if kind != "lost" => Err(kind),1318
RunError::Stop(_) => Ok(CoreError::new("SESSION_NOT_CONNECTED", "The SSH connection was lost.")),1319
RunError::Error(_) if !requested.is_empty() && requested != "lost" => Err(requested),1320
RunError::Error(e) => Ok(e),1321
};1322
let error = match error {1323
Err(kind) => {1324
match kind.as_str() {1325
"cancel" => self.finish_cancel(id).await,1326
"pause" => self.set(id, "paused", None),1327
_ => self.set(id, "interrupted", Some(JobError { code: "APP_RESTARTED".into(), message: "The application was closed during the transfer.".into() })),1328
}1329
return;1330
}1331
Ok(error) => error,1332
};1333
let code = error_code(&error);1334
let message: String = Regex::strip_code(&error.message).chars().take(400).collect();1335
let (state, transferred, size, direction) = self.with_job(id, |j| (j.state.clone(), j.transferred, j.size, j.direction.clone())).unwrap_or_default();1336
self.logger.warn("Transfer interrupted.", json!({ "jobId": id, "direction": direction, "state": state, "code": code, "transferred": transferred, "size": size }));1337
if state == "finalizing" && (code == "FILE_RESULT_UNKNOWN" || CONNECTION_CODES.contains(&code.as_str())) {1338
self.update(id, |j| j.finalize["state"] = json!("unknown"));1339
}1340
if AUTH_CODES.contains(&code.as_str()) {1341
self.update(id, |j| j.auth_required = true);1342
self.set(id, "interrupted", Some(JobError { code, message }));1343
return;1344
}1345
if CONNECTION_CODES.contains(&code.as_str()) || code == "FILE_RESULT_UNKNOWN" {1346
self.set(id, "interrupted", Some(JobError { code, message }));1347
let auto = self.inner.lock().settings.auto_retry;1348
let retries = self.with_job(id, |j| j.total_retries).unwrap_or(0);1349
if auto && retries < MAX_TOTAL_RETRIES {1350
self.schedule_retry(id);1351
}1352
return;1353
}1354
let resumable = ["FILE_NO_SPACE", "FILE_EXISTS", "FILE_CONFLICT", "LOCAL_NO_SPACE", "ENOSPC", "FILE_READ_ONLY_FILESYSTEM", "FILE_BUSY"].contains(&code.as_str());1355
self.update(id, |j| j.resumable = resumable);1356
self.set(id, "failed", Some(JobError { code: if code == "ENOSPC" { "LOCAL_NO_SPACE".into() } else { code }, message }));1357
}1359
fn schedule_retry(self: &Arc<Self>, id: &str) {1360
self.clear_retry(id);1361
let delay = {1362
let mut inner = self.inner.lock();1363
let Some(job) = inner.jobs.get_mut(id) else { return };1364
if job.attempts as usize >= RETRY_DELAYS_MS.len() {1365
return; // Manual resume from here on: no endless loop.1366
}1367
let delay = RETRY_DELAYS_MS[job.attempts as usize];1368
job.attempts += 1;1369
job.total_retries += 1;1370
job.next_retry_at = Some(now() + delay as i64);1371
delay1372
};1373
let error = self.with_job(id, |j| j.error.clone()).flatten();1374
self.set(id, "reconnecting", error.clone());1375
let manager = self.clone();1376
let job_id = id.to_string();1377
let task = tokio::spawn(async move {1378
tokio::time::sleep(Duration::from_millis(delay)).await;1379
if let Some(rt) = manager.inner.lock().runtime.get_mut(&job_id) {1380
rt.retry_task = None;1381
}1382
if manager.job_state(&job_id) != "reconnecting" {1383
return;1384
}1385
let attempts = manager.with_job(&job_id, |j| j.attempts).unwrap_or(0) as usize;1386
if manager.find_session(&job_id).is_some() {1387
manager.set(&job_id, "waiting", None);1388
manager.schedule();1389
} else if attempts < RETRY_DELAYS_MS.len() {1390
manager.schedule_retry(&job_id);1391
} else {1392
let error = manager.with_job(&job_id, |j| j.error.clone()).flatten();1393
manager.set(&job_id, "interrupted", error);1394
}1395
});1396
let mut inner = self.inner.lock();1397
let transferred = inner.jobs.get(id).map(|j| j.transferred).unwrap_or(0);1398
inner.runtime.entry(id.to_string()).or_insert_with(|| Runtime::new(transferred)).retry_task = Some(task);1399
}1401
fn clear_retry(&self, id: &str) {1402
let mut inner = self.inner.lock();1403
if let Some(task) = inner.runtime.get_mut(id).and_then(|r| r.retry_task.take()) {1404
task.abort();1405
}1406
if let Some(job) = inner.jobs.get_mut(id) {1407
job.next_retry_at = None;1408
}1409
}1411
fn session_state(self: &Arc<Self>, status: &Value) {1412
if status["state"] == "disconnected" {1413
// A dropped SSH connection does not answer outstanding SFTP requests: end those jobs now.1414
let session = status["sessionId"].clone();1415
let ids: Vec<String> = {1416
let inner = self.inner.lock();1417
inner.jobs.values().filter(|j| (j.session_id == session || j.extra.get("relaySource").is_some_and(|r| r["sessionId"] == session)) && inner.runtime.get(&j.id).is_some_and(|r| r.running)).map(|j| j.id.clone()).collect()1418
};1419
for id in ids {1420
self.stop(&id, "lost");1421
}1422
return;1423
}1424
if status["state"] != "connected" {1425
return;1426
}1427
let identity = SessionIdentity::of_status(status).key();1428
let candidates: Vec<String> = {1429
let inner = self.inner.lock();1430
inner1431
.jobs1432
.values()1433
.filter(|j| j.identity.key() == identity || relay_source(j).is_some_and(|(_, source)| source.key() == identity))1434
.filter(|j| {1435
j.state == "reconnecting"1436
|| (j.state == "interrupted" && inner.settings.auto_retry && j.error.as_ref().is_some_and(|e| CONNECTION_CODES.contains(&e.code.as_str())) && j.total_retries < MAX_TOTAL_RETRIES)1437
})1438
.map(|j| j.id.clone())1439
.collect()1440
};1441
for id in candidates {1442
self.clear_retry(&id);1443
self.set(&id, "waiting", None);1444
}1445
self.inner.lock().cleaned_identities.remove(&identity);1446
self.cleanup_orphans_later();1447
self.schedule();1448
}1450
fn stop(&self, id: &str, kind: &str) {1451
let inner = self.inner.lock();1452
if let Some(rt) = inner.runtime.get(id) {1453
let _ = rt.stop.send(kind.to_string());1454
}1455
}1457
// ---- transfer engine -----------------------------------------------------------------------1459
fn helper_auth(&self, id: &str, privileged_override: bool, timeout: Option<u64>) -> Authorization {1460
let inner = self.inner.lock();1461
let privileged = privileged_override || inner.jobs.get(id).is_some_and(|j| j.mode == "privileged");1462
let mut auth = if privileged { Authorization { sudo: true, sudo_password: inner.secrets.get(id).map(|s| s.to_string()).filter(|s| !s.is_empty()), ..Default::default() } } else { Authorization::default() };1463
auth.timeout_ms = timeout;1464
auth1465
}1467
async fn open_channel(&self, session_id: &str, stop: &watch::Receiver<String>) -> RunResult<Arc<Sftp>> {1468
let sftp = self.files.sftp(session_id).await?;1469
let kind = stop.borrow().clone();1470
if !kind.is_empty() {1471
return Err(RunError::Stop(kind));1472
}1473
Ok(sftp)1474
}1476
async fn run_upload(self: &Arc<Self>, id: &str, session_id: &str, stop: watch::Receiver<String>) -> RunResult<()> {1477
let job = self.snapshot(id)?;1478
let meta = std::fs::symlink_metadata(&job.local_path).map_err(|e| file_error(if e.kind() == std::io::ErrorKind::NotFound { "LOCAL_NOT_FOUND" } else { "LOCAL_READ_FAILED" }, &format!("{} cannot be read.", job.local_path)))?;1479
if !meta.is_file() {1480
return Err(file_error("LOCAL_NOT_FILE", &format!("{} is no longer a regular file.", job.local_path)).into());1481
}1482
let current = LocalIdentity::of(&meta);1483
if !job.local_identity.as_ref().is_some_and(|i| i.same(¤t)) {1484
if job.transferred > 0 {1485
return Err(file_error("FILE_SOURCE_CHANGED", "The local file changed since the transfer started; restart it from the beginning.").into());1486
}1487
self.update(id, |j| {1488
j.local_identity = Some(current.clone());1489
j.size = current.size;1490
});1491
}1492
let job = self.snapshot(id)?;1493
if job.finalize_state() == "unknown" || (job.hash.is_some() && job.transferred == job.size && job.part_path.is_some()) {1494
return self.finalize_upload(id, session_id, &stop).await;1495
}1496
self.prepare_upload_target(id, session_id).await?;1497
let sftp = self.open_channel(session_id, &stop).await?;1498
let local = LocalFile::open(Path::new(&job.local_path), "r").map_err(io_core)?;1499
let handle = self.open_remote_part(&sftp, id).await?;1500
let result: RunResult<()> = async {1501
let job = self.snapshot(id)?;1502
if job.transferred > 0 {1503
let remote = sftp.fstat(&handle).await.map_err(sftp_core)?;1504
let transferred = job.transferred.min(remote.size);1505
self.update(id, |j| j.transferred = transferred);1506
let window = TAIL_CHECK.min(transferred);1507
let position = transferred - window;1508
verify_tail(1509
transferred,1510
async { local.read_at(position, window as usize).await.map_err(io_core) }.boxed(),1511
async { sftp.read_full(&handle, position, window as usize).await.map_err(sftp_core) }.boxed(),1512
window,1513
)1514
.await?;1515
}1516
self.set(id, "transferring", None);1517
let size = self.with_job(id, |j| j.size).unwrap_or(0);1518
let produce: Producer = {1519
let local = local.clone();1520
Arc::new(move |pos, len| {1521
let local = local.clone();1522
Box::pin(async move {1523
let data = local.read_at(pos, len).await.map_err(io_core)?;1524
if data.len() != len {1525
return Err(file_error("FILE_SOURCE_CHANGED", "The local file shrank during the transfer."));1526
}1527
Ok(data)1528
})1529
})1530
};1531
let consume: Consumer = {1532
let (sftp, handle) = (sftp.clone(), handle.clone());1533
Arc::new(move |buf, pos| {1534
let (sftp, handle) = (sftp.clone(), handle.clone());1535
Box::pin(async move { sftp.write_at(&handle, pos, &buf).await.map_err(sftp_core) })1536
})1537
};1538
self.run_pump(id, size, produce, consume, stop.clone()).await?;1539
let remote = sftp.fstat(&handle).await.map_err(sftp_core)?;1540
if remote.size > size {1541
sftp.truncate(&handle, size).await.map_err(sftp_core)?;1542
}1543
let _ = sftp.fsync(&handle).await;1544
let _ = sftp.close_file(&handle).await;1545
let after = local.metadata().map(|m| LocalIdentity::of(&m)).map_err(io_core)?;1546
if !self.with_job(id, |j| j.local_identity.as_ref().is_some_and(|i| i.same(&after))).unwrap_or(false) {1547
return Err(file_error("FILE_SOURCE_CHANGED", "The local file changed during the transfer.").into());1548
}1549
self.set(id, "verifying", None);1550
let hash = local_hash(Path::new(&job.local_path), stop.clone()).await?;1551
self.update(id, |j| {1552
j.hash = Some(hash.clone());1553
j.verification = Some(json!({ "method": "sha256", "local": hash }));1554
});1555
Ok(())1556
}1557
.await;1558
if result.is_err() {1559
let _ = tokio::time::timeout(Duration::from_secs(3), sftp.close_file(&handle)).await;1560
}1561
drop(sftp);1562
result?;1563
self.finalize_upload(id, session_id, &stop).await1564
}1566
/// Like an upload, with an SFTP file as the source: same part file, publish step and resume rules.1567
async fn run_relay(self: &Arc<Self>, id: &str, session_id: &str, stop: watch::Receiver<String>) -> RunResult<()> {1568
let job = self.snapshot(id)?;1569
if job.finalize_state() == "unknown" || (job.hash.is_some() && job.transferred == job.size && job.part_path.is_some()) {1570
return self.finalize_upload(id, session_id, &stop).await;1571
}1572
let source_session = self.relay_session(id)?;1573
let source = self.open_channel(&source_session, &stop).await?;1574
let stat = source.stat(&job.local_path).await.map_err(|e| file_error(e.transfer_code(), &format!("{} is not accessible.", job.local_path)))?;1575
if !stat.is_file() {1576
return Err(file_error("FILE_UNSAFE_TYPE_OR_LINK", &format!("{} is no longer a regular file.", job.local_path)).into());1577
}1578
let current = RemoteIdentity { size: stat.size, mtime: stat.mtime as u64 };1579
if job.remote_identity.as_ref() != Some(¤t) {1580
if job.transferred > 0 {1581
return Err(file_error("FILE_SOURCE_CHANGED", "The source file changed since the transfer started; restart it from the beginning.").into());1582
}1583
self.update(id, |j| {1584
j.remote_identity = Some(current.clone());1585
j.size = current.size;1586
});1587
}1588
self.prepare_upload_target(id, session_id).await?;1589
let destination = self.open_channel(session_id, &stop).await?;1590
let input = source.open_file(&job.local_path, "r", None).await.map_err(|e| file_error(e.transfer_code(), &format!("{} cannot be read.", job.local_path)))?;1591
let output = match self.open_remote_part(&destination, id).await {1592
Ok(handle) => handle,1593
Err(error) => {1594
let _ = tokio::time::timeout(Duration::from_secs(3), source.close_file(&input)).await;1595
return Err(error);1596
}1597
};1598
let result: RunResult<()> = async {1599
let job = self.snapshot(id)?;1600
if job.transferred > 0 {1601
let written = destination.fstat(&output).await.map_err(sftp_core)?;1602
let transferred = job.transferred.min(written.size);1603
self.update(id, |j| j.transferred = transferred);1604
let window = TAIL_CHECK.min(transferred);1605
let position = transferred - window;1606
verify_tail(1607
transferred,1608
async { source.read_full(&input, position, window as usize).await.map_err(sftp_core) }.boxed(),1609
async { destination.read_full(&output, position, window as usize).await.map_err(sftp_core) }.boxed(),1610
window,1611
)1612
.await?;1613
}1614
self.set(id, "transferring", None);1615
let size = self.with_job(id, |j| j.size).unwrap_or(0);1616
let produce: Producer = {1617
let (source, input) = (source.clone(), input.clone());1618
Arc::new(move |pos, len| {1619
let (source, input) = (source.clone(), input.clone());1620
Box::pin(async move {1621
let data = source.read_full(&input, pos, len).await.map_err(sftp_core)?;1622
if data.len() != len {1623
return Err(file_error("FILE_SOURCE_CHANGED", "The source file shrank during the transfer."));1624
}1625
Ok(data)1626
})1627
})1628
};1629
let consume: Consumer = {1630
let (destination, output) = (destination.clone(), output.clone());1631
Arc::new(move |buf, pos| {1632
let (destination, output) = (destination.clone(), output.clone());1633
Box::pin(async move { destination.write_at(&output, pos, &buf).await.map_err(sftp_core) })1634
})1635
};1636
self.run_pump(id, size, produce, consume, stop.clone()).await?;1637
let written = destination.fstat(&output).await.map_err(sftp_core)?;1638
if written.size > size {1639
destination.truncate(&output, size).await.map_err(sftp_core)?;1640
}1641
let _ = destination.fsync(&output).await;1642
let _ = destination.close_file(&output).await;1643
self.set(id, "verifying", None);1644
let hash = self.relay_hash(&source_session, &source, &input, &job.local_path, size, &stop).await?;1645
let after = source.stat(&job.local_path).await.map_err(sftp_core)?;1646
if (RemoteIdentity { size: after.size, mtime: after.mtime as u64 }) != current {1647
return Err(file_error("FILE_SOURCE_CHANGED", "The source file changed during the transfer.").into());1648
}1649
self.update(id, |j| {1650
j.hash = Some(hash.clone());1651
j.verification = Some(json!({ "method": "sha256", "local": hash }));1652
});1653
Ok(())1654
}1655
.await;1656
let _ = tokio::time::timeout(Duration::from_secs(3), source.close_file(&input)).await;1657
if result.is_err() {1658
let _ = tokio::time::timeout(Duration::from_secs(3), destination.close_file(&output)).await;1659
}1660
drop((source, destination));1661
if let Err(RunError::Error(_)) = &result {1662
// Either end may drop before its disconnect event arrives: that is a lost connection, not a broken file.1663
if [source_session.as_str(), session_id].iter().any(|s| self.sessions.get(s).is_none_or(|r| *r.state.lock() != "connected")) {1664
return Err(file_error("SESSION_NOT_CONNECTED", "A connection of this transfer ended.").into());1665
}1666
}1667
result?;1668
self.finalize_upload(id, session_id, &stop).await1669
}1671
fn relay_session(&self, id: &str) -> RunResult<String> {1672
let job = self.snapshot(id)?;1673
let (preferred, identity) = relay_source(&job).ok_or_else(|| RunError::Error(file_error("TRANSFER_INVALID_SELECTION", "The transfer has no source server.")))?;1674
let session = matching_session(&self.sessions.list(), &identity, &preferred).ok_or_else(|| {1675
RunError::Error(file_error("SESSION_NOT_CONNECTED", &format!("Connect to {}@{} to continue.", identity.username.as_str().unwrap_or(""), identity.host.as_str().unwrap_or(""))))1676
})?;1677
self.update(id, |j| {1678
if let Some(relay) = j.extra.get_mut("relaySource") {1679
relay["sessionId"] = json!(session);1680
}1681
});1682
Ok(session)1683
}1685
/// SHA-256 of the source: on the source server if it can, otherwise read over SFTP here.1686
async fn relay_hash(&self, session: &str, sftp: &Sftp, handle: &FileHandle, path: &str, size: u64, stop: &watch::Receiver<String>) -> RunResult<String> {1687
let auth = Authorization::default().with_timeout(hash_timeout(size));1688
if let Ok(value) = self.files.helper(session, json!({ "operation": "hash", "path": path }), &auth).await {1689
if let Some(hash) = value["hash"].as_str() {1690
return Ok(hash.to_string());1691
}1692
}1693
if let Ok(hash) = remote_checksum(&self.sessions, session, path).await {1694
return Ok(hash);1695
}1696
let mut hasher = Sha256::new();1697
let mut position = 0;1698
while position < size {1699
let kind = stop.borrow().clone();1700
if !kind.is_empty() {1701
return Err(RunError::Stop(kind));1702
}1703
let length = (size - position).min(1024 * 1024) as usize;1704
let data = sftp.read_full(handle, position, length).await.map_err(sftp_core)?;1705
if data.len() != length {1706
return Err(file_error("FILE_SOURCE_CHANGED", "The source file shrank during the transfer.").into());1707
}1708
hasher.update(&data);1709
position += length as u64;1710
}1711
Ok(hex::encode(hasher.finalize()))1712
}1714
/// A move removes its source only after a completed transfer whose SHA-256 matched and only if1715
/// the source is still the file that was copied. In any other case the source stays.1716
async fn finish_move(self: &Arc<Self>, id: &str) {1717
let Ok(job) = self.snapshot(id) else { return };1718
if job.state != "completed" || job.extra.get("moveSource") != Some(&Value::Bool(true)) || job.extra.contains_key("moveResult") {1719
return;1720
}1721
let verified = job.verification.as_ref().is_some_and(|v| v["local"].is_string() && v["local"] == v["remote"]);1722
let result: Result<(), &'static str> = if verified { self.remove_source(&job).await } else { Err("unverified") };1723
if let Err(reason) = result {1724
self.logger.warn("Moved file kept at its source.", json!({ "jobId": id, "reason": reason }));1725
}1726
self.update(id, |j| {1727
j.extra.insert("moveResult".into(), match result {1728
Ok(()) => json!({ "state": "deleted" }),1729
Err(reason) => json!({ "state": "kept", "reason": reason }),1730
});1731
});1732
self.mark_job(id);1733
self.emit_soon();1734
self.persist(true, false);1735
}1737
async fn remove_source(self: &Arc<Self>, job: &Job) -> Result<(), &'static str> {1738
let root = job.extra.get("moveRoot").and_then(Value::as_str).unwrap_or("").to_string();1739
if job.direction == "upload" {1740
let path = Path::new(&job.local_path);1741
let meta = std::fs::symlink_metadata(path).map_err(|_| "source-gone")?;1742
if !meta.is_file() || !job.local_identity.as_ref().is_some_and(|i| i.same(&LocalIdentity::of(&meta))) {1743
return Err("source-changed");1744
}1745
std::fs::remove_file(path).map_err(|_| "remove-failed")?;1746
let mut parent = path.parent();1747
while let Some(directory) = parent.filter(|d| !root.is_empty() && d.starts_with(&root) && *d != Path::new(&root)) {1748
if std::fs::remove_dir(directory).is_err() {1749
break;1750
}1751
parent = directory.parent();1752
}1753
return Ok(());1754
}1755
let (session, path) = if job.direction == "relay" {1756
let (preferred, identity) = relay_source(job).ok_or("source-gone")?;1757
(matching_session(&self.sessions.list(), &identity, &preferred).ok_or("source-disconnected")?, job.local_path.clone())1758
} else {1759
(self.find_session(&job.id).ok_or("source-disconnected")?, job.remote_path.clone())1760
};1761
let sftp = self.files.sftp(&session).await.map_err(|_| "source-disconnected")?;1762
let stat = sftp.lstat(&path).await.map_err(|_| "source-gone")?;1763
if !stat.is_file() || job.remote_identity.as_ref() != Some(&RemoteIdentity { size: stat.size, mtime: stat.mtime as u64 }) {1764
return Err("source-changed");1765
}1766
sftp.unlink(&path).await.map_err(|_| "remove-failed")?;1767
let mut directory = posix::dirname(&path);1768
while !root.is_empty() && directory != root && directory.starts_with(&format!("{}/", root.trim_end_matches('/'))) {1769
if sftp.rmdir(&directory).await.is_err() {1770
break;1771
}1772
directory = posix::dirname(&directory);1773
}1774
Ok(())1775
}1777
/// After every file of a moved batch is gone from its source, removes the source folders, deepest1778
/// first. Removing a folder fails while it holds anything, such as a skipped symlink or a kept file.1779
async fn finish_batch_move(self: &Arc<Self>, batch: &str) {1780
let pending = {1781
let mut inner = self.inner.lock();1782
let done = inner.jobs.values().filter(|j| j.batch_id == batch).all(|j| j.state == "completed" && j.extra.get("moveResult").is_some_and(|r| r["state"] == "deleted"));1783
if !done { return; }1784
inner.batch_moves.remove(batch)1785
};1786
let Some(mut pending) = pending else { return };1787
pending.directories.sort_by_key(|d| std::cmp::Reverse(d.len()));1788
match pending.session {1789
None => {1790
for directory in &pending.directories {1791
let _ = std::fs::remove_dir(directory);1792
}1793
}1794
Some(session) => {1795
if let Ok(sftp) = self.files.sftp(&session).await {1796
for directory in &pending.directories {1797
let _ = sftp.rmdir(directory).await;1798
}1799
}1800
}1801
}1802
}1804
/// A new connection is verified against the trusted key, but that key can have been replaced since1805
/// the job started: then this is possibly another machine, and the job must not write or delete there.1806
fn check_host_keys(&self, id: &str) -> RunResult<()> {1807
let job = self.snapshot(id)?;1808
let session = self.find_session(id);1809
let source = relay_source(&job).and_then(|(preferred, identity)| matching_session(&self.sessions.list(), &identity, &preferred));1810
for (stored, current) in [("hostKey", session), ("sourceHostKey", source)] {1811
let (Some(expected), Some(current)) = (job.extra.get(stored).and_then(Value::as_str), current) else { continue };1812
if self.sessions.trusted_host_key(¤t).as_deref() != Some(expected) {1813
return Err(file_error("TRANSFER_HOST_KEY_CHANGED", "The server's host key changed after this transfer started. Start a new transfer if this is the same server.").into());1814
}1815
}1816
Ok(())1817
}1819
async fn run_pump(self: &Arc<Self>, id: &str, size: u64, produce: Producer, consume: Consumer, stop: watch::Receiver<String>) -> RunResult<()> {1820
let start = self.with_job(id, |j| j.transferred).unwrap_or(0);1821
let manager = self.clone();1822
let job_id = id.to_string();1823
let progress = move |confirmed: u64| manager.progress(&job_id, confirmed);1824
match pump(start, size, produce, consume, stop, progress).await {1825
Ok(_) => Ok(()),1826
Err(PumpStop::Stopped(kind)) => Err(RunError::Stop(kind)),1827
Err(PumpStop::Failed(error)) => Err(RunError::Error(error)),1828
}1829
}1831
async fn prepare_upload_target(self: &Arc<Self>, id: &str, session_id: &str) -> RunResult<()> {1832
let job = self.snapshot(id)?;1833
if job.resolved_path.is_some() && job.part_path.is_some() && job.target_baseline.is_some() {1834
return Ok(());1835
}1836
let auth = self.helper_auth(id, false, None);1837
let stat = match self.files.helper(session_id, json!({ "operation": "stat", "path": job.remote_path }), &auth).await {1838
Ok(stat) => {1839
self.update(id, |j| j.helper = Some(true));1840
stat1841
}1842
Err(error) => {1843
if !NO_HELPER_CODES.contains(&error.code.as_str()) || job.mode == "privileged" {1844
return Err(error.into());1845
}1846
// No python3 / Windows server: plain SFTP is enough for new files; existing ones need1847
// the metadata-preserving Linux helper.1848
self.update(id, |j| j.helper = Some(false));1849
let sftp = self.files.sftp(session_id).await?;1850
if sftp.exists(&job.remote_path).await {1851
let message = if error.code == "FEATURE_PLATFORM_UNSUPPORTED" {1852
"Replacing an existing file on this server is not supported (no metadata-preserving helper); remove it first or choose another name."1853
} else {1854
"Replacing an existing file requires python3 on the server to preserve its owner, permissions and ACLs."1855
};1856
return Err(file_error(&error.code, message).into());1857
}1858
json!({ "exists": false, "resolvedPath": job.remote_path })1859
}1860
};1861
let exists = stat["exists"] == true;1862
if exists && stat["type"] != "file" {1863
return Err(file_error("FILE_UNSAFE_TYPE_OR_LINK", &format!("{} is not a regular file.", job.remote_path)).into());1864
}1865
if exists && number(&stat["baseline"]["nlink"]).unwrap_or(1.0) > 1.0 {1866
return Err(file_error("FILE_HARDLINK_UNSUPPORTED", &format!("{} has several hard links.", job.remote_path)).into());1867
}1868
if exists && !job.overwrite {1869
return Err(file_error("FILE_EXISTS", &format!("{} already exists.", job.remote_path)).into());1870
}1871
let resolved = stat["resolvedPath"].as_str().filter(|s| !s.is_empty()).unwrap_or(&job.remote_path).to_string();1872
let baseline = if exists { stat["baseline"].clone() } else { Value::Null };1873
let helper = self.with_job(id, |j| j.helper == Some(true)).unwrap_or(false);1874
let (owner_uid, part_path) = if job.mode == "privileged" {1875
let workspace = self.files.helper(session_id, json!({ "operation": "transfer-workspace" }), &Authorization::default()).await?;1876
(Some(workspace["uid"].clone()), posix::join(workspace["path"].as_str().unwrap_or(""), &format!("{}.part", job.id)))1877
} else {1878
let uid = if helper { Some(self.files.helper(session_id, json!({ "operation": "transfer-workspace" }), &Authorization::default()).await?["uid"].clone()) } else { None };1879
(uid, posix::join(&posix::dirname(&resolved), &part_name(&posix::basename(&resolved), &job.id)))1880
};1881
self.update(id, |j| {1882
j.resolved_path = Some(resolved.clone());1883
j.target_baseline = Some(baseline.clone());1884
if owner_uid.is_some() {1885
j.owner_uid = owner_uid.clone();1886
}1887
j.part_path = Some(part_path.clone());1888
});1889
self.persist(true, false);1890
Ok(())1891
}1893
async fn open_remote_part(&self, sftp: &Sftp, id: &str) -> RunResult<FileHandle> {1894
let job = self.snapshot(id)?;1895
let part = job.part_path.clone().unwrap_or_default();1896
if job.transferred > 0 {1897
match sftp.lstat(&part).await {1898
Ok(stat) if stat.is_file() => return sftp.open_file(&part, "r+", None).await.map_err(|e| sftp_core(e).into()),1899
_ => self.update(id, |j| j.transferred = 0), // The partial file disappeared; start this file again.1900
}1901
}1902
match sftp.open_file(&part, "wx", Some(0o600)).await {1903
Ok(handle) => Ok(handle),1904
Err(error) => {1905
// Only our own leftover (same job id) can exist here; remove the entry itself, never follow it.1906
if error.transfer_code() == "FILE_PERMISSION_DENIED" {1907
return Err(file_error("FILE_PERMISSION_DENIED", &format!("{} is not writable.", posix::dirname(&part))).into());1908
}1909
let _ = sftp.unlink(&part).await;1910
sftp.open_file(&part, "wx", Some(0o600)).await.map_err(|e| sftp_core(e).into())1911
}1912
}1913
}1915
async fn finalize_upload(self: &Arc<Self>, id: &str, session_id: &str, stop: &watch::Receiver<String>) -> RunResult<()> {1916
let kind = stop.borrow().clone();1917
if !kind.is_empty() {1918
return Err(RunError::Stop(kind));1919
}1920
self.set(id, "finalizing", None);1921
let retry = self.with_job(id, |j| ["unknown", "running"].contains(&j.finalize_state())).unwrap_or(false);1922
self.update(id, |j| j.finalize = json!({ "state": "running", "startedAt": now() }));1923
self.persist(true, false);1924
let job = self.snapshot(id)?;1925
if job.helper == Some(false) {1926
return self.publish_without_helper(id, session_id).await;1927
}1928
let auth = self.helper_auth(id, job.finalize_with_sudo, Some(hash_timeout(job.size)));1929
let request = json!({1930
"operation": "publish",1931
"source": job.part_path,1932
"uid": job.owner_uid,1933
"path": job.remote_path,1934
"hash": job.hash,1935
"size": job.size,1936
"baseline": job.target_baseline.clone().unwrap_or(Value::Null),1937
"overwrite": job.overwrite,1938
"strategy": if job.mode == "privileged" { "copy" } else { "rename" },1939
"parents": true,1940
"retry": retry,1941
"mode": 0o6441942
});1943
let result = self.files.helper(session_id, request, &auth).await?;1944
self.update(id, |j| {1945
j.finalize = json!({1946
"state": "done",1947
"resolvedPath": result["resolvedPath"],1948
"created": result["created"] == true,1949
"metadataPreserved": result["metadataPreserved"] == true,1950
"alreadyPublished": result["alreadyPublished"] == true,1951
"finishedAt": now()1952
});1953
let mut verification = j.verification.clone().unwrap_or_else(|| json!({}));1954
verification["remote"] = result["hash"].clone();1955
verification["method"] = json!("sha256");1956
j.verification = Some(verification);1957
});1958
if result["alreadyPublished"] == true && job.mode == "privileged" {1959
let _ = self.files.helper(session_id, json!({ "operation": "transfer-remove", "id": job.id }), &Authorization::default()).await;1960
}1961
self.inner.lock().secrets.remove(id);1962
self.set(id, "completed", None);1963
self.logger.info("Transfer completed.", json!({ "jobId": id, "direction": "upload", "bytes": job.size, "mode": job.mode, "path": job.remote_path }));1964
Ok(())1965
}1967
async fn publish_without_helper(self: &Arc<Self>, id: &str, session_id: &str) -> RunResult<()> {1968
let job = self.snapshot(id)?;1969
let part = job.part_path.clone().unwrap_or_default();1970
let remote_hash = remote_checksum(&self.sessions, session_id, &part).await.ok();1971
if remote_hash.as_ref().is_some_and(|h| Some(h) != job.hash.as_ref()) {1972
let _ = self.discard_partial(id).await;1973
self.update(id, |j| j.transferred = 0);1974
return Err(file_error("FILE_VERIFY_FAILED", "The uploaded data does not match the local file.").into());1975
}1976
self.update(id, |j| {1977
let mut verification = j.verification.clone().unwrap_or_else(|| json!({}));1978
verification["remote"] = json!(remote_hash);1979
verification["method"] = json!(if remote_hash.is_some() { "sha256" } else { "size" });1980
j.verification = Some(verification);1981
});1982
let sftp = self.files.sftp(session_id).await?;1983
let stat = sftp.stat(&part).await.map_err(sftp_core)?;1984
if stat.size != job.size {1985
return Err(file_error("FILE_VERIFY_FAILED", "The uploaded size does not match the local file.").into());1986
}1987
if sftp.exists(&job.remote_path).await {1988
return Err(file_error("FILE_EXISTS", &format!("{} appeared during the transfer.", job.remote_path)).into());1989
}1990
sftp.rename(&part, &job.remote_path).await.map_err(sftp_core)?;1991
self.update(id, |j| j.finalize = json!({ "state": "done", "resolvedPath": j.remote_path, "created": true, "metadataPreserved": false, "finishedAt": now() }));1992
self.set(id, "completed", None);1993
Ok(())1994
}1996
async fn run_download(self: &Arc<Self>, id: &str, session_id: &str, stop: watch::Receiver<String>) -> RunResult<()> {1997
let sftp = self.open_channel(session_id, &stop).await?;1998
let job = self.snapshot(id)?;1999
let mut source = job.remote_path.clone();2000
if job.mode == "privileged" {2001
source = self.ensure_staged(id, &sftp, session_id).await?;2002
} else {2003
let stat = sftp.stat(&job.remote_path).await.map_err(|e| file_error(e.transfer_code(), &format!("{} is not accessible.", job.remote_path)))?;2004
let current = RemoteIdentity { size: stat.size, mtime: stat.mtime as u64 };2005
if job.remote_identity.as_ref().is_some_and(|r| *r != current) && job.transferred > 0 {2006
return Err(file_error("FILE_SOURCE_CHANGED", "The remote file changed since the transfer started; restart it from the beginning.").into());2007
}2008
self.update(id, |j| {2009
j.remote_identity = Some(current.clone());2010
j.size = current.size;2011
j.remote_mode = Some(json!(stat.mode & 0o777));2012
});2013
}2014
let handle = match sftp.open_file(&source, "r", None).await {2015
Ok(handle) => handle,2016
Err(error) if error.transfer_code() == "FILE_PERMISSION_DENIED" && self.with_job(id, |j| j.mode == "privileged-allowed").unwrap_or(false) => {2017
self.update(id, |j| j.mode = "privileged".into());2018
source = self.ensure_staged(id, &sftp, session_id).await?;2019
sftp.open_file(&source, "r", None).await.map_err(sftp_core)?2020
}2021
Err(error) => return Err(file_error(error.transfer_code(), &format!("{} cannot be read.", job.remote_path)).into()),2022
};2023
let local_part = {2024
let job = self.snapshot(id)?;2025
job.local_part.clone().unwrap_or_else(|| {2026
let local = Path::new(&job.local_path);2027
local.with_file_name(part_name(&local.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default(), &job.id)).to_string_lossy().to_string()2028
})2029
};2030
self.update(id, |j| j.local_part = Some(local_part.clone()));2031
let result: RunResult<()> = async {2032
let local = self.open_local_part(id, &local_part)?;2033
let transferred = self.with_job(id, |j| j.transferred).unwrap_or(0);2034
if transferred > 0 {2035
let window = TAIL_CHECK.min(transferred);2036
let position = transferred - window;2037
verify_tail(2038
transferred,2039
async { local.read_at(position, window as usize).await.map_err(io_core) }.boxed(),2040
async { sftp.read_full(&handle, position, window as usize).await.map_err(sftp_core) }.boxed(),2041
window,2042
)2043
.await?;2044
}2045
self.set(id, "transferring", None);2046
let size = self.with_job(id, |j| j.size).unwrap_or(0);2047
let produce: Producer = {2048
let (sftp, handle) = (sftp.clone(), handle.clone());2049
Arc::new(move |pos, len| {2050
let (sftp, handle) = (sftp.clone(), handle.clone());2051
Box::pin(async move {2052
let data = sftp.read_full(&handle, pos, len).await.map_err(sftp_core)?;2053
if data.len() != len {2054
return Err(file_error("FILE_SOURCE_CHANGED", "The remote file shrank during the transfer."));2055
}2056
Ok(data)2057
})2058
})2059
};2060
let consume: Consumer = {2061
let local = local.clone();2062
Arc::new(move |buf, pos| {2063
let local = local.clone();2064
Box::pin(async move { local.write_at(pos, buf).await.map_err(io_core) })2065
})2066
};2067
self.run_pump(id, size, produce, consume, stop.clone()).await?;2068
local.finish(size).await.map_err(io_core)?;2069
drop(local);2070
let _ = sftp.close_file(&handle).await;2071
self.set(id, "verifying", None);2072
let hash = local_hash(Path::new(&local_part), stop.clone()).await?;2073
let job = self.snapshot(id)?;2074
let mut verification = json!({ "method": "size", "local": hash });2075
if job.mode == "privileged" {2076
verification = json!({ "method": "sha256", "local": hash, "remote": job.staged.as_ref().map(|s| s["hash"].clone()).unwrap_or(Value::Null) });2077
} else {2078
let stat = sftp.stat(&job.remote_path).await.map_err(sftp_core)?;2079
let identity = job.remote_identity.clone().unwrap_or_default();2080
if stat.size != identity.size || stat.mtime as u64 != identity.mtime {2081
return Err(file_error("FILE_SOURCE_CHANGED", "The remote file changed during the transfer; restart it from the beginning.").into());2082
}2083
if self.inner.lock().settings.verify {2084
let remote = match self.files.helper(session_id, json!({ "operation": "hash", "path": job.remote_path }), &Authorization::default().with_timeout(hash_timeout(job.size))).await {2085
Ok(value) => value["hash"].as_str().map(str::to_string),2086
Err(_) => remote_checksum(&self.sessions, session_id, &job.remote_path).await.ok(),2087
};2088
if let Some(remote) = remote {2089
verification = json!({ "method": "sha256", "local": hash, "remote": remote });2090
}2091
}2092
}2093
self.update(id, |j| {2094
j.hash = Some(hash.clone());2095
j.verification = Some(verification.clone());2096
});2097
if verification["remote"].as_str().is_some_and(|r| r != hash) {2098
let _ = self.discard_partial(id).await;2099
self.update(id, |j| j.transferred = 0);2100
return Err(file_error("FILE_VERIFY_FAILED", "The downloaded data does not match the remote file.").into());2101
}2102
Ok(())2103
}2104
.await;2105
if result.is_err() {2106
let _ = tokio::time::timeout(Duration::from_secs(3), sftp.close_file(&handle)).await;2107
}2108
drop(sftp);2109
result?;2110
let kind = stop.borrow().clone();2111
if !kind.is_empty() {2112
return Err(RunError::Stop(kind));2113
}2114
self.set(id, "finalizing", None);2115
let job = self.snapshot(id)?;2116
if std::fs::symlink_metadata(&job.local_path).is_ok() && !job.overwrite {2117
return Err(file_error("FILE_EXISTS", &format!("{} already exists.", job.local_path)).into());2118
}2119
#[cfg(unix)]2120
{2121
use std::os::unix::fs::PermissionsExt;2122
let remote_mode = job.remote_mode.as_ref().and_then(|m| number(m)).map(|m| m as u32).unwrap_or(0o644);2123
let mode = if job.mode == "privileged" { 0o600 } else { (remote_mode & 0o755) | 0o600 };2124
let _ = std::fs::set_permissions(&local_part, std::fs::Permissions::from_mode(mode));2125
}2126
std::fs::rename(&local_part, &job.local_path).map_err(io_core)?;2127
if job.mode == "privileged" && self.files.helper(session_id, json!({ "operation": "transfer-remove", "id": job.id }), &Authorization::default()).await.is_err() {2128
self.update(id, |j| j.cleanup_pending = true);2129
}2130
self.update(id, |j| j.finalize = json!({ "state": "done", "finishedAt": now() }));2131
self.inner.lock().secrets.remove(id);2132
self.set(id, "completed", None);2133
self.logger.info("Transfer completed.", json!({ "jobId": id, "direction": "download", "bytes": job.size, "mode": job.mode }));2134
Ok(())2135
}2137
async fn ensure_staged(self: &Arc<Self>, id: &str, sftp: &Sftp, session_id: &str) -> RunResult<String> {2138
let job = self.snapshot(id)?;2139
if let Some(staged) = &job.staged {2140
let path = staged["path"].as_str().unwrap_or("");2141
if sftp.stat(path).await.is_ok_and(|s| Some(s.size as f64) == number(&staged["size"])) {2142
return Ok(path.to_string());2143
}2144
self.update(id, |j| j.transferred = 0);2145
}2146
self.set(id, "connecting", None);2147
let workspace = self.files.helper(session_id, json!({ "operation": "transfer-workspace" }), &Authorization::default()).await?;2148
let auth = self.helper_auth(id, true, Some(hash_timeout(job.size)));2149
let staged = self.files.helper(session_id, json!({ "operation": "stage-out", "path": job.remote_path, "id": job.id, "uid": workspace["uid"] }), &auth).await?;2150
let path = staged["path"].as_str().unwrap_or("").to_string();2151
self.update(id, |j| {2152
j.staged = Some(json!({ "path": staged["path"], "size": staged["size"], "hash": staged["hash"], "mtimeNs": staged["mtimeNs"] }));2153
j.size = number(&staged["size"]).unwrap_or(0.0) as u64;2154
j.transferred = 0;2155
});2156
self.persist(true, false);2157
Ok(path)2158
}2160
fn open_local_part(&self, id: &str, part: &str) -> RunResult<LocalFile> {2161
let transferred = self.with_job(id, |j| j.transferred).unwrap_or(0);2162
if transferred > 0 {2163
match std::fs::symlink_metadata(part) {2164
Ok(meta) if meta.is_file() => {2165
self.update(id, |j| j.transferred = j.transferred.min(meta.len()));2166
return LocalFile::open(Path::new(part), "r+").map_err(|e| io_core(e).into());2167
}2168
_ => self.update(id, |j| j.transferred = 0),2169
}2170
}2171
let _ = std::fs::remove_file(part);2172
LocalFile::open(Path::new(part), "wx").map_err(|e| io_core(e).into())2173
}2175
// ---- cancel / cleanup ------------------------------------------------------------------------2177
async fn finish_cancel(self: &Arc<Self>, id: &str) {2178
self.inner.lock().secrets.remove(id);2179
let cleaned = self.discard_partial(id).await.is_ok();2180
self.update(id, |j| j.cleanup_pending = !cleaned);2181
self.set(id, "cancelled", None);2182
}2184
/// Removes only temporary data this job created; existing destination files are never touched.2185
async fn discard_partial(&self, id: &str) -> CoreResult<()> {2186
let Some(job) = self.with_job(id, |j| j.clone()) else { return Ok(()) };2187
if job.direction == "download" {2188
if let Some(part) = &job.local_part {2189
match std::fs::remove_file(part) {2190
Ok(()) => {}2191
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}2192
Err(e) => return Err(io_core(e)),2193
}2194
}2195
if job.staged.is_some() {2196
let session = self.find_session(id).ok_or_else(|| file_error("SESSION_NOT_CONNECTED", "Staged data is removed after reconnecting."))?;2197
self.files.helper(&session, json!({ "operation": "transfer-remove", "id": job.id }), &Authorization::default()).await?;2198
self.update(id, |j| j.staged = None);2199
}2200
return Ok(());2201
}2202
let Some(part) = job.part_path.clone() else { return Ok(()) };2203
let session = self.find_session(id).ok_or_else(|| file_error("SESSION_NOT_CONNECTED", "Partial data is removed after reconnecting."))?;2204
if job.mode == "privileged" {2205
self.files.helper(&session, json!({ "operation": "transfer-remove", "id": job.id }), &Authorization::default()).await?;2206
} else {2207
let sftp = self.files.sftp(&session).await?;2208
if let Ok(stat) = sftp.lstat(&part).await {2209
if !stat.is_dir() {2210
sftp.unlink(&part).await.map_err(sftp_core)?;2211
}2212
}2213
}2214
self.update(id, |j| j.part_path = None);2215
Ok(())2216
}2218
fn cleanup_orphans_later(self: &Arc<Self>) {2219
if self.cleanup_scheduled.swap(true, Ordering::SeqCst) {2220
return;2221
}2222
let manager = self.clone();2223
tokio::spawn(async move {2224
tokio::task::yield_now().await;2225
manager.cleanup_scheduled.store(false, Ordering::SeqCst);2226
manager.cleanup_orphans().await;2227
});2228
}2230
/// After a restart: delete temporary transfer data that no resumable job owns any more.2231
async fn cleanup_orphans(self: &Arc<Self>) {2232
for session in self.sessions.list() {2233
if session["state"] != "connected" {2234
continue;2235
}2236
let session_id = session["sessionId"].as_str().unwrap_or("").to_string();2237
let identity = SessionIdentity::of_status(&session).key();2238
if !self.inner.lock().cleaned_identities.insert(identity.clone()) {2239
continue;2240
}2241
let pending: Vec<String> = self.inner.lock().jobs.values().filter(|j| j.identity.key() == identity && j.cleanup_pending).map(|j| j.id.clone()).collect();2242
for id in pending {2243
if self.discard_partial(&id).await.is_ok() {2244
self.update(&id, |j| j.cleanup_pending = false);2245
self.mark_job(&id);2246
}2247
}2248
let keep: HashSet<String> = self.inner.lock().jobs.values().filter(|j| j.identity.key() == identity && !FINAL.contains(&j.state.as_str())).map(|j| j.id.clone()).collect();2249
let Ok(listed) = self.files.helper(&session_id, json!({ "operation": "transfer-list" }), &Authorization::default()).await else { continue };2250
for item in listed["items"].as_array().cloned().unwrap_or_default() {2251
let item_id = item["id"].as_str().unwrap_or("").to_string();2252
let known = self.with_job(&item_id, |j| FINAL.contains(&j.state.as_str()));2253
let abandoned = match known {2254
Some(final_state) => final_state,2255
None => now() - (number(&item["mtime"]).unwrap_or(0.0) * 1000.0) as i64 > ORPHAN_AGE_MS,2256
};2257
if keep.contains(&item_id) || !abandoned {2258
continue;2259
}2260
let _ = self.files.helper(&session_id, json!({ "operation": "transfer-remove", "id": item_id }), &Authorization::default()).await;2261
self.logger.info("Abandoned transfer data removed.", json!({ "sessionId": session_id, "id": item_id, "bytes": item["size"] }));2262
}2263
self.persist(false, false);2264
}2265
}2267
// ---- state, events, persistence ----------------------------------------------------------------2269
fn authorize(&self, id: &str, options: &Value) {2270
let mut inner = self.inner.lock();2271
let password = options["sudoPassword"].as_str().filter(|s| !s.is_empty()).map(str::to_string);2272
if let Some(job) = inner.jobs.get_mut(id) {2273
if options["sudo"] == true || password.is_some() {2274
// Data is already next to the target: only the metadata-preserving replacement needs root.2275
if job.direction != "download" && job.mode == "direct" && job.transferred > 0 && job.transferred == job.size {2276
job.finalize_with_sudo = true;2277
}2278
if job.direction == "download" && job.mode == "direct" {2279
job.mode = "privileged-allowed".into();2280
}2281
if job.direction != "download" && job.mode == "direct" && job.transferred == 0 && job.error.as_ref().is_some_and(|e| AUTH_CODES.contains(&e.code.as_str())) {2282
job.mode = "privileged".into();2283
job.part_path = None;2284
job.target_baseline = None;2285
}2286
}2287
}2288
if let Some(password) = password {2289
inner.secrets.insert(id.to_string(), Zeroizing::new(password));2290
}2291
}2293
fn set(self: &Arc<Self>, id: &str, state: &str, error: Option<JobError>) {2294
if !STATES.contains(&state) {2295
return;2296
}2297
{2298
let mut inner = self.inner.lock();2299
let t = now();2300
let Inner { jobs, runtime, .. } = &mut *inner;2301
let Some(job) = jobs.get_mut(id) else { return };2302
let rt = runtime.get_mut(id);2303
if job.state == "transferring" {2304
if let Some(rt) = rt.as_deref() {2305
if let Some(since) = rt.active_since {2306
job.active_ms += (t - since).max(0) as u64;2307
}2308
}2309
}2310
if let Some(rt) = rt {2311
if job.state == "transferring" {2312
rt.active_since = None;2313
}2314
if state == "transferring" {2315
rt.active_since = Some(t);2316
rt.last_time = t;2317
rt.last_bytes = job.transferred;2318
}2319
}2320
job.state = state.into();2321
job.error = error;2322
job.updated_at = t;2323
if state == "transferring" && job.started_at.is_none() {2324
job.started_at = Some(t);2325
}2326
if FINAL.contains(&state) {2327
job.finished_at = Some(t);2328
}2329
let owner = job.owner.clone();2330
Self::mark(&mut inner, id, &owner);2331
}2332
self.emit_soon();2333
self.persist(!RUNNING.contains(&state), false);2334
}2336
fn progress(self: &Arc<Self>, id: &str, confirmed: u64) {2337
{2338
let mut inner = self.inner.lock();2339
let t = now();2340
let Inner { jobs, runtime, .. } = &mut *inner;2341
let Some(job) = jobs.get_mut(id) else { return };2342
let moved = confirmed.saturating_sub(job.transferred);2343
job.transferred = confirmed;2344
job.moved_bytes += moved;2345
if let Some(rt) = runtime.get_mut(id) {2346
rt.progress_since_resume += moved;2347
if rt.progress_since_resume > 1024 * 1024 {2348
job.attempts = 0;2349
}2350
let elapsed = t - rt.last_time;2351
if elapsed >= 500 {2352
let instant = (job.transferred.saturating_sub(rt.last_bytes)) as f64 * 1000.0 / elapsed as f64;2353
rt.speed = if rt.samples > 0 { 0.3 * instant + 0.7 * rt.speed } else { instant };2354
rt.samples += 1;2355
rt.last_time = t;2356
rt.last_bytes = job.transferred;2357
}2358
}2359
job.updated_at = t;2360
let owner = job.owner.clone();2361
Self::mark(&mut inner, id, &owner);2362
}2363
self.emit_soon();2364
self.persist(false, false);2365
}2367
fn public(inner: &Inner, job: &Job) -> Value {2368
let rt = inner.runtime.get(&job.id);2369
let t = now();2370
let mut speed = 0.0;2371
let mut eta = Value::Null;2372
if job.state == "transferring" {2373
if let Some(rt) = rt {2374
let idle = t - rt.last_time;2375
speed = if idle > 2000 { rt.speed * 2000.0 / idle as f64 } else { rt.speed };2376
if rt.samples >= 3 && speed > 1.0 {2377
eta = json!((job.size.saturating_sub(job.transferred) as f64 / speed).ceil() as u64);2378
}2379
}2380
}2381
let active_ms = job.active_ms + if job.state == "transferring" { rt.and_then(|r| r.active_since).map(|s| (t - s).max(0) as u64).unwrap_or(0) } else { 0 };2382
let state = job.state.as_str();2383
let overwrite_error = job.error.as_ref().is_some_and(|e| e.code == "FILE_EXISTS" || e.code == "FILE_CONFLICT");2384
json!({2385
"id": job.id,2386
"batchId": job.batch_id,2387
"sessionId": job.session_id,2388
"identity": job.identity,2389
"direction": job.direction,2390
"name": job.name,2391
"state": state,2392
"source": if job.direction != "download" { &job.local_path } else { &job.remote_path },2393
"target": if job.direction != "download" { &job.remote_path } else { &job.local_path },2394
"resolvedTarget": job.finalize.get("resolvedPath").filter(|v| v.is_string()).cloned().or_else(|| job.resolved_path.clone().map(Value::from)).unwrap_or(Value::Null),2395
"mode": if job.mode == "privileged" { "privileged" } else { "direct" },2396
"size": job.size,2397
"transferred": job.transferred,2398
"percent": percent(job.transferred, job.size),2399
"speed": speed.round() as u64,2400
"averageSpeed": if active_ms > 1000 { json!((job.moved_bytes as f64 * 1000.0 / active_ms as f64).round() as u64) } else { Value::Null },2401
"eta": eta,2402
"finalize": { "state": job.finalize_state(), "metadataPreserved": job.finalize["metadataPreserved"] == true, "created": job.finalize["created"] == true },2403
"verification": job.verification.as_ref().map(|v| json!({ "method": v["method"], "matched": v["remote"].is_string() && v["remote"] == v["local"] })),2404
"error": job.error.as_ref().map(|e| json!({ "code": e.code, "message": e.message })),2405
"authRequired": job.auth_required,2406
"attempts": job.attempts,2407
"nextRetryAt": job.next_retry_at,2408
"cleanupPending": job.cleanup_pending,2409
"sourceIdentity": job.extra.get("relaySource").map(|r| r["identity"].clone()).unwrap_or(Value::Null),2410
"sourceSessionId": job.extra.get("relaySource").map(|r| r["sessionId"].clone()).unwrap_or(Value::Null),2411
"move": job.extra.get("moveSource") == Some(&Value::Bool(true)),2412
"moveResult": job.extra.get("moveResult").cloned().unwrap_or(Value::Null),2413
"actions": {2414
"pause": (["waiting", "reconnecting", "interrupted", "connecting", "transferring", "verifying"].contains(&state)),2415
"resume": (["paused", "interrupted", "reconnecting"].contains(&state) || (state == "failed" && job.resumable)),2416
"retry": (["failed", "interrupted", "paused", "cancelled"].contains(&state)),2417
"cancel": (!FINAL.contains(&state) && state != "finalizing"),2418
"overwrite": (overwrite_error && ["failed", "interrupted"].contains(&state)),2419
"remove": (FINAL.contains(&state) && !job.cleanup_pending)2420
},2421
"createdAt": job.created_at,2422
"updatedAt": job.updated_at,2423
"finishedAt": job.finished_at2424
})2425
}2427
fn mark(inner: &mut Inner, id: &str, owner: &str) {2428
inner.changed.insert(id.to_string());2429
inner.pending_owners.insert(owner.to_string());2430
}2432
fn mark_job(self: &Arc<Self>, id: &str) {2433
{2434
let mut inner = self.inner.lock();2435
let Some(owner) = inner.jobs.get(id).map(|j| j.owner.clone()) else { return };2436
Self::mark(&mut inner, id, &owner);2437
}2438
self.emit_soon();2439
}2441
fn emit_soon(self: &Arc<Self>) {2442
{2443
let mut inner = self.inner.lock();2444
if inner.emit_scheduled {2445
return;2446
}2447
inner.emit_scheduled = true;2448
}2449
let manager = self.clone();2450
tokio::spawn(async move {2451
tokio::time::sleep(Duration::from_millis(EMIT_MS)).await;2452
let updates: Vec<(String, Value)> = {2453
let mut inner = manager.inner.lock();2454
inner.emit_scheduled = false;2455
let owners: Vec<String> = inner.pending_owners.drain().collect();2456
let full: HashSet<String> = inner.full_owners.drain().collect();2457
let changed: HashSet<String> = inner.changed.drain().collect();2458
owners2459
.into_iter()2460
.map(|owner| {2461
let all: Vec<Value> = Self::owned(&inner, &owner).iter().map(|j| Self::public(&inner, j)).collect();2462
let is_full = full.contains(&owner);2463
let jobs: Vec<Value> = if is_full { all.clone() } else { all.iter().filter(|j| changed.contains(j["id"].as_str().unwrap_or("")) || RUNNING.contains(&j["state"].as_str().unwrap_or(""))).cloned().collect() };2464
let removed: Vec<&String> = changed.iter().filter(|id| !inner.jobs.contains_key(*id)).collect();2465
let update = json!({ "owner": owner, "jobs": jobs, "removed": removed, "full": is_full, "summary": summarize(&all), "settings": inner.settings });2466
(owner, update)2467
})2468
.collect()2469
};2470
for (owner, update) in updates {2471
(manager.emitter)(&owner, update);2472
}2473
});2474
}2476
fn owned<'a>(inner: &'a Inner, owner: &str) -> Vec<&'a Job> {2477
let mut jobs: Vec<&Job> = inner.jobs.values().filter(|j| j.owner == owner).collect();2478
jobs.sort_by_key(|j| j.order);2479
jobs2480
}2482
fn select_ids(inner: &Inner, owner: &str, ids: &Value) -> Vec<String> {2483
let owned = Self::owned(inner, owner);2484
if *ids == "all" {2485
return owned.iter().map(|j| j.id.clone()).collect();2486
}2487
let wanted: HashSet<String> = match ids {2488
Value::Array(list) => list.iter().take(20000).map(|v| v.as_str().map(str::to_string).unwrap_or_else(|| v.to_string())).collect(),2489
Value::Null => HashSet::new(),2490
other => [other.as_str().map(str::to_string).unwrap_or_else(|| other.to_string())].into_iter().collect(),2491
};2492
owned.iter().filter(|j| wanted.contains(&j.id)).map(|j| j.id.clone()).collect()2493
}2495
fn select(&self, owner: &str, ids: &Value) -> Vec<String> {2496
Self::select_ids(&self.inner.lock(), owner, ids)2497
}2499
/// Moves a server-to-server job to a new connection to its source server, on request only and2500
/// only when the old connection is gone.2501
fn bind_relay_source(&self, id: &str, options: &Value) -> bool {2502
let Some(requested) = options["sourceSessionId"].as_str() else { return true };2503
let Ok(identity) = self.identity(requested) else { return false };2504
let Some(Some((previous, expected))) = self.with_job(id, relay_source) else { return false };2505
if identity != expected || (previous != requested && self.identity(&previous).is_ok()) {2506
return false;2507
}2508
if let Some(relay) = self.inner.lock().jobs.get_mut(id).and_then(|job| job.extra.get_mut("relaySource")) {2509
relay["sessionId"] = json!(requested);2510
}2511
true2512
}2514
fn bind_session(&self, id: &str, options: &Value) -> bool {2515
if !self.bind_relay_source(id, options) {2516
return false;2517
}2518
let Some(requested) = options["sessionId"].as_str() else { return true };2519
let Ok(identity) = self.identity(requested) else { return false };2520
let Some((previous, expected)) = self.with_job(id, |j| (j.session(), j.identity.clone())) else { return false };2521
if identity != expected || (previous != requested && self.identity(&previous).is_ok()) {2522
return false;2523
}2524
if let Some(job) = self.inner.lock().jobs.get_mut(id) {2525
job.session_id = json!(requested);2526
}2527
true2528
}2530
fn find_session(&self, id: &str) -> Option<String> {2531
let (identity, preferred) = self.with_job(id, |j| (j.identity.clone(), j.session()))?;2532
matching_session(&self.sessions.list(), &identity, &preferred)2533
}2535
fn session_for(&self, id: &str) -> RunResult<String> {2536
self.find_session(id).ok_or_else(|| {2537
let (user, host) = self.with_job(id, |j| (j.identity.username.as_str().unwrap_or("").to_string(), j.identity.host.as_str().unwrap_or("").to_string())).unwrap_or_default();2538
RunError::Error(file_error("SESSION_NOT_CONNECTED", &format!("Connect to {user}@{host} to continue.")))2539
})2540
}2542
fn trim_history(inner: &mut Inner) {2543
let mut finished: Vec<(i64, String)> = inner.jobs.values().filter(|j| FINAL.contains(&j.state.as_str()) && !j.cleanup_pending).map(|j| (j.finished_at.unwrap_or(0), j.id.clone())).collect();2544
if finished.len() <= MAX_HISTORY {2545
return;2546
}2547
finished.sort();2548
let excess = finished.len() - MAX_HISTORY;2549
for (_, id) in finished.into_iter().take(excess) {2550
inner.jobs.remove(&id);2551
inner.changed.insert(id);2552
}2553
}2555
fn with_job<T>(&self, id: &str, f: impl FnOnce(&Job) -> T) -> Option<T> {2556
self.inner.lock().jobs.get(id).map(f)2557
}2559
fn snapshot(&self, id: &str) -> RunResult<Job> {2560
self.with_job(id, |j| j.clone()).ok_or_else(|| RunError::Stop("cancel".into()))2561
}2563
fn update(&self, id: &str, f: impl FnOnce(&mut Job)) {2564
if let Some(job) = self.inner.lock().jobs.get_mut(id) {2565
f(job);2566
}2567
}2569
fn job_state(&self, id: &str) -> String {2570
self.with_job(id, |j| j.state.clone()).unwrap_or_default()2571
}2573
fn snapshot_file(&self) -> Option<(PathBuf, Value)> {2574
let file = self.file.clone()?;2575
let mut inner = self.inner.lock();2576
if inner.closed {2577
return None;2578
}2579
Self::trim_history(&mut inner);2580
// Persisted state never contains passwords, keys or runtime handles.2581
let mut jobs: Vec<&Job> = inner.jobs.values().collect();2582
jobs.sort_by_key(|j| j.order);2583
Some((file, json!({ "version": 1, "settings": inner.settings, "order": inner.order, "jobs": jobs })))2584
}2586
fn persist(self: &Arc<Self>, soon: bool, sync: bool) {2587
if self.file.is_none() {2588
return;2589
}2590
let write = |manager: &TransferManager| {2591
if let Some((file, data)) = manager.snapshot_file() {2592
if let Err(error) = crate::core::storage::atomic_write_json(&file, &data) {2593
manager.logger.warn("Transfer state could not be saved.", json!({ "message": error.message }));2594
}2595
}2596
};2597
if sync {2598
write(self);2599
return;2600
}2601
if !soon && self.persist_pending.swap(true, Ordering::SeqCst) {2602
return;2603
}2604
if soon {2605
self.persist_pending.store(true, Ordering::SeqCst);2606
}2607
let manager = self.clone();2608
tokio::spawn(async move {2609
tokio::time::sleep(Duration::from_millis(if soon { 50 } else { PERSIST_MS })).await;2610
if !manager.persist_pending.swap(false, Ordering::SeqCst) {2611
return;2612
}2613
let _ = tokio::task::spawn_blocking(move || write(&manager)).await;2614
});2615
}2617
fn load(&self) {2618
let Some(file) = &self.file else { return };2619
let Ok(text) = std::fs::read_to_string(file) else { return };2620
let Ok(data) = serde_json::from_str::<Value>(&text) else { return };2621
if data["version"] != 1 || !data["jobs"].is_array() {2622
return;2623
}2624
let mut inner = self.inner.lock();2625
if let Some(settings) = data["settings"].as_object() {2626
if let Some(v) = settings.get("maxConcurrent").and_then(number) {2627
inner.settings.max_concurrent = if v.fract() == 0.0 && (1.0..=8.0).contains(&v) { v as u32 } else { Settings::default().max_concurrent };2628
}2629
if let Some(v) = settings.get("autoRetry").and_then(Value::as_bool) {2630
inner.settings.auto_retry = v;2631
}2632
if let Some(v) = settings.get("verify").and_then(Value::as_bool) {2633
inner.settings.verify = v;2634
}2635
}2636
inner.order = number(&data["order"]).unwrap_or(0.0) as u64;2637
let jobs = data["jobs"].as_array().cloned().unwrap_or_default();2638
let skip = jobs.len().saturating_sub(MAX_JOBS);2639
for raw in jobs.into_iter().skip(skip) {2640
let Ok(mut job) = serde_json::from_value::<Job>(raw) else { continue };2641
if job.id.len() != 32 || !job.id.chars().all(|c| c.is_ascii_digit() || ('a'..='f').contains(&c)) || !STATES.contains(&job.state.as_str()) {2642
continue;2643
}2644
// Nothing restarts on its own after the application was closed: the user is offered to resume.2645
if RUNNING.contains(&job.state.as_str()) || job.state == "waiting" || job.state == "reconnecting" {2646
if job.state == "finalizing" || job.finalize_state() == "running" {2647
if !job.finalize.is_object() {2648
job.finalize = json!({});2649
}2650
job.finalize["state"] = json!("unknown");2651
}2652
job.state = "interrupted".into();2653
job.error = Some(JobError { code: "APP_RESTARTED".into(), message: "The application was closed during the transfer.".into() });2654
}2655
job.next_retry_at = None;2656
job.attempts = 0;2657
inner.order = inner.order.max(job.order);2658
inner.jobs.insert(job.id.clone(), job);2659
}2660
}2661
}2663
struct Regex;2665
impl Regex {2666
/// `String(message).replace(/^[A-Z_]+: /, "")`.2667
fn strip_code(message: &str) -> String {2668
if let Some((prefix, rest)) = message.split_once(": ") {2669
if !prefix.is_empty() && prefix.chars().all(|c| c.is_ascii_uppercase() || c == '_') {2670
return rest.to_string();2671
}2672
}2673
message.to_string()2674
}2675
}2677
#[derive(Debug, Clone)]2678
struct RemoteInfo {2679
kind: &'static str,2680
size: u64,2681
mtime: u64,2682
mode: u32,2683
}2685
fn relay_source(job: &Job) -> Option<(String, SessionIdentity)> {2686
let relay = job.extra.get("relaySource")?;2687
Some((relay["sessionId"].as_str().unwrap_or("").to_string(), serde_json::from_value(relay["identity"].clone()).ok()?))2688
}2690
/// Every name is checked: a hostile source server cannot place a file outside the destination.2691
/// Symlinks and special files are skipped and listed, never followed.2692
fn walk_relay<'a>(sftp: &'a Sftp, plan: &'a mut Plan, source: &'a str, destination: &'a str, depth: u32) -> BoxFuture<'a, CoreResult<()>> {2693
async move {2694
if plan.files.len() + plan.directories.len() > MAX_PLAN_ENTRIES {2695
return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", &format!("A transfer can contain at most {MAX_PLAN_ENTRIES} entries.")));2696
}2697
let stat = sftp.lstat(source).await.map_err(|e| file_error(e.transfer_code(), &format!("{source} is not accessible.")))?;2698
if stat.is_symlink() || (!stat.is_dir() && !stat.is_file()) {2699
plan.skipped.push(json!({ "path": source, "reason": if stat.is_symlink() { "symlink" } else { "type" } }));2700
return Ok(());2701
}2702
if stat.is_file() {2703
plan.bytes += stat.size;2704
plan.files.push(PlanFile {2705
name: posix::basename(destination),2706
local_path: source.to_string(),2707
remote_path: destination.to_string(),2708
size: stat.size,2709
local_identity: None,2710
remote_identity: Some(RemoteIdentity { size: stat.size, mtime: stat.mtime as u64 }),2711
mode: None,2712
conflict: false,2713
overwrite_confirmed: false,2714
});2715
return Ok(());2716
}2717
if depth > 64 {2718
return Err(file_error("TRANSFER_TOO_DEEP", "The selected directory tree is too deep."));2719
}2720
plan.directories.push(destination.to_string());2721
plan.source_directories.push(source.to_string());2722
let entries = sftp.read_dir(source).await.map_err(|e| file_error(e.transfer_code(), &format!("{source} cannot be read.")))?;2723
for entry in entries {2724
if entry.name == "." || entry.name == ".." {2725
continue;2726
}2727
let Ok(name) = safe_entry_name(&entry.name) else {2728
plan.skipped.push(json!({ "path": posix::join(source, &entry.name), "reason": "name" }));2729
continue;2730
};2731
walk_relay(sftp, plan, &posix::join(source, &name), &posix::join(destination, &name), depth + 1).await?;2732
}2733
Ok(())2734
}2735
.boxed()2736
}2738
fn add_download(plan: &mut Plan, remote: &str, local: &str, info: &RemoteInfo) {2739
plan.bytes += info.size;2740
plan.files.push(PlanFile {2741
name: posix::basename(remote),2742
local_path: local.to_string(),2743
remote_path: remote.to_string(),2744
size: info.size,2745
local_identity: None,2746
remote_identity: Some(RemoteIdentity { size: info.size, mtime: info.mtime }),2747
mode: Some(info.mode & 0o777),2748
conflict: false,2749
overwrite_confirmed: false,2750
});2751
}2753
#[derive(Default)]2754
struct LocalWalk {2755
files: Vec<PlanFile>,2756
directories: Vec<String>,2757
source_directories: Vec<String>,2758
skipped: Vec<Value>,2759
bytes: u64,2760
}2762
impl LocalWalk {2763
fn visit(&mut self, local: &Path, remote: &str, depth: u32) -> CoreResult<()> {2764
if self.files.len() + self.directories.len() > MAX_PLAN_ENTRIES {2765
return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", &format!("A transfer can contain at most {MAX_PLAN_ENTRIES} entries.")));2766
}2767
let meta = std::fs::symlink_metadata(local).map_err(io_core)?;2768
let shown = local.to_string_lossy().to_string();2769
if meta.file_type().is_symlink() {2770
self.skipped.push(json!({ "path": shown, "reason": "symlink" }));2771
return Ok(());2772
}2773
if meta.is_dir() {2774
self.directories.push(remote.to_string());2775
self.source_directories.push(shown);2776
if depth > 64 {2777
return Err(file_error("TRANSFER_TOO_DEEP", "The selected directory tree is too deep."));2778
}2779
for entry in std::fs::read_dir(local).map_err(io_core)? {2780
let entry = entry.map_err(io_core)?;2781
let raw = entry.file_name().to_string_lossy().to_string();2782
let Ok(name) = safe_entry_name(&raw) else {2783
self.skipped.push(json!({ "path": local.join(&raw).to_string_lossy(), "reason": "name" }));2784
continue;2785
};2786
self.visit(&local.join(&name), &posix::join(remote, &name), depth + 1)?;2787
}2788
return Ok(());2789
}2790
if !meta.is_file() {2791
self.skipped.push(json!({ "path": shown, "reason": "type" }));2792
return Ok(());2793
}2794
self.bytes += meta.len();2795
self.files.push(PlanFile {2796
name: local.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default(),2797
local_path: shown,2798
remote_path: remote.to_string(),2799
size: meta.len(),2800
local_identity: Some(LocalIdentity::of(&meta)),2801
remote_identity: None,2802
mode: None,2803
conflict: false,2804
overwrite_confirmed: false,2805
});2806
Ok(())2807
}2808
}2810
/// Writable by the SSH account: direct SFTP; otherwise the privileged staging route.2811
async fn directory_mode(sftp: &Sftp, directory: &str) -> CoreResult<&'static str> {2812
let stat: Stat = sftp.stat(directory).await.map_err(|e| {2813
let code = e.transfer_code();2814
file_error(if code == "FILE_NOT_FOUND" { "FILE_NOT_FOUND" } else { code }, &format!("{directory} is not accessible."))2815
})?;2816
if !stat.is_dir() {2817
return Err(file_error("FILE_NOT_DIRECTORY", &format!("{directory} is not a directory.")));2818
}2819
let probe = posix::join(directory, &format!(".chj-probe-{}", hex_id(6)));2820
match sftp.open_file(&probe, "wx", Some(0o600)).await {2821
Ok(handle) => {2822
let _ = sftp.close_file(&handle).await;2823
let _ = sftp.unlink(&probe).await;2824
Ok("direct")2825
}2826
Err(error) if error.transfer_code() == "FILE_PERMISSION_DENIED" => Ok("privileged"),2827
Err(error) => {2828
let code = error.transfer_code();2829
Err(file_error(if code == "TRANSFER_FAILED" { "FILE_REMOTE_IO_FAILED" } else { code }, &format!("{directory} is not writable.")))2830
}2831
}2832
}2834
async fn remote_conflicts(sftp: &Arc<Sftp>, plan: &mut Plan) {2835
let paths: Vec<(usize, String)> = plan.files.iter().enumerate().map(|(i, f)| (i, f.remote_path.clone())).collect();2836
let checks = futures::stream::iter(paths.into_iter().map(|(i, path)| {2837
let sftp = sftp.clone();2838
async move { (i, sftp.lstat(&path).await.ok()) }2839
}))2840
.buffer_unordered(16)2841
.collect::<Vec<_>>()2842
.await;2843
let mut blocked = HashSet::new();2844
for (i, stat) in checks {2845
let Some(stat) = stat else { continue };2846
if stat.is_dir() {2847
blocked.insert(i);2848
plan.skipped.push(json!({ "path": plan.files[i].remote_path, "reason": "directory-exists" }));2849
} else {2850
plan.files[i].conflict = true;2851
}2852
}2853
let mut index = 0;2854
plan.files.retain(|_| {2855
let keep = !blocked.contains(&index);2856
index += 1;2857
keep2858
});2859
}2861
async fn local_hash(path: &Path, stop: watch::Receiver<String>) -> RunResult<String> {2862
let path = path.to_path_buf();2863
let result = tokio::task::spawn_blocking(move || -> Result<String, RunError> {2864
use std::io::Read;2865
let mut file = std::fs::File::open(&path).map_err(|e| RunError::Error(io_core(e)))?;2866
let mut digest = Sha256::new();2867
let mut buffer = vec![0u8; 1024 * 1024];2868
loop {2869
let kind = stop.borrow().clone();2870
if !kind.is_empty() {2871
return Err(RunError::Stop(kind));2872
}2873
let n = file.read(&mut buffer).map_err(|e| RunError::Error(io_core(e)))?;2874
if n == 0 {2875
break;2876
}2877
digest.update(&buffer[..n]);2878
}2879
Ok(hex::encode(digest.finalize()))2880
})2881
.await2882
.map_err(|_| RunError::Error(CoreError::plain("Local hashing failed.")))?;2883
result2884
}2886
#[cfg(test)]2887
mod tests {2888
use super::*;2890
#[test]2891
fn transfers_never_choose_another_connection_to_the_same_server() {2892
let a = json!({ "sessionId": "A", "state": "connected", "profileId": "p", "host": "h", "port": 22, "username": "u" });2893
let mut b = a.clone(); b["sessionId"] = json!("B");2894
let identity = SessionIdentity::of_status(&a);2895
assert_eq!(matching_session(&[a.clone(), b.clone()], &identity, "A"), Some("A".into()));2896
assert_eq!(matching_session(&[b.clone()], &identity, "A"), None);2897
assert_eq!(matching_session(&[a.clone(), b.clone()], &identity, "B"), Some("B".into()));2898
b["host"] = json!("another-host");2899
assert_eq!(matching_session(&[b], &identity, "B"), None);2900
let mut disconnected = a; disconnected["state"] = json!("disconnected");2901
assert_eq!(matching_session(&[disconnected], &identity, "A"), None);2902
}2904
/// A `transfers.json` written by 0.1.0 loads, keeps unknown fields and comes back unchanged2905
/// (except the restart transition 0.1.0 applies on load as well).2906
#[test]2907
fn loads_and_rewrites_0_1_0_state() {2908
let legacy = json!({2909
"version": 1,2910
"settings": { "maxConcurrent": 3, "autoRetry": false, "verify": true },2911
"order": 7,2912
"jobs": [{2913
"id": "0123456789abcdef0123456789abcdef", "owner": "chj.file-manager", "batchId": "b", "direction": "upload",2914
"identity": { "profileId": "p", "host": "h", "port": 22, "username": "u" }, "sessionId": "s1", "order": 7,2915
"name": "a.bin", "localPath": "/tmp/a.bin", "remotePath": "/srv/a.bin", "size": 10, "transferred": 4, "movedBytes": 4, "activeMs": 12,2916
"state": "transferring", "mode": "direct", "overwrite": false,2917
"localIdentity": { "size": 10, "mtimeMs": 1700000000123.0, "ino": 42, "dev": 16777220 },2918
"remoteIdentity": null, "remoteMode": null, "finalize": { "state": "pending" }, "attempts": 2, "totalRetries": 3, "error": null,2919
"createdAt": 1, "updatedAt": 2, "resolvedPath": "/srv/a.bin", "partPath": "/srv/.a.bin.chj-part-x", "targetBaseline": null,2920
"helper": true, "ownerUid": 1000, "futureField": { "kept": true }2921
}]2922
});2923
let dir = tempfile::tempdir().unwrap();2924
std::fs::write(dir.path().join("transfers.json"), serde_json::to_string(&legacy).unwrap()).unwrap();2925
let job: Job = serde_json::from_value(legacy["jobs"][0].clone()).unwrap();2926
assert_eq!(job.target_baseline, Some(Value::Null));2927
assert_eq!(job.local_identity.as_ref().unwrap().mtime_ms, 1700000000123);2928
let back = serde_json::to_value(&job).unwrap();2929
assert_eq!(back["futureField"], json!({ "kept": true }));2930
assert_eq!(back["targetBaseline"], Value::Null);2931
assert_eq!(back["ownerUid"], 1000);2932
let mut fresh = job.clone();2933
fresh.target_baseline = None;2934
assert!(serde_json::to_value(&fresh).unwrap().get("targetBaseline").is_none());2935
assert_eq!(Regex::strip_code("FILE_EXISTS: /x already exists."), "/x already exists.");2936
assert_eq!(part_name("abc", "id1"), ".abc.chj-part-id1");2937
}2939
#[test]2940
fn summary_matches_0_1_0_rules() {2941
let jobs = vec![json!({ "state": "completed", "size": 10, "transferred": 10 }), json!({ "state": "transferring", "size": 10, "transferred": 5, "speed": 5, "eta": 1 })];2942
let s = summarize(&jobs);2943
assert_eq!(s["percent"], 75.0);2944
assert_eq!(s["eta"], 1);2945
let done = summarize(&[json!({ "state": "completed", "size": 0, "transferred": 0 })]);2946
assert_eq!(done["percent"], 100.0);2947
}2948
}2950
#[cfg(test)]2951
#[path = "../../../tests/support/transfer_route_identity.rs"]2952
mod route_identity_tests;2954
#[cfg(test)]2955
#[path = "../../../tests/support/transfer_safety.rs"]2956
mod safety_tests;SHA-256
173d098a667c781da79b526b2fa4d6ccbfa9f46cd15102a189d8fa3f58f915a3CH-J Proprietary Software License 1.14 · Zdroje jsou zveřejněny pod CH-J Proprietary Software License 1.14. Jejich dostupnost nemění licenční podmínky ani neposkytuje další oprávnění.
SHA-256 archivu
77380d4805485ffa4027fbe7858b5836f197a4aa9d52358a43e2d12d018744df