1use super::*;
2use std::{
3 fs::File,
4 io::{Read, Write},
5 os::unix::ffi::OsStrExt,
6};
7impl RemoteServices {
8 pub(crate) fn source_bytes(
9 &self,
10 source: &ByteSource,
11 maximum: usize,
12 verify: bool,
13 ) -> Result<Vec<u8>, CoreError> {
14 buntrail_services::files::document_io::source_bytes(&self.spools, source, maximum, verify)
15 }
16 pub fn load_snapshot(
17 &self,
18 snapshot: FileSnapshot,
19 document: DocumentID,
20 recovery: Option<&buntrail_services::recovery::store::RecoveryStore>,
21 ) -> Result<
22 (
23 buntrail_core::documents::LoadedDocument,
24 buntrail_services::files::LoadBudget,
25 ),
26 CoreError,
27 > {
28 self.check(&snapshot.file.target)?;
29 buntrail_services::files::document_io::load_snapshot(
30 snapshot,
31 document,
32 recovery,
33 &self.spools,
34 )
35 }
36 pub fn save_prepared(
37 &self,
38 recovery: &buntrail_services::recovery::store::RecoveryStore,
39 prepared: buntrail_core::documents::PreparedSave,
40 ) -> OperationReceipt {
41 let operation = prepared.operation;
42 let target = prepared.recovery.file.target.clone();
43 if let Err(error) = self.check(&target) {
44 return buntrail_services::files::failure(operation, target, error);
45 }
46 match buntrail_services::files::document_io::prepare_save(recovery, prepared) {
47 Ok((request, _budget)) => self.upload(request, false),
48 Err(error) => buntrail_services::files::failure(operation, target, error),
49 }
50 }
51 fn read_sync(&self, mut file: FileRef) -> Result<FileSnapshot, CoreError> {
52 self.check(&file.target)?;
53 let mut request = self.transport.start(
54 Message::Read {
55 path_base64: w::path(&file.path)?,
56 },
57 None,
58 None,
59 )?;
60 let begin = request.next()?;
61 let Message::ReadBegin {
62 transfer_id,
63 version,
64 total_length,
65 path_base64,
66 } = begin.header.message
67 else {
68 return Err(response_error(begin.header.message));
69 };
70 if path_base64 != w::path(&file.path)?
71 || begin.header.stream_id.as_deref() != Some(&transfer_id)
72 {
73 return Err(w::invalid());
74 }
75 if total_length > buntrail_services::files::MAX_ENCODED_BYTES as u64 {
76 return Err(CoreError::new(
77 ErrorCode::Unavailable,
78 "Remote file exceeds bounded spool capacity",
79 ));
80 }
81 let version: buntrail_protocol::types::WireVersion =
82 serde_json::from_str(&version).map_err(io_error)?;
83 let version = version.core(&self.target)?;
84 let (mut spool, path, lease) = self.spools.create_spool()?;
85 let mut verifier =
86 buntrail_protocol::frame::TransferVerifier::new(transfer_id.clone(), total_length)
87 .map_err(|_| w::invalid())?;
88 let mut prefix = Vec::new();
89 loop {
90 let f = request.next()?;
91 if f.header.stream_id.as_deref() != Some(&transfer_id) {
92 return Err(w::invalid());
93 }
94 match f.header.message {
95 Message::ReadChunk {
96 transfer_id: ref tid,
97 offset,
98 } => {
99 verifier
100 .chunk(tid, offset, &f.payload)
101 .map_err(|_| w::invalid())?;
102 prefix.extend_from_slice(&f.payload[..f.payload.len().min(4 - prefix.len())]);
103 spool.write_all(&f.payload).map_err(io_error)?;
104 request.credit(&transfer_id, f.payload.len())?;
105 }
106 Message::ReadEnd {
107 transfer_id: tid,
108 sha256,
109 } => {
110 let digest = w::digest(&sha256)?;
111 verifier.end(&tid, digest).map_err(|_| w::invalid())?;
112 if digest != version.content_digest {
113 return Err(w::invalid());
114 }
115 request.finish();
116 break;
117 }
118 m => return Err(response_error(m)),
119 }
120 }
121 spool.sync_all().map_err(io_error)?;
122 file.resolved_identity = Some(version.file_identity.clone());
123 let encoding = if prefix.starts_with(&[0xff, 0xfe]) {
124 Encoding::Other("UTF-16LE".into())
125 } else if prefix.starts_with(&[0xfe, 0xff]) {
126 Encoding::Other("UTF-16BE".into())
127 } else {
128 Encoding::Utf8 {
129 bom: prefix.starts_with(&[0xef, 0xbb, 0xbf]),
130 }
131 };
132 Ok(FileSnapshot {
133 file,
134 version: version.clone(),
135 size: total_length,
136 encoding,
137 source: ByteSource::PrivateSpool {
138 local_path: TargetPath::posix(path.as_os_str().as_bytes().to_vec())?,
139 length: total_length,
140 digest: version.content_digest,
141 lease: Some(lease),
142 },
143 })
144 }
145 fn upload(&self, r: SaveRequest, create: bool) -> OperationReceipt {
146 let id = r.operation;
147 let target = r.file.target.clone();
148 let mut commit_submitted = false;
149 let outcome = (|| -> Result<OperationReceipt, CoreError> {
150 self.check(&r.file.target)?;
151 if !create
152 && !self
153 .negotiated_capabilities()
154 .iter()
155 .any(|cap| cap == "file.save_proof")
156 {
157 return Err(CoreError::new(
158 ErrorCode::Unavailable,
159 "active helper lacks authenticated save provenance; preserve its existing sessions",
160 ));
161 }
162 if let Some(v) = &r.expected_version {
163 self.check(&v.target)?;
164 }
165 let (mut input, length): (Box<dyn Read>, u64) = match &r.source {
166 ByteSource::Memory(b) => {
167 (Box::new(std::io::Cursor::new(b.clone())), b.len() as u64)
168 }
169 ByteSource::PrivateSpool {
170 local_path,
171 length,
172 lease,
173 ..
174 } => {
175 let TargetPath::PosixBytes(bytes) = local_path else {
176 return Err(w::invalid());
177 };
178 let path = std::path::Path::new(std::ffi::OsStr::from_bytes(bytes));
179 if lease.is_none() || self.spools.retain(path).is_none() {
180 return Err(CoreError::new(
181 ErrorCode::Unavailable,
182 "Upload source is not retained by this connection",
183 ));
184 }
185 (Box::new(File::open(path).map_err(io_error)?), *length)
186 }
187 };
188 let transfer = w::uuid(buntrail_services::connections::local::new_id()?);
189 let mut request = self.transport.start(
190 Message::UploadBegin {
191 transfer_id: transfer.clone(),
192 path_base64: w::path(&r.file.path)?,
193 total_length: length,
194 },
195 Some(id),
196 Some(transfer.clone()),
197 )?;
198 match request.next()?.header.message {
199 Message::Ack {} => {}
200 message => return Err(response_error(message)),
201 }
202 let mut offset = 0;
203 let mut hash = sha2::Sha256::new();
204 let mut chunk = [0; 65536];
205 while offset < length {
206 let n = (length - offset).min(65536) as usize;
207 input.read_exact(&mut chunk[..n]).map_err(io_error)?;
208 hash.update(&chunk[..n]);
209 request.send(
210 Message::UploadChunk {
211 transfer_id: transfer.clone(),
212 offset,
213 },
214 chunk[..n].to_vec(),
215 )?;
216 match request.next()?.header.message {
217 Message::Credit { bytes } if bytes as usize == n => {}
218 m => return Err(response_error(m)),
219 }
220 offset += n as u64;
221 }
222 if input.read(&mut [0; 1]).map_err(io_error)? != 0 {
223 return Err(w::invalid());
224 }
225 let digest = w::hex(&hash.finalize());
226 if let ByteSource::PrivateSpool {
227 digest: expected, ..
228 } = &r.source
229 && digest != w::hex(expected)
230 {
231 return Err(w::invalid());
232 }
233 request.send(
234 Message::UploadEnd {
235 transfer_id: transfer.clone(),
236 sha256: digest,
237 },
238 vec![],
239 )?;
240 match request.next()?.header.message {
241 Message::Ack {} => {}
242 message => return Err(response_error(message)),
243 }
244 request.send(
245 if create {
246 Message::CreateCommit {
247 transfer_id: transfer,
248 }
249 } else {
250 Message::SaveCommitOwned {
251 transfer_id: transfer,
252 lease_id: w::uuid(self.lease.lease.0),
253 submitted_revision: r.submitted_revision.0,
254 expected_version: r
255 .expected_version
256 .as_ref()
257 .map(|v| serde_json::to_string(&w::version(v)).map_err(io_error))
258 .transpose()?,
259 }
260 },
261 vec![],
262 )?;
263 commit_submitted = true;
264 if create {
265 let mut replies = Replies {
266 request,
267 stream_id: None,
268 bytes: Vec::new(),
269 length: None,
270 hash: None,
271 sequence: 0,
272 done: false,
273 };
274 loop {
275 let f = replies.request.next()?;
276 if matches!(f.header.message, Message::Receipt { .. }) {
277 let receipt = legacy_receipt(id, &target, f.header.message)?;
278 replies.request.finish();
279 return Ok(receipt);
280 }
281 if let Some(result) = replies.accept(f)? {
282 return match result? {
283 w::Reply::Receipt { receipt } => receipt.core(&target),
284 _ => Err(w::invalid()),
285 };
286 }
287 }
288 }
289 let message = request.next()?.header.message;
290 request.finish();
291 legacy_receipt(id, &target, message)
292 })();
293 outcome.unwrap_or_else(|error| {
294 if commit_submitted {
295 super::operations::uncertain(id, target, error)
296 } else {
297 buntrail_services::files::failure(id, target, error)
298 }
299 })
300 }
301}
302fn response_error(m: Message) -> CoreError {
303 match m {
304 Message::Error { code, safe_message } => w::core_error(&code, &safe_message),
305 _ => w::invalid(),
306 }
307}
308pub(super) fn legacy_receipt(
309 id: OperationID,
310 t: &TargetIdentity,
311 m: Message,
312) -> Result<OperationReceipt, CoreError> {
313 use buntrail_protocol::types::{ReceiptState as S, WireEffect, WireObservation as O};
314 let Message::Receipt {
315 state,
316 saved_version,
317 saved_revision,
318 safe_message,
319 failure,
320 observed,
321 } = m
322 else {
323 return Err(response_error(m));
324 };
325 let observed = match observed {
326 O::Unobserved => ObservedState::Unobserved,
327 O::File {
328 path_base64,
329 version,
330 digest,
331 } => ObservedState::File {
332 file: FileRef {
333 target: t.clone(),
334 path: w::target_path(&path_base64)?,
335 resolved_identity: None,
336 },
337 version: version.as_ref().map(|v| v.core(t)).transpose()?,
338 digest: digest.as_deref().map(w::digest).transpose()?,
339 },
340 };
341 Ok(OperationReceipt {
342 id,
343 state: match state {
344 S::Accepted => OperationState::Accepted,
345 S::Running => OperationState::Running,
346 S::Confirmed => OperationState::Confirmed(MutationResult::Saved(SaveResult {
347 saved_revision: BufferRevision(saved_revision.ok_or_else(w::invalid)?),
348 version: saved_version.ok_or_else(w::invalid)?.core(t)?,
349 })),
350 S::Failed => {
351 let f = failure.ok_or_else(w::invalid)?;
352 OperationState::Failed(OperationFailure {
353 error: w::core_error(&f.code, &safe_message),
354 target: t.clone(),
355 observed,
356 effect: if f.effect == WireEffect::ProvenNoEffect {
357 FailureEffect::ProvenNoEffect
358 } else {
359 FailureEffect::PossiblePartialEffects
360 },
361 })
362 }
363 S::Unknown => OperationState::Unknown(ReconciliationEvidence {
364 target: t.clone(),
365 observed,
366 explanation: safe_message,
367 }),
368 },
369 })
370}
371impl FileService for RemoteServices {
372 fn resolve(
373 &self,
374 root: FileRef,
375 purpose: FileResolvePurpose,
376 ) -> ServiceFuture<'_, Result<FileResolution, CoreError>> {
377 Box::pin(async move {
378 self.check(&root.target)?;
379 if !self
380 .negotiated_capabilities()
381 .iter()
382 .any(|c| c == "file.resolve")
383 {
384 return Err(CoreError::new(
385 ErrorCode::Unavailable,
386 "Target path resolution is unavailable",
387 ));
388 }
389 match self.call(
390 w::Call::FileResolve {
391 root: w::File::from_core(&root)?,
392 purpose: w::ResolvePurpose::from_core(&purpose)?,
393 },
394 None,
395 )? {
396 w::Reply::FileResolved {
397 root,
398 file,
399 resolved_path,
400 node_kind,
401 version,
402 } => Ok(FileResolution {
403 root: root.core(&self.target)?,
404 file: file.core(&self.target)?,
405 resolved_path: w::target_path(&resolved_path)?,
406 kind: node_kind
407 .map(|k| match k.as_str() {
408 "File" => Ok(FileKind::File),
409 "Directory" => Ok(FileKind::Directory),
410 "Symlink" => Ok(FileKind::Symlink),
411 "Other" => Ok(FileKind::Other),
412 _ => Err(w::invalid()),
413 })
414 .transpose()?,
415 version: version.map(|v| v.core(&self.target)).transpose()?,
416 }),
417 _ => Err(w::invalid()),
418 }
419 })
420 }
421 fn list(&self, file: FileRef) -> ServiceFuture<'_, Result<Vec<FileEntry>, CoreError>> {
422 Box::pin(async move {
423 self.check(&file.target)?;
424 let w::Reply::List { entries } = self.call(
425 w::Call::List {
426 file: w::File::from_core(&file)?,
427 },
428 None,
429 )?
430 else {
431 return Err(w::invalid());
432 };
433 entries
434 .into_iter()
435 .map(|e| {
436 Ok(FileEntry {
437 file: e.file.core(&self.target)?,
438 display_name: e.display_name,
439 kind: match e.kind.as_str() {
440 "File" => FileKind::File,
441 "Directory" => FileKind::Directory,
442 "Symlink" => FileKind::Symlink,
443 "Other" => FileKind::Other,
444 _ => return Err(w::invalid()),
445 },
446 })
447 })
448 .collect()
449 })
450 }
451 fn read(&self, file: FileRef) -> ServiceFuture<'_, Result<FileSnapshot, CoreError>> {
452 Box::pin(async move { self.read_sync(file) })
453 }
454 fn save(&self, r: SaveRequest) -> ServiceFuture<'_, OperationReceipt> {
455 Box::pin(async move { self.upload(r, false) })
456 }
457 fn mutate(&self, r: FileMutationRequest) -> ServiceFuture<'_, OperationReceipt> {
458 Box::pin(async move {
459 let id = r.operation;
460 let target = r.file.target.clone();
461 if let FileMutation::Create { source } = r.command {
462 if r.expected != ExpectedPathState::Absent {
463 return buntrail_services::files::failure(id, target, w::invalid());
464 }
465 return self.upload(
466 SaveRequest {
467 operation: id,
468 file: r.file,
469 expected_version: None,
470 submitted_revision: BufferRevision(0),
471 source,
472 },
473 true,
474 );
475 }
476 let result = (|| -> Result<OperationReceipt, CoreError> {
477 self.check(&target)?;
478 let file = w::File::from_core(&r.file)?;
479 let command = match r.command {
480 FileMutation::CreateDirectory => w::FileCommand::CreateDirectory,
481 FileMutation::Rename { destination } => w::FileCommand::Rename {
482 destination: w::path(&destination)?,
483 },
484 FileMutation::Move { destination } => w::FileCommand::Move {
485 destination: w::path(&destination)?,
486 },
487 FileMutation::Remove => {
488 if !self
489 .negotiated_capabilities()
490 .iter()
491 .any(|c| c == "file.remove")
492 {
493 return Err(CoreError::new(
494 ErrorCode::Unavailable,
495 "Active helper lacks nonrecursive removal; preserve its sessions and update after they end",
496 ));
497 }
498 w::FileCommand::Remove {
499 confirmed_target: self.transport.context.target_binding.clone(),
500 confirmed_path: file.path_base64.clone(),
501 }
502 }
503 FileMutation::Delete => w::FileCommand::Delete {
504 confirmed_target: self.transport.context.target_binding.clone(),
505 confirmed_path: file.path_base64.clone(),
506 },
507 FileMutation::Trash => {
508 return Err(CoreError::new(
509 ErrorCode::Unavailable,
510 "Remote deletion is permanent on the displayed target; local Trash is unavailable",
511 ));
512 }
513 FileMutation::Create { .. } => unreachable!(),
514 };
515 let expected = match r.expected {
516 ExpectedPathState::Absent => None,
517 ExpectedPathState::Present(v) => {
518 self.check(&v.target)?;
519 Some(w::version(&v))
520 }
521 };
522 Ok(self.mutation_call(
523 w::Call::FileMutate {
524 file,
525 expected,
526 command,
527 },
528 id,
529 &target,
530 ))
531 })();
532 result.unwrap_or_else(|e| buntrail_services::files::failure(id, target, e))
533 })
534 }
535 fn watch(&self, file: FileRef) -> ServiceFuture<'_, Result<EventStream<FileEvent>, CoreError>> {
536 Box::pin(async move {
537 self.check(&file.target)?;
538 Ok(Box::pin(Watch {
539 source: self.stream(
540 w::Call::Watch {
541 file: w::File::from_core(&file)?,
542 },
543 None,
544 )?,
545 target: self.target.clone(),
546 }) as EventStream<FileEvent>)
547 })
548 }
549}
550struct Watch {
551 source: Replies,
552 target: TargetIdentity,
553}
554impl Stream for Watch {
555 type Item = Result<FileEvent, CoreError>;
556 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
557 let result = Pin::new(&mut self.source).poll_next(cx);
558 result.map(|v| {
559 v.map(|r| {
560 r.and_then(|r| {
561 let w::Reply::Watch { file, event } = r else {
562 return Err(w::invalid());
563 };
564 let file = file.core(&self.target)?;
565 Ok(match event.as_str() {
566 "changed" => FileEvent::Changed(file),
567 "removed" => FileEvent::Removed(file),
568 "rescan_required" => FileEvent::RescanRequired(file),
569 _ => return Err(w::invalid()),
570 })
571 })
572 })
573 })
574 }
575}
576
577pub struct RemoteOwnedSave {
580 remote: RemoteServices,
581 request: SaveRequest,
582 approval: GitSaveApproval,
583 _budget: buntrail_services::files::LoadBudget,
584}
585impl RemoteOwnedSave {
586 pub fn approval(&self) -> &GitSaveApproval {
587 &self.approval
588 }
589 pub fn execute(self) -> OperationReceipt {
590 if let Err(error) = self.remote.owned_save_ready() {
591 return buntrail_services::files::failure(
592 self.request.operation,
593 self.request.file.target,
594 error,
595 );
596 }
597 self.remote.upload(self.request, false)
598 }
599}
600impl RemoteServices {
601 fn owned_save_ready(&self) -> Result<(), CoreError> {
602 if !self.transport.owns_registration(&self.lease) {
603 return Err(CoreError::new(
604 ErrorCode::Closed,
605 "exact remote save registration is no longer active",
606 ));
607 }
608 if !self
609 .negotiated_capabilities()
610 .iter()
611 .any(|c| c == "file.save_proof")
612 {
613 return Err(CoreError::new(
614 ErrorCode::Unavailable,
615 "active helper lacks authenticated save provenance; preserve its existing sessions",
616 ));
617 }
618 Ok(())
619 }
620 pub fn prepare_owned_save(&self, request: SaveRequest) -> Result<RemoteOwnedSave, CoreError> {
621 self.owned_save_ready()?;
622 let length = match &request.source {
623 ByteSource::Memory(b) => b.len() as u64,
624 ByteSource::PrivateSpool { length, .. } => *length,
625 };
626 if length > (buntrail_services::files::MAX_DOCUMENT_BYTES * 2 + 4) as u64 {
627 return Err(CoreError::new(
628 ErrorCode::Unavailable,
629 "save source exceeds staging bound",
630 ));
631 }
632 let budget =
633 buntrail_services::files::LoadBudget::acquire(length as usize * 2 + 1024 * 1024)?;
634 self.freeze_owned_save(request, budget)
635 }
636 pub fn prepare_owned_document(
637 &self,
638 recovery: &buntrail_services::recovery::store::RecoveryStore,
639 prepared: buntrail_core::documents::PreparedSave,
640 ) -> Result<RemoteOwnedSave, CoreError> {
641 self.owned_save_ready()?;
642 self.check(&prepared.recovery.file.target)?;
643 let (request, budget) =
644 buntrail_services::files::document_io::prepare_save(recovery, prepared)?;
645 self.freeze_owned_save(request, budget)
646 }
647 fn freeze_owned_save(
648 &self,
649 mut request: SaveRequest,
650 budget: buntrail_services::files::LoadBudget,
651 ) -> Result<RemoteOwnedSave, CoreError> {
652 self.owned_save_ready()?;
653 self.check(&request.file.target)?;
654 if let Some(expected) = &request.expected_version {
655 self.check(&expected.target)?;
656 }
657 let bytes = match &request.source {
658 ByteSource::Memory(bytes) => bytes.clone(),
659 source => Arc::from(self.source_bytes(
660 source,
661 buntrail_services::files::MAX_DOCUMENT_BYTES * 2 + 4,
662 true,
663 )?),
664 };
665 request.file.resolved_identity = None;
666 let approval = GitSaveApproval {
667 operation: request.operation,
668 actor: self.lease.clone(),
669 request_digest: buntrail_services::files::save_request_digest(
670 &request.file,
671 &request.expected_version,
672 request.submitted_revision,
673 sha2::Sha256::digest(&bytes).into(),
674 )?,
675 };
676 request.source = ByteSource::Memory(bytes);
677 Ok(RemoteOwnedSave {
678 remote: self.clone(),
679 request,
680 approval,
681 _budget: budget,
682 })
683 }
684}