Enforce hot TCP authorization and bounded reader admission

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::<u8>(), 1..1024),
+        position in any::<usize>(),
+        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());
+    }
 }

crates/protocol/src/harden.rs

@@ -63,18 +63,29 @@ impl SecurityLog {
     }
 }
 
+#[derive(Clone)]
 struct AclEntry {
     tenant: TenantId,
     trust: TrustDomain,
     fingerprint: [u8; 32],
     write: bool,
 }
 
+#[derive(Clone)]
 pub struct Acl {
     entries: Vec<AclEntry>,
     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<Self, CacheError> {
         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(),

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;

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<u64, CacheError> {
+    // 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");

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<TenantPolicy> {
+        self.inner.tenant_policies()
+    }
+
     pub fn limits(&self) -> ResourceLimits {
         self.inner.limits()
     }
@@ -223,12 +229,30 @@ impl FileStore {
         items: Vec<GetItem>,
     ) -> Result<BatchReply<GetHit>, 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<GetItem>,
+        max_reply_bytes: usize,
+    ) -> Result<BatchReply<GetHit>, 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<GetHit>) -> BatchReply<GetHit> {
         for item in &reply.items {
             if let Ok(hit) = &item.body {
                 self.touch(hit.generation);
             }
         }
-        Ok(reply)
+        reply
     }
 
     pub fn release_cancel(