falsegreen_agent/
artifact.rs

1//! Integrity-checked, bounded, atomic acquisition of pinned remote artifacts.
2
3use std::fs::{self, File, OpenOptions};
4use std::io::{Read, Seek, SeekFrom, Write};
5use std::path::{Path, PathBuf};
6use std::thread;
7use std::time::{Duration, Instant};
8
9use fs2::FileExt;
10use sha2::{Digest, Sha256};
11use thiserror::Error;
12use uuid::Uuid;
13
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct ArtifactSpec {
16    pub id: String,
17    pub version: String,
18    pub file_name: String,
19    pub url: String,
20    pub sha256: String,
21    pub size_bytes: u64,
22}
23
24impl ArtifactSpec {
25    pub fn validate(&self) -> Result<(), ArtifactError> {
26        for (field, value) in [
27            ("id", self.id.as_str()),
28            ("version", self.version.as_str()),
29            ("file_name", self.file_name.as_str()),
30        ] {
31            if !safe_component(value) {
32                return Err(ArtifactError::InvalidSpec(format!(
33                    "{field} must be one non-empty portable path component"
34                )));
35            }
36        }
37        if !(self.url.starts_with("https://") || self.url.starts_with("http://127.0.0.1:")) {
38            return Err(ArtifactError::InvalidSpec(
39                "artifact URL must use HTTPS (loopback HTTP is permitted for tests)".to_owned(),
40            ));
41        }
42        if self.sha256.len() != 64 || !self.sha256.bytes().all(|byte| byte.is_ascii_hexdigit()) {
43            return Err(ArtifactError::InvalidSpec(
44                "SHA-256 must contain exactly 64 hexadecimal characters".to_owned(),
45            ));
46        }
47        if self.size_bytes == 0 {
48            return Err(ArtifactError::InvalidSpec(
49                "expected artifact size must be nonzero".to_owned(),
50            ));
51        }
52        Ok(())
53    }
54}
55
56#[derive(Debug, Clone, Copy, PartialEq, Eq)]
57pub struct DownloadPolicy {
58    pub request_timeout: Duration,
59    pub total_timeout: Duration,
60    pub lock_timeout: Duration,
61}
62
63impl Default for DownloadPolicy {
64    fn default() -> Self {
65        Self {
66            request_timeout: Duration::from_secs(60),
67            total_timeout: Duration::from_secs(60 * 60),
68            lock_timeout: Duration::from_secs(30),
69        }
70    }
71}
72
73#[derive(Debug, Clone, PartialEq, Eq)]
74pub struct AcquiredFile {
75    pub path: PathBuf,
76    pub reused: bool,
77    pub sha256: String,
78    pub size_bytes: u64,
79}
80
81#[derive(Debug, Clone)]
82pub struct ArtifactFetcher {
83    policy: DownloadPolicy,
84}
85
86impl ArtifactFetcher {
87    #[must_use]
88    pub const fn new(policy: DownloadPolicy) -> Self {
89        Self { policy }
90    }
91
92    /// Ensure `destination` contains exactly the pinned bytes in `spec`.
93    ///
94    /// Concurrent processes serialize through an advisory lock. A damaged cache entry is retained
95    /// beside the target with an `.invalid-*` suffix for diagnosis, and the replacement becomes
96    /// visible only after its size and SHA-256 have both been verified and synced.
97    pub fn ensure_file(
98        &self,
99        spec: &ArtifactSpec,
100        destination: impl AsRef<Path>,
101    ) -> Result<AcquiredFile, ArtifactError> {
102        spec.validate()?;
103        let destination = destination.as_ref();
104        let parent = destination
105            .parent()
106            .ok_or_else(|| ArtifactError::InvalidDestination(destination.to_path_buf()))?;
107        fs::create_dir_all(parent)
108            .map_err(|source| io_error("create cache directory", parent, source))?;
109        let _lock = CacheLock::acquire(&lock_path(destination)?, self.policy.lock_timeout)?;
110
111        match fs::symlink_metadata(destination) {
112            Ok(metadata) if metadata.file_type().is_symlink() => {
113                quarantine(destination)?;
114            }
115            Ok(_) => match verify_file(destination, spec) {
116                Ok(()) => {
117                    return Ok(AcquiredFile {
118                        path: destination.to_path_buf(),
119                        reused: true,
120                        sha256: spec.sha256.to_ascii_lowercase(),
121                        size_bytes: spec.size_bytes,
122                    });
123                }
124                Err(ArtifactError::DigestMismatch { .. } | ArtifactError::SizeMismatch { .. }) => {
125                    quarantine(destination)?;
126                }
127                Err(error) => return Err(error),
128            },
129            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
130            Err(source) => {
131                return Err(io_error("inspect cached artifact", destination, source));
132            }
133        }
134
135        let temp_path = temporary_sibling(destination, "partial")?;
136        let mut temp = TemporaryFile::create(temp_path)?;
137        let diagnostic_path = temp.path().to_path_buf();
138        self.download(spec, temp.file_mut(), &diagnostic_path)?;
139        temp.file_mut()
140            .sync_all()
141            .map_err(|source| io_error("sync downloaded artifact", temp.path(), source))?;
142        verify_file(temp.path(), spec)?;
143        temp.persist(destination)?;
144
145        Ok(AcquiredFile {
146            path: destination.to_path_buf(),
147            reused: false,
148            sha256: spec.sha256.to_ascii_lowercase(),
149            size_bytes: spec.size_bytes,
150        })
151    }
152
153    fn download(
154        &self,
155        spec: &ArtifactSpec,
156        output: &mut File,
157        output_path: &Path,
158    ) -> Result<(), ArtifactError> {
159        let agent = ureq::AgentBuilder::new()
160            // ureq's aggregate `timeout` covers the entire response, which makes
161            // a healthy multi-gigabyte transfer fail after the short inactivity
162            // budget. These are per-I/O inactivity limits; `total_timeout` below
163            // remains the explicit end-to-end bound.
164            .timeout_connect(self.policy.request_timeout)
165            .timeout_read(self.policy.request_timeout)
166            .timeout_write(self.policy.request_timeout)
167            .build();
168        let response = match agent
169            .get(&spec.url)
170            .set("accept", "application/octet-stream")
171            .call()
172        {
173            Ok(response) => response,
174            Err(ureq::Error::Status(status, _)) => {
175                return Err(ArtifactError::HttpStatus { status });
176            }
177            Err(ureq::Error::Transport(error)) => {
178                return Err(ArtifactError::Network(error.to_string()));
179            }
180        };
181        if let Some(length) = response.header("content-length") {
182            let length = length.parse::<u64>().map_err(|_| {
183                ArtifactError::Network("server returned an invalid Content-Length".to_owned())
184            })?;
185            if length != spec.size_bytes {
186                return Err(ArtifactError::SizeMismatch {
187                    expected: spec.size_bytes,
188                    actual: length,
189                });
190            }
191        }
192
193        let started = Instant::now();
194        let mut next_progress = started + Duration::from_secs(5);
195        let mut reader = response.into_reader();
196        let mut digest = Sha256::new();
197        let mut total = 0_u64;
198        let mut buffer = [0_u8; 1024 * 1024];
199        loop {
200            if started.elapsed() > self.policy.total_timeout {
201                return Err(ArtifactError::Timeout {
202                    seconds: self.policy.total_timeout.as_secs(),
203                });
204            }
205            let count = reader
206                .read(&mut buffer)
207                .map_err(|source| ArtifactError::Network(source.to_string()))?;
208            if count == 0 {
209                break;
210            }
211            total = total.saturating_add(count as u64);
212            if total > spec.size_bytes {
213                return Err(ArtifactError::SizeMismatch {
214                    expected: spec.size_bytes,
215                    actual: total,
216                });
217            }
218            digest.update(&buffer[..count]);
219            output
220                .write_all(&buffer[..count])
221                .map_err(|source| io_error("write downloaded artifact", output_path, source))?;
222            if Instant::now() >= next_progress {
223                let percent = total.saturating_mul(100) / spec.size_bytes;
224                eprintln!(
225                    "Downloading {}: {} MiB / {} MiB ({}%)",
226                    spec.file_name,
227                    total / (1024 * 1024),
228                    spec.size_bytes / (1024 * 1024),
229                    percent
230                );
231                next_progress = Instant::now() + Duration::from_secs(5);
232            }
233        }
234        if total != spec.size_bytes {
235            return Err(ArtifactError::SizeMismatch {
236                expected: spec.size_bytes,
237                actual: total,
238            });
239        }
240        let actual = format!("{:x}", digest.finalize());
241        if !actual.eq_ignore_ascii_case(&spec.sha256) {
242            return Err(ArtifactError::DigestMismatch {
243                expected: spec.sha256.to_ascii_lowercase(),
244                actual,
245            });
246        }
247        Ok(())
248    }
249}
250
251impl Default for ArtifactFetcher {
252    fn default() -> Self {
253        Self::new(DownloadPolicy::default())
254    }
255}
256
257#[derive(Debug, Error)]
258pub enum ArtifactError {
259    #[error("invalid artifact specification: {0}")]
260    InvalidSpec(String),
261    #[error("artifact destination has no parent directory: {0}")]
262    InvalidDestination(PathBuf),
263    #[error("artifact acquisition timed out after {seconds} seconds")]
264    Timeout { seconds: u64 },
265    #[error("timed out waiting for cache lock {path} after {seconds} seconds")]
266    LockTimeout { path: PathBuf, seconds: u64 },
267    #[error("artifact server returned HTTP {status}")]
268    HttpStatus { status: u16 },
269    #[error("artifact download failed: {0}")]
270    Network(String),
271    #[error("artifact size mismatch: expected {expected} bytes, found {actual}")]
272    SizeMismatch { expected: u64, actual: u64 },
273    #[error("artifact SHA-256 mismatch: expected {expected}, found {actual}")]
274    DigestMismatch { expected: String, actual: String },
275    #[error("could not {operation} at {path}: {source}")]
276    Io {
277        operation: &'static str,
278        path: PathBuf,
279        #[source]
280        source: std::io::Error,
281    },
282}
283
284pub(crate) fn hash_file(path: &Path) -> Result<(String, u64), ArtifactError> {
285    let mut file = File::open(path).map_err(|source| io_error("open artifact", path, source))?;
286    let mut digest = Sha256::new();
287    let mut size = 0_u64;
288    let mut buffer = [0_u8; 1024 * 1024];
289    loop {
290        let count = file
291            .read(&mut buffer)
292            .map_err(|source| io_error("read artifact", path, source))?;
293        if count == 0 {
294            break;
295        }
296        size = size.saturating_add(count as u64);
297        digest.update(&buffer[..count]);
298    }
299    Ok((format!("{:x}", digest.finalize()), size))
300}
301
302pub(crate) fn verify_file(path: &Path, spec: &ArtifactSpec) -> Result<(), ArtifactError> {
303    if !path.is_file() {
304        return Err(ArtifactError::InvalidDestination(path.to_path_buf()));
305    }
306    let (actual_digest, actual_size) = hash_file(path)?;
307    if actual_size != spec.size_bytes {
308        return Err(ArtifactError::SizeMismatch {
309            expected: spec.size_bytes,
310            actual: actual_size,
311        });
312    }
313    if !actual_digest.eq_ignore_ascii_case(&spec.sha256) {
314        return Err(ArtifactError::DigestMismatch {
315            expected: spec.sha256.to_ascii_lowercase(),
316            actual: actual_digest,
317        });
318    }
319    Ok(())
320}
321
322fn safe_component(value: &str) -> bool {
323    !value.is_empty()
324        && value != "."
325        && value != ".."
326        && value
327            .bytes()
328            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
329}
330
331fn lock_path(destination: &Path) -> Result<PathBuf, ArtifactError> {
332    let name = destination
333        .file_name()
334        .and_then(|name| name.to_str())
335        .ok_or_else(|| ArtifactError::InvalidDestination(destination.to_path_buf()))?;
336    Ok(destination.with_file_name(format!(".{name}.lock")))
337}
338
339fn temporary_sibling(path: &Path, label: &str) -> Result<PathBuf, ArtifactError> {
340    let name = path
341        .file_name()
342        .and_then(|name| name.to_str())
343        .ok_or_else(|| ArtifactError::InvalidDestination(path.to_path_buf()))?;
344    Ok(path.with_file_name(format!(".{name}.{label}-{}", Uuid::new_v4())))
345}
346
347fn quarantine(path: &Path) -> Result<PathBuf, ArtifactError> {
348    let target = temporary_sibling(path, "invalid")?;
349    fs::rename(path, &target)
350        .map_err(|source| io_error("quarantine invalid artifact", path, source))?;
351    Ok(target)
352}
353
354fn io_error(operation: &'static str, path: &Path, source: std::io::Error) -> ArtifactError {
355    ArtifactError::Io {
356        operation,
357        path: path.to_path_buf(),
358        source,
359    }
360}
361
362struct CacheLock {
363    file: File,
364}
365
366impl CacheLock {
367    fn acquire(path: &Path, timeout: Duration) -> Result<Self, ArtifactError> {
368        let started = Instant::now();
369        match fs::symlink_metadata(path) {
370            Ok(metadata) if metadata.file_type().is_symlink() => {
371                return Err(ArtifactError::InvalidDestination(path.to_path_buf()));
372            }
373            Ok(_) => {}
374            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
375            Err(source) => return Err(io_error("inspect cache lock", path, source)),
376        }
377        let mut file = OpenOptions::new()
378            .create(true)
379            .read(true)
380            .write(true)
381            .truncate(false)
382            .open(path)
383            .map_err(|source| io_error("open cache lock", path, source))?;
384        loop {
385            match FileExt::try_lock_exclusive(&file) {
386                Ok(()) => break,
387                Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
388                    if started.elapsed() >= timeout {
389                        return Err(ArtifactError::LockTimeout {
390                            path: path.to_path_buf(),
391                            seconds: timeout.as_secs(),
392                        });
393                    }
394                    thread::sleep(Duration::from_millis(50));
395                }
396                Err(source) => return Err(io_error("lock cache entry", path, source)),
397            }
398        }
399        file.set_len(0)
400            .and_then(|()| file.seek(SeekFrom::Start(0)).map(|_| ()))
401            .and_then(|()| writeln!(file, "pid={}", std::process::id()))
402            .and_then(|()| file.sync_all())
403            .map_err(|source| io_error("record cache lock owner", path, source))?;
404        Ok(Self { file })
405    }
406}
407
408impl Drop for CacheLock {
409    fn drop(&mut self) {
410        let _ = FileExt::unlock(&self.file);
411    }
412}
413
414struct TemporaryFile {
415    path: PathBuf,
416    file: Option<File>,
417}
418
419impl TemporaryFile {
420    fn create(path: PathBuf) -> Result<Self, ArtifactError> {
421        let file = OpenOptions::new()
422            .create_new(true)
423            .read(true)
424            .write(true)
425            .open(&path)
426            .map_err(|source| io_error("create temporary artifact", &path, source))?;
427        Ok(Self {
428            path,
429            file: Some(file),
430        })
431    }
432
433    fn path(&self) -> &Path {
434        &self.path
435    }
436
437    fn file_mut(&mut self) -> &mut File {
438        self.file.as_mut().expect("temporary file is present")
439    }
440
441    fn persist(mut self, destination: &Path) -> Result<(), ArtifactError> {
442        self.file.take();
443        fs::rename(&self.path, destination)
444            .map_err(|source| io_error("publish verified artifact", destination, source))?;
445        self.path = PathBuf::new();
446        Ok(())
447    }
448}
449
450impl Drop for TemporaryFile {
451    fn drop(&mut self) {
452        if !self.path.as_os_str().is_empty() {
453            let _ = fs::remove_file(&self.path);
454        }
455    }
456}
457
458#[cfg(test)]
459mod tests {
460    use std::io::{Read, Write};
461    use std::net::TcpListener;
462    use std::thread;
463    use std::time::Duration;
464
465    use sha2::{Digest, Sha256};
466
467    use super::{ArtifactError, ArtifactFetcher, ArtifactSpec, DownloadPolicy};
468
469    fn serve(body: Vec<u8>, requests: usize) -> String {
470        let listener = TcpListener::bind("127.0.0.1:0").expect("listener");
471        let address = listener.local_addr().expect("address");
472        thread::spawn(move || {
473            for _ in 0..requests {
474                let (mut stream, _) = listener.accept().expect("accept");
475                let mut request = [0_u8; 4096];
476                let _ = stream.read(&mut request);
477                write!(
478                    stream,
479                    "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
480                    body.len()
481                )
482                .expect("headers");
483                stream.write_all(&body).expect("body");
484            }
485        });
486        format!("http://{address}/artifact")
487    }
488
489    fn spec(url: String, body: &[u8]) -> ArtifactSpec {
490        ArtifactSpec {
491            id: "test-artifact".to_owned(),
492            version: "v1".to_owned(),
493            file_name: "test.bin".to_owned(),
494            url,
495            sha256: format!("{:x}", Sha256::digest(body)),
496            size_bytes: body.len() as u64,
497        }
498    }
499
500    fn serve_slow(body: Vec<u8>, pause: Duration) -> String {
501        let listener = TcpListener::bind("127.0.0.1:0").expect("listener");
502        let address = listener.local_addr().expect("address");
503        thread::spawn(move || {
504            let (mut stream, _) = listener.accept().expect("accept");
505            let mut request = [0_u8; 4096];
506            let _ = stream.read(&mut request);
507            write!(
508                stream,
509                "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
510                body.len()
511            )
512            .expect("headers");
513            for byte in body {
514                stream.write_all(&[byte]).expect("body byte");
515                stream.flush().expect("flush");
516                thread::sleep(pause);
517            }
518        });
519        format!("http://{address}/artifact")
520    }
521
522    #[test]
523    fn acquires_atomically_then_reuses_verified_cache() {
524        let body = b"trusted artifact bytes";
525        let fetcher = ArtifactFetcher::default();
526        let directory = tempfile::tempdir().expect("tempdir");
527        let destination = directory.path().join("test.bin");
528        let artifact = spec(serve(body.to_vec(), 1), body);
529
530        let first = fetcher
531            .ensure_file(&artifact, &destination)
532            .expect("download");
533        assert!(!first.reused);
534        let second = fetcher.ensure_file(&artifact, &destination).expect("reuse");
535        assert!(second.reused);
536        assert_eq!(std::fs::read(destination).expect("read"), body);
537    }
538
539    #[test]
540    fn request_timeout_is_inactivity_not_whole_download_deadline() {
541        let body = b"slow-but-continuous";
542        let fetcher = ArtifactFetcher::new(DownloadPolicy {
543            request_timeout: Duration::from_millis(100),
544            total_timeout: Duration::from_secs(3),
545            lock_timeout: Duration::from_secs(1),
546        });
547        let directory = tempfile::tempdir().expect("tempdir");
548        let destination = directory.path().join("test.bin");
549        let artifact = spec(serve_slow(body.to_vec(), Duration::from_millis(25)), body);
550
551        let acquired = fetcher
552            .ensure_file(&artifact, &destination)
553            .expect("continuous transfer may exceed one inactivity interval overall");
554        assert!(!acquired.reused);
555        assert_eq!(std::fs::read(destination).expect("read"), body);
556    }
557
558    #[test]
559    fn quarantines_corrupt_cache_before_reacquiring() {
560        let body = b"trusted bytes";
561        let directory = tempfile::tempdir().expect("tempdir");
562        let destination = directory.path().join("test.bin");
563        std::fs::write(&destination, b"corrupt").expect("corrupt cache");
564        let artifact = spec(serve(body.to_vec(), 1), body);
565
566        ArtifactFetcher::default()
567            .ensure_file(&artifact, &destination)
568            .expect("replace");
569        let invalid = std::fs::read_dir(directory.path())
570            .expect("entries")
571            .filter_map(Result::ok)
572            .any(|entry| entry.file_name().to_string_lossy().contains(".invalid-"));
573        assert!(invalid);
574        assert_eq!(std::fs::read(destination).expect("read"), body);
575    }
576
577    #[test]
578    fn refuses_digest_mismatch_without_publishing_partial_file() {
579        let body = b"unexpected";
580        let directory = tempfile::tempdir().expect("tempdir");
581        let destination = directory.path().join("test.bin");
582        let mut artifact = spec(serve(body.to_vec(), 1), body);
583        artifact.sha256 = "0".repeat(64);
584        let fetcher = ArtifactFetcher::new(DownloadPolicy {
585            request_timeout: std::time::Duration::from_secs(1),
586            total_timeout: std::time::Duration::from_secs(1),
587            lock_timeout: std::time::Duration::from_secs(1),
588        });
589
590        assert!(matches!(
591            fetcher.ensure_file(&artifact, &destination),
592            Err(ArtifactError::DigestMismatch { .. })
593        ));
594        assert!(!destination.exists());
595    }
596
597    #[test]
598    fn rejects_path_components_and_non_https_remote_urls() {
599        let artifact = ArtifactSpec {
600            id: "../escape".to_owned(),
601            version: "v1".to_owned(),
602            file_name: "file".to_owned(),
603            url: "http://example.com/file".to_owned(),
604            sha256: "0".repeat(64),
605            size_bytes: 1,
606        };
607        assert!(matches!(
608            artifact.validate(),
609            Err(ArtifactError::InvalidSpec(_))
610        ));
611    }
612
613    #[cfg(unix)]
614    #[test]
615    fn never_reuses_a_symlink_as_a_cache_object() {
616        use std::os::unix::fs::symlink;
617
618        let body = b"trusted bytes";
619        let directory = tempfile::tempdir().expect("tempdir");
620        let external = directory.path().join("external.bin");
621        std::fs::write(&external, body).expect("external");
622        let cache = directory.path().join("cache");
623        std::fs::create_dir(&cache).expect("cache");
624        let destination = cache.join("test.bin");
625        symlink(&external, &destination).expect("symlink");
626        let artifact = spec(serve(body.to_vec(), 1), body);
627
628        let acquired = ArtifactFetcher::default()
629            .ensure_file(&artifact, &destination)
630            .expect("acquire owned cache object");
631        assert!(!acquired.reused);
632        assert!(
633            !std::fs::symlink_metadata(&destination)
634                .expect("metadata")
635                .file_type()
636                .is_symlink()
637        );
638        assert_eq!(std::fs::read(external).expect("external read"), body);
639    }
640}
falsegreen_agent/artifact.rs

falsegreen_agent/
artifact.rs
