Veröffentlichte Quellen

CH-J Tools

Durchsuchen Sie Verzeichnisse und Dateien einer bestimmten Anwendungsausgabe.

Quellen als ZIP herunterladen

CH-J / Server Manager 1704 Dateien

0.1.2-beta Hinzugefügt: 08.11.2026, 02:53 · Erstellt: 07.11.2026, 19:32

service.rs Hinzugefügt: Datum nicht erfasst 92,7 KB · 1778 Zeilen

Code Datei herunterladen

1//! Previews, runs, saved profiles, history and schedules. Every remote command is an operation of the matrix.

3use super::bridge;

4use super::parse::{self, Summary};

5use super::spec::{self, Capabilities, Endpoint, Mode, Place, Spec, Transport};

6use crate::core::error::{CoreError, CoreResult};

7use crate::core::files::local;

8use crate::core::logging::Logger;

9use crate::core::matrix;

10use crate::core::ssh::privileged::{Frame, PrivilegedExecutor, RunOptions, StreamEvent};

11use crate::core::ssh::session::SessionManager;

12use parking_lot::Mutex;

13use serde_json::{json, Value};

14use std::collections::HashMap;

15use std::path::{Path, PathBuf};

16use std::process::Stdio;

17use std::sync::Arc;

18use std::time::{Duration, Instant};

19use tokio::io::AsyncReadExt;

20use tokio::sync::watch;

21use zeroize::Zeroizing;

23const MAX_ITEMS: usize = 5000;

24const MAX_OUTPUT: usize = 64 * 1024;

25const MAX_HISTORY: usize = 300;

26const PLAN_LIFETIME: Duration = Duration::from_secs(3600);

27const IDLE_SESSION: Duration = Duration::from_secs(1800);

28const CANCEL_GRACE: Duration = Duration::from_secs(10);

29const INTERVALS: &[u64] = &[5, 10, 15, 20, 30, 60, 120, 180, 240, 360, 720, 1440];

31/// Kept in memory only, for the time between a preview and its run.

32#[derive(Clone, Default)]

33pub struct Secrets {

34 pub source_sudo: Option<Zeroizing<String>>,

35 pub target_sudo: Option<Zeroizing<String>>,

36 pub direct_password: Option<Zeroizing<String>>,

37}

39impl Secrets {

40 fn read(payload: &Value) -> Self {

        let text = |name: &str| payload[name].as_str().filter(|value| !value.is_empty()).map(|value| Zeroizing::new(value.to_string()));

        Secrets { source_sudo: text("sourceSudo"), target_sudo: text("targetSudo"), direct_password: text("directPassword") }

43 }

44}

46struct Plan {

47 spec: Spec,

48 secrets: Secrets,

49 deletes: u64,

50 empty_source: bool,

51 protected_target: bool,

52 created: Instant,

53 source_value: Value,

54}

56struct Running {

57 cancel: watch::Sender<bool>,

58 remote_pid: Arc<Mutex<Option<(String, String, bool, Option<Zeroizing<String>>)>>>,

59}

61#[derive(Default)]

62struct Inner {

63 plans: HashMap<String, Plan>,

64 jobs: Vec<Value>,

65 running: HashMap<String, Running>,

66 opened_sessions: HashMap<String, Instant>,

67 handoff: Option<Value>,

68 last_app_runs: HashMap<String, Instant>,

69 last_stamp: Option<String>,

70}

72pub struct RsyncService {

73 sessions: Arc<SessionManager>,

74 executor: Arc<PrivilegedExecutor>,

75 logger: Arc<Logger>,

76 root: PathBuf,

77 protected: PathBuf,

78 bridge_executable: PathBuf,

79 inner: Mutex<Inner>,

80}

82/// What one run produced.

83#[derive(Default)]

84struct Collected {

85 summary: Summary,

86 items: Vec<Value>,

87 total_items: u64,

88 stats_lines: Vec<String>,

89 code: Option<i64>,

90 pid: Option<String>,

91 bridge: Option<String>,

92}

94fn job_error(code: &str, message: &str) -> CoreError {

95 CoreError::new(code, message)

96}

98fn new_id(prefix: &str) -> String {

99 format!("{prefix}-{}", uuid::Uuid::new_v4().simple())

100}

101

102fn stamp() -> String {

103 chrono::Utc::now().format("%Y%m%dT%H%M%SZ").to_string()

104}

105

106fn now() -> String {

107 chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true)

108}

109

110/// Never finishes without a deadline, so it can sit in a `select!`.

111async fn wait_until(deadline: Option<tokio::time::Instant>) {

112 match deadline {

113 Some(deadline) => tokio::time::sleep_until(deadline).await,

114 None => std::future::pending().await,

115 }

116}

117

118fn parse_key_values(text: &str) -> HashMap<String, String> {

119

    text.lines().filter_map(|line| line.split_once('=')).map(|(key, value)| (key.trim_start_matches('\u{1e}').to_string(), value.to_string())).collect()

120}

121

122/// A schedule as the two forms a server understands. Only fixed shapes pass.

123pub fn schedule_forms(schedule: &Value) -> CoreResult<(String, String)> {

124 let invalid = || job_error("RSYNC_SCHEDULE_INVALID", "The schedule is not valid.");

125 let clock = |text: &str| -> CoreResult<(u32, u32)> {

126 let (hour, minute) = text.split_once(':').ok_or_else(invalid)?;

127 let (hour, minute): (u32, u32) = (hour.parse().map_err(|_| invalid())?, minute.parse().map_err(|_| invalid())?);

128 if hour > 23 || minute > 59 {

129 return Err(invalid());

130 }

131 Ok((hour, minute))

132 };

133 match schedule["kind"].as_str() {

134 Some("interval") => {

135 let minutes = schedule["minutes"].as_u64().unwrap_or(0);

136 if !INTERVALS.contains(&minutes) {

137 return Err(invalid());

138 }

139

            let cron = if minutes < 60 { format!("*/{minutes} * * * *") } else if minutes == 1440 { "0 0 * * *".to_string() } else { format!("0 */{} * * *", minutes / 60) };

140 Ok((format!("OnBootSec=5min;OnUnitActiveSec={minutes}min"), cron))

141 }

142 Some("daily") => {

143 let (hour, minute) = clock(schedule["time"].as_str().unwrap_or(""))?;

144 Ok((format!("OnCalendar=*-*-* {hour:02}:{minute:02}:00"), format!("{minute} {hour} * * *")))

145 }

146 Some("weekly") => {

147 let (hour, minute) = clock(schedule["time"].as_str().unwrap_or(""))?;

148 let day = schedule["day"].as_u64().filter(|day| *day <= 6).ok_or_else(invalid)?;

149 let name = ["Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"][day as usize];

150 Ok((format!("OnCalendar={name} *-*-* {hour:02}:{minute:02}:00"), format!("{minute} {hour} * * {day}")))

151 }

152 _ => Err(invalid()),

153 }

154}

155

156impl RsyncService {

157

    pub fn new(sessions: Arc<SessionManager>, executor: Arc<PrivilegedExecutor>, logger: Arc<Logger>, root: &Path, bridge_executable: PathBuf) -> Arc<Self> {

158

        Arc::new(RsyncService { sessions, executor, logger, root: root.join("rsync"), protected: root.to_path_buf(), bridge_executable, inner: Mutex::new(Inner::default()) })

159 }

160

161 pub fn data_path(&self) -> String {

162 self.protected.to_string_lossy().to_string()

163 }

164

165 // --- detection -----------------------------------------------------------------------------------

166

167 /// Installations of rsync on this computer, newest kind first. A graphical app on macOS has a minimal PATH.

168 pub async fn local_probe(&self) -> Value {

169 if cfg!(windows) {

170 return json!({ "available": false, "candidates": [], "reason": "windows" });

171 }

172

        let mut directories: Vec<PathBuf> = std::env::var_os("PATH").map(|path| std::env::split_paths(&path).collect()).unwrap_or_default();

173 directories.extend(["/opt/homebrew/bin", "/usr/local/bin", "/usr/bin", "/bin"].map(PathBuf::from));

174 let mut candidates: Vec<Value> = Vec::new();

175 for directory in directories {

176 let path = directory.join("rsync");

177 let text = path.to_string_lossy().to_string();

178 if !path.is_file() || candidates.iter().any(|candidate| candidate["path"] == text) {

179 continue;

180 }

181

            let Ok(Ok(output)) = tokio::time::timeout(Duration::from_secs(5), tokio::process::Command::new(&path).arg("--version").stdin(Stdio::null()).output()).await else { continue };

182 let all = format!("{}{}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr));

183 let version = all.lines().next().unwrap_or("").trim().to_string();

184

            let capabilities = Capabilities::parse(&version, !all.contains("no ACLs") && all.contains("ACLs"), !all.contains("no xattrs") && all.contains("xattrs"));

185

            candidates.push(json!({ "path": text, "version": version, "realRsync": capabilities.real_rsync, "major": capabilities.major, "minor": capabilities.minor, "acls": capabilities.acls, "xattrs": capabilities.xattrs }));

186 }

187

        candidates.sort_by_key(|candidate| std::cmp::Reverse((candidate["major"].as_u64().unwrap_or(0), candidate["minor"].as_u64().unwrap_or(0))));

188 json!({ "available": !candidates.is_empty(), "candidates": candidates })

189 }

190

191 pub async fn remote_probe(&self, session_id: &str) -> CoreResult<Value> {

192 let output = self.operation(session_id, "rsync.probe", vec![], false, None, Duration::from_secs(30)).await?;

193 let values = parse_key_values(&output);

194 let get = |name: &str| values.get(name).cloned().unwrap_or_default();

195 let version = get("version");

196 let capabilities = Capabilities::parse(&version, get("acls") == "1", get("xattrs") == "1");

197 Ok(json!({

198 "available": !get("rsync").is_empty(),

199 "path": get("rsync"),

200 "version": version,

201 "realRsync": capabilities.real_rsync,

202 "major": capabilities.major,

203 "minor": capabilities.minor,

204 "acls": capabilities.acls,

205 "xattrs": capabilities.xattrs,

206 "installer": get("installer"),

207 "ssh": get("ssh"),

208 "scheduler": get("scheduler"),

209 }))

210 }

211

212 pub async fn install(&self, session_id: &str, payload: &Value) -> CoreResult<Value> {

213 if payload["confirm"] != true {

214 return Err(job_error("RSYNC_CONFIRMATION_REQUIRED", "Installing rsync must be confirmed."));

215 }

216 let record = self.sessions.require_connected(session_id)?;

217

        let output = self.privileged(session_id, "rsync.install", vec![], payload["sudoPassword"].as_str(), Duration::from_secs(900)).await?;

218

        self.logger.warn("rsync installed on the remote server.", json!({ "sessionId": record.session_id, "host": record.host }));

219 Ok(json!({ "output": output.trim() }))

220 }

221

222 pub async fn inspect(&self, payload: &Value) -> CoreResult<Value> {

223 let path = payload["path"].as_str().unwrap_or("");

224 if payload["kind"] == "local" {

225 let checked = local::checked_path(path, &self.protected)?;

226 let meta = std::fs::symlink_metadata(&checked);

227 return Ok(match meta {

228

                Err(_) => json!({ "type": "missing", "writable": checked.parent().is_some_and(|parent| parent.exists()), "real": real_local(&checked) }),

229

                Ok(meta) if meta.is_dir() => json!({ "type": "directory", "entries": std::fs::read_dir(&checked).map(|list| list.count()).unwrap_or(0), "writable": !meta.permissions().readonly(), "real": real_local(&checked) }),

230

                Ok(meta) => json!({ "type": "file", "size": meta.len(), "writable": !meta.permissions().readonly(), "real": real_local(&checked) }),

231 });

232 }

233 let session_id = payload["sessionId"].as_str().unwrap_or("");

234

        let output = self.operation(session_id, "rsync.inspect", vec![path.to_string()], payload["privileged"] == true, payload["sudoPassword"].as_str(), Duration::from_secs(60)).await?;

235 let values = parse_key_values(&output);

236 let get = |name: &str| values.get(name).cloned().unwrap_or_default();

237 let number = |name: &str| get(name).parse::<u64>().ok();

238

        Ok(json!({ "type": get("type"), "entries": number("entries"), "size": number("size"), "writable": get("writable") == "1", "availableKiB": number("availableKiB"), "link": values.get("link"), "real": values.get("real") }))

239 }

240

241

    async fn operation(&self, session_id: &str, operation: &str, args: Vec<String>, privileged: bool, password: Option<&str>, timeout: Duration) -> CoreResult<String> {

242 let options = RunOptions {

243 operation: operation.into(),

244 args,

245 privileged,

246 sudo_password: password.filter(|p| !p.is_empty()).map(str::to_string),

247 use_cached_password: privileged,

248 purpose: operation.into(),

249 error_prefix: "RSYNC_".into(),

250 timeout,

251 output_limit: 512 * 1024,

252 ..Default::default()

253 };

254 let result = self.executor.run(session_id, options).await?;

255 Ok(result.stdout)

256 }

257

258

    async fn privileged(&self, session_id: &str, operation: &str, args: Vec<String>, password: Option<&str>, timeout: Duration) -> CoreResult<String> {

259 self.operation(session_id, operation, args, true, password, timeout).await

260 }

261

262 // --- connections ---------------------------------------------------------------------------------

263

264 /// A login for rsync: a saved profile, or a new target with a password or a key. The host key goes through the

265 /// same trust check as the terminal.

266 pub async fn connect(&self, payload: &Value) -> CoreResult<Value> {

267 let session_id = payload["sessionId"].as_str().map(str::to_string).unwrap_or_else(|| new_id("rsync"));

268 let mut options = json!({ "sessionId": session_id, "cols": 80, "rows": 24 });

269 if let Some(profile) = payload["profileId"].as_str().filter(|id| !id.is_empty()) {

270 options["profileId"] = json!(profile);

271 } else {

272 let adhoc = &payload["adhoc"];

273

            options["adhoc"] = json!({ "host": adhoc["host"], "port": adhoc["port"], "username": adhoc["username"], "authMethod": adhoc["authMethod"] });

274 if adhoc["authMethod"] == "privateKey" {

275 options["adhoc"]["privateKeyPath"] = adhoc["privateKeyPath"].clone();

276 }

277 }

278 for name in ["password", "passphrase"] {

279 if let Some(value) = payload[name].as_str().filter(|value| !value.is_empty()) {

280 options[name] = json!(value);

281 }

282 }

283 match self.sessions.connect(&options).await {

284 Ok(status) => {

285 self.inner.lock().opened_sessions.insert(session_id.clone(), Instant::now());

286 Ok(json!({ "sessionId": session_id, "status": status }))

287 }

288 // Module errors carry only text, and the module has to show this fingerprint.

289 Err(error) if error.code == "HOST_KEY_UNKNOWN" || error.code == "HOST_KEY_MISMATCH" => {

290

                Ok(json!({ "sessionId": session_id, "hostKey": { "code": error.code, "fingerprint": error.details.get("fingerprint"), "knownFingerprint": error.details.get("knownFingerprint") } }))

291 }

292 Err(error) => Err(error),

293 }

294 }

295

296 /// Saved SSH logins to choose from; never their secrets.

297 pub fn ssh_profiles(&self) -> CoreResult<Value> {

298 let list = self.sessions.profiles.list()?;

299

        Ok(Value::Array(list.iter().map(|profile| json!({ "id": profile["id"], "label": profile["label"], "host": profile["host"], "port": profile["port"], "username": profile["username"], "authMethod": profile["authMethod"] })).collect()))

300 }

301

302 pub fn trust_host(&self, payload: &Value) -> CoreResult<Value> {

303 self.sessions.trust_pending(payload)

304 }

305

306 pub async fn disconnect(&self, session_id: &str) -> CoreResult<Value> {

307 if self.inner.lock().opened_sessions.remove(session_id).is_none() {

308 return Err(job_error("RSYNC_SESSION_NOT_OWNED", "Only connections opened for rsync can be closed here."));

309 }

310 Ok(self.sessions.disconnect(session_id, "rsync").await)

311 }

312

313 /// A saved job names SSH profiles: a live session is used, otherwise one is opened with what the vault holds.

314 async fn resolve_endpoint(&self, endpoint: &Value) -> CoreResult<Value> {

315 let mut endpoint = endpoint.clone();

316 if endpoint["kind"] != "ssh" {

317 return Ok(endpoint);

318 }

319

        if endpoint["sessionId"].as_str().is_some_and(|id| self.sessions.get(id).is_some_and(|record| record.is_connected())) {

320 return Ok(endpoint);

321 }

322 let Some(profile) = endpoint["profileId"].as_str().filter(|id| !id.is_empty()).map(str::to_string) else {

323 return Err(CoreError::new("SESSION_NOT_CONNECTED", "The server of this job is not connected."));

324 };

325

        let live = self.sessions.list().into_iter().find(|status| status["profileId"] == profile.as_str() && status["state"] == "connected");

326 let session_id = match live {

327 Some(status) => status["sessionId"].as_str().unwrap_or("").to_string(),

328

            None => self.connect(&json!({ "profileId": profile })).await?["sessionId"].as_str().unwrap_or("").to_string(),

329 };

330 endpoint["sessionId"] = json!(session_id);

331 Ok(endpoint)

332 }

333

334 async fn resolve(&self, payload: &Value) -> CoreResult<Value> {

335 let mut value = payload.clone();

336 value["source"] = self.resolve_endpoint(&payload["source"]).await?;

337 if payload["transport"]["kind"] != "direct" {

338 value["target"] = self.resolve_endpoint(&payload["target"]).await?;

339 }

340 Ok(value)

341 }

342

343 fn check_local_paths(&self, spec: &Spec) -> CoreResult<()> {

344 for endpoint in [&spec.source, &spec.target] {

345 if endpoint.place == Place::Local {

346 local::checked_path(&endpoint.path, &self.protected)?;

347 }

348 }

349 Ok(())

350 }

351

352 fn touch(&self, spec: &Spec) {

353 let mut inner = self.inner.lock();

354 for endpoint in [&spec.source, &spec.target] {

355 if let Place::Remote { session_id } = &endpoint.place {

356 if let Some(used) = inner.opened_sessions.get_mut(session_id) {

357 *used = Instant::now();

358 }

359 }

360 }

361 }

362

363 // --- jobs ----------------------------------------------------------------------------------------

364

365 pub fn jobs(&self) -> Value {

366 let inner = self.inner.lock();

367 Value::Array(inner.jobs.iter().rev().take(50).cloned().collect())

368 }

369

370 pub fn job(&self, id: &str) -> Value {

371 self.inner.lock().jobs.iter().find(|job| job["id"] == id).cloned().unwrap_or(Value::Null)

372 }

373

374 fn update_job(&self, id: &str, change: impl FnOnce(&mut Value)) {

375 let mut inner = self.inner.lock();

376 if let Some(job) = inner.jobs.iter_mut().find(|job| job["id"] == id) {

377 change(job);

378 }

379 }

380

381 fn append_output(&self, id: &str, text: &str) {

382 self.update_job(id, |job| {

383 let mut output = job["output"].as_str().unwrap_or("").to_string();

384 output.push_str(text);

385 if output.len() > MAX_OUTPUT {

386 let mut cut = output.len() - MAX_OUTPUT;

387 while !output.is_char_boundary(cut) {

388 cut += 1;

389 }

390 output.drain(..cut);

391 }

392 job["output"] = json!(output);

393 });

394 }

395

396 fn register_job(&self, id: &str, kind: &str, spec: &Spec, label: &str, owner: &str) {

397 let job = json!({

398 "id": id, "kind": kind, "state": "running", "label": label, "owner": owner, "mode": spec.mode.name(),

399 "source": describe_endpoint(&spec.source), "target": describe_endpoint(&spec.target),

400

            "started": now(), "progress": { "percent": null }, "summary": {}, "items": [], "itemsTotal": 0, "output": "",

401 });

402 let mut inner = self.inner.lock();

403 inner.jobs.push(job);

404 if inner.jobs.len() > 200 {

405 inner.jobs.remove(0);

406 }

407 }

408

409 /// The options a run would use, for the preview of the command. No secret is ever among them.

410 pub fn describe(&self, payload: &Value) -> CoreResult<Value> {

411 let spec = spec::parse(payload)?;

412 let capabilities = Capabilities { real_rsync: true, major: 3, minor: 2, acls: true, xattrs: true };

413 let mut args = spec::flags(&spec, &capabilities, false, "<stamp>", Some(0))?;

414 args.push(spec::source_argument(&spec));

415 args.push(spec::target_argument(&spec, "<stamp>"));

416 Ok(json!({ "arguments": args }))

417 }

418

419 pub async fn preview(self: &Arc<Self>, owner: &str, payload: &Value) -> CoreResult<Value> {

420 let resolved = self.resolve(payload).await?;

421 let spec = spec::parse(&resolved)?;

422 self.check_local_paths(&spec)?;

423 self.touch(&spec);

424 let secrets = Secrets::read(payload);

425 let id = new_id("job");

426 let label = payload["label"].as_str().unwrap_or("").chars().take(120).collect::<String>();

427 self.register_job(&id, "preview", &spec, &label, owner);

428 self.spawn(id.clone(), spec, secrets, true, None, false, resolved);

429 Ok(json!({ "jobId": id }))

430 }

431

432 /// Starts what a preview showed. Deleting needs the number the user saw, a protected target its exact path.

433 pub fn run(self: &Arc<Self>, owner: &str, payload: &Value) -> CoreResult<Value> {

434 if payload["confirm"] != true {

435 return Err(job_error("RSYNC_CONFIRMATION_REQUIRED", "The synchronization must be confirmed."));

436 }

437 let plan_id = payload["planId"].as_str().unwrap_or("");

438 let plan = {

439 let mut inner = self.inner.lock();

440 inner.plans.remove(plan_id)

441 };

442

        let plan = plan.filter(|plan| plan.created.elapsed() < PLAN_LIFETIME).ok_or_else(|| job_error("RSYNC_PLAN_EXPIRED", "The preview is no longer valid. Run it again."))?;

443 let restore = |service: &Self, plan: Plan| {

444 service.inner.lock().plans.insert(plan_id.to_string(), plan);

445 };

446 if plan.spec.mode == Mode::Compare {

447 return Err(job_error("RSYNC_COMPARE_ONLY", "A comparison transfers nothing."));

448 }

449 let removes = plan.deletes > 0 && matches!(plan.spec.mode, Mode::Mirror | Mode::Versioned);

450 if removes && payload["acceptDeletes"].as_u64() != Some(plan.deletes) {

451 let error = job_error("RSYNC_DELETE_NOT_ACCEPTED", "The number of files to delete must be confirmed.");

452 restore(self, plan);

453 return Err(error);

454 }

455

        if removes && plan.protected_target && payload["confirmProtected"].as_str().map(spec::normalize) != Some(spec::normalize(&plan.spec.target.path)) {

456

            let error = job_error("RSYNC_TARGET_PROTECTED", "Deleting in a system or home directory needs its exact path as a confirmation.");

457 restore(self, plan);

458 return Err(error);

459 }

460 if removes && plan.empty_source && payload["allowEmptySource"] != true {

461

            let error = job_error("RSYNC_SOURCE_EMPTY", "The source is empty, so everything in the target would be deleted. This must be confirmed separately.");

462 restore(self, plan);

463 return Err(error);

464 }

465 // Two runs into one place would mix their versions and snapshots.

466 if self.target_busy(&plan.spec) {

467 let error = job_error("RSYNC_TARGET_BUSY", "Another transfer is writing to this target.");

468 restore(self, plan);

469 return Err(error);

470 }

471 let id = new_id("job");

472 let label = payload["label"].as_str().unwrap_or("").chars().take(120).collect::<String>();

473 self.register_job(&id, "run", &plan.spec, &label, owner);

474 self.touch(&plan.spec);

475 let allowed = Some(plan.deletes);

476 let allow_empty = payload["allowEmptySource"] == true;

477 self.spawn(id.clone(), plan.spec, plan.secrets, false, allowed, allow_empty, plan.source_value);

478 Ok(json!({ "jobId": id }))

479 }

480

481 pub async fn cancel(&self, id: &str) -> CoreResult<Value> {

482 let (cancel, remote) = {

483 let inner = self.inner.lock();

484 match inner.running.get(id) {

485 Some(running) => (running.cancel.clone(), running.remote_pid.lock().clone()),

486 None => return Ok(json!({ "cancelled": false })),

487 }

488 };

489 let _ = cancel.send(true);

490 if let Some((session, pid, privileged, password)) = remote {

491

            let _ = self.operation(&session, "rsync.cancel", vec![pid], privileged, password.as_ref().map(|p| p.as_str()), Duration::from_secs(20)).await;

492 }

493 self.update_job(id, |job| job["cancelRequested"] = json!(true));

494 Ok(json!({ "cancelled": true }))

495 }

496

497 fn target_busy(&self, spec: &Spec) -> bool {

498 let description = describe_endpoint(&spec.target);

499 let inner = self.inner.lock();

500 inner.jobs.iter().any(|job| job["state"] == "running" && job["kind"] == "run" && job["target"] == description)

501 }

502

503

    fn spawn(self: &Arc<Self>, id: String, spec: Spec, secrets: Secrets, dry_run: bool, allowed: Option<u64>, allow_empty: bool, source_value: Value) {

504 let (cancel, cancelled) = watch::channel(false);

505 let remote_pid = Arc::new(Mutex::new(None));

506 self.inner.lock().running.insert(id.clone(), Running { cancel, remote_pid: remote_pid.clone() });

507 let service = self.clone();

508 tokio::spawn(async move {

509

            let result = service.execute(&id, &spec, &secrets, dry_run, allowed, allow_empty, cancelled.clone(), remote_pid).await;

510 let was_cancelled = *cancelled.borrow();

511 service.finish(&id, spec, secrets, dry_run, result, was_cancelled, source_value).await;

512 });

513 }

514

515 #[allow(clippy::too_many_arguments)]

516

    async fn finish(self: &Arc<Self>, id: &str, spec: Spec, secrets: Secrets, dry_run: bool, result: CoreResult<(Collected, String)>, cancelled: bool, source_value: Value) {

517 self.inner.lock().running.remove(id);

518 let (state, code, message, collected, run_stamp) = match result {

519 Err(error) => {

520 self.update_job(id, |job| {

521 job["errorCode"] = json!(error.code);

522 job["errorDetails"] = json!(error.details);

523 });

524 ("failed", None, error.message.clone(), None, String::new())

525 }

526 Ok((collected, run_stamp)) => {

527 let code = collected.code.unwrap_or(255);

528 let (state, message) = parse::exit_meaning(code);

529 let state = if cancelled { "cancelled" } else { state };

530 (state, Some(code), message.to_string(), Some(collected), run_stamp)

531 }

532 };

533 let mut plan_id = Value::Null;

534 let mut snapshot_failed = false;

535 if let Some(collected) = &collected {

536 let succeeded = matches!(state, "ok" | "warning");

537 if dry_run && succeeded {

538 let empty_source = self.source_is_empty(&spec, &secrets).await;

539 // A symbolic link must not hide a system folder from this check.

540

                let protected = spec::is_protected(&spec.target.path) || self.real_path(&spec.target, secrets.target_sudo.as_ref().map(|p| p.as_str())).await.is_some_and(|real| spec::is_protected(&real));

541 let new_plan = new_id("plan");

542 plan_id = json!(new_plan);

543 let mut inner = self.inner.lock();

544 inner.plans.retain(|_, plan| plan.created.elapsed() < PLAN_LIFETIME);

545

                inner.plans.insert(new_plan, Plan { spec: spec.clone(), secrets: secrets.clone(), deletes: collected.summary.delete, empty_source, protected_target: protected, created: Instant::now(), source_value });

546 drop(inner);

547 self.update_job(id, |job| {

548 job["emptySource"] = json!(empty_source);

549 job["protectedTarget"] = json!(protected);

550 });

551 }

552 let mut published = true;

553 if !dry_run && succeeded && spec.mode == Mode::Snapshot {

554 if let Err(error) = self.finish_snapshot(&spec, &secrets, &run_stamp).await {

555 self.append_output(id, &format!("\n{}\n", error.message));

556 published = false;

557 snapshot_failed = true;

558 }

559 }

560

            if !dry_run && succeeded && published && (spec.options.keep_days > 0 || spec.options.keep_count > 0) && matches!(spec.mode, Mode::Versioned | Mode::Snapshot) {

561

                match self.prune(&spec.target, &secrets, spec.options.keep_days, spec.options.keep_count, Some(&spec.mode)).await {

562

                    Ok(output) if !output.trim().is_empty() => self.append_output(id, &format!("\n{}\n", output.trim())),

563 Ok(_) => {}

564 Err(error) => self.append_output(id, &format!("\n{}\n", error.message)),

565 }

566 }

567 }

568 let stats = collected.as_ref().map(|c| parse::statistics(&c.stats_lines)).unwrap_or_else(|| json!({}));

569 let summary = collected.as_ref().map(|c| c.summary.to_json()).unwrap_or_else(|| json!({}));

570 let total = collected.as_ref().map(|c| c.total_items).unwrap_or(0);

571

        let (state, message) = if snapshot_failed { ("failed", "The snapshot was transferred, but the latest link could not be updated.".to_string()) } else { (state, message) };

572 self.update_job(id, |job| {

573 job["state"] = json!(state);

574 job["code"] = json!(code);

575 job["message"] = json!(message);

576 job["finished"] = json!(now());

577 job["stats"] = stats.clone();

578 job["summary"] = summary.clone();

579 job["itemsTotal"] = json!(total);

580 job["planId"] = plan_id.clone();

581 if state == "ok" {

582 job["progress"]["percent"] = json!(100);

583 }

584 });

585 if !dry_run {

586 let entry = {

587 let mut entry = self.job(id);

588 entry.as_object_mut().map(|object| {

589 object.remove("items");

590 object.remove("output");

591 });

592 entry

593 };

594 let _ = self.append_history(entry);

595 self.logger.info("rsync job finished.", json!({ "jobId": id, "state": state, "code": code }));

596 }

597 }

598

599 /// "missing", "empty" or "present".

600 async fn source_state(&self, spec: &Spec, secrets: &Secrets) -> &'static str {

601 let payload = match &spec.source.place {

602 Place::Local => json!({ "kind": "local", "path": spec.source.path }),

603

            Place::Remote { session_id } => json!({ "kind": "ssh", "sessionId": session_id, "path": spec.source.path, "privileged": spec.source.privileged, "sudoPassword": secrets.source_sudo.as_ref().map(|p| p.as_str()) }),

604 };

605 match self.inspect(&payload).await {

606 Ok(value) if value["type"] == "missing" => "missing",

607 Ok(value) if value["type"] == "directory" && value["entries"] == 0 => "empty",

608 Ok(_) => "present",

609 Err(_) => "unknown",

610 }

611 }

612

613 async fn real_path(&self, endpoint: &Endpoint, password: Option<&str>) -> Option<String> {

614 let payload = match &endpoint.place {

615 Place::Local => json!({ "kind": "local", "path": endpoint.path }),

616 Place::Remote { session_id } if session_id == spec::DIRECT_TARGET => return None,

617

            Place::Remote { session_id } => json!({ "kind": "ssh", "sessionId": session_id, "path": endpoint.path, "privileged": endpoint.privileged, "sudoPassword": password }),

618 };

619 self.inspect(&payload).await.ok()?["real"].as_str().filter(|real| real.starts_with('/')).map(str::to_string)

620 }

621

622 /// Two runs of this application never share a folder of versions or a snapshot.

623 async fn unique_stamp(&self) -> String {

624 loop {

625 let next = stamp();

626 {

627 let mut inner = self.inner.lock();

628 if inner.last_stamp.as_deref() != Some(next.as_str()) {

629 inner.last_stamp = Some(next.clone());

630 return next;

631 }

632 }

633 tokio::time::sleep(Duration::from_millis(200)).await;

634 }

635 }

636

637 async fn source_is_empty(&self, spec: &Spec, secrets: &Secrets) -> bool {

638 matches!(spec.mode, Mode::Mirror | Mode::Versioned) && self.source_state(spec, secrets).await == "empty"

639 }

640

641 async fn finish_snapshot(&self, spec: &Spec, secrets: &Secrets, run_stamp: &str) -> CoreResult<()> {

642 let base = spec::normalize(&spec.target.path);

643 match &spec.target.place {

644 Place::Local => point_latest_local(Path::new(&base), run_stamp),

645 Place::Remote { session_id } => {

646

                self.operation(session_id, "rsync.snapshot.finish", vec![base, run_stamp.to_string()], spec.target.privileged, target_password(spec, secrets), Duration::from_secs(60)).await?;

647 Ok(())

648 }

649 }

650 }

651

652

    async fn prune(&self, target: &Endpoint, secrets: &Secrets, keep_days: u64, keep_count: u64, mode: Option<&Mode>) -> CoreResult<String> {

653 let base = spec::normalize(&target.path);

654 let password = secrets.target_sudo.as_ref().map(|p| p.as_str());

655 let kinds: &[&str] = match mode {

656 Some(Mode::Versioned) => &["versions"],

657 Some(Mode::Snapshot) => &["snapshots"],

658 _ => &["versions", "snapshots"],

659 };

660 match &target.place {

661 Place::Local => prune_local(Path::new(&base), keep_days, keep_count, kinds),

662 Place::Remote { session_id } => {

663 let mut output = String::new();

664 for kind in kinds {

665

                    output.push_str(&self.operation(session_id, "rsync.versions.prune", vec![base.clone(), keep_days.to_string(), keep_count.to_string(), kind.to_string()], target.privileged, password, Duration::from_secs(1800)).await?);

666 }

667 Ok(output)

668 }

669 }

670 }

671

672 async fn capabilities(&self, place: &Place, local_binary: &mut Option<PathBuf>) -> CoreResult<Capabilities> {

673 match place {

674 Place::Local => {

675 let probe = self.local_probe().await;

676

                let first = probe["candidates"].get(0).cloned().ok_or_else(|| job_error("RSYNC_LOCAL_MISSING", "rsync was not found on this computer."))?;

677 *local_binary = first["path"].as_str().map(PathBuf::from);

678

                Ok(Capabilities { real_rsync: first["realRsync"] == true, major: first["major"].as_u64().unwrap_or(0) as u32, minor: first["minor"].as_u64().unwrap_or(0) as u32, acls: first["acls"] == true, xattrs: first["xattrs"] == true })

679 }

680 Place::Remote { session_id } => {

681 let probe = self.remote_probe(session_id).await?;

682 if probe["available"] != true {

683

                    return Err(CoreError::new("RSYNC_MISSING", "rsync is not installed on this server.").with("sessionId", session_id.as_str()).with("installer", probe["installer"].as_str().unwrap_or("none")));

684 }

685

                Ok(Capabilities { real_rsync: probe["realRsync"] == true, major: probe["major"].as_u64().unwrap_or(0) as u32, minor: probe["minor"].as_u64().unwrap_or(0) as u32, acls: probe["acls"] == true, xattrs: probe["xattrs"] == true })

686 }

687 }

688 }

689

690 async fn ensure_target(&self, spec: &Spec, secrets: &Secrets) -> CoreResult<()> {

691 let path = spec::normalize(&spec.target.path);

692 match &spec.target.place {

693 Place::Local => std::fs::create_dir_all(&path).map_err(|e| job_error("RSYNC_MKDIR_FAILED", &e.to_string())),

694 Place::Remote { session_id } => {

695

                self.operation(session_id, "rsync.target.ensure", vec![path], spec.target.privileged, target_password(spec, secrets), Duration::from_secs(60)).await?;

696 Ok(())

697 }

698 }

699 }

700

701 #[allow(clippy::too_many_arguments)]

702

    async fn execute(self: &Arc<Self>, id: &str, spec: &Spec, secrets: &Secrets, dry_run: bool, allowed: Option<u64>, allow_empty: bool, cancelled: watch::Receiver<bool>, remote_pid: Arc<Mutex<Option<(String, String, bool, Option<Zeroizing<String>>)>>>) -> CoreResult<(Collected, String)> {

703 let run_stamp = self.unique_stamp().await;

704 // On one machine a symbolic link can hide that the two paths lie inside each other.

705 let one_machine = match (&spec.source.place, &spec.target.place) {

706 (Place::Local, Place::Local) => true,

707 (Place::Remote { session_id: a }, Place::Remote { session_id: b }) => a == b || self.same_login(a, b),

708 _ => false,

709 };

710 if one_machine && !matches!(spec.transport, Transport::Direct { .. }) {

711 spec::check_overlap(spec)?;

712 let source_real = self.real_path(&spec.source, secrets.source_sudo.as_ref().map(|p| p.as_str())).await;

713 let target_real = self.real_path(&spec.target, target_password(spec, secrets)).await;

714 if let (Some(source_real), Some(target_real)) = (source_real, target_real) {

715 let mut resolved = spec.clone();

716 resolved.source.path = source_real;

717 resolved.target.path = target_real;

718 spec::check_overlap(&resolved)?;

719 }

720 }

721 // The state may have changed since the preview: an unmounted disk looks like an empty folder.

722 if !dry_run && matches!(spec.mode, Mode::Mirror | Mode::Versioned) {

723 let source = self.source_state(spec, secrets).await;

724 if source == "missing" {

725

                return Err(job_error("RSYNC_SOURCE_MISSING", "The source does not exist any more, so nothing was changed."));

726 }

727 if source == "empty" && !allow_empty {

728

                return Err(job_error("RSYNC_SOURCE_EMPTY", "The source is empty, so everything in the target would be deleted. This must be confirmed separately."));

729 }

730 if source == "unknown" {

731

                return Err(job_error("RSYNC_SOURCE_UNKNOWN", "The source could not be checked, so nothing was changed."));

732 }

733 }

734 if *cancelled.borrow() {

735 let collected = Collected { code: Some(20), ..Default::default() };

736 return Ok((collected, run_stamp));

737 }

738 let mut local_binary = None;

739 // Both ends need rsync, and the one that runs the client decides the options.

740 let source_capabilities = self.capabilities(&spec.source.place, &mut local_binary).await?;

741 let direct = matches!(spec.transport, Transport::Direct { .. });

742

        let target_capabilities = if direct { source_capabilities.clone() } else { self.capabilities(&spec.target.place, &mut local_binary).await? };

743

        let client = if spec.source.place == Place::Local || spec.target.place == Place::Local { if spec.source.place == Place::Local { source_capabilities } else { target_capabilities } } else { source_capabilities };

744 let mut flags = spec::flags(spec, &client, dry_run, &run_stamp, allowed)?;

745 if !dry_run && !direct {

746 self.ensure_target(spec, secrets).await?;

747 }

748 let source_arg = spec::source_argument(spec);

749 let target_arg = spec::target_argument(spec, &run_stamp);

750 let collected = match (&spec.source.place, &spec.target.place) {

751 (Place::Local, Place::Local) => {

752 flags.push(source_arg);

753 flags.push(target_arg);

754 self.run_local(id, local_binary, flags, None, cancelled).await?

755 }

756 (Place::Local, Place::Remote { session_id }) => {

757 flags.push(source_arg);

758 flags.push(format!("x:{target_arg}"));

759

                self.run_local(id, local_binary, flags, Some((session_id.clone(), spec.target.privileged, secrets.target_sudo.clone())), cancelled).await?

760 }

761 (Place::Remote { session_id }, Place::Local) => {

762 flags.push(format!("x:{source_arg}"));

763 flags.push(target_arg);

764

                self.run_local(id, local_binary, flags, Some((session_id.clone(), spec.source.privileged, secrets.source_sudo.clone())), cancelled).await?

765 }

766 (Place::Remote { session_id: a }, Place::Remote { session_id: b }) if a == b || self.same_login(a, b) => {

767 flags.push(source_arg);

768 flags.push(target_arg);

769 let privileged = spec.source.privileged || spec.target.privileged;

770 let password = secrets.source_sudo.clone().or_else(|| secrets.target_sudo.clone());

771 self.run_remote(id, a, "rsync.run", flags, privileged, password, None, cancelled, remote_pid).await?

772 }

773 (Place::Remote { session_id: a }, Place::Remote { session_id: b }) => match &spec.transport {

774 Transport::Relay => {

775 flags.push(source_arg);

776 flags.push(format!("x:{target_arg}"));

777 self.run_relay(id, a, b, flags, spec, secrets, cancelled, remote_pid).await?

778 }

779 Transport::Direct { host, port, username, method, key_path, accept_new_host_key } => {

780 flags.push(source_arg);

781 let host = if host.contains(':') { format!("[{host}]") } else { host.clone() };

782 flags.push(format!("{username}@{host}:{target_arg}"));

783

                    let mut args = vec![method.clone(), port.to_string(), key_path.clone(), if *accept_new_host_key { "accept-new".to_string() } else { "yes".to_string() }];

784 args.extend(flags);

785

                    let password = (method == "password").then(|| secrets.direct_password.clone().ok_or_else(|| job_error("RSYNC_PASSWORD_REQUIRED", "Enter the password for the login between the servers."))).transpose()?;

786

                    self.run_remote(id, a, "rsync.direct", args, spec.source.privileged, secrets.source_sudo.clone(), password, cancelled, remote_pid).await?

787 }

788 },

789 };

790 Ok((collected, run_stamp))

791 }

792

793 fn same_login(&self, a: &str, b: &str) -> bool {

794 match (self.sessions.get(a), self.sessions.get(b)) {

795 (Some(a), Some(b)) => a.host == b.host && a.port == b.port && a.username == b.username,

796 _ => false,

797 }

798 }

799

800 fn take_line(&self, id: &str, line: &str, collected: &mut Collected) {

801 if let Some(value) = line.strip_prefix("\u{1e}CHJ_PID=") {

802 collected.pid = Some(value.trim().to_string());

803 return;

804 }

805 if let Some(value) = line.strip_prefix("\u{1e}CHJ_EXIT=") {

806 collected.code = value.trim().parse().ok();

807 return;

808 }

809 if let Some(value) = line.strip_prefix("\u{1e}CHJ_BRIDGE=") {

810 collected.bridge = Some(value.trim().to_string());

811 return;

812 }

813 if line.starts_with('\u{1e}') {

814 return;

815 }

816 if let Some(progress) = parse::progress(line) {

817 self.update_job(id, |job| job["progress"] = progress);

818 return;

819 }

820 if let Some(item) = parse::item(line) {

821 collected.summary.add(&item);

822 collected.total_items += 1;

823 let summary = collected.summary.to_json();

824 let total = collected.total_items;

825

            let value = (collected.items.len() < MAX_ITEMS).then(|| json!({ "change": item.change.name(), "size": item.size, "path": item.path, "flags": item.flags }));

826 if let Some(value) = &value {

827 collected.items.push(value.clone());

828 }

829 self.update_job(id, |job| {

830 job["summary"] = summary;

831 job["itemsTotal"] = json!(total);

832 if let (Some(value), Some(items)) = (value, job["items"].as_array_mut()) {

833 items.push(value);

834 }

835 });

836 return;

837 }

838 if line.trim().is_empty() {

839 return;

840 }

841 collected.stats_lines.push(line.to_string());

842 if collected.stats_lines.len() > 200 {

843 collected.stats_lines.remove(0);

844 }

845 self.append_output(id, &format!("{line}\n"));

846 }

847

848 fn feed(&self, id: &str, pending: &mut Vec<u8>, data: &[u8], collected: &mut Collected) {

849 pending.extend_from_slice(data);

850 while let Some(end) = pending.iter().position(|byte| *byte == b'\n' || *byte == b'\r') {

851 let line: Vec<u8> = pending.drain(..=end).collect();

852 let text = String::from_utf8_lossy(&line[..line.len() - 1]).into_owned();

853 self.take_line(id, &text, collected);

854 }

855 if pending.len() > 1024 * 1024 {

856 pending.clear();

857 }

858 }

859

860 /// The local rsync. With a session, its remote shell is this application's bridge over that session.

861

    async fn run_local(self: &Arc<Self>, id: &str, binary: Option<PathBuf>, flags: Vec<String>, remote: Option<(String, bool, Option<Zeroizing<String>>)>, mut cancelled: watch::Receiver<bool>) -> CoreResult<Collected> {

862 let binary = binary.ok_or_else(|| job_error("RSYNC_LOCAL_MISSING", "rsync was not found on this computer."))?;

863 let mut command = tokio::process::Command::new(&binary);

864

        command.stdin(Stdio::null()).stdout(Stdio::piped()).stderr(Stdio::piped()).kill_on_drop(true).env("LC_ALL", "C");

865 let mut link_dir = None;

866 let mut bridge_task = None;

867 if let Some((session_id, privileged, password)) = remote {

868 // rsync splits its remote shell on spaces, and the path of an installed app has them.

869 let directory = LinkDirectory::create().map_err(|e| job_error("RSYNC_BRIDGE_FAILED", &e.to_string()))?;

870 let link = directory.0.join("bridge");

871 make_link(&self.bridge_executable, &link).map_err(|e| job_error("RSYNC_BRIDGE_FAILED", &e.to_string()))?;

872

            let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)).await.map_err(|e| job_error("RSYNC_BRIDGE_FAILED", &e.to_string()))?;

873 let port = listener.local_addr().map_err(|e| job_error("RSYNC_BRIDGE_FAILED", &e.to_string()))?.port();

874 let token = uuid::Uuid::new_v4().simple().to_string();

875 command.env(bridge::BRIDGE_ENV, format!("{port}:{token}"));

876 command.arg("-e").arg(format!("{} {}", link.display(), bridge::BRIDGE_FLAG));

877 let service = self.clone();

878 let job = id.to_string();

879 let executor = self.executor.clone();

880 bridge_task = Some(tokio::spawn(async move {

881 let result = bridge::serve_local(listener, token, executor, session_id, privileged, password, {

882 let service = service.clone();

883 let job = job.clone();

884 move |text| service.append_output(&job, &text)

885 })

886 .await;

887 if let Err(error) = result {

888 service.append_output(&job, &format!("{}\n", error.message));

889 }

890 }));

891 link_dir = Some(directory);

892 }

893 command.args(&flags);

894 let mut child = command.spawn().map_err(|e| job_error("RSYNC_LOCAL_FAILED", &e.to_string()))?;

895 let mut stdout = child.stdout.take().unwrap();

896 let mut stderr = child.stderr.take().unwrap();

897 let mut collected = Collected::default();

898 let mut pending = Vec::new();

899 let mut buffer = vec![0u8; 65536];

900 let mut error_buffer = vec![0u8; 8192];

901 let (mut out_open, mut err_open) = (true, true);

902 let mut deadline = None;

903 while out_open || err_open {

904 tokio::select! {

905 read = stdout.read(&mut buffer), if out_open => match read {

906 Ok(0) | Err(_) => out_open = false,

907 Ok(count) => self.feed(id, &mut pending, &buffer[..count], &mut collected),

908 },

909 read = stderr.read(&mut error_buffer), if err_open => match read {

910 Ok(0) | Err(_) => err_open = false,

911 Ok(count) => self.append_output(id, &String::from_utf8_lossy(&error_buffer[..count])),

912 },

913 _ = cancelled.changed() => {

914 let _ = child.start_kill();

915 if let Some(task) = &bridge_task {

916 task.abort();

917 }

918 deadline = Some(tokio::time::Instant::now() + CANCEL_GRACE);

919 }

920 _ = wait_until(deadline) => break,

921 }

922 }

923 let status = child.wait().await.map_err(|e| job_error("RSYNC_LOCAL_FAILED", &e.to_string()))?;

924 if !pending.is_empty() {

925 let text = String::from_utf8_lossy(&pending).into_owned();

926 self.take_line(id, &text, &mut collected);

927 }

928 collected.code = Some(status.code().map(i64::from).unwrap_or(20));

929 if let Some(task) = bridge_task {

930 task.abort();

931 }

932 drop(link_dir);

933 Ok(collected)

934 }

935

936 /// rsync on a server: both paths there, or a direct login from there.

937 #[allow(clippy::too_many_arguments)]

938

    async fn run_remote(self: &Arc<Self>, id: &str, session_id: &str, operation: &str, args: Vec<String>, privileged: bool, sudo: Option<Zeroizing<String>>, secret_line: Option<Zeroizing<String>>, mut cancelled: watch::Receiver<bool>, remote_pid: Arc<Mutex<Option<(String, String, bool, Option<Zeroizing<String>>)>>>) -> CoreResult<Collected> {

939 let record = self.sessions.require_connected(session_id)?;

940 let mut lost = record.lost();

941

        let mut stream = self.executor.stream(session_id, operation, &args, privileged, sudo.as_ref().map(|p| p.as_str())).await?;

942 if let Some(secret) = secret_line {

943

            stream.write.data_bytes(format!("{}\n", secret.as_str()).into_bytes()).await.map_err(|e| job_error("RSYNC_REMOTE_FAILED", &e.to_string()))?;

944 }

945 let mut collected = Collected::default();

946 let mut pending = Vec::new();

947 let mut exit = None;

948 let mut deadline = None;

949 loop {

950 tokio::select! {

951 event = stream.next() => match event {

952 StreamEvent::Out(data) => {

953 self.feed(id, &mut pending, &data, &mut collected);

954 if let Some(pid) = collected.pid.take() {

955 *remote_pid.lock() = Some((session_id.to_string(), pid, privileged, sudo.clone()));

956 // Cancelled before its number was known.

957 if *cancelled.borrow() {

958 self.stop_remote(&remote_pid).await;

959 }

960 }

961 }

962 StreamEvent::Err(data) => self.append_output(id, &String::from_utf8_lossy(&data)),

963 StreamEvent::Exit(code) => exit = Some(code),

964

                    StreamEvent::BadPassword => return Err(CoreError::new("SUDO_PASSWORD_INCORRECT", "The sudo password was rejected.")),

965 StreamEvent::Closed => break,

966 },

967 _ = lost.changed() => {

968 if *lost.borrow() {

969 return Err(CoreError::new("SESSION_NOT_CONNECTED", "The SSH connection was lost."));

970 }

971 }

972 _ = cancelled.changed() => { deadline = Some(tokio::time::Instant::now() + CANCEL_GRACE); }

973 _ = wait_until(deadline) => {

974 let _ = stream.write.close().await;

975 collected.code = Some(20);

976 break;

977 }

978 }

979 }

980 if collected.code.is_none() {

981 if exit == Some(3) {

982 let output = self.job(id)["output"].as_str().unwrap_or("").to_string();

983 return Err(job_error("RSYNC_REMOTE_FAILED", output.trim()));

984 }

985 collected.code = exit.or(Some(255));

986 }

987 Ok(collected)

988 }

989

990 /// rsync on the source server, its remote end carried by the application to the other server.

991 #[allow(clippy::too_many_arguments)]

992

    async fn run_relay(self: &Arc<Self>, id: &str, near: &str, far: &str, flags: Vec<String>, spec: &Spec, secrets: &Secrets, mut cancelled: watch::Receiver<bool>, remote_pid: Arc<Mutex<Option<(String, String, bool, Option<Zeroizing<String>>)>>>) -> CoreResult<Collected> {

993 let near_record = self.sessions.require_connected(near)?;

994 let far_record = self.sessions.require_connected(far)?;

995 let (mut near_lost, mut far_lost) = (near_record.lost(), far_record.lost());

996 let near_sudo = secrets.source_sudo.clone();

997

        let mut client = self.executor.stream(near, "rsync.client", &flags, spec.source.privileged, near_sudo.as_ref().map(|p| p.as_str())).await?;

998 let mut collected = Collected::default();

999 let mut pending = Vec::new();

1000 let mut task: Option<tokio::task::JoinHandle<CoreResult<()>>> = None;

1001 let mut failure: Option<CoreError> = None;

1002 let mut exit = None;

1003 let mut deadline = None;

1004 loop {

1005 tokio::select! {

1006 event = client.next() => match event {

1007 StreamEvent::Out(data) => {

1008 self.feed(id, &mut pending, &data, &mut collected);

1009 if let Some(pid) = collected.pid.take() {

1010

                            *remote_pid.lock() = Some((near.to_string(), pid, spec.source.privileged, near_sudo.clone()));

1011 if *cancelled.borrow() {

1012 self.stop_remote(&remote_pid).await;

1013 }

1014 }

1015 if task.is_none() && failure.is_none() {

1016 if let Some(work) = collected.bridge.clone() {

1017 task = Some(self.start_relay(id, near, far, work, spec, secrets));

1018 }

1019 }

1020 }

1021 StreamEvent::Err(data) => self.append_output(id, &String::from_utf8_lossy(&data)),

1022 StreamEvent::Exit(code) => exit = Some(code),

1023

                    StreamEvent::BadPassword => return Err(CoreError::new("SUDO_PASSWORD_INCORRECT", "The sudo password was rejected.")),

1024 StreamEvent::Closed => break,

1025 },

1026 // The far side could not start or broke off: the client here would wait for it for ever.

1027 result = async { task.as_mut().unwrap().await }, if task.is_some() => {

1028 task = None;

1029 if let Ok(Err(error)) = result {

1030 failure = Some(error);

1031 self.stop_remote(&remote_pid).await;

1032 deadline = Some(tokio::time::Instant::now() + CANCEL_GRACE);

1033 }

1034 }

1035

                _ = near_lost.changed() => { if *near_lost.borrow() { return Err(CoreError::new("SESSION_NOT_CONNECTED", "The SSH connection was lost.")); } }

1036

                _ = far_lost.changed() => { if *far_lost.borrow() { self.stop_remote(&remote_pid).await; return Err(CoreError::new("SESSION_NOT_CONNECTED", "The SSH connection was lost.")); } }

1037 _ = cancelled.changed() => {

1038 if let Some(task) = &task {

1039 task.abort();

1040 }

1041 deadline = Some(tokio::time::Instant::now() + CANCEL_GRACE);

1042 }

1043 _ = wait_until(deadline) => {

1044 let _ = client.write.close().await;

1045 collected.code = Some(20);

1046 break;

1047 }

1048 }

1049 }

1050 // The bridge ends by itself now that the client is gone. Only one that does not gets cut off.

1051 if let Some(mut task) = task {

1052 if tokio::time::timeout(CANCEL_GRACE, &mut task).await.is_err() {

1053 task.abort();

1054 self.stop_remote(&remote_pid).await;

1055 }

1056 }

1057 if let Some(error) = failure {

1058 return Err(error);

1059 }

1060 if collected.code.is_none() {

1061 collected.code = exit.or(Some(255));

1062 }

1063 Ok(collected)

1064 }

1065

1066 async fn stop_remote(&self, remote_pid: &Arc<Mutex<Option<(String, String, bool, Option<Zeroizing<String>>)>>>) {

1067 let known = remote_pid.lock().clone();

1068 if let Some((session, pid, privileged, password)) = known {

1069

            let _ = self.operation(&session, "rsync.cancel", vec![pid], privileged, password.as_ref().map(|p| p.as_str()), Duration::from_secs(20)).await;

1070 }

1071 }

1072

1073

    fn start_relay(self: &Arc<Self>, id: &str, near: &str, far: &str, work: String, spec: &Spec, secrets: &Secrets) -> tokio::task::JoinHandle<CoreResult<()>> {

1074 let service = self.clone();

1075 let job = id.to_string();

1076 let (near, far) = (near.to_string(), far.to_string());

1077 let (near_privileged, far_privileged) = (spec.source.privileged, spec.target.privileged);

1078 let (near_sudo, far_sudo) = (secrets.source_sudo.clone(), secrets.target_sudo.clone());

1079 tokio::spawn(async move {

1080 async {

1081

                let mut bridge_stream = service.executor.stream(&near, "rsync.bridge", std::slice::from_ref(&work), near_privileged, near_sudo.as_ref().map(|p| p.as_str())).await?;

1082 // The first lines are the arguments rsync gave its remote shell.

1083 let mut header = Vec::new();

1084 let marker = b"CHJ_ARGS_END\n";

1085 let (arguments, leftover) = loop {

1086 if let Some(end) = header.windows(marker.len()).position(|window| window == marker) {

1087 let text = String::from_utf8_lossy(&header[..end]).into_owned();

1088 let leftover = header[end + marker.len()..].to_vec();

1089 break (text.lines().map(str::to_string).collect::<Vec<_>>(), leftover);

1090 }

1091 match bridge_stream.next().await {

1092 StreamEvent::Out(data) => header.extend_from_slice(&data),

1093 StreamEvent::Err(data) => service.append_output(&job, &String::from_utf8_lossy(&data)),

1094 StreamEvent::Exit(_) => {}

1095

                        StreamEvent::BadPassword => return Err(CoreError::new("SUDO_PASSWORD_INCORRECT", "The sudo password was rejected.")),

1096 StreamEvent::Closed => return Ok::<(), CoreError>(()),

1097 }

1098 };

1099 let mut with_host = vec!["x".to_string()];

1100 with_host.extend(arguments);

1101 let args = bridge::server_arguments(&with_host)?;

1102

                let far_stream = service.executor.stream(&far, "rsync.server", &args, far_privileged, far_sudo.as_ref().map(|p| p.as_str())).await?;

1103 let (errors, errors_job) = (service.clone(), job.clone());

1104

                bridge::serve_relay(bridge_stream, far_stream, move |text| errors.append_output(&errors_job, &text), leftover).await

1105 }

1106 .await

1107 })

1108 }

1109

1110 // --- versions ------------------------------------------------------------------------------------

1111

1112 pub async fn versions(&self, payload: &Value) -> CoreResult<Value> {

1113 let base = spec::normalize(payload["path"].as_str().unwrap_or(""));

1114 if !base.starts_with('/') {

1115 return Err(job_error("RSYNC_SPEC_INVALID", "The path must be absolute."));

1116 }

1117 let mut versions = Vec::new();

1118 let mut latest = Value::Null;

1119 if payload["kind"] == "local" {

1120 let checked = local::checked_path(&base, &self.protected)?;

1121 for (kind, directory) in [(spec::VERSIONS_DIR, checked.join(spec::VERSIONS_DIR)), (".", checked.clone())] {

1122 for name in local_stamps(&directory) {

1123

                    versions.push(json!({ "kind": kind, "name": name, "path": format!("{}/{name}", directory.display()) }));

1124 }

1125 }

1126 if let Ok(target) = std::fs::read_link(checked.join("latest")) {

1127 latest = json!(target.to_string_lossy());

1128 }

1129 } else {

1130 let session_id = payload["sessionId"].as_str().unwrap_or("");

1131

            let output = self.operation(session_id, "rsync.versions.list", vec![base.clone()], payload["privileged"] == true, payload["sudoPassword"].as_str(), Duration::from_secs(60)).await?;

1132 for line in output.lines() {

1133 if let Some(rest) = line.strip_prefix("version=") {

1134 let parts: Vec<&str> = rest.split('|').collect();

1135 if let [kind, name, time] = parts[..] {

1136 let directory = if kind == "." { base.clone() } else { format!("{base}/{kind}") };

1137

                        versions.push(json!({ "kind": kind, "name": name, "time": time.parse::<u64>().unwrap_or(0), "path": format!("{directory}/{name}") }));

1138 }

1139 } else if let Some(rest) = line.strip_prefix("latest=") {

1140 latest = json!(rest);

1141 }

1142 }

1143 }

1144 versions.sort_by_key(|version| std::cmp::Reverse(version["name"].as_str().unwrap_or("").to_string()));

1145 Ok(json!({ "versions": versions, "latest": latest }))

1146 }

1147

1148 pub async fn prune_versions(&self, payload: &Value) -> CoreResult<Value> {

1149 if payload["confirm"] != true {

1150 return Err(job_error("RSYNC_CONFIRMATION_REQUIRED", "Removing old versions must be confirmed."));

1151 }

1152 let endpoint = match payload["kind"].as_str() {

1153

            Some("local") => Endpoint { place: Place::Local, path: payload["path"].as_str().unwrap_or("").to_string(), privileged: false },

1154

            _ => Endpoint { place: Place::Remote { session_id: payload["sessionId"].as_str().unwrap_or("").to_string() }, path: payload["path"].as_str().unwrap_or("").to_string(), privileged: payload["privileged"] == true },

1155 };

1156

        let secrets = Secrets { target_sudo: payload["sudoPassword"].as_str().filter(|p| !p.is_empty()).map(|p| Zeroizing::new(p.to_string())), ..Default::default() };

1157

        let output = self.prune(&endpoint, &secrets, payload["keepDays"].as_u64().unwrap_or(0).min(36_500), payload["keepCount"].as_u64().unwrap_or(0).min(100_000), None).await?;

1158 Ok(json!({ "output": output.trim() }))

1159 }

1160

1161 // --- profiles and history ------------------------------------------------------------------------

1162

1163 fn read_json(&self, name: &str) -> Value {

1164

        std::fs::read(self.root.join(name)).ok().and_then(|bytes| serde_json::from_slice(&bytes).ok()).unwrap_or_else(|| json!([]))

1165 }

1166

1167 fn write_json(&self, name: &str, value: &Value) -> CoreResult<()> {

1168 std::fs::create_dir_all(&self.root).map_err(|e| job_error("RSYNC_STORAGE_FAILED", &e.to_string()))?;

1169 let temporary = self.root.join(format!("{name}.tmp"));

1170

        std::fs::write(&temporary, serde_json::to_vec_pretty(value).unwrap_or_default()).map_err(|e| job_error("RSYNC_STORAGE_FAILED", &e.to_string()))?;

1171 std::fs::rename(&temporary, self.root.join(name)).map_err(|e| job_error("RSYNC_STORAGE_FAILED", &e.to_string()))

1172 }

1173

1174 pub fn profiles(&self) -> Value {

1175 self.read_json("profiles.json")

1176 }

1177

1178 /// A saved job keeps SSH profiles and options. Live sessions and every secret stay out of the file.

1179 pub fn save_profile(&self, payload: &Value) -> CoreResult<Value> {

1180 let name = payload["name"].as_str().unwrap_or("").trim().chars().take(120).collect::<String>();

1181 if name.is_empty() {

1182 return Err(job_error("RSYNC_SPEC_INVALID", "Enter the name of the profile."));

1183 }

1184 let mut clean = payload["spec"].clone();

1185 for side in ["source", "target"] {

1186 if let Some(endpoint) = clean[side].as_object_mut() {

1187 endpoint.remove("sessionId");

1188 endpoint.remove("sudoPassword");

1189

                if endpoint.get("kind").and_then(Value::as_str) == Some("ssh") && endpoint.get("profileId").and_then(Value::as_str).unwrap_or("").is_empty() {

1190

                    return Err(job_error("RSYNC_PROFILE_NEEDS_SAVED_SERVER", "A saved job needs servers with a saved SSH profile, because a manual login is not stored."));

1191 }

1192 }

1193 }

1194 if let Some(object) = clean.as_object_mut() {

1195 for secret in ["sourceSudo", "targetSudo", "directPassword"] {

1196 object.remove(secret);

1197 }

1198 }

1199 // The check needs some session to pass; profiles name theirs later.

1200 let mut check = clean.clone();

1201 for side in ["source", "target"] {

1202 if check[side]["kind"] == "ssh" {

1203 check[side]["sessionId"] = json!(format!("check-{side}"));

1204 }

1205 }

1206 spec::parse(&check)?;

1207 let schedule = payload["schedule"].clone();

1208 if !schedule.is_null() {

1209 schedule_forms(&schedule)?;

1210 }

1211 if payload["scheduleInApp"] == true && schedule["kind"] != "interval" {

1212

            return Err(job_error("RSYNC_SCHEDULE_INVALID", "A job scheduled inside the application runs in intervals. For a fixed time, use a server's timer."));

1213 }

1214

        let id = payload["id"].as_str().filter(|id| !id.is_empty()).map(str::to_string).unwrap_or_else(|| new_id("profile"));

1215 let mut list = self.profiles().as_array().cloned().unwrap_or_default();

1216 list.retain(|profile| profile["id"] != id.as_str());

1217

        let profile = json!({ "id": id, "name": name, "spec": clean, "schedule": schedule, "scheduleInApp": payload["scheduleInApp"] == true, "maxDeletes": payload["maxDeletes"].as_u64().unwrap_or(0), "saved": now() });

1218 list.push(profile.clone());

1219 self.write_json("profiles.json", &Value::Array(list))?;

1220 Ok(profile)

1221 }

1222

1223 pub fn delete_profile(&self, id: &str) -> CoreResult<Value> {

1224 let mut list = self.profiles().as_array().cloned().unwrap_or_default();

1225 list.retain(|profile| profile["id"] != id);

1226 self.write_json("profiles.json", &Value::Array(list))?;

1227 Ok(json!({ "deleted": true }))

1228 }

1229

1230 pub fn history(&self) -> Value {

1231 self.read_json("history.json")

1232 }

1233

1234 fn append_history(&self, entry: Value) -> CoreResult<()> {

1235 let mut list = self.history().as_array().cloned().unwrap_or_default();

1236 list.push(entry);

1237 if list.len() > MAX_HISTORY {

1238 let extra = list.len() - MAX_HISTORY;

1239 list.drain(..extra);

1240 }

1241 self.write_json("history.json", &Value::Array(list))

1242 }

1243

1244 pub fn clear_history(&self) -> CoreResult<Value> {

1245 self.write_json("history.json", &json!([]))?;

1246 Ok(json!({ "cleared": true }))

1247 }

1248

1249 /// From the File Manager to the Rsync Manager: the prepared job waits until the manager asks for it.

1250 pub fn put_handoff(&self, spec: Value) {

1251 self.inner.lock().handoff = Some(spec);

1252 }

1253

1254 pub fn take_handoff(&self) -> Value {

1255 self.inner.lock().handoff.take().unwrap_or(Value::Null)

1256 }

1257

1258 // --- schedules on servers ------------------------------------------------------------------------

1259

1260 /// A job on the server's own timer, independent of this application. Only what the server can do alone qualifies:

1261 /// both paths there, or a login it can make without a stored password.

1262 pub async fn schedule_install(&self, payload: &Value) -> CoreResult<Value> {

1263 if payload["confirm"] != true {

1264 return Err(job_error("RSYNC_CONFIRMATION_REQUIRED", "Installing a scheduled job must be confirmed."));

1265 }

1266 let resolved = self.resolve(&payload["spec"]).await?;

1267 let spec = spec::parse(&resolved)?;

1268 let (systemd, cron) = schedule_forms(&payload["schedule"])?;

1269 if spec.mode == Mode::Compare {

1270

            return Err(job_error("RSYNC_SCHEDULE_NOT_ON_SERVER", "A comparison transfers nothing, so it has no place on a timer."));

1271 }

1272 let Place::Remote { session_id } = &spec.source.place else {

1273

            return Err(job_error("RSYNC_SCHEDULE_NOT_ON_SERVER", "This job needs this computer, so it cannot run on a server's timer."));

1274 };

1275 let Place::Remote { session_id: other } = &spec.target.place else {

1276

            return Err(job_error("RSYNC_SCHEDULE_NOT_ON_SERVER", "This job needs this computer, so it cannot run on a server's timer."));

1277 };

1278 let (mut ssh_port, mut ssh_key, mut host_keys) = (String::new(), String::new(), String::new());

1279 let target_argument = spec::target_argument(&spec, "{stamp}");

1280 let one_server = session_id == other || self.same_login(session_id, other);

1281 if one_server {

1282 spec::check_overlap(&spec)?;

1283 let password = payload["sudoPassword"].as_str();

1284

            if let (Some(source_real), Some(target_real)) = (self.real_path(&spec.source, password).await, self.real_path(&spec.target, password).await) {

1285 let mut resolved = spec.clone();

1286 resolved.source.path = source_real;

1287 resolved.target.path = target_real;

1288 spec::check_overlap(&resolved)?;

1289 }

1290 }

1291 let target = if one_server {

1292 target_argument

1293

        } else if let Transport::Direct { host, port, username, method, key_path, accept_new_host_key } = &spec.transport {

1294 if method == "password" {

1295

                return Err(job_error("RSYNC_SCHEDULE_NEEDS_KEY", "A scheduled job cannot store a password on the server. Use a key."));

1296 }

1297 ssh_port = port.to_string();

1298 ssh_key = key_path.clone();

1299 host_keys = if *accept_new_host_key { "accept-new".into() } else { "yes".into() };

1300 let host = if host.contains(':') { format!("[{host}]") } else { host.clone() };

1301 format!("{username}@{host}:{target_argument}")

1302 } else {

1303

            return Err(job_error("RSYNC_SCHEDULE_NEEDS_DIRECT", "Between two servers, a scheduled job needs a direct login, because the transfer through this application stops with it."));

1304 };

1305 let capabilities = self.capabilities(&spec.source.place, &mut None).await?;

1306 let max_deletes = payload["maxDeletes"].as_u64().unwrap_or(0);

1307 let mut args = spec::flags(&spec, &capabilities, false, "{stamp}", Some(max_deletes))?;

1308 args.retain(|flag| !flag.starts_with("--info="));

1309 args.push(spec::source_argument(&spec));

1310 args.push(target);

1311

        let id = payload["id"].as_str().filter(|id| !id.is_empty()).map(str::to_string).unwrap_or_else(|| format!("{:x}", uuid::Uuid::new_v4().as_u128() >> 64));

1312 if !id.bytes().all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-') || id.len() > 40 {

1313 return Err(job_error("RSYNC_SPEC_INVALID", "The identifier of the scheduled job is not valid."));

1314 }

1315

        let label = payload["label"].as_str().unwrap_or("").replace(['\n', '\r'], " ").chars().take(120).collect::<String>();

1316 let base = spec::normalize(&spec.target.path);

1317

        let mut all = vec![id.clone(), label, systemd, cron, spec.mode.name().to_string(), if one_server { base } else { String::new() }, spec.options.keep_days.to_string(), spec.options.keep_count.to_string(), ssh_port, ssh_key, host_keys, spec::normalize(&spec.source.path)];

1318 all.extend(args);

1319 let options = RunOptions {

1320 operation: "rsync.schedule.install".into(),

1321 args: all,

1322 frames: vec![Frame { tag: "RUNNER".into(), data: matrix::script("rsync.sh").as_bytes().to_vec() }],

1323 privileged: true,

1324 sudo_password: payload["sudoPassword"].as_str().filter(|p| !p.is_empty()).map(str::to_string),

1325 use_cached_password: true,

1326 purpose: "rsync.schedule.install".into(),

1327 error_prefix: "RSYNC_".into(),

1328 timeout: Duration::from_secs(60),

1329 output_limit: 64 * 1024,

1330 ..Default::default()

1331 };

1332 self.executor.run(session_id, options).await?;

1333 self.logger.warn("rsync job scheduled on a server.", json!({ "sessionId": session_id, "jobId": id }));

1334 Ok(json!({ "id": id }))

1335 }

1336

1337 pub async fn schedule_list(&self, session_id: &str, payload: &Value) -> CoreResult<Value> {

1338

        let output = self.privileged(session_id, "rsync.schedule.list", vec![], payload["sudoPassword"].as_str(), Duration::from_secs(30)).await?;

1339 let mut jobs: Vec<Value> = Vec::new();

1340 for line in output.lines() {

1341 if let Some(id) = line.strip_prefix("\u{1e}CHJ_JOB=") {

1342 jobs.push(json!({ "id": id.trim() }));

1343 } else if let (Some(job), Some((key, value))) = (jobs.last_mut(), line.split_once('=')) {

1344 job[key] = json!(value);

1345 }

1346 }

1347 Ok(Value::Array(jobs))

1348 }

1349

1350 /// What else runs rsync on this server: cron, systemd, a daemon, a running process, a script. It only reads.

1351 pub async fn discover(&self, session_id: &str, payload: &Value) -> CoreResult<Value> {

1352

        let output = self.privileged(session_id, "rsync.discover", vec![], payload["sudoPassword"].as_str(), Duration::from_secs(60)).await?;

1353 Ok(Value::Array(parse_discovered(&output)))

1354 }

1355

1356 pub async fn schedule_action(&self, session_id: &str, action: &str, payload: &Value) -> CoreResult<Value> {

1357 let id = payload["id"].as_str().unwrap_or("").to_string();

1358 let operation = match action {

1359 "log" => "rsync.schedule.log",

1360 "start" => "rsync.schedule.start",

1361 "remove" => "rsync.schedule.remove",

1362 _ => return Err(job_error("RSYNC_SPEC_INVALID", "Unknown action.")),

1363 };

1364 if action != "log" && payload["confirm"] != true {

1365 return Err(job_error("RSYNC_CONFIRMATION_REQUIRED", "This action must be confirmed."));

1366 }

1367

        let output = self.privileged(session_id, operation, vec![id], payload["sudoPassword"].as_str(), Duration::from_secs(30)).await?;

1368 Ok(json!({ "output": output.trim() }))

1369 }

1370

1371 // --- schedules in this application ---------------------------------------------------------------

1372

1373 /// Runs jobs scheduled inside the application, while it runs, and closes connections nothing uses any more.

1374 pub fn start_scheduler(self: &Arc<Self>) {

1375 let service = self.clone();

1376 tokio::spawn(async move {

1377 let mut tick = tokio::time::interval(Duration::from_secs(30));

1378 loop {

1379 tick.tick().await;

1380 service.scheduler_tick().await;

1381 }

1382 });

1383 }

1384

1385 async fn scheduler_tick(self: &Arc<Self>) {

1386 let idle: Vec<String> = {

1387 let mut inner = self.inner.lock();

1388 let busy = !inner.running.is_empty();

1389

            let list: Vec<String> = if busy { vec![] } else { inner.opened_sessions.iter().filter(|(_, used)| used.elapsed() > IDLE_SESSION).map(|(id, _)| id.clone()).collect() };

1390 for id in &list {

1391 inner.opened_sessions.remove(id);

1392 }

1393 list

1394 };

1395 for id in idle {

1396 self.sessions.disconnect(&id, "rsync idle").await;

1397 }

1398 for profile in self.profiles().as_array().cloned().unwrap_or_default() {

1399 if profile["scheduleInApp"] != true || !self.due(&profile) {

1400 continue;

1401 }

1402 let id = profile["id"].as_str().unwrap_or("").to_string();

1403 self.inner.lock().last_app_runs.insert(id.clone(), Instant::now());

1404 if let Err(error) = self.run_unattended(&profile).await {

1405

                let _ = self.append_history(json!({ "id": new_id("job"), "kind": "run", "state": "failed", "label": profile["name"], "message": error.message, "errorCode": error.code, "finished": now() }));

1406 }

1407 }

1408 }

1409

1410 fn due(&self, profile: &Value) -> bool {

1411 let id = profile["id"].as_str().unwrap_or("");

1412 let minutes = profile["schedule"]["minutes"].as_u64().unwrap_or(0);

1413 if profile["schedule"]["kind"] != "interval" || minutes == 0 {

1414 return false;

1415 }

1416 // The interval counts from the start of the application, not from a run nobody remembers.

1417 let mut inner = self.inner.lock();

1418 let last = *inner.last_app_runs.entry(id.to_string()).or_insert_with(Instant::now);

1419 last.elapsed() >= Duration::from_secs(minutes * 60)

1420 }

1421

1422 /// Nobody is there to look at a preview: deleting stops at the limit saved with the profile.

1423 async fn run_unattended(self: &Arc<Self>, profile: &Value) -> CoreResult<()> {

1424 let resolved = self.resolve(&profile["spec"]).await?;

1425 let spec = spec::parse(&resolved)?;

1426 self.check_local_paths(&spec)?;

1427 if spec.mode == Mode::Compare {

1428 return Ok(());

1429 }

1430 if self.target_busy(&spec) {

1431 return Ok(());

1432 }

1433 if matches!(spec.mode, Mode::Mirror | Mode::Versioned) && spec::is_protected(&spec.target.path) {

1434

            return Err(job_error("RSYNC_TARGET_PROTECTED", "A scheduled mirror never deletes in a system or home directory."));

1435 }

1436 let id = new_id("job");

1437 self.register_job(&id, "run", &spec, profile["name"].as_str().unwrap_or(""), "scheduler");

1438 self.touch(&spec);

1439 let limit = profile["maxDeletes"].as_u64().unwrap_or(0);

1440 self.spawn(id, spec, Secrets::default(), false, Some(limit), false, resolved);

1441 Ok(())

1442 }

1443

1444}

1445

1446/// A private place for the link to the helper, removed with the transfer.

1447struct LinkDirectory(PathBuf);

1448

1449impl LinkDirectory {

1450 fn create() -> std::io::Result<Self> {

1451 let path = std::env::temp_dir().join(new_id("chj-rsync"));

1452 std::fs::create_dir(&path)?;

1453 #[cfg(unix)]

1454 {

1455 use std::os::unix::fs::PermissionsExt;

1456 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700))?;

1457 }

1458 Ok(LinkDirectory(path))

1459 }

1460}

1461

1462impl Drop for LinkDirectory {

1463 fn drop(&mut self) {

1464 let _ = std::fs::remove_dir_all(&self.0);

1465 }

1466}

1467

1468/// On one server a single sudo password serves both paths.

1469fn target_password<'a>(spec: &Spec, secrets: &'a Secrets) -> Option<&'a str> {

1470 let shared = if spec.source.place == spec.target.place { secrets.source_sudo.as_ref() } else { None };

1471 secrets.target_sudo.as_ref().or(shared).map(|p| p.as_str())

1472}

1473

1474fn describe_endpoint(endpoint: &Endpoint) -> Value {

1475 match &endpoint.place {

1476 Place::Local => json!({ "kind": "local", "path": endpoint.path }),

1477 Place::Remote { session_id } => json!({ "kind": "ssh", "sessionId": session_id, "path": endpoint.path }),

1478 }

1479}

1480

1481/// The nearest existing folder resolved, plus what does not exist yet.

1482fn real_local(path: &Path) -> String {

1483 let mut existing = path.to_path_buf();

1484 let mut rest = Vec::new();

1485 while !existing.exists() {

1486 match (existing.file_name().map(|name| name.to_os_string()), existing.parent().map(Path::to_path_buf)) {

1487 (Some(name), Some(parent)) => {

1488 rest.push(name);

1489 existing = parent;

1490 }

1491 _ => break,

1492 }

1493 }

1494 let mut real = std::fs::canonicalize(&existing).unwrap_or(existing);

1495 for name in rest.iter().rev() {

1496 real.push(name);

1497 }

1498 real.to_string_lossy().to_string()

1499}

1500

1501fn is_stamp(name: &str) -> bool {

1502 let bytes = name.as_bytes();

1503

    bytes.len() == 16 && bytes[8] == b'T' && bytes[15] == b'Z' && bytes[..8].iter().chain(&bytes[9..15]).all(u8::is_ascii_digit)

1504}

1505

1506fn local_stamps(directory: &Path) -> Vec<String> {

1507

    let mut names: Vec<String> = std::fs::read_dir(directory).into_iter().flatten().flatten().map(|entry| entry.file_name().to_string_lossy().to_string()).filter(|name| is_stamp(name)).collect();

1508 names.sort();

1509 names

1510}

1511

1512#[cfg(unix)]

1513fn make_link(original: &Path, link: &Path) -> std::io::Result<()> {

1514 std::os::unix::fs::symlink(original, link)

1515}

1516

1517#[cfg(not(unix))]

1518fn make_link(original: &Path, link: &Path) -> std::io::Result<()> {

1519 std::fs::copy(original, link).map(|_| ())

1520}

1521

1522fn point_latest_local(base: &Path, run_stamp: &str) -> CoreResult<()> {

1523 use std::io::Write;

1524 let latest = base.join("latest");

1525 if std::fs::symlink_metadata(&latest).is_ok_and(|meta| !meta.file_type().is_symlink()) {

1526 return Err(job_error("RSYNC_SNAPSHOT_FAILED", "The target contains a real entry named latest."));

1527 }

1528

    let mut record = std::fs::OpenOptions::new().create(true).append(true).open(base.join(".chj-snapshots")).map_err(|e| job_error("RSYNC_SNAPSHOT_FAILED", &e.to_string()))?;

1529 writeln!(record, "{run_stamp}").map_err(|e| job_error("RSYNC_SNAPSHOT_FAILED", &e.to_string()))?;

1530 let temporary = base.join(format!(".latest.{}", std::process::id()));

1531 make_link(Path::new(run_stamp), &temporary).map_err(|e| job_error("RSYNC_SNAPSHOT_FAILED", &e.to_string()))?;

1532 std::fs::rename(&temporary, base.join("latest")).map_err(|e| job_error("RSYNC_SNAPSHOT_FAILED", &e.to_string()))

1533}

1534

1535/// Which of the stamped names to remove: never more than leaves `keep_count`, never newer than `keep_days` allow.

1536

pub fn prune_choice(names: &[String], keep_days: u64, keep_count: u64, keep_name: Option<&str>, today: chrono::DateTime<chrono::Utc>) -> Vec<String> {

1537 let limit = (today - chrono::Duration::days(keep_days as i64)).format("%Y%m%dT%H%M%SZ").to_string();

1538 let mut removable = names.len().saturating_sub(keep_count as usize);

1539 let mut chosen = Vec::new();

1540 for name in names {

1541 if removable == 0 {

1542 break;

1543 }

1544 if Some(name.as_str()) == keep_name {

1545 continue;

1546 }

1547 if keep_days == 0 || name.as_str() < limit.as_str() {

1548 chosen.push(name.clone());

1549 removable -= 1;

1550 }

1551 }

1552 chosen

1553}

1554

1555fn prune_local(base: &Path, keep_days: u64, keep_count: u64, kinds: &[&str]) -> CoreResult<String> {

1556 let mut output = String::new();

1557 // Through a symbolic link anywhere on the way, the removal would reach another place.

1558

    let plain_directory = |path: &Path| std::fs::symlink_metadata(path).is_ok_and(|meta| meta.is_dir()) && std::fs::canonicalize(path).is_ok_and(|real| real == path);

1559 if !plain_directory(base) {

1560 return Ok(output);

1561 }

1562 let mut remove = |directory: &Path, keep: Option<&str>, owned: Option<&Vec<String>>| -> CoreResult<()> {

1563 // Through a symbolic link, the removal would reach another place.

1564 if !plain_directory(directory) {

1565 return Ok(());

1566 }

1567

        let names: Vec<String> = local_stamps(directory).into_iter().filter(|name| owned.is_none_or(|list: &Vec<String>| list.contains(name))).collect();

1568 for name in prune_choice(&names, keep_days, keep_count, keep, chrono::Utc::now()) {

1569 let entry = directory.join(&name);

1570 if plain_directory(&entry) {

1571 std::fs::remove_dir_all(entry).map_err(|e| job_error("RSYNC_PRUNE_FAILED", &e.to_string()))?;

1572 output.push_str(&format!("removed {name}\n"));

1573 }

1574 }

1575 Ok(())

1576 };

1577 if kinds.contains(&"versions") {

1578 remove(&base.join(spec::VERSIONS_DIR), None, None)?;

1579 }

1580 if kinds.contains(&"snapshots") {

1581 // A folder of the data can look like a time too: only what this application recorded is a snapshot.

1582 if let Ok(latest) = std::fs::read_link(base.join("latest")) {

1583 let latest = latest.to_string_lossy().to_string();

1584

            let owned: Vec<String> = std::fs::read_to_string(base.join(".chj-snapshots")).unwrap_or_default().lines().map(str::to_string).collect();

1585 if is_stamp(&latest) {

1586 remove(base, Some(&latest), Some(&owned))?;

1587 }

1588 }

1589 }

1590 Ok(output)

1591}

1592

1593/// A command line may carry a password, and it goes no further than this.

1594fn hide_secrets(line: &str) -> String {

1595 static SECRETS: std::sync::LazyLock<regex::Regex> = std::sync::LazyLock::new(|| {

1596

        regex::Regex::new(r#"(?i)(\b[a-z0-9_]*(?:pass|passwd|password|token|secret|api_?key)[a-z0-9_]*=|--password=|sshpass\s+-p\s*)(?:"[^"]*"|'[^']*'|\S+)|(://[^/\s:@]+:)[^/\s@]+@"#).unwrap()

1597 });

1598 SECRETS.replace_all(line, |found: &regex::Captures| match (found.get(1), found.get(2)) {

1599 (Some(name), _) => format!("{}***", name.as_str()),

1600 (_, Some(user)) => format!("{}***@", user.as_str()),

1601 _ => "***".to_string(),

1602 }).into_owned()

1603}

1604

1605pub fn parse_discovered(output: &str) -> Vec<Value> {

1606 // On Alpine /etc/crontabs is a link to the spool, so one line is found under both names.

1607 let mut seen = std::collections::HashSet::new();

1608 output.lines().filter_map(|line| line.strip_prefix("\u{1e}CHJ_FOUND=")).filter_map(|line| {

1609 let mut fields = line.splitn(3, '\t');

1610 let kind = fields.next()?;

1611 let place = fields.next()?;

1612 let detail = hide_secrets(fields.next().unwrap_or(""));

1613

        (kind != "cron" || seen.insert(detail.clone())).then(|| json!({ "kind": kind, "place": place, "detail": detail }))

1614 }).collect()

1615}

1616

1617#[cfg(test)]

1618mod tests {

1619 #[test]

1620 fn found_rsync_is_listed_without_its_secrets() {

1621

        let output = "\u{1e}CHJ_READY\nnoise\n\u{1e}CHJ_FOUND=cron\t/etc/cron.d/mirror\t0 3 * * * root RSYNC_PASSWORD=abc rsync -a /a b@c::d/\n\u{1e}CHJ_FOUND=running\t-\t\n";

1622 let list = super::parse_discovered(output);

1623 assert_eq!(list.len(), 2);

1624 assert_eq!(list[0]["kind"], "cron");

1625 assert_eq!(list[0]["place"], "/etc/cron.d/mirror");

1626 assert_eq!(list[0]["detail"], "0 3 * * * root RSYNC_PASSWORD=*** rsync -a /a b@c::d/");

1627

        assert_eq!(super::parse_discovered("\u{1e}CHJ_FOUND=cron\t/etc/crontabs/root\t1 2 * * * rsync\n\u{1e}CHJ_FOUND=cron\t/var/spool/cron/crontabs/root\t1 2 * * * rsync\n").len(), 1);

1628 for (line, wanted) in [

1629 ("sshpass -p 'a b' rsync -e ssh /x y:/z", "sshpass -p *** rsync -e ssh /x y:/z"),

1630 ("rsync --password=abc /x y", "rsync --password=*** /x y"),

1631 ("rsync --password-file=/etc/rs.pw /x rsync://u@h/m", "rsync --password-file=/etc/rs.pw /x rsync://u@h/m"),

1632 ("rsync /x rsync://user:abc@host/mod", "rsync /x rsync://user:***@host/mod"),

1633 ("MY_TOKEN=\"a b\" rsync -a /x /y", "MY_TOKEN=*** rsync -a /x /y"),

1634 ] {

1635 assert_eq!(super::hide_secrets(line), wanted);

1636 }

1637 }

1638

1639 use super::*;

1640

1641 fn names(list: &[&str]) -> Vec<String> {

1642 list.iter().map(|name| name.to_string()).collect()

1643 }

1644

1645 #[test]

1646 fn schedules_become_both_forms() {

1647

        assert_eq!(schedule_forms(&json!({ "kind": "interval", "minutes": 30 })).unwrap(), ("OnBootSec=5min;OnUnitActiveSec=30min".to_string(), "*/30 * * * *".to_string()));

1648 assert_eq!(schedule_forms(&json!({ "kind": "interval", "minutes": 120 })).unwrap().1, "0 */2 * * *");

1649 assert_eq!(schedule_forms(&json!({ "kind": "interval", "minutes": 1440 })).unwrap().1, "0 0 * * *");

1650

        assert_eq!(schedule_forms(&json!({ "kind": "daily", "time": "03:05" })).unwrap(), ("OnCalendar=*-*-* 03:05:00".to_string(), "5 3 * * *".to_string()));

1651

        assert_eq!(schedule_forms(&json!({ "kind": "weekly", "time": "23:59", "day": 1 })).unwrap(), ("OnCalendar=Mon *-*-* 23:59:00".to_string(), "59 23 * * 1".to_string()));

1652 }

1653

1654 #[test]

1655 fn schedules_cannot_inject_anything() {

1656

        for schedule in [json!({ "kind": "interval", "minutes": 7 }), json!({ "kind": "daily", "time": "25:00" }), json!({ "kind": "daily", "time": "1:1; rm" }), json!({ "kind": "weekly", "time": "10:00", "day": 9 }), json!({ "kind": "x" }), json!(null)] {

1657 assert_eq!(schedule_forms(&schedule).unwrap_err().code, "RSYNC_SCHEDULE_INVALID", "{schedule}");

1658 }

1659 }

1660

1661 #[test]

1662 fn a_real_entry_named_latest_is_not_overwritten() {

1663 let dir = tempfile::tempdir().unwrap();

1664 std::fs::create_dir(dir.path().join("latest")).unwrap();

1665 assert_eq!(point_latest_local(dir.path(), "20200101T000000Z").unwrap_err().code, "RSYNC_SNAPSHOT_FAILED");

1666 assert!(dir.path().join("latest").is_dir());

1667 }

1668

1669 #[test]

1670 fn stamps_are_recognized_exactly() {

1671 assert!(is_stamp("20261102T010203Z"));

1672 for name in ["latest", "20261102T010203", "2026110xT010203Z", ".partial", "../20261102T010203Z"] {

1673 assert!(!is_stamp(name), "{name}");

1674 }

1675 }

1676

1677 #[test]

1678 fn pruning_keeps_the_count_and_the_latest() {

1679 let today = chrono::DateTime::parse_from_rfc3339("2026-11-02T00:00:00Z").unwrap().with_timezone(&chrono::Utc);

1680 let all = names(&["20261001T000000Z", "20261015T000000Z", "20261101T000000Z"]);

1681 assert_eq!(prune_choice(&all, 0, 1, None, today), names(&["20261001T000000Z", "20261015T000000Z"]));

1682 assert_eq!(prune_choice(&all, 7, 0, None, today), names(&["20261001T000000Z", "20261015T000000Z"]));

1683 assert_eq!(prune_choice(&all, 30, 0, None, today), names(&["20261001T000000Z"]));

1684 assert_eq!(prune_choice(&all, 0, 3, None, today), Vec::<String>::new());

1685 assert_eq!(prune_choice(&all, 0, 2, Some("20261001T000000Z"), today), names(&["20261015T000000Z"]));

1686 assert_eq!(prune_choice(&[], 0, 0, None, today), Vec::<String>::new());

1687 }

1688

1689 #[test]

1690 fn local_pruning_touches_only_stamped_directories() {

1691 let dir = tempfile::tempdir().unwrap();

1692 let base = std::fs::canonicalize(dir.path()).unwrap();

1693 for name in ["20200101T000000Z", "20200102T000000Z", "20200103T000000Z", "keep-me"] {

1694 std::fs::create_dir(base.join(name)).unwrap();

1695 }

1696 make_link(Path::new("20200101T000000Z"), &base.join("latest")).unwrap();

1697 std::fs::write(base.join(".chj-snapshots"), "20200101T000000Z\n20200102T000000Z\n").unwrap();

1698 let output = prune_local(&base, 0, 0, &["versions", "snapshots"]).unwrap();

1699 assert_eq!(output, "removed 20200102T000000Z\n");

1700 assert!(base.join("keep-me").exists() && base.join("20200101T000000Z").exists());

1701 assert!(base.join("20200103T000000Z").exists(), "not recorded as a snapshot");

1702 }

1703

1704 #[test]

1705 fn data_that_looks_like_a_version_stays() {

1706 let dir = tempfile::tempdir().unwrap();

1707 let base = std::fs::canonicalize(dir.path()).unwrap();

1708 std::fs::create_dir(base.join("20200101T000000Z")).unwrap();

1709 std::fs::create_dir_all(base.join(".chj-versions/20200102T000000Z")).unwrap();

1710 // Without a record nothing proves that the folder is a snapshot.

1711 assert_eq!(prune_local(&base, 0, 0, &["versions", "snapshots"]).unwrap(), "removed 20200102T000000Z\n");

1712 assert!(base.join("20200101T000000Z").exists());

1713 }

1714

1715 #[test]

1716 #[cfg(unix)]

1717 fn pruning_does_not_follow_a_link_out_of_the_backup() {

1718 let outside = tempfile::tempdir().unwrap();

1719 std::fs::create_dir(outside.path().join("20200101T000000Z")).unwrap();

1720 let dir = tempfile::tempdir().unwrap();

1721 let base = std::fs::canonicalize(dir.path()).unwrap();

1722 make_link(outside.path(), &base.join(".chj-versions")).unwrap();

1723 assert_eq!(prune_local(&base, 0, 0, &["versions"]).unwrap(), "");

1724 // A link further up the path counts as well.

1725 std::fs::create_dir_all(base.join("real/.chj-versions/20200101T000000Z")).unwrap();

1726 make_link(&base.join("real"), &base.join("alias")).unwrap();

1727 assert_eq!(prune_local(&base.join("alias"), 0, 0, &["versions"]).unwrap(), "");

1728 assert!(base.join("real/.chj-versions/20200101T000000Z").exists());

1729 assert!(outside.path().join("20200101T000000Z").exists());

1730 }

1731

1732 #[test]

1733 fn profiles_store_no_session_and_no_secret() {

1734 let dir = tempfile::tempdir().unwrap();

1735 let root = dir.path().to_path_buf();

1736 let vault = Arc::new(crate::core::vault::VaultStore::new(&root));

1737 let profiles = Arc::new(crate::core::profiles::ProfileService::new(vault));

1738 let logger = Arc::new(Logger::new(&root));

1739 let sessions = SessionManager::new(profiles, logger.clone(), Arc::new(|_: &str, _: Value| {}));

1740 let executor = Arc::new(PrivilegedExecutor::new(sessions.clone()));

1741 let service = RsyncService::new(sessions, executor, logger, &root, PathBuf::from("/bin/false"));

1742

        let saved = service.save_profile(&json!({ "name": "a", "spec": { "mode": "copy", "sourceSudo": "secret-1", "directPassword": "secret-2", "source": { "kind": "local", "path": "/data" }, "target": { "kind": "ssh", "profileId": "p1", "sessionId": "live", "sudoPassword": "secret-3", "path": "/backup" } } })).unwrap();

1743 let text = std::fs::read_to_string(root.join("rsync/profiles.json")).unwrap();

1744 assert!(!text.contains("secret-") && !text.contains("live"), "{text}");

1745 assert_eq!(saved["spec"]["target"]["profileId"], "p1");

1746

        assert_eq!(service.save_profile(&json!({ "name": "b", "spec": { "source": { "kind": "local", "path": "/a" }, "target": { "kind": "ssh", "sessionId": "manual", "path": "/b" } } })).unwrap_err().code, "RSYNC_PROFILE_NEEDS_SAVED_SERVER");

1747 assert_eq!(service.save_profile(&json!({ "name": "", "spec": {} })).unwrap_err().code, "RSYNC_SPEC_INVALID");

1748 assert_eq!(service.profiles().as_array().unwrap().len(), 1);

1749 service.delete_profile(saved["id"].as_str().unwrap()).unwrap();

1750 assert_eq!(service.profiles().as_array().unwrap().len(), 0);

1751 }

1752

1753 #[test]

1754 fn a_run_needs_the_numbers_of_its_preview() {

1755 let dir = tempfile::tempdir().unwrap();

1756 let root = dir.path().to_path_buf();

1757 let vault = Arc::new(crate::core::vault::VaultStore::new(&root));

1758 let profiles = Arc::new(crate::core::profiles::ProfileService::new(vault));

1759 let logger = Arc::new(Logger::new(&root));

1760 let sessions = SessionManager::new(profiles, logger.clone(), Arc::new(|_: &str, _: Value| {}));

1761 let executor = Arc::new(PrivilegedExecutor::new(sessions.clone()));

1762 let service = RsyncService::new(sessions, executor, logger, &root, PathBuf::from("/bin/false"));

1763 let plan = |target: &str, deletes: u64, empty: bool| Plan {

1764

            spec: spec::parse(&json!({ "mode": "mirror", "source": { "kind": "local", "path": "/a" }, "target": { "kind": "local", "path": target } })).unwrap(),

1765

            secrets: Secrets::default(), deletes, empty_source: empty, protected_target: spec::is_protected(target), created: Instant::now(), source_value: json!({}),

1766 };

1767 service.inner.lock().plans.insert("p".into(), plan("/srv/b", 3, false));

1768 assert_eq!(service.run("o", &json!({ "planId": "p" })).unwrap_err().code, "RSYNC_CONFIRMATION_REQUIRED");

1769

        assert_eq!(service.run("o", &json!({ "planId": "p", "confirm": true })).unwrap_err().code, "RSYNC_DELETE_NOT_ACCEPTED");

1770

        assert_eq!(service.run("o", &json!({ "planId": "p", "confirm": true, "acceptDeletes": 2 })).unwrap_err().code, "RSYNC_DELETE_NOT_ACCEPTED");

1771 service.inner.lock().plans.insert("q".into(), plan("/home", 1, false));

1772

        assert_eq!(service.run("o", &json!({ "planId": "q", "confirm": true, "acceptDeletes": 1 })).unwrap_err().code, "RSYNC_TARGET_PROTECTED");

1773

        assert_eq!(service.run("o", &json!({ "planId": "q", "confirm": true, "acceptDeletes": 1, "confirmProtected": "/etc" })).unwrap_err().code, "RSYNC_TARGET_PROTECTED");

1774 service.inner.lock().plans.insert("r".into(), plan("/srv/c", 5, true));

1775

        assert_eq!(service.run("o", &json!({ "planId": "r", "confirm": true, "acceptDeletes": 5 })).unwrap_err().code, "RSYNC_SOURCE_EMPTY");

1776

        assert_eq!(service.run("o", &json!({ "planId": "none", "confirm": true })).unwrap_err().code, "RSYNC_PLAN_EXPIRED");

1777 }

1778}

1779

SHA-256

7518965f7ed4bbd701f9158c2b1420df5d25a2e6cdf52819b12e27da3503909d

CH-J Proprietary Software License 1.14 · Die Quellen werden unter der CH-J Proprietary Software License 1.14 bereitgestellt. Ihre Verfügbarkeit ändert die Lizenzbedingungen nicht und gewährt keine zusätzlichen Rechte.

SHA-256 des Archivs

ac8fe790f4ba60fc098f8c985fc2656cfcc6cdc21fb34b30e308a2b54e62258f