diff --git a/crates/protocol/src/fuzz.rs b/crates/protocol/src/fuzz.rs index a36259e..c6ca006 100644 --- a/crates/protocol/src/fuzz.rs +++ b/crates/protocol/src/fuzz.rs @@ -107,4 +107,29 @@ proptest! { let _ = decode_hello(&bytes); let _ = decode_hello_ack(&bytes); } + + #[test] + fn valid_exports_reject_every_generated_truncation_and_payload_bit_flip( + payload in prop::collection::vec(any::(), 1..1024), + position in any::(), + bit in 0u8..8, + ) { + let block = CommittedExport { + identity: synthetic_identity(1, 1), + source_generation: 1, + checksum: content_checksum(&payload), + payload: payload.clone(), + }; + let frame = encode_export(&block).unwrap(); + prop_assert_eq!(decode_export(&frame).unwrap(), block); + let cut = position % frame.len(); + prop_assert!(decode_export(&frame[..cut]).is_err()); + let mut changed = frame.clone(); + let offset = changed.len() - payload.len() + position % payload.len(); + changed[offset] ^= 1 << bit; + prop_assert!(decode_export(&changed).is_err()); + let mut trailing = frame; + trailing.push(0); + prop_assert!(decode_export(&trailing).is_err()); + } } diff --git a/crates/protocol/src/harden.rs b/crates/protocol/src/harden.rs index fc3c0f2..8807b83 100644 --- a/crates/protocol/src/harden.rs +++ b/crates/protocol/src/harden.rs @@ -63,6 +63,7 @@ impl SecurityLog { } } +#[derive(Clone)] struct AclEntry { tenant: TenantId, trust: TrustDomain, @@ -70,11 +71,21 @@ struct AclEntry { write: bool, } +#[derive(Clone)] pub struct Acl { entries: Vec, max: usize, } +impl std::fmt::Debug for Acl { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Acl") + .field("entries", &self.entries.len()) + .field("capacity", &self.max) + .finish() + } +} + impl Acl { pub fn new(max: usize) -> Result { if max == 0 || max > 1024 { @@ -435,6 +446,25 @@ mod tests { use super::*; use gpucache_core_types::TenantId; + #[test] + fn cloned_acl_keeps_authority_without_exposing_fingerprints_in_debug() { + let ctx = AuthContext::new( + TenantId::new(b"tenant".to_vec()).unwrap(), + TrustDomain::new(b"trust".to_vec()).unwrap(), + [173; 32], + ) + .unwrap(); + let mut acl = Acl::new(1).unwrap(); + acl.grant(ctx.clone(), false).unwrap(); + let copy = acl.clone(); + assert_eq!(copy.check(&ctx, false), Ok(())); + assert_eq!(copy.check(&ctx, true), Err(CacheError::Unauthorized)); + let debug = format!("{copy:?}"); + assert!(debug.contains("entries: 1")); + assert!(!debug.contains("173")); + assert!(!debug.contains("tenant")); + } + fn ctx(tenant: &[u8], fingerprint: u8) -> AuthContext { AuthContext::new( TenantId::new(tenant.to_vec()).unwrap(), diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index dcc320f..f849084 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -35,6 +35,6 @@ pub use harden::{ SecretRing, SecurityLog, }; pub use negotiate::{negotiate, HELLO_OP, WIRE_VERSION}; -pub use reply::{decode_reply, encode_reply, ReplyBody, REPLY_MAGIC}; +pub use reply::{decode_reply, encode_reply, get_reply_payload_budget, ReplyBody, REPLY_MAGIC}; pub use gpucache_core_types::Attestation; diff --git a/crates/protocol/src/reply.rs b/crates/protocol/src/reply.rs index cbb4ba9..c9d7e09 100644 --- a/crates/protocol/src/reply.rs +++ b/crates/protocol/src/reply.rs @@ -15,6 +15,34 @@ use crate::negotiate::WIRE_VERSION; pub const REPLY_MAGIC: &[u8; 4] = b"GCKR"; const OP_FAILED: u16 = 0; +/// Aggregate Get payload allowance after reserving GCKR batch/item metadata. +/// +/// Reserve the exact successful-item metadata for every item. Every encoded +/// error is smaller, so failed items cannot consume another item's reservation. +/// This conservative budget must be checked before creating any reader pins. +/// The limit covers the reply body, excluding an outer transport envelope. +pub fn get_reply_payload_budget( + max_reply_bytes: usize, + item_count: usize, +) -> Result { + // Magic/version/op/epoch/count, then request-id/status/GetHit/bytes length. + const BATCH_BYTES: usize = 4 + 2 + 2 + 8 + 4; + const ITEM_BYTES: usize = 16 + 1 + 8 + 8 + 32 + 8 + 8 + 4; + u32::try_from(item_count).map_err(|_| CacheError::QuotaExceeded { + kind: QuotaKind::BatchItems, + })?; + let available = item_count + .checked_mul(ITEM_BYTES) + .and_then(|bytes| bytes.checked_add(BATCH_BYTES)) + .and_then(|metadata| max_reply_bytes.checked_sub(metadata)) + .ok_or(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + })?; + // ByteWriter::bytes also has an absolute field cap. Keep even an unlimited + // caller from pinning data whose subsequent reply encoding would fail. + Ok(available.min(HARD_BYTES_CAP) as u64) +} + /// Encoded store reply. Request frames stay `GCK1`; this is the response side. #[derive(Clone, Debug, PartialEq, Eq)] pub enum ReplyBody { @@ -549,6 +577,69 @@ mod tests { RequestId::from_counter(n).unwrap() } + #[test] + fn get_budget_matches_the_wire_and_covers_error_metadata() { + for count in [0, 1, 2, 64, 1024] { + let batch = BatchReply { + store_epoch: u64::MAX, + items: (0..count) + .map(|n| ItemResult { + request_id: id(n as u128 + 1), + body: Ok(GetHit { + generation: u64::MAX, + length: 3, + checksum: content_checksum(b"abc"), + region_id: u64::MAX, + lease_deadline_ms: u64::MAX, + payload: b"abc".to_vec(), + }), + }) + .collect(), + }; + let frame_bytes = encode_reply(&ReplyBody::Get(batch.clone())).unwrap().len(); + assert_eq!( + get_reply_payload_budget(frame_bytes, count), + Ok(count as u64 * 3) + ); + // Include the widest error encoding (two u64 fields) and the + // quota error emitted when the successful payload would not fit. + for error in [ + CacheError::EpochMismatch { + store: u64::MAX, + request: u64::MAX, + }, + CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + }, + CacheError::Unauthorized, + ] { + let mut denied = batch.clone(); + for item in &mut denied.items { + item.body = Err(error.clone()); + } + assert!(encode_reply(&ReplyBody::Get(denied)).unwrap().len() <= frame_bytes); + } + } + assert!(get_reply_payload_budget(104, 1).is_err()); + assert_eq!(get_reply_payload_budget(105, 1), Ok(0)); + assert!(get_reply_payload_budget(usize::MAX, usize::MAX).is_err()); + assert_eq!( + get_reply_payload_budget(usize::MAX, 1), + Ok(HARD_BYTES_CAP as u64) + ); + let failed = encode_reply(&ReplyBody::Failed { + op: OP_GET, + store_epoch: u64::MAX, + error: CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + }, + }) + .unwrap(); + // Smaller than even the server's minimum 64-byte reply-body ceiling. + assert_eq!(failed.len(), 21); + assert!(decode_reply(&failed).is_ok()); + } + #[test] fn reply_roundtrip_covers_hit_prefix_and_failure() { let sum = content_checksum(b"abc"); diff --git a/crates/storage/src/file.rs b/crates/storage/src/file.rs index dc38253..55a667b 100644 --- a/crates/storage/src/file.rs +++ b/crates/storage/src/file.rs @@ -103,6 +103,12 @@ impl FileStore { self.inner.epoch() } + /// Configured bearer-fingerprint policies for compiling a transport ACL. + /// The returned credentials must never be logged or sent to a peer. + pub fn tenant_policies(&self) -> Vec { + self.inner.tenant_policies() + } + pub fn limits(&self) -> ResourceLimits { self.inner.limits() } @@ -223,12 +229,30 @@ impl FileStore { items: Vec, ) -> Result, CacheError> { let reply = self.inner.get(auth, now_ms, items)?; + Ok(self.touch_get_reply(reply)) + } + + /// Bound the encoded GCKR body before copying payloads or creating pins. + pub fn get_with_reply_limit( + &mut self, + auth: &AuthContext, + now_ms: u64, + items: Vec, + max_reply_bytes: usize, + ) -> Result, CacheError> { + let reply = self + .inner + .get_with_reply_limit(auth, now_ms, items, max_reply_bytes)?; + Ok(self.touch_get_reply(reply)) + } + + fn touch_get_reply(&mut self, reply: BatchReply) -> BatchReply { for item in &reply.items { if let Ok(hit) = &item.body { self.touch(hit.generation); } } - Ok(reply) + reply } pub fn release_cancel( diff --git a/crates/storage/src/memory.rs b/crates/storage/src/memory.rs index 8c98d67..f67ada4 100644 --- a/crates/storage/src/memory.rs +++ b/crates/storage/src/memory.rs @@ -9,9 +9,10 @@ use gpucache_core_types::{ RequestId, ResourceLimits, }; use gpucache_protocol::{ - ensure_deadline, fingerprints_equal, AuthContext, BatchReply, CommitItem, CommitOutcome, - GetHit, GetItem, InvalidateItem, InvalidateOutcome, ItemResult, LookupItem, LookupOutcome, - MissReason, ReleaseCancelItem, ReleaseOutcome, ReserveOutcome, ReservePutItem, TenantPolicy, + ensure_deadline, fingerprints_equal, get_reply_payload_budget, AuthContext, BatchReply, + CommitItem, CommitOutcome, GetHit, GetItem, InvalidateItem, InvalidateOutcome, ItemResult, + LookupItem, LookupOutcome, MissReason, ReleaseCancelItem, ReleaseOutcome, ReserveOutcome, + ReservePutItem, TenantPolicy, }; #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -143,6 +144,8 @@ pub struct StoreStats { pub quarantined: u32, pub committed: u32, pub idempotency_entries: u32, + /// Active and stuck pins both retain admission credit until explicit drain. + pub reader_pins: u32, } impl MemoryStore { @@ -183,6 +186,10 @@ impl MemoryStore { self.epoch } + pub(crate) fn tenant_policies(&self) -> Vec { + self.policies.values().cloned().collect() + } + pub fn stats(&self) -> StoreStats { let mut quarantined = 0; let mut committed = 0; @@ -200,6 +207,7 @@ impl MemoryStore { quarantined, committed, idempotency_entries: u32::try_from(self.idem.len()).unwrap_or(u32::MAX), + reader_pins: self.reader_pin_count(), } } @@ -250,8 +258,8 @@ impl MemoryStore { now_ms: u64, request: ReclaimRequest, ) -> Result { - self.poll(now_ms); self.authorize(auth, None)?; + self.poll(now_ms); let ReclaimRequest { request_id, epoch, @@ -365,16 +373,45 @@ impl MemoryStore { auth: &AuthContext, now_ms: u64, items: Vec, + ) -> Result, CacheError> { + self.get_inner(auth, now_ms, items, None) + } + + /// Get with a GCKR reply-body ceiling, including all batch/item metadata. + /// Over-budget items fail before payload copying or reader pin creation. + pub fn get_with_reply_limit( + &mut self, + auth: &AuthContext, + now_ms: u64, + items: Vec, + max_reply_bytes: usize, + ) -> Result, CacheError> { + self.get_inner(auth, now_ms, items, Some(max_reply_bytes)) + } + + fn get_inner( + &mut self, + auth: &AuthContext, + now_ms: u64, + items: Vec, + max_reply_bytes: Option, ) -> Result, CacheError> { self.prepare( auth, now_ms, &request_ids(items.iter().map(|i| i.header.request_id)), )?; + let mut remaining = self.limits.max_object_bytes; + if let Some(limit) = max_reply_bytes { + remaining = remaining.min(get_reply_payload_budget(limit, items.len())?); + } let mut out = Vec::with_capacity(items.len()); for item in items { let id = item.header.request_id; - let result = self.get_one(auth, now_ms, &item); + let result = self.get_one(auth, now_ms, &item, remaining); + if let Ok(hit) = &result { + remaining -= hit.length; + } out.push(ItemResult { request_id: id, body: result, @@ -1060,6 +1097,7 @@ impl MemoryStore { auth: &AuthContext, now_ms: u64, item: &GetItem, + remaining_reply_bytes: u64, ) -> Result { self.check_header(&item.header, now_ms)?; self.authorize(auth, Some(&item.identity))?; @@ -1068,8 +1106,16 @@ impl MemoryStore { reason: "pin ttl required", }); } + // Independent reader ceiling uses the existing request limit, so no + // public wire/config schema changes are needed. Check before cloning + // a payload. Expiry and cancellation do not release unknown pins. + if self.reader_pin_count() >= self.limits.max_in_flight_requests { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests, + }); + } let generation = item.generation; - let (visible, fail, state, region_id, payload, checksum, length, same_identity) = { + let (visible, fail, state, region_id, checksum, length, same_identity) = { let slot = self .slots .get(&generation) @@ -1080,7 +1126,6 @@ impl MemoryStore { slot.fail, slot.life.state, slot.region_id, - slot.bytes.clone().unwrap_or_default(), slot.checksum, slot.length, slot.identity == item.identity, @@ -1095,6 +1140,16 @@ impl MemoryStore { let region_id = region_id.ok_or(CacheError::Malformed { reason: "missing region", })?; + if length > remaining_reply_bytes { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + }); + } + let payload = self + .slots + .get(&generation) + .and_then(|slot| slot.bytes.clone()) + .unwrap_or_default(); if payload.len() as u64 != length || verify_checksum(&payload, checksum).is_err() { self.apply_event(generation, Event::CorruptHold, Some(FailKind::Corrupt))?; return Err(CacheError::ChecksumMismatch); @@ -1122,6 +1177,13 @@ impl MemoryStore { }) } + fn reader_pin_count(&self) -> u32 { + self.slots + .values() + .map(|slot| slot.pins.len() as u32) + .fold(0, u32::saturating_add) + } + fn release_one( &mut self, auth: &AuthContext, diff --git a/crates/storage/src/props.rs b/crates/storage/src/props.rs index 129cfed..45c15b6 100644 --- a/crates/storage/src/props.rs +++ b/crates/storage/src/props.rs @@ -164,6 +164,9 @@ fn random_operations_never_return_uncommitted_or_foreign_bytes() { next_id += 1; match &got.items[0].body { Ok(hit) => assert_eq!(hit.payload, *payload), + Err(CacheError::QuotaExceeded { + kind: gpucache_core_types::QuotaKind::InFlightRequests, + }) => assert_eq!(store.stats().reader_pins, limits.max_in_flight_requests), Err(err) => panic!("get of committed block failed: {err}"), } } @@ -171,6 +174,7 @@ fn random_operations_never_return_uncommitted_or_foreign_bytes() { assert!(stats.retained_bytes <= limits.max_retained_bytes); assert!(stats.registrations <= limits.max_registrations); assert!(stats.in_flight_requests <= limits.max_in_flight_requests); + assert!(stats.reader_pins <= limits.max_in_flight_requests); } let _ = CacheError::TimedOut; } diff --git a/crates/storage/src/tests.rs b/crates/storage/src/tests.rs index 2e01c60..7e392b4 100644 --- a/crates/storage/src/tests.rs +++ b/crates/storage/src/tests.rs @@ -623,6 +623,174 @@ fn quotas_reject_without_growing_past_the_ceiling() { assert!(store.stats().registrations <= 1); } +#[test] +fn reader_pin_cancel_storm_is_bounded_and_expiry_keeps_credit() { + let mut tight = limits(); + tight.max_in_flight_requests = 2; + let identity = synthetic_identity(1, 1); + let policy = + TenantPolicy::new(identity.tenant.clone(), identity.trust_domain.clone(), FP).unwrap(); + let auth = + AuthContext::new(identity.tenant.clone(), identity.trust_domain.clone(), FP).unwrap(); + let mut store = MemoryStore::new(tight, vec![policy]).unwrap(); + let generation = commit_bytes(&mut store, &auth, 1, b"pins"); + let mut region = 0; + for n in 10..12 { + let reply = store + .get( + &auth, + 0, + vec![GetItem { + header: hdr(&store, n, 0), + generation, + identity: identity.clone(), + pin_ttl_ms: 10, + }], + ) + .unwrap(); + region = reply.items[0].body.as_ref().unwrap().region_id; + } + for n in 12..140 { + let reply = store + .get( + &auth, + 0, + vec![GetItem { + header: hdr(&store, n, 0), + generation, + identity: identity.clone(), + pin_ttl_ms: 10, + }], + ) + .unwrap(); + assert!(matches!( + reply.items[0].body, + Err(CacheError::QuotaExceeded { + kind: gpucache_core_types::QuotaKind::InFlightRequests + }) + )); + assert_eq!(store.stats().reader_pins, 2); + } + for n in 200..328 { + let reply = store + .release_cancel( + &auth, + 20, + vec![ReleaseCancelItem { + header: hdr(&store, n, 20), + generation, + region_id: Some(region), + pin_owner: Some(RequestId::from_counter(10).unwrap()), + attestation: Attestation::Unknown, + }], + ) + .unwrap(); + assert!(reply.items[0].body.is_ok()); + assert_eq!(store.stats().reader_pins, 2); + } + let epoch = store.bump_epoch(20); + store + .reclaim_fenced( + &auth, + 20, + ReclaimRequest { + request_id: RequestId::from_counter(400).unwrap(), + epoch, + deadline_ms: 1000, + generation, + observed_epoch: epoch, + }, + ) + .unwrap(); + assert_eq!(store.stats().reader_pins, 0); +} + +#[test] +fn batched_get_bounds_reply_payload_before_copy_or_pin() { + let identity = synthetic_identity(1, 1); + let mut tight = limits(); + tight.max_object_bytes = 4; + let policy = + TenantPolicy::new(identity.tenant.clone(), identity.trust_domain.clone(), FP).unwrap(); + let auth = + AuthContext::new(identity.tenant.clone(), identity.trust_domain.clone(), FP).unwrap(); + let mut store = MemoryStore::new(tight, vec![policy]).unwrap(); + let generation = commit_bytes(&mut store, &auth, 1, b"data"); + let items = (10..12) + .map(|n| GetItem { + header: hdr(&store, n, 0), + generation, + identity: identity.clone(), + pin_ttl_ms: 100, + }) + .collect(); + let reply = store.get(&auth, 0, items).unwrap(); + assert_eq!(reply.items[0].body.as_ref().unwrap().payload, b"data"); + assert!(matches!( + reply.items[1].body, + Err(CacheError::QuotaExceeded { + kind: gpucache_core_types::QuotaKind::InFlightBytes + }) + )); + assert_eq!(store.stats().reader_pins, 1); +} + +#[test] +fn get_reply_limit_rejects_metadata_and_payload_overflow_without_pins() { + let (mut store, auth) = open(None); + let generation = commit_bytes(&mut store, &auth, 1, b"data"); + let item = |store: &MemoryStore, n| GetItem { + header: hdr(store, n, 0), + generation, + identity: synthetic_identity(1, 1), + pin_ttl_ms: 100, + }; + let quota = CacheError::QuotaExceeded { + kind: gpucache_core_types::QuotaKind::InFlightBytes, + }; + for (n, limit) in [0, 20, 64, 104].into_iter().enumerate() { + let request = item(&store, 10 + n as u128); + assert_eq!( + store.get_with_reply_limit(&auth, 0, vec![request], limit), + Err(quota.clone()) + ); + assert_eq!(store.stats().reader_pins, 0); + } + for (n, limit) in [105, 108].into_iter().enumerate() { + let request = item(&store, 20 + n as u128); + let reply = store + .get_with_reply_limit(&auth, 0, vec![request], limit) + .unwrap(); + assert_eq!(reply.items[0].body, Err(quota.clone())); + assert_eq!(store.stats().reader_pins, 0); + assert!( + gpucache_protocol::encode_reply(&gpucache_protocol::ReplyBody::Get(reply)) + .unwrap() + .len() + <= limit + ); + } + let mut wrong_auth = auth.clone(); + wrong_auth.credential_fingerprint = [8; 32]; + let request = item(&store, 30); + assert_eq!( + store.get_with_reply_limit(&wrong_auth, 0, vec![request], 0), + Err(CacheError::Unauthorized) + ); + let request = item(&store, 31); + let reply = store + .get_with_reply_limit(&auth, 0, vec![request], 109) + .unwrap(); + assert_eq!(reply.items[0].body.as_ref().unwrap().payload, b"data"); + assert_eq!(store.stats().reader_pins, 1); + assert_eq!( + gpucache_protocol::encode_reply(&gpucache_protocol::ReplyBody::Get(reply)) + .unwrap() + .len(), + 109 + ); +} + #[test] fn idempotent_replay_and_conflict() { let (mut store, auth) = open(None); diff --git a/crates/transport/src/rate.rs b/crates/transport/src/rate.rs index 010ec92..e627c20 100644 --- a/crates/transport/src/rate.rs +++ b/crates/transport/src/rate.rs @@ -18,19 +18,21 @@ pub struct RateGate { default_limit: u32, window_ms: u64, max_tenants: usize, + last_now_ms: Option, limits: BTreeMap, u32>, windows: BTreeMap, Window>, } impl RateGate { pub fn new(default_limit: u32, window_ms: u64, max_tenants: usize) -> Result { - if default_limit == 0 || window_ms == 0 || max_tenants == 0 { + if default_limit == 0 || window_ms == 0 || max_tenants == 0 || max_tenants > 1024 { return Err(CacheError::Malformed { reason: "rate" }); } Ok(Self { default_limit, window_ms, max_tenants, + last_now_ms: None, limits: BTreeMap::new(), windows: BTreeMap::new(), }) @@ -50,14 +52,44 @@ impl RateGate { } pub fn admit(&mut self, tenant: &[u8], now_ms: u64) -> Result<(), CacheError> { + self.admit_batch(tenant, now_ms, 1) + } + + /// Charge a complete batch atomically, so a batch cannot bypass the item + /// budget or consume a partial admission when it exceeds the budget. + pub fn admit_batch( + &mut self, + tenant: &[u8], + now_ms: u64, + items: usize, + ) -> Result<(), CacheError> { if tenant.is_empty() || tenant.len() > 128 { return Err(CacheError::Malformed { reason: "rate" }); } + let items = u32::try_from(items) + .ok() + .filter(|items| *items != 0) + .ok_or(CacheError::Malformed { reason: "rate" })?; + if self.last_now_ms.is_some_and(|last| now_ms < last) { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::RateLimit, + }); + } + self.last_now_ms = Some(now_ms); let limit = self .limits .get(tenant) .copied() .unwrap_or(self.default_limit); + if items > limit { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::RateLimit, + }); + } + // Expired tenants can release table capacity. Never evict a live + // window: doing so would reset that tenant's admission budget. + self.windows + .retain(|_, window| now_ms.saturating_sub(window.start_ms) < self.window_ms); if !self.windows.contains_key(tenant) && self.windows.len() >= self.max_tenants { return Err(CacheError::QuotaExceeded { kind: QuotaKind::RateLimit, @@ -67,16 +99,12 @@ impl RateGate { start_ms: now_ms, used: 0, }); - if now_ms.saturating_sub(window.start_ms) >= self.window_ms { - window.start_ms = now_ms; - window.used = 0; - } - if window.used >= limit { + let Some(used) = window.used.checked_add(items).filter(|used| *used <= limit) else { return Err(CacheError::QuotaExceeded { kind: QuotaKind::RateLimit, }); - } - window.used = window.used.saturating_add(1); + }; + window.used = used; Ok(()) } } @@ -120,4 +148,36 @@ mod tests { } ); } + + #[test] + fn batch_failure_preserves_the_remaining_budget() { + let mut gate = RateGate::new(3, 100, 1).unwrap(); + gate.admit_batch(b"tenant-a", 0, 2).unwrap(); + assert_eq!( + gate.admit_batch(b"tenant-a", 1, 2), + Err(CacheError::QuotaExceeded { + kind: QuotaKind::RateLimit, + }) + ); + gate.admit(b"tenant-a", 2).unwrap(); + assert!(gate.admit(b"tenant-b", 3).is_err()); + gate.admit(b"tenant-b", 100).unwrap(); + assert_eq!(gate.windows.len(), 1); + assert!(gate.admit_batch(b"tenant-b", 101, 0).is_err()); + assert!(RateGate::new(1, 100, 1025).is_err()); + } + + #[test] + fn oversized_first_batch_does_not_take_the_only_tenant_slot() { + let mut gate = RateGate::new(1, 100, 1).unwrap(); + assert_eq!( + gate.admit_batch(b"tenant-a", 0, 2), + Err(CacheError::QuotaExceeded { + kind: QuotaKind::RateLimit + }) + ); + assert!(gate.windows.is_empty()); + gate.admit(b"tenant-b", 0).unwrap(); + assert_eq!(gate.windows.len(), 1); + } } diff --git a/crates/transport/src/server.rs b/crates/transport/src/server.rs index 1fec976..1215ce9 100644 --- a/crates/transport/src/server.rs +++ b/crates/transport/src/server.rs @@ -1,22 +1,25 @@ // Copyright 2024-2026 RustFS, Inc. // SPDX-License-Identifier: Apache-2.0 -use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream}; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::thread::JoinHandle; use std::time::Duration; -use gpucache_core_types::CacheError; +use gpucache_core_types::{Attestation, CacheError, QuotaKind}; use gpucache_protocol::{ - decode_hello, decode_request, encode_hello_ack, encode_reply, negotiate, request_op, ReplyBody, - RequestBody, OP_HELLO, WIRE_VERSION, + decode_hello, decode_request, encode_hello_ack, encode_reply, negotiate, request_op, Acl, + AuditEvent, AuthContext, ReplyBody, RequestBody, SecurityLog, OP_HELLO, WIRE_VERSION, }; use gpucache_storage::FileStore; use crate::clock::mono_ms; use crate::frame::{read_envelope, write_envelope}; use crate::pool::Pool; +use crate::rate::RateGate; + +const MAX_TENANTS: usize = 1024; #[derive(Clone, Debug)] pub struct ServerConfig { @@ -26,6 +29,13 @@ pub struct ServerConfig { pub max_frame_bytes: usize, pub pool_buffers: usize, pub io_timeout: Duration, + /// Optional restriction of the store's policies. `None` grants configured + /// store principals read/write; `Some(empty_acl)` denies every data call. + /// An ACL grant never bypasses the store's credential/tenant check. + pub acl: Option, + /// Maximum admitted data items per tenant per window, across connections. + pub rate_limit_items: u32, + pub rate_window_ms: u64, } impl Default for ServerConfig { @@ -37,6 +47,9 @@ impl Default for ServerConfig { max_frame_bytes: 1024 * 1024, pool_buffers: 16, io_timeout: Duration::from_secs(5), + acl: None, + rate_limit_items: 16_384, + rate_window_ms: 1_000, } } } @@ -47,6 +60,8 @@ impl ServerConfig { || self.pool_buffers == 0 || self.max_frame_bytes < 64 || self.io_timeout.is_zero() + || self.rate_limit_items == 0 + || self.rate_window_ms == 0 { return Err(CacheError::Malformed { reason: "server config", @@ -71,18 +86,52 @@ pub struct ServerHandle { addr: SocketAddr, shutdown: Arc, thread: Option>, + security: Arc>, } impl ServerHandle { pub fn addr(&self) -> SocketAddr { self.addr } + + /// Bounded audit snapshot; no fingerprints, request identities or payloads. + pub fn security_events(&self) -> Vec { + self.security + .lock() + .unwrap_or_else(|err| err.into_inner()) + .log + .events() + .copied() + .collect() + } + + pub fn denied_requests(&self) -> u64 { + self.security + .lock() + .unwrap_or_else(|err| err.into_inner()) + .denied + } + + pub fn rate_limited_requests(&self) -> u64 { + self.security + .lock() + .unwrap_or_else(|err| err.into_inner()) + .rate_limited + } } impl Drop for ServerHandle { fn drop(&mut self) { self.shutdown.store(true, Ordering::Relaxed); - let _ = TcpStream::connect(self.addr); + let mut wake_addr = self.addr; + if wake_addr.ip().is_unspecified() { + wake_addr.set_ip(if wake_addr.is_ipv4() { + std::net::Ipv4Addr::LOCALHOST.into() + } else { + std::net::Ipv6Addr::LOCALHOST.into() + }); + } + let _ = TcpStream::connect(wake_addr); if let Some(thread) = self.thread.take() { let _ = thread.join(); } @@ -95,11 +144,101 @@ struct Shared { cfg: ServerConfig, shutdown: Arc, connections: AtomicUsize, + security: Arc>, +} + +struct HotSecurity { + principals: Acl, + restriction: Option, + rate: RateGate, + log: SecurityLog, + denied: u64, + rate_limited: u64, +} + +impl HotSecurity { + fn open(store: &FileStore, cfg: &ServerConfig) -> Result { + let mut principals = Acl::new(MAX_TENANTS)?; + for policy in store.tenant_policies() { + principals.grant( + AuthContext::new( + policy.tenant, + policy.trust_domain, + policy.credential_fingerprint, + )?, + true, + )?; + } + Ok(Self { + principals, + restriction: cfg.acl.clone(), + rate: RateGate::new(cfg.rate_limit_items, cfg.rate_window_ms, MAX_TENANTS)?, + log: SecurityLog::new(256)?, + denied: 0, + rate_limited: 0, + }) + } + + fn admit( + &mut self, + auth: &AuthContext, + request: &RequestBody, + now: u64, + ) -> Result<(), CacheError> { + // Reader pin releases must remain available to a read-only principal. + // A mixed batch that includes a writer cancellation requires write. + let write = match request { + RequestBody::Lookup(_) | RequestBody::Get(_) => false, + RequestBody::ReleaseCancel(items) => items.iter().any(|item| item.pin_owner.is_none()), + RequestBody::ReservePut(_) | RequestBody::Commit(_) | RequestBody::Invalidate(_) => { + true + } + }; + let authorized = self.principals.check(auth, write).and_then(|()| { + self.restriction + .as_ref() + .map_or(Ok(()), |acl| acl.check(auth, write)) + }); + if let Err(error) = authorized { + self.denied = self.denied.saturating_add(1); + self.log.push(AuditEvent::Deny); + return Err(error); + } + if request.is_empty() { + return Err(CacheError::Malformed { + reason: "empty batch", + }); + } + // A full admission window must not strand an existing reader pin. + // Storage still checks tenant, region and pin owner, and its bounded + // idempotency table applies. Unknown completion does not bypass rate + // admission because it cannot release the held reference. + if matches!(request, RequestBody::ReleaseCancel(items) if items.iter().all(|item| { + item.pin_owner.is_some() + && item.region_id.is_some() + && item.attestation == Attestation::Drained + })) { + return Ok(()); + } + let result = self + .rate + .admit_batch(auth.tenant.as_bytes(), now, request.len()); + if matches!( + result, + Err(CacheError::QuotaExceeded { + kind: QuotaKind::RateLimit + }) + ) { + self.rate_limited = self.rate_limited.saturating_add(1); + } + result + } } /// Bind a loopback TCP cache service. `store` must already have tenant policies. pub fn serve(store: FileStore, cfg: ServerConfig) -> Result { cfg.validate()?; + let security = Arc::new(Mutex::new(HotSecurity::open(&store, &cfg)?)); let listener = TcpListener::bind(cfg.bind).map_err(|_| CacheError::Storage { reason: "bind" })?; let addr = listener @@ -112,6 +251,7 @@ pub fn serve(store: FileStore, cfg: ServerConfig) -> Result Result) { + // Keep bounded ownership of workers. Drop must interrupt and join them + // before the file-backed directory can safely be reopened after restart. + let mut workers: Vec<(JoinHandle<()>, TcpStream)> = Vec::new(); loop { if shared.shutdown.load(Ordering::Relaxed) { break; @@ -136,8 +280,20 @@ fn accept_loop(listener: TcpListener, shared: Arc) { if shared.shutdown.load(Ordering::Relaxed) { break; } + let mut index = 0; + while index < workers.len() { + if workers[index].0.is_finished() { + let (worker, _) = workers.swap_remove(index); + let _ = worker.join(); + } else { + index += 1; + } + } + let Ok(shutdown_stream) = stream.try_clone() else { + continue; + }; let previous = shared.connections.fetch_add(1, Ordering::Relaxed); - if previous >= shared.cfg.max_connections { + if previous >= shared.cfg.max_connections || workers.len() >= shared.cfg.max_connections { shared.connections.fetch_sub(1, Ordering::Relaxed); continue; } @@ -154,10 +310,19 @@ fn accept_loop(listener: TcpListener, shared: Arc) { let _guard = Guard(Arc::clone(&shared_for_thread)); connection(stream, shared_for_thread); }); - if spawned.is_err() { - shared.connections.fetch_sub(1, Ordering::Relaxed); + match spawned { + Ok(worker) => workers.push((worker, shutdown_stream)), + Err(_) => { + shared.connections.fetch_sub(1, Ordering::Relaxed); + } } } + for (_, stream) in &workers { + let _ = stream.shutdown(Shutdown::Both); + } + for (worker, _) in workers { + let _ = worker.join(); + } } fn connection(mut stream: TcpStream, shared: Arc) { @@ -176,7 +341,7 @@ fn connection(mut stream: TcpStream, shared: Arc) { Ok(envelope) => envelope, Err(_) => break, }; - let reply = handle_frame(&shared.store, envelope.body()); + let reply = handle_frame(&shared, envelope.body()); // The decoder owns the request; return the read buffer before writing. drop(envelope); if write_envelope(&mut stream, mono_ms(), &reply, shared.cfg.max_frame_bytes).is_err() { @@ -185,7 +350,8 @@ fn connection(mut stream: TcpStream, shared: Arc) { } } -fn handle_frame(store: &Mutex, body: &[u8]) -> Vec { +fn handle_frame(shared: &Shared, body: &[u8]) -> Vec { + let store = &shared.store; if body.len() >= 8 && &body[..4] == b"GCK1" { let op = u16::from_le_bytes([body[6], body[7]]); if op == OP_HELLO { @@ -203,7 +369,22 @@ fn handle_frame(store: &Mutex, body: &[u8]) -> Vec { let mut store = store.lock().unwrap_or_else(|err| err.into_inner()); // Time is checked after waiting for the store, not before contention. match decoded { - Ok((auth, request)) => dispatch(&mut store, &auth, request, mono_ms()), + Ok((auth, request)) => { + let now = mono_ms(); + let admission = shared + .security + .lock() + .unwrap_or_else(|err| err.into_inner()) + .admit(&auth, &request, now); + match admission { + Ok(()) => dispatch(&mut store, &auth, request, now, shared.cfg.max_frame_bytes), + Err(error) => ReplyBody::Failed { + op: request_op(&request), + store_epoch: store.epoch(), + error, + }, + } + } Err(error) => ReplyBody::Failed { op: 0, store_epoch: store.epoch(), @@ -222,6 +403,7 @@ fn dispatch( auth: &gpucache_protocol::AuthContext, request: RequestBody, now: u64, + max_reply_bytes: usize, ) -> ReplyBody { let op = request_op(&request); let result = match request { @@ -230,7 +412,9 @@ fn dispatch( store.reserve_put(auth, now, items).map(ReplyBody::Reserve) } RequestBody::Commit(items) => store.commit(auth, now, items).map(ReplyBody::Commit), - RequestBody::Get(items) => store.get(auth, now, items).map(ReplyBody::Get), + RequestBody::Get(items) => store + .get_with_reply_limit(auth, now, items, max_reply_bytes) + .map(ReplyBody::Get), RequestBody::ReleaseCancel(items) => store .release_cancel(auth, now, items) .map(ReplyBody::Release), diff --git a/crates/transport/tests/hot_security.rs b/crates/transport/tests/hot_security.rs new file mode 100644 index 0000000..d158905 --- /dev/null +++ b/crates/transport/tests/hot_security.rs @@ -0,0 +1,585 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! Actual loopback TCP gates. Synthetic CPU bytes, not an engine or TLS test. + +use std::fs; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use gpucache_core_types::{ + content_checksum, synthetic_identity, Attestation, CacheError, CacheIdentity, QuotaKind, + RequestId, ResourceLimits, TenantId, TrustDomain, +}; +use gpucache_protocol::{ + Acl, AuditEvent, AuthContext, CommitItem, GetItem, InvalidateItem, ItemHeader, LookupItem, + LookupOutcome, ReleaseCancelItem, ReservePutItem, TenantPolicy, +}; +use gpucache_storage::FileStore; +use gpucache_transport::{serve, ServerConfig, ServerHandle, TcpClient}; + +const PAYLOAD: &[u8] = b"synthetic-cpu-kv"; + +struct Fixture { + path: PathBuf, + identity: CacheIdentity, + auth: AuthContext, + other_auth: AuthContext, +} + +impl Fixture { + fn new() -> Self { + static NEXT: AtomicU64 = AtomicU64::new(0); + let path = std::env::temp_dir().join(format!( + "gpucache-hot-security-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + let identity = synthetic_identity(1, 1); + let auth = AuthContext::new( + identity.tenant.clone(), + identity.trust_domain.clone(), + [9; 32], + ) + .unwrap(); + let other_auth = AuthContext::new( + TenantId::new(b"tenant-b".to_vec()).unwrap(), + identity.trust_domain.clone(), + [8; 32], + ) + .unwrap(); + Self { + path, + identity, + auth, + other_auth, + } + } + + fn open(&self) -> FileStore { + let policies = [&self.auth, &self.other_auth] + .into_iter() + .map(|auth| { + TenantPolicy::new( + auth.tenant.clone(), + auth.trust_domain.clone(), + auth.credential_fingerprint, + ) + .unwrap() + }) + .collect(); + FileStore::open(&self.path, ResourceLimits::dev_defaults(), policies).unwrap() + } + + fn seeded(&self) -> (FileStore, u64) { + let mut store = self.open(); + let head = |n| ItemHeader { + request_id: RequestId::from_counter(n).unwrap(), + epoch: store.epoch(), + deadline_ms: u64::MAX, + }; + let put_header = head(1); + let commit_header = head(2); + let generation = store + .reserve_put( + &self.auth, + 0, + vec![ReservePutItem { + header: put_header, + identity: self.identity.clone(), + generation: 0, + length: PAYLOAD.len() as u64, + checksum: content_checksum(PAYLOAD), + payload: Some(PAYLOAD.to_vec()), + lease_ttl_ms: 30_000, + }], + ) + .unwrap() + .items[0] + .body + .as_ref() + .unwrap() + .generation; + store + .commit( + &self.auth, + 0, + vec![CommitItem { + header: commit_header, + generation, + length: PAYLOAD.len() as u64, + checksum: content_checksum(PAYLOAD), + }], + ) + .unwrap() + .items[0] + .body + .as_ref() + .unwrap(); + (store, generation) + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.path); + } +} + +fn connect(server: &ServerHandle, auth: AuthContext) -> TcpClient { + TcpClient::connect(server.addr(), auth, Duration::from_secs(5), 1024 * 1024).unwrap() +} + +fn header(client: &TcpClient, n: u128) -> ItemHeader { + ItemHeader { + request_id: RequestId::from_counter(n).unwrap(), + epoch: client.epoch(), + deadline_ms: client.deadline(30_000), + } +} + +fn lookup(client: &TcpClient, n: u128, identity: &CacheIdentity) -> LookupItem { + LookupItem { + header: header(client, n), + identity: identity.clone(), + } +} + +fn put(client: &TcpClient, n: u128, identity: &CacheIdentity) -> ReservePutItem { + ReservePutItem { + header: header(client, n), + identity: identity.clone(), + generation: 0, + length: PAYLOAD.len() as u64, + checksum: content_checksum(PAYLOAD), + payload: Some(PAYLOAD.to_vec()), + lease_ttl_ms: 30_000, + } +} + +fn commit(client: &TcpClient, n: u128, generation: u64) -> CommitItem { + CommitItem { + header: header(client, n), + generation, + length: PAYLOAD.len() as u64, + checksum: content_checksum(PAYLOAD), + } +} + +fn get(client: &TcpClient, n: u128, identity: &CacheIdentity, generation: u64) -> GetItem { + GetItem { + header: header(client, n), + generation, + identity: identity.clone(), + pin_ttl_ms: 30_000, + } +} + +fn cancel(client: &TcpClient, n: u128, generation: u64) -> ReleaseCancelItem { + ReleaseCancelItem { + header: header(client, n), + generation, + region_id: None, + pin_owner: None, + attestation: Attestation::Drained, + } +} + +fn invalidate(client: &TcpClient, n: u128, generation: u64) -> InvalidateItem { + InvalidateItem { + header: header(client, n), + generation, + } +} + +fn every_entry_denies(client: &mut TcpClient, identity: &CacheIdentity, generation: u64) { + assert_eq!( + client.lookup(vec![lookup(client, 10, identity)]), + Err(CacheError::Unauthorized) + ); + assert_eq!( + client.reserve_put(vec![put(client, 11, identity)]), + Err(CacheError::Unauthorized) + ); + assert_eq!( + client.commit(vec![commit(client, 12, generation)]), + Err(CacheError::Unauthorized) + ); + assert_eq!( + client.get(vec![get(client, 13, identity, generation)]), + Err(CacheError::Unauthorized) + ); + assert_eq!( + client.release_cancel(vec![cancel(client, 14, generation)]), + Err(CacheError::Unauthorized) + ); + assert_eq!( + client.invalidate(vec![invalidate(client, 15, generation)]), + Err(CacheError::Unauthorized) + ); +} + +#[test] +fn empty_acl_denies_every_tcp_entry_even_with_a_valid_store_credential() { + let fixture = Fixture::new(); + let (store, generation) = fixture.seeded(); + let server = serve( + store, + ServerConfig { + acl: Some(Acl::new(1).unwrap()), + ..ServerConfig::default() + }, + ) + .unwrap(); + let mut client = connect(&server, fixture.auth.clone()); + every_entry_denies(&mut client, &fixture.identity, generation); + assert_eq!(server.denied_requests(), 6); + assert_eq!(server.security_events(), vec![AuditEvent::Deny; 6]); +} + +#[test] +fn readonly_tcp_acl_allows_get_and_pin_release_but_denies_all_writer_paths() { + let fixture = Fixture::new(); + let (store, generation) = fixture.seeded(); + let mut acl = Acl::new(2).unwrap(); + acl.grant(fixture.auth.clone(), false).unwrap(); + let server = serve( + store, + ServerConfig { + acl: Some(acl), + ..ServerConfig::default() + }, + ) + .unwrap(); + let mut client = connect(&server, fixture.auth.clone()); + assert!(matches!( + client + .lookup(vec![lookup(&client, 10, &fixture.identity)]) + .unwrap() + .items[0] + .body, + Ok(LookupOutcome::Hit { .. }) + )); + for denied in [ + client + .reserve_put(vec![put(&client, 11, &fixture.identity)]) + .err(), + client.commit(vec![commit(&client, 12, generation)]).err(), + client + .release_cancel(vec![cancel(&client, 13, generation)]) + .err(), + client + .invalidate(vec![invalidate(&client, 14, generation)]) + .err(), + ] { + assert_eq!(denied, Some(CacheError::Unauthorized)); + } + let request = get(&client, 15, &fixture.identity, generation); + let owner = request.header.request_id; + let hit = client + .get(vec![request]) + .unwrap() + .items + .remove(0) + .body + .unwrap(); + assert_eq!(hit.payload, PAYLOAD); + let release = ReleaseCancelItem { + header: header(&client, 16), + generation, + region_id: Some(hit.region_id), + pin_owner: Some(owner), + attestation: Attestation::Drained, + }; + assert_eq!( + client.release_cancel(vec![release.clone(), cancel(&client, 17, generation)]), + Err(CacheError::Unauthorized) + ); + // Denying the mixed batch did not consume or release the first pin. + client.release_cancel(vec![release]).unwrap().items[0] + .body + .as_ref() + .unwrap(); + assert_eq!(server.denied_requests(), 5); +} + +#[test] +fn an_acl_grant_cannot_override_store_tenant_trust_or_fingerprint_policies() { + let fixture = Fixture::new(); + let (store, generation) = fixture.seeded(); + let mut wrong_fingerprint = fixture.auth.clone(); + wrong_fingerprint.credential_fingerprint = [7; 32]; + let mut wrong_trust = fixture.auth.clone(); + wrong_trust.trust_domain = TrustDomain::new(b"other-trust".to_vec()).unwrap(); + let mut unknown_tenant = fixture.auth.clone(); + unknown_tenant.tenant = TenantId::new(b"unknown-tenant".to_vec()).unwrap(); + let mut acl = Acl::new(4).unwrap(); + for auth in [ + &wrong_fingerprint, + &wrong_trust, + &unknown_tenant, + &fixture.auth, + ] { + acl.grant(auth.clone(), true).unwrap(); + } + let server = serve( + store, + ServerConfig { + acl: Some(acl), + ..ServerConfig::default() + }, + ) + .unwrap(); + for auth in [wrong_fingerprint, wrong_trust, unknown_tenant] { + let mut client = connect(&server, auth); + every_entry_denies(&mut client, &fixture.identity, generation); + } + let mut valid = connect(&server, fixture.auth.clone()); + let hit = valid + .get(vec![get(&valid, 100, &fixture.identity, generation)]) + .unwrap(); + assert_eq!(hit.items[0].body.as_ref().unwrap().payload, PAYLOAD); + assert_eq!(server.denied_requests(), 18); +} + +#[test] +fn valid_other_tenant_cannot_read_or_mutate_a_foreign_generation() { + let fixture = Fixture::new(); + let (store, generation) = fixture.seeded(); + let server = serve(store, ServerConfig::default()).unwrap(); + let mut other = connect(&server, fixture.other_auth.clone()); + assert_eq!( + other + .lookup(vec![lookup(&other, 10, &fixture.identity)]) + .unwrap() + .items[0] + .body, + Err(CacheError::Unauthorized) + ); + assert_eq!( + other + .reserve_put(vec![put(&other, 11, &fixture.identity)]) + .unwrap() + .items[0] + .body, + Err(CacheError::Unauthorized) + ); + let mut own_identity = fixture.identity.clone(); + own_identity.tenant = fixture.other_auth.tenant.clone(); + assert_eq!( + other + .get(vec![get(&other, 12, &own_identity, generation)]) + .unwrap() + .items[0] + .body, + Err(CacheError::Unauthorized) + ); + assert_eq!( + other + .commit(vec![commit(&other, 13, generation)]) + .unwrap() + .items[0] + .body, + Err(CacheError::Unauthorized) + ); + assert_eq!( + other + .release_cancel(vec![cancel(&other, 14, generation)]) + .unwrap() + .items[0] + .body, + Err(CacheError::Unauthorized) + ); + assert_eq!( + other + .invalidate(vec![invalidate(&other, 15, generation)]) + .unwrap() + .items[0] + .body, + Err(CacheError::Unauthorized) + ); + let mut owner = connect(&server, fixture.auth.clone()); + for mut wrong in [ + synthetic_identity(2, 1), + synthetic_identity(1, 2), + fixture.identity.clone(), + ] { + if wrong == fixture.identity { + wrong.trust_domain = TrustDomain::new(b"wrong-trust".to_vec()).unwrap(); + } + let hit = owner + .get(vec![get(&owner, 100, &wrong, generation)]) + .unwrap(); + assert!(matches!( + hit.items[0].body, + Err(CacheError::Incompatible | CacheError::Unauthorized) + )); + } + assert_eq!( + owner + .get(vec![get(&owner, 101, &fixture.identity, generation)]) + .unwrap() + .items[0] + .body + .as_ref() + .unwrap() + .payload, + PAYLOAD + ); +} + +#[test] +fn tcp_rate_budget_is_atomic_shared_by_connections_and_separate_by_tenant() { + let fixture = Fixture::new(); + let server = serve( + fixture.open(), + ServerConfig { + rate_limit_items: 3, + rate_window_ms: 60_000, + ..ServerConfig::default() + }, + ) + .unwrap(); + let mut first = connect(&server, fixture.auth.clone()); + let mut second = connect(&server, fixture.auth.clone()); + first + .lookup(vec![ + lookup(&first, 10, &fixture.identity), + lookup(&first, 11, &fixture.identity), + ]) + .unwrap(); + assert_eq!( + second.lookup(vec![ + lookup(&second, 12, &fixture.identity), + lookup(&second, 13, &fixture.identity), + ]), + Err(CacheError::QuotaExceeded { + kind: QuotaKind::RateLimit + }) + ); + second + .lookup(vec![lookup(&second, 14, &fixture.identity)]) + .unwrap(); + assert_eq!( + first.lookup(vec![lookup(&first, 15, &fixture.identity)]), + Err(CacheError::QuotaExceeded { + kind: QuotaKind::RateLimit + }) + ); + let mut quiet = connect(&server, fixture.other_auth.clone()); + let mut quiet_identity = fixture.identity.clone(); + quiet_identity.tenant = fixture.other_auth.tenant.clone(); + quiet + .lookup(vec![lookup(&quiet, 16, &quiet_identity)]) + .unwrap(); + assert_eq!(server.rate_limited_requests(), 2); +} + +#[test] +fn exhausted_tcp_rate_budget_still_allows_authorized_reader_pin_drain() { + let fixture = Fixture::new(); + let (store, generation) = fixture.seeded(); + let mut acl = Acl::new(1).unwrap(); + acl.grant(fixture.auth.clone(), false).unwrap(); + let server = serve( + store, + ServerConfig { + acl: Some(acl), + rate_limit_items: 1, + rate_window_ms: 60_000, + ..ServerConfig::default() + }, + ) + .unwrap(); + let mut client = connect(&server, fixture.auth.clone()); + let request = get(&client, 10, &fixture.identity, generation); + let pin_owner = request.header.request_id; + let hit = client + .get(vec![request]) + .unwrap() + .items + .remove(0) + .body + .unwrap(); + let release = ReleaseCancelItem { + header: header(&client, 11), + generation, + region_id: Some(hit.region_id), + pin_owner: Some(pin_owner), + attestation: Attestation::Drained, + }; + let mut bad_auth = fixture.auth.clone(); + bad_auth.credential_fingerprint = [7; 32]; + let mut attacker = connect(&server, bad_auth); + assert_eq!( + attacker.release_cancel(vec![release.clone()]), + Err(CacheError::Unauthorized) + ); + client.release_cancel(vec![release]).unwrap().items[0] + .body + .as_ref() + .unwrap(); + assert_eq!(server.rate_limited_requests(), 0); + assert_eq!(server.denied_requests(), 1); + assert_eq!( + client.get(vec![get(&client, 12, &fixture.identity, generation)]), + Err(CacheError::QuotaExceeded { + kind: QuotaKind::RateLimit + }) + ); +} + +#[test] +fn tcp_deny_storm_has_a_bounded_audit_and_does_not_spend_the_victims_rate_budget() { + let fixture = Fixture::new(); + let server = serve( + fixture.open(), + ServerConfig { + rate_limit_items: 1, + rate_window_ms: 60_000, + ..ServerConfig::default() + }, + ) + .unwrap(); + let mut wrong = fixture.auth.clone(); + wrong.credential_fingerprint = [7; 32]; + let mut attacker = connect(&server, wrong); + for n in 1..=300 { + assert_eq!( + attacker.lookup(vec![lookup(&attacker, n, &fixture.identity)]), + Err(CacheError::Unauthorized) + ); + } + assert_eq!(server.denied_requests(), 300); + assert_eq!(server.security_events(), vec![AuditEvent::Deny; 256]); + let mut victim = connect(&server, fixture.auth.clone()); + victim + .lookup(vec![lookup(&victim, 500, &fixture.identity)]) + .unwrap(); +} + +#[test] +fn server_shutdown_closes_idle_connections_before_reopening_the_store() { + let fixture = Fixture::new(); + let (store, generation) = fixture.seeded(); + let server = serve(store, ServerConfig::default()).unwrap(); + let mut stale = connect(&server, fixture.auth.clone()); + let previous_epoch = stale.epoch(); + drop(server); + let store = fixture.open(); + assert_eq!(store.epoch(), previous_epoch + 1); + let server = serve(store, ServerConfig::default()).unwrap(); + assert!(stale + .invalidate(vec![invalidate(&stale, 10, generation)]) + .is_err()); + let mut fresh = connect(&server, fixture.auth.clone()); + assert!(matches!( + fresh + .lookup(vec![lookup(&fresh, 11, &fixture.identity)]) + .unwrap() + .items[0] + .body, + Ok(LookupOutcome::Hit { .. }) + )); +} diff --git a/crates/transport/tests/reply_budget.rs b/crates/transport/tests/reply_budget.rs new file mode 100644 index 0000000..0460a5a --- /dev/null +++ b/crates/transport/tests/reply_budget.rs @@ -0,0 +1,250 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! Framed Get admission over real loopback TCP with synthetic CPU payloads. + +use std::fs; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use gpucache_core_types::{ + content_checksum, synthetic_identity, Attestation, CacheError, QuotaKind, RequestId, + ResourceLimits, +}; +use gpucache_protocol::{ + AuthContext, CommitItem, GetItem, ItemHeader, LookupItem, LookupOutcome, ReleaseCancelItem, + ReservePutItem, TenantPolicy, +}; +use gpucache_storage::FileStore; +use gpucache_transport::{serve, ServerConfig, ServerHandle, TcpClient}; + +const FRAME_BYTES: usize = 2048; + +struct Fixture { + path: PathBuf, + auth: AuthContext, +} + +impl Fixture { + fn new() -> Self { + static NEXT: AtomicU64 = AtomicU64::new(0); + let identity = synthetic_identity(1, 1); + Self { + path: std::env::temp_dir().join(format!( + "gpucache-reply-budget-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )), + auth: AuthContext::new(identity.tenant, identity.trust_domain, [9; 32]).unwrap(), + } + } + + fn open(&self, max_pins: u32) -> FileStore { + let mut limits = ResourceLimits::dev_defaults(); + limits.max_in_flight_requests = max_pins; + FileStore::open( + &self.path, + limits, + vec![TenantPolicy::new( + self.auth.tenant.clone(), + self.auth.trust_domain.clone(), + self.auth.credential_fingerprint, + ) + .unwrap()], + ) + .unwrap() + } + + fn seed(&self, store: &mut FileStore, family: u16, payload: &[u8]) -> u64 { + let header = |n| ItemHeader { + request_id: RequestId::from_counter(n).unwrap(), + epoch: store.epoch(), + deadline_ms: u64::MAX, + }; + let reserve = header(u128::from(family) * 2); + let commit = header(u128::from(family) * 2 + 1); + let generation = store + .reserve_put( + &self.auth, + 0, + vec![ReservePutItem { + header: reserve, + identity: synthetic_identity(1, family), + generation: 0, + length: payload.len() as u64, + checksum: content_checksum(payload), + payload: Some(payload.to_vec()), + lease_ttl_ms: 30_000, + }], + ) + .unwrap() + .items[0] + .body + .as_ref() + .unwrap() + .generation; + store + .commit( + &self.auth, + 0, + vec![CommitItem { + header: commit, + generation, + length: payload.len() as u64, + checksum: content_checksum(payload), + }], + ) + .unwrap() + .items[0] + .body + .as_ref() + .unwrap(); + generation + } + + fn serve(&self, store: FileStore) -> (ServerHandle, TcpClient) { + let server = serve( + store, + ServerConfig { + max_frame_bytes: FRAME_BYTES, + ..ServerConfig::default() + }, + ) + .unwrap(); + let client = TcpClient::connect( + server.addr(), + self.auth.clone(), + Duration::from_secs(5), + FRAME_BYTES, + ) + .unwrap(); + (server, client) + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.path); + } +} + +fn header(client: &TcpClient, n: u128) -> ItemHeader { + ItemHeader { + request_id: RequestId::from_counter(n).unwrap(), + epoch: client.epoch(), + deadline_ms: client.deadline(30_000), + } +} + +fn get(client: &TcpClient, n: u128, family: u16, generation: u64) -> GetItem { + GetItem { + header: header(client, n), + identity: synthetic_identity(1, family), + generation, + pin_ttl_ms: 30_000, + } +} + +fn release(client: &mut TcpClient, n: u128, owner: u128, generation: u64, region_id: u64) { + client + .release_cancel(vec![ReleaseCancelItem { + header: header(client, n), + generation, + region_id: Some(region_id), + pin_owner: Some(RequestId::from_counter(owner).unwrap()), + attestation: Attestation::Drained, + }]) + .unwrap() + .items[0] + .body + .as_ref() + .unwrap(); +} + +#[test] +fn oversized_get_creates_no_pin_and_keeps_the_tcp_connection_synchronized() { + let fixture = Fixture::new(); + let mut store = fixture.open(1); + // A successful single-item GCKR reply adds exactly 105 metadata bytes. + let too_large = fixture.seed(&mut store, 1, &vec![1; FRAME_BYTES - 104]); + let exact = fixture.seed(&mut store, 2, &vec![2; FRAME_BYTES - 105]); + assert_eq!(store.stats().reader_pins, 0); + let (_server, mut client) = fixture.serve(store); + + for n in 100..103 { + let denied = client.get(vec![get(&client, n, 1, too_large)]).unwrap(); + assert_eq!( + denied.items[0].body, + Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + }) + ); + } + assert!(matches!( + client + .lookup(vec![LookupItem { + header: header(&client, 103), + identity: synthetic_identity(1, 2), + }]) + .unwrap() + .items[0] + .body, + Ok(LookupOutcome::Hit { generation, .. }) if generation == exact + )); + // With a global ceiling of one, this success proves that no rejected Get + // retained a hidden pin. Its reply fills the frame exactly. + let hit = client.get(vec![get(&client, 104, 2, exact)]).unwrap().items[0] + .body + .clone() + .unwrap(); + assert_eq!(hit.payload, vec![2; FRAME_BYTES - 105]); + release(&mut client, 105, 104, exact, hit.region_id); + assert!( + client.get(vec![get(&client, 106, 2, exact)]).unwrap().items[0] + .body + .is_ok() + ); +} + +#[test] +fn multi_item_get_accounts_for_every_header_before_creating_pins() { + let fixture = Fixture::new(); + let mut store = fixture.open(2); + // Two successful Get items add 20 + 2 * 85 = 190 metadata bytes. + let exact = fixture.seed(&mut store, 1, &vec![1; (FRAME_BYTES - 190) / 2]); + let over = fixture.seed(&mut store, 2, &vec![2; (FRAME_BYTES - 190) / 2 + 1]); + let (_server, mut client) = fixture.serve(store); + + let reply = client + .get(vec![get(&client, 100, 2, over), get(&client, 101, 2, over)]) + .unwrap(); + let hit = reply.items[0].body.as_ref().unwrap(); + assert_eq!( + reply.items[1].body, + Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + }) + ); + release(&mut client, 102, 100, over, hit.region_id); + + // Both credits remain usable: the rejected second item did not pin. + // The complete two-item reply is exactly FRAME_BYTES, not just its payload. + let reply = client + .get(vec![ + get(&client, 103, 1, exact), + get(&client, 104, 1, exact), + ]) + .unwrap(); + for (index, owner) in [103, 104].into_iter().enumerate() { + let hit = reply.items[index].body.as_ref().unwrap(); + assert_eq!(hit.payload, vec![1; (FRAME_BYTES - 190) / 2]); + release( + &mut client, + 105 + index as u128, + owner, + exact, + hit.region_id, + ); + } +} diff --git a/docs/SECURITY.md b/docs/SECURITY.md index e337a91..58c5d40 100644 --- a/docs/SECURITY.md +++ b/docs/SECURITY.md @@ -1,4 +1,4 @@ -# Security (V1 C, V3 hooks) +# Security (V1 C, V3 CPU/TCP) The library calls still require an `AuthContext`. The TCP service does too. It binds `127.0.0.1` unless the operator sets `--allow-remote`, and it refuses to start with an empty tenant policy. A data request with an unknown tenant, a mismatched trust domain, or a wrong fingerprint is `Unauthorized` and returns no payload. Hello carries versions and the store epoch only. @@ -6,7 +6,7 @@ The fingerprint is derived from a credential, but the current protocol accepts t ## What is enforced -- Every store call requires an `AuthContext`: tenant, trust domain, and a 32-byte credential fingerprint. The fingerprint is compared in constant time. An all-zero fingerprint is rejected. +- Every store call requires an `AuthContext`: tenant, trust domain, and a 32-byte credential fingerprint. The comparison examines all 32 fingerprint bytes. An all-zero fingerprint is rejected. Every hot TCP data operation also passes a mandatory server ACL compiled from these store policies before dispatch. A successful Hello does not authorize a later data frame. - Empty policy configuration fails closed. Unknown tenants fail closed. - An identity whose tenant or trust domain does not match the caller is `Unauthorized`. The store does not return that block's bytes. - Physical keys include the tenant. There is no cross-tenant dedup and no list-all-keys API. @@ -14,27 +14,32 @@ The fingerprint is derived from a credential, but the current protocol accepts t - Only `COMMITTED`, checksum-valid, same-identity blocks are returned. - Quarantine plus fencing is the rule for buffers whose completion is unknown. A timer is not treated as proof that a copy has stopped. -## V3 hooks +## V3 admission and library hooks -These sit beside the V1 gate. They do not replace it and they do not add mTLS. See [adr/0007-trusted-fabric.md](adr/0007-trusted-fabric.md). +ACL and rate admission run on the TCP hot path alongside the store gate. The remaining hooks are explicitly in-process. None adds mTLS. See [adr/0007-trusted-fabric.md](adr/0007-trusted-fabric.md) and [adr/0010-hot-auth.md](adr/0010-hot-auth.md). -- `Acl` fails closed. A grant is read or write. Write is not implied by read. Unknown tenants and unknown fingerprints are `Unauthorized` and return no payload. -- `ReplayGuard` rejects a repeated nonce inside a bounded window and rejects a stamp outside the caller-supplied skew. The window forgets the oldest nonce when it is full. That is not permanent replay protection. +- `Acl` fails closed. `ServerConfig::acl` restricts the configured store policies; an empty ACL denies every data call. Lookup/Get and reader pin release require read access. ReservePut/Commit/Invalidate and writer cancellation require write access. A mixed pin-release/writer-cancel batch requires write before any mutation. Additional grants cannot override the store credential policy. The mandatory ACL has at most 1,024 principals. +- In-process `ReplayGuard` rejects repeated nonces inside a bounded window and stamps outside the configured skew. It rejects new admissions at capacity instead of evicting live nonces. The authorization helper scopes replay by tenant, trust domain and fingerprint; callers must use a server clock. Expired entries can be forgotten. This is not permanent replay protection and it is not wired into GCK1. - `seal` / `open_seal` mix a key id and generation into a SHA-256. A mismatch is `ChecksumMismatch` and returns no bytes. This is not an HMAC. Anyone who can write the bytes can recompute it. It is an integrity hook for a trusted fabric, not a claim of cryptographic authenticity against a hostile store. -- `SecretRing` rotates fingerprints. The previous fingerprint stays readable until `retire`. Writes use the current fingerprint only. Audit events store `Rotate`, `Retire`, `Deny`, and `Replay` plus a generation id. They do not store the fingerprint, the payload, or a prompt. -- `RateGate` is a per-tenant admit window. A noisy tenant does not consume another tenant's budget. Refusal is `QuotaExceeded { RateLimit }`. +- In-process `SecretRing` rotates fingerprints. The previous fingerprint stays readable until `retire`. Writes use the current fingerprint only. Audit events store `Rotate`, `Retire`, `Deny`, and `Replay` plus a generation id. They do not store the fingerprint, the payload, or a prompt. Rotation history is bounded at 1,024 generations; capacity exhaustion and attempts to revive retired credentials fail closed. +- TCP `RateGate` atomically charges every batch item against a per-tenant budget shared by all connections. The default is 16,384 items per 1,000 ms, configurable in ServerConfig. Rejected credentials cannot spend the claimed tenant's budget. Active windows are never evicted to bypass the limit; at most 1,024 tenant windows exist. Refusal is `QuotaExceeded { RateLimit }`. ServerHandle exposes denial/rate counters and the bounded 256-entry ACL-denial audit. +- Authorized reader pin releases with `Drained` attestation bypass the rate budget so a Get that consumes the final credit can still release its pin. Tenant, region, pin-owner and storage idempotency limits still apply. A mixed batch containing writer cancellation does not receive this exemption. - Control-plane loss on the simulated DPU queue hides slots and refuses new submits. It does not return a hit while membership is unknown. ## What is not enforced -- mTLS is not linked. The TCP service does not call `Acl`, `ReplayGuard`, `RateGate`, or `SecretRing` yet; those are library hooks a caller can use beside the V1 fingerprint. `--allow-remote` does not add them and does not terminate TLS. Treat a non-loopback bind as a trusted fabric, or do not bind it. +- mTLS is not linked. The wire fingerprint is a bearer-equivalent secret on plaintext TCP. `--allow-remote` does not add encryption or cryptographic peer authentication. Treat a non-loopback bind as a trusted fabric, or do not bind it. - Replay protection beyond the idempotency window and the bounded nonce window is not a permanent cache. - `Attestation::Drained` is accepted from any caller who already passed the tenant gate. It is not a device measurement. A buggy client inside the tenant can reclaim its own in-flight buffer early. -- The idempotency table forgets entries after `idempotency_ttl_ms` or when it is full (new ids are then rejected until expiry). Reusing an id after expiry can run the operation again. +- The idempotency table forgets entries after `idempotency_ttl_ms`. At capacity, new IDs are rejected until expiry rather than evicting a live entry. Reusing an ID after expiry can run the operation again. - This tree does not encrypt KV bytes at rest. `MemoryStore` keeps them in process memory. `FileStore` writes them as ordinary files in the directory you name. That directory must not be a production disk or a device node. The open path rejects `/dev`, `/proc`, `/sys`, `/`, and `..`. -- TCP has no mTLS. The V3 nonce window is a separate hook; the TCP server does not call it yet. Do not expose the listener past a trusted fabric. `--allow-remote` does not add authentication; the fingerprint check is still required. +- TCP does not call ReplayGuard or SecretRing: GCK1 carries no authenticated nonce, issue time or credential generation. The sender-controlled envelope time is not trusted as proof of freshness. ACL configuration is fixed at startup. Versioned authentication/freshness and live rotation remain deferred; the existing request-ID/deadline contract is unchanged. - `PrefixHit` bytes cover only `matched_prefix_blocks` of the query chain. A client that ignores that field and decodes them as the full query will mis-align. The server will not do that alignment for the client. +## Shutdown and restart + +Dropping ServerHandle closes active sockets and joins their workers before returning. Reopen the FileStore only after this completes. The restart advances the persisted epoch; stale connections cannot keep writing through the old store instance. This does not provide a distributed leader election or a fence against independent processes opening the same directory. + ## Fallback Callers must treat every error and every miss as "recompute in the engine". They must not stitch a partial or quarantined block into a decode. V1 C checks that a wrong tenant, a wrong model, a corrupt file, a short write, and a failed disk write return no foreign payload. The synthetic check is exact token-id equality. See [CORRECTNESS.md](CORRECTNESS.md). diff --git a/docs/adr/0007-trusted-fabric.md b/docs/adr/0007-trusted-fabric.md index b021422..ed4382d 100644 --- a/docs/adr/0007-trusted-fabric.md +++ b/docs/adr/0007-trusted-fabric.md @@ -17,21 +17,21 @@ The V3 boundary is a **trusted fabric**, documented here, plus CPU hooks. mTLS i Operators: - Leave the listener on `127.0.0.1` unless the host network is already restricted to cache peers. -- `--allow-remote` does not add authentication and does not start TLS. The V1 fingerprint gate still applies. The TCP service does not call the V3 ACL, replay guard, rate gate, or secret ring. +- `--allow-remote` does not add cryptographic authentication and does not start TLS. The fingerprint gate still applies. TCP now enforces the ACL and per-tenant rate gate described in [0010-hot-auth.md](0010-hot-auth.md). ReplayGuard and SecretRing remain in-process hooks. - Do not place this TCP service on an open network and call that mTLS. -In-process hooks (the ACL, nonce window, rate windows, and audit log have caps; secret-rotation history currently grows without a cap): +Bounded CPU security mechanisms (ACL and rate admission are enforced on TCP; replay, seals and rotation remain in-process hooks): -- `Acl` maps a fingerprint to read or write. An empty ACL fails closed. A read grant is not a write grant. The fingerprint compare scans every entry. -- `ReplayGuard` accepts a nonce once inside a caller-supplied skew. The window is bounded. Evicting the oldest nonce means that nonce can appear again later. This is not an infinite replay cache. +- `Acl` maps a fingerprint to read or write. An empty ACL fails closed. A read grant is not a write grant. The ACL scans every entry; each fingerprint comparison examines all 32 bytes. +- `ReplayGuard` accepts a nonce once inside a caller-supplied skew. The window is bounded and retains admitted nonces for the full skew window. Capacity exhaustion refuses new admissions rather than evicting a live nonce. After expiry the nonce can be admitted again; this is not an infinite replay cache. Authenticated helper calls scope the nonce by tenant, trust domain and fingerprint, and require server-clock time. - `seal` / `open_seal` bind a key id and a generation into a SHA-256 of a length-delimited buffer. A mismatch is `ChecksumMismatch` and returns no bytes. This is not an HMAC and not a transport MAC. On a trusted fabric it catches accidental corruption. It does not stop a caller who can recompute SHA-256. - `RateGate` limits admits per tenant per window. A noisy tenant spends only its own budget. -- `SecretRing::rotate` moves the previous fingerprint to overlap (read, not write) and audits the generation id. `retire` applies to overlap only. The audit log stores event kinds and generation ids, not fingerprints, payloads, or prompts. +- `SecretRing::rotate` moves the previous fingerprint to overlap (read, not write) and audits the generation id. `retire` applies to overlap only. The audit log stores event kinds and generation ids, not fingerprints, payloads, or prompts. Rotation history is capped at 1,024 generations and fails closed at capacity; a retired fingerprint cannot be revived by rotation. Shard membership, epoch fencing, rebalance, rolling upgrade, and fault-domain isolation extend the V2 directory. They fence in-flight fills. They do not copy payloads and they are not replicas. See [0005-shard-fencing.md](0005-shard-fencing.md). ## Consequences -- `cargo test --workspace --all-targets` covers these hooks without certificates. +- `cargo test --workspace --all-targets` covers these hooks and the TCP ACL/rate gates without certificates. ADR 0010 supersedes the earlier statement that all V3 mechanisms were unused by TCP. - A later milestone may add mTLS behind a feature flag. Until that lands, security docs must keep saying the fabric is the boundary. - Wrong-tenant and replay failures are not hits. Callers recompute. diff --git a/docs/adr/0010-hot-auth.md b/docs/adr/0010-hot-auth.md new file mode 100644 index 0000000..3ec9dbe --- /dev/null +++ b/docs/adr/0010-hot-auth.md @@ -0,0 +1,112 @@ +# ADR 0010: Enforce ACL and bounded admission on the hot TCP path + +- Status: **Accepted** +- Date: 2026-10-20 +- Milestone: V3 CPU/TCP + +## Context + +The FileStore already checked tenant, trust domain, fingerprint and the identity +of each requested block. The V3 ACL and rate gate existed only as library hooks. +A TCP deployment could not restrict a configured tenant to read access, and +independent client connections did not share an admission window. + +GCK1 has no authenticated nonce, secret generation, or issued-time field. +The envelope timestamp is supplied by an untrusted sender. Treating it as proof +of freshness would make replay protection misleading and would break legitimate +idempotent retries. The wire and CacheIdentity semantics must remain unchanged. + +## Decision + +Every decoded TCP data request passes a server-owned ACL before any store call. +The mandatory ACL is compiled from the FileStore policies, with a maximum of +1,024 entries. `ServerConfig::acl` optionally restricts these grants. `None` +preserves read/write access for the configured store principals; an explicit +empty ACL denies all data calls. Additional ACL entries cannot authorize a +credential absent from the store policies. + +Lookup and Get require read access. ReservePut, Commit and Invalidate require +write access. ReleaseCancel requires write if **any** item cancels a writer +(`pin_owner` absent). A batch of reader pin releases requires read, so read-only +clients can drain their own references. The storage tenant and pin-owner checks +still apply. Denying a mixed batch happens before releasing any pin. + +After authorization, each batch atomically charges its item count against a +server-owned per-tenant RateGate. Connections share this budget. Defaults are +16,384 items per 1,000 ms, configurable through ServerConfig. A failed batch +admission does not partially consume the budget. Unauthorized credentials cannot +spend the claimed tenant's budget. Tables are limited to 1,024 tenants; active +windows are never evicted to admit another tenant. Expired windows release table +capacity. An oversized first batch does not occupy a tenant window. Authorized +reader pin releases with `Drained` attestation bypass this admission budget, so a +Get that uses the final credit can still release its reference. Storage still +checks the tenant, region and pin owner and bounds its idempotency table. +Time comes from the server's monotonic clock. + +ACL denials increment a saturating counter and a 256-entry audit ring. +Rate-limit refusals increment a separate counter. ServerHandle exposes snapshots; +they contain no credentials, payloads or identities. These are admission counters, +not a claim to count every per-item storage error. + +Server shutdown closes the bounded set of active TCP sockets and joins every +worker before returning. An old worker cannot continue modifying the FileStore +after its directory is reopened by the supported stop-then-start sequence. + +## Reader and reply admission + +`MemoryStore` now applies `max_in_flight_requests` as an independent global +reader-pin ceiling as well as the existing writer ceiling. Expired or unknown +completion pins remain charged. Each Get batch may copy at most +`max_object_bytes` total payload bytes, checked before allocation/pin admission. +Later over-budget items receive existing quota errors, preserving item-scoped +batch semantics. `StoreStats::reader_pins` exposes the charged count. No new +wire tag, identity field or configuration schema is introduced. Direct fenced +reclaim authorizes before running timer maintenance. + +TCP additionally passes its `max_frame_bytes` reply-body ceiling to the store +before Get admission. The protocol-owned budget reserves the exact 20-byte GCKR +batch header and 85 metadata bytes per possible successful item, then allows +payloads only within the remaining bytes and the codec's absolute field limit. +Every item error is smaller than a successful item's metadata reservation; +failed items conservatively keep that reservation for the rest of the batch. +If metadata alone does not fit, the batch fails with the existing quota error, +whose 21-byte failed reply fits even the minimum server frame limit. Otherwise +over-budget items fail without copying payloads or creating pins. Successfully +admitted items and their pin handles always fit the configured frame before +encoding begins. The outer TCP envelope is separate from this body limit. +Network failures after admission still retain unknown-completion pins under +the existing drain/fence contract. + +## Wire compatibility and deferred authentication + +GCK1 requests, GCKR replies and CacheIdentity are unchanged. Hello remains an +unauthenticated version/epoch exchange with no data access. A data frame still +requires its own complete AuthContext; a successful Hello does not authenticate +the connection. + +The bearer-equivalent fingerprint travels on plaintext TCP. This is authorization +on a trusted fabric, **not mTLS** or cryptographic peer authentication. The ACL +is fixed at server start. The SecretRing and ReplayGuard remain in-process hooks; +neither is advertised as enforced by TCP. Request IDs and deadlines retain the +existing bounded idempotency contract, including replay after expiry. Adding +authenticated freshness or live credential rotation needs a separate versioned +protocol/authentication design. SHA-256 integrity seals are not transport MACs. + +## Evidence and limits + +`crates/transport/tests/hot_security.rs` exercises all six operations on actual +loopback TCP: empty-ACL denial, read-only denial and pin drain, conflicting ACL +grants versus store policies, cross-tenant generations, cross-model/layout +reads, batch rate limits across connections, a bounded denial audit, and restart +with an old connection still open. Unit tests cover the bounded rate table and +credential-safe ACL debugging. + +`crates/transport/tests/reply_budget.rs` uses 2 KiB frames and one/two-pin ceilings +to cover oversized Gets, exact-size single and multi-item replies, mixed success +and quota errors, pin drain, and subsequent requests on the same TCP connection. +Storage tests assert zero pins after metadata/payload rejection. Codec tests +check the budget against actual GCKR encodings and the absolute field cap. + +These tests use synthetic CPU payloads. They do not qualify encryption, hardware +DMA drain, a distributed control-plane authority or any GPU/DPU transport. +Hardware paths remain `NOT_RUN_HARDWARE`.