feat(cubesys): R5 — read-gating in daemon path (owner/grant-native)

Closes the read-privacy gap flagged in plan R1/R5. Rather than fork the
FUSE uid/gid Acl model (whose identity is POSIX uid, incompatible with the
daemon's name-based HELLO identity), read-gating is done owner/grant-native:

- commands.rs: new admit_read(coord) mirroring admit_mutate (owner ->
  read-grant -> deny; unowned records world-readable). Wired into
  (read+execute) and  (metadata read), gated by enforce_owner exactly
  like writes.
- Tests: read_gate_blocks_non_owner_and_allows_read_grant (bob denied,
  allowed after alice grants read), read_gate_requires_identity_under_enforce
  (anonymous stat rejected under --require-identity).
- Full ./check quick green: 39 cubesys lib tests, clippy -D warnings clean.

Note in plan: Acl lift (R1) is NOT a literal fork; the daemon reuses the
owner/grant enforcement concept, not the POSIX Acl struct.

cubesys/src/commands.rs

@@ -196,6 +196,60 @@ impl Session {
         }
     }
 
+    /// Authorization gate for a read op (R5 — closes the read-privacy gap in
+    /// the daemon path). A read command (`run`/`stat`, and the metadata read in
+    /// `ls`) may proceed only when EITHER the session's identity owner matches
+    /// the record's `owner_local_user` (owner), OR the caller holds a grant
+    /// authorizing `Perm::Read` over the coordinate. Returns `None` to allow, or
+    /// an error message to reject.
+    ///
+    /// Mirrors [`admit_mutate`] (Task 6 + 6b): same anonymous policy, same
+    /// owner→grant→deny ordering. An unowned record is world-readable (first
+    /// writer claims ownership on write; reads never create state).
+    ///
+    /// Note: the FUSE mount enforces POSIX `mode` bits via
+    /// `cubefs::nullspace::Acl`; the daemon deliberately does NOT import that
+    /// uid/gid model — its identity is name-based (`Owner`) from `HELLO`, so
+    /// read gating here is owner/grant-native rather than a literal `Acl` lift.
+    /// (See plan R1/R5 for the rationale.)
+    fn admit_read(&self, coord: Czyx) -> Option<String> {
+        match &self.identity {
+            None => {
+                if self.enforce_owner {
+                    Some(
+                        "owner enforcement: read operations require a HELLO identity \
+                         (daemon policy --require-identity)"
+                            .to_string(),
+                    )
+                } else {
+                    None
+                }
+            }
+            Some(id) => {
+                // (1) owner match — or unowned record, world-readable.
+                if let Some(record_owner) = self.store.owner(&coord) {
+                    if record_owner == id.owner_local {
+                        return None;
+                    }
+                } else {
+                    return None; // unowned -> readable by anyone
+                }
+                // (2) read grant.
+                let who = Owner::from(id);
+                if grant_allows(&self.store, &who, &coord, Perm::Read) {
+                    return None;
+                }
+                // (3) deny.
+                Some(format!(
+                    "owner violation: record at {} owned by '{}', you are '{}' (and hold no matching read grant)",
+                    coord.pack_u32(),
+                    self.store.owner(&coord).unwrap_or_default(),
+                    id.owner_local
+                ))
+            }
+        }
+    }
+
     /// Shared handle to the underlying concurrent store.
     pub fn store(&self) -> Arc<ConcurrentStore> {
         self.store.clone()
@@ -534,6 +588,11 @@ impl Session {
             "run" => {
                 let path = it.next().ok_or_else(|| "run needs <path>".to_string())?;
                 let _coord = crate::path_to_czyx(path).map_err(|e| e.to_string())?;
+                // R5: reads (this one also *executes*) honor the owner/grantee
+                // read gate, same as writes honor the mutate gate.
+                if let Some(msg) = self.admit_read(_coord) {
+                    return Err(msg);
+                }
                 let sn = txn_snapshot(self);
                 let cell = crate::load_code_cell(&sn, path).map_err(|e| e.to_string())?;
                 let mut vm = Vm::new(sn);
@@ -561,6 +620,11 @@ impl Session {
             }
             "stat" => {
                 let path = it.next().ok_or_else(|| "stat needs <path>".to_string())?;
+                // R5: metadata reads honor the read gate.
+                let coord = crate::path_to_czyx(path).map_err(|e| e.to_string())?;
+                if let Some(msg) = self.admit_read(coord) {
+                    return Err(msg);
+                }
                 let sn = txn_snapshot(self);
                 let fs = cubefs::CubeFs::new(sn);
                 let a = fs
@@ -1077,4 +1141,86 @@ mod tests {
         );
         let _ = std::fs::remove_dir_all(&dir);
     }
+
+    #[test]
+    fn read_gate_blocks_non_owner_and_allows_read_grant() {
+        // R5: `run` (a read+execute op) honors the same owner/grantee gate as
+        // writes. alice owns a cell; bob cannot read it under enforce_owner;
+        // once alice grants bob read, bob's read succeeds.
+        use crate::grants::grant;
+        use crate::tenant::{TenantConfig, TenantId, TenantIdentity, TenantRegistry};
+        use std::str::FromStr;
+
+        let registry = TenantRegistry::with_config(TenantConfig::Memory);
+        let ts = registry
+            .get_or_provision(TenantId::from_str("alpha").unwrap())
+            .unwrap();
+        ts.set_identity(TenantIdentity {
+            tenant: TenantId::from_str("alpha").unwrap(),
+            owner_local: "alice".to_string(),
+            owner_remote: None,
+        });
+        let mut a = Session::for_tenant(&ts);
+        a.set_enforce_owner(true);
+        // Alice writes a runnable cell she owns.
+        assert!(a.exec("prog /c070/z001/y001/x001 const 7 halt").is_ok());
+
+        // Bob is a different owner on the same (shared) store.
+        ts.set_identity(TenantIdentity {
+            tenant: TenantId::from_str("alpha").unwrap(),
+            owner_local: "bob".to_string(),
+            owner_remote: None,
+        });
+        let mut b = Session::for_tenant(&ts);
+        b.set_enforce_owner(true);
+        let r = b.exec("run /c070/z001/y001/x001");
+        assert!(
+            r.is_err(),
+            "non-owner read must be gated under --require-identity"
+        );
+        assert!(r.unwrap_err().contains("owner violation"));
+
+        // Alice grants bob READ on the coord.
+        ts.set_identity(TenantIdentity {
+            tenant: TenantId::from_str("alpha").unwrap(),
+            owner_local: "alice".to_string(),
+            owner_remote: None,
+        });
+        let a2 = Session::for_tenant(&ts);
+        let _ = grant(
+            &a2.store,
+            &crate::grants::Owner::from(&a2.identity().unwrap()),
+            &crate::grants::Owner {
+                local: "bob".to_string(),
+                remote: None,
+            },
+            crate::grants::PERM_READ,
+            Some(Czyx::new(70, 1, 1, 1)),
+        );
+
+        // Now bob's read passes via grant.
+        let r = b.exec("run /c070/z001/y001/x001");
+        assert!(r.is_ok(), "read grant must allow bob's read: {r:?}");
+    }
+
+    #[test]
+    fn read_gate_requires_identity_under_enforce() {
+        // R5: a `stat` (pure metadata read) from an anonymous session is
+        // rejected under --require-identity, exactly like a write.
+        let dir = std::env::temp_dir().join(format!("cube2-rgate-{}", std::process::id()));
+        let _ = std::fs::create_dir_all(&dir);
+        let cs = ConcurrentStore::open(
+            dir.join("db.cubedb").to_str().unwrap(),
+            dir.join("wal.ndjson").to_str().unwrap(),
+            dir.join("recovery.jsonl").to_str().unwrap(),
+            DurabilityConfig::default(),
+        )
+        .unwrap();
+        let mut s = Session::with_store(Arc::new(cs));
+        s.set_enforce_owner(true);
+        let r = s.exec("stat /c071/z001/y001/x001");
+        assert!(r.is_err(), "anonymous stat must be gated");
+        assert!(r.unwrap_err().contains("require a HELLO identity"));
+        let _ = std::fs::remove_dir_all(&dir);
+    }
 }