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 if !self
509 .negotiated_capabilities()
510 .iter()
511 .any(|c| c == "file.trash")
512 {
513 return Err(CoreError::new(
514 ErrorCode::Unavailable,
515 "Active helper lacks Trash; update after retained sessions end",
516 ));
517 }
518 w::FileCommand::Trash {
519 confirmed_target: self.transport.context.target_binding.clone(),
520 confirmed_path: file.path_base64.clone(),
521 }
522 }
523 FileMutation::Create { .. } => unreachable!(),
524 };
525 let expected = match r.expected {
526 ExpectedPathState::Absent => None,
527 ExpectedPathState::Present(v) => {
528 self.check(&v.target)?;
529 Some(w::version(&v))
530 }
531 };
532 Ok(self.mutation_call(
533 w::Call::FileMutate {
534 file,
535 expected,
536 command,
537 },
538 id,
539 &target,
540 ))
541 })();
542 result.unwrap_or_else(|e| buntrail_services::files::failure(id, target, e))
543 })
544 }
545 fn watch(&self, file: FileRef) -> ServiceFuture<'_, Result<EventStream<FileEvent>, CoreError>> {
546 Box::pin(async move {
547 self.check(&file.target)?;
548 Ok(Box::pin(Watch {
549 source: self.stream(
550 w::Call::Watch {
551 file: w::File::from_core(&file)?,
552 },
553 None,
554 )?,
555 target: self.target.clone(),
556 }) as EventStream<FileEvent>)
557 })
558 }
559}
560struct Watch {
561 source: Replies,
562 target: TargetIdentity,
563}
564impl Stream for Watch {
565 type Item = Result<FileEvent, CoreError>;
566 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
567 let result = Pin::new(&mut self.source).poll_next(cx);
568 result.map(|v| {
569 v.map(|r| {
570 r.and_then(|r| {
571 let w::Reply::Watch { file, event } = r else {
572 return Err(w::invalid());
573 };
574 let file = file.core(&self.target)?;
575 Ok(match event.as_str() {
576 "changed" => FileEvent::Changed(file),
577 "removed" => FileEvent::Removed(file),
578 "rescan_required" => FileEvent::RescanRequired(file),
579 _ => return Err(w::invalid()),
580 })
581 })
582 })
583 })
584 }
585}
586
587pub struct RemoteOwnedSave {
590 remote: RemoteServices,
591 request: SaveRequest,
592 approval: GitSaveApproval,
593 _budget: buntrail_services::files::LoadBudget,
594}
595impl RemoteOwnedSave {
596 pub fn approval(&self) -> &GitSaveApproval {
597 &self.approval
598 }
599 pub fn execute(self) -> OperationReceipt {
600 if let Err(error) = self.remote.owned_save_ready() {
601 return buntrail_services::files::failure(
602 self.request.operation,
603 self.request.file.target,
604 error,
605 );
606 }
607 self.remote.upload(self.request, false)
608 }
609}
610impl RemoteServices {
611 fn owned_save_ready(&self) -> Result<(), CoreError> {
612 if !self.transport.owns_registration(&self.lease) {
613 return Err(CoreError::new(
614 ErrorCode::Closed,
615 "exact remote save registration is no longer active",
616 ));
617 }
618 if !self
619 .negotiated_capabilities()
620 .iter()
621 .any(|c| c == "file.save_proof")
622 {
623 return Err(CoreError::new(
624 ErrorCode::Unavailable,
625 "active helper lacks authenticated save provenance; preserve its existing sessions",
626 ));
627 }
628 Ok(())
629 }
630 pub fn prepare_owned_save(&self, request: SaveRequest) -> Result<RemoteOwnedSave, CoreError> {
631 self.owned_save_ready()?;
632 let length = match &request.source {
633 ByteSource::Memory(b) => b.len() as u64,
634 ByteSource::PrivateSpool { length, .. } => *length,
635 };
636 if length > (buntrail_services::files::MAX_DOCUMENT_BYTES * 2 + 4) as u64 {
637 return Err(CoreError::new(
638 ErrorCode::Unavailable,
639 "save source exceeds staging bound",
640 ));
641 }
642 let budget =
643 buntrail_services::files::LoadBudget::acquire(length as usize * 2 + 1024 * 1024)?;
644 self.freeze_owned_save(request, budget)
645 }
646 pub fn prepare_owned_document(
647 &self,
648 recovery: &buntrail_services::recovery::store::RecoveryStore,
649 prepared: buntrail_core::documents::PreparedSave,
650 ) -> Result<RemoteOwnedSave, CoreError> {
651 self.owned_save_ready()?;
652 self.check(&prepared.recovery.file.target)?;
653 let (request, budget) =
654 buntrail_services::files::document_io::prepare_save(recovery, prepared)?;
655 self.freeze_owned_save(request, budget)
656 }
657 fn freeze_owned_save(
658 &self,
659 mut request: SaveRequest,
660 budget: buntrail_services::files::LoadBudget,
661 ) -> Result<RemoteOwnedSave, CoreError> {
662 self.owned_save_ready()?;
663 self.check(&request.file.target)?;
664 if let Some(expected) = &request.expected_version {
665 self.check(&expected.target)?;
666 }
667 let bytes = match &request.source {
668 ByteSource::Memory(bytes) => bytes.clone(),
669 source => Arc::from(self.source_bytes(
670 source,
671 buntrail_services::files::MAX_DOCUMENT_BYTES * 2 + 4,
672 true,
673 )?),
674 };
675 request.file.resolved_identity = None;
676 let approval = GitSaveApproval {
677 operation: request.operation,
678 actor: self.lease.clone(),
679 request_digest: buntrail_services::files::save_request_digest(
680 &request.file,
681 &request.expected_version,
682 request.submitted_revision,
683 sha2::Sha256::digest(&bytes).into(),
684 )?,
685 };
686 request.source = ByteSource::Memory(bytes);
687 Ok(RemoteOwnedSave {
688 remote: self.clone(),
689 request,
690 approval,
691 _budget: budget,
692 })
693 }
694}