strop_remote/filesystem/
mod.rs

1//! Fixed protected filesystem operations, invoked only by owned workers.
2//! Directory authority does not transfer remote file-save permits.
3use ropey::Rope;
4#[cfg(all(test, target_os = "linux"))]
5mod tests;
6use serde::{de::DeserializeOwned, Deserialize};
7use serde_json::{json, Value};
8use sha2::{Digest, Sha256};
9use strop_core::worker::CancelToken;
10use strop_workspace::operation::*;
11use strop_workspace::{Filesystem, Observation, RemoteEndpoint, ResourceLocation};
12
13const HELPER: &str = concat!(
14    include_str!("../protected.py"),
15    "\n",
16    include_str!("observe.py"),
17    "\n",
18    include_str!("cleanup.py"),
19    "\n",
20    include_str!("mutate.py"),
21    "\n",
22    include_str!("main.py")
23);
24const HEADER_LIMIT: usize = 64 * 1024;
25const REPLY_LIMIT: usize = 64 * 1024;
26const BODY_LIMIT: usize = 256 * 1024 * 1024;
27
28fn failure(kind: FsFailureKind, detail: impl Into<String>) -> FsFailure {
29    FsFailure::new(kind, detail)
30}
31fn endpoint(intent: &OperationIntent) -> Result<&RemoteEndpoint, FsFailure> {
32    let location = intent.location().ok_or_else(|| {
33        failure(
34            FsFailureKind::InvalidPath,
35            "filesystem operation has no resource",
36        )
37    })?;
38    let Filesystem::Remote(endpoint) = &location.filesystem else {
39        return Err(failure(
40            FsFailureKind::Unsupported,
41            "protected SSH operations require a remote namespace",
42        ));
43    };
44    if intent
45        .source
46        .iter()
47        .chain(intent.destination.iter())
48        .any(|other| other.filesystem != location.filesystem)
49    {
50        return Err(failure(
51            FsFailureKind::Unsupported,
52            "cross-namespace transfer is unsupported",
53        ));
54    }
55    Ok(endpoint)
56}
57fn path(location: Option<&ResourceLocation>) -> Option<&[u8]> {
58    location.map(|location| strop_workspace::addr::uri::path_bytes(&location.path))
59}
60
61#[derive(Deserialize)]
62struct NativeObservation {
63    path: Vec<u8>,
64    value: Option<Observation>,
65}
66impl NativeObservation {
67    fn locate(self, endpoint: &RemoteEndpoint) -> Result<LocatedObservation, FsFailure> {
68        let path = strop_workspace::addr::uri::bytes_to_path(self.path)
69            .map_err(|error| failure(FsFailureKind::Protocol, error.to_string()))?;
70        let file = strop_workspace::RemoteFile::from_path(endpoint.clone(), path)
71            .map_err(|error| failure(FsFailureKind::Protocol, error.to_string()))?;
72        Ok(LocatedObservation {
73            location: ResourceLocation::remote(endpoint.clone(), file.path().to_owned()),
74            value: self.value,
75        })
76    }
77}
78#[derive(Deserialize)]
79struct PreparedReply {
80    source: Option<NativeObservation>,
81    destination: Option<NativeObservation>,
82    parents: Vec<NativeObservation>,
83    capability: OperationCapability,
84}
85#[derive(Deserialize)]
86struct Reply<T> {
87    version: u8,
88    value: Option<T>,
89    error: Option<HelperError>,
90}
91#[derive(Deserialize)]
92struct HelperError {
93    kind: String,
94    detail: String,
95    unconfirmed: bool,
96    observed_destination: Option<Observation>,
97    publication: Option<PublicationWitness>,
98}
99impl HelperError {
100    fn failure(self) -> FsFailure {
101        let kind = match self.kind.as_str() {
102            "unsupported" => FsFailureKind::Unsupported,
103            "permission" => FsFailureKind::Permission,
104            "conflict" => FsFailureKind::Conflict,
105            "busy" => FsFailureKind::Busy,
106            "invalid_path" => FsFailureKind::InvalidPath,
107            "incomplete" => FsFailureKind::Incomplete,
108            "cancelled" => FsFailureKind::Cancelled,
109            "io" => FsFailureKind::Io,
110            _ => FsFailureKind::Protocol,
111        };
112        failure(kind, self.detail)
113    }
114}
115
116enum InvokeError {
117    Before(FsFailure),
118    Unconfirmed {
119        detail: String,
120        observed_destination: Option<Box<Observation>>,
121        publication: Option<PublicationWitness>,
122    },
123}
124fn invoke<T: DeserializeOwned>(
125    endpoint: &RemoteEndpoint,
126    request: Value,
127    contents: Option<&Rope>,
128    mutation: bool,
129    token: &CancelToken,
130) -> Result<T, InvokeError> {
131    if token.is_cancelled() {
132        return Err(InvokeError::Before(failure(
133            FsFailureKind::Cancelled,
134            "cancelled before helper launch",
135        )));
136    }
137    let mut header = serde_json::to_vec(&request).map_err(|error| {
138        InvokeError::Before(failure(FsFailureKind::Protocol, error.to_string()))
139    })?;
140    header.push(b'\n');
141    if header.len() > HEADER_LIMIT {
142        return Err(InvokeError::Before(failure(
143            FsFailureKind::Unsupported,
144            "filesystem request header exceeds its bound",
145        )));
146    }
147    let command = crate::RemoteCommand::python(HELPER, Vec::new(), std::path::Path::new("/"))
148        .map_err(|error| {
149            InvokeError::Before(failure(FsFailureKind::Unsupported, error.to_string()))
150        })?;
151    let mut chunks = vec![header.as_slice()];
152    if let Some(contents) = contents {
153        chunks.extend(contents.chunks().map(str::as_bytes));
154    }
155    let lost = |detail: String| {
156        if mutation {
157            InvokeError::Unconfirmed {
158                detail,
159                observed_destination: None,
160                publication: None,
161            }
162        } else {
163            InvokeError::Before(failure(FsFailureKind::Io, detail))
164        }
165    };
166    let output = crate::run_with_input(endpoint, &command, token, &chunks)
167        .map_err(|error| lost(error.to_string()))?;
168    if output.stdout_dropped != 0 || output.stdout.len() > REPLY_LIMIT {
169        return Err(lost("filesystem response exceeded its bound".into()));
170    }
171    let reply: Reply<T> = serde_json::from_slice(&output.stdout).map_err(|error| {
172        lost(format!(
173            "{error}; {}",
174            String::from_utf8_lossy(&output.stderr)
175        ))
176    })?;
177    if reply.version != 1 {
178        return Err(lost("filesystem protocol version differs".into()));
179    }
180    if let Some(error) = reply.error {
181        return Err(if error.unconfirmed {
182            InvokeError::Unconfirmed {
183                detail: error.detail,
184                observed_destination: error.observed_destination.map(Box::new),
185                publication: error.publication,
186            }
187        } else {
188            InvokeError::Before(error.failure())
189        });
190    }
191    if !output.status.success() || output.stdin_error.is_some() {
192        return Err(lost(format!(
193            "filesystem helper transport did not complete: {:?}; {:?}",
194            output.status, output.stdin_error
195        )));
196    }
197    reply
198        .value
199        .ok_or_else(|| lost("filesystem response omitted its result".into()))
200}
201fn read_error(error: InvokeError) -> FsFailure {
202    match error {
203        InvokeError::Before(error) => error,
204        InvokeError::Unconfirmed { detail, .. } => failure(FsFailureKind::Protocol, detail),
205    }
206}
207
208pub fn prepare(
209    intent: &OperationIntent,
210    allow_occupied: bool,
211    token: &CancelToken,
212) -> Result<PreparedOperation, FsFailure> {
213    let endpoint = endpoint(intent)?;
214    let buffer_copy =
215        intent.kind == OperationKind::Copy && intent.copy_version == CopyVersion::Buffer;
216    let reply: PreparedReply = invoke(endpoint, json!({"version":1,"action":"prepare","kind":intent.kind,
217        "source":path(intent.source.as_ref()),"destination":path(intent.destination.as_ref()),
218        "buffer_copy":buffer_copy,"allow_occupied":allow_occupied,"expected_content":intent.expected_content}), None, false, token).map_err(read_error)?;
219    if reply.parents.len() > 2 {
220        return Err(failure(
221            FsFailureKind::Protocol,
222            "helper returned too many parent observations",
223        ));
224    }
225    Ok(PreparedOperation {
226        intent: intent.clone(),
227        source: reply
228            .source
229            .map(|value| value.locate(endpoint))
230            .transpose()?,
231        destination: reply
232            .destination
233            .map(|value| value.locate(endpoint))
234            .transpose()?,
235        parents: reply
236            .parents
237            .into_iter()
238            .map(|value| value.locate(endpoint))
239            .collect::<Result<_, _>>()?,
240        dependencies: Vec::new(),
241        capability: reply.capability,
242    })
243}
244
245fn request(operation: &PreparedOperation, action: &str) -> Value {
246    json!({"version":1,"action":action,"kind":operation.intent.kind,
247        "source":path(operation.source.as_ref().map(|value| &value.location)),
248        "destination":path(operation.destination.as_ref().map(|value| &value.location)),
249        "logical_source":path(operation.intent.source.as_ref()),"logical_destination":path(operation.intent.destination.as_ref()),
250        "before_source":operation.source.as_ref().and_then(|value| value.value.as_ref()),
251        "before_destination":operation.destination.as_ref().and_then(|value| value.value.as_ref()),
252        "parents":operation.parents.iter().map(|parent| json!({"path":path(Some(&parent.location)),"value":parent.value})).collect::<Vec<_>>(),
253        "capability":operation.capability,"buffer_copy":operation.intent.copy_version == CopyVersion::Buffer})
254}
255
256pub fn execute(
257    operation: &PreparedOperation,
258    contents: Option<&Rope>,
259    receipts: &[StepReceipt],
260    token: &CancelToken,
261) -> StepOutcome {
262    let endpoint = match endpoint(&operation.intent) {
263        Ok(endpoint) => endpoint,
264        Err(error) => return StepOutcome::Refused(error),
265    };
266    let mut request = request(operation, "apply");
267    if operation.intent.kind == OperationKind::Copy
268        && operation.intent.copy_version == CopyVersion::Buffer
269    {
270        let Some(contents) = contents else {
271            return StepOutcome::Refused(failure(
272                FsFailureKind::Conflict,
273                "buffer copy snapshot is missing",
274            ));
275        };
276        if contents.len_bytes() > BODY_LIMIT {
277            return StepOutcome::Refused(failure(
278                FsFailureKind::Unsupported,
279                "buffer copy exceeds the 256 MiB upload bound",
280            ));
281        }
282        let mut digest = Sha256::new();
283        for chunk in contents.chunks() {
284            digest.update(chunk.as_bytes());
285        }
286        let digest: [u8; 32] = digest.finalize().into();
287        request["length"] = json!(contents.len_bytes());
288        request["digest"] = json!(digest);
289    }
290    request["parents"] = Value::Array(
291        operation
292            .parents
293            .iter()
294            .map(|parent| {
295                let value = parent.value.as_ref().or_else(|| {
296                    operation.dependencies.iter().find_map(|step| {
297                        receipts.iter().find_map(|receipt| {
298                            if receipt.step != *step
299                                || receipt
300                                    .operation
301                                    .destination
302                                    .as_ref()
303                                    .map(|value| &value.location)
304                                    != Some(&parent.location)
305                            {
306                                return None;
307                            }
308                            match &receipt.outcome {
309                                StepOutcome::Committed {
310                                    destination_after, ..
311                                } => destination_after.as_ref(),
312                                _ => None,
313                            }
314                        })
315                    })
316                });
317                json!({"path":path(Some(&parent.location)),"value":value})
318            })
319            .collect(),
320    );
321    let vacated = operation.destination.as_ref().is_some_and(|destination| {
322        destination.value.as_ref().is_some_and(|expected| {
323            operation.dependencies.iter().any(|step| {
324                receipts.iter().any(|receipt| {
325                    receipt.step == *step
326                        && receipt.outcome.is_committed()
327                        && receipt.operation.source.as_ref().is_some_and(|source| {
328                            source.location == destination.location
329                                && source
330                                    .value
331                                    .as_ref()
332                                    .is_some_and(|value| value.same_object(expected))
333                        })
334                })
335            })
336        })
337    });
338    request["vacated"] = json!(vacated);
339    let contents = contents.filter(|_| {
340        operation.intent.kind == OperationKind::Copy
341            && operation.intent.copy_version == CopyVersion::Buffer
342    });
343    match invoke(endpoint, request, contents, true, token) {
344        Ok(outcome) => outcome,
345        Err(InvokeError::Before(error)) if error.kind == FsFailureKind::Cancelled => {
346            StepOutcome::Cancelled {
347                detail: error.detail,
348            }
349        }
350        Err(InvokeError::Before(error)) => StepOutcome::Refused(error),
351        Err(InvokeError::Unconfirmed {
352            detail,
353            observed_destination,
354            publication,
355        }) => StepOutcome::Unconfirmed {
356            detail,
357            observed_destination: observed_destination.map(|value| *value),
358            recovery: None,
359            publication,
360        },
361    }
362}
363
364pub fn verify(receipt: &StepReceipt, token: &CancelToken) -> Result<VerifiedOutcome, FsFailure> {
365    let endpoint = endpoint(&receipt.operation.intent)?;
366    let mut request = request(&receipt.operation, "verify");
367    request["observed_destination"] = json!(match &receipt.outcome {
368        StepOutcome::Committed {
369            destination_after, ..
370        } => destination_after.as_ref(),
371        StepOutcome::Unconfirmed {
372            observed_destination,
373            ..
374        } => observed_destination.as_ref(),
375        _ => None,
376    });
377    request["publication"] = json!(match &receipt.outcome {
378        StepOutcome::Committed { publication, .. }
379        | StepOutcome::Unconfirmed { publication, .. } => publication.as_ref(),
380        _ => None,
381    });
382    invoke(endpoint, request, None, false, token).map_err(read_error)
383}
strop_remote/filesystem/mod.rs

strop_remote/filesystem/
mod.rs
