From abc1b56d241f259326e06a46ca7daede9e6703a2 Mon Sep 17 00:00:00 2001 From: CUBELinux-2 Date: Tue, 11 Aug 2026 13:41:24 -0400 Subject: [PATCH] =?UTF-8?q?feat(cubesys):=20Task=206=20=E2=80=94=20owner?= =?UTF-8?q?=20enforcement=20on=20mutating=20commands?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stamp owner_local_user on records written via prog/write and gate the mutating paths (prog, write, del — both live and buffered txn) so a session may only create or overwrite a record whose owner_local_user matches its HELLO-declared identity. Design (logical + expedient for the whole project): - Owner is the durable record-level CubeHeader.owner_local_user field, so enforcement is replay-safe and works across daemon restart. - Enforcement is opt-in/non-breaking: gated only when the session has a stamped identity AND the record has an owner. First write by an owner claims an unowned coord; a session with no identity (tests, legacy) writes freely. - COMMIT re-checks owner on each buffered op before applying, so a concurrent cross-owner commit between BEGIN and COMMIT is rejected (txn is restored for retry, not silently dropped). - seal/open (encrypted raw put/del) left ungated for now: their headers are not owner-stamped yet — tracked as follow-up. Verification: - ./check quick: EXIT=0, fmt+clippy clean, 26 cubesys tests (added owner_enforcement_blocks_cross_owner_overwrite, owner_enforcement_allows_first_claim_and_same_owner, for_tenant_carries_identity). - Ad-hoc daemon verifier over real cube-server socket (LE framing): cross-owner overwrite + delete rejected, same-owner + first-claim allowed, no-HELLO legacy writes allowed. ALL PASS. --- cubesys/src/commands.rs | 174 ++++++++++++++++++++++++++++++++++++++-- cubesys/src/lib.rs | 6 +- cubesys/src/store.rs | 17 +++- 3 files changed, 186 insertions(+), 11 deletions(-) diff --git a/cubesys/src/commands.rs b/cubesys/src/commands.rs index 68da0e8..3325248 100644 --- a/cubesys/src/commands.rs +++ b/cubesys/src/commands.rs @@ -207,6 +207,19 @@ impl Session { .take() .ok_or_else(|| "commit: no transaction is open".to_string())?; let n = txn.ops.len(); + // Task 6: owner enforcement for buffered txn ops. The live path + // already checked at `prog`/`write`/`del` time, but a concurrent + // commit from another owner could have claimed the coord between + // our BEGIN and COMMIT, so re-check now (the buffer is re-read + // here, consistent with the single WAL entry T5 writes). + for op in &txn.ops { + if let Some(msg) = owner_violation(&self.identity, &store.owner(&op.coord)) { + // Roll the txn back: restore it so the caller can retry + // after resolving the conflict (do not silently drop). + self.txn = Some(txn); + return Err(msg); + } + } // Apply every buffered op atomically under one store write lock // and append a SINGLE WAL entry (WalOp::Txn) so the whole batch // is durable as one unit and replays idempotently. @@ -266,13 +279,18 @@ impl Session { // a throwaway store so we can buffer (or apply) them without // duplicating the record codec. let coord = scratch_code_coord(path, Kind::Fn, name, &ops)?; + // Task 6: reject overwriting a record owned by a different owner. + if let Some(msg) = owner_violation(&self.identity, &store.owner(&coord)) { + return Err(msg); + } + let owner = self.identity.as_ref().map(|i| i.owner_local.as_str()); let value = { let mut scratch = CubeStore::new(HashBackend::new()); - crate::store_code_cell(&mut scratch, path, Kind::Fn, name, &[], &ops) + crate::store_code_cell(&mut scratch, path, Kind::Fn, name, &[], &ops, owner) .map_err(|e| e.to_string())?; scratch.get_raw(&coord).unwrap_or_default() }; - let header = header_for_code(Kind::Fn, name, &ops); + let header = header_for_code(Kind::Fn, name, &ops, owner); if let Some(txn) = self.txn.as_mut() { txn.ops.push(TxnOp { coord, @@ -284,7 +302,7 @@ impl Session { )); } let coord = store - .put_code_cell(path, Kind::Fn, name, &[], &ops) + .put_code_cell(path, Kind::Fn, name, &[], &ops, owner) .map_err(|e| e.to_string())?; Ok(format!( "wrote program {path} -> coord {} ({} ops)", @@ -302,13 +320,17 @@ impl Session { .map_err(|e| format!("bytecode decode error: {e:?}"))?; let name = path.rsplit('/').next().unwrap_or(path); let coord = scratch_code_coord(path, Kind::Fn, name, &code)?; + if let Some(msg) = owner_violation(&self.identity, &store.owner(&coord)) { + return Err(msg); + } + let owner = self.identity.as_ref().map(|i| i.owner_local.as_str()); let value = { let mut scratch = CubeStore::new(HashBackend::new()); - crate::store_code_cell(&mut scratch, path, Kind::Fn, name, &[], &code) + crate::store_code_cell(&mut scratch, path, Kind::Fn, name, &[], &code, owner) .map_err(|e| e.to_string())?; scratch.get_raw(&coord).unwrap_or_default() }; - let header = header_for_code(Kind::Fn, name, &code); + let header = header_for_code(Kind::Fn, name, &code, owner); if let Some(txn) = self.txn.as_mut() { txn.ops.push(TxnOp { coord, @@ -320,13 +342,18 @@ impl Session { )); } let coord = store - .put_code_cell(path, Kind::Fn, name, &[], &code) + .put_code_cell(path, Kind::Fn, name, &[], &code, owner) .map_err(|e| e.to_string())?; Ok(format!("wrote {path} -> coord {}", coord.pack_u32())) } "del" => { let path = it.next().ok_or_else(|| "del needs ".to_string())?; let coord = crate::path_to_czyx(path).map_err(|e| e.to_string())?; + // Task 6: reject deleting a record owned by a different owner + // (unless unowned, which any identity may take over). + if let Some(msg) = owner_violation(&self.identity, &store.owner(&coord)) { + return Err(msg); + } if let Some(txn) = self.txn.as_mut() { txn.ops.push(TxnOp { coord, put: None }); return Ok(format!("buffered del {path} — commit to apply")); @@ -455,16 +482,48 @@ pub fn txn_snapshot(s: &Session) -> CubeStore { /// writing — used to buffer `prog`/`write` mutations during a transaction. fn scratch_code_coord(path: &str, kind: Kind, name: &str, code: &[Op]) -> Result { let mut scratch = CubeStore::new(HashBackend::new()); - crate::store_code_cell(&mut scratch, path, kind, name, &[], code).map_err(|e| e.to_string()) + crate::store_code_cell(&mut scratch, path, kind, name, &[], code, None).map_err(|e| e.to_string()) +} + +/// Owner-enforcement check (Task 6): a mutating command may only create or +/// overwrite a record when EITHER the record is unowned OR the session's +/// identity owner matches the record's `owner_local_user`. Returns `None` when +/// the write is allowed, or an `Err`-style message when it must be rejected. +/// +/// Enforcement is opt-in / non-breaking: if the session has no stamped +/// identity (e.g. a `Session::with_store` used by tests, or a client that +/// never sent `HELLO`), every write is permitted. The daemon stamps identity +/// via `HELLO` (Task 3) before any mutating op, so live traffic is gated. +fn owner_violation(identity: &Option, record_owner: &Option) -> Option { + let owner = identity.as_ref().map(|i| i.owner_local.as_str()); + match (owner, record_owner) { + // No session identity -> no gate (legacy / test path). + (None, _) => None, + // Session has an identity but the record is unowned -> allow (first write + // by this owner claims it). + (Some(_), None) => None, + // Both set: must match exactly. + (Some(session_owner), Some(record_owner)) => { + if session_owner == record_owner.as_str() { + None + } else { + Some(format!( + "owner violation: record owned by '{record_owner}', you are '{session_owner}'" + )) + } + } + } } /// Build the `CubeHeader` a `store_code_cell` call would attach (mirrors /// `crate::store_code_cell`), so a buffered txn put carries the same header. -fn header_for_code(kind: Kind, name: &str, code: &[Op]) -> CubeHeader { +/// `owner` (when set) is stamped on `owner_local_user` for Task 6 enforcement. +fn header_for_code(kind: Kind, name: &str, code: &[Op], owner: Option<&str>) -> CubeHeader { let mut h = CubeHeader::new(); h.title = Some(name.to_string()); h.doc_type = Some(kind.as_str().to_string()); h.linked_records = Vec::new(); + h.owner_local_user = owner.map(|o| o.to_string()); if h.doc_type.as_deref() == Some("fn") { h.size_bytes = Some(cubecode::encode(code).len() as u64); } @@ -559,6 +618,7 @@ pub fn parse_transform(s: &str) -> Option { mod tests { use super::*; use crate::store::DurabilityConfig; + use std::str::FromStr; use std::sync::Arc; fn session() -> Session { @@ -591,6 +651,104 @@ mod tests { assert!(s.exec("commit").is_ok()); } + #[test] + fn owner_enforcement_blocks_cross_owner_overwrite() { + use crate::tenant::{TenantIdentity, TenantId}; + let mut s = session(); + // Stamp an identity (Task 3 path) for owner "alice". + s.set_identity(TenantIdentity { + tenant: TenantId::from_str("alpha").unwrap(), + owner_local: "alice".to_string(), + owner_remote: None, + }); + // Alice writes a record — it should be stamped + succeed. + assert!(s.exec("prog /c050/z001/y001/x001 const 7 halt").is_ok()); + assert_eq!( + s.store.owner(&Czyx::new(50, 1, 1, 1)).as_deref(), + Some("alice") + ); + // Now "bob" tries to overwrite alice's coord — must be rejected. + s.set_identity(TenantIdentity { + tenant: TenantId::from_str("alpha").unwrap(), + owner_local: "bob".to_string(), + owner_remote: None, + }); + let r = s.exec("prog /c050/z001/y001/x001 const 8 halt"); + assert!(r.is_err(), "cross-owner overwrite must be rejected"); + assert!(r.unwrap_err().contains("owner violation")); + // The record must still be alice's (unchanged value). + assert_eq!( + s.store.owner(&Czyx::new(50, 1, 1, 1)).as_deref(), + Some("alice") + ); + // And a delete by bob is likewise rejected. + let d = s.exec("del /c050/z001/y001/x001"); + assert!(d.is_err()); + assert!(d.unwrap_err().contains("owner violation")); + } + + #[test] + fn owner_enforcement_allows_first_claim_and_same_owner() { + use crate::tenant::{TenantIdentity, TenantId}; + let mut s = session(); + // A session with NO identity writes freely (legacy / test path). + assert!(s.exec("prog /c051/z001/y001/x001 const 1 halt").is_ok()); + assert_eq!(s.store.owner(&Czyx::new(51, 1, 1, 1)), None); + // Alice claims the previously-unowned coord — allowed (first write). + s.set_identity(TenantIdentity { + tenant: TenantId::from_str("alpha").unwrap(), + owner_local: "alice".to_string(), + owner_remote: None, + }); + assert!(s.exec("prog /c051/z001/y001/x001 const 2 halt").is_ok()); + assert_eq!( + s.store.owner(&Czyx::new(51, 1, 1, 1)).as_deref(), + Some("alice") + ); + // Same owner overwrites freely. + assert!(s.exec("prog /c051/z001/y001/x001 const 3 halt").is_ok()); + } + + #[test] + fn for_tenant_carries_identity_to_session() { + // Mirrors the cube-server daemon flow: a connection resolves its tenant + // session, stamps identity via HELLO, then runs a per-connection + // `Session::for_tenant` for the rest of the connection. The identity + // MUST propagate so owner enforcement fires on the live path. + 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 s = Session::for_tenant(&ts); + // The per-connection session must carry the stamped identity. + assert_eq!(s.identity().unwrap().owner_local, "alice"); + // Alice writes her coord (stamped). + assert!(s.exec("prog /c052/z001/y001/x001 const 7 halt").is_ok()); + assert_eq!( + s.store.owner(&Czyx::new(52, 1, 1, 1)).as_deref(), + Some("alice") + ); + // A second connection as bob shares the same store (legacy mode) but + // must be blocked from overwriting alice's record. + 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); + let r = b.exec("prog /c052/z001/y001/x001 const 8 halt"); + assert!(r.is_err(), "cross-owner overwrite must be rejected live"); + assert!(r.unwrap_err().contains("owner violation")); + } + #[test] fn txn_isolation_begin_snapshot_hides_live_writer() { // Stream A opens a txn and snapshots; stream B mutates the live store. diff --git a/cubesys/src/lib.rs b/cubesys/src/lib.rs index 7281af1..f9e4998 100644 --- a/cubesys/src/lib.rs +++ b/cubesys/src/lib.rs @@ -132,12 +132,14 @@ pub fn store_code_cell( name: &str, links: &[Czyx], code: &[cubecode::Op], + owner: Option<&str>, ) -> Result { let coord = path_to_czyx(path)?; let mut h = CubeHeader::new(); h.title = Some(name.to_string()); h.doc_type = Some(kind.as_str().to_string()); h.linked_records = links.to_vec(); + h.owner_local_user = owner.map(|o| o.to_string()); if h.doc_type.as_deref() == Some("fn") { h.size_bytes = Some(cubecode::encode(code).len() as u64); } @@ -176,6 +178,7 @@ mod tests { "add", &[], &[Op::Const(2), Op::Const(3), Op::Add, Op::Halt], + None, ) .unwrap(); @@ -282,7 +285,7 @@ pub mod demo { let entry_code = vec![Op::Const(21), Op::CallLink(0), Op::Halt]; let double_coord = - super::store_code_cell(&mut store, double, Kind::Fn, "double", &[], &double_code) + super::store_code_cell(&mut store, double, Kind::Fn, "double", &[], &double_code, None) .expect("store double"); let entry_coord = super::store_code_cell( &mut store, @@ -291,6 +294,7 @@ pub mod demo { "entry", &[double_coord], &entry_code, + None, ) .expect("store entry"); println!(" wrote {double} -> coord {}", double_coord.pack_u32()); diff --git a/cubesys/src/store.rs b/cubesys/src/store.rs index 0479811..0bc853b 100644 --- a/cubesys/src/store.rs +++ b/cubesys/src/store.rs @@ -491,6 +491,17 @@ impl ConcurrentStore { out } + /// The `owner_local_user` stamped on the record at `key`, if it has one. + /// Used by owner enforcement (Task 6): a mutating command may only + /// overwrite a record whose owner matches the session's identity owner. + pub fn owner(&self, key: &Czyx) -> Option { + self.inner + .read() + .unwrap() + .get_record(key) + .and_then(|(h, _)| h.owner_local_user.clone()) + } + /// A consistent point-in-time snapshot of the whole store. Used by the VM /// and cubefs, which take a `CubeStore` by value. A read-lock clone, so it /// does not block concurrent readers (it only waits for an in-flight @@ -580,6 +591,8 @@ impl ConcurrentStore { } /// Store a code cell at `path` (the path->code bridge), durability-logged. + /// `owner` (when set) is stamped on the record's `owner_local_user` field + /// so owner enforcement (Task 6) can later reject cross-owner overwrites. pub fn put_code_cell( &self, path: &str, @@ -587,9 +600,9 @@ impl ConcurrentStore { name: &str, links: &[Czyx], code: &[Op], + owner: Option<&str>, ) -> Result { - let coord = - self.with_mut(|store| crate::store_code_cell(store, path, kind, name, links, code))?; + let coord = self.with_mut(|store| crate::store_code_cell(store, path, kind, name, links, code, owner))?; if let Some(v) = self.get_raw(&coord) { self.log_put(coord, v); }