Публикуемый источник
Загрузка источника ZIP
CH-J Tools
Просмотр каталогов и файлов для конкретного выпуска приложения.
CH-J / Server Manager
1704 файлы
0.1.2-beta
Добавлено: · Построено:
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
pub(crate) fn copy_download_options(options: &Value) -> Value {54
let mut options = options.as_object().cloned().unwrap_or_default();55
options.insert("noSaveDialog".into(), json!(true));56
Value::Object(options)57
}59
fn now() -> i64 {60
chrono::Utc::now().timestamp_millis()61
}63
fn hex_id(bytes: usize) -> String {64
let mut buffer = vec![0u8; bytes];65
rand::rngs::OsRng.fill_bytes(&mut buffer);66
hex::encode(buffer)67
}69
fn part_name(name: &str, id: &str) -> String {70
format!(".{}.chj-part-{id}", name.chars().take(120).collect::<String>())71
}73
fn hash_timeout(size: u64) -> u64 {74
120_000 + (size as f64 / HASH_RATE as f64 * 1000.0).ceil() as u6475
}77
// --- lenient JSON numbers (0.1.0 wrote JavaScript numbers) ------------------------------------------79
fn number(value: &Value) -> Option<f64> {80
match value {81
Value::Number(n) => n.as_f64(),82
Value::String(s) => s.parse().ok(),83
_ => None,84
}85
}87
fn de_u64<'de, D: Deserializer<'de>>(d: D) -> Result<u64, D::Error> {88
Ok(number(&Value::deserialize(d)?).filter(|n| n.is_finite() && *n >= 0.0).map(|n| n as u64).unwrap_or(0))89
}90
fn de_i64<'de, D: Deserializer<'de>>(d: D) -> Result<i64, D::Error> {91
Ok(number(&Value::deserialize(d)?).filter(|n| n.is_finite()).map(|n| n as i64).unwrap_or(0))92
}93
fn de_opt_i64<'de, D: Deserializer<'de>>(d: D) -> Result<Option<i64>, D::Error> {94
Ok(number(&Value::deserialize(d)?).filter(|n| n.is_finite()).map(|n| n as i64))95
}96
fn de_u32<'de, D: Deserializer<'de>>(d: D) -> Result<u32, D::Error> {97
Ok(number(&Value::deserialize(d)?).filter(|n| n.is_finite() && *n >= 0.0).map(|n| n as u32).unwrap_or(0))98
}99
/// Keeps JSON `null` as `Some(Null)` (0.1.0 distinguishes `undefined` from `null`).100
fn de_present<'de, D: Deserializer<'de>>(d: D) -> Result<Option<Value>, D::Error> {101
Ok(Some(Value::deserialize(d)?))102
}103
fn de_opt_string<'de, D: Deserializer<'de>>(d: D) -> Result<Option<String>, D::Error> {104
Ok(match Value::deserialize(d)? {105
Value::String(s) => Some(s),106
_ => None,107
})108
}110
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]111
#[serde(rename_all = "camelCase")]112
pub struct SessionIdentity {113
#[serde(default)]114
pub profile_id: Value,115
#[serde(default)]116
pub host: Value,117
#[serde(default)]118
pub port: Value,119
#[serde(default)]120
pub username: Value,121
#[serde(default)]122
pub jump: Vec<Value>,123
}125
impl SessionIdentity {126
fn key(&self) -> String {127
let profile = self.profile_id.as_str().filter(|id| !id.starts_with("adhoc:")).unwrap_or("");128
serde_json::to_string(&json!([profile, self.host, self.port, self.username, self.jump])).unwrap_or_default()129
}130
fn of_status(status: &Value) -> SessionIdentity {131
SessionIdentity {132
profile_id: status["profileId"].clone(), host: status["host"].clone(), port: status["port"].clone(), username: status["username"].clone(),133
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(),134
}135
}136
}138
fn matching_session(sessions: &[Value], identity: &SessionIdentity, preferred: &str) -> Option<String> {139
sessions.iter().find(|s| s["state"] == "connected" && s["sessionId"] == preferred && SessionIdentity::of_status(s) == *identity)140
.and_then(|s| s["sessionId"].as_str().map(str::to_string))141
}143
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]144
#[serde(rename_all = "camelCase")]145
pub struct LocalIdentity {146
#[serde(default, deserialize_with = "de_u64")]147
pub size: u64,148
#[serde(default, deserialize_with = "de_i64")]149
pub mtime_ms: i64,150
#[serde(default, deserialize_with = "de_u64")]151
pub ino: u64,152
#[serde(default, deserialize_with = "de_u64")]153
pub dev: u64,154
}156
impl LocalIdentity {157
pub fn of(meta: &std::fs::Metadata) -> LocalIdentity {158
let mtime_ms = meta159
.modified()160
.ok()161
.map(|t| match t.duration_since(std::time::UNIX_EPOCH) {162
Ok(d) => d.as_millis() as i64,163
Err(e) => -(e.duration().as_millis() as i64),164
})165
.unwrap_or(0);166
#[cfg(unix)]167
let (ino, dev) = {168
use std::os::unix::fs::MetadataExt;169
(meta.ino(), meta.dev())170
};171
// Stable Rust exposes no file index on Windows; size + mtime identify the source.172
#[cfg(not(unix))]173
let (ino, dev) = (0, 0);174
LocalIdentity { size: meta.len(), mtime_ms, ino, dev }175
}177
fn same(&self, other: &LocalIdentity) -> bool {178
let loose = cfg!(not(unix)) || self.ino == 0 || other.ino == 0;179
self.size == other.size && self.mtime_ms == other.mtime_ms && (loose || (self.ino == other.ino && self.dev == other.dev))180
}181
}183
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]184
pub struct RemoteIdentity {185
#[serde(default, deserialize_with = "de_u64")]186
pub size: u64,187
#[serde(default, deserialize_with = "de_u64")]188
pub mtime: u64,189
}191
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]192
pub struct JobError {193
#[serde(default)]194
pub code: String,195
#[serde(default)]196
pub message: String,197
}199
#[derive(Debug, Clone, Serialize, Deserialize)]200
#[serde(rename_all = "camelCase")]201
pub struct Job {202
pub id: String,203
#[serde(default)]204
pub owner: String,205
#[serde(default)]206
pub batch_id: String,207
#[serde(default)]208
pub direction: String,209
#[serde(default)]210
pub identity: SessionIdentity,211
#[serde(default)]212
pub session_id: Value,213
#[serde(default, deserialize_with = "de_u64")]214
pub order: u64,215
#[serde(default)]216
pub name: String,217
#[serde(default)]218
pub local_path: String,219
#[serde(default)]220
pub remote_path: String,221
#[serde(default, deserialize_with = "de_u64")]222
pub size: u64,223
#[serde(default, deserialize_with = "de_u64")]224
pub transferred: u64,225
#[serde(default, deserialize_with = "de_u64")]226
pub moved_bytes: u64,227
#[serde(default, deserialize_with = "de_u64")]228
pub active_ms: u64,229
pub state: String,230
#[serde(default)]231
pub mode: String,232
#[serde(default)]233
pub overwrite: bool,234
#[serde(default)]235
pub local_identity: Option<LocalIdentity>,236
#[serde(default)]237
pub remote_identity: Option<RemoteIdentity>,238
#[serde(default)]239
pub remote_mode: Option<Value>,240
#[serde(default)]241
pub finalize: Value,242
#[serde(default, deserialize_with = "de_u32")]243
pub attempts: u32,244
#[serde(default, deserialize_with = "de_u32")]245
pub total_retries: u32,246
#[serde(default)]247
pub error: Option<JobError>,248
#[serde(default, deserialize_with = "de_i64")]249
pub created_at: i64,250
#[serde(default, deserialize_with = "de_i64")]251
pub updated_at: i64,252
#[serde(default, deserialize_with = "de_opt_i64", skip_serializing_if = "Option::is_none")]253
pub started_at: Option<i64>,254
#[serde(default, deserialize_with = "de_opt_i64", skip_serializing_if = "Option::is_none")]255
pub finished_at: Option<i64>,256
#[serde(default, deserialize_with = "de_opt_i64")]257
pub next_retry_at: Option<i64>,258
#[serde(default, deserialize_with = "de_opt_string", skip_serializing_if = "Option::is_none")]259
pub resolved_path: Option<String>,260
#[serde(default, deserialize_with = "de_opt_string", skip_serializing_if = "Option::is_none")]261
pub part_path: Option<String>,262
/// `None` = not prepared yet (0.1.0 `undefined`), `Some(Null)` = target did not exist.263
#[serde(default, deserialize_with = "de_present", skip_serializing_if = "Option::is_none")]264
pub target_baseline: Option<Value>,265
#[serde(default, skip_serializing_if = "Option::is_none")]266
pub helper: Option<bool>,267
#[serde(default, skip_serializing_if = "Option::is_none")]268
pub owner_uid: Option<Value>,269
#[serde(default, deserialize_with = "de_opt_string", skip_serializing_if = "Option::is_none")]270
pub hash: Option<String>,271
#[serde(default, skip_serializing_if = "Option::is_none")]272
pub verification: Option<Value>,273
#[serde(default, skip_serializing_if = "Option::is_none")]274
pub staged: Option<Value>,275
#[serde(default, deserialize_with = "de_opt_string", skip_serializing_if = "Option::is_none")]276
pub local_part: Option<String>,277
#[serde(default)]278
pub auth_required: bool,279
#[serde(default)]280
pub resumable: bool,281
#[serde(default)]282
pub finalize_with_sudo: bool,283
#[serde(default)]284
pub cleanup_pending: bool,285
/// Unknown fields written by other versions are kept unchanged.286
#[serde(flatten)]287
pub extra: Map<String, Value>,288
}290
impl Job {291
fn finalize_state(&self) -> &str {292
self.finalize["state"].as_str().unwrap_or("pending")293
}294
fn session(&self) -> String {295
self.session_id.as_str().unwrap_or("").to_string()296
}297
}299
#[derive(Debug, Clone, Serialize, Deserialize)]300
#[serde(rename_all = "camelCase")]301
pub struct Settings {302
pub max_concurrent: u32,303
pub auto_retry: bool,304
pub verify: bool,305
}307
impl Default for Settings {308
fn default() -> Self {309
Settings { max_concurrent: 2, auto_retry: true, verify: true }310
}311
}313
#[derive(Debug, Clone)]314
struct PlanFile {315
name: String,316
local_path: String,317
remote_path: String,318
size: u64,319
local_identity: Option<LocalIdentity>,320
remote_identity: Option<RemoteIdentity>,321
mode: Option<u32>,322
conflict: bool,323
overwrite_confirmed: bool,324
}326
#[derive(Debug, Clone)]327
struct Plan {328
id: String,329
owner: String,330
direction: &'static str,331
session_id: String,332
identity: SessionIdentity,333
directory: String,334
local_directory: String,335
files: Vec<PlanFile>,336
directories: Vec<String>,337
skipped: Vec<Value>,338
bytes: u64,339
created_at: i64,340
mode: String,341
privilege: Option<Value>,342
/// Server-to-server: the source session and its identity. The plan's session is the destination.343
relay: Option<(String, SessionIdentity)>,344
/// Source folders of a move, removed when empty after the batch (a job exists only per file).345
source_directories: Vec<String>,346
source_sudo: bool,347
}349
struct Runtime {350
running: bool,351
stop: watch::Sender<String>,352
generation: u64,353
speed: f64,354
samples: u32,355
last_bytes: u64,356
last_time: i64,357
active_since: Option<i64>,358
progress_since_resume: u64,359
retry_task: Option<tokio::task::JoinHandle<()>>,360
}362
impl Runtime {363
fn new(transferred: u64) -> Self {364
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 }365
}366
}368
struct BatchMove {369
/// `None` = the folders are on this computer.370
session: Option<String>,371
directories: Vec<String>,372
}374
#[derive(Default)]375
struct Inner {376
batch_moves: HashMap<String, BatchMove>,377
jobs: HashMap<String, Job>,378
runtime: HashMap<String, Runtime>,379
secrets: HashMap<String, Zeroizing<String>>,380
plans: HashMap<String, Plan>,381
settings: Settings,382
order: u64,383
changed: HashSet<String>,384
pending_owners: HashSet<String>,385
full_owners: HashSet<String>,386
emit_scheduled: bool,387
cleaned_identities: HashSet<String>,388
closed: bool,389
}391
/// Stop requested (pause/cancel/shutdown/lost) or a real failure.392
enum RunError {393
Stop(String),394
Error(CoreError),395
}397
impl From<CoreError> for RunError {398
fn from(error: CoreError) -> Self {399
RunError::Error(error)400
}401
}403
type RunResult<T> = Result<T, RunError>;405
/// 0.1.0 `errorCode` for Core errors.406
fn error_code(error: &CoreError) -> String {407
if !error.code.is_empty() {408
return error.code.clone();409
}410
let lower = error.message.to_lowercase();411
if ["no response", "not connected", "connection lost", "connection closed", "channel", "econnreset", "socket"].iter().any(|p| lower.contains(p)) {412
return "SESSION_NOT_CONNECTED".into();413
}414
"TRANSFER_FAILED".into()415
}417
fn sftp_core(error: SftpError) -> CoreError {418
CoreError::new(error.transfer_code(), error.message())419
}421
fn io_core(error: std::io::Error) -> CoreError {422
let code = match error.kind() {423
std::io::ErrorKind::NotFound => "ENOENT",424
std::io::ErrorKind::PermissionDenied => "EACCES",425
std::io::ErrorKind::StorageFull => "ENOSPC",426
std::io::ErrorKind::AlreadyExists => "EEXIST",427
_ => "EIO",428
};429
CoreError::new(code, error.to_string())430
}432
fn percent(transferred: u64, size: u64) -> Value {433
if size == 0 {434
return Value::Null;435
}436
json!(((transferred.min(size) as f64 / size as f64) * 1000.0).floor().min(1000.0) / 10.0)437
}439
/// Aggregate view: percentages are byte based.440
pub fn summarize(jobs: &[Value]) -> Value {441
let counted: Vec<&Value> = jobs.iter().filter(|j| j["state"] != "cancelled").collect();442
let sizes_known = counted.iter().all(|j| j["size"].is_number());443
let total: f64 = counted.iter().map(|j| j["size"].as_f64().unwrap_or(0.0)).sum();444
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();445
let completed = counted.iter().filter(|j| j["state"] == "completed").count();446
let active: Vec<&Value> = jobs.iter().filter(|j| RUNNING.contains(&j["state"].as_str().unwrap_or(""))).collect();447
let speed: f64 = active.iter().map(|j| j["speed"].as_f64().unwrap_or(0.0)).sum();448
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();449
let eta_reliable = !active.is_empty() && active.iter().all(|j| j["state"] != "transferring" || !j["eta"].is_null());450
let mut overall: Option<f64> = None;451
if sizes_known && total > 0.0 {452
overall = Some((transferred / total * 1000.0).floor() / 10.0);453
} else if sizes_known && !counted.is_empty() && completed == counted.len() {454
overall = Some(100.0);455
}456
if overall.is_some_and(|o| o >= 100.0) && completed != counted.len() {457
overall = Some(99.9);458
}459
let count = |state: &str| jobs.iter().filter(|j| j["state"] == state).count();460
json!({461
"total": jobs.len(),462
"completed": completed,463
"active": active.len(),464
"waiting": count("waiting") + count("reconnecting"),465
"paused": count("paused"),466
"interrupted": count("interrupted"),467
"failed": count("failed"),468
"cancelled": jobs.len() - counted.len(),469
"totalBytes": total as u64,470
"transferredBytes": transferred as u64,471
"sizesKnown": sizes_known,472
"percent": overall,473
"speed": speed,474
"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 }475
})476
}478
/// Native dialogs for one planning call (shown by the command layer for the calling module).479
pub type PickPaths = Box<dyn FnOnce() -> BoxFuture<'static, Option<Vec<PathBuf>>> + Send>;480
pub type PickSavePath = Box<dyn FnOnce(String) -> BoxFuture<'static, Option<PathBuf>> + Send>;481
pub type PickDirectory = Box<dyn FnOnce() -> BoxFuture<'static, Option<PathBuf>> + Send>;483
pub type UpdateEmitter = Arc<dyn Fn(&str, Value) + Send + Sync>;485
pub struct TransferManager {486
sessions: Arc<SessionManager>,487
files: Arc<RemoteFileService>,488
logger: Arc<Logger>,489
file: Option<PathBuf>,490
inner: Mutex<Inner>,491
emitter: UpdateEmitter,492
persist_pending: AtomicBool,493
cleanup_scheduled: AtomicBool,494
scheduling: AtomicBool,495
}497
impl TransferManager {498
pub fn new(sessions: Arc<SessionManager>, files: Arc<RemoteFileService>, storage_root: Option<&Path>, logger: Arc<Logger>, emitter: UpdateEmitter) -> Arc<Self> {499
let manager = Arc::new(TransferManager {500
sessions,501
files,502
logger,503
file: storage_root.map(|r| r.join("transfers.json")),504
inner: Mutex::new(Inner::default()),505
emitter,506
persist_pending: AtomicBool::new(false),507
cleanup_scheduled: AtomicBool::new(false),508
scheduling: AtomicBool::new(false),509
});510
manager.load();511
manager512
}514
/// Subscribes to session state changes (must run inside the async runtime).515
pub fn start_session_listener(self: &Arc<Self>) {516
let mut events = self.sessions.state_events.subscribe();517
let weak = Arc::downgrade(self);518
tokio::spawn(async move {519
loop {520
match events.recv().await {521
Ok(status) => match weak.upgrade() {522
Some(manager) => manager.session_state(&status),523
None => break,524
},525
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,526
Err(_) => break,527
}528
}529
});530
}532
// ---- public API (owner = module id) ----------------------------------------------------------534
pub fn list(self: &Arc<Self>, owner: &str) -> Value {535
let jobs = {536
let inner = self.inner.lock();537
Self::owned(&inner, owner).iter().map(|j| Self::public(&inner, j)).collect::<Vec<_>>()538
};539
self.cleanup_orphans_later();540
let settings = self.inner.lock().settings.clone();541
json!({ "jobs": jobs, "summary": summarize(&jobs), "settings": settings })542
}544
pub fn update_settings(self: &Arc<Self>, patch: &Value) -> CoreResult<Value> {545
let settings = {546
let mut inner = self.inner.lock();547
if let Some(value) = patch.get("maxConcurrent").filter(|v| !v.is_null()) {548
match number(value) {549
Some(n) if n.fract() == 0.0 && (1.0..=8.0).contains(&n) => inner.settings.max_concurrent = n as u32,550
_ => return Err(file_error("TRANSFER_INVALID_SETTINGS", "Concurrent transfers must be between 1 and 8.")),551
}552
}553
if let Some(value) = patch.get("autoRetry").filter(|v| !v.is_null()) {554
inner.settings.auto_retry = *value == true;555
}556
if let Some(value) = patch.get("verify").filter(|v| !v.is_null()) {557
inner.settings.verify = *value == true;558
}559
inner.settings.clone()560
};561
self.persist(true, false);562
self.schedule();563
Ok(serde_json::to_value(settings).unwrap_or_default())564
}566
pub fn identity(&self, session_id: &str) -> CoreResult<SessionIdentity> {567
let session = self568
.sessions569
.list()570
.into_iter()571
.find(|s| s["sessionId"] == session_id && s["state"] == "connected")572
.ok_or_else(|| file_error("SESSION_NOT_CONNECTED", "The SSH session is not connected."))?;573
Ok(SessionIdentity::of_status(&session))574
}576
pub async fn plan_upload(self: &Arc<Self>, owner: &str, session_id: &str, directory: &Value, pick: PickPaths) -> CoreResult<Value> {577
let directory = normalize_remote_path(&or_root(directory))?;578
let identity = self.identity(session_id)?;579
let mut local_paths: Vec<PathBuf> = Vec::new();580
for path in pick().await.unwrap_or_default() {581
if !local_paths.contains(&path) {582
local_paths.push(path);583
}584
}585
if local_paths.is_empty() {586
return Ok(json!({ "canceled": true }));587
}588
let mut plan = Plan {589
id: hex_id(16),590
owner: owner.into(),591
direction: "upload",592
session_id: session_id.into(),593
identity,594
directory: directory.clone(),595
local_directory: String::new(),596
files: Vec::new(),597
directories: Vec::new(),598
skipped: Vec::new(),599
bytes: 0,600
created_at: now(),601
mode: String::new(),602
privilege: None,603
relay: None,604
source_directories: Vec::new(),605
source_sudo: false,606
};607
let walked = {608
let directory = directory.clone();609
tokio::task::spawn_blocking(move || -> CoreResult<(Vec<PlanFile>, Vec<String>, Vec<String>, Vec<Value>, u64)> {610
let mut walk = LocalWalk::default();611
for local in &local_paths {612
let name = safe_entry_name(&local.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default())?;613
walk.visit(local, &posix::join(&directory, &name), 0)?;614
}615
Ok((walk.files, walk.directories, walk.source_directories, walk.skipped, walk.bytes))616
})617
.await618
.map_err(|_| CoreError::plain("Local folder scan failed."))??619
};620
(plan.files, plan.directories, plan.source_directories, plan.skipped, plan.bytes) = walked;621
let sftp = self.files.sftp(session_id).await?;622
plan.mode = directory_mode(&sftp, &directory).await?.into();623
remote_conflicts(&sftp, &mut plan).await;624
drop(sftp);625
if plan.mode == "privileged" {626
let privilege = self.files.privileges(session_id).await?;627
if privilege["mode"] == "none" {628
let code = if privilege["sudoInstalled"] == true { "SUDO_NOT_PERMITTED" } else { "SUDO_UNAVAILABLE" };629
return Err(file_error(code, "The destination is not writable and the SSH account cannot use sudo."));630
}631
plan.privilege = Some(privilege);632
}633
self.store_plan(plan)634
}636
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> {637
let identity = self.identity(session_id)?;638
let values: Vec<Value> = match paths {639
Value::Array(list) => list.clone(),640
other => vec![other.clone()],641
};642
let mut targets: Vec<String> = Vec::new();643
for value in values.into_iter().filter(|v| !v.is_null()) {644
let target = normalize_remote_path(&value)?;645
if !targets.contains(&target) {646
targets.push(target);647
}648
}649
if targets.is_empty() || targets.len() > 100 {650
return Err(file_error("TRANSFER_INVALID_SELECTION", "Select between 1 and 100 remote entries."));651
}652
if targets.iter().any(|t| t == "/") {653
return Err(file_error("TRANSFER_INVALID_SELECTION", "Downloading the entire remote root is not allowed in one batch."));654
}655
let sudo = options["sudo"] == true;656
let mut plan = Plan {657
id: hex_id(16),658
owner: owner.into(),659
direction: "download",660
session_id: session_id.into(),661
identity,662
directory: String::new(),663
local_directory: String::new(),664
files: Vec::new(),665
directories: Vec::new(),666
skipped: Vec::new(),667
bytes: 0,668
created_at: now(),669
mode: if sudo { "privileged-allowed".into() } else { "direct".into() },670
privilege: None,671
relay: None,672
source_directories: Vec::new(),673
source_sudo: false,674
};675
let sftp = self.files.sftp(session_id).await?;676
let first = if targets.len() == 1 { Some(self.remote_stat(&sftp, &targets[0], options, session_id).await?) } else { None };677
if let Some(info) = first.filter(|i| i.kind == "file") {678
let Some(chosen) = pick_file(posix::basename(&targets[0])).await else { return Ok(json!({ "canceled": true })) };679
plan.local_directory = chosen.parent().map(|p| p.to_string_lossy().to_string()).unwrap_or_default();680
add_download(&mut plan, &targets[0], &chosen.to_string_lossy(), &info);681
plan.files[0].overwrite_confirmed = options["noSaveDialog"] != true; // The native save dialog already confirmed replacing.682
} else {683
let Some(chosen) = pick_directory().await else { return Ok(json!({ "canceled": true })) };684
plan.local_directory = chosen.to_string_lossy().to_string();685
for target in &targets {686
let local = chosen.join(safe_entry_name(&posix::basename(target))?);687
self.walk_remote(&sftp, session_id, &mut plan, target, &local.to_string_lossy(), options, true, 0).await?;688
}689
}690
if !sudo {691
for file in &plan.files {692
let handle = sftp.open_file(&file.remote_path, "r", None).await.map_err(sftp_core)?;693
sftp.close_file(&handle).await.map_err(sftp_core)?;694
}695
}696
drop(sftp);697
for file in plan.files.iter_mut() {698
if !file.overwrite_confirmed && std::fs::symlink_metadata(&file.local_path).is_ok() {699
file.conflict = true;700
}701
}702
self.store_plan(plan)703
}705
/// Server-to-server plan. The data streams through this computer: nothing is stored on its disk.706
pub async fn plan_relay(self: &Arc<Self>, owner: &str, source_session: &str, paths: &Value, dest_session: &str, directory: &Value, options: &Value) -> CoreResult<Value> {707
let directory = normalize_remote_path(&or_root(directory))?;708
let identity = self.identity(dest_session)?;709
let source_identity = self.identity(source_session)?;710
let mut targets: Vec<String> = Vec::new();711
for value in paths.as_array().cloned().unwrap_or_default() {712
let target = normalize_remote_path(&value)?;713
if !targets.contains(&target) {714
targets.push(target);715
}716
}717
if targets.is_empty() || targets.len() > 100 || targets.iter().any(|t| t == "/") {718
return Err(file_error("TRANSFER_INVALID_SELECTION", "Select between 1 and 100 remote entries, not the remote root."));719
}720
if source_identity == identity {721
for target in &targets {722
if posix::dirname(target) == directory {723
return Err(file_error("TRANSFER_SAME_PATH", "The source and the destination are the same folder."));724
}725
if directory == *target || directory.starts_with(&format!("{target}/")) {726
return Err(file_error("TRANSFER_INTO_ITSELF", "A folder cannot be copied into itself."));727
}728
}729
}730
let mut plan = Plan {731
id: hex_id(16),732
owner: owner.into(),733
direction: "relay",734
session_id: dest_session.into(),735
identity,736
directory: directory.clone(),737
local_directory: String::new(),738
files: Vec::new(),739
directories: Vec::new(),740
skipped: Vec::new(),741
bytes: 0,742
created_at: now(),743
mode: String::new(),744
privilege: None,745
relay: Some((source_session.to_string(), source_identity)),746
source_directories: Vec::new(),747
source_sudo: options["sudo"] == true,748
};749
let source = self.files.sftp(source_session).await?;750
for target in &targets {751
let destination = posix::join(&directory, &safe_entry_name(&posix::basename(target))?);752
self.walk_relay(&source, source_session, &mut plan, target, &destination, options, 0).await?;753
}754
drop(source);755
let sftp = self.files.sftp(dest_session).await?;756
plan.mode = directory_mode(&sftp, &directory).await?.into();757
remote_conflicts(&sftp, &mut plan).await;758
drop(sftp);759
if plan.mode == "privileged" {760
let privilege = self.files.privileges(dest_session).await?;761
if privilege["mode"] == "none" {762
let code = if privilege["sudoInstalled"] == true { "SUDO_NOT_PERMITTED" } else { "SUDO_UNAVAILABLE" };763
return Err(file_error(code, "The destination is not writable and the SSH account cannot use sudo."));764
}765
plan.privilege = Some(privilege);766
}767
self.store_plan(plan)768
}770
pub async fn start(self: &Arc<Self>, owner: &str, plan_id: &Value, options: &Value) -> CoreResult<Value> {771
let plan = {772
let inner = self.inner.lock();773
inner.plans.get(plan_id.as_str().unwrap_or("")).filter(|p| p.owner == owner && now() - p.created_at <= PLAN_TTL_MS).cloned()774
}775
.ok_or_else(|| file_error("TRANSFER_PLAN_EXPIRED", "The transfer plan expired; select the files again."))?;776
if options["sessionId"].as_str().is_some_and(|id| id != plan.session_id)777
|| self.identity(&plan.session_id)? != plan.identity778
{779
return Err(file_error("TRANSFER_PLAN_SERVER_CHANGED", "The transfer plan belongs to a different server or connection."));780
}781
let mut job_extra = Map::new();782
if let Some((source, source_identity)) = &plan.relay {783
if self.identity(source)? != *source_identity {784
return Err(file_error("TRANSFER_PLAN_SERVER_CHANGED", "The transfer plan belongs to a different server or connection."));785
}786
job_extra.insert("sourceSudo".into(), json!(plan.source_sudo));787
job_extra.insert("relaySource".into(), json!({ "sessionId": source, "identity": source_identity }));788
}789
job_extra.insert("hostKey".into(), json!(self.sessions.trusted_host_key(&plan.session_id)));790
if let Some((source, _)) = &plan.relay {791
job_extra.insert("sourceHostKey".into(), json!(self.sessions.trusted_host_key(source)));792
}793
if options["move"] == true {794
job_extra.insert("moveSource".into(), json!(true));795
job_extra.insert("moveRoot".into(), json!(options["sourceDirectory"].as_str().unwrap_or("")));796
}797
let conflict = options["conflict"].as_str().filter(|c| ["overwrite", "skip"].contains(c)).unwrap_or("fail");798
let conflicts = plan.files.iter().filter(|f| f.conflict).count();799
if conflicts > 0 && conflict == "fail" {800
return Err(file_error("FILE_EXISTS", &format!("{conflicts} destination file(s) already exist.")));801
}802
if self.inner.lock().jobs.len() + plan.files.len() > MAX_JOBS {803
return Err(file_error("TRANSFER_QUEUE_FULL", "Remove finished transfers before starting more."));804
}805
let password = options["sudoPassword"].as_str().unwrap_or("").to_string();806
let privileged = plan.mode == "privileged" || (plan.direction == "download" && (plan.mode == "privileged-allowed" || options["sudo"] == true));807
if plan.mode == "privileged" {808
// Verifies the sudo authorization once before any data moves (and caches it on request).809
let auth = Authorization { sudo: true, sudo_password: (!password.is_empty()).then(|| password.clone()), remember_sudo: options["rememberSudo"] == true, ..Default::default() };810
self.files.helper(&plan.session_id, json!({ "operation": "stat", "path": plan.directory }), &auth).await?;811
}812
let source_options = options.get("sourceOptions").filter(|o| o.is_object()).unwrap_or(options);813
if plan.source_sudo {814
let auth = Authorization::from_options(source_options).sudo();815
if let Some(file) = plan.files.first() {816
self.files.helper(&plan.relay.as_ref().unwrap().0, json!({ "operation": "source-stat", "path": file.local_path }), &auth).await.map_err(|e| e.with("sourceSudo", true))?;817
}818
}819
self.create_directories(&plan, plan.mode == "privileged", &password).await?;820
let mut created = 0;821
{822
let mut inner = self.inner.lock();823
for file in &plan.files {824
if file.conflict && conflict == "skip" {825
continue;826
}827
inner.order += 1;828
let order = inner.order;829
let job = Job {830
id: hex_id(16),831
owner: owner.into(),832
batch_id: plan.id.clone(),833
direction: plan.direction.into(),834
identity: plan.identity.clone(),835
session_id: json!(plan.session_id),836
order,837
name: file.name.clone(),838
local_path: file.local_path.clone(),839
remote_path: file.remote_path.clone(),840
size: file.size,841
transferred: 0,842
moved_bytes: 0,843
active_ms: 0,844
state: "waiting".into(),845
mode: if plan.direction != "download" { plan.mode.clone() } else if privileged { "privileged-allowed".into() } else { "direct".into() },846
overwrite: conflict == "overwrite" || file.overwrite_confirmed,847
local_identity: file.local_identity.clone(),848
remote_identity: file.remote_identity.clone(),849
remote_mode: file.mode.map(Value::from),850
finalize: json!({ "state": "pending" }),851
attempts: 0,852
total_retries: 0,853
error: None,854
created_at: now(),855
updated_at: now(),856
started_at: None,857
finished_at: None,858
next_retry_at: None,859
resolved_path: None,860
part_path: None,861
target_baseline: None,862
helper: None,863
owner_uid: None,864
hash: None,865
verification: None,866
staged: None,867
local_part: None,868
auth_required: false,869
resumable: false,870
finalize_with_sudo: false,871
cleanup_pending: false,872
extra: job_extra.clone(),873
};874
if !password.is_empty() && privileged {875
inner.secrets.insert(job.id.clone(), Zeroizing::new(password.clone()));876
}877
if plan.source_sudo {878
if let Some(password) = source_options["sudoPassword"].as_str().filter(|s| !s.is_empty()) {879
inner.secrets.insert(format!("{}:source", job.id), Zeroizing::new(password.to_string()));880
}881
}882
let id = job.id.clone();883
inner.jobs.insert(id.clone(), job);884
created += 1;885
Self::mark(&mut inner, &id, owner);886
}887
inner.plans.remove(&plan.id);888
if options["move"] == true && !plan.source_directories.is_empty() {889
let session = plan.relay.as_ref().map(|(source, _)| source.clone()).or_else(|| (plan.direction == "download").then(|| plan.session_id.clone()));890
inner.batch_moves.insert(plan.id.clone(), BatchMove { session, directories: plan.source_directories.clone() });891
}892
Self::trim_history(&mut inner);893
}894
self.finish_batch_move(&plan.id).await;895
self.emit_soon();896
self.persist(true, false);897
self.logger.info("Transfer batch queued.", json!({ "direction": plan.direction, "files": created, "bytes": plan.bytes, "mode": plan.mode, "sessionId": plan.session_id }));898
self.schedule();899
Ok(json!({ "batchId": plan.id, "jobs": created, "skipped": plan.files.len() - created }))900
}902
pub fn pause(self: &Arc<Self>, owner: &str, ids: &Value) -> Value {903
let mut count = 0;904
for id in self.select(owner, ids) {905
let state = self.job_state(&id);906
if ["waiting", "reconnecting", "interrupted"].contains(&state.as_str()) {907
self.clear_retry(&id);908
self.set(&id, "paused", None);909
count += 1;910
} else if RUNNING.contains(&state.as_str()) && state != "finalizing" {911
self.stop(&id, "pause");912
count += 1;913
}914
}915
self.persist(true, false);916
json!({ "paused": count })917
}919
pub fn resume(self: &Arc<Self>, owner: &str, ids: &Value, options: &Value) -> Value {920
let mut count = 0;921
for id in self.select(owner, ids) {922
if self.with_job(&id, move_recovery_pending).unwrap_or(false) {923
if self.inner.lock().runtime.get(&id).is_some_and(|r| r.running) || !self.bind_session(&id, options) { continue; }924
self.update(&id, |j| { if j.direction == "relay" { j.extra.insert("sourceAuthRequired".into(), json!(true)); } });925
self.authorize(&id, options);926
self.inner.lock().runtime.entry(id.clone()).or_insert_with(|| Runtime::new(0)).running = true;927
let manager = self.clone();928
tokio::spawn(async move { manager.recover_move(&id).await; });929
count += 1;930
continue;931
}932
let resumable = {933
let inner = self.inner.lock();934
inner.jobs.get(&id).is_some_and(|j| ["paused", "interrupted", "reconnecting"].contains(&j.state.as_str()) || (j.state == "failed" && j.resumable))935
};936
if !resumable {937
continue;938
}939
if !self.bind_session(&id, options) { continue; }940
self.authorize(&id, options);941
{942
let mut inner = self.inner.lock();943
if let Some(job) = inner.jobs.get_mut(&id) {944
if options["overwrite"] == true && job.error.as_ref().is_some_and(|e| e.code == "FILE_EXISTS" || e.code == "FILE_CONFLICT") {945
job.overwrite = true;946
job.target_baseline = None;947
job.finalize = json!({ "state": "pending" });948
}949
job.attempts = 0;950
job.auth_required = false;951
}952
}953
self.clear_retry(&id);954
self.set(&id, "waiting", None);955
count += 1;956
}957
self.persist(true, false);958
self.schedule();959
json!({ "resumed": count })960
}962
/// Restart from offset zero (used when the partial data cannot be safely continued).963
pub fn retry(self: &Arc<Self>, owner: &str, ids: &Value, options: &Value) -> Value {964
let mut count = 0;965
for id in self.select(owner, ids) {966
let local_part = {967
let inner = self.inner.lock();968
match inner.jobs.get(&id) {969
Some(j) if ["failed", "interrupted", "paused", "cancelled"].contains(&j.state.as_str()) => Some(j.local_part.clone()),970
_ => None,971
}972
};973
let Some(local_part) = local_part else { continue };974
if !self.bind_session(&id, options) { continue; }975
// The deterministic partial file (same job id) is truncated and rewritten from offset zero.976
if let Some(part) = local_part {977
let _ = std::fs::remove_file(part);978
}979
self.authorize(&id, options);980
{981
let mut inner = self.inner.lock();982
if let Some(job) = inner.jobs.get_mut(&id) {983
job.transferred = 0;984
job.staged = None;985
job.target_baseline = None;986
job.hash = None;987
job.verification = None;988
job.finalize = json!({ "state": "pending" });989
job.attempts = 0;990
job.auth_required = false;991
job.error = None;992
job.resumable = false;993
job.finalize_with_sudo = false;994
if job.direction == "upload" {995
if let Ok(meta) = std::fs::metadata(&job.local_path) {996
job.local_identity = Some(LocalIdentity::of(&meta));997
job.size = meta.len();998
}999
} else {1000
job.remote_identity = None;1001
}1002
if options["overwrite"] == true {1003
job.overwrite = true;1004
}1005
}1006
}1007
self.set(&id, "waiting", None);1008
count += 1;1009
}1010
self.persist(true, false);1011
self.schedule();1012
json!({ "retried": count })1013
}1015
pub async fn cancel(self: &Arc<Self>, owner: &str, ids: &Value) -> Value {1016
let mut count = 0;1017
for id in self.select(owner, ids) {1018
let state = self.job_state(&id);1019
if FINAL.contains(&state.as_str()) {1020
continue;1021
}1022
count += 1;1023
if RUNNING.contains(&state.as_str()) {1024
if state == "finalizing" {1025
continue; // The remote replacement must finish or fail on its own.1026
}1027
self.stop(&id, "cancel");1028
continue;1029
}1030
self.clear_retry(&id);1031
self.finish_cancel(&id).await;1032
}1033
self.persist(true, false);1034
json!({ "cancelled": count })1035
}1037
pub fn remove(self: &Arc<Self>, owner: &str, ids: &Value) -> Value {1038
let mut count = 0;1039
{1040
let mut inner = self.inner.lock();1041
let selected = Self::select_ids(&inner, owner, ids);1042
for id in selected {1043
let removable = inner.jobs.get(&id).is_some_and(|j| FINAL.contains(&j.state.as_str()) && !j.cleanup_pending);1044
if !removable {1045
continue;1046
}1047
inner.jobs.remove(&id);1048
inner.secrets.remove(&id);1049
inner.secrets.remove(&format!("{id}:source"));1050
inner.runtime.remove(&id);1051
inner.changed.insert(id);1052
count += 1;1053
}1054
inner.pending_owners.insert(owner.to_string());1055
inner.full_owners.insert(owner.to_string());1056
}1057
self.persist(true, false);1058
self.emit_soon();1059
json!({ "removed": count })1060
}1062
pub fn shutdown(self: &Arc<Self>) {1063
let ids: Vec<(String, String)> = {1064
let inner = self.inner.lock();1065
if inner.closed {1066
return;1067
}1068
inner.jobs.values().map(|j| (j.id.clone(), j.state.clone())).collect()1069
};1070
for (id, state) in ids {1071
self.clear_retry(&id);1072
if RUNNING.contains(&state.as_str()) {1073
self.stop(&id, "shutdown");1074
}1075
}1076
// Running jobs are recorded as interrupted so the next start offers to resume them.1077
{1078
let mut inner = self.inner.lock();1079
for job in inner.jobs.values_mut() {1080
if RUNNING.contains(&job.state.as_str()) || job.state == "waiting" || job.state == "reconnecting" {1081
if job.state == "finalizing" || job.finalize_state() == "running" {1082
job.finalize["state"] = json!("unknown");1083
}1084
job.state = "interrupted".into();1085
job.error = Some(JobError { code: "APP_RESTARTED".into(), message: "The application was closed during the transfer.".into() });1086
}1087
}1088
}1089
self.persist(true, true);1090
let mut inner = self.inner.lock();1091
inner.closed = true; // Late completions of stopped jobs must not overwrite the saved state.1092
inner.secrets.clear();1093
}1095
// ---- planning helpers ----------------------------------------------------------------------1097
fn store_plan(&self, plan: Plan) -> CoreResult<Value> {1098
if plan.files.len() + plan.directories.len() > MAX_PLAN_ENTRIES {1099
return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", &format!("A transfer can contain at most {MAX_PLAN_ENTRIES} entries.")));1100
}1101
if plan.bytes > MAX_PLAN_BYTES {1102
return Err(file_error("TRANSFER_TOO_LARGE", "The selection exceeds the transfer size limit."));1103
}1104
let conflicts: Vec<&PlanFile> = plan.files.iter().filter(|f| f.conflict).collect();1105
let result = json!({1106
"canceled": false,1107
"planId": plan.id,1108
"direction": plan.direction,1109
"files": plan.files.len(),1110
"directories": plan.directories.len(),1111
"bytes": plan.bytes,1112
"mode": plan.mode,1113
"requiresSudo": plan.mode == "privileged" && plan.privilege.as_ref().is_some_and(|p| p["mode"] != "root"),1114
"sudoPasswordRequired": plan.mode == "privileged" && plan.privilege.as_ref().is_some_and(|p| p["mode"] == "sudo-password" && p["sudoCached"] != true),1115
"conflicts": conflicts.iter().take(50).map(|f| if plan.direction != "download" { f.remote_path.clone() } else { f.local_path.clone() }).collect::<Vec<_>>(),1116
"conflictCount": conflicts.len(),1117
"skipped": plan.skipped.iter().take(50).cloned().collect::<Vec<_>>(),1118
"skippedCount": plan.skipped.len(),1119
"destination": if plan.direction != "download" { plan.directory.clone() } else { plan.local_directory.clone() }1120
});1121
let mut inner = self.inner.lock();1122
let cutoff = now() - PLAN_TTL_MS;1123
inner.plans.retain(|_, p| p.created_at >= cutoff);1124
inner.plans.insert(plan.id.clone(), plan);1125
Ok(result)1126
}1128
async fn source_info(&self, sftp: &Sftp, session: &str, path: &str, auth: &Authorization) -> CoreResult<RemoteInfo> {1129
if !auth.sudo {1130
let stat = sftp.lstat(path).await.map_err(|e| sftp_core(e).with("sourceSudo", true))?;1131
return Ok(RemoteInfo { kind: if stat.is_symlink() { "symlink" } else if stat.is_file() { "file" } else if stat.is_dir() { "directory" } else { "other" }, size: stat.size, mtime: stat.mtime as u64, mode: stat.mode });1132
}1133
let value = self.files.helper(session, json!({ "operation": "source-stat", "path": path }), auth).await.map_err(|e| e.with("sourceSudo", true))?;1134
if value["exists"] != true {1135
return Err(file_error("FILE_NOT_FOUND", &format!("{path} does not exist.")));1136
}1137
Ok(RemoteInfo {1138
kind: match value["type"].as_str() { Some("file") => "file", Some("directory") => "directory", Some("symlink") => "symlink", _ => "other" },1139
size: number(&value["baseline"]["size"]).unwrap_or(0.0) as u64,1140
mtime: (number(&value["baseline"]["mtimeNs"]).unwrap_or(0.0) / 1e9).trunc() as u64,1141
mode: number(&value["baseline"]["mode"]).unwrap_or(0.0) as u32,1142
})1143
}1145
/// Every name is checked; symlinks and special files are skipped, never followed.1146
#[allow(clippy::too_many_arguments)]1147
fn walk_relay<'a>(&'a self, sftp: &'a Sftp, session: &'a str, plan: &'a mut Plan, source: &'a str, destination: &'a str, options: &'a Value, depth: u32) -> BoxFuture<'a, CoreResult<()>> {1148
async move {1149
if plan.files.len() + plan.directories.len() > MAX_PLAN_ENTRIES {1150
return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", &format!("A transfer can contain at most {MAX_PLAN_ENTRIES} entries.")));1151
}1152
let stat = self.source_info(sftp, session, source, &Authorization::from_options(options)).await?;1153
if stat.kind == "symlink" || stat.kind == "other" {1154
plan.skipped.push(json!({ "path": source, "reason": if stat.kind == "symlink" { "symlink" } else { "type" } }));1155
return Ok(());1156
}1157
if stat.kind == "file" {1158
if options["sudo"] != true {1159
let handle = sftp.open_file(source, "r", None).await.map_err(sftp_core)?;1160
sftp.close_file(&handle).await.map_err(sftp_core)?;1161
}1162
plan.bytes += stat.size;1163
plan.files.push(PlanFile {1164
name: posix::basename(destination),1165
local_path: source.to_string(),1166
remote_path: destination.to_string(),1167
size: stat.size,1168
local_identity: None,1169
remote_identity: Some(RemoteIdentity { size: stat.size, mtime: stat.mtime }),1170
mode: None,1171
conflict: false,1172
overwrite_confirmed: false,1173
});1174
return Ok(());1175
}1176
if depth > 64 {1177
return Err(file_error("TRANSFER_TOO_DEEP", "The selected directory tree is too deep."));1178
}1179
plan.directories.push(destination.to_string());1180
plan.source_directories.push(source.to_string());1181
let names: Vec<String> = if options["sudo"] == true {1182
let listed = self.files.helper(session, json!({ "operation": "listdir", "path": source, "noFollow": true }), &Authorization::from_options(options)).await?;1183
if listed["truncated"] == true {1184
return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", "The source directory listing is truncated."));1185
}1186
listed["entries"].as_array().cloned().unwrap_or_default().iter().filter_map(|e| e["name"].as_str().map(str::to_string)).collect()1187
} else {1188
sftp.read_dir(source).await.map_err(|e| file_error(e.transfer_code(), &format!("{source} cannot be read.")))?.into_iter().map(|e| e.name).collect()1189
};1190
for raw in names {1191
if raw == "." || raw == ".." {1192
continue;1193
}1194
let Ok(name) = safe_entry_name(&raw) else {1195
plan.skipped.push(json!({ "path": posix::join(source, &raw), "reason": "name" }));1196
continue;1197
};1198
self.walk_relay(sftp, session, plan, &posix::join(source, &name), &posix::join(destination, &name), options, depth + 1).await?;1199
}1200
Ok(())1201
}1202
.boxed()1203
}1205
async fn remote_stat(&self, sftp: &Sftp, target: &str, options: &Value, session_id: &str) -> CoreResult<RemoteInfo> {1206
match sftp.stat(target).await {1207
Ok(stat) => Ok(RemoteInfo { kind: if stat.is_dir() { "directory" } else { "file" }, size: stat.size, mtime: stat.mtime as u64, mode: stat.mode }),1208
Err(error) => {1209
let code = error.transfer_code();1210
if code != "FILE_PERMISSION_DENIED" || options["sudo"] != true {1211
return Err(file_error(code, &format!("{target} is not accessible.")));1212
}1213
let value = self.files.helper(session_id, json!({ "operation": "stat", "path": target }), &Authorization::from_options(options).sudo()).await?;1214
if value["exists"] != true {1215
return Err(file_error("FILE_NOT_FOUND", &format!("{target} does not exist.")));1216
}1217
let mtime_ns = number(&value["baseline"]["mtimeNs"]).unwrap_or(0.0);1218
Ok(RemoteInfo {1219
kind: if value["type"] == "directory" { "directory" } else { "file" },1220
size: number(&value["baseline"]["size"]).unwrap_or(0.0) as u64,1221
mtime: (mtime_ns / 1e9).trunc() as u64,1222
mode: number(&value["baseline"]["mode"]).unwrap_or(0.0) as u32,1223
})1224
}1225
}1226
}1228
#[allow(clippy::too_many_arguments)]1229
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<()>> {1230
async move {1231
if plan.files.len() + plan.directories.len() > MAX_PLAN_ENTRIES {1232
return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", &format!("A transfer can contain at most {MAX_PLAN_ENTRIES} entries.")));1233
}1234
let info = if top {1235
self.remote_stat(sftp, remote, options, session_id).await?1236
} else {1237
let stat = sftp.lstat(remote).await.map_err(|e| file_error(e.transfer_code(), &format!("{remote} is not accessible.")))?;1238
if stat.is_symlink() {1239
plan.skipped.push(json!({ "path": remote, "reason": "symlink" }));1240
return Ok(());1241
}1242
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 }1243
};1244
if info.kind == "other" {1245
plan.skipped.push(json!({ "path": remote, "reason": "type" }));1246
return Ok(());1247
}1248
if info.kind != "directory" {1249
add_download(plan, remote, local, &info);1250
return Ok(());1251
}1252
if depth > 64 {1253
return Err(file_error("TRANSFER_TOO_DEEP", "The selected directory tree is too deep."));1254
}1255
plan.directories.push(local.to_string());1256
plan.source_directories.push(remote.to_string());1257
match sftp.read_dir(remote).await {1258
Ok(entries) => {1259
for entry in entries {1260
if entry.name == "." || entry.name == ".." {1261
continue;1262
}1263
let Ok(name) = safe_entry_name(&entry.name) else {1264
plan.skipped.push(json!({ "path": posix::join(remote, &entry.name), "reason": "name" }));1265
continue;1266
};1267
let child_local = Path::new(local).join(&name).to_string_lossy().to_string();1268
self.walk_remote(sftp, session_id, plan, &posix::join(remote, &name), &child_local, options, false, depth + 1).await?;1269
}1270
}1271
Err(error) => {1272
if error.transfer_code() != "FILE_PERMISSION_DENIED" || options["sudo"] != true {1273
return Err(file_error(error.transfer_code(), &format!("Directory {remote} cannot be listed.")));1274
}1275
let listed = self.files.helper(session_id, json!({ "operation": "listdir", "path": remote }), &Authorization::from_options(options).sudo()).await?;1276
for entry in listed["entries"].as_array().cloned().unwrap_or_default() {1277
let raw = entry["name"].as_str().unwrap_or("").to_string();1278
let Ok(name) = safe_entry_name(&raw) else {1279
plan.skipped.push(json!({ "path": posix::join(remote, &raw), "reason": "name" }));1280
continue;1281
};1282
let child = posix::join(remote, &name);1283
let child_local = Path::new(local).join(&name).to_string_lossy().to_string();1284
match entry["type"].as_str() {1285
Some(kind @ ("symlink" | "other")) => plan.skipped.push(json!({ "path": child, "reason": kind })),1286
Some("directory") => self.walk_remote(sftp, session_id, plan, &child, &child_local, options, true, depth + 1).await?,1287
_ => add_download(1288
plan,1289
&child,1290
&child_local,1291
&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 },1292
),1293
}1294
}1295
}1296
}1297
Ok(())1298
}1299
.boxed()1300
}1302
async fn create_directories(&self, plan: &Plan, privileged: bool, password: &str) -> CoreResult<()> {1303
if plan.direction == "download" {1304
for directory in &plan.directories {1305
std::fs::create_dir_all(directory).map_err(io_core)?;1306
#[cfg(unix)]1307
{1308
use std::os::unix::fs::PermissionsExt;1309
let _ = std::fs::set_permissions(directory, std::fs::Permissions::from_mode(0o700));1310
}1311
}1312
return Ok(());1313
}1314
if plan.directories.is_empty() {1315
return Ok(());1316
}1317
if privileged {1318
let auth = Authorization { sudo: true, sudo_password: (!password.is_empty()).then(|| password.to_string()), ..Default::default() };1319
for directory in &plan.directories {1320
if let Err(error) = self.files.helper(&plan.session_id, json!({ "operation": "mkdir", "path": directory, "parents": true }), &auth).await {1321
if error.code != "FILE_EXISTS" {1322
return Err(error);1323
}1324
}1325
}1326
return Ok(());1327
}1328
let sftp = self.files.sftp(&plan.session_id).await?;1329
for directory in &plan.directories {1330
if let Err(error) = sftp.mkdir(directory, 0o755).await {1331
if !sftp.stat(directory).await.map(|s| s.is_dir()).unwrap_or(false) {1332
return Err(file_error(error.transfer_code(), &format!("Directory {directory} could not be created.")));1333
}1334
}1335
}1336
Ok(())1337
}1339
// ---- scheduling ------------------------------------------------------------------------------1341
fn schedule(self: &Arc<Self>) {1342
if self.scheduling.swap(true, Ordering::SeqCst) {1343
return;1344
}1345
let manager = self.clone();1346
tokio::spawn(async move {1347
tokio::task::yield_now().await;1348
manager.scheduling.store(false, Ordering::SeqCst);1349
let start: Vec<String> = {1350
let mut inner = manager.inner.lock();1351
if inner.closed {1352
return;1353
}1354
let mut running = inner.runtime.values().filter(|r| r.running).count();1355
let mut waiting: Vec<(u64, String)> = inner.jobs.values().filter(|j| j.state == "waiting").map(|j| (j.order, j.id.clone())).collect();1356
waiting.sort();1357
let max = inner.settings.max_concurrent as usize;1358
let mut start = Vec::new();1359
for (_, id) in waiting {1360
if running >= max {1361
break;1362
}1363
if inner.runtime.get(&id).is_some_and(|r| r.running) {1364
continue;1365
}1366
running += 1;1367
let transferred = inner.jobs[&id].transferred;1368
inner.runtime.entry(id.clone()).or_insert_with(|| Runtime::new(transferred)).running = true;1369
start.push(id);1370
}1371
start1372
};1373
for id in start {1374
let manager = manager.clone();1375
tokio::spawn(async move { manager.run(id).await });1376
}1377
});1378
}1380
async fn run(self: Arc<Self>, id: String) {1381
let stop_rx = {1382
let mut inner = self.inner.lock();1383
let transferred = inner.jobs.get(&id).map(|j| j.transferred).unwrap_or(0);1384
let rt = inner.runtime.entry(id.clone()).or_insert_with(|| Runtime::new(transferred));1385
rt.running = true;1386
rt.stop = watch::channel(String::new()).0;1387
rt.generation += 1;1388
rt.speed = 0.0;1389
rt.samples = 0;1390
rt.last_bytes = transferred;1391
rt.last_time = now();1392
rt.progress_since_resume = 0;1393
let rx = rt.stop.subscribe();1394
if let Some(job) = inner.jobs.get_mut(&id) {1395
job.error = None;1396
job.resumable = false;1397
}1398
rx1399
};1400
self.set(&id, "connecting", None);1401
let outcome: RunResult<()> = async {1402
self.check_host_keys(&id)?;1403
let session_id = self.session_for(&id)?;1404
self.update(&id, |j| j.session_id = json!(session_id));1405
let direction = self.with_job(&id, |j| j.direction.clone()).unwrap_or_default();1406
if direction == "upload" {1407
self.run_upload(&id, &session_id, stop_rx.clone()).await1408
} else if direction == "relay" {1409
self.run_relay(&id, &session_id, stop_rx.clone()).await1410
} else {1411
self.run_download(&id, &session_id, stop_rx.clone()).await1412
}1413
}1414
.await;1415
match outcome {1416
Ok(()) => {1417
self.finish_move(&id).await;1418
if let Some(batch) = self.with_job(&id, |j| j.batch_id.clone()) {1419
self.finish_batch_move(&batch).await;1420
}1421
}1422
Err(error) => self.failure(&id, error, &stop_rx).await,1423
}1424
if self.job_state(&id) == "completed" {1425
self.forget_secrets(&id);1426
}1427
if let Some(rt) = self.inner.lock().runtime.get_mut(&id) {1428
rt.running = false;1429
}1430
self.persist(true, false);1431
self.schedule();1432
}1434
async fn failure(self: &Arc<Self>, id: &str, error: RunError, stop: &watch::Receiver<String>) {1435
let requested = stop.borrow().clone();1436
let error = match error {1437
RunError::Stop(kind) if kind != "lost" => Err(kind),1438
RunError::Stop(_) => Ok(CoreError::new("SESSION_NOT_CONNECTED", "The SSH connection was lost.")),1439
RunError::Error(_) if !requested.is_empty() && requested != "lost" => Err(requested),1440
RunError::Error(e) => Ok(e),1441
};1442
let error = match error {1443
Err(kind) => {1444
match kind.as_str() {1445
"cancel" => self.finish_cancel(id).await,1446
"pause" => self.set(id, "paused", None),1447
_ => self.set(id, "interrupted", Some(JobError { code: "APP_RESTARTED".into(), message: "The application was closed during the transfer.".into() })),1448
}1449
return;1450
}1451
Ok(error) => error,1452
};1453
let code = error_code(&error);1454
let message: String = Regex::strip_code(&error.message).chars().take(400).collect();1455
let (state, transferred, size, direction) = self.with_job(id, |j| (j.state.clone(), j.transferred, j.size, j.direction.clone())).unwrap_or_default();1456
self.logger.warn("Transfer interrupted.", json!({ "jobId": id, "direction": direction, "state": state, "code": code, "transferred": transferred, "size": size }));1457
if state == "finalizing" && (code == "FILE_RESULT_UNKNOWN" || CONNECTION_CODES.contains(&code.as_str())) {1458
self.update(id, |j| j.finalize["state"] = json!("unknown"));1459
}1460
if AUTH_CODES.contains(&code.as_str()) {1461
self.update(id, |j| j.auth_required = true);1462
self.set(id, "interrupted", Some(JobError { code, message }));1463
return;1464
}1465
if CONNECTION_CODES.contains(&code.as_str()) || code == "FILE_RESULT_UNKNOWN" {1466
self.set(id, "interrupted", Some(JobError { code, message }));1467
let auto = self.inner.lock().settings.auto_retry;1468
let retries = self.with_job(id, |j| j.total_retries).unwrap_or(0);1469
if auto && retries < MAX_TOTAL_RETRIES {1470
self.schedule_retry(id);1471
}1472
return;1473
}1474
let resumable = ["FILE_NO_SPACE", "FILE_EXISTS", "FILE_CONFLICT", "LOCAL_NO_SPACE", "ENOSPC", "FILE_READ_ONLY_FILESYSTEM", "FILE_BUSY"].contains(&code.as_str());1475
self.update(id, |j| j.resumable = resumable);1476
self.set(id, "failed", Some(JobError { code: if code == "ENOSPC" { "LOCAL_NO_SPACE".into() } else { code }, message }));1477
}1479
fn schedule_retry(self: &Arc<Self>, id: &str) {1480
self.clear_retry(id);1481
let delay = {1482
let mut inner = self.inner.lock();1483
let Some(job) = inner.jobs.get_mut(id) else { return };1484
if job.attempts as usize >= RETRY_DELAYS_MS.len() {1485
return; // Manual resume from here on: no endless loop.1486
}1487
let delay = RETRY_DELAYS_MS[job.attempts as usize];1488
job.attempts += 1;1489
job.total_retries += 1;1490
job.next_retry_at = Some(now() + delay as i64);1491
delay1492
};1493
let error = self.with_job(id, |j| j.error.clone()).flatten();1494
self.set(id, "reconnecting", error.clone());1495
let manager = self.clone();1496
let job_id = id.to_string();1497
let task = tokio::spawn(async move {1498
tokio::time::sleep(Duration::from_millis(delay)).await;1499
if let Some(rt) = manager.inner.lock().runtime.get_mut(&job_id) {1500
rt.retry_task = None;1501
}1502
if manager.job_state(&job_id) != "reconnecting" {1503
return;1504
}1505
let attempts = manager.with_job(&job_id, |j| j.attempts).unwrap_or(0) as usize;1506
if manager.find_session(&job_id).is_some() {1507
manager.set(&job_id, "waiting", None);1508
manager.schedule();1509
} else if attempts < RETRY_DELAYS_MS.len() {1510
manager.schedule_retry(&job_id);1511
} else {1512
let error = manager.with_job(&job_id, |j| j.error.clone()).flatten();1513
manager.set(&job_id, "interrupted", error);1514
}1515
});1516
let mut inner = self.inner.lock();1517
let transferred = inner.jobs.get(id).map(|j| j.transferred).unwrap_or(0);1518
inner.runtime.entry(id.to_string()).or_insert_with(|| Runtime::new(transferred)).retry_task = Some(task);1519
}1521
fn clear_retry(&self, id: &str) {1522
let mut inner = self.inner.lock();1523
if let Some(task) = inner.runtime.get_mut(id).and_then(|r| r.retry_task.take()) {1524
task.abort();1525
}1526
if let Some(job) = inner.jobs.get_mut(id) {1527
job.next_retry_at = None;1528
}1529
}1531
fn session_state(self: &Arc<Self>, status: &Value) {1532
if status["state"] == "disconnected" {1533
// A dropped SSH connection does not answer outstanding SFTP requests: end those jobs now.1534
let session = status["sessionId"].clone();1535
let ids: Vec<String> = {1536
let inner = self.inner.lock();1537
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()1538
};1539
for id in ids {1540
self.stop(&id, "lost");1541
}1542
return;1543
}1544
if status["state"] != "connected" {1545
return;1546
}1547
let identity = SessionIdentity::of_status(status).key();1548
let candidates: Vec<String> = {1549
let inner = self.inner.lock();1550
inner1551
.jobs1552
.values()1553
.filter(|j| j.identity.key() == identity || relay_source(j).is_some_and(|(_, source)| source.key() == identity))1554
.filter(|j| {1555
j.state == "reconnecting"1556
|| (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)1557
})1558
.map(|j| j.id.clone())1559
.collect()1560
};1561
for id in candidates {1562
self.clear_retry(&id);1563
self.set(&id, "waiting", None);1564
}1565
self.inner.lock().cleaned_identities.remove(&identity);1566
self.cleanup_orphans_later();1567
self.schedule();1568
}1570
fn stop(&self, id: &str, kind: &str) {1571
let inner = self.inner.lock();1572
if let Some(rt) = inner.runtime.get(id) {1573
let _ = rt.stop.send(kind.to_string());1574
}1575
}1577
// ---- transfer engine -----------------------------------------------------------------------1579
fn helper_auth(&self, id: &str, privileged_override: bool, timeout: Option<u64>) -> Authorization {1580
let inner = self.inner.lock();1581
let privileged = privileged_override || inner.jobs.get(id).is_some_and(|j| j.mode == "privileged");1582
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() };1583
auth.timeout_ms = timeout;1584
auth1585
}1587
fn source_auth(&self, job: &Job) -> Authorization {1588
if job.direction != "relay" {1589
return self.helper_auth(&job.id, job.mode == "privileged", Some(hash_timeout(job.size)));1590
}1591
Authorization {1592
sudo: job.extra.get("sourceSudo") == Some(&Value::Bool(true)),1593
sudo_password: self.inner.lock().secrets.get(&format!("{}:source", job.id)).map(|s| s.to_string()),1594
timeout_ms: Some(hash_timeout(job.size)),1595
..Default::default()1596
}1597
}1599
fn forget_secrets(&self, id: &str) {1600
let mut inner = self.inner.lock();1601
inner.secrets.remove(id);1602
inner.secrets.remove(&format!("{id}:source"));1603
}1605
async fn open_channel(&self, session_id: &str, stop: &watch::Receiver<String>) -> RunResult<Arc<Sftp>> {1606
let sftp = self.files.sftp(session_id).await?;1607
let kind = stop.borrow().clone();1608
if !kind.is_empty() {1609
return Err(RunError::Stop(kind));1610
}1611
Ok(sftp)1612
}1614
async fn run_upload(self: &Arc<Self>, id: &str, session_id: &str, stop: watch::Receiver<String>) -> RunResult<()> {1615
let job = self.snapshot(id)?;1616
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)).with("sourceSudo", true))?;1617
if !meta.is_file() {1618
return Err(file_error("LOCAL_NOT_FILE", &format!("{} is no longer a regular file.", job.local_path)).into());1619
}1620
let current = LocalIdentity::of(&meta);1621
if !job.local_identity.as_ref().is_some_and(|i| i.same(¤t)) {1622
if job.transferred > 0 {1623
return Err(file_error("FILE_SOURCE_CHANGED", "The local file changed since the transfer started; restart it from the beginning.").into());1624
}1625
self.update(id, |j| {1626
j.local_identity = Some(current.clone());1627
j.size = current.size;1628
});1629
}1630
let job = self.snapshot(id)?;1631
if job.finalize_state() == "unknown" || (job.hash.is_some() && job.transferred == job.size && job.part_path.is_some()) {1632
return self.finalize_upload(id, session_id, &stop).await;1633
}1634
self.prepare_upload_target(id, session_id).await?;1635
let sftp = self.open_channel(session_id, &stop).await?;1636
let local = LocalFile::open(Path::new(&job.local_path), "r").map_err(io_core)?;1637
let handle = self.open_remote_part(&sftp, id).await?;1638
let result: RunResult<()> = async {1639
let job = self.snapshot(id)?;1640
if job.transferred > 0 {1641
let remote = sftp.fstat(&handle).await.map_err(sftp_core)?;1642
let transferred = job.transferred.min(remote.size);1643
self.update(id, |j| j.transferred = transferred);1644
let window = TAIL_CHECK.min(transferred);1645
let position = transferred - window;1646
verify_tail(1647
transferred,1648
async { local.read_at(position, window as usize).await.map_err(io_core) }.boxed(),1649
async { sftp.read_full(&handle, position, window as usize).await.map_err(sftp_core) }.boxed(),1650
window,1651
)1652
.await?;1653
}1654
self.set(id, "transferring", None);1655
let size = self.with_job(id, |j| j.size).unwrap_or(0);1656
let produce: Producer = {1657
let local = local.clone();1658
Arc::new(move |pos, len| {1659
let local = local.clone();1660
Box::pin(async move {1661
let data = local.read_at(pos, len).await.map_err(io_core)?;1662
if data.len() != len {1663
return Err(file_error("FILE_SOURCE_CHANGED", "The local file shrank during the transfer."));1664
}1665
Ok(data)1666
})1667
})1668
};1669
let consume: Consumer = {1670
let (sftp, handle) = (sftp.clone(), handle.clone());1671
Arc::new(move |buf, pos| {1672
let (sftp, handle) = (sftp.clone(), handle.clone());1673
Box::pin(async move { sftp.write_at(&handle, pos, &buf).await.map_err(sftp_core) })1674
})1675
};1676
self.run_pump(id, size, produce, consume, stop.clone()).await?;1677
let remote = sftp.fstat(&handle).await.map_err(sftp_core)?;1678
if remote.size > size {1679
sftp.truncate(&handle, size).await.map_err(sftp_core)?;1680
}1681
let _ = sftp.fsync(&handle).await;1682
let _ = sftp.close_file(&handle).await;1683
let after = local.metadata().map(|m| LocalIdentity::of(&m)).map_err(io_core)?;1684
if !self.with_job(id, |j| j.local_identity.as_ref().is_some_and(|i| i.same(&after))).unwrap_or(false) {1685
return Err(file_error("FILE_SOURCE_CHANGED", "The local file changed during the transfer.").into());1686
}1687
self.set(id, "verifying", None);1688
let hash = local_hash(Path::new(&job.local_path), stop.clone()).await?;1689
self.update(id, |j| {1690
j.hash = Some(hash.clone());1691
j.verification = Some(json!({ "method": "sha256", "local": hash }));1692
});1693
Ok(())1694
}1695
.await;1696
if result.is_err() {1697
let _ = tokio::time::timeout(Duration::from_secs(3), sftp.close_file(&handle)).await;1698
}1699
drop(sftp);1700
result?;1701
self.finalize_upload(id, session_id, &stop).await1702
}1704
/// Like an upload, with an SFTP file as the source: same part file, publish step and resume rules.1705
async fn run_relay(self: &Arc<Self>, id: &str, session_id: &str, stop: watch::Receiver<String>) -> RunResult<()> {1706
let result = self.run_relay_inner(id, session_id, stop).await;1707
if let Err(RunError::Error(error)) = &result {1708
if AUTH_CODES.contains(&error.code.as_str()) {1709
self.update(id, |j| { j.extra.insert("sourceAuthRequired".into(), json!(error.details.get("sourceSudo") == Some(&Value::Bool(true)))); });1710
}1711
}1712
if self.cleanup_staged(id).await.is_err() {1713
self.update(id, |j| j.cleanup_pending = true);1714
} else if result.is_ok() {1715
self.update(id, |j| j.cleanup_pending = false);1716
}1717
result1718
}1720
async fn run_relay_inner(self: &Arc<Self>, id: &str, session_id: &str, stop: watch::Receiver<String>) -> RunResult<()> {1721
let job = self.snapshot(id)?;1722
if job.extra.get("sourceSudo") == Some(&Value::Bool(true)) && (job.finalize_state() == "unknown" || job.hash.is_some()) {1723
let session = self.relay_session(id)?;1724
let sftp = self.open_channel(&session, &stop).await?;1725
let info = self.source_info(&sftp, &session, &job.local_path, &self.source_auth(&job)).await?;1726
if info.kind != "file" || job.remote_identity.as_ref() != Some(&RemoteIdentity { size: info.size, mtime: info.mtime }) {1727
return Err(file_error("FILE_SOURCE_CHANGED", "The source file changed since the transfer started.").into());1728
}1729
}1730
if job.finalize_state() == "unknown" || (job.hash.is_some() && job.transferred == job.size && job.part_path.is_some()) {1731
return self.finalize_upload(id, session_id, &stop).await;1732
}1733
let source_session = self.relay_session(id)?;1734
let source = self.open_channel(&source_session, &stop).await?;1735
let auth = self.source_auth(&job);1736
let stat = self.source_info(&source, &source_session, &job.local_path, &auth).await?;1737
if stat.kind != "file" {1738
return Err(file_error("FILE_UNSAFE_TYPE_OR_LINK", &format!("{} is no longer a regular file.", job.local_path)).into());1739
}1740
let current = RemoteIdentity { size: stat.size, mtime: stat.mtime as u64 };1741
if job.remote_identity.as_ref() != Some(¤t) {1742
if relay_source_changed(&job, ¤t, auth.sudo) {1743
return Err(file_error("FILE_SOURCE_CHANGED", "The source file changed since the transfer started; restart it from the beginning.").into());1744
}1745
self.update(id, |j| {1746
j.remote_identity = Some(current.clone());1747
j.size = current.size;1748
});1749
}1750
let source_path = if auth.sudo {1751
self.ensure_staged(id, &source, &source_session).await?1752
} else { job.local_path.clone() };1753
self.prepare_upload_target(id, session_id).await?;1754
let destination = self.open_channel(session_id, &stop).await?;1755
let input = source.open_file(&source_path, "r", None).await.map_err(|e| file_error(e.transfer_code(), &format!("{} cannot be read.", job.local_path)).with("sourceSudo", true))?;1756
let output = match self.open_remote_part(&destination, id).await {1757
Ok(handle) => handle,1758
Err(error) => {1759
let _ = tokio::time::timeout(Duration::from_secs(3), source.close_file(&input)).await;1760
return Err(error);1761
}1762
};1763
let result: RunResult<()> = async {1764
let job = self.snapshot(id)?;1765
if job.transferred > 0 {1766
let written = destination.fstat(&output).await.map_err(sftp_core)?;1767
let transferred = job.transferred.min(written.size);1768
self.update(id, |j| j.transferred = transferred);1769
let window = TAIL_CHECK.min(transferred);1770
let position = transferred - window;1771
verify_tail(1772
transferred,1773
async { source.read_full(&input, position, window as usize).await.map_err(sftp_core) }.boxed(),1774
async { destination.read_full(&output, position, window as usize).await.map_err(sftp_core) }.boxed(),1775
window,1776
)1777
.await?;1778
}1779
self.set(id, "transferring", None);1780
let size = self.with_job(id, |j| j.size).unwrap_or(0);1781
let produce: Producer = {1782
let (source, input) = (source.clone(), input.clone());1783
Arc::new(move |pos, len| {1784
let (source, input) = (source.clone(), input.clone());1785
Box::pin(async move {1786
let data = source.read_full(&input, pos, len).await.map_err(sftp_core)?;1787
if data.len() != len {1788
return Err(file_error("FILE_SOURCE_CHANGED", "The source file shrank during the transfer."));1789
}1790
Ok(data)1791
})1792
})1793
};1794
let consume: Consumer = {1795
let (destination, output) = (destination.clone(), output.clone());1796
Arc::new(move |buf, pos| {1797
let (destination, output) = (destination.clone(), output.clone());1798
Box::pin(async move { destination.write_at(&output, pos, &buf).await.map_err(sftp_core) })1799
})1800
};1801
self.run_pump(id, size, produce, consume, stop.clone()).await?;1802
let written = destination.fstat(&output).await.map_err(sftp_core)?;1803
if written.size > size {1804
destination.truncate(&output, size).await.map_err(sftp_core)?;1805
}1806
let _ = destination.fsync(&output).await;1807
let _ = destination.close_file(&output).await;1808
self.set(id, "verifying", None);1809
let hash = if auth.sudo {1810
self.snapshot(id)?.staged.as_ref().and_then(|s| s["hash"].as_str()).filter(|h| valid_hash(h)).map(str::to_string)1811
.ok_or_else(|| file_error("FILE_VERIFY_FAILED", "The staged source has no verified hash."))?1812
} else { self.relay_hash(&source_session, &source, &input, &source_path, size, &stop).await? };1813
let after = self.source_info(&source, &source_session, &job.local_path, &auth).await?;1814
if after.kind != "file" || (RemoteIdentity { size: after.size, mtime: after.mtime }) != current {1815
return Err(file_error("FILE_SOURCE_CHANGED", "The source file changed during the transfer.").into());1816
}1817
self.update(id, |j| {1818
j.hash = Some(hash.clone());1819
j.verification = Some(json!({ "method": "sha256", "local": hash }));1820
});1821
Ok(())1822
}1823
.await;1824
let _ = tokio::time::timeout(Duration::from_secs(3), source.close_file(&input)).await;1825
if result.is_err() {1826
let _ = tokio::time::timeout(Duration::from_secs(3), destination.close_file(&output)).await;1827
}1828
drop((source, destination));1829
if let Err(RunError::Error(_)) = &result {1830
// Either end may drop before its disconnect event arrives: that is a lost connection, not a broken file.1831
if [source_session.as_str(), session_id].iter().any(|s| self.sessions.get(s).is_none_or(|r| *r.state.lock() != "connected")) {1832
return Err(file_error("SESSION_NOT_CONNECTED", "A connection of this transfer ended.").into());1833
}1834
}1835
result?;1836
self.finalize_upload(id, session_id, &stop).await1837
}1839
fn relay_session(&self, id: &str) -> RunResult<String> {1840
let job = self.snapshot(id)?;1841
let (preferred, identity) = relay_source(&job).ok_or_else(|| RunError::Error(file_error("TRANSFER_INVALID_SELECTION", "The transfer has no source server.")))?;1842
let session = matching_session(&self.sessions.list(), &identity, &preferred).ok_or_else(|| {1843
RunError::Error(file_error("SESSION_NOT_CONNECTED", &format!("Connect to {}@{} to continue.", identity.username.as_str().unwrap_or(""), identity.host.as_str().unwrap_or(""))))1844
})?;1845
self.update(id, |j| {1846
if let Some(relay) = j.extra.get_mut("relaySource") {1847
relay["sessionId"] = json!(session);1848
}1849
});1850
Ok(session)1851
}1853
/// SHA-256 of the source: on the source server if it can, otherwise read over SFTP here.1854
async fn relay_hash(&self, session: &str, sftp: &Sftp, handle: &FileHandle, path: &str, size: u64, stop: &watch::Receiver<String>) -> RunResult<String> {1855
let auth = Authorization::default().with_timeout(hash_timeout(size));1856
if let Ok(value) = self.files.helper(session, json!({ "operation": "hash", "path": path }), &auth).await {1857
if let Some(hash) = value["hash"].as_str() {1858
return Ok(hash.to_string());1859
}1860
}1861
if let Ok(hash) = remote_checksum(&self.sessions, session, path).await {1862
return Ok(hash);1863
}1864
let mut hasher = Sha256::new();1865
let mut position = 0;1866
while position < size {1867
let kind = stop.borrow().clone();1868
if !kind.is_empty() {1869
return Err(RunError::Stop(kind));1870
}1871
let length = (size - position).min(1024 * 1024) as usize;1872
let data = sftp.read_full(handle, position, length).await.map_err(sftp_core)?;1873
if data.len() != length {1874
return Err(file_error("FILE_SOURCE_CHANGED", "The source file shrank during the transfer.").into());1875
}1876
hasher.update(&data);1877
position += length as u64;1878
}1879
Ok(hex::encode(hasher.finalize()))1880
}1882
/// A move removes its source only after a completed transfer whose SHA-256 matched and only if1883
/// the source is still the file that was copied. In any other case the source stays.1884
async fn finish_move(self: &Arc<Self>, id: &str) {1885
let Ok(job) = self.snapshot(id) else { return };1886
if job.state != "completed" || job.extra.get("moveSource") != Some(&Value::Bool(true)) || job.extra.contains_key("moveResult") {1887
return;1888
}1889
let verified = move_verified(&job);1890
let result: Result<(), &'static str> = if verified { self.remove_source(&job).await } else { Err("unverified") };1891
if let Err(reason) = result {1892
self.logger.warn("Move source removal was not confirmed.", json!({ "jobId": id, "reason": reason }));1893
}1894
self.update(id, |j| {1895
j.extra.entry("moveResult".to_string()).or_insert_with(|| match result {1896
Ok(()) => json!({ "state": "deleted" }),1897
Err(reason) => json!({ "state": "kept", "reason": reason }),1898
});1899
});1900
self.mark_job(id);1901
self.emit_soon();1902
self.persist(true, false);1903
}1905
async fn remove_source(self: &Arc<Self>, job: &Job) -> Result<(), &'static str> {1906
if !move_verified(job) { return Err("unverified"); }1907
let root = job.extra.get("moveRoot").and_then(Value::as_str).unwrap_or("").to_string();1908
if job.direction == "upload" {1909
let path = Path::new(&job.local_path);1910
let meta = std::fs::symlink_metadata(path).map_err(|_| "source-gone")?;1911
if !meta.is_file() || !job.local_identity.as_ref().is_some_and(|i| i.same(&LocalIdentity::of(&meta))) {1912
return Err("source-changed");1913
}1914
std::fs::remove_file(path).map_err(|_| "remove-failed")?;1915
let mut parent = path.parent();1916
while let Some(directory) = parent.filter(|d| !root.is_empty() && d.starts_with(&root) && *d != Path::new(&root)) {1917
if std::fs::remove_dir(directory).is_err() {1918
break;1919
}1920
parent = directory.parent();1921
}1922
return Ok(());1923
}1924
let (session, path) = if job.direction == "relay" {1925
let (preferred, identity) = relay_source(job).ok_or("source-gone")?;1926
(matching_session(&self.sessions.list(), &identity, &preferred).ok_or("source-disconnected")?, job.local_path.clone())1927
} else {1928
(self.find_session(&job.id).ok_or("source-disconnected")?, job.remote_path.clone())1929
};1930
let auth = self.source_auth(job);1931
if auth.sudo {1932
let expected = job.remote_identity.as_ref().ok_or("source-changed")?;1933
// Persist before the call: a helper or app crash must never claim the1934
// original name was kept. The adjacent remote journal permits recovery.1935
self.update(&job.id, |j| { j.extra.insert("moveResult".into(), remote_move_result(Err("FILE_RESULT_UNKNOWN"))); });1936
self.persist(true, true);1937
let response = self.files.helper(&session, json!({ "operation": "remove-source", "path": path, "root": root,1938
"expected": expected, "hash": job.verification.as_ref().unwrap()["local"] }), &auth).await;1939
let result = remote_move_result(response.as_ref().map_err(|e| e.code.as_str()));1940
let removed = result["state"] == "deleted";1941
self.update(&job.id, |j| { j.extra.insert("moveResult".into(), result); });1942
return if removed { Ok(()) } else { Err("remove-failed") };1943
}1944
let sftp = self.files.sftp(&session).await.map_err(|_| "source-disconnected")?;1945
let stat = sftp.lstat(&path).await.map_err(|_| "source-gone")?;1946
if !stat.is_file() || job.remote_identity.as_ref() != Some(&RemoteIdentity { size: stat.size, mtime: stat.mtime as u64 }) {1947
return Err("source-changed");1948
}1949
sftp.unlink(&path).await.map_err(|_| "remove-failed")?;1950
let mut directory = posix::dirname(&path);1951
while !root.is_empty() && directory != root && directory.starts_with(&format!("{}/", root.trim_end_matches('/'))) {1952
if sftp.rmdir(&directory).await.is_err() {1953
break;1954
}1955
directory = posix::dirname(&directory);1956
}1957
Ok(())1958
}1960
// Deliberate no-clobber recovery through the existing Resume action. This1961
// restores a claimed entry only; it never retries source deletion or copying.1962
async fn recover_move(self: &Arc<Self>, id: &str) {1963
let response: RunResult<Value> = async {1964
let job = self.snapshot(id)?;1965
self.check_host_keys(id)?;1966
let session = if job.direction == "relay" { self.relay_session(id)? } else { self.session_for(id)? };1967
let path = if job.direction == "relay" { &job.local_path } else { &job.remote_path };1968
Ok(self.files.helper(&session, json!({ "operation": "recover-source", "path": path }), &self.source_auth(&job)).await?)1969
}.await;1970
match response {1971
Ok(value) => self.update(id, |j| {1972
j.extra.insert("moveResult".into(), remote_move_result(Ok(&value)));1973
j.error = None;1974
}),1975
Err(RunError::Error(error)) => self.update(id, |j| {1976
j.error = Some(JobError { code: error_code(&error), message: Regex::strip_code(&error.message).chars().take(400).collect() });1977
}),1978
Err(RunError::Stop(_)) => {}1979
}1980
self.forget_secrets(id);1981
if let Some(rt) = self.inner.lock().runtime.get_mut(id) { rt.running = false; }1982
self.mark_job(id);1983
self.persist(true, false);1984
}1986
/// After every file of a moved batch is gone from its source, removes the source folders, deepest1987
/// first. Removing a folder fails while it holds anything, such as a skipped symlink or a kept file.1988
async fn finish_batch_move(self: &Arc<Self>, batch: &str) {1989
let pending = {1990
let mut inner = self.inner.lock();1991
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"));1992
if !done { return; }1993
inner.batch_moves.remove(batch)1994
};1995
let Some(mut pending) = pending else { return };1996
pending.directories.sort_by_key(|d| std::cmp::Reverse(d.len()));1997
match pending.session {1998
None => {1999
for directory in &pending.directories {2000
let _ = std::fs::remove_dir(directory);2001
}2002
}2003
Some(session) => {2004
if let Ok(sftp) = self.files.sftp(&session).await {2005
for directory in &pending.directories {2006
let _ = sftp.rmdir(directory).await;2007
}2008
}2009
}2010
}2011
}2013
/// A new connection is verified against the trusted key, but that key can have been replaced since2014
/// the job started: then this is possibly another machine, and the job must not write or delete there.2015
fn check_host_keys(&self, id: &str) -> RunResult<()> {2016
let job = self.snapshot(id)?;2017
let session = self.find_session(id);2018
let source = relay_source(&job).and_then(|(preferred, identity)| matching_session(&self.sessions.list(), &identity, &preferred));2019
for (stored, current) in [("hostKey", session), ("sourceHostKey", source)] {2020
let (Some(expected), Some(current)) = (job.extra.get(stored).and_then(Value::as_str), current) else { continue };2021
if self.sessions.trusted_host_key(¤t).as_deref() != Some(expected) {2022
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());2023
}2024
}2025
Ok(())2026
}2028
async fn run_pump(self: &Arc<Self>, id: &str, size: u64, produce: Producer, consume: Consumer, stop: watch::Receiver<String>) -> RunResult<()> {2029
let start = self.with_job(id, |j| j.transferred).unwrap_or(0);2030
let manager = self.clone();2031
let job_id = id.to_string();2032
let progress = move |confirmed: u64| manager.progress(&job_id, confirmed);2033
match pump(start, size, produce, consume, stop, progress).await {2034
Ok(_) => Ok(()),2035
Err(PumpStop::Stopped(kind)) => Err(RunError::Stop(kind)),2036
Err(PumpStop::Failed(error)) => Err(RunError::Error(error)),2037
}2038
}2040
async fn prepare_upload_target(self: &Arc<Self>, id: &str, session_id: &str) -> RunResult<()> {2041
let job = self.snapshot(id)?;2042
if job.resolved_path.is_some() && job.part_path.is_some() && job.target_baseline.is_some() {2043
return Ok(());2044
}2045
let auth = self.helper_auth(id, false, None);2046
let stat = match self.files.helper(session_id, json!({ "operation": "stat", "path": job.remote_path }), &auth).await {2047
Ok(stat) => {2048
self.update(id, |j| j.helper = Some(true));2049
stat2050
}2051
Err(error) => {2052
if !NO_HELPER_CODES.contains(&error.code.as_str()) || job.mode == "privileged" {2053
return Err(error.into());2054
}2055
// No python3 / Windows server: plain SFTP is enough for new files; existing ones need2056
// the metadata-preserving Linux helper.2057
self.update(id, |j| j.helper = Some(false));2058
let sftp = self.files.sftp(session_id).await?;2059
if sftp.exists(&job.remote_path).await {2060
let message = if error.code == "FEATURE_PLATFORM_UNSUPPORTED" {2061
"Replacing an existing file on this server is not supported (no metadata-preserving helper); remove it first or choose another name."2062
} else {2063
"Replacing an existing file requires python3 on the server to preserve its owner, permissions and ACLs."2064
};2065
return Err(file_error(&error.code, message).into());2066
}2067
json!({ "exists": false, "resolvedPath": job.remote_path })2068
}2069
};2070
let exists = stat["exists"] == true;2071
if exists && stat["type"] != "file" {2072
return Err(file_error("FILE_UNSAFE_TYPE_OR_LINK", &format!("{} is not a regular file.", job.remote_path)).into());2073
}2074
if exists && number(&stat["baseline"]["nlink"]).unwrap_or(1.0) > 1.0 {2075
return Err(file_error("FILE_HARDLINK_UNSUPPORTED", &format!("{} has several hard links.", job.remote_path)).into());2076
}2077
if exists && !job.overwrite {2078
return Err(file_error("FILE_EXISTS", &format!("{} already exists.", job.remote_path)).into());2079
}2080
let resolved = stat["resolvedPath"].as_str().filter(|s| !s.is_empty()).unwrap_or(&job.remote_path).to_string();2081
let baseline = if exists { stat["baseline"].clone() } else { Value::Null };2082
let helper = self.with_job(id, |j| j.helper == Some(true)).unwrap_or(false);2083
let (owner_uid, part_path) = if job.mode == "privileged" {2084
let workspace = self.files.helper(session_id, json!({ "operation": "transfer-workspace" }), &Authorization::default()).await?;2085
(Some(workspace["uid"].clone()), posix::join(workspace["path"].as_str().unwrap_or(""), &format!("{}.part", job.id)))2086
} else {2087
let uid = if helper { Some(self.files.helper(session_id, json!({ "operation": "transfer-workspace" }), &Authorization::default()).await?["uid"].clone()) } else { None };2088
(uid, posix::join(&posix::dirname(&resolved), &part_name(&posix::basename(&resolved), &job.id)))2089
};2090
self.update(id, |j| {2091
j.resolved_path = Some(resolved.clone());2092
j.target_baseline = Some(baseline.clone());2093
if owner_uid.is_some() {2094
j.owner_uid = owner_uid.clone();2095
}2096
j.part_path = Some(part_path.clone());2097
});2098
self.persist(true, false);2099
Ok(())2100
}2102
async fn open_remote_part(&self, sftp: &Sftp, id: &str) -> RunResult<FileHandle> {2103
let job = self.snapshot(id)?;2104
let part = job.part_path.clone().unwrap_or_default();2105
if job.transferred > 0 {2106
match sftp.lstat(&part).await {2107
Ok(stat) if stat.is_file() => return sftp.open_file(&part, "r+", None).await.map_err(|e| sftp_core(e).into()),2108
_ => self.update(id, |j| j.transferred = 0), // The partial file disappeared; start this file again.2109
}2110
}2111
match sftp.open_file(&part, "wx", Some(0o600)).await {2112
Ok(handle) => Ok(handle),2113
Err(error) => {2114
// Only our own leftover (same job id) can exist here; remove the entry itself, never follow it.2115
if error.transfer_code() == "FILE_PERMISSION_DENIED" {2116
return Err(file_error("FILE_PERMISSION_DENIED", &format!("{} is not writable.", posix::dirname(&part))).into());2117
}2118
let _ = sftp.unlink(&part).await;2119
sftp.open_file(&part, "wx", Some(0o600)).await.map_err(|e| sftp_core(e).into())2120
}2121
}2122
}2124
async fn finalize_upload(self: &Arc<Self>, id: &str, session_id: &str, stop: &watch::Receiver<String>) -> RunResult<()> {2125
let kind = stop.borrow().clone();2126
if !kind.is_empty() {2127
return Err(RunError::Stop(kind));2128
}2129
self.set(id, "finalizing", None);2130
let retry = self.with_job(id, |j| ["unknown", "running"].contains(&j.finalize_state())).unwrap_or(false);2131
self.update(id, |j| j.finalize = json!({ "state": "running", "startedAt": now() }));2132
self.persist(true, false);2133
let job = self.snapshot(id)?;2134
if job.helper == Some(false) {2135
return self.publish_without_helper(id, session_id).await;2136
}2137
let auth = self.helper_auth(id, job.finalize_with_sudo, Some(hash_timeout(job.size)));2138
let request = json!({2139
"operation": "publish",2140
"source": job.part_path,2141
"uid": job.owner_uid,2142
"path": job.remote_path,2143
"hash": job.hash,2144
"size": job.size,2145
"baseline": job.target_baseline.clone().unwrap_or(Value::Null),2146
"overwrite": job.overwrite,2147
"strategy": if job.mode == "privileged" { "copy" } else { "rename" },2148
"parents": true,2149
"retry": retry,2150
"mode": 0o6442151
});2152
let result = self.files.helper(session_id, request, &auth).await?;2153
self.update(id, |j| {2154
j.finalize = json!({2155
"state": "done",2156
"resolvedPath": result["resolvedPath"],2157
"created": result["created"] == true,2158
"metadataPreserved": result["metadataPreserved"] == true,2159
"alreadyPublished": result["alreadyPublished"] == true,2160
"finishedAt": now()2161
});2162
let mut verification = j.verification.clone().unwrap_or_else(|| json!({}));2163
verification["remote"] = result["hash"].clone();2164
verification["method"] = json!("sha256");2165
j.verification = Some(verification);2166
});2167
if result["alreadyPublished"] == true && job.mode == "privileged" {2168
let _ = self.files.helper(session_id, json!({ "operation": "transfer-remove", "id": job.id }), &Authorization::default()).await;2169
}2170
self.set(id, "completed", None);2171
self.logger.info("Transfer completed.", json!({ "jobId": id, "direction": "upload", "bytes": job.size, "mode": job.mode, "path": job.remote_path }));2172
Ok(())2173
}2175
async fn publish_without_helper(self: &Arc<Self>, id: &str, session_id: &str) -> RunResult<()> {2176
let job = self.snapshot(id)?;2177
let part = job.part_path.clone().unwrap_or_default();2178
let remote_hash = remote_checksum(&self.sessions, session_id, &part).await.ok();2179
if job.direction == "relay" && job.extra.get("sourceSudo") == Some(&Value::Bool(true)) && remote_hash.is_none() {2180
return Err(file_error("FILE_VERIFY_FAILED", "The destination hash could not be verified against the staged source.").into());2181
}2182
if remote_hash.as_ref().is_some_and(|h| Some(h) != job.hash.as_ref()) {2183
let _ = self.discard_partial(id).await;2184
self.update(id, |j| j.transferred = 0);2185
return Err(file_error("FILE_VERIFY_FAILED", "The uploaded data does not match the local file.").into());2186
}2187
self.update(id, |j| {2188
let mut verification = j.verification.clone().unwrap_or_else(|| json!({}));2189
verification["remote"] = json!(remote_hash);2190
verification["method"] = json!(if remote_hash.is_some() { "sha256" } else { "size" });2191
j.verification = Some(verification);2192
});2193
let sftp = self.files.sftp(session_id).await?;2194
let stat = sftp.stat(&part).await.map_err(sftp_core)?;2195
if stat.size != job.size {2196
return Err(file_error("FILE_VERIFY_FAILED", "The uploaded size does not match the local file.").into());2197
}2198
if sftp.exists(&job.remote_path).await {2199
return Err(file_error("FILE_EXISTS", &format!("{} appeared during the transfer.", job.remote_path)).into());2200
}2201
sftp.rename(&part, &job.remote_path).await.map_err(sftp_core)?;2202
self.update(id, |j| j.finalize = json!({ "state": "done", "resolvedPath": j.remote_path, "created": true, "metadataPreserved": false, "finishedAt": now() }));2203
self.set(id, "completed", None);2204
Ok(())2205
}2207
async fn run_download(self: &Arc<Self>, id: &str, session_id: &str, stop: watch::Receiver<String>) -> RunResult<()> {2208
let sftp = self.open_channel(session_id, &stop).await?;2209
let job = self.snapshot(id)?;2210
let mut source = job.remote_path.clone();2211
if job.mode == "privileged" {2212
source = self.ensure_staged(id, &sftp, session_id).await?;2213
} else if job.mode == "privileged-allowed" && sftp.stat(&job.remote_path).await.is_err_and(|e| e.transfer_code() == "FILE_PERMISSION_DENIED") {2214
self.update(id, |j| j.mode = "privileged".into());2215
source = self.ensure_staged(id, &sftp, session_id).await?;2216
} else {2217
let stat = sftp.stat(&job.remote_path).await.map_err(|e| file_error(e.transfer_code(), &format!("{} is not accessible.", job.remote_path)))?;2218
let current = RemoteIdentity { size: stat.size, mtime: stat.mtime as u64 };2219
if job.remote_identity.as_ref().is_some_and(|r| *r != current) && job.transferred > 0 {2220
return Err(file_error("FILE_SOURCE_CHANGED", "The remote file changed since the transfer started; restart it from the beginning.").into());2221
}2222
self.update(id, |j| {2223
j.remote_identity = Some(current.clone());2224
j.size = current.size;2225
j.remote_mode = Some(json!(stat.mode & 0o777));2226
});2227
}2228
let handle = match sftp.open_file(&source, "r", None).await {2229
Ok(handle) => handle,2230
Err(error) if error.transfer_code() == "FILE_PERMISSION_DENIED" && self.with_job(id, |j| j.mode == "privileged-allowed").unwrap_or(false) => {2231
self.update(id, |j| j.mode = "privileged".into());2232
source = self.ensure_staged(id, &sftp, session_id).await?;2233
sftp.open_file(&source, "r", None).await.map_err(sftp_core)?2234
}2235
Err(error) => return Err(file_error(error.transfer_code(), &format!("{} cannot be read.", job.remote_path)).into()),2236
};2237
let local_part = {2238
let job = self.snapshot(id)?;2239
job.local_part.clone().unwrap_or_else(|| {2240
let local = Path::new(&job.local_path);2241
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()2242
})2243
};2244
self.update(id, |j| j.local_part = Some(local_part.clone()));2245
let result: RunResult<()> = async {2246
let local = self.open_local_part(id, &local_part)?;2247
let transferred = self.with_job(id, |j| j.transferred).unwrap_or(0);2248
if transferred > 0 {2249
let window = TAIL_CHECK.min(transferred);2250
let position = transferred - window;2251
verify_tail(2252
transferred,2253
async { local.read_at(position, window as usize).await.map_err(io_core) }.boxed(),2254
async { sftp.read_full(&handle, position, window as usize).await.map_err(sftp_core) }.boxed(),2255
window,2256
)2257
.await?;2258
}2259
self.set(id, "transferring", None);2260
let size = self.with_job(id, |j| j.size).unwrap_or(0);2261
let produce: Producer = {2262
let (sftp, handle) = (sftp.clone(), handle.clone());2263
Arc::new(move |pos, len| {2264
let (sftp, handle) = (sftp.clone(), handle.clone());2265
Box::pin(async move {2266
let data = sftp.read_full(&handle, pos, len).await.map_err(sftp_core)?;2267
if data.len() != len {2268
return Err(file_error("FILE_SOURCE_CHANGED", "The remote file shrank during the transfer."));2269
}2270
Ok(data)2271
})2272
})2273
};2274
let consume: Consumer = {2275
let local = local.clone();2276
Arc::new(move |buf, pos| {2277
let local = local.clone();2278
Box::pin(async move { local.write_at(pos, buf).await.map_err(io_core) })2279
})2280
};2281
self.run_pump(id, size, produce, consume, stop.clone()).await?;2282
local.finish(size).await.map_err(io_core)?;2283
drop(local);2284
let _ = sftp.close_file(&handle).await;2285
self.set(id, "verifying", None);2286
let hash = local_hash(Path::new(&local_part), stop.clone()).await?;2287
let job = self.snapshot(id)?;2288
let mut verification = json!({ "method": "size", "local": hash });2289
if job.mode == "privileged" {2290
verification = json!({ "method": "sha256", "local": hash, "remote": job.staged.as_ref().map(|s| s["hash"].clone()).unwrap_or(Value::Null) });2291
} else {2292
let stat = sftp.stat(&job.remote_path).await.map_err(sftp_core)?;2293
let identity = job.remote_identity.clone().unwrap_or_default();2294
if stat.size != identity.size || stat.mtime as u64 != identity.mtime {2295
return Err(file_error("FILE_SOURCE_CHANGED", "The remote file changed during the transfer; restart it from the beginning.").into());2296
}2297
if self.inner.lock().settings.verify {2298
let remote = match self.files.helper(session_id, json!({ "operation": "hash", "path": job.remote_path }), &Authorization::default().with_timeout(hash_timeout(job.size))).await {2299
Ok(value) => value["hash"].as_str().map(str::to_string),2300
Err(_) => remote_checksum(&self.sessions, session_id, &job.remote_path).await.ok(),2301
};2302
if let Some(remote) = remote {2303
verification = json!({ "method": "sha256", "local": hash, "remote": remote });2304
}2305
}2306
}2307
self.update(id, |j| {2308
j.hash = Some(hash.clone());2309
j.verification = Some(verification.clone());2310
});2311
if verification["remote"].as_str().is_some_and(|r| r != hash) {2312
let _ = self.discard_partial(id).await;2313
self.update(id, |j| j.transferred = 0);2314
return Err(file_error("FILE_VERIFY_FAILED", "The downloaded data does not match the remote file.").into());2315
}2316
Ok(())2317
}2318
.await;2319
if result.is_err() {2320
let _ = tokio::time::timeout(Duration::from_secs(3), sftp.close_file(&handle)).await;2321
}2322
drop(sftp);2323
result?;2324
let kind = stop.borrow().clone();2325
if !kind.is_empty() {2326
return Err(RunError::Stop(kind));2327
}2328
self.set(id, "finalizing", None);2329
let job = self.snapshot(id)?;2330
if std::fs::symlink_metadata(&job.local_path).is_ok() && !job.overwrite {2331
return Err(file_error("FILE_EXISTS", &format!("{} already exists.", job.local_path)).into());2332
}2333
#[cfg(unix)]2334
{2335
use std::os::unix::fs::PermissionsExt;2336
let remote_mode = job.remote_mode.as_ref().and_then(|m| number(m)).map(|m| m as u32).unwrap_or(0o644);2337
let mode = if job.mode == "privileged" { 0o600 } else { (remote_mode & 0o755) | 0o600 };2338
let _ = std::fs::set_permissions(&local_part, std::fs::Permissions::from_mode(mode));2339
}2340
std::fs::rename(&local_part, &job.local_path).map_err(io_core)?;2341
if job.mode == "privileged" && self.files.helper(session_id, json!({ "operation": "transfer-remove", "id": job.id }), &Authorization::default()).await.is_err() {2342
self.update(id, |j| j.cleanup_pending = true);2343
}2344
self.update(id, |j| j.finalize = json!({ "state": "done", "finishedAt": now() }));2345
self.set(id, "completed", None);2346
self.logger.info("Transfer completed.", json!({ "jobId": id, "direction": "download", "bytes": job.size, "mode": job.mode }));2347
Ok(())2348
}2350
async fn ensure_staged(self: &Arc<Self>, id: &str, sftp: &Sftp, session_id: &str) -> RunResult<String> {2351
let job = self.snapshot(id)?;2352
if let Some(staged) = &job.staged {2353
let path = staged["path"].as_str().unwrap_or("");2354
if sftp.stat(path).await.is_ok_and(|s| Some(s.size as f64) == number(&staged["size"])) {2355
return Ok(path.to_string());2356
}2357
self.update(id, |j| j.transferred = 0);2358
}2359
self.set(id, "connecting", None);2360
let workspace = self.files.helper(session_id, json!({ "operation": "transfer-workspace" }), &Authorization::default()).await?;2361
let auth = self.helper_auth(id, true, Some(hash_timeout(job.size)));2362
let path = if job.direction == "relay" { &job.local_path } else { &job.remote_path };2363
let auth = if job.direction == "relay" { self.source_auth(&job) } else { auth };2364
if job.direction == "relay" {2365
self.update(id, |j| j.staged = Some(json!({})));2366
}2367
let staged = self.files.helper(session_id, json!({ "operation": "stage-out", "path": path, "id": source_stage_id(&job), "uid": workspace["uid"],2368
"noFollow": job.direction == "relay", "expected": if job.direction == "relay" { serde_json::to_value(&job.remote_identity).unwrap_or(Value::Null) } else { Value::Null } }), &auth).await.map_err(|e| if job.direction == "relay" { e.with("sourceSudo", true) } else { e })?;2369
let path = staged["path"].as_str().unwrap_or("").to_string();2370
self.update(id, |j| {2371
j.staged = Some(json!({ "path": staged["path"], "size": staged["size"], "hash": staged["hash"], "mtimeNs": staged["mtimeNs"] }));2372
j.size = number(&staged["size"]).unwrap_or(0.0) as u64;2373
if job.direction != "relay" { j.transferred = 0; }2374
});2375
self.persist(true, false);2376
Ok(path)2377
}2379
async fn cleanup_staged(&self, id: &str) -> CoreResult<()> {2380
let Some(job) = self.with_job(id, |j| j.clone()).filter(|j| j.staged.is_some()) else { return Ok(()) };2381
let session = if job.direction == "relay" {2382
self.relay_session(id).map_err(|_| file_error("SESSION_NOT_CONNECTED", "Staged data is removed after reconnecting."))?2383
} else {2384
self.find_session(id).ok_or_else(|| file_error("SESSION_NOT_CONNECTED", "Staged data is removed after reconnecting."))?2385
};2386
self.files.helper(&session, json!({ "operation": "transfer-remove", "id": source_stage_id(&job) }), &Authorization::default()).await?;2387
self.update(id, |j| j.staged = None);2388
Ok(())2389
}2391
fn open_local_part(&self, id: &str, part: &str) -> RunResult<LocalFile> {2392
let transferred = self.with_job(id, |j| j.transferred).unwrap_or(0);2393
if transferred > 0 {2394
match std::fs::symlink_metadata(part) {2395
Ok(meta) if meta.is_file() => {2396
self.update(id, |j| j.transferred = j.transferred.min(meta.len()));2397
return LocalFile::open(Path::new(part), "r+").map_err(|e| io_core(e).into());2398
}2399
_ => self.update(id, |j| j.transferred = 0),2400
}2401
}2402
let _ = std::fs::remove_file(part);2403
LocalFile::open(Path::new(part), "wx").map_err(|e| io_core(e).into())2404
}2406
// ---- cancel / cleanup ------------------------------------------------------------------------2408
async fn finish_cancel(self: &Arc<Self>, id: &str) {2409
self.forget_secrets(id);2410
let cleaned = self.discard_partial(id).await.is_ok();2411
self.update(id, |j| j.cleanup_pending = !cleaned);2412
self.set(id, "cancelled", None);2413
}2415
/// Removes only temporary data this job created; existing destination files are never touched.2416
async fn discard_partial(&self, id: &str) -> CoreResult<()> {2417
let Some(job) = self.with_job(id, |j| j.clone()) else { return Ok(()) };2418
self.cleanup_staged(id).await?;2419
if job.direction == "download" {2420
if let Some(part) = &job.local_part {2421
match std::fs::remove_file(part) {2422
Ok(()) => {}2423
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}2424
Err(e) => return Err(io_core(e)),2425
}2426
}2427
return Ok(());2428
}2429
let Some(part) = job.part_path.clone() else { return Ok(()) };2430
let session = self.find_session(id).ok_or_else(|| file_error("SESSION_NOT_CONNECTED", "Partial data is removed after reconnecting."))?;2431
if job.mode == "privileged" {2432
self.files.helper(&session, json!({ "operation": "transfer-remove", "id": job.id }), &Authorization::default()).await?;2433
} else {2434
let sftp = self.files.sftp(&session).await?;2435
if let Ok(stat) = sftp.lstat(&part).await {2436
if !stat.is_dir() {2437
sftp.unlink(&part).await.map_err(sftp_core)?;2438
}2439
}2440
}2441
self.update(id, |j| j.part_path = None);2442
Ok(())2443
}2445
fn cleanup_orphans_later(self: &Arc<Self>) {2446
if self.cleanup_scheduled.swap(true, Ordering::SeqCst) {2447
return;2448
}2449
let manager = self.clone();2450
tokio::spawn(async move {2451
tokio::task::yield_now().await;2452
manager.cleanup_scheduled.store(false, Ordering::SeqCst);2453
manager.cleanup_orphans().await;2454
});2455
}2457
/// After a restart: delete temporary transfer data that no resumable job owns any more.2458
async fn cleanup_orphans(self: &Arc<Self>) {2459
for session in self.sessions.list() {2460
if session["state"] != "connected" {2461
continue;2462
}2463
let session_id = session["sessionId"].as_str().unwrap_or("").to_string();2464
let identity = SessionIdentity::of_status(&session).key();2465
if !self.inner.lock().cleaned_identities.insert(identity.clone()) {2466
continue;2467
}2468
let pending: Vec<String> = self.inner.lock().jobs.values().filter(|j| (j.identity.key() == identity || relay_source(j).is_some_and(|(_, source)| source.key() == identity)) && j.cleanup_pending).map(|j| j.id.clone()).collect();2469
for id in pending {2470
if self.discard_partial(&id).await.is_ok() {2471
self.update(&id, |j| j.cleanup_pending = false);2472
self.mark_job(&id);2473
}2474
}2475
let keep: HashSet<String> = self.inner.lock().jobs.values().filter(|j| (j.identity.key() == identity || relay_source(j).is_some_and(|(_, source)| source.key() == identity)) && !FINAL.contains(&j.state.as_str())).flat_map(|j| [j.id.clone(), source_stage_id(j)]).collect();2476
let Ok(listed) = self.files.helper(&session_id, json!({ "operation": "transfer-list" }), &Authorization::default()).await else { continue };2477
for item in listed["items"].as_array().cloned().unwrap_or_default() {2478
let item_id = item["id"].as_str().unwrap_or("").to_string();2479
let known = self.inner.lock().jobs.values().find(|j| j.id == item_id || source_stage_id(j) == item_id).map(|j| FINAL.contains(&j.state.as_str()));2480
let abandoned = match known {2481
Some(final_state) => final_state,2482
None => now() - (number(&item["mtime"]).unwrap_or(0.0) * 1000.0) as i64 > ORPHAN_AGE_MS,2483
};2484
if keep.contains(&item_id) || !abandoned {2485
continue;2486
}2487
let _ = self.files.helper(&session_id, json!({ "operation": "transfer-remove", "id": item_id }), &Authorization::default()).await;2488
self.logger.info("Abandoned transfer data removed.", json!({ "sessionId": session_id, "id": item_id, "bytes": item["size"] }));2489
}2490
self.persist(false, false);2491
}2492
}2494
// ---- state, events, persistence ----------------------------------------------------------------2496
fn authorize(&self, id: &str, options: &Value) {2497
let mut inner = self.inner.lock();2498
let password = options["sudoPassword"].as_str().filter(|s| !s.is_empty()).map(str::to_string);2499
let source = inner.jobs.get(id).is_some_and(|j| j.direction == "relay" && j.extra.get("sourceAuthRequired") == Some(&Value::Bool(true)));2500
if source {2501
if options["sudo"] == true || password.is_some() {2502
if let Some(job) = inner.jobs.get_mut(id) { job.extra.insert("sourceSudo".into(), json!(true)); }2503
}2504
if let Some(password) = password {2505
inner.secrets.insert(format!("{id}:source"), Zeroizing::new(password));2506
}2507
return;2508
}2509
if let Some(job) = inner.jobs.get_mut(id) {2510
if options["sudo"] == true || password.is_some() {2511
// Data is already next to the target: only the metadata-preserving replacement needs root.2512
if job.direction != "download" && job.mode == "direct" && job.transferred > 0 && job.transferred == job.size {2513
job.finalize_with_sudo = true;2514
}2515
if job.direction == "download" && job.mode == "direct" {2516
job.mode = "privileged-allowed".into();2517
}2518
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())) {2519
job.mode = "privileged".into();2520
job.part_path = None;2521
job.target_baseline = None;2522
}2523
}2524
}2525
if let Some(password) = password {2526
inner.secrets.insert(id.to_string(), Zeroizing::new(password));2527
}2528
}2530
fn set(self: &Arc<Self>, id: &str, state: &str, error: Option<JobError>) {2531
if !STATES.contains(&state) {2532
return;2533
}2534
{2535
let mut inner = self.inner.lock();2536
let t = now();2537
let Inner { jobs, runtime, .. } = &mut *inner;2538
let Some(job) = jobs.get_mut(id) else { return };2539
let rt = runtime.get_mut(id);2540
if job.state == "transferring" {2541
if let Some(rt) = rt.as_deref() {2542
if let Some(since) = rt.active_since {2543
job.active_ms += (t - since).max(0) as u64;2544
}2545
}2546
}2547
if let Some(rt) = rt {2548
if job.state == "transferring" {2549
rt.active_since = None;2550
}2551
if state == "transferring" {2552
rt.active_since = Some(t);2553
rt.last_time = t;2554
rt.last_bytes = job.transferred;2555
}2556
}2557
job.state = state.into();2558
job.error = error;2559
job.updated_at = t;2560
if state == "transferring" && job.started_at.is_none() {2561
job.started_at = Some(t);2562
}2563
if FINAL.contains(&state) {2564
job.finished_at = Some(t);2565
}2566
let owner = job.owner.clone();2567
Self::mark(&mut inner, id, &owner);2568
}2569
self.emit_soon();2570
self.persist(!RUNNING.contains(&state), false);2571
}2573
fn progress(self: &Arc<Self>, id: &str, confirmed: u64) {2574
{2575
let mut inner = self.inner.lock();2576
let t = now();2577
let Inner { jobs, runtime, .. } = &mut *inner;2578
let Some(job) = jobs.get_mut(id) else { return };2579
let moved = confirmed.saturating_sub(job.transferred);2580
job.transferred = confirmed;2581
job.moved_bytes += moved;2582
if let Some(rt) = runtime.get_mut(id) {2583
rt.progress_since_resume += moved;2584
if rt.progress_since_resume > 1024 * 1024 {2585
job.attempts = 0;2586
}2587
let elapsed = t - rt.last_time;2588
if elapsed >= 500 {2589
let instant = (job.transferred.saturating_sub(rt.last_bytes)) as f64 * 1000.0 / elapsed as f64;2590
rt.speed = if rt.samples > 0 { 0.3 * instant + 0.7 * rt.speed } else { instant };2591
rt.samples += 1;2592
rt.last_time = t;2593
rt.last_bytes = job.transferred;2594
}2595
}2596
job.updated_at = t;2597
let owner = job.owner.clone();2598
Self::mark(&mut inner, id, &owner);2599
}2600
self.emit_soon();2601
self.persist(false, false);2602
}2604
fn public(inner: &Inner, job: &Job) -> Value {2605
let rt = inner.runtime.get(&job.id);2606
let t = now();2607
let mut speed = 0.0;2608
let mut eta = Value::Null;2609
if job.state == "transferring" {2610
if let Some(rt) = rt {2611
let idle = t - rt.last_time;2612
speed = if idle > 2000 { rt.speed * 2000.0 / idle as f64 } else { rt.speed };2613
if rt.samples >= 3 && speed > 1.0 {2614
eta = json!((job.size.saturating_sub(job.transferred) as f64 / speed).ceil() as u64);2615
}2616
}2617
}2618
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 };2619
let state = job.state.as_str();2620
let overwrite_error = job.error.as_ref().is_some_and(|e| e.code == "FILE_EXISTS" || e.code == "FILE_CONFLICT");2621
json!({2622
"id": job.id,2623
"batchId": job.batch_id,2624
"sessionId": job.session_id,2625
"identity": job.identity,2626
"direction": job.direction,2627
"name": job.name,2628
"state": state,2629
"source": if job.direction != "download" { &job.local_path } else { &job.remote_path },2630
"target": if job.direction != "download" { &job.remote_path } else { &job.local_path },2631
"resolvedTarget": job.finalize.get("resolvedPath").filter(|v| v.is_string()).cloned().or_else(|| job.resolved_path.clone().map(Value::from)).unwrap_or(Value::Null),2632
"mode": if job.mode == "privileged" { "privileged" } else { "direct" },2633
"size": job.size,2634
"transferred": job.transferred,2635
"percent": percent(job.transferred, job.size),2636
"speed": speed.round() as u64,2637
"averageSpeed": if active_ms > 1000 { json!((job.moved_bytes as f64 * 1000.0 / active_ms as f64).round() as u64) } else { Value::Null },2638
"eta": eta,2639
"finalize": { "state": job.finalize_state(), "metadataPreserved": job.finalize["metadataPreserved"] == true, "created": job.finalize["created"] == true },2640
"verification": job.verification.as_ref().map(|v| json!({ "method": v["method"], "matched": v["remote"].is_string() && v["remote"] == v["local"] })),2641
"error": job.error.as_ref().map(|e| json!({ "code": e.code, "message": e.message })),2642
"authRequired": job.auth_required || move_recovery_pending(job),2643
"sourceAuthRequired": job.extra.get("sourceAuthRequired") == Some(&Value::Bool(true)) || (move_recovery_pending(job) && job.direction == "relay"),2644
"attempts": job.attempts,2645
"nextRetryAt": job.next_retry_at,2646
"cleanupPending": job.cleanup_pending,2647
"sourceIdentity": job.extra.get("relaySource").map(|r| r["identity"].clone()).unwrap_or(Value::Null),2648
"sourceSessionId": job.extra.get("relaySource").map(|r| r["sessionId"].clone()).unwrap_or(Value::Null),2649
"move": job.extra.get("moveSource") == Some(&Value::Bool(true)),2650
"moveResult": job.extra.get("moveResult").cloned().unwrap_or(Value::Null),2651
"actions": {2652
"pause": (["waiting", "reconnecting", "interrupted", "connecting", "transferring", "verifying"].contains(&state)),2653
"resume": (["paused", "interrupted", "reconnecting"].contains(&state) || (state == "failed" && job.resumable) || (move_recovery_pending(job) && !rt.is_some_and(|r| r.running))),2654
"retry": (["failed", "interrupted", "paused", "cancelled"].contains(&state)),2655
"cancel": (!FINAL.contains(&state) && state != "finalizing"),2656
"overwrite": (overwrite_error && ["failed", "interrupted"].contains(&state)),2657
"remove": (FINAL.contains(&state) && !job.cleanup_pending)2658
},2659
"createdAt": job.created_at,2660
"updatedAt": job.updated_at,2661
"finishedAt": job.finished_at2662
})2663
}2665
fn mark(inner: &mut Inner, id: &str, owner: &str) {2666
inner.changed.insert(id.to_string());2667
inner.pending_owners.insert(owner.to_string());2668
}2670
fn mark_job(self: &Arc<Self>, id: &str) {2671
{2672
let mut inner = self.inner.lock();2673
let Some(owner) = inner.jobs.get(id).map(|j| j.owner.clone()) else { return };2674
Self::mark(&mut inner, id, &owner);2675
}2676
self.emit_soon();2677
}2679
fn emit_soon(self: &Arc<Self>) {2680
{2681
let mut inner = self.inner.lock();2682
if inner.emit_scheduled {2683
return;2684
}2685
inner.emit_scheduled = true;2686
}2687
let manager = self.clone();2688
tokio::spawn(async move {2689
tokio::time::sleep(Duration::from_millis(EMIT_MS)).await;2690
let updates: Vec<(String, Value)> = {2691
let mut inner = manager.inner.lock();2692
inner.emit_scheduled = false;2693
let owners: Vec<String> = inner.pending_owners.drain().collect();2694
let full: HashSet<String> = inner.full_owners.drain().collect();2695
let changed: HashSet<String> = inner.changed.drain().collect();2696
owners2697
.into_iter()2698
.map(|owner| {2699
let all: Vec<Value> = Self::owned(&inner, &owner).iter().map(|j| Self::public(&inner, j)).collect();2700
let is_full = full.contains(&owner);2701
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() };2702
let removed: Vec<&String> = changed.iter().filter(|id| !inner.jobs.contains_key(*id)).collect();2703
let update = json!({ "owner": owner, "jobs": jobs, "removed": removed, "full": is_full, "summary": summarize(&all), "settings": inner.settings });2704
(owner, update)2705
})2706
.collect()2707
};2708
for (owner, update) in updates {2709
(manager.emitter)(&owner, update);2710
}2711
});2712
}2714
fn owned<'a>(inner: &'a Inner, owner: &str) -> Vec<&'a Job> {2715
let mut jobs: Vec<&Job> = inner.jobs.values().filter(|j| j.owner == owner).collect();2716
jobs.sort_by_key(|j| j.order);2717
jobs2718
}2720
fn select_ids(inner: &Inner, owner: &str, ids: &Value) -> Vec<String> {2721
let owned = Self::owned(inner, owner);2722
if *ids == "all" {2723
return owned.iter().map(|j| j.id.clone()).collect();2724
}2725
let wanted: HashSet<String> = match ids {2726
Value::Array(list) => list.iter().take(20000).map(|v| v.as_str().map(str::to_string).unwrap_or_else(|| v.to_string())).collect(),2727
Value::Null => HashSet::new(),2728
other => [other.as_str().map(str::to_string).unwrap_or_else(|| other.to_string())].into_iter().collect(),2729
};2730
owned.iter().filter(|j| wanted.contains(&j.id)).map(|j| j.id.clone()).collect()2731
}2733
fn select(&self, owner: &str, ids: &Value) -> Vec<String> {2734
Self::select_ids(&self.inner.lock(), owner, ids)2735
}2737
/// Moves a server-to-server job to a new connection to its source server, on request only and2738
/// only when the old connection is gone.2739
fn bind_relay_source(&self, id: &str, options: &Value) -> bool {2740
let Some(requested) = options["sourceSessionId"].as_str() else { return true };2741
let Ok(identity) = self.identity(requested) else { return false };2742
let Some(Some((previous, expected))) = self.with_job(id, relay_source) else { return false };2743
if identity != expected || (previous != requested && self.identity(&previous).is_ok()) {2744
return false;2745
}2746
if let Some(relay) = self.inner.lock().jobs.get_mut(id).and_then(|job| job.extra.get_mut("relaySource")) {2747
relay["sessionId"] = json!(requested);2748
}2749
true2750
}2752
fn bind_session(&self, id: &str, options: &Value) -> bool {2753
if !self.bind_relay_source(id, options) {2754
return false;2755
}2756
let Some(requested) = options["sessionId"].as_str() else { return true };2757
let Ok(identity) = self.identity(requested) else { return false };2758
let Some((previous, expected)) = self.with_job(id, |j| (j.session(), j.identity.clone())) else { return false };2759
if identity != expected || (previous != requested && self.identity(&previous).is_ok()) {2760
return false;2761
}2762
if let Some(job) = self.inner.lock().jobs.get_mut(id) {2763
job.session_id = json!(requested);2764
}2765
true2766
}2768
fn find_session(&self, id: &str) -> Option<String> {2769
let (identity, preferred) = self.with_job(id, |j| (j.identity.clone(), j.session()))?;2770
matching_session(&self.sessions.list(), &identity, &preferred)2771
}2773
fn session_for(&self, id: &str) -> RunResult<String> {2774
self.find_session(id).ok_or_else(|| {2775
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();2776
RunError::Error(file_error("SESSION_NOT_CONNECTED", &format!("Connect to {user}@{host} to continue.")))2777
})2778
}2780
fn trim_history(inner: &mut Inner) {2781
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();2782
if finished.len() <= MAX_HISTORY {2783
return;2784
}2785
finished.sort();2786
let excess = finished.len() - MAX_HISTORY;2787
for (_, id) in finished.into_iter().take(excess) {2788
inner.jobs.remove(&id);2789
inner.changed.insert(id);2790
}2791
}2793
fn with_job<T>(&self, id: &str, f: impl FnOnce(&Job) -> T) -> Option<T> {2794
self.inner.lock().jobs.get(id).map(f)2795
}2797
fn snapshot(&self, id: &str) -> RunResult<Job> {2798
self.with_job(id, |j| j.clone()).ok_or_else(|| RunError::Stop("cancel".into()))2799
}2801
fn update(&self, id: &str, f: impl FnOnce(&mut Job)) {2802
if let Some(job) = self.inner.lock().jobs.get_mut(id) {2803
f(job);2804
}2805
}2807
fn job_state(&self, id: &str) -> String {2808
self.with_job(id, |j| j.state.clone()).unwrap_or_default()2809
}2811
fn snapshot_file(&self) -> Option<(PathBuf, Value)> {2812
let file = self.file.clone()?;2813
let mut inner = self.inner.lock();2814
if inner.closed {2815
return None;2816
}2817
Self::trim_history(&mut inner);2818
// Persisted state never contains passwords, keys or runtime handles.2819
let mut jobs: Vec<&Job> = inner.jobs.values().collect();2820
jobs.sort_by_key(|j| j.order);2821
Some((file, json!({ "version": 1, "settings": inner.settings, "order": inner.order, "jobs": jobs })))2822
}2824
fn persist(self: &Arc<Self>, soon: bool, sync: bool) {2825
if self.file.is_none() {2826
return;2827
}2828
let write = |manager: &TransferManager| {2829
if let Some((file, data)) = manager.snapshot_file() {2830
if let Err(error) = crate::core::storage::atomic_write_json(&file, &data) {2831
manager.logger.warn("Transfer state could not be saved.", json!({ "message": error.message }));2832
}2833
}2834
};2835
if sync {2836
write(self);2837
return;2838
}2839
if !soon && self.persist_pending.swap(true, Ordering::SeqCst) {2840
return;2841
}2842
if soon {2843
self.persist_pending.store(true, Ordering::SeqCst);2844
}2845
let manager = self.clone();2846
tokio::spawn(async move {2847
tokio::time::sleep(Duration::from_millis(if soon { 50 } else { PERSIST_MS })).await;2848
if !manager.persist_pending.swap(false, Ordering::SeqCst) {2849
return;2850
}2851
let _ = tokio::task::spawn_blocking(move || write(&manager)).await;2852
});2853
}2855
fn load(&self) {2856
let Some(file) = &self.file else { return };2857
let Ok(text) = std::fs::read_to_string(file) else { return };2858
let Ok(data) = serde_json::from_str::<Value>(&text) else { return };2859
if data["version"] != 1 || !data["jobs"].is_array() {2860
return;2861
}2862
let mut inner = self.inner.lock();2863
if let Some(settings) = data["settings"].as_object() {2864
if let Some(v) = settings.get("maxConcurrent").and_then(number) {2865
inner.settings.max_concurrent = if v.fract() == 0.0 && (1.0..=8.0).contains(&v) { v as u32 } else { Settings::default().max_concurrent };2866
}2867
if let Some(v) = settings.get("autoRetry").and_then(Value::as_bool) {2868
inner.settings.auto_retry = v;2869
}2870
if let Some(v) = settings.get("verify").and_then(Value::as_bool) {2871
inner.settings.verify = v;2872
}2873
}2874
inner.order = number(&data["order"]).unwrap_or(0.0) as u64;2875
let jobs = data["jobs"].as_array().cloned().unwrap_or_default();2876
let skip = jobs.len().saturating_sub(MAX_JOBS);2877
for raw in jobs.into_iter().skip(skip) {2878
let Ok(mut job) = serde_json::from_value::<Job>(raw) else { continue };2879
if job.id.len() != 32 || !job.id.chars().all(|c| c.is_ascii_digit() || ('a'..='f').contains(&c)) || !STATES.contains(&job.state.as_str()) {2880
continue;2881
}2882
// Nothing restarts on its own after the application was closed: the user is offered to resume.2883
if RUNNING.contains(&job.state.as_str()) || job.state == "waiting" || job.state == "reconnecting" {2884
if job.state == "finalizing" || job.finalize_state() == "running" {2885
if !job.finalize.is_object() {2886
job.finalize = json!({});2887
}2888
job.finalize["state"] = json!("unknown");2889
}2890
job.state = "interrupted".into();2891
job.error = Some(JobError { code: "APP_RESTARTED".into(), message: "The application was closed during the transfer.".into() });2892
}2893
job.next_retry_at = None;2894
job.attempts = 0;2895
inner.order = inner.order.max(job.order);2896
inner.jobs.insert(job.id.clone(), job);2897
}2898
}2899
}2901
struct Regex;2903
impl Regex {2904
/// `String(message).replace(/^[A-Z_]+: /, "")`.2905
fn strip_code(message: &str) -> String {2906
if let Some((prefix, rest)) = message.split_once(": ") {2907
if !prefix.is_empty() && prefix.chars().all(|c| c.is_ascii_uppercase() || c == '_') {2908
return rest.to_string();2909
}2910
}2911
message.to_string()2912
}2913
}2915
#[derive(Debug, Clone)]2916
struct RemoteInfo {2917
kind: &'static str,2918
size: u64,2919
mtime: u64,2920
mode: u32,2921
}2923
fn source_stage_id(job: &Job) -> String {2924
if job.direction == "relay" { hex::encode(Sha256::digest(format!("source:{}", job.id)))[..32].to_string() } else { job.id.clone() }2925
}2927
fn valid_hash(hash: &str) -> bool {2928
hash.len() == 64 && hash.bytes().all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(&c))2929
}2931
fn move_verified(job: &Job) -> bool {2932
job.state == "completed" && job.finalize_state() == "done" && job.verification.as_ref().is_some_and(|v| v["local"].as_str().is_some_and(valid_hash) && v["local"] == v["remote"])2933
}2935
fn relay_source(job: &Job) -> Option<(String, SessionIdentity)> {2936
let relay = job.extra.get("relaySource")?;2937
Some((relay["sessionId"].as_str().unwrap_or("").to_string(), serde_json::from_value(relay["identity"].clone()).ok()?))2938
}2940
fn add_download(plan: &mut Plan, remote: &str, local: &str, info: &RemoteInfo) {2941
plan.bytes += info.size;2942
plan.files.push(PlanFile {2943
name: posix::basename(remote),2944
local_path: local.to_string(),2945
remote_path: remote.to_string(),2946
size: info.size,2947
local_identity: None,2948
remote_identity: Some(RemoteIdentity { size: info.size, mtime: info.mtime }),2949
mode: Some(info.mode & 0o777),2950
conflict: false,2951
overwrite_confirmed: false,2952
});2953
}2955
#[derive(Default)]2956
struct LocalWalk {2957
files: Vec<PlanFile>,2958
directories: Vec<String>,2959
source_directories: Vec<String>,2960
skipped: Vec<Value>,2961
bytes: u64,2962
}2964
impl LocalWalk {2965
fn visit(&mut self, local: &Path, remote: &str, depth: u32) -> CoreResult<()> {2966
if self.files.len() + self.directories.len() > MAX_PLAN_ENTRIES {2967
return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", &format!("A transfer can contain at most {MAX_PLAN_ENTRIES} entries.")));2968
}2969
let meta = std::fs::symlink_metadata(local).map_err(io_core)?;2970
let shown = local.to_string_lossy().to_string();2971
if meta.file_type().is_symlink() {2972
self.skipped.push(json!({ "path": shown, "reason": "symlink" }));2973
return Ok(());2974
}2975
if meta.is_dir() {2976
self.directories.push(remote.to_string());2977
self.source_directories.push(shown);2978
if depth > 64 {2979
return Err(file_error("TRANSFER_TOO_DEEP", "The selected directory tree is too deep."));2980
}2981
for entry in std::fs::read_dir(local).map_err(io_core)? {2982
let entry = entry.map_err(io_core)?;2983
let raw = entry.file_name().to_string_lossy().to_string();2984
let Ok(name) = safe_entry_name(&raw) else {2985
self.skipped.push(json!({ "path": local.join(&raw).to_string_lossy(), "reason": "name" }));2986
continue;2987
};2988
self.visit(&local.join(&name), &posix::join(remote, &name), depth + 1)?;2989
}2990
return Ok(());2991
}2992
if !meta.is_file() {2993
self.skipped.push(json!({ "path": shown, "reason": "type" }));2994
return Ok(());2995
}2996
self.bytes += meta.len();2997
self.files.push(PlanFile {2998
name: local.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default(),2999
local_path: shown,3000
remote_path: remote.to_string(),3001
size: meta.len(),3002
local_identity: Some(LocalIdentity::of(&meta)),3003
remote_identity: None,3004
mode: None,3005
conflict: false,3006
overwrite_confirmed: false,3007
});3008
Ok(())3009
}3010
}3012
/// Writable by the SSH account: direct SFTP; otherwise the privileged staging route.3013
async fn directory_mode(sftp: &Sftp, directory: &str) -> CoreResult<&'static str> {3014
let stat: Stat = sftp.stat(directory).await.map_err(|e| {3015
let code = e.transfer_code();3016
file_error(if code == "FILE_NOT_FOUND" { "FILE_NOT_FOUND" } else { code }, &format!("{directory} is not accessible."))3017
})?;3018
if !stat.is_dir() {3019
return Err(file_error("FILE_NOT_DIRECTORY", &format!("{directory} is not a directory.")));3020
}3021
let probe = posix::join(directory, &format!(".chj-probe-{}", hex_id(6)));3022
match sftp.open_file(&probe, "wx", Some(0o600)).await {3023
Ok(handle) => {3024
let _ = sftp.close_file(&handle).await;3025
let _ = sftp.unlink(&probe).await;3026
Ok("direct")3027
}3028
Err(error) if error.transfer_code() == "FILE_PERMISSION_DENIED" => Ok("privileged"),3029
Err(error) => {3030
let code = error.transfer_code();3031
Err(file_error(if code == "TRANSFER_FAILED" { "FILE_REMOTE_IO_FAILED" } else { code }, &format!("{directory} is not writable.")))3032
}3033
}3034
}3036
async fn remote_conflicts(sftp: &Arc<Sftp>, plan: &mut Plan) {3037
let paths: Vec<(usize, String)> = plan.files.iter().enumerate().map(|(i, f)| (i, f.remote_path.clone())).collect();3038
let checks = futures::stream::iter(paths.into_iter().map(|(i, path)| {3039
let sftp = sftp.clone();3040
async move { (i, sftp.lstat(&path).await.ok()) }3041
}))3042
.buffer_unordered(16)3043
.collect::<Vec<_>>()3044
.await;3045
let mut blocked = HashSet::new();3046
for (i, stat) in checks {3047
let Some(stat) = stat else { continue };3048
if stat.is_dir() {3049
blocked.insert(i);3050
plan.skipped.push(json!({ "path": plan.files[i].remote_path, "reason": "directory-exists" }));3051
} else {3052
plan.files[i].conflict = true;3053
}3054
}3055
let mut index = 0;3056
plan.files.retain(|_| {3057
let keep = !blocked.contains(&index);3058
index += 1;3059
keep3060
});3061
}3063
async fn local_hash(path: &Path, stop: watch::Receiver<String>) -> RunResult<String> {3064
let path = path.to_path_buf();3065
let result = tokio::task::spawn_blocking(move || -> Result<String, RunError> {3066
use std::io::Read;3067
let mut file = std::fs::File::open(&path).map_err(|e| RunError::Error(io_core(e)))?;3068
let mut digest = Sha256::new();3069
let mut buffer = vec![0u8; 1024 * 1024];3070
loop {3071
let kind = stop.borrow().clone();3072
if !kind.is_empty() {3073
return Err(RunError::Stop(kind));3074
}3075
let n = file.read(&mut buffer).map_err(|e| RunError::Error(io_core(e)))?;3076
if n == 0 {3077
break;3078
}3079
digest.update(&buffer[..n]);3080
}3081
Ok(hex::encode(digest.finalize()))3082
})3083
.await3084
.map_err(|_| RunError::Error(CoreError::plain("Local hashing failed.")))?;3085
result3086
}3088
// A privileged plan binds an existing identity even at offset zero. A full retry3089
// deliberately clears it; establish a fresh identity before staging in that case.3090
fn relay_source_changed(job: &Job, current: &RemoteIdentity, sudo: bool) -> bool {3091
job.remote_identity.as_ref() != Some(current)3092
&& (job.transferred > 0 || (sudo && job.remote_identity.is_some()))3093
}3095
fn remote_move_result(response: Result<&Value, &str>) -> Value {3096
match response {3097
Ok(value) if value["removed"] == true && value["unknown"] != true => json!({ "state": "deleted" }),3098
Ok(value) if value["removed"] == false && value["unknown"] != true => {3099
let mut kept = value.clone();3100
kept["state"] = json!("kept");3101
kept3102
}3103
Err("FILE_SOURCE_CHANGED" | "FILE_UNSAFE_TYPE_OR_LINK") => json!({ "state": "kept", "reason": "source-changed" }),3104
_ => json!({ "state": "unknown", "reason": "remove-failed" }),3105
}3106
}3108
fn move_recovery_pending(job: &Job) -> bool {3109
job.state == "completed" && job.direction != "upload" && job.extra.get("moveSource") == Some(&Value::Bool(true))3110
&& job.extra.get("moveResult").is_some_and(|r| r["state"] == "unknown" || r["temporaryPath"].is_string())3111
}3113
#[cfg(test)]3114
mod tests {3115
use super::*;3117
fn manager(root: &Path) -> Arc<TransferManager> {3118
use crate::core::{profiles::ProfileService, ssh::privileged::PrivilegedExecutor, vault::VaultStore};3119
let profiles = Arc::new(ProfileService::new(Arc::new(VaultStore::new(root))));3120
let logger = Arc::new(Logger::new(root));3121
let sessions = SessionManager::new(profiles, logger.clone(), Arc::new(|_, _| {}));3122
let files = Arc::new(RemoteFileService::new(sessions.clone(), Arc::new(PrivilegedExecutor::new(sessions.clone())), logger.clone()));3123
TransferManager::new(sessions, files, Some(root), logger, Arc::new(|_, _| {}))3124
}3126
fn move_job() -> Job {3127
let hash = "a".repeat(64);3128
serde_json::from_value(json!({3129
"id": "0123456789abcdef0123456789abcdef", "owner": "chj.file-manager", "direction": "relay",3130
"state": "completed", "mode": "direct", "sourceSudo": true, "finalize": { "state": "done" },3131
"verification": { "local": hash, "remote": hash }, "size": 10,3132
"remoteIdentity": { "size": 10, "mtime": 20 }3133
})).unwrap()3134
}3136
#[test]3137
fn remote_move_unknown_results_are_never_reported_kept() {3138
for code in ["FILE_RESULT_UNKNOWN", "SESSION_NOT_CONNECTED", "FILE_REMOTE_IO_FAILED", "SUDO_PASSWORD_REQUIRED"] {3139
assert_eq!(remote_move_result(Err(code))["state"], "unknown");3140
}3141
for value in [Value::Null, json!({}), json!({ "removed": false, "unknown": true })] {3142
assert_eq!(remote_move_result(Ok(&value))["state"], "unknown");3143
}3144
assert_eq!(remote_move_result(Ok(&json!({ "removed": true })))["state"], "deleted");3145
assert_eq!(remote_move_result(Err("FILE_SOURCE_CHANGED"))["state"], "kept");3146
let retained = remote_move_result(Ok(&json!({ "removed": false, "reason": "source-changed", "temporaryPath": "/source/.chj-move-test", "restorationFailed": true })));3147
assert_eq!(retained["state"], "kept");3148
assert_eq!(retained["temporaryPath"], "/source/.chj-move-test");3149
assert_eq!(retained["restorationFailed"], true);3150
}3152
#[tokio::test]3153
async fn move_claim_recovery_survives_restart_and_resume_does_not_restart_the_transfer() {3154
let dir = tempfile::tempdir().unwrap();3155
let m = manager(dir.path());3156
let mut j = move_job();3157
j.extra.insert("moveSource".into(), json!(true));3158
j.extra.insert("moveResult".into(), remote_move_result(Err("FILE_RESULT_UNKNOWN")));3159
m.inner.lock().jobs.insert(j.id.clone(), j.clone());3160
m.persist(true, true);3161
let restarted = manager(dir.path());3162
let loaded = restarted.snapshot(&j.id).ok().unwrap();3163
assert!(move_recovery_pending(&loaded));3164
assert_eq!(loaded.state, "completed");3165
let view = TransferManager::public(&restarted.inner.lock(), &loaded);3166
assert_eq!(view["actions"]["resume"], true);3167
assert_eq!(view["authRequired"], true);3168
assert_eq!(view["sourceAuthRequired"], true);3169
assert_eq!(restarted.resume(&loaded.owner, &json!([loaded.id]), &json!({ "sudo": true, "sudoPassword": "source-only" }))["resumed"], 1);3170
assert_eq!(restarted.resume(&loaded.owner, &json!([loaded.id]), &json!({}))["resumed"], 0);3171
assert_eq!(restarted.source_auth(&loaded).sudo_password.as_deref(), Some("source-only"));3172
tokio::task::yield_now().await; // No session: recovery stays unknown, without a network call.3173
let unchanged = restarted.snapshot(&j.id).ok().unwrap();3174
assert_eq!(unchanged.state, "completed");3175
assert_eq!(unchanged.transferred, loaded.transferred);3176
assert_eq!(unchanged.verification, loaded.verification);3177
assert_eq!(unchanged.extra["moveResult"]["state"], "unknown");3178
assert!(unchanged.error.is_some()); // Failed recovery is visible and can be reauthorized.3179
assert!(restarted.inner.lock().secrets.is_empty());3180
let mut retained = unchanged;3181
retained.extra.insert("moveResult".into(), json!({ "state": "kept", "temporaryPath": "/source/.chj-move-test" }));3182
assert!(move_recovery_pending(&retained));3183
retained.extra.insert("moveResult".into(), json!({ "state": "deleted" }));3184
assert!(!move_recovery_pending(&retained));3185
}3187
#[test]3188
fn panel_download_planning_keeps_sudo_options_and_requires_conflict_confirmation() {3189
let supplied = json!({ "sudo": true, "sudoPassword": "memory-only", "rememberSudo": true, "noSaveDialog": false });3190
let options = copy_download_options(&supplied);3191
assert_eq!(options["sudo"], true);3192
assert_eq!(options["sudoPassword"], "memory-only");3193
assert_eq!(options["rememberSudo"], true);3194
assert_eq!(options["noSaveDialog"], true); // No native save dialog has confirmed an overwrite.3195
assert_eq!(supplied["noSaveDialog"], false);3196
for absent in [Value::Null, json!([]), json!(false)] {3197
assert_eq!(copy_download_options(&absent), json!({ "noSaveDialog": true }));3198
}3199
}3201
#[test]3202
fn source_planning_authorization_does_not_elevate_the_destination() {3203
let dir = tempfile::tempdir().unwrap();3204
let m = manager(dir.path());3205
let plan = Plan {3206
id: "plan".into(), owner: "chj.file-manager".into(), direction: "relay", session_id: "destination".into(),3207
identity: SessionIdentity::default(), directory: "/out".into(), local_directory: String::new(),3208
files: Vec::new(), directories: Vec::new(), skipped: Vec::new(), bytes: 0, created_at: now(),3209
mode: "direct".into(), privilege: None, relay: Some(("source".into(), SessionIdentity::default())),3210
source_directories: Vec::new(), source_sudo: true,3211
};3212
let view = m.store_plan(plan).unwrap();3213
assert_eq!(view["mode"], "direct");3214
assert_eq!(view["requiresSudo"], false);3215
assert_eq!(view["sudoPasswordRequired"], false);3216
assert!(m.inner.lock().plans["plan"].source_sudo);3217
let j = move_job();3218
assert!(m.source_auth(&j).sudo);3219
assert!(!m.helper_auth(&j.id, false, None).sudo);3220
assert_ne!(source_stage_id(&j), j.id);3221
assert_eq!(source_stage_id(&j), source_stage_id(&j.clone()));3222
assert_eq!(source_stage_id(&j).len(), 32);3223
let mut download = j;3224
download.direction = "download".into();3225
assert_eq!(source_stage_id(&download), download.id);3226
}3228
#[test]3229
fn retry_authorization_only_updates_the_endpoint_that_failed() {3230
let dir = tempfile::tempdir().unwrap();3231
let m = manager(dir.path());3232
let mut j = move_job();3233
j.extra.insert("sourceAuthRequired".into(), json!(true));3234
j.extra.insert("relaySource".into(), json!({ "sessionId": "source", "identity": SessionIdentity::default() }));3235
m.inner.lock().jobs.insert(j.id.clone(), j.clone());3236
m.inner.lock().secrets.insert(j.id.clone(), Zeroizing::new("destination-secret".into()));3237
m.authorize(&j.id, &json!({ "sudo": true, "sudoPassword": "source-secret" }));3238
let j = m.snapshot(&j.id).ok().unwrap();3239
assert_eq!(j.mode, "direct");3240
assert!(m.source_auth(&j).sudo);3241
assert_eq!(m.source_auth(&j).sudo_password.as_deref(), Some("source-secret"));3242
assert_eq!(m.inner.lock().secrets[&j.id].as_str(), "destination-secret");3243
assert_eq!(TransferManager::public(&m.inner.lock(), &j)["sourceSessionId"], "source");3244
assert_eq!(TransferManager::public(&m.inner.lock(), &j)["sourceAuthRequired"], true);3245
m.update(&j.id, |j| { j.extra.insert("sourceAuthRequired".into(), json!(false)); });3246
m.authorize(&j.id, &json!({ "sudo": true, "sudoPassword": "new-destination-secret" }));3247
assert_eq!(m.inner.lock().secrets[&j.id].as_str(), "new-destination-secret");3248
assert_eq!(m.source_auth(&j).sudo_password.as_deref(), Some("source-secret"));3249
}3251
#[test]3252
fn privileged_relay_retry_establishes_identity_but_resume_keeps_the_binding() {3253
let mut job = move_job();3254
let original = RemoteIdentity { size: 10, mtime: 20 };3255
let changed = RemoteIdentity { size: 12, mtime: 21 };3256
assert!(!relay_source_changed(&job, &original, true));3257
assert!(relay_source_changed(&job, &changed, true)); // Even before any bytes are sent.3258
job.transferred = 5;3259
assert!(relay_source_changed(&job, &changed, true));3260
// The full retry path clears progress and remote_identity, unlike resume.3261
job.transferred = 0;3262
job.remote_identity = None;3263
assert!(!relay_source_changed(&job, &original, true));3264
assert!(!relay_source_changed(&job, &changed, true));3265
job.remote_identity = Some(changed.clone());3266
assert!(!relay_source_changed(&job, &changed, true));3267
assert!(relay_source_changed(&job, &original, true));3268
job.transferred = 1;3269
job.remote_identity = None;3270
assert!(relay_source_changed(&job, &original, true));3271
assert!(relay_source_changed(&job, &original, false));3272
}3274
#[test]3275
fn source_password_is_memory_only_and_missing_after_restart() {3276
let dir = tempfile::tempdir().unwrap();3277
let m = manager(dir.path());3278
let j = move_job();3279
m.inner.lock().jobs.insert(j.id.clone(), j.clone());3280
m.inner.lock().secrets.insert(j.id.clone(), Zeroizing::new("destination-secret".into()));3281
m.inner.lock().secrets.insert(format!("{}:source", j.id), Zeroizing::new("source-secret".into()));3282
assert_eq!(m.source_auth(&j).sudo_password.as_deref(), Some("source-secret"));3283
assert!(!m.helper_auth(&j.id, false, None).sudo);3284
let (file, data) = m.snapshot_file().unwrap();3285
let text = serde_json::to_string(&data).unwrap();3286
assert!(!text.contains("source-secret") && !text.contains("destination-secret") && !text.contains("sudoPassword"));3287
std::fs::write(file, text).unwrap();3288
let restarted = manager(dir.path());3289
let loaded = restarted.snapshot(&j.id).ok().unwrap();3290
let auth = restarted.source_auth(&loaded);3291
assert!(auth.sudo); // Missing credentials must still go to sudo, never to a direct read.3292
assert!(auth.sudo_password.is_none());3293
assert!(restarted.inner.lock().secrets.is_empty());3294
m.forget_secrets(&j.id);3295
assert!(m.inner.lock().secrets.is_empty());3296
}3298
#[tokio::test]3299
async fn move_never_removes_an_unverified_or_unknown_hash() {3300
let dir = tempfile::tempdir().unwrap();3301
let m = manager(dir.path());3302
for verification in [Value::Null, json!({ "local": null, "remote": null }),3303
json!({ "local": "", "remote": "" }), json!({ "local": "unknown", "remote": "unknown" }),3304
json!({ "local": "a".repeat(64), "remote": "b".repeat(64) })] {3305
let mut j = move_job();3306
j.verification = Some(verification);3307
assert_eq!(m.remove_source(&j).await, Err("unverified"));3308
}3309
let mut j = move_job();3310
j.finalize["state"] = json!("unknown");3311
assert_eq!(m.remove_source(&j).await, Err("unverified"));3312
j.finalize["state"] = json!("done");3313
j.state = "transferring".into();3314
assert_eq!(m.remove_source(&j).await, Err("unverified"));3315
}3317
#[tokio::test]3318
async fn local_move_rechecks_identity_and_only_prunes_below_move_root() {3319
let dir = tempfile::tempdir().unwrap();3320
let m = manager(dir.path());3321
let root = dir.path().join("root");3322
let parent = root.join("nested");3323
std::fs::create_dir_all(&parent).unwrap();3324
let source = parent.join("a");3325
std::fs::write(&source, b"source").unwrap();3326
let mut j = move_job();3327
j.direction = "upload".into();3328
j.local_path = source.to_string_lossy().into();3329
j.local_identity = Some(LocalIdentity::of(&std::fs::symlink_metadata(&source).unwrap()));3330
j.extra.insert("moveRoot".into(), json!(root));3331
std::fs::write(&source, b"changed-size").unwrap();3332
assert_eq!(m.remove_source(&j).await, Err("source-changed"));3333
assert!(source.exists());3334
j.local_identity = Some(LocalIdentity::of(&std::fs::symlink_metadata(&source).unwrap()));3335
assert_eq!(m.remove_source(&j).await, Ok(()));3336
assert!(!source.exists() && !parent.exists() && root.exists());3337
#[cfg(unix)]3338
{3339
let link = root.join("link");3340
let target = root.join("target");3341
std::fs::write(&target, b"keep").unwrap();3342
std::os::unix::fs::symlink(&target, &link).unwrap();3343
j.local_path = link.to_string_lossy().into();3344
j.local_identity = Some(LocalIdentity::of(&std::fs::metadata(&link).unwrap()));3345
assert_eq!(m.remove_source(&j).await, Err("source-changed"));3346
assert_eq!(std::fs::read(target).unwrap(), b"keep");3347
assert!(std::fs::symlink_metadata(link).unwrap().file_type().is_symlink());3348
}3349
}3351
#[test]3352
fn transfers_never_choose_another_connection_to_the_same_server() {3353
let a = json!({ "sessionId": "A", "state": "connected", "profileId": "p", "host": "h", "port": 22, "username": "u" });3354
let mut b = a.clone(); b["sessionId"] = json!("B");3355
let identity = SessionIdentity::of_status(&a);3356
assert_eq!(matching_session(&[a.clone(), b.clone()], &identity, "A"), Some("A".into()));3357
assert_eq!(matching_session(&[b.clone()], &identity, "A"), None);3358
assert_eq!(matching_session(&[a.clone(), b.clone()], &identity, "B"), Some("B".into()));3359
b["host"] = json!("another-host");3360
assert_eq!(matching_session(&[b], &identity, "B"), None);3361
let mut disconnected = a; disconnected["state"] = json!("disconnected");3362
assert_eq!(matching_session(&[disconnected], &identity, "A"), None);3363
}3365
/// A `transfers.json` written by 0.1.0 loads, keeps unknown fields and comes back unchanged3366
/// (except the restart transition 0.1.0 applies on load as well).3367
#[test]3368
fn loads_and_rewrites_0_1_0_state() {3369
let legacy = json!({3370
"version": 1,3371
"settings": { "maxConcurrent": 3, "autoRetry": false, "verify": true },3372
"order": 7,3373
"jobs": [{3374
"id": "0123456789abcdef0123456789abcdef", "owner": "chj.file-manager", "batchId": "b", "direction": "upload",3375
"identity": { "profileId": "p", "host": "h", "port": 22, "username": "u" }, "sessionId": "s1", "order": 7,3376
"name": "a.bin", "localPath": "/tmp/a.bin", "remotePath": "/srv/a.bin", "size": 10, "transferred": 4, "movedBytes": 4, "activeMs": 12,3377
"state": "transferring", "mode": "direct", "overwrite": false,3378
"localIdentity": { "size": 10, "mtimeMs": 1700000000123.0, "ino": 42, "dev": 16777220 },3379
"remoteIdentity": null, "remoteMode": null, "finalize": { "state": "pending" }, "attempts": 2, "totalRetries": 3, "error": null,3380
"createdAt": 1, "updatedAt": 2, "resolvedPath": "/srv/a.bin", "partPath": "/srv/.a.bin.chj-part-x", "targetBaseline": null,3381
"helper": true, "ownerUid": 1000, "futureField": { "kept": true }3382
}]3383
});3384
let dir = tempfile::tempdir().unwrap();3385
std::fs::write(dir.path().join("transfers.json"), serde_json::to_string(&legacy).unwrap()).unwrap();3386
let job: Job = serde_json::from_value(legacy["jobs"][0].clone()).unwrap();3387
assert_eq!(job.target_baseline, Some(Value::Null));3388
assert_eq!(job.local_identity.as_ref().unwrap().mtime_ms, 1700000000123);3389
let back = serde_json::to_value(&job).unwrap();3390
assert_eq!(back["futureField"], json!({ "kept": true }));3391
assert_eq!(back["targetBaseline"], Value::Null);3392
assert_eq!(back["ownerUid"], 1000);3393
let mut fresh = job.clone();3394
fresh.target_baseline = None;3395
assert!(serde_json::to_value(&fresh).unwrap().get("targetBaseline").is_none());3396
assert_eq!(Regex::strip_code("FILE_EXISTS: /x already exists."), "/x already exists.");3397
assert_eq!(part_name("abc", "id1"), ".abc.chj-part-id1");3398
}3400
#[test]3401
fn summary_matches_0_1_0_rules() {3402
let jobs = vec![json!({ "state": "completed", "size": 10, "transferred": 10 }), json!({ "state": "transferring", "size": 10, "transferred": 5, "speed": 5, "eta": 1 })];3403
let s = summarize(&jobs);3404
assert_eq!(s["percent"], 75.0);3405
assert_eq!(s["eta"], 1);3406
let done = summarize(&[json!({ "state": "completed", "size": 0, "transferred": 0 })]);3407
assert_eq!(done["percent"], 100.0);3408
}3409
}3411
#[cfg(test)]3412
#[path = "../../../tests/support/transfer_route_identity.rs"]3413
mod route_identity_tests;3415
#[cfg(test)]3416
#[path = "../../../tests/support/transfer_safety.rs"]3417
mod safety_tests;SHA-256
78f262a2bf8705c03881e959c01ea8c762c09fb9dab14753ba1d25d5e9fdc32aCH-J Proprietary Software License 1.14 · Источник представлен в CH-J Proprietary Software License 1.14. Его наличие не изменяет лицензионных условий и не предоставляет дополнительных прав.
Архив SHA-256
ac8fe790f4ba60fc098f8c985fc2656cfcc6cdc21fb34b30e308a2b54e62258f