diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index c724cbd7..94daf10c 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -99,6 +99,13 @@ matched the bytes, or `Unpinned` for an open-world key with no pin. Callers must verified one. A pin mismatch returns `IntegrityViolation` rather than the bytes, so the gate fails closed. +The pin itself belongs to the backend, which keeps it with the bytes: an NDJSON +sidecar (`.nora-pins.ndjson`) on the local filesystem, and the user-defined +`sha256` object metadata on S3/GCS, written atomically with the object and read +back on GET/HEAD. The wrapper only validates keys and runs the gate; an object +stored without a digest (or written before pins existed) simply has none and +stays open-world. + The curation layer is a second trust boundary for proxy traffic. When mode is `enforce`, a package must pass all filters (blocklist, allowlist, namespace, integrity) before reaching storage. When mode is `audit`, blocked packages @@ -149,9 +156,9 @@ nora/ │ │ └── mod.rs # Re-exports: docker_routes(), maven_routes(), ... │ │ │ ├── storage/ -│ │ ├── mod.rs # StorageBackend trait + Storage wrapper (validate + pin gate) -│ │ ├── local.rs # Local filesystem implementation -│ │ └── object.rs # Object-store implementation (S3-compatible + GCS) +│ │ ├── mod.rs # StorageBackend trait + Storage wrapper (validate + verify gate) +│ │ ├── local.rs # Local filesystem implementation (pins in an NDJSON sidecar) +│ │ └── object.rs # Object-store implementation, S3-compatible + GCS (pins in object metadata) │ │ │ ├── auth/ # Authentication (middleware + providers) │ │ ├── mod.rs # auth_middleware, provider dispatch @@ -165,7 +172,7 @@ nora/ │ ├── validation.rs # Input validation: storage keys, package names, null bytes │ │ │ ├── verified.rs # Compile-time integrity witnesses (GateOutcome typestate) -│ ├── hash_pin_store.rs # SHA-256 pins recorded on put(), verified on get() +│ ├── hash_pin_store.rs # SHA-256 pin sidecar for the local backend (NDJSON) │ ├── digest_quarantine.rs # First-seen tracking for proxy-fetched digests │ ├── circuit_breaker.rs # Per-registry circuit breaker for upstream proxy calls │ ├── proxy_coalesce.rs # Single-flight coalescing on the proxy cache-miss path diff --git a/CHANGELOG.md b/CHANGELOG.md index a8a4a828..7888d166 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,9 @@ # Changelog ## [Unreleased] -### Fixed -- **GC: tag-rooted mark walk kept a tag manifest's children but swept the digest-named copy of the manifest itself**, so pull-by-digest of a tagged image 404'd after the first GC run while pull-by-tag kept working. The walk now marks `manifests/sha256:.json` for every tag manifest — the digest alias the OCI distribution spec requires to stay pullable. Orphaned digest manifests now also take their `.meta.json` sidecar with them instead of leaking it. (#949) +### Added +- **Hash pins on S3/GCS via object metadata** — the SHA-256 integrity pin is no longer a local-filesystem-only feature. On object-store backends it is written as the user-defined `sha256` object metadata, atomically with the object, and read back on GET/HEAD, so buffered reads verify at rest and raw files get `ETag`, `If-None-Match` (304) and `If-Match` conditional overwrite on every backend. Pins are now a backend concern: the local backend keeps its NDJSON sidecar (same path and format, no migration), the object-store backend keeps object metadata, and the storage wrapper only validates keys and runs the fail-closed verify gate. Objects written before the upgrade carry no metadata and stay open-world until they are rewritten; `nora re-pin` rewrites the object on an object store, since object metadata cannot be changed in place. +- **Raw upload integrity via `Repr-Digest` (RFC 9530)** — a raw `PUT` may declare `Repr-Digest: sha-256=:BASE64:`; NORA verifies the received body against it before committing, so a corrupted or truncated upload is rejected with `400` instead of being pinned. The pin itself is always the server-computed hash; the header only gates the commit. A `Repr-Digest` without a sha-256 entry is rejected rather than silently skipped. ## [1.2.0] - 2026-08-23 diff --git a/COMPAT.md b/COMPAT.md index a3bcfe47..da602633 100644 --- a/COMPAT.md +++ b/COMPAT.md @@ -389,7 +389,7 @@ Helm charts are stored as OCI artifacts via the Docker registry endpoints. `helm | Health check | Full | `/health` | | Swagger/OpenAPI | Full | `/api-docs` | | S3 backend | Full | AWS S3, Ceph RGW. Basic storage works on any S3-compatible; multi-replica write-serialization has a caveat — see note below. | -| GCS backend | Full | Native Google Cloud Storage (`storage.mode = "gcs"`): Workload Identity / service-account JSON / ambient credentials; endpoint override for emulators and Private Google Access. Same single-writer caveat as S3 for rpm/deb publishing (in-process publish lock). Hash-pinning (at-rest integrity verification) is unavailable on ALL object-store backends, not only S3. | +| GCS backend | Full | Native Google Cloud Storage (`storage.mode = "gcs"`): Workload Identity / service-account JSON / ambient credentials; endpoint override for emulators and Private Google Access. Same single-writer caveat as S3 for rpm/deb publishing (in-process publish lock). Hash-pinning (at-rest integrity verification) works on every backend: the pin is the `sha256` object metadata on S3/GCS. | | Local filesystem backend | Full | Default, content-addressable | | Activity log | Full | Recent push/pull in dashboard | | Backup/restore | Full | CLI commands | diff --git a/README.md b/README.md index 3d5174d5..018f9745 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,7 @@ All endpoints require authentication. Anonymous read is opt-in via `anonymous_re | Cargo | ✅ | ✅ | `crates.io` (sparse index) | hosted + proxy (sparse index) | | PyPI | ✅ | ✅ | `pypi.org/simple/` | hosted + proxy | | Go Modules | ✅ | — | `proxy.golang.org` | proxy only (modules immutable, push not in protocol) | -| Raw files | ❌ | ✅ | — (no upstream) | hosted only; conditional `PUT` (ETag/`If-Match` — local backend only; `If-None-Match: *` works on any backend) | +| Raw files | ❌ | ✅ | — (no upstream) | hosted only; conditional `PUT`/`GET` (ETag, `If-Match`, `If-None-Match`) on every backend; upload verification via `Repr-Digest` (RFC 9530) | | RubyGems | ✅ | ❌ | `rubygems.org` | proxy only — `gem push` not implemented in NORA v1.1.0 | | Terraform | ✅ | — | `registry.terraform.io` | proxy only; client configuration notes in COMPAT.md | | Ansible Galaxy | ✅ | ❌ | `galaxy.ansible.com` | proxy only — `ansible-galaxy collection publish` not implemented | diff --git a/nora-registry/src/gc.rs b/nora-registry/src/gc.rs index 23447e85..06da6eeb 100644 --- a/nora-registry/src/gc.rs +++ b/nora-registry/src/gc.rs @@ -1366,12 +1366,18 @@ mod tests { Vec::new() }) } - async fn put(&self, _key: &str, _data: &[u8]) -> crate::storage::Result<()> { + async fn put(&self, _key: &str, _data: &[u8], _sha256: &str) -> crate::storage::Result<()> { Ok(()) } - async fn get(&self, _key: &str) -> crate::storage::Result { + async fn get( + &self, + _key: &str, + ) -> crate::storage::Result<(axum::body::Bytes, Option)> { Err(crate::storage::StorageError::NotFound) } + async fn pin(&self, _key: &str) -> Option { + None + } async fn delete(&self, _key: &str) -> crate::storage::Result<()> { Ok(()) } @@ -1388,6 +1394,7 @@ mod tests { &self, _key: &str, _src: &std::path::Path, + _sha256: Option<&str>, ) -> crate::storage::Result<()> { Ok(()) } @@ -1396,11 +1403,17 @@ mod tests { _key: &str, ) -> crate::storage::Result<( u64, + Option, std::pin::Pin>, )> { Err(crate::storage::StorageError::NotFound) } - async fn copy(&self, _src: &str, _dst: &str) -> crate::storage::Result<()> { + async fn copy( + &self, + _src: &str, + _dst: &str, + _sha256: Option<&str>, + ) -> crate::storage::Result<()> { Err(crate::storage::StorageError::NotFound) } } diff --git a/nora-registry/src/hash_pin_store.rs b/nora-registry/src/hash_pin_store.rs index ccc687a2..3e06c783 100644 --- a/nora-registry/src/hash_pin_store.rs +++ b/nora-registry/src/hash_pin_store.rs @@ -1,11 +1,13 @@ // Copyright (c) 2026 The NORA Authors // SPDX-License-Identifier: MIT -//! Hash Pin Store — immutable hash verification for stored artifacts. +//! Hash Pin Store — the local filesystem backend's record of the SHA-256 pin of +//! every artifact it stores. //! -//! Records SHA-256 hashes on every `Storage::put()` and verifies them on -//! `Storage::get()`. Detects tampering at the storage layer (e.g. direct -//! filesystem modification bypassing NORA). +//! `LocalStorage` records a pin on every write and hands it back on every read; +//! the `Storage` wrapper is what compares it against the bytes. Together they +//! detect tampering at the storage layer (e.g. direct filesystem modification +//! bypassing NORA). //! //! Persistence: append-only NDJSON file (`.nora-pins.ndjson`) compacted on //! startup. Each line: `{"k":"storage/key","h":"sha256hex"}`. An empty `h` @@ -23,7 +25,6 @@ use parking_lot::RwLock; use serde::{Deserialize, Serialize}; -use sha2::{Digest, Sha256}; use std::collections::HashMap; use std::io::{self, BufRead, Write}; use std::path::PathBuf; @@ -95,18 +96,9 @@ impl HashPinStore { store } - /// Compute SHA-256 hex digest. - fn sha256_hex(data: &[u8]) -> String { - let mut hasher = Sha256::new(); - hasher.update(data); - hex::encode(hasher.finalize()) - } - - /// Record the hash for a storage key. Called on every `put()`. + /// Record a pre-computed SHA-256 hash for a storage key. /// - /// If the key is new, the hash is pinned. If the key exists with the same - /// hash, this is a no-op. If the hash changed (normal metadata update), - /// the pin is updated. + /// `hash` must be a lowercase hex-encoded SHA-256 (64 chars). /// /// Returns the I/O error if the pin append fails, so the caller can fail /// closed rather than serve an artifact it could not pin. The in-memory @@ -114,40 +106,18 @@ impl HashPinStore { /// claim a pin the disk does not hold, or a `get()` after a failed `put()` /// would verify against a RAM-only pin that vanishes on restart, and a /// retried `put()` would skip the (still-missing) append. - pub fn record(&self, key: &str, data: &[u8]) -> io::Result<()> { - let hash = Self::sha256_hex(data); - // Atomic per key: hold the write lock across check → append → insert. - // The disk append happens before the in-memory update (durability), and - // no concurrent record() for the same key can interleave its append and - // insert with ours, so disk and memory cannot diverge. (An earlier - // two-lock version — read-check, release, append, write-insert — had a - // TOCTOU where two same-key writers' append and insert orders disagreed.) - // The append is a ~100-byte line and record() runs on a blocking thread - // (`spawn_blocking`), so holding the lock across it trades a little read - // contention for correctness — the right call for a tamper-detection store. - let mut pins = self.pins.write(); - if pins.get(key).is_none_or(|existing| *existing != hash) { - Self::append_to_file(&self.path, key, &hash)?; - pins.insert(key.to_string(), hash); - } - Ok(()) - } - - /// Record a pre-computed SHA-256 hash for a storage key. - /// - /// Used by streaming paths where the hash was already computed - /// incrementally during download — avoids re-reading the file (#580). - /// - /// `hash` must be a lowercase hex-encoded SHA-256 (64 chars). Durability and - /// ordering match [`HashPinStore::record`]: the pin is appended before the - /// in-memory index is updated, and an I/O failure is returned to the caller. pub fn record_hash(&self, key: &str, hash: &str) -> io::Result<()> { debug_assert!( hash.len() == 64 && hash.chars().all(|c| c.is_ascii_hexdigit()), "record_hash: expected 64-char hex SHA-256, got: {hash}" ); - // Same atomic check → append → insert under one write lock as record(). + // Atomic per key: hold the write lock across check → append → insert. + // The disk append happens before the in-memory update (durability), and + // no concurrent record_hash() for the same key can interleave its append + // and insert with ours, so disk and memory cannot diverge. (An earlier + // two-lock version — read-check, release, append, write-insert — had a + // TOCTOU where two same-key writers' append and insert orders disagreed.) let mut pins = self.pins.write(); if pins.get(key).is_none_or(|existing| *existing != hash) { Self::append_to_file(&self.path, key, hash)?; @@ -156,28 +126,6 @@ impl HashPinStore { Ok(()) } - /// Verify data integrity against pinned hash. Called on every `get()`. - /// - /// Returns `true` if the hash matches or no pin exists for this key. - /// Returns `false` and logs a warning if tampering is detected. - #[must_use = "ignoring verification result may allow tampered data"] - pub fn verify(&self, key: &str, data: &[u8]) -> bool { - let pins = self.pins.read(); - if let Some(expected) = pins.get(key) { - let actual = Self::sha256_hex(data); - if *expected != actual { - warn!( - key = key, - expected = expected.as_str(), - actual = actual.as_str(), - "INTEGRITY VIOLATION: stored artifact hash mismatch" - ); - return false; - } - } - true - } - /// Remove a pin entry. Called on `delete()`. /// /// Appends a tombstone before dropping the in-memory entry, returning any @@ -186,7 +134,7 @@ impl HashPinStore { /// before verification, and a later `put()` of the key overwrites the pin — /// so callers may treat a remove failure as non-fatal. pub fn remove(&self, key: &str) -> io::Result<()> { - // Atomic tombstone: append + drop under one write lock (see record()). + // Atomic tombstone: append + drop under one write lock (see record_hash()). let mut pins = self.pins.write(); if pins.contains_key(key) { Self::append_to_file(&self.path, key, "")?; @@ -200,11 +148,6 @@ impl HashPinStore { self.pins.read().get(key).cloned() } - /// Number of pinned entries. - pub fn len(&self) -> usize { - self.pins.read().len() - } - /// Compact the NDJSON file: rewrite with only live entries via a temp file /// and an atomic rename. Returns any I/O error; the caller decides whether a /// compaction failure is fatal (it is not — see [`HashPinStore::new`]). @@ -255,45 +198,41 @@ impl HashPinStore { #[allow(clippy::unwrap_used)] mod tests { use super::*; + use sha2::{Digest, Sha256}; use tempfile::TempDir; fn pin_path(dir: &TempDir) -> PathBuf { dir.path().join(".nora-pins.ndjson") } - #[test] - fn test_record_and_verify() { - let dir = TempDir::new().unwrap(); - let store = HashPinStore::new(pin_path(&dir)); - - store - .record("maven/com/example/1.0/app.jar", b"jar-content") - .unwrap(); - assert!(store.verify("maven/com/example/1.0/app.jar", b"jar-content")); - assert!(!store.verify("maven/com/example/1.0/app.jar", b"tampered")); + fn sha(data: &[u8]) -> String { + hex::encode(Sha256::digest(data)) } #[test] - fn test_verify_unknown_key_passes() { + fn test_record_and_get() { let dir = TempDir::new().unwrap(); let store = HashPinStore::new(pin_path(&dir)); + let key = "maven/com/example/1.0/app.jar"; - // No pin exists — verification passes (open world) - assert!(store.verify("unknown/key", b"anything")); + store.record_hash(key, &sha(b"jar-content")).unwrap(); + assert_eq!( + store.get(key).as_deref(), + Some(sha(b"jar-content").as_str()) + ); + assert_eq!(store.get("unknown/key"), None); } #[test] fn test_record_update_overwrites_pin() { let dir = TempDir::new().unwrap(); let store = HashPinStore::new(pin_path(&dir)); + let key = "npm/meta/express"; - store.record("npm/meta/express", b"v1").unwrap(); - assert!(store.verify("npm/meta/express", b"v1")); - + store.record_hash(key, &sha(b"v1")).unwrap(); // Metadata update — pin is updated - store.record("npm/meta/express", b"v2").unwrap(); - assert!(store.verify("npm/meta/express", b"v2")); - assert!(!store.verify("npm/meta/express", b"v1")); + store.record_hash(key, &sha(b"v2")).unwrap(); + assert_eq!(store.get(key).as_deref(), Some(sha(b"v2").as_str())); } #[test] @@ -301,14 +240,9 @@ mod tests { let dir = TempDir::new().unwrap(); let store = HashPinStore::new(pin_path(&dir)); - store.record("key", b"data").unwrap(); - assert_eq!(store.len(), 1); - + store.record_hash("key", &sha(b"data")).unwrap(); store.remove("key").unwrap(); - assert_eq!(store.len(), 0); - - // After removal, any data passes verification (no pin) - assert!(store.verify("key", b"whatever")); + assert_eq!(store.get("key"), None); } #[test] @@ -318,16 +252,15 @@ mod tests { { let store = HashPinStore::new(&path); - store.record("a", b"data-a").unwrap(); - store.record("b", b"data-b").unwrap(); + store.record_hash("a", &sha(b"data-a")).unwrap(); + store.record_hash("b", &sha(b"data-b")).unwrap(); store.remove("b").unwrap(); } // Reload from disk let store = HashPinStore::new(&path); - assert_eq!(store.len(), 1); - assert!(store.verify("a", b"data-a")); - assert!(store.verify("b", b"anything")); // removed, no pin + assert_eq!(store.get("a").as_deref(), Some(sha(b"data-a").as_str())); + assert_eq!(store.get("b"), None); } #[test] @@ -337,14 +270,14 @@ mod tests { { let store = HashPinStore::new(&path); - store.record("keep", b"data").unwrap(); - store.record("remove", b"data").unwrap(); + store.record_hash("keep", &sha(b"data")).unwrap(); + store.record_hash("remove", &sha(b"data")).unwrap(); store.remove("remove").unwrap(); } // After reload + compact, file should only have 1 entry let store = HashPinStore::new(&path); - assert_eq!(store.len(), 1); + assert!(store.get("keep").is_some()); let content = std::fs::read_to_string(&path).unwrap(); let lines: Vec<&str> = content.lines().collect(); @@ -358,13 +291,13 @@ mod tests { let path = pin_path(&dir); let store = HashPinStore::new(&path); - // Same data twice — should not append duplicate - store.record("key", b"data").unwrap(); - store.record("key", b"data").unwrap(); + // Same hash twice — should not append duplicate + store.record_hash("key", &sha(b"data")).unwrap(); + store.record_hash("key", &sha(b"data")).unwrap(); let content = std::fs::read_to_string(&path).unwrap(); let lines: Vec<&str> = content.lines().collect(); - assert_eq!(lines.len(), 1, "duplicate record should be idempotent"); + assert_eq!(lines.len(), 1, "duplicate record_hash should be idempotent"); } #[test] @@ -373,66 +306,10 @@ mod tests { let path = pin_path(&dir); let store = HashPinStore::new(&path); - assert_eq!(store.len(), 0); + assert_eq!(store.get("anything"), None); assert!(!path.exists(), "empty store should not create file"); } - #[test] - fn test_record_hash_and_verify() { - let dir = TempDir::new().unwrap(); - let store = HashPinStore::new(pin_path(&dir)); - - // Pre-computed SHA-256 of b"streaming-data" - let hash = HashPinStore::sha256_hex(b"streaming-data"); - store.record_hash("docker/blob/sha256:abc", &hash).unwrap(); - - assert_eq!(store.len(), 1); - assert!(store.verify("docker/blob/sha256:abc", b"streaming-data")); - assert!(!store.verify("docker/blob/sha256:abc", b"tampered")); - } - - #[test] - fn test_record_hash_persists_on_reload() { - let dir = TempDir::new().unwrap(); - let path = pin_path(&dir); - - let hash = HashPinStore::sha256_hex(b"persistent"); - { - let store = HashPinStore::new(&path); - store.record_hash("key/hash", &hash).unwrap(); - } - - // Reload - let store = HashPinStore::new(&path); - assert_eq!(store.len(), 1); - assert!(store.verify("key/hash", b"persistent")); - } - - #[test] - fn test_record_hash_idempotent() { - let dir = TempDir::new().unwrap(); - let path = pin_path(&dir); - let store = HashPinStore::new(&path); - - let hash = HashPinStore::sha256_hex(b"data"); - store.record_hash("key", &hash).unwrap(); - store.record_hash("key", &hash).unwrap(); - - let content = std::fs::read_to_string(&path).unwrap(); - let lines: Vec<&str> = content.lines().collect(); - assert_eq!(lines.len(), 1, "duplicate record_hash should be idempotent"); - } - - #[test] - fn test_sha256_correctness() { - // Known test vector: SHA-256 of empty string - let hash = HashPinStore::sha256_hex(b""); - assert_eq!( - hash, - "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" - ); - } - /// A pin write to an unwritable path must surface the I/O error, not swallow /// it — otherwise `put()` reports success while the pin never lands, /// silently downgrading the key to open-world on the next restart. @@ -450,26 +327,23 @@ mod tests { let store = HashPinStore::new(&unwritable); assert!( - store.record("k", b"data").is_err(), + store.record_hash("k", &sha(b"data")).is_err(), "pin write to an unwritable path must return an error, not swallow it" ); // The in-memory index must not claim a pin the disk never accepted. - assert_eq!(store.len(), 0, "failed pin write must not update memory"); - - let hash = HashPinStore::sha256_hex(b"data"); - assert!( - store.record_hash("k", &hash).is_err(), - "record_hash must propagate the same I/O error" + assert_eq!( + store.get("k"), + None, + "failed pin write must not update memory" ); - assert_eq!(store.len(), 0, "failed record_hash must not update memory"); } - /// Regression for the disk-first TOCTOU: concurrent record() calls for the - /// SAME key with DIFFERENT data must leave the in-memory pin equal to what a - /// fresh reload from disk sees — disk and memory cannot diverge. Holding the - /// write lock across check → append → insert makes each record() atomic per - /// key; the earlier two-lock version could append in one order but insert in - /// the other. + /// Regression for the disk-first TOCTOU: concurrent record_hash() calls for + /// the SAME key with DIFFERENT hashes must leave the in-memory pin equal to + /// what a fresh reload from disk sees — disk and memory cannot diverge. + /// Holding the write lock across check → append → insert makes each call + /// atomic per key; the earlier two-lock version could append in one order + /// but insert in the other. #[test] fn test_concurrent_same_key_disk_memory_consistent() { use std::sync::Arc; @@ -482,7 +356,7 @@ mod tests { .map(|i| { let s = Arc::clone(&store); std::thread::spawn(move || { - let _ = s.record(key, format!("data-{i}").as_bytes()); + let _ = s.record_hash(key, &sha(format!("data-{i}").as_bytes())); }) }) .collect(); diff --git a/nora-registry/src/import/mod.rs b/nora-registry/src/import/mod.rs index 47fbceae..4890fe03 100644 --- a/nora-registry/src/import/mod.rs +++ b/nora-registry/src/import/mod.rs @@ -31,9 +31,6 @@ //! `import-key-format-equals-handler-key-format`. //! - **SSRF guard** on the source URL and every redirect hop (DNS-pinned) — //! review R2, contract `import-ssrf-per-redirect-hop`. -//! - **at-rest integrity degrades on S3** (the sha256 pin is local-only): a loud -//! WARN is emitted — review R6, accepted contract -//! `import-s3-integrity-at-rest-degraded`. use async_trait::async_trait; use axum::body::Bytes; @@ -206,7 +203,7 @@ pub async fn run( config: &crate::config::Config, ) -> Result<()> { match cmd { - ImportCommand::Assess(args) => assess(args, storage, config).await, + ImportCommand::Assess(args) => assess(args, config).await, ImportCommand::Run(args) => run_import(args, storage, config).await, } } @@ -221,11 +218,7 @@ fn read_auth() -> Option { /// `nora import assess` — read-only per-repo compatibility table plus a /// connectivity/SSRF/auth smoke test. Writes nothing, sets no markers. -async fn assess( - args: AssessArgs, - storage: &crate::storage::Storage, - config: &crate::config::Config, -) -> Result<()> { +async fn assess(args: AssessArgs, config: &crate::config::Config) -> Result<()> { // Default-deny SSRF on the operator URL (assess has no opt-out flag). http::precheck_url(&args.url, false)?; let client = http::build_import_client(&config.tls, CONNECT_TIMEOUT, READ_TIMEOUT, false)?; @@ -281,14 +274,6 @@ async fn assess( repos.len() ); - // R6: on S3 the at-rest hash pin is unavailable — say so loudly. - if storage.backend_name() == "s3" { - println!( - "\nWARNING: target storage is S3 — at-rest hash pin is UNAVAILABLE. \ - verify-before-commit closes TRANSFER integrity only; imported artifacts \ - are unpinned at rest (accepted: import-s3-integrity-at-rest-degraded)." - ); - } // R3: permission import is Artifactory-only — flag Nexus before a run. if matches!(args.source, SourceKind::Nexus) { println!("\nNOTE: --with-permissions is unsupported for Nexus (no permission API)."); @@ -318,13 +303,6 @@ async fn run_import( let auth = read_auth(); let on_s3 = storage.backend_name() == "s3"; - if on_s3 { - tracing::warn!( - "S3 target: at-rest hash pin unavailable — verify-before-commit closes TRANSFER \ - integrity only; imported artifacts are UNPINNED at rest \ - (accepted: import-s3-integrity-at-rest-degraded)" - ); - } let curation = crate::build_curation_engine(config)?; let source = source::build_source( @@ -789,7 +767,7 @@ mod integration_tests { ); assert_eq!(h.storage.get(MAVEN_KEY).await.unwrap().as_ref(), body); assert_eq!( - h.storage.get_pin_hash(MAVEN_KEY).as_deref(), + h.storage.pin(MAVEN_KEY).await.as_deref(), Some(sha.as_str()) ); // Repo marked done; rerun is idempotent (resume skip, no re-download). diff --git a/nora-registry/src/import/transfer.rs b/nora-registry/src/import/transfer.rs index 4ccff81b..dd5321a4 100644 --- a/nora-registry/src/import/transfer.rs +++ b/nora-registry/src/import/transfer.rs @@ -615,7 +615,7 @@ mod tests { o => panic!("expected Imported, got {o:?}"), } assert_eq!(storage.get(KEY).await.unwrap().as_ref(), body); - assert_eq!(storage.get_pin_hash(KEY).as_deref(), Some(sha.as_str())); + assert_eq!(storage.pin(KEY).await.as_deref(), Some(sha.as_str())); } #[tokio::test] diff --git a/nora-registry/src/main.rs b/nora-registry/src/main.rs index fd64fccc..23fcc266 100644 --- a/nora-registry/src/main.rs +++ b/nora-registry/src/main.rs @@ -933,9 +933,6 @@ async fn main() { std::process::exit(2); } match storage.repin(&key, &expected, yes).await { - Ok(storage::RepinOutcome::NoPinStore) => { - println!("Backend has no pin store (S3) — nothing to re-pin."); - } Ok(storage::RepinOutcome::DiskMismatch { disk, expected }) => { error!( key = %key, diff --git a/nora-registry/src/metrics.rs b/nora-registry/src/metrics.rs index 7bf94e63..3e460cd8 100644 --- a/nora-registry/src/metrics.rs +++ b/nora-registry/src/metrics.rs @@ -261,11 +261,10 @@ pub static UPSTREAM_REQUEST_DURATION: LazyLock = LazyLock::new(|| }); /// Wall-clock time of the integrity-verify step on a buffered `Storage::get()` -/// — the `spawn_blocking(pins.verify(..))` call. Includes blocking-pool queue -/// time, so a rising p99 under read load signals pool saturation, not just hash -/// cost. Recorded whenever a pin store is configured (Local backend); a key -/// with no pin returns early inside `verify()` and contributes a near-zero -/// sample. Quantifies the #602 perf question before any change is made (#602). +/// — the `spawn_blocking` re-hash. Includes blocking-pool queue time, so a +/// rising p99 under read load signals pool saturation, not just hash cost. +/// Recorded only for a pinned key; an unpinned one skips the gate and observes +/// nothing. Quantifies the #602 perf question before any change is made (#602). pub static STORAGE_VERIFY_DURATION_SECONDS: LazyLock = LazyLock::new(|| { register_histogram_vec!( "nora_storage_verify_duration_seconds", diff --git a/nora-registry/src/registry/cargo_registry.rs b/nora-registry/src/registry/cargo_registry.rs index b93317f7..2e857fcd 100644 --- a/nora-registry/src/registry/cargo_registry.rs +++ b/nora-registry/src/registry/cargo_registry.rs @@ -1076,12 +1076,15 @@ mod tests { "cargo/index-entries/fa/il/failcrate/0.2.0.json".to_string(), ]) } - async fn get(&self, _key: &str) -> StorageResult { + async fn get(&self, _key: &str) -> StorageResult<(Bytes, Option)> { Err(StorageError::Io(std::io::Error::other( "injected transient read error", ))) } - async fn put(&self, _key: &str, _data: &[u8]) -> StorageResult<()> { + async fn pin(&self, _key: &str) -> Option { + None + } + async fn put(&self, _key: &str, _data: &[u8], _sha256: &str) -> StorageResult<()> { panic!("regenerate must abort before writing a truncated index"); } async fn delete(&self, _key: &str) -> StorageResult<()> { @@ -1099,16 +1102,27 @@ mod tests { fn backend_name(&self) -> &'static str { "failing-get-test" } - async fn put_from_path(&self, _key: &str, _src: &Path) -> StorageResult<()> { + async fn put_from_path( + &self, + _key: &str, + _src: &Path, + _sha256: Option<&str>, + ) -> StorageResult<()> { Ok(()) } async fn get_reader( &self, _key: &str, - ) -> StorageResult<(u64, Pin>)> { + ) -> StorageResult<(u64, Option, Pin>)> + { Err(StorageError::NotFound) } - async fn copy(&self, _src: &str, _dst: &str) -> StorageResult<()> { + async fn copy( + &self, + _src: &str, + _dst: &str, + _sha256: Option<&str>, + ) -> StorageResult<()> { Err(StorageError::NotFound) } } diff --git a/nora-registry/src/registry/docker.rs b/nora-registry/src/registry/docker.rs index 1dad51d7..c82e7711 100644 --- a/nora-registry/src/registry/docker.rs +++ b/nora-registry/src/registry/docker.rs @@ -244,6 +244,9 @@ async fn storage_get_reader_with_fallback( } other => other, } + // Docker blobs are content-addressed: the URL digest is the integrity + // check, so the stored pin is not needed here. + .map(|(size, _pin, reader)| (size, reader)) } /// An `AsyncRead` wrapper that hashes the bytes it streams and, on a SHA-256 @@ -1370,7 +1373,7 @@ async fn download_blob( if fetched._guard.path.is_none() { // Successfully stored — stream from storage match state.storage.get_reader(&key).await { - Ok((size, reader)) => { + Ok((size, _pin, reader)) => { let stream = ReaderStream::new(VerifyingReader::new(reader, &digest)); return Response::builder() @@ -5762,15 +5765,21 @@ mod integration_tests { #[async_trait::async_trait] impl crate::storage::StorageBackend for FailingPrefixBackend { - async fn put(&self, key: &str, data: &[u8]) -> crate::storage::Result<()> { - self.inner.put(key, data).await + async fn put(&self, key: &str, data: &[u8], sha256: &str) -> crate::storage::Result<()> { + self.inner.put(key, data, sha256).await } - async fn get(&self, key: &str) -> crate::storage::Result { + async fn get( + &self, + key: &str, + ) -> crate::storage::Result<(axum::body::Bytes, Option)> { if key.starts_with(&self.fail_prefix) { return Err(crate::storage::StorageError::Network("injected".into())); } self.inner.get(key).await } + async fn pin(&self, key: &str) -> Option { + self.inner.pin(key).await + } async fn delete(&self, key: &str) -> crate::storage::Result<()> { self.inner.delete(key).await } @@ -5789,21 +5798,28 @@ mod integration_tests { fn backend_name(&self) -> &'static str { "failing-prefix-test" } - async fn copy(&self, src: &str, dst: &str) -> crate::storage::Result<()> { - self.inner.copy(src, dst).await + async fn copy( + &self, + src: &str, + dst: &str, + sha256: Option<&str>, + ) -> crate::storage::Result<()> { + self.inner.copy(src, dst, sha256).await } async fn put_from_path( &self, key: &str, src: &std::path::Path, + sha256: Option<&str>, ) -> crate::storage::Result<()> { - self.inner.put_from_path(key, src).await + self.inner.put_from_path(key, src, sha256).await } async fn get_reader( &self, key: &str, ) -> crate::storage::Result<( u64, + Option, std::pin::Pin>, )> { if key.starts_with(&self.fail_prefix) { diff --git a/nora-registry/src/registry/raw.rs b/nora-registry/src/registry/raw.rs index 7574ee3f..eff67314 100644 --- a/nora-registry/src/registry/raw.rs +++ b/nora-registry/src/registry/raw.rs @@ -88,31 +88,32 @@ async fn download( } } - // Conditional GET — If-None-Match - if let Some(inm) = headers - .get(header::IF_NONE_MATCH) - .and_then(|v| v.to_str().ok()) - { - if let Some(stored_hash) = state.storage.get_pin_hash(&key) { - let etag_val = format!("\"{}\"", stored_hash); - if inm.trim() == etag_val || inm.trim() == "*" { - return (StatusCode::NOT_MODIFIED, [(header::ETAG, etag_val)]).into_response(); - } - } - } - // Streamed serve with STREAMING integrity verification. The buffered // `get_verified` gate would hold the whole object in memory — unusable for // multi-GB artifacts — so raw hashes the stream as it is served and // compares against the recorded pin at EOF. On a mismatch the body is // aborted BEFORE its final frame: the client observes a connection error / // Content-Length shortfall instead of a completed corrupt download — - // fail-closed, in streaming form. A key with no pin (object-store backend) - // is served without a cryptographic check, exactly like the buffered - // gate's `Unpinned` arm. - let pin = state.storage.get_pin_hash(&key); + // fail-closed, in streaming form. A key with no pin is served without a + // cryptographic check, exactly like the buffered gate's `Unpinned` arm. + // + // The pin comes back with the reader, so bytes and pin are one object + // version and one backend round-trip. match state.storage.get_reader(&key).await { - Ok((len, reader)) => { + Ok((len, pin, reader)) => { + // Conditional GET — If-None-Match + if let (Some(inm), Some(stored_hash)) = ( + headers + .get(header::IF_NONE_MATCH) + .and_then(|v| v.to_str().ok()), + pin.as_deref(), + ) { + let etag_val = format!("\"{}\"", stored_hash); + if inm.trim() == etag_val || inm.trim() == "*" { + return (StatusCode::NOT_MODIFIED, [(header::ETAG, etag_val)]).into_response(); + } + } + let content_type = guess_content_type(&key); let etag = pin.as_ref().map(|h| format!("\"{}\"", h)); @@ -280,6 +281,56 @@ fn verify_while_streaming( } } +/// Verify an RFC 9530 `Repr-Digest` header against the server-computed sha-256. +/// +/// The header gates the commit but never sets the pin, so the stored pin is +/// always the hash of the bytes the server received. A header without a +/// sha-256 entry is rejected rather than ignored: a client that asked for +/// verification must not get a silent skip. +fn verify_repr_digest(headers: &axum::http::HeaderMap, computed: &str) -> Option { + use base64::{engine::general_purpose::STANDARD, Engine}; + + let value = headers.get("repr-digest").and_then(|v| v.to_str().ok())?; + // Last sha-256 entry wins, per structured-field dictionary semantics. + let Some(b64) = value + .split(',') + .filter_map(|e| e.trim().strip_prefix("sha-256=:")?.strip_suffix(':')) + .next_back() + else { + return Some( + ( + StatusCode::BAD_REQUEST, + "Repr-Digest must carry a sha-256 entry: sha-256=:BASE64:", + ) + .into_response(), + ); + }; + let declared = match STANDARD.decode(b64) { + Ok(bytes) => hex::encode(bytes), + Err(_) => { + return Some( + ( + StatusCode::BAD_REQUEST, + "Repr-Digest sha-256 value is not valid base64", + ) + .into_response(), + ) + } + }; + if declared != computed { + return Some( + ( + StatusCode::BAD_REQUEST, + format!( + "Repr-Digest mismatch: declared sha-256 {declared}, body hashes to {computed}" + ), + ) + .into_response(), + ); + } + None +} + async fn upload( State(state): State, Path(path): Path, @@ -361,6 +412,10 @@ async fn upload( } }; + if let Some(resp) = verify_repr_digest(&headers, &sha256) { + return resp; + } + let if_none_match = headers .get(header::IF_NONE_MATCH) .and_then(|v| v.to_str().ok()) @@ -418,8 +473,7 @@ async fn upload( // If-Match: "" → update only if ETag matches (true, _, Some(etag)) => { - let stored_hash = state.storage.get_pin_hash(&key); - match stored_hash { + match state.storage.pin(&key).await { Some(hash) => { let expected = format!("\"{}\"", hash); if etag == expected { @@ -436,7 +490,7 @@ async fn upload( return (StatusCode::PRECONDITION_FAILED, "ETag mismatch").into_response(); } None => { - // No pin hash available (e.g. S3 backend) — cannot verify + // Stored before pins existed — nothing to compare against. return ( StatusCode::PRECONDITION_FAILED, "ETag not available for this resource", @@ -566,7 +620,7 @@ async fn check_exists(State(state): State, Path(path): Path) - .header(header::CONTENT_LENGTH, meta.size.to_string()) .header(header::CONTENT_TYPE, guess_content_type(&key)) .header(header::CACHE_CONTROL, &state.config.raw.cache_control); - if let Some(hash) = state.storage.get_pin_hash(&key) { + if let Some(hash) = state.storage.pin(&key).await { builder = builder.header(header::ETAG, format!("\"{}\"", hash)); } if meta.modified > 0 { @@ -1045,17 +1099,6 @@ mod integration_tests { let ctx = create_test_context(); send(&ctx.app, Method::PUT, "/raw/etag.txt", b"hello".to_vec()).await; - // The ETag is the hash-pin, recorded fire-and-forget after PUT. Wait for - // it to land so HEAD deterministically sees the ETag — otherwise the - // pin task can be starved under a full parallel suite and the header is - // absent (#603). Polls (fast path = immediate), no fixed sleep. - for _ in 0..200 { - if ctx.state.storage.get_pin_hash("raw/etag.txt").is_some() { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - } - let head = send(&ctx.app, Method::HEAD, "/raw/etag.txt", "").await; assert_eq!(head.status(), StatusCode::OK); let etag = head.headers().get("etag").expect("HEAD must return ETag"); @@ -1346,13 +1389,6 @@ mod integration_tests { b"0123456789".to_vec(), ) .await; - // The ETag is the hash-pin, recorded fire-and-forget after PUT (#603). - for _ in 0..200 { - if ctx.state.storage.get_pin_hash("raw/ifr.bin").is_some() { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - } let head = send(&ctx.app, Method::HEAD, "/raw/ifr.bin", "").await; let etag = head .headers() @@ -1392,6 +1428,141 @@ mod integration_tests { assert_eq!(&body_bytes(fresh).await[..], b"2345"); } + /// The ETag flows are backend-agnostic since the pin became object + /// metadata: HEAD advertises it, a matching `If-None-Match` is a 304, and a + /// conditional overwrite matches on it — all on an object store. + #[tokio::test] + async fn test_raw_conditional_requests_on_object_backend() { + use crate::storage::ObjectStorage; + use sha2::{Digest, Sha256}; + + let ctx = crate::test_helpers::create_test_context_with_storage(Storage::from_backend( + std::sync::Arc::new(ObjectStorage::in_memory()), + )); + + let put = send(&ctx.app, Method::PUT, "/raw/obj.txt", b"v1".to_vec()).await; + assert_eq!(put.status(), StatusCode::CREATED); + + let head = send(&ctx.app, Method::HEAD, "/raw/obj.txt", "").await; + assert_eq!(head.status(), StatusCode::OK); + let etag = head + .headers() + .get("etag") + .expect("object metadata must carry the pin") + .to_str() + .unwrap() + .to_string(); + assert_eq!(etag, format!("\"{}\"", hex::encode(Sha256::digest(b"v1")))); + + let cached = send_with_headers( + &ctx.app, + Method::GET, + "/raw/obj.txt", + vec![("if-none-match", &etag)], + "", + ) + .await; + assert_eq!(cached.status(), StatusCode::NOT_MODIFIED); + + let stale = send_with_headers( + &ctx.app, + Method::PUT, + "/raw/obj.txt", + vec![( + "if-match", + "\"0000000000000000000000000000000000000000000000000000000000000000\"", + )], + b"v2".to_vec(), + ) + .await; + assert_eq!(stale.status(), StatusCode::PRECONDITION_FAILED); + + let overwrite = send_with_headers( + &ctx.app, + Method::PUT, + "/raw/obj.txt", + vec![("if-match", &etag)], + b"v2".to_vec(), + ) + .await; + assert_eq!(overwrite.status(), StatusCode::OK); + + let head = send(&ctx.app, Method::HEAD, "/raw/obj.txt", "").await; + assert_eq!( + head.headers().get("etag").unwrap().to_str().unwrap(), + format!("\"{}\"", hex::encode(Sha256::digest(b"v2"))) + ); + assert_eq!( + &body_bytes(send(&ctx.app, Method::GET, "/raw/obj.txt", "").await).await[..], + b"v2" + ); + } + + #[tokio::test] + async fn test_raw_put_repr_digest() { + use base64::{engine::general_purpose::STANDARD, Engine}; + use sha2::{Digest, Sha256}; + let ctx = create_test_context(); + let good = format!("sha-256=:{}:", STANDARD.encode(Sha256::digest(b"hello"))); + let wrong = format!("sha-256=:{}:", STANDARD.encode(Sha256::digest(b"other"))); + + // Matching digest commits. + let ok = send_with_headers( + &ctx.app, + Method::PUT, + "/raw/rd-ok.txt", + vec![("repr-digest", good.as_str())], + b"hello".to_vec(), + ) + .await; + assert_eq!(ok.status(), StatusCode::CREATED); + + // Mismatch is rejected before anything is stored. + let bad = send_with_headers( + &ctx.app, + Method::PUT, + "/raw/rd-bad.txt", + vec![("repr-digest", wrong.as_str())], + b"hello".to_vec(), + ) + .await; + assert_eq!(bad.status(), StatusCode::BAD_REQUEST); + assert!(ctx.state.storage.get("raw/rd-bad.txt").await.is_err()); + + // Unsupported-algorithm-only header fails closed. + let sha512_only = send_with_headers( + &ctx.app, + Method::PUT, + "/raw/rd-512.txt", + vec![("repr-digest", "sha-512=:AAAA:")], + b"hello".to_vec(), + ) + .await; + assert_eq!(sha512_only.status(), StatusCode::BAD_REQUEST); + + // Malformed base64 is rejected. + let malformed = send_with_headers( + &ctx.app, + Method::PUT, + "/raw/rd-mal.txt", + vec![("repr-digest", "sha-256=:not base64!:")], + b"hello".to_vec(), + ) + .await; + assert_eq!(malformed.status(), StatusCode::BAD_REQUEST); + + // Multi-algorithm dictionary: the sha-256 member is the one verified. + let multi = send_with_headers( + &ctx.app, + Method::PUT, + "/raw/rd-multi.txt", + vec![("repr-digest", format!("sha-512=:AAAA:, {good}").as_str())], + b"hello".to_vec(), + ) + .await; + assert_eq!(multi.status(), StatusCode::CREATED); + } + #[tokio::test] async fn test_raw_cache_control_default() { let ctx = create_test_context(); diff --git a/nora-registry/src/repo_index.rs b/nora-registry/src/repo_index.rs index 7c9d64ae..72d9fb28 100644 --- a/nora-registry/src/repo_index.rs +++ b/nora-registry/src/repo_index.rs @@ -1015,12 +1015,18 @@ mod tests { .cloned() .collect()) } - async fn put(&self, _k: &str, _d: &[u8]) -> crate::storage::Result<()> { + async fn put(&self, _k: &str, _d: &[u8], _sha256: &str) -> crate::storage::Result<()> { Ok(()) } - async fn get(&self, _k: &str) -> crate::storage::Result { + async fn get( + &self, + _k: &str, + ) -> crate::storage::Result<(axum::body::Bytes, Option)> { Err(crate::storage::StorageError::NotFound) } + async fn pin(&self, _k: &str) -> Option { + None + } async fn delete(&self, _k: &str) -> crate::storage::Result<()> { Ok(()) } @@ -1037,6 +1043,7 @@ mod tests { &self, _k: &str, _s: &std::path::Path, + _sha256: Option<&str>, ) -> crate::storage::Result<()> { Ok(()) } @@ -1045,11 +1052,17 @@ mod tests { _k: &str, ) -> crate::storage::Result<( u64, + Option, std::pin::Pin>, )> { Err(crate::storage::StorageError::NotFound) } - async fn copy(&self, _src: &str, _dst: &str) -> crate::storage::Result<()> { + async fn copy( + &self, + _src: &str, + _dst: &str, + _sha256: Option<&str>, + ) -> crate::storage::Result<()> { Err(crate::storage::StorageError::NotFound) } } diff --git a/nora-registry/src/storage/local.rs b/nora-registry/src/storage/local.rs index ab71f70e..2943d2e4 100644 --- a/nora-registry/src/storage/local.rs +++ b/nora-registry/src/storage/local.rs @@ -5,10 +5,16 @@ use async_trait::async_trait; use axum::body::Bytes; use std::path::{Path, PathBuf}; use std::pin::Pin; +use std::sync::Arc; use tokio::fs; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; use super::{FileMeta, Result, StorageBackend, StorageError}; +use crate::hash_pin_store::HashPinStore; + +/// The hash-pin sidecar is backend bookkeeping, not a stored artifact: it is +/// held out of listings and the size gauge, like the `tmp/` staging directory. +const PIN_FILE: &str = ".nora-pins.ndjson"; /// Monotonic counter for unique temp file names (atomic — no collisions). static TMP_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); @@ -28,16 +34,41 @@ async fn sync_parent_dir(path: &Path) -> Result<()> { Ok(()) } -/// Local filesystem storage backend (zero-config default) +/// Local filesystem storage backend (zero-config default). Hash pins live in an +/// NDJSON sidecar next to the artifacts. pub struct LocalStorage { base_path: PathBuf, + pins: Arc, } impl LocalStorage { pub fn new(path: &str) -> Self { - Self { - base_path: PathBuf::from(path), - } + let base_path = PathBuf::from(path); + let pins = Arc::new(HashPinStore::new(base_path.join(PIN_FILE))); + Self { base_path, pins } + } + + /// Record `sha256` for `key`, on the blocking pool (the append is + /// filesystem I/O). Fails closed: an artifact whose pin did not reach the + /// disk would silently downgrade to open-world after the next restart — the + /// #582/#604 bypass — so the write reports failure instead. + /// + /// On an immutable registry the client's retry hits the 409 guard and never + /// re-runs this, so the orphaned body stays unpinned until an operator + /// `repin`s it — still strictly better than a silent success. + async fn record_pin(&self, key: &str, sha256: &str) -> Result<()> { + let pins = Arc::clone(&self.pins); + let key_owned = key.to_string(); + let hash = sha256.to_ascii_lowercase(); + tokio::task::spawn_blocking(move || pins.record_hash(&key_owned, &hash)) + .await + .unwrap_or_else(|e| Err(std::io::Error::other(e.to_string()))) + .map_err(|e| { + tracing::error!(error = %e, key = %key, "hash-pin record failed"); + StorageError::Io(std::io::Error::other(format!( + "hash-pin record failed: {e}" + ))) + }) } fn key_to_path(&self, key: &str) -> PathBuf { @@ -52,7 +83,7 @@ impl LocalStorage { if path.is_file() { if let Ok(rel_path) = path.strip_prefix(base) { let key = rel_path.to_string_lossy().replace('\\', "/"); - if key.starts_with(prefix) || prefix.is_empty() { + if key != PIN_FILE && (key.starts_with(prefix) || prefix.is_empty()) { results.push(key); } } @@ -82,7 +113,7 @@ impl LocalStorage { if metadata.is_file() { if let Ok(rel_path) = path.strip_prefix(base) { let key = rel_path.to_string_lossy().replace('\\', "/"); - if key.starts_with(prefix) || prefix.is_empty() { + if key != PIN_FILE && (key.starts_with(prefix) || prefix.is_empty()) { let modified = metadata .modified() .ok() @@ -108,7 +139,7 @@ impl LocalStorage { #[async_trait] impl StorageBackend for LocalStorage { - async fn put(&self, key: &str, data: &[u8]) -> Result<()> { + async fn put(&self, key: &str, data: &[u8], sha256: &str) -> Result<()> { let path = self.key_to_path(key); // Create parent directories @@ -137,10 +168,11 @@ impl StorageBackend for LocalStorage { if write_result.is_err() { let _ = fs::remove_file(&tmp).await; } - write_result + write_result?; + self.record_pin(key, sha256).await } - async fn get(&self, key: &str) -> Result { + async fn get(&self, key: &str) -> Result<(Bytes, Option)> { let path = self.key_to_path(key); let mut file = fs::File::open(&path).await.map_err(|e| { @@ -154,7 +186,11 @@ impl StorageBackend for LocalStorage { let mut buffer = Vec::new(); file.read_to_end(&mut buffer).await?; - Ok(Bytes::from(buffer)) + Ok((Bytes::from(buffer), self.pins.get(key))) + } + + async fn pin(&self, key: &str) -> Option { + self.pins.get(key) } async fn delete(&self, key: &str) -> Result<()> { @@ -168,6 +204,22 @@ impl StorageBackend for LocalStorage { } })?; + // A lost tombstone is fail-safe — a stale pin at worst yields a future + // IntegrityViolation, healable via `repin` — so the delete still + // reports success: the authoritative action (byte removal) is done. + let pins = Arc::clone(&self.pins); + let key_owned = key.to_string(); + if let Err(e) = tokio::task::spawn_blocking(move || pins.remove(&key_owned)) + .await + .unwrap_or_else(|e| Err(std::io::Error::other(e.to_string()))) + { + tracing::warn!( + error = %e, + key = %key, + "hash-pin tombstone write failed; stale pin left (repin to heal)" + ); + } + Ok(()) } @@ -248,6 +300,9 @@ impl StorageBackend for LocalStorage { for entry in entries.flatten() { let path = entry.path(); if path.is_file() { + if is_root && path.file_name().is_some_and(|n| n == PIN_FILE) { + continue; + } total += entry.metadata().map(|m| m.len()).unwrap_or(0); } else if path.is_dir() { // `/tmp/` holds in-flight streamed uploads — @@ -272,7 +327,7 @@ impl StorageBackend for LocalStorage { "local" } - async fn put_from_path(&self, key: &str, src: &Path) -> Result<()> { + async fn put_from_path(&self, key: &str, src: &Path, sha256: Option<&str>) -> Result<()> { let dest = self.key_to_path(key); if let Some(parent) = dest.parent() { fs::create_dir_all(parent).await?; @@ -283,7 +338,6 @@ impl StorageBackend for LocalStorage { Ok(()) => { // Durability: make the rename's directory entry survive power-loss. sync_parent_dir(&dest).await?; - Ok(()) } Err(e) if e.raw_os_error() == Some(18 /* EXDEV */) => { let mut reader = fs::File::open(src).await?; @@ -315,13 +369,16 @@ impl StorageBackend for LocalStorage { } copy_result?; let _ = fs::remove_file(src).await; - Ok(()) } - Err(e) => Err(StorageError::Io(e)), + Err(e) => return Err(StorageError::Io(e)), + } + match sha256 { + Some(hash) => self.record_pin(key, hash).await, + None => Ok(()), } } - async fn copy(&self, src: &str, dst: &str) -> Result<()> { + async fn copy(&self, src: &str, dst: &str, sha256: Option<&str>) -> Result<()> { let src_path = self.key_to_path(src); let dst_path = self.key_to_path(dst); if let Some(parent) = dst_path.parent() { @@ -337,8 +394,10 @@ impl StorageBackend for LocalStorage { other => other, }; match linked { - Ok(()) => sync_parent_dir(&dst_path).await, - Err(e) if e.kind() == std::io::ErrorKind::NotFound => Err(StorageError::NotFound), + Ok(()) => sync_parent_dir(&dst_path).await?, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + return Err(StorageError::NotFound) + } // Cross-device, or a filesystem without links — copy the bytes. Err(_) => { fs::copy(&src_path, &dst_path).await.map_err(|e| { @@ -348,12 +407,22 @@ impl StorageBackend for LocalStorage { StorageError::Io(e) } })?; - sync_parent_dir(&dst_path).await + sync_parent_dir(&dst_path).await?; } } + match sha256 + .map(str::to_ascii_lowercase) + .or_else(|| self.pins.get(src)) + { + Some(hash) => self.record_pin(dst, &hash).await, + None => Ok(()), + } } - async fn get_reader(&self, key: &str) -> Result<(u64, Pin>)> { + async fn get_reader( + &self, + key: &str, + ) -> Result<(u64, Option, Pin>)> { let path = self.key_to_path(key); let file = fs::File::open(&path).await.map_err(|e| { if e.kind() == std::io::ErrorKind::NotFound { @@ -363,7 +432,7 @@ impl StorageBackend for LocalStorage { } })?; let meta = file.metadata().await?; - Ok((meta.len(), Box::pin(file))) + Ok((meta.len(), self.pins.get(key), Box::pin(file))) } async fn get_range( @@ -394,15 +463,28 @@ impl StorageBackend for LocalStorage { #[allow(clippy::unwrap_used)] mod tests { use super::*; + use sha2::{Digest, Sha256}; use tempfile::TempDir; + /// The backend pins what it stores, so every test write carries the digest + /// of its own bytes. + async fn put(storage: &LocalStorage, key: &str, data: &[u8]) -> Result<()> { + storage + .put(key, data, &hex::encode(Sha256::digest(data))) + .await + } + + async fn get(storage: &LocalStorage, key: &str) -> Result { + storage.get(key).await.map(|(data, _pin)| data) + } + #[tokio::test] async fn test_put_and_get() { let temp_dir = TempDir::new().unwrap(); let storage = LocalStorage::new(temp_dir.path().to_str().unwrap()); - storage.put("test/key", b"test data").await.unwrap(); - let data = storage.get("test/key").await.unwrap(); + put(&storage, "test/key", b"test data").await.unwrap(); + let data = get(&storage, "test/key").await.unwrap(); assert_eq!(&*data, b"test data"); } @@ -411,7 +493,7 @@ mod tests { let temp_dir = TempDir::new().unwrap(); let storage = LocalStorage::new(temp_dir.path().to_str().unwrap()); - let result = storage.get("nonexistent").await; + let result = get(&storage, "nonexistent").await; assert!(matches!(result, Err(StorageError::NotFound))); } @@ -420,9 +502,9 @@ mod tests { let temp_dir = TempDir::new().unwrap(); let storage = LocalStorage::new(temp_dir.path().to_str().unwrap()); - storage.put("docker/image/blob1", b"data1").await.unwrap(); - storage.put("docker/image/blob2", b"data2").await.unwrap(); - storage.put("maven/artifact", b"data3").await.unwrap(); + put(&storage, "docker/image/blob1", b"data1").await.unwrap(); + put(&storage, "docker/image/blob2", b"data2").await.unwrap(); + put(&storage, "maven/artifact", b"data3").await.unwrap(); let docker_keys = storage.list("docker/").await.unwrap(); assert_eq!(docker_keys.len(), 2); @@ -437,7 +519,7 @@ mod tests { let temp_dir = TempDir::new().unwrap(); let storage = LocalStorage::new(temp_dir.path().to_str().unwrap()); - storage.put("test", b"12345").await.unwrap(); + put(&storage, "test", b"12345").await.unwrap(); let meta = storage.stat("test").await.unwrap(); assert_eq!(meta.size, 5); assert!(meta.modified > 0); @@ -491,8 +573,8 @@ mod tests { let temp_dir = TempDir::new().unwrap(); let storage = LocalStorage::new(temp_dir.path().to_str().unwrap()); - storage.put("a/b/c/d/e/file", b"deep").await.unwrap(); - let data = storage.get("a/b/c/d/e/file").await.unwrap(); + put(&storage, "a/b/c/d/e/file", b"deep").await.unwrap(); + let data = get(&storage, "a/b/c/d/e/file").await.unwrap(); assert_eq!(&*data, b"deep"); } @@ -501,10 +583,10 @@ mod tests { let temp_dir = TempDir::new().unwrap(); let storage = LocalStorage::new(temp_dir.path().to_str().unwrap()); - storage.put("key", b"original").await.unwrap(); - storage.put("key", b"updated").await.unwrap(); + put(&storage, "key", b"original").await.unwrap(); + put(&storage, "key", b"updated").await.unwrap(); - let data = storage.get("key").await.unwrap(); + let data = get(&storage, "key").await.unwrap(); assert_eq!(&*data, b"updated"); } @@ -525,7 +607,7 @@ mod tests { let s = storage.clone(); handles.push(tokio::spawn(async move { let data = vec![i; 1024]; - s.put("shared/key", &data).await + put(&s, "shared/key", &data).await })); } @@ -533,7 +615,7 @@ mod tests { h.await.expect("task panicked").expect("put failed"); } - let data = storage.get("shared/key").await.expect("get failed"); + let data = get(&storage, "shared/key").await.expect("get failed"); assert_eq!(data.len(), 1024); let first = data[0]; assert!( @@ -552,7 +634,7 @@ mod tests { let s = storage.clone(); handles.push(tokio::spawn(async move { let key = format!("key/{}", i); - s.put(&key, format!("data-{}", i).as_bytes()).await + put(&s, &key, format!("data-{}", i).as_bytes()).await })); } @@ -562,7 +644,7 @@ mod tests { for i in 0..10u32 { let key = format!("key/{}", i); - let data = storage.get(&key).await.expect("get failed"); + let data = get(&storage, &key).await.expect("get failed"); assert_eq!(&*data, format!("data-{}", i).as_bytes()); } } @@ -582,8 +664,7 @@ mod tests { let temp_dir = TempDir::new().unwrap(); let storage = std::sync::Arc::new(LocalStorage::new(temp_dir.path().to_str().unwrap())); - storage - .put("rw/key", &vec![0u8; LEN]) + put(&storage, "rw/key", &vec![0u8; LEN]) .await .expect("seed put"); @@ -596,7 +677,7 @@ mod tests { // visible mix of the two. for i in 0..100u32 { let byte = if i % 2 == 0 { 0u8 } else { 1u8 }; - sw.put("rw/key", &vec![byte; LEN]) + put(&sw, "rw/key", &vec![byte; LEN]) .await .expect("put failed"); } @@ -608,7 +689,7 @@ mod tests { let reader = tokio::spawn(async move { // Spin for the whole write loop so the concurrent window is exercised. while !dr.load(Ordering::Acquire) { - match sr.get("rw/key").await { + match get(&sr, "rw/key").await { Ok(data) => { assert_eq!(data.len(), LEN, "torn/partial read: wrong object length"); let first = data[0]; @@ -631,7 +712,7 @@ mod tests { reader.await.expect("reader panicked"); // Final state is a complete, uniform object. - let data = storage.get("rw/key").await.expect("final get"); + let data = get(&storage, "rw/key").await.expect("final get"); assert_eq!(data.len(), LEN); let first = data[0]; assert!( @@ -652,8 +733,8 @@ mod tests { let temp_dir = TempDir::new().unwrap(); let storage = LocalStorage::new(temp_dir.path().to_str().unwrap()); - storage.put("a/file1", b"hello").await.unwrap(); // 5 bytes - storage.put("b/file2", b"world!").await.unwrap(); // 6 bytes + put(&storage, "a/file1", b"hello").await.unwrap(); // 5 bytes + put(&storage, "b/file2", b"world!").await.unwrap(); // 6 bytes let size = storage.total_size().await; assert_eq!(size, 11); @@ -664,8 +745,8 @@ mod tests { let temp_dir = TempDir::new().unwrap(); let storage = LocalStorage::new(temp_dir.path().to_str().unwrap()); - storage.put("file1", b"12345").await.unwrap(); - storage.put("file2", b"67890").await.unwrap(); + put(&storage, "file1", b"12345").await.unwrap(); + put(&storage, "file2", b"67890").await.unwrap(); assert_eq!(storage.total_size().await, 10); storage.delete("file1").await.unwrap(); @@ -677,7 +758,7 @@ mod tests { let temp_dir = TempDir::new().unwrap(); let storage = std::sync::Arc::new(LocalStorage::new(temp_dir.path().to_str().unwrap())); - storage.put("del/key", b"ephemeral").await.expect("put"); + put(&storage, "del/key", b"ephemeral").await.expect("put"); let mut handles = Vec::new(); for _ in 0..10 { @@ -692,7 +773,7 @@ mod tests { } assert!(matches!( - storage.get("del/key").await, + get(&storage, "del/key").await, Err(crate::storage::StorageError::NotFound) )); } diff --git a/nora-registry/src/storage/mod.rs b/nora-registry/src/storage/mod.rs index 551b838e..d956fd4c 100644 --- a/nora-registry/src/storage/mod.rs +++ b/nora-registry/src/storage/mod.rs @@ -7,13 +7,12 @@ mod object; pub use local::LocalStorage; pub use object::ObjectStorage; -use crate::hash_pin_store::HashPinStore; use crate::metrics::{STORAGE_GET_BYTES, STORAGE_OPERATIONS, STORAGE_VERIFY_DURATION_SECONDS}; use crate::validation::{validate_storage_key, ValidationError}; use async_trait::async_trait; use axum::body::Bytes; use sha2::{Digest, Sha256}; -use std::path::{Path, PathBuf}; +use std::path::Path; use std::pin::Pin; use std::sync::Arc; use thiserror::Error; @@ -65,6 +64,10 @@ fn registry_label(key: &str) -> &str { } } +fn sha256_hex(data: &[u8]) -> String { + hex::encode(Sha256::digest(data)) +} + /// Outcome of [`Storage::repin`] — an operator integrity-recovery action (#601). #[derive(Debug, PartialEq, Eq)] pub enum RepinOutcome { @@ -79,15 +82,24 @@ pub enum RepinOutcome { /// is genuinely corrupt/tampered — re-pin cannot heal it; restore from /// backup. The pin is left unchanged. DiskMismatch { disk: String, expected: String }, - /// This backend has no pin store (S3) — there is nothing to re-pin. - NoPinStore, } -/// Storage backend trait +/// Storage backend trait. +/// +/// Every backend owns the SHA-256 integrity pin of the artifacts it stores and +/// keeps it beside the bytes — an NDJSON sidecar on the local filesystem, +/// user-defined object metadata on an object store. `sha256` arguments are +/// lowercase hex (64 chars). #[async_trait] pub trait StorageBackend: Send + Sync { - async fn put(&self, key: &str, data: &[u8]) -> Result<()>; - async fn get(&self, key: &str) -> Result; + /// Store `data` under `key`, pinned to `sha256`. Fails closed: a body whose + /// pin cannot be recorded is reported as a failed write (#582/#604). + async fn put(&self, key: &str, data: &[u8], sha256: &str) -> Result<()>; + /// Bytes plus the recorded pin, from ONE backend round-trip. `None` means + /// the object carries no pin (open-world). + async fn get(&self, key: &str) -> Result<(Bytes, Option)>; + /// The recorded pin for `key` without reading the bytes. + async fn pin(&self, key: &str) -> Option; async fn delete(&self, key: &str) -> Result<()>; async fn list(&self, prefix: &str) -> Result>; async fn stat(&self, key: &str) -> Option; @@ -114,28 +126,36 @@ pub trait StorageBackend: Send + Sync { fn backend_name(&self) -> &'static str; /// Refresh any cached size data. No-op for backends without caching. async fn refresh_total_size(&self) {} - /// Move or copy a file from `src` into storage under `key`. + /// Move or copy a file from `src` into storage under `key`, pinned to + /// `sha256` when the caller computed one (streaming paths do; legacy + /// callers that verified integrity separately pass `None`, leaving the + /// object open-world). /// /// Local backend: atomic `rename`, with streaming copy fallback on EXDEV. - /// S3 backend: multipart upload from file. + /// Object store: multipart upload from file. /// The caller is responsible for deleting `src` on error. - async fn put_from_path(&self, key: &str, src: &Path) -> Result<()>; + async fn put_from_path(&self, key: &str, src: &Path, sha256: Option<&str>) -> Result<()>; /// Server-side copy of `src` to `dst` inside the backend — the bytes never - /// transit this process. + /// transit this process. `sha256` is the digest of the copied bytes when the + /// caller knows it; otherwise `dst` inherits the pin of `src`. /// /// Returns [`StorageError::NotFound`] when `src` does not exist; an existing /// `dst` is overwritten. - async fn copy(&self, src: &str, dst: &str) -> Result<()>; + async fn copy(&self, src: &str, dst: &str, sha256: Option<&str>) -> Result<()>; /// Open an artifact for streaming read without loading it into memory (#580). /// - /// Returns `(size_bytes, reader)`. The caller converts the reader to a - /// streaming HTTP response via `ReaderStream` + `Body::from_stream()`. + /// Returns `(size_bytes, pin, reader)`. The caller converts the reader to a + /// streaming HTTP response via `ReaderStream` + `Body::from_stream()`, and + /// checks the pin as it streams. /// /// Local backend: `tokio::fs::File::open` + metadata. - /// S3 backend: `object_store::get` → byte-stream wrapped in `StreamReader`. - async fn get_reader(&self, key: &str) -> Result<(u64, Pin>)>; + /// Object store: `object_store::get` → byte-stream wrapped in `StreamReader`. + async fn get_reader( + &self, + key: &str, + ) -> Result<(u64, Option, Pin>)>; /// Stream the inclusive byte range `[start, end]` of an object, returning the object's /// total size and a reader over exactly those bytes. The default reads from the start and @@ -147,7 +167,7 @@ pub trait StorageBackend: Send + Sync { end: u64, ) -> Result<(u64, Pin>)> { use tokio::io::AsyncReadExt; - let (size, mut reader) = self.get_reader(key).await?; + let (size, _, mut reader) = self.get_reader(key).await?; let mut to_skip = start; let mut buf = [0u8; 64 * 1024]; while to_skip > 0 { @@ -164,18 +184,18 @@ pub trait StorageBackend: Send + Sync { } /// Storage wrapper for dynamic dispatch with integrity verification. +/// +/// Owns key validation, metrics and the fail-closed verify gate; the pin itself +/// belongs to the backend, which stores it beside the bytes. #[derive(Clone)] pub struct Storage { inner: Arc, - pin_store: Option>, } impl Storage { pub fn new_local(path: &str) -> Self { - let pin_path = PathBuf::from(path).join(".nora-pins.ndjson"); Self { inner: Arc::new(LocalStorage::new(path)), - pin_store: Some(Arc::new(HashPinStore::new(pin_path))), } } @@ -187,9 +207,6 @@ impl Storage { secret_key: Option<&str>, virtual_hosted: bool, ) -> Self { - tracing::warn!( - "Hash pin store disabled for S3 backend — integrity verification unavailable" - ); Self { inner: Arc::new(ObjectStorage::new( s3_url, @@ -199,7 +216,6 @@ impl Storage { secret_key, virtual_hosted, )), - pin_store: None, } } @@ -208,93 +224,45 @@ impl Storage { service_account_path: Option<&str>, base_url: Option<&str>, ) -> Self { - tracing::warn!( - "Hash pin store disabled for GCS backend — integrity verification unavailable" - ); Self { inner: Arc::new(ObjectStorage::new_gcs( bucket, service_account_path, base_url, )), - pin_store: None, } } /// Test-only: wrap an arbitrary backend so unit tests can inject behaviour /// the real backends can't easily produce — e.g. a `stat`-failing backend /// driving GC's fail-closed "age unknown → keep and count" branch (#610). - /// No pin store. #[cfg(test)] pub(crate) fn from_backend(inner: Arc) -> Self { - Self { - inner, - pin_store: None, - } + Self { inner } } pub async fn put(&self, key: &str, data: &[u8]) -> Result<()> { validate_storage_key(key)?; - match self.inner.put(key, data).await { + // A buffered body can be large; hashing it inline would stall the tokio + // worker for the hash duration, so the digest is computed on the + // blocking pool. The backend records it together with the bytes, so a + // completed `put()` is never readable-but-unpinned (#604). + let buffered = data.to_vec(); + let hash = match tokio::task::spawn_blocking(move || sha256_hex(&buffered)).await { + Ok(h) => h, + Err(e) => { + STORAGE_OPERATIONS + .with_label_values(&["put", "error"]) + .inc(); + tracing::error!(error = %e, key = %key, "hash task for put failed"); + return Err(StorageError::Io(std::io::Error::other(format!( + "hash task failed: {e}" + )))); + } + }; + match self.inner.put(key, data, &hash).await { Ok(()) => { STORAGE_OPERATIONS.with_label_values(&["put", "ok"]).inc(); - if let Some(ref pins) = self.pin_store { - let pins = Arc::clone(pins); - let key_owned = key.to_string(); - let data_owned = data.to_vec(); - // Await the pin record (SHA-256 + ndjson append, offloaded - // from the tokio worker) so `put()` does not return until the - // pin is durable. Previously this was fire-and-forget, which - // left a window where the artifact was readable but unpinned — - // a `get()` after a completed `put()` could serve it - // unverified (#604) — and silently dropped the pin if the task - // panicked. Fail-closed (mirroring `get()`): if integrity - // cannot be recorded, the write reports failure rather than - // leaving an unverifiable artifact. - // - // NOTE: a `get()` racing *during* an in-flight `put()` (between - // the inner write and this record) can still briefly observe - // the artifact unpinned. Fully closing that requires - // serializing get/put per key; it is benign (it serves NORA's - // own just-written bytes) and out of scope here. - // `record` now returns its I/O result: handle the inner - // failure (ENOSPC/EACCES/EIO/read-only FS) the same way as a - // panicked task — fail closed. Previously that error was - // swallowed inside `record`, so `put()` returned Ok while the - // pin never reached disk, silently downgrading the key to - // open-world after the next restart (the #582/#604 bypass). - // - // NOTE (immutable registries): `self.inner.put` already - // succeeded, so on an immutable registry the client's retry - // hits the immutability guard (409) and never re-runs this pin - // write — the orphaned body stays unpinned until an operator - // `repin`s it. That is still strictly better than the prior - // silent success and is the documented recovery path; - // auto-cleanup of the orphan body is a separate follow-up. - match tokio::task::spawn_blocking(move || pins.record(&key_owned, &data_owned)) - .await - { - Ok(Ok(())) => {} - Ok(Err(e)) => { - STORAGE_OPERATIONS - .with_label_values(&["put", "pin_error"]) - .inc(); - tracing::error!(error = %e, key = %key, "hash-pin record failed"); - return Err(StorageError::Io(std::io::Error::other(format!( - "hash-pin record failed: {e}" - )))); - } - Err(e) => { - STORAGE_OPERATIONS - .with_label_values(&["put", "pin_error"]) - .inc(); - tracing::error!(error = %e, key = %key, "hash-pin record task panicked"); - return Err(StorageError::Io(std::io::Error::other(format!( - "hash-pin record failed: {e}" - )))); - } - } - } Ok(()) } Err(e) => { @@ -306,82 +274,86 @@ impl Storage { } } - pub async fn get(&self, key: &str) -> Result { + /// Buffered read through the fail-closed integrity gate, returning the bytes + /// and the pin they were checked against (`None` = open-world key). + async fn get_pinned(&self, key: &str) -> Result<(Bytes, Option)> { validate_storage_key(key)?; - match self.inner.get(key).await { - Ok(data) => { - STORAGE_OPERATIONS.with_label_values(&["get", "ok"]).inc(); - let label = registry_label(key); - STORAGE_GET_BYTES - .with_label_values(&[label]) - .observe(data.len() as f64); - if let Some(ref pins) = self.pin_store { - let pins = Arc::clone(pins); - let key_owned = key.to_string(); - let data_ref = data.clone(); - // SHA-256 verification — offloaded from the tokio worker. - // A buffered `get()` may hold a large artifact in memory; - // hashing it inline would stall the async worker for the - // hash duration, so we keep the blocking pool. The panic - // path is handled fail-closed below — see #582. - // - // INVARIANT (#582): a *positive* verify result is NEVER - // cached — the hash is recomputed on every read. Caching - // "verified" by mtime/size would re-open the bypass #582 - // closed (bit-rot does not bump mtime; an on-disk tamperer - // can forge it via `utimes`). The recompute is the - // deliberate cost of fail-closed delivery; #602 instruments - // that cost via STORAGE_VERIFY_DURATION_SECONDS rather than - // weakening the guarantee. - let verify_start = std::time::Instant::now(); - let outcome = - tokio::task::spawn_blocking(move || pins.verify(&key_owned, &data_ref)) - .await; - STORAGE_VERIFY_DURATION_SECONDS - .with_label_values(&[label]) - .observe(verify_start.elapsed().as_secs_f64()); - match outcome { - // Genuine hash mismatch — tampering or on-disk corruption. - // Fail-closed: never serve the tampered bytes (#582). - Ok(false) => { - STORAGE_OPERATIONS - .with_label_values(&["get", "integrity_fail"]) - .inc(); - tracing::error!( - key = %key, - "integrity violation: refusing to serve tampered artifact" - ); - return Err(StorageError::IntegrityViolation); - } - // Verification task itself panicked. We cannot prove the - // bytes are intact, so fail-closed too — a crashed - // verifier must not become an integrity bypass (#582). - Err(e) => { - STORAGE_OPERATIONS - .with_label_values(&["get", "verify_error"]) - .inc(); - tracing::error!( - error = %e, - key = %key, - "hash verification task failed: refusing to serve unverified artifact" - ); - return Err(StorageError::IntegrityViolation); - } - // Hash matched, or no pin exists for this key (open-world). - Ok(true) => {} - } - } - Ok(data) - } + let (data, pin) = match self.inner.get(key).await { + Ok(v) => v, Err(e) => { STORAGE_OPERATIONS .with_label_values(&["get", "error"]) .inc(); - Err(e) + return Err(e); + } + }; + STORAGE_OPERATIONS.with_label_values(&["get", "ok"]).inc(); + let label = registry_label(key); + STORAGE_GET_BYTES + .with_label_values(&[label]) + .observe(data.len() as f64); + + let Some(expected) = pin else { + return Ok((data, None)); + }; + + // SHA-256 verification — offloaded from the tokio worker for the same + // reason as `put`. The panic path is handled fail-closed below (#582). + // + // INVARIANT (#582): a *positive* verify result is NEVER cached — the + // hash is recomputed on every read. Caching "verified" by mtime/size + // would re-open the bypass #582 closed (bit-rot does not bump mtime; an + // at-rest tamperer can forge it via `utimes`). The recompute is the + // deliberate cost of fail-closed delivery; #602 instruments that cost + // via STORAGE_VERIFY_DURATION_SECONDS rather than weakening the + // guarantee. + let bytes = data.clone(); + let verify_start = std::time::Instant::now(); + let outcome = tokio::task::spawn_blocking(move || sha256_hex(&bytes)).await; + STORAGE_VERIFY_DURATION_SECONDS + .with_label_values(&[label]) + .observe(verify_start.elapsed().as_secs_f64()); + match outcome { + Ok(actual) if actual == expected => Ok((data, Some(expected))), + // Genuine hash mismatch — tampering or at-rest corruption. + // Fail-closed: never serve the tampered bytes (#582). + Ok(actual) => { + STORAGE_OPERATIONS + .with_label_values(&["get", "integrity_fail"]) + .inc(); + tracing::error!( + key = %key, + expected = %expected, + actual = %actual, + "INTEGRITY VIOLATION: refusing to serve tampered artifact" + ); + Err(StorageError::IntegrityViolation) + } + // Verification task itself panicked. We cannot prove the bytes are + // intact, so fail closed too — a crashed verifier must not become an + // integrity bypass (#582). + Err(e) => { + STORAGE_OPERATIONS + .with_label_values(&["get", "verify_error"]) + .inc(); + tracing::error!( + error = %e, + key = %key, + "hash verification task failed: refusing to serve unverified artifact" + ); + Err(StorageError::IntegrityViolation) } } } + pub async fn get(&self, key: &str) -> Result { + // Validate locally as the first act, like every Storage wrapper method: a choke point + // that does not depend on the transitive get_pinned() validation below surviving a future + // refactor (trust-boundary invariant — mirrors get_verified). + validate_storage_key(key)?; + self.get_pinned(key).await.map(|(data, _)| data) + } + /// Buffered, integrity-gated read returning a compile-time integrity /// witness (typestate pilot — see [`crate::verified`]). /// @@ -392,9 +364,9 @@ impl Storage { /// - a pin existed and the bytes matched it → [`GateOutcome::Verified`] /// carrying a `Blob` (a proof the bytes hash to the recorded /// pin); - /// - no pin existed for the key, or this backend has no pin store (S3) → - /// [`GateOutcome::Unpinned`], so a caller cannot mistake the open-world - /// case for a verified read. + /// - the object carries no pin (written before pins, or stored without a + /// digest) → [`GateOutcome::Unpinned`], so a caller cannot mistake the + /// open-world case for a verified read. /// /// [`GateOutcome::Verified`]: nora_registry::verified::GateOutcome::Verified /// [`GateOutcome::Unpinned`]: nora_registry::verified::GateOutcome::Unpinned @@ -411,15 +383,15 @@ impl Storage { // below surviving a future refactor (trust-boundary invariant). validate_storage_key(key)?; use nora_registry::verified::{Blob, GateOutcome}; - // Reuse the fail-closed gate: get() returns Ok only after a digest - // match (Ok(true)) or the open-world / no-pin branch. - let data = self.get(key).await?; - match self.get_pin_hash(key) { + // Reuse the fail-closed gate: it returns Ok only after a digest match or + // on the open-world / no-pin branch. + let (data, pin) = self.get_pinned(key).await?; + match pin { Some(pin) => match Blob::verify(data, &pin) { Ok(blob) => Ok(GateOutcome::Verified(blob)), - // get() already verified these bytes against the same in-memory - // pin, so a mismatch here is unreachable in practice; treat it - // as a tamper signal and fail closed rather than downgrade. + // The gate already verified these bytes against the same pin, so + // a mismatch here is unreachable in practice; treat it as a + // tamper signal and fail closed rather than downgrade. Err(_) => Err(StorageError::IntegrityViolation), }, None => Ok(GateOutcome::Unpinned(Blob::raw(data))), @@ -433,34 +405,6 @@ impl Storage { STORAGE_OPERATIONS .with_label_values(&["delete", "ok"]) .inc(); - if let Some(ref pins) = self.pin_store { - let pins = Arc::clone(pins); - let key_owned = key.to_string(); - // Await the tombstone write so a failure is observable rather - // than fire-and-forget. A lost tombstone is fail-safe (a stale - // pin at worst yields a future IntegrityViolation, healable via - // `repin`), so `delete()` still reports success — the - // authoritative action (byte removal) already succeeded. - match tokio::task::spawn_blocking(move || pins.remove(&key_owned)).await { - Ok(Ok(())) => {} - Ok(Err(e)) => { - STORAGE_OPERATIONS - .with_label_values(&["delete", "pin_error"]) - .inc(); - tracing::warn!( - error = %e, - key = %key, - "hash-pin tombstone write failed; stale pin left (repin to heal)" - ); - } - Err(e) => { - STORAGE_OPERATIONS - .with_label_values(&["delete", "pin_error"]) - .inc(); - tracing::warn!(error = %e, key = %key, "hash-pin tombstone task panicked"); - } - } - } Ok(()) } Err(e) => { @@ -517,16 +461,20 @@ impl Storage { self.inner.backend_name() } - /// Look up the pinned SHA-256 hash for a storage key (None if pin store is disabled or key is unknown). - pub fn get_pin_hash(&self, key: &str) -> Option { - self.pin_store.as_ref().and_then(|p| p.get(key)) + /// The recorded SHA-256 pin for a storage key, or `None` when the object + /// carries no pin (written before pins, or stored without a digest). + pub async fn pin(&self, key: &str) -> Option { + if validate_storage_key(key).is_err() { + return None; + } + self.inner.pin(key).await } /// Operator recovery for an artifact whose hash pin no longer matches its /// stored bytes (#601). Reads the raw bytes *bypassing* verification — the /// whole point, since [`Storage::get`] fails closed on the very mismatch we /// are recovering from — and updates the pin to `expected` **only if the - /// on-disk bytes already hash to `expected`**. + /// stored bytes already hash to `expected`**. /// /// `expected` is the SHA-256 the operator independently knows to be /// canonical for this key (from a CI manifest, upstream checksum, lockfile, @@ -534,48 +482,38 @@ impl Storage { /// integrity bypass: a plain "recompute the hash from disk" would let /// corrupted or tampered bytes silently re-bless themselves, re-opening the /// hole #582 closed. By demanding `disk == expected`, re-pin can only ever - /// set the pin to a hash the disk *already* has **and** the operator has - /// vouched for. If the disk is genuinely corrupt (`disk != expected`) it - /// refuses — re-pin cannot heal corruption; the operator must restore from - /// backup first. + /// set the pin to a hash the stored bytes *already* have **and** the + /// operator has vouched for. If the bytes are genuinely corrupt + /// (`disk != expected`) it refuses — re-pin cannot heal corruption; the + /// operator must restore from backup first. /// /// `apply == false` is a dry run (computes and compares, writes nothing). - /// Local backend only — S3 has no pin store. + /// A pin travels with the bytes, and object metadata cannot be changed + /// without re-writing the object, so applying a re-pin rewrites the + /// artifact: the file and its sidecar entry on the local backend, the whole + /// object on an object store. pub async fn repin(&self, key: &str, expected: &str, apply: bool) -> Result { validate_storage_key(key)?; - let Some(ref pins) = self.pin_store else { - return Ok(RepinOutcome::NoPinStore); - }; let expected = expected.to_ascii_lowercase(); // Raw read — deliberately bypasses `Storage::get()`'s verification, // which would fail closed on the mismatch we are recovering from. - let data = self.inner.get(key).await?; - let disk = hex::encode(Sha256::digest(&data)); + let (data, old) = self.inner.get(key).await?; + let disk = sha256_hex(&data); if disk != expected { - // The bytes on disk are not the ones the operator vouched for — + // The stored bytes are not the ones the operator vouched for — // genuine corruption/tampering. Re-pin must NOT bless them. return Ok(RepinOutcome::DiskMismatch { disk, expected }); } - let old = pins.get(key); if old.as_deref() == Some(expected.as_str()) { return Ok(RepinOutcome::AlreadyPinned { hash: expected }); } if !apply { return Ok(RepinOutcome::WouldUpdate { old, new: expected }); } - pins.record_hash(key, &expected).map_err(|e| { - StorageError::Io(std::io::Error::other(format!( - "hash-pin record failed: {e}" - ))) - })?; + self.inner.put(key, &data, &expected).await?; Ok(RepinOutcome::Updated { old, new: expected }) } - /// Number of pinned hashes (0 if pin store is disabled). - pub fn pinned_count(&self) -> usize { - self.pin_store.as_ref().map_or(0, |p| p.len()) - } - /// Refresh cached total_size. No-op for local storage, computes for S3. pub async fn refresh_total_size_cache(&self) { self.inner.refresh_total_size().await; @@ -583,49 +521,17 @@ impl Storage { /// Move or copy a file from `src` into storage under `key`. /// - /// When `sha256` is `Some`, the hash is recorded in the pin store without - /// re-reading the file — used by streaming download paths where the hash - /// was already computed incrementally (#580). - /// - /// When `sha256` is `None`, the pin store is not updated (legacy behavior - /// for callers that have already verified integrity separately). + /// When `sha256` is `Some`, the backend pins the object to that digest + /// without re-reading the file — used by streaming paths where the hash was + /// computed incrementally (#580). When it is `None` the object is stored + /// unpinned (legacy behaviour for callers that verified integrity + /// separately). pub async fn put_from_path(&self, key: &str, src: &Path, sha256: Option<&str>) -> Result<()> { validate_storage_key(key)?; - match self.inner.put_from_path(key, src).await { + let sha256 = sha256.map(str::to_ascii_lowercase); + match self.inner.put_from_path(key, src, sha256.as_deref()).await { Ok(()) => { STORAGE_OPERATIONS.with_label_values(&["put", "ok"]).inc(); - if let (Some(hash), Some(ref pins)) = (sha256, &self.pin_store) { - let pins = Arc::clone(pins); - let key_owned = key.to_string(); - let hash = hash.to_string(); - // Await the pin record so it is durable before this returns — - // the streaming write counterpart of the `put()` fix (#604). - // `record_hash` uses the pre-computed digest (no re-hash), so - // the await is near-free. Fail-closed on a panicking task. - match tokio::task::spawn_blocking(move || pins.record_hash(&key_owned, &hash)) - .await - { - Ok(Ok(())) => {} - Ok(Err(e)) => { - STORAGE_OPERATIONS - .with_label_values(&["put", "pin_error"]) - .inc(); - tracing::error!(error = %e, key = %key, "hash-pin record failed"); - return Err(StorageError::Io(std::io::Error::other(format!( - "hash-pin record failed: {e}" - )))); - } - Err(e) => { - STORAGE_OPERATIONS - .with_label_values(&["put", "pin_error"]) - .inc(); - tracing::error!(error = %e, key = %key, "hash-pin record task panicked"); - return Err(StorageError::Io(std::io::Error::other(format!( - "hash-pin record failed: {e}" - )))); - } - } - } Ok(()) } Err(e) => { @@ -640,47 +546,16 @@ impl Storage { /// Server-side copy of `src` to `dst` (see [`StorageBackend::copy`]). /// /// `sha256` is the lowercase hex digest of the copied bytes; when present it - /// is pinned for `dst` without reading the object back. When it is `None`, - /// `dst` inherits the pin of `src` if `src` carries one, and otherwise stays + /// pins `dst` without reading the object back. When it is `None`, `dst` + /// inherits the pin of `src` if `src` carries one, and otherwise stays /// open-world — as [`put_from_path`](Self::put_from_path) does. pub async fn copy(&self, src: &str, dst: &str, sha256: Option<&str>) -> Result<()> { validate_storage_key(src)?; validate_storage_key(dst)?; - match self.inner.copy(src, dst).await { + let sha256 = sha256.map(str::to_ascii_lowercase); + match self.inner.copy(src, dst, sha256.as_deref()).await { Ok(()) => { STORAGE_OPERATIONS.with_label_values(&["copy", "ok"]).inc(); - let hash = sha256 - .map(str::to_ascii_lowercase) - .or_else(|| self.get_pin_hash(src)); - if let (Some(hash), Some(ref pins)) = (hash, &self.pin_store) { - let pins = Arc::clone(pins); - let key_owned = dst.to_string(); - // Fail closed like `put`/`put_from_path`: a copied-but-unpinned - // blob would be served without verification (#582/#604). - match tokio::task::spawn_blocking(move || pins.record_hash(&key_owned, &hash)) - .await - { - Ok(Ok(())) => {} - Ok(Err(e)) => { - STORAGE_OPERATIONS - .with_label_values(&["copy", "pin_error"]) - .inc(); - tracing::error!(error = %e, key = %dst, "hash-pin record failed"); - return Err(StorageError::Io(std::io::Error::other(format!( - "hash-pin record failed: {e}" - )))); - } - Err(e) => { - STORAGE_OPERATIONS - .with_label_values(&["copy", "pin_error"]) - .inc(); - tracing::error!(error = %e, key = %dst, "hash-pin record task panicked"); - return Err(StorageError::Io(std::io::Error::other(format!( - "hash-pin record failed: {e}" - )))); - } - } - } Ok(()) } Err(e) => { @@ -692,16 +567,16 @@ impl Storage { } } - /// Open an artifact for streaming read without loading into memory (#580). + /// Open an artifact for streaming read without loading it into memory (#580). /// - /// Returns `(size_bytes, reader)`. Pin-store integrity is NOT checked here - /// because streaming prevents full-data hashing. Callers that need integrity - /// verification should use `verify_integrity_by_hash` with the content digest - /// (available from the URL for Docker blobs). + /// Returns `(size_bytes, pin, reader)`. The bytes are NOT verified here — + /// streaming prevents a full-body hash before the first frame — so callers + /// hash the stream and check it against `pin` at EOF (see raw's + /// `verify_while_streaming`), or rely on a content-addressed digest. pub async fn get_reader( &self, key: &str, - ) -> Result<(u64, Pin>)> { + ) -> Result<(u64, Option, Pin>)> { validate_storage_key(key)?; match self.inner.get_reader(key).await { Ok(reader) => { @@ -751,20 +626,18 @@ mod tests { use std::time::Duration; use tempfile::TempDir; - /// The GCS wrapper constructs without credentials or network and disables - /// the pin store, matching the S3 wrapper's at-rest posture. + /// The GCS wrapper constructs without credentials or network. #[test] fn test_new_gcs_wrapper() { let storage = Storage::new_gcs("test-bucket", None, Some("http://localhost:4443")); assert_eq!(storage.backend_name(), "gcs"); - assert!(storage.get_pin_hash("any/key").is_none()); } /// Wait until the pin record from `put()` is visible. Since #604 `put()` /// awaits the pin, so this returns on the first poll; kept for robustness. async fn await_pin(storage: &Storage, key: &str) { for _ in 0..200 { - if storage.get_pin_hash(key).is_some() { + if storage.pin(key).await.is_some() { return; } tokio::time::sleep(Duration::from_millis(10)).await; @@ -772,10 +645,16 @@ mod tests { panic!("pin for {key} was never recorded"); } + fn object_storage() -> (Arc, Storage) { + let backend = Arc::new(ObjectStorage::in_memory()); + let storage = Storage::from_backend(backend.clone()); + (backend, storage) + } + /// Regression for #604: `put()` must record the hash-pin BEFORE it returns, /// so there is no window where a completed put leaves the artifact readable /// but unpinned (which a later `get()` would serve unverified). Exercises - /// the real call path `Storage::put()` → `get_pin_hash()` — the pin is + /// the real call path `Storage::put()` → `pin()` — the pin is /// observable synchronously, with no polling. #[tokio::test] async fn put_records_pin_before_returning() { @@ -785,7 +664,7 @@ mod tests { storage.put("raw/x/app.bin", b"payload").await.unwrap(); assert!( - storage.get_pin_hash("raw/x/app.bin").is_some(), + storage.pin("raw/x/app.bin").await.is_some(), "put() must record the hash-pin before returning (#604)" ); // And the recorded pin must match the bytes (a subsequent get verifies). @@ -813,7 +692,7 @@ mod tests { b"layer" ); assert_eq!( - storage.get_pin_hash("docker/dst/blobs/x").as_deref(), + storage.pin("docker/dst/blobs/x").await.as_deref(), Some(hash.as_str()) ); assert!(matches!( @@ -894,9 +773,10 @@ mod tests { ); } - /// Regression for #604: the streaming write path `put_from_path()` must also - /// record its pin BEFORE returning (same fire-and-forget gap as `put()`, - /// on the path that handles Docker blobs). Exercises the real call path. + /// Regression for #604: the streaming write path `put_from_path()` — the one + /// that handles Docker blobs — must record its pin BEFORE returning, so the + /// pin is observable the moment the call completes. Exercises the real call + /// path. #[tokio::test] async fn put_from_path_records_pin_before_returning() { use sha2::{Digest, Sha256}; @@ -911,7 +791,7 @@ mod tests { storage.put_from_path(key, &src, Some(&sha)).await.unwrap(); assert_eq!( - storage.get_pin_hash(key).as_deref(), + storage.pin(key).await.as_deref(), Some(sha.as_str()), "put_from_path must record the hash-pin before returning (#604)" ); @@ -1087,7 +967,7 @@ mod tests { storage.get(key).await, Err(StorageError::IntegrityViolation) )); - assert_eq!(storage.get_pin_hash(key).as_deref(), Some(genuine.as_str())); + assert_eq!(storage.pin(key).await.as_deref(), Some(genuine.as_str())); } /// Re-pinning a key whose pin already equals `expected` is a no-op. @@ -1125,7 +1005,7 @@ mod tests { std::fs::write(&src, b"orphan-bytes").unwrap(); storage.put_from_path(key, &src, None).await.unwrap(); assert_eq!( - storage.get_pin_hash(key), + storage.pin(key).await, None, "precondition: the orphaned body must be stored without a pin" ); @@ -1140,10 +1020,161 @@ mod tests { new: expected.clone(), } ); + assert_eq!(storage.pin(key).await.as_deref(), Some(expected.as_str())); + assert_eq!(&storage.get(key).await.unwrap()[..], b"orphan-bytes"); + } + + // --- Object-store backend: the pin is user-defined object metadata --- + + /// A pinned object round-trips its pin through the store's metadata, so the + /// gate verifies it exactly as it does on the local backend. + #[tokio::test] + async fn object_put_pins_and_verifies() { + use nora_registry::verified::GateOutcome; + let (_, storage) = object_storage(); + let key = "raw/obj/app.bin"; + + storage.put(key, b"object-bytes").await.unwrap(); + + assert_eq!(&storage.get(key).await.unwrap()[..], b"object-bytes"); assert_eq!( - storage.get_pin_hash(key).as_deref(), - Some(expected.as_str()) + storage.pin(key).await.as_deref(), + Some(sha_hex(b"object-bytes").as_str()) + ); + assert!(matches!( + storage.get_verified(key).await.unwrap(), + GateOutcome::Verified(_) + )); + } + + /// The streaming write path pins when the caller computed a digest, and + /// leaves the object open-world when it did not. + #[tokio::test] + async fn object_put_from_path_pins_only_with_a_digest() { + use nora_registry::verified::GateOutcome; + let (_, storage) = object_storage(); + let dir = TempDir::new().unwrap(); + + let src = dir.path().join("pinned.bin"); + std::fs::write(&src, b"streamed").unwrap(); + let sha = sha_hex(b"streamed"); + storage + .put_from_path("raw/obj/pinned.bin", &src, Some(&sha)) + .await + .unwrap(); + assert_eq!( + storage.pin("raw/obj/pinned.bin").await.as_deref(), + Some(sha.as_str()) + ); + + let src = dir.path().join("unpinned.bin"); + std::fs::write(&src, b"streamed").unwrap(); + storage + .put_from_path("raw/obj/unpinned.bin", &src, None) + .await + .unwrap(); + assert_eq!(storage.pin("raw/obj/unpinned.bin").await, None); + assert!(matches!( + storage.get_verified("raw/obj/unpinned.bin").await.unwrap(), + GateOutcome::Unpinned(_) + )); + } + + /// A store-side copy carries the source's user metadata, so the destination + /// inherits the pin without the wrapper re-writing it. + #[tokio::test] + async fn object_copy_inherits_the_source_pin() { + let (_, storage) = object_storage(); + storage.put("docker/src/blobs/x", b"layer").await.unwrap(); + + storage + .copy("docker/src/blobs/x", "docker/dst/blobs/x", None) + .await + .unwrap(); + + assert_eq!( + storage.pin("docker/dst/blobs/x").await.as_deref(), + Some(sha_hex(b"layer").as_str()) + ); + assert_eq!( + &storage.get("docker/dst/blobs/x").await.unwrap()[..], + b"layer" + ); + } + + /// Regression for #582 on an object store: bytes replaced under an unchanged + /// pin must not be served. Writes through the backend directly, bypassing + /// the wrapper — exactly the out-of-band tamper the pin exists to catch. + #[tokio::test] + async fn object_get_fails_closed_on_integrity_mismatch() { + let (backend, storage) = object_storage(); + let key = "raw/obj/tampered.bin"; + + storage.put(key, b"genuine-bytes").await.unwrap(); + backend + .put(key, b"TAMPERED", &sha_hex(b"genuine-bytes")) + .await + .unwrap(); + + assert!(matches!( + storage.get(key).await, + Err(StorageError::IntegrityViolation) + )); + } + + /// Objects written before pins existed carry no metadata: they stay + /// readable and open-world, with no migration. + #[tokio::test] + async fn object_without_metadata_is_open_world() { + use object_store::{ObjectStoreExt, PutPayload}; + let (backend, storage) = object_storage(); + let key = "raw/obj/legacy.bin"; + + backend + .store() + .put( + &object_store::path::Path::from(key), + PutPayload::from_static(b"pre-existing"), + ) + .await + .unwrap(); + + assert_eq!(&storage.get(key).await.unwrap()[..], b"pre-existing"); + assert_eq!(storage.pin(key).await, None); + } + + /// #601 on an object store: re-pin still refuses bytes the operator did not + /// vouch for, and applying it rewrites the object with the new pin. + #[tokio::test] + async fn object_repin_refuses_mismatch_and_pins_on_match() { + let (_, storage) = object_storage(); + let dir = TempDir::new().unwrap(); + let key = "raw/obj/repin.bin"; + + // An unpinned object (streamed in without a digest) is the recoverable + // orphan case. + let src = dir.path().join("orphan.bin"); + std::fs::write(&src, b"orphan-bytes").unwrap(); + storage.put_from_path(key, &src, None).await.unwrap(); + + let wrong = sha_hex(b"something-else"); + assert_eq!( + storage.repin(key, &wrong, true).await.unwrap(), + RepinOutcome::DiskMismatch { + disk: sha_hex(b"orphan-bytes"), + expected: wrong, + } + ); + + let expected = sha_hex(b"orphan-bytes"); + assert_eq!( + storage.repin(key, &expected, true).await.unwrap(), + RepinOutcome::Updated { + old: None, + new: expected.clone(), + } ); + assert_eq!(storage.pin(key).await.as_deref(), Some(expected.as_str())); assert_eq!(&storage.get(key).await.unwrap()[..], b"orphan-bytes"); } } diff --git a/nora-registry/src/storage/object.rs b/nora-registry/src/storage/object.rs index e624dd2c..27f95384 100644 --- a/nora-registry/src/storage/object.rs +++ b/nora-registry/src/storage/object.rs @@ -7,7 +7,10 @@ use futures::TryStreamExt; use object_store::aws::AmazonS3Builder; use object_store::gcp::GoogleCloudStorageBuilder; use object_store::path::Path; -use object_store::{ObjectStore, ObjectStoreExt, PutPayload, WriteMultipart}; +use object_store::{ + Attribute, AttributeValue, Attributes, GetOptions, ObjectStore, ObjectStoreExt, + PutMultipartOptions, PutOptions, PutPayload, WriteMultipart, +}; use std::pin::Pin; use tokio::io::{AsyncRead, AsyncReadExt}; @@ -162,6 +165,27 @@ impl ObjectStorage { last_refresh_unix: std::sync::atomic::AtomicU64::new(0), } } + + /// In-process object store for tests: same code path as S3/GCS, including + /// user-metadata round-trips and store-side copy. + #[cfg(test)] + pub(crate) fn in_memory() -> Self { + Self { + store: Box::new(object_store::memory::InMemory::new()), + name: "s3", + cached_total_size: std::sync::atomic::AtomicU64::new(0), + size_cache_initialized: std::sync::atomic::AtomicBool::new(false), + cached_reachable: std::sync::atomic::AtomicBool::new(true), + last_refresh_unix: std::sync::atomic::AtomicU64::new(0), + } + } + + /// Test-only handle on the raw store, for writing objects NORA itself would + /// never write (e.g. one with no pin metadata). + #[cfg(test)] + pub(crate) fn store(&self) -> &dyn ObjectStore { + self.store.as_ref() + } } /// Encode `@` in object keys to `%40` for SeaweedFS compatibility (shared by @@ -203,6 +227,30 @@ fn decode_object_key(key: &str) -> String { key.replace("%2540", "@").replace("%40", "@") } +/// User-defined object metadata carrying the SHA-256 integrity pin — written +/// atomically with the object and returned with every GET/HEAD, so the pin +/// travels with the bytes (`x-amz-meta-sha256` / `x-goog-meta-sha256`). +const PIN_METADATA_KEY: &str = "sha256"; + +fn pin_attribute() -> Attribute { + Attribute::Metadata(PIN_METADATA_KEY.into()) +} + +fn pin_attributes(sha256: &str) -> Attributes { + [( + pin_attribute(), + AttributeValue::from(sha256.to_ascii_lowercase()), + )] + .into_iter() + .collect() +} + +fn read_pin(attributes: &Attributes) -> Option { + attributes + .get(&pin_attribute()) + .map(|v| v.as_ref().to_string()) +} + /// Map object_store errors to StorageError. fn map_err(e: object_store::Error) -> StorageError { match e { @@ -213,32 +261,55 @@ fn map_err(e: object_store::Error) -> StorageError { #[async_trait] impl StorageBackend for ObjectStorage { - async fn put(&self, key: &str, data: &[u8]) -> Result<()> { + async fn put(&self, key: &str, data: &[u8], sha256: &str) -> Result<()> { let encoded = encode_object_key(key); let path = Path::from(encoded); let payload = PutPayload::from(data.to_vec()); - self.store.put(&path, payload).await.map_err(map_err)?; + let opts = PutOptions { + attributes: pin_attributes(sha256), + ..Default::default() + }; + self.store + .put_opts(&path, payload, opts) + .await + .map_err(map_err)?; Ok(()) } - async fn get(&self, key: &str) -> Result { + async fn get(&self, key: &str) -> Result<(Bytes, Option)> { let encoded = encode_object_key(key); let path = Path::from(encoded); - match self.store.get(&path).await { - Ok(result) => { - let bytes = result.bytes().await.map_err(map_err)?; - Ok(bytes) - } + let result = match self.store.get(&path).await { + Ok(result) => result, Err(object_store::Error::NotFound { .. }) if key.contains('@') => { // Fallback: try legacy _at_ encoding for pre-#534 data. // Only needed when key contains @, since otherwise both schemes produce the same output. let legacy_path = Path::from(encode_object_key_legacy(key)); - let result = self.store.get(&legacy_path).await.map_err(map_err)?; - let bytes = result.bytes().await.map_err(map_err)?; - Ok(bytes) + self.store.get(&legacy_path).await.map_err(map_err)? } - Err(e) => Err(map_err(e)), - } + Err(e) => return Err(map_err(e)), + }; + let pin = read_pin(&result.attributes); + let bytes = result.bytes().await.map_err(map_err)?; + Ok((bytes, pin)) + } + + async fn pin(&self, key: &str) -> Option { + let head = || GetOptions { + head: true, + ..Default::default() + }; + let path = Path::from(encode_object_key(key)); + let result = match self.store.get_opts(&path, head()).await { + Ok(r) => r, + Err(_) if key.contains('@') => { + // Fallback: try legacy _at_ encoding for pre-#534 data. + let legacy_path = Path::from(encode_object_key_legacy(key)); + self.store.get_opts(&legacy_path, head()).await.ok()? + } + Err(_) => return None, + }; + read_pin(&result.attributes) } async fn delete(&self, key: &str) -> Result<()> { @@ -369,7 +440,12 @@ impl StorageBackend for ObjectStorage { } } - async fn put_from_path(&self, key: &str, src: &std::path::Path) -> Result<()> { + async fn put_from_path( + &self, + key: &str, + src: &std::path::Path, + sha256: Option<&str>, + ) -> Result<()> { let encoded = encode_object_key(key); let s3_path = Path::from(encoded); @@ -384,7 +460,17 @@ impl StorageBackend for ObjectStorage { // No partial objects are visible to readers (upload never completed). // finish() calls abort() on its own errors; cancellation (future // dropped) relies on lifecycle policy only. - let upload = self.store.put_multipart(&s3_path).await.map_err(map_err)?; + // The pin rides on the multipart init, so it lands with the object or + // not at all — no window where the bytes exist unpinned. + let opts = PutMultipartOptions { + attributes: sha256.map(pin_attributes).unwrap_or_default(), + ..Default::default() + }; + let upload = self + .store + .put_multipart_opts(&s3_path, opts) + .await + .map_err(map_err)?; let mut writer = WriteMultipart::new(upload); let mut buf = vec![0u8; 8 * 1024 * 1024]; // 8 MiB read buffer @@ -401,15 +487,19 @@ impl StorageBackend for ObjectStorage { Ok(()) } - async fn copy(&self, src: &str, dst: &str) -> Result<()> { + async fn copy(&self, src: &str, dst: &str, _sha256: Option<&str>) -> Result<()> { // Store-side copy (S3 CopyObject / GCS rewrite) — no bytes cross the - // network through this process. + // network through this process. A store-side copy carries the source's + // user metadata, so `dst` inherits the pin without re-writing it. let from_path = Path::from(encode_object_key(src)); let to_path = Path::from(encode_object_key(dst)); self.store.copy(&from_path, &to_path).await.map_err(map_err) } - async fn get_reader(&self, key: &str) -> Result<(u64, Pin>)> { + async fn get_reader( + &self, + key: &str, + ) -> Result<(u64, Option, Pin>)> { let encoded = encode_object_key(key); let path = Path::from(encoded); let result = match self.store.get(&path).await { @@ -421,9 +511,10 @@ impl StorageBackend for ObjectStorage { Err(e) => return Err(map_err(e)), }; let size = result.meta.size; + let pin = read_pin(&result.attributes); let stream = result.into_stream().map_err(std::io::Error::other); let reader = tokio_util::io::StreamReader::new(stream); - Ok((size as u64, Box::pin(reader))) + Ok((size as u64, pin, Box::pin(reader))) } async fn get_range( @@ -432,7 +523,7 @@ impl StorageBackend for ObjectStorage { start: u64, end: u64, ) -> Result<(u64, Pin>)> { - let make_opts = || object_store::GetOptions { + let make_opts = || GetOptions { range: Some(object_store::GetRange::Bounded(start..(end + 1))), ..Default::default() }; @@ -458,6 +549,7 @@ impl StorageBackend for ObjectStorage { #[cfg(test)] mod tests { use super::*; + use sha2::Digest; #[test] fn test_backend_name() { @@ -531,21 +623,16 @@ mod tests { /// packument (`npm install` -> ENOVERSIONS). #[tokio::test] async fn scoped_key_lists_and_gets_through_path_encoding() { - let storage = ObjectStorage { - store: Box::new(object_store::memory::InMemory::new()), - name: "s3", - cached_total_size: std::sync::atomic::AtomicU64::new(0), - size_cache_initialized: std::sync::atomic::AtomicBool::new(false), - cached_reachable: std::sync::atomic::AtomicBool::new(true), - last_refresh_unix: std::sync::atomic::AtomicU64::new(0), - }; + let storage = ObjectStorage::in_memory(); + let body = br#"{"version":"1.0.0"}"#; + let hash = hex::encode(sha2::Sha256::digest(body)); let key = "npm/@scope/pkg/versions/1.0.0.json"; - storage.put(key, br#"{"version":"1.0.0"}"#).await.unwrap(); + storage.put(key, body, &hash).await.unwrap(); // Control: a non-scoped key (no `@`, no `%`) is unaffected. let plain = "npm/plainpkg/versions/1.0.0.json"; - storage.put(plain, br#"{"version":"1.0.0"}"#).await.unwrap(); + storage.put(plain, body, &hash).await.unwrap(); // list() must return the ORIGINAL logical key (with `@`), not the `%2540` form. let listed = storage.list("npm/@scope/pkg/versions/").await.unwrap(); @@ -557,11 +644,12 @@ mod tests { // The listed key must be directly get-able — the exact scan-regenerate step that // silently dropped scoped versions before the fix. - let got = storage + let (got, pin) = storage .get(&listed[0]) .await .expect("get on the listed key must succeed"); - assert_eq!(&got[..], br#"{"version":"1.0.0"}"#); + assert_eq!(&got[..], body); + assert_eq!(pin.as_deref(), Some(hash.as_str())); let listed_plain = storage.list("npm/plainpkg/versions/").await.unwrap(); assert_eq!(listed_plain, vec![plain.to_string()]); diff --git a/nora-registry/src/test_helpers.rs b/nora-registry/src/test_helpers.rs index c0c42c0c..f92145a6 100644 --- a/nora-registry/src/test_helpers.rs +++ b/nora-registry/src/test_helpers.rs @@ -73,6 +73,12 @@ pub fn create_test_context_with_raw_disabled() -> TestContext { build_context(false, &[], false, |cfg| cfg.raw.enabled = false) } +/// Build a test context over a caller-supplied backend, so handler tests can +/// run against an object store instead of the local filesystem. +pub fn create_test_context_with_storage(storage: Storage) -> TestContext { + build_context_with(false, &[], false, |_| {}, Some(storage)) +} + /// Build a test context with custom config tweaks. pub fn create_test_context_with_config(customize: impl FnOnce(&mut Config)) -> TestContext { build_context(false, &[], false, customize) @@ -91,6 +97,16 @@ fn build_context( users: &[(&str, &str)], anonymous_read: bool, customize: impl FnOnce(&mut Config), +) -> TestContext { + build_context_with(auth_enabled, users, anonymous_read, customize, None) +} + +fn build_context_with( + auth_enabled: bool, + users: &[(&str, &str)], + anonymous_read: bool, + customize: impl FnOnce(&mut Config), + storage: Option, ) -> TestContext { let tempdir = TempDir::new().expect("failed to create tempdir"); let storage_path = tempdir.path().to_str().unwrap().to_string(); @@ -218,7 +234,7 @@ fn build_context( // Apply any custom config tweaks customize(&mut config); - let storage = Storage::new_local(&storage_path); + let storage = storage.unwrap_or_else(|| Storage::new_local(&storage_path)); let auth = if auth_enabled && !users.is_empty() { let htpasswd_path = tempdir.path().join("users.htpasswd"); diff --git a/nora-registry/src/verified.rs b/nora-registry/src/verified.rs index 186e5da0..d232b949 100644 --- a/nora-registry/src/verified.rs +++ b/nora-registry/src/verified.rs @@ -7,7 +7,7 @@ //! [`Storage::get`](crate::storage::Storage::get) gate hash-pin-verifies bytes //! before returning them (`#582`/`#604`), the streaming docker path //! tamper-detects at EOF, and `put` couples the body write with a hash-pin -//! record on the local backend. This module pushes those guarantees one step +//! record on every backend. This module pushes those guarantees one step //! **left** — into the type system — and, just as importantly, makes the //! *known holes* impossible to hide. //! @@ -22,7 +22,7 @@ //! no privileged constructor, no cross-crate seal to trust. What `expected` //! *means* is the caller's to state: for a hash-pinned cache read it is the //! digest NORA recorded at store time (tamper-evidence against on-disk -//! corruption — `src/storage/mod.rs:260`), and for a content-addressed +//! corruption — `src/storage/mod.rs`), and for a content-addressed //! artifact (a docker blob, whose key *is* its digest) it is the canonical //! upstream digest. //! @@ -41,8 +41,8 @@ //! A serve sink that demands `Blob` (see [`verified_body`]) *cannot* //! be handed raw or merely-tamper-evident bytes — that is a compile error, not //! a runtime check that a future refactor might skip. And the open-world hole -//! (an unpinned key, or the S3 backend which has no pin store at all) is forced -//! into the open by [`GateOutcome`]: a caller of +//! (an object stored with no pin) is forced into the open by [`GateOutcome`]: +//! a caller of //! [`Storage::get_verified`](crate::storage::Storage::get_verified) must //! `match` and decide what to do with [`GateOutcome::Unpinned`] — it can never //! be mistaken for a cryptographically verified read. @@ -215,15 +215,14 @@ pub enum IntegrityError { /// A caller cannot get bytes out without `match`ing, so the open-world hole is /// impossible to ignore: either the bytes matched a recorded pin /// ([`Verified`](GateOutcome::Verified)), or no pin existed for the key and they -/// were served without a cryptographic check ([`Unpinned`](GateOutcome::Unpinned)) -/// — the latter covers both genuinely-unpinned keys and the S3 backend, which -/// has no pin store at all. +/// were served without a cryptographic check +/// ([`Unpinned`](GateOutcome::Unpinned)). #[derive(Debug)] pub enum GateOutcome { /// A pin existed and the bytes matched it: cryptographically [`Verified`]. Verified(Blob), /// No pin existed for this key — served open-world (no cryptographic - /// guarantee). The honest name for the gate's no-pin / S3 branch. + /// guarantee). The honest name for the gate's no-pin branch. Unpinned(Blob), } @@ -304,14 +303,12 @@ pub trait Durability: sealed::Sealed { const TIER: &'static str; } -/// Body **and** hash-pin both landed durably — the local-backend store path. -/// Uninhabited. +/// Body **and** hash-pin both landed durably. Uninhabited. #[derive(Debug)] pub enum Pinned {} -/// Stored on a backend with no pin store (S3): integrity cannot be recorded, so -/// the artifact is served open-world. Names the documented S3 limitation in the -/// type system. Uninhabited. +/// Stored without a digest, so no pin was recorded and the artifact is served +/// open-world. Names that hole in the type system. Uninhabited. #[derive(Debug)] pub enum Unpinnable {} @@ -347,7 +344,7 @@ impl StoreReceipt

{ impl StoreReceipt { /// Mint a [`Pinned`] receipt — call only after both the body and the - /// hash-pin have durably landed (the local-backend `put` path). + /// hash-pin have durably landed. #[must_use] pub fn pinned(key: impl Into) -> Self { Self { @@ -358,7 +355,7 @@ impl StoreReceipt { } impl StoreReceipt { - /// Mint an [`Unpinnable`] receipt — the backend has no pin store (S3). + /// Mint an [`Unpinnable`] receipt — the store recorded no pin. #[must_use] pub fn unpinnable(key: impl Into) -> Self { Self { @@ -370,8 +367,8 @@ impl StoreReceipt { /// An operation that requires a durable integrity pin (e.g. an immutable /// publish that must be re-verifiable). Accepts **only** a -/// [`StoreReceipt`] — passing an [`StoreReceipt`] (an S3 -/// store) is a compile error, so the S3 hole cannot be silently relied upon. +/// [`StoreReceipt`] — passing an [`StoreReceipt`] is a +/// compile error, so the open-world hole cannot be silently relied upon. /// /// # A pinned store type-checks /// @@ -381,12 +378,12 @@ impl StoreReceipt { /// assert_eq!(require_pinned(local), "raw/x"); /// ``` /// -/// # An unpinnable (S3) store does NOT compile here +/// # An unpinnable store does NOT compile here /// /// ```compile_fail /// use nora_registry::verified::{require_pinned, StoreReceipt}; -/// let s3 = StoreReceipt::unpinnable("raw/x"); -/// let _ = require_pinned(s3); // expected StoreReceipt, found +/// let unpinned = StoreReceipt::unpinnable("raw/x"); +/// let _ = require_pinned(unpinned); // expected StoreReceipt, found /// ``` pub fn require_pinned(receipt: StoreReceipt) -> String { receipt.key diff --git a/tests/s3-backends/test.sh b/tests/s3-backends/test.sh index 7de870a9..70ff3992 100755 --- a/tests/s3-backends/test.sh +++ b/tests/s3-backends/test.sh @@ -24,6 +24,15 @@ skip() { SKIPPED=$((SKIPPED + 1)) } +# SHA-256 of stdin, hex digest only. +sha256_hex() { + if command -v sha256sum >/dev/null 2>&1; then + sha256sum | awk '{print $1}' + else + shasum -a 256 | awk '{print $1}' + fi +} + # Wait for a NORA instance to be healthy (up to 30s) wait_healthy() { local url="$1" @@ -70,6 +79,26 @@ test_backend() { fail "${name}: raw download (simple key) mismatch" fi + # 2b. ETag + conditional GET. The hash pin is stored as `sha256` object + # metadata, so the ETag is the object's digest on every S3 implementation. + local expected_etag etag + expected_etag="\"$(printf '%s\n' "$payload" | sha256_hex)\"" + etag=$(curl -sf --head "${base}/raw/s3test/simple.txt" 2>/dev/null \ + | tr -d '\r' | awk 'tolower($1) == "etag:" {print $2}') + if [ "$etag" = "$expected_etag" ]; then + pass "${name}: HEAD returns sha256 ETag" + else + fail "${name}: HEAD ETag ${etag:-} != ${expected_etag}" + fi + + http_code=$(curl -s -o /dev/null -w "%{http_code}" \ + -H "If-None-Match: ${etag}" "${base}/raw/s3test/simple.txt") + if [ "$http_code" = "304" ]; then + pass "${name}: If-None-Match returns 304" + else + fail "${name}: If-None-Match returned ${http_code}, expected 304" + fi + # 3. Raw upload/download — key with @ (scoped package path) local at_payload="at-test-data-$(date +%s)" http_code=$(echo "$at_payload" | curl -s -o /dev/null -w "%{http_code}" \