soma_infra/
linux_filesystem.rs

1use std::fs::{File, Metadata};
2use std::io::{Read, Seek, SeekFrom};
3use std::os::fd::OwnedFd;
4use std::path::{Path, PathBuf};
5use std::time::SystemTime;
6
7use async_trait::async_trait;
8use rustix::fs::{Mode, OFlags, ResolveFlags, open, openat2};
9use sha2::{Digest, Sha256};
10use soma_fleet::{HostEndpoint, HostRecord};
11use tokio_util::sync::CancellationToken;
12
13use crate::{
14    FileHash, FileKind, FileMetadata, FilePreview, FileReadPolicy, FilesystemInspector, InfraError,
15    InfraResult,
16};
17
18/// Linux descriptor-confined filesystem reader.
19#[derive(Debug, Clone)]
20pub struct LinuxFilesystemInspector {
21    policy: FileReadPolicy,
22}
23
24impl LinuxFilesystemInspector {
25    /// Creates an inspector from an explicit read policy.
26    #[must_use]
27    pub fn new(policy: FileReadPolicy) -> Self {
28        Self { policy }
29    }
30
31    /// Returns the active read policy.
32    #[must_use]
33    pub fn policy(&self) -> &FileReadPolicy {
34        &self.policy
35    }
36}
37
38#[async_trait]
39impl FilesystemInspector for LinuxFilesystemInspector {
40    async fn stat(
41        &self,
42        host: &HostRecord,
43        path: &Path,
44        cancellation: &CancellationToken,
45    ) -> InfraResult<FileMetadata> {
46        ensure_local(host)?;
47        ensure_not_cancelled(cancellation)?;
48        let policy = self.policy.clone();
49        let path = path.to_path_buf();
50        let host = host.clone();
51        run_blocking(cancellation, move || {
52            let bound = bind_path(&policy, &path, OFlags::PATH | OFlags::CLOEXEC)?;
53            metadata_for(
54                &host,
55                &path,
56                &bound
57                    .file
58                    .metadata()
59                    .map_err(|error| fs_error("stat", &path, error))?,
60            )
61        })
62        .await
63    }
64
65    async fn read(
66        &self,
67        host: &HostRecord,
68        path: &Path,
69        cancellation: &CancellationToken,
70    ) -> InfraResult<FilePreview> {
71        ensure_local(host)?;
72        ensure_not_cancelled(cancellation)?;
73        let policy = self.policy.clone();
74        let path = path.to_path_buf();
75        let host = host.clone();
76        let operation_cancellation = cancellation.clone();
77        run_blocking(cancellation, move || {
78            let mut bound = bind_path(
79                &policy,
80                &path,
81                OFlags::RDONLY | OFlags::NONBLOCK | OFlags::CLOEXEC,
82            )?;
83            let metadata = bound
84                .file
85                .metadata()
86                .map_err(|error| fs_error("read", &path, error))?;
87            let typed = metadata_for(&host, &path, &metadata)?;
88            if typed.kind != FileKind::File {
89                return Err(InfraError::Filesystem {
90                    operation: "read",
91                    path,
92                    message: "path is not a regular file".into(),
93                });
94            }
95            let limit = policy.max_preview_bytes();
96            let mut content = Vec::with_capacity(limit.min(8192));
97            let mut buffer = [0_u8; 64 * 1024];
98            while content.len() <= limit {
99                ensure_not_cancelled(&operation_cancellation)?;
100                let remaining = limit.saturating_add(1).saturating_sub(content.len());
101                let read_limit = remaining.min(buffer.len());
102                let read = bound
103                    .file
104                    .read(&mut buffer[..read_limit])
105                    .map_err(|error| fs_error("read", &typed.path, error))?;
106                if read == 0 {
107                    break;
108                }
109                content.extend_from_slice(&buffer[..read]);
110            }
111            let truncated = content.len() > limit;
112            content.truncate(limit);
113            Ok(FilePreview {
114                metadata: typed,
115                content,
116                truncated,
117            })
118        })
119        .await
120    }
121
122    async fn hash(
123        &self,
124        host: &HostRecord,
125        path: &Path,
126        cancellation: &CancellationToken,
127    ) -> InfraResult<FileHash> {
128        ensure_local(host)?;
129        ensure_not_cancelled(cancellation)?;
130        let policy = self.policy.clone();
131        let path = path.to_path_buf();
132        let host = host.clone();
133        let operation_cancellation = cancellation.clone();
134        run_blocking(cancellation, move || {
135            let mut bound = bind_path(
136                &policy,
137                &path,
138                OFlags::RDONLY | OFlags::NONBLOCK | OFlags::CLOEXEC,
139            )?;
140            let metadata = bound
141                .file
142                .metadata()
143                .map_err(|error| fs_error("hash", &path, error))?;
144            let typed = metadata_for(&host, &path, &metadata)?;
145            if typed.kind != FileKind::File {
146                return Err(InfraError::Filesystem {
147                    operation: "hash",
148                    path,
149                    message: "path is not a regular file".into(),
150                });
151            }
152            if typed.size_bytes > policy.max_hash_bytes() {
153                return Err(InfraError::InvalidRequest {
154                    domain: "filesystem",
155                    message: format!(
156                        "file is {} bytes; hash limit is {}",
157                        typed.size_bytes,
158                        policy.max_hash_bytes()
159                    ),
160                });
161            }
162            bound
163                .file
164                .seek(SeekFrom::Start(0))
165                .map_err(|error| fs_error("hash", &typed.path, error))?;
166            let (sha256, bytes_hashed) = hash_reader(
167                &mut bound.file,
168                &typed.path,
169                policy.max_hash_bytes(),
170                &operation_cancellation,
171            )?;
172            Ok(FileHash {
173                metadata: typed,
174                sha256,
175                bytes_hashed,
176            })
177        })
178        .await
179    }
180}
181
182fn hash_reader(
183    reader: &mut impl Read,
184    path: &Path,
185    max_bytes: u64,
186    cancellation: &CancellationToken,
187) -> InfraResult<(String, u64)> {
188    let mut hasher = Sha256::new();
189    let mut buffer = [0_u8; 64 * 1024];
190    let mut bytes_hashed = 0_u64;
191    loop {
192        ensure_not_cancelled(cancellation)?;
193        let read_limit_u64 = max_bytes
194            .saturating_sub(bytes_hashed)
195            .saturating_add(1)
196            .min(buffer.len() as u64);
197        let read_limit = usize::try_from(read_limit_u64).unwrap_or(buffer.len());
198        let read = reader
199            .read(&mut buffer[..read_limit])
200            .map_err(|error| fs_error("hash", path, error))?;
201        if read == 0 {
202            break;
203        }
204        let next_bytes = bytes_hashed.saturating_add(read as u64);
205        if next_bytes > max_bytes {
206            return Err(InfraError::InvalidRequest {
207                domain: "filesystem",
208                message: format!("file grew beyond hash limit of {max_bytes} bytes"),
209            });
210        }
211        bytes_hashed = next_bytes;
212        hasher.update(&buffer[..read]);
213    }
214    let digest = hasher.finalize();
215    let digest = digest.iter().map(|byte| format!("{byte:02x}")).collect();
216    Ok((digest, bytes_hashed))
217}
218
219struct BoundFile {
220    file: File,
221}
222
223fn bind_path(policy: &FileReadPolicy, path: &Path, target_flags: OFlags) -> InfraResult<BoundFile> {
224    let (root, relative) = policy.resolve(path)?;
225    let slash: OwnedFd = open(
226        "/",
227        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC,
228        Mode::empty(),
229    )
230    .map_err(|error| fs_error("open-root", path, error))?;
231    let root_relative = root.strip_prefix("/").unwrap_or(root.as_path());
232    let root_fd = openat2(
233        &slash,
234        if root_relative.as_os_str().is_empty() {
235            Path::new(".")
236        } else {
237            root_relative
238        },
239        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC,
240        Mode::empty(),
241        ResolveFlags::BENEATH | ResolveFlags::NO_SYMLINKS | ResolveFlags::NO_MAGICLINKS,
242    )
243    .map_err(|error| fs_error("open-root", path, error))?;
244    let target = if relative.as_os_str().is_empty() {
245        Path::new(".")
246    } else {
247        relative.as_path()
248    };
249    let fd = openat2(
250        &root_fd,
251        target,
252        target_flags,
253        Mode::empty(),
254        ResolveFlags::BENEATH | ResolveFlags::NO_SYMLINKS | ResolveFlags::NO_MAGICLINKS,
255    )
256    .map_err(|error| fs_error("open", path, error))?;
257    Ok(BoundFile { file: fd.into() })
258}
259
260fn metadata_for(host: &HostRecord, path: &Path, metadata: &Metadata) -> InfraResult<FileMetadata> {
261    let kind = if metadata.is_file() {
262        FileKind::File
263    } else if metadata.is_dir() {
264        FileKind::Directory
265    } else {
266        return Err(InfraError::Filesystem {
267            operation: "stat",
268            path: path.to_path_buf(),
269            message: "path is neither a regular file nor a directory".into(),
270        });
271    };
272    let modified_unix_millis = metadata.modified().ok().and_then(system_time_millis);
273    Ok(FileMetadata {
274        host: host.id().clone(),
275        topology_revision: host.revision().clone(),
276        path: path.to_path_buf(),
277        kind,
278        size_bytes: if kind == FileKind::File {
279            metadata.len()
280        } else {
281            0
282        },
283        readonly: metadata.permissions().readonly(),
284        modified_unix_millis,
285    })
286}
287
288fn system_time_millis(value: SystemTime) -> Option<i64> {
289    let duration = value.duration_since(SystemTime::UNIX_EPOCH).ok()?;
290    i64::try_from(duration.as_millis()).ok()
291}
292
293fn ensure_local(host: &HostRecord) -> InfraResult<()> {
294    if matches!(host.endpoint(), HostEndpoint::Local) {
295        Ok(())
296    } else {
297        Err(InfraError::UnsupportedTarget {
298            domain: "filesystem",
299            host: host.id().clone(),
300        })
301    }
302}
303
304fn ensure_not_cancelled(cancellation: &CancellationToken) -> InfraResult<()> {
305    if cancellation.is_cancelled() {
306        Err(soma_fleet::FleetError::Cancelled.into())
307    } else {
308        Ok(())
309    }
310}
311
312async fn run_blocking<T, F>(cancellation: &CancellationToken, operation: F) -> InfraResult<T>
313where
314    T: Send + 'static,
315    F: FnOnce() -> InfraResult<T> + Send + 'static,
316{
317    let task = tokio::task::spawn_blocking(operation);
318    tokio::select! {
319        () = cancellation.cancelled() => Err(soma_fleet::FleetError::Cancelled.into()),
320        result = task => result.map_err(|error| InfraError::Filesystem {
321            operation: "join",
322            path: PathBuf::new(),
323            message: error.to_string(),
324        })?,
325    }
326}
327
328fn fs_error(operation: &'static str, path: &Path, error: impl std::fmt::Display) -> InfraError {
329    InfraError::Filesystem {
330        operation,
331        path: path.to_path_buf(),
332        message: error.to_string(),
333    }
334}
335
336#[cfg(test)]
337#[path = "linux_filesystem_tests.rs"]
338mod tests;
soma_infra/linux_filesystem.rs

soma_infra/
linux_filesystem.rs
