From 295bbeadaccde596426d031e17d4ac80243ee98f Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Wed, 29 Jul 2026 14:57:56 +0200 Subject: [PATCH] implemented share_access, POSIX read-only mode --- Cargo.lock | 8 +- Cargo.toml | 4 +- crates/pdfs-core/src/access.rs | 108 ++++ crates/pdfs-core/src/cache.rs | 290 ++++++++- crates/pdfs-core/src/db/maintenance.rs | 1 + crates/pdfs-core/src/db/migrations.rs | 16 +- crates/pdfs-core/src/db/mod.rs | 3 +- crates/pdfs-core/src/db/nodes.rs | 26 + crates/pdfs-core/src/db/ops.rs | 100 ++++ crates/pdfs-core/src/db/share_access.rs | 143 +++++ crates/pdfs-core/src/db/tests.rs | 311 ++++++++++ crates/pdfs-core/src/lib.rs | 2 + crates/pdfs-fuse/src/background.rs | 194 +++++- crates/pdfs-fuse/src/devices.rs | 3 + crates/pdfs-fuse/src/drain.rs | 278 ++++++++- crates/pdfs-fuse/src/filesystem.rs | 232 +++++-- crates/pdfs-fuse/src/lib.rs | 763 ++++++++++++++++++++++-- crates/pdfs-fuse/src/mount.rs | 5 + crates/pdfs-fuse/src/state.rs | 692 ++++++++++++++++++++- crates/pdfs-fuse/src/sweep.rs | 6 +- crates/pdfs-fuse/src/upload.rs | 4 + docs/BUGS.md | 12 +- 22 files changed, 3055 insertions(+), 146 deletions(-) create mode 100644 crates/pdfs-core/src/access.rs create mode 100644 crates/pdfs-core/src/db/share_access.rs diff --git a/Cargo.lock b/Cargo.lock index 3e320a8..35d2574 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3830,9 +3830,9 @@ dependencies = [ [[package]] name = "proton-drive-rs" -version = "0.2.2" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "692610fbead83735a0415d9d068c1e9b467b12cce013ea19b3063f05206ce3d6" +checksum = "a02afd5a53ecbf0e1ea18f3102e7b99e5ad61977ee0709bd214279b2376e0077" dependencies = [ "async-trait", "base64", @@ -3857,9 +3857,9 @@ dependencies = [ [[package]] name = "proton-sdk" -version = "0.2.2" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6b3f2ad9cd1ba69b67cbafcaf7c159b3a200162a3c518b1bc7e25df837343b41" +checksum = "46b5f84f0c33ac7a19b0131e971e7a17f10f12d6dc6df64c3b9dda7f37aa35bb" dependencies = [ "aes-gcm 0.11.0", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index da299db..71d47e9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,8 +10,8 @@ authors = ["Nils Pukropp "] rust-version = "1.96" [workspace.dependencies] -proton-sdk = { version = "0.2.2" } -proton-drive-rs = { version = "0.2.2" } +proton-sdk = { version = "0.3.1" } +proton-drive-rs = { version = "0.3.1" } # Internal crates. pdfs-core = { path = "crates/pdfs-core" } diff --git a/crates/pdfs-core/src/access.rs b/crates/pdfs-core/src/access.rs new file mode 100644 index 0000000..f3061a6 --- /dev/null +++ b/crates/pdfs-core/src/access.rs @@ -0,0 +1,108 @@ +//! Effective access carried by mounted Drive nodes. + +use proton_drive_rs::MemberRole; + +/// Effective write access for a node in a mounted tree. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Access { + Owner, + Editor, + Viewer, + Unknown, +} + +impl Access { + pub fn writable(self) -> bool { + matches!(self, Self::Owner | Self::Editor) + } + + pub(crate) fn as_db_str(self) -> &'static str { + match self { + Self::Owner => "owner", + Self::Editor => "editor", + Self::Viewer => "viewer", + Self::Unknown => "unknown", + } + } + + pub(crate) fn from_db_str(value: &str) -> Option { + match value { + "owner" => Some(Self::Owner), + "editor" => Some(Self::Editor), + "viewer" => Some(Self::Viewer), + "unknown" => Some(Self::Unknown), + _ => None, + } + } +} + +/// Resolve a share root's effective access from its exact membership role. +/// +/// Owned trees fail open. A share whose role is absent or unrecognised fails +/// closed. Descendants normally inherit directly at intern time; `Inherited` +/// exists for the explicit wire role and therefore returns `parent`. +pub fn access_for(role: Option, parent: Access, under_share: bool) -> Access { + match role { + Some(MemberRole::Viewer) => Access::Viewer, + Some(MemberRole::Editor | MemberRole::Admin) => Access::Editor, + Some(MemberRole::Inherited) => parent, + None if under_share => Access::Viewer, + None => Access::Owner, + } +} + +/// POSIX permission bits advertised for a mounted node. +pub fn perm_bits(is_dir: bool, access: Access) -> u16 { + match (is_dir, access.writable()) { + (true, true) => 0o755, + (false, true) => 0o644, + (true, false) => 0o555, + (false, false) => 0o444, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn access_for_covers_the_fail_open_and_fail_closed_table() { + for parent in [ + Access::Owner, + Access::Editor, + Access::Viewer, + Access::Unknown, + ] { + assert_eq!(access_for(None, parent, false), Access::Owner); + assert_eq!(access_for(None, parent, true), Access::Viewer); + assert_eq!( + access_for(Some(MemberRole::Viewer), parent, true), + Access::Viewer + ); + assert_eq!( + access_for(Some(MemberRole::Editor), parent, true), + Access::Editor + ); + assert_eq!( + access_for(Some(MemberRole::Admin), parent, true), + Access::Editor + ); + assert_eq!( + access_for(Some(MemberRole::Inherited), parent, true), + parent + ); + } + } + + #[test] + fn permission_bits_follow_effective_access() { + for writable in [Access::Owner, Access::Editor] { + assert_eq!(perm_bits(true, writable), 0o755); + assert_eq!(perm_bits(false, writable), 0o644); + } + for read_only in [Access::Viewer, Access::Unknown] { + assert_eq!(perm_bits(true, read_only), 0o555); + assert_eq!(perm_bits(false, read_only), 0o444); + } + } +} diff --git a/crates/pdfs-core/src/cache.rs b/crates/pdfs-core/src/cache.rs index 789d6c4..ae77a7d 100644 --- a/crates/pdfs-core/src/cache.rs +++ b/crates/pdfs-core/src/cache.rs @@ -17,6 +17,7 @@ //! [`Node`]: proton_drive_rs::Node use std::collections::{BTreeMap, HashSet}; +use std::fmt; use std::io::{Read, Seek, SeekFrom}; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -28,10 +29,11 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use crate::db::{CacheEntryInput, Db}; -use crate::error::Result; +use crate::error::{Error, Result}; /// `cache_entries.kind` tag for whole-file content blobs. const KIND_BLOB: &str = "blob"; +static STAGING_N: AtomicU64 = AtomicU64::new(0); /// `cache_entries.kind` tag for on-demand block-cache chunks. const KIND_BLOCK: &str = "block"; /// `cache_entries.kind` tag for cached thumbnails. @@ -169,6 +171,35 @@ pub struct Baseline { pub revision_id: Option, } +/// A denied-write rescue that retained the bytes but could not fully publish +/// their recovery metadata. +#[derive(Debug)] +pub struct PreserveWriteError { + /// The actual path that still contains the accepted bytes. + pub surviving_path: PathBuf, + /// The recovery metadata that still describes those bytes, when one could + /// be made durable. Callers must never clear this after an error. + pub marker_path: Option, + error: Error, +} + +impl fmt::Display for PreserveWriteError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "{} (bytes retained at {})", + self.error, + self.surviving_path.display() + ) + } +} + +impl std::error::Error for PreserveWriteError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(&self.error) + } +} + /// Content cache rooted at a directory, with a sibling pin-registry file. pub struct ContentCache { /// Directory holding `` blobs and `.meta` tags. @@ -225,6 +256,18 @@ pub struct ContentCache { } impl ContentCache { + fn staging_path(&self, meta: &StagedWrite) -> PathBuf { + // The uid goes in the name so a staged file can be tied back to its node + // without a database; `/` in a uid would otherwise open a path. + let safe_uid = meta.uid.replace(['/', '\\'], "_"); + self.staging_dir.join(format!( + "{}-{}-{}", + safe_uid, + now_millis(), + STAGING_N.fetch_add(1, Ordering::Relaxed) + )) + } + /// Open (and create) a cache under `content_dir`, with the pin registry at /// `pins_path` and a `max_bytes` size cap (`0` = unlimited). Both parent /// directories are created if missing. `db` is the shared metadata database @@ -910,16 +953,7 @@ impl ContentCache { pub fn stage_write(&self, meta: &StagedWrite, scratch: &Path) -> Result { use std::io::Write as _; - static N: AtomicU64 = AtomicU64::new(0); - // The uid goes in the name so a staged file can be tied back to its node - // without a database; `/` in a uid would otherwise open a path. - let safe_uid = meta.uid.replace(['/', '\\'], "_"); - let path = self.staging_dir.join(format!( - "{}-{}-{}", - safe_uid, - now_millis(), - N.fetch_add(1, Ordering::Relaxed) - )); + let path = self.staging_path(meta); let copied = if let Err(e) = std::fs::rename(scratch, &path) { if e.kind() != std::io::ErrorKind::CrossesDevices { return Err(e.into()); @@ -949,6 +983,154 @@ impl ContentCache { Ok(path) } + /// Preserve accepted bytes which cannot be queued. + /// + /// Scratch and staging normally share a filesystem, so the normal path is + /// an atomic rename. If that is impossible, the source first receives a + /// durable recovery marker and remains in place until a copied destination, + /// its sidecar, and the staging directory are all synced. + pub fn preserve_write( + &self, + meta: &StagedWrite, + scratch: &Path, + ) -> std::result::Result { + let path = self.staging_path(meta); + self.preserve_write_at(meta, scratch, &path) + } + + fn preserve_write_at( + &self, + meta: &StagedWrite, + scratch: &Path, + path: &Path, + ) -> std::result::Result { + use std::io::Write as _; + + let retained = |surviving_path: &Path, error: Error| { + let adjacent = Self::scratch_sidecar(surviving_path); + let source = Self::scratch_sidecar(scratch); + let marker_path = [adjacent, source].into_iter().find(|path| path.is_file()); + PreserveWriteError { + surviving_path: surviving_path.to_path_buf(), + marker_path, + error, + } + }; + // If rename cannot happen, this marker is what lets the next open move + // the still-authoritative source into recovery instead of deleting it. + let marker_result = self.mark_scratch_durable(scratch, meta); + let side = path.with_extension("json"); + let side_tmp = path.with_extension("json.tmp"); + let publish_sidecar = || -> Result<()> { + let mut side_file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&side_tmp)?; + side_file.write_all(&serde_json::to_vec_pretty(meta)?)?; + side_file.sync_all()?; + drop(side_file); + std::fs::rename(&side_tmp, &side)?; + std::fs::File::open(&self.staging_dir)?.sync_all()?; + Ok(()) + }; + + match std::fs::rename(scratch, path) { + Ok(()) => { + if let Err(error) = std::fs::File::open(path).and_then(|file| file.sync_all()) { + if std::fs::rename(path, scratch).is_ok() { + if !Self::scratch_sidecar(scratch).is_file() { + let _ = self.mark_scratch_durable(scratch, meta); + } + let _ = + std::fs::File::open(&self.scratch_dir).and_then(|dir| dir.sync_all()); + return Err(retained(scratch, error.into())); + } + return Err(retained(path, error.into())); + } + let scratch_side = Self::scratch_sidecar(scratch); + let side_result = if marker_result.is_ok() { + std::fs::rename(&scratch_side, &side) + .map_err(Error::from) + .and_then(|()| { + std::fs::File::open(&self.staging_dir)? + .sync_all() + .map_err(Error::from) + }) + } else { + publish_sidecar() + }; + if let Err(error) = side_result { + let _ = std::fs::remove_file(&side_tmp); + // Publication did not complete. Put the blob back beside + // the source marker so startup recovery still has a + // reconstructable pair, rather than an unexplained staging + // blob and a marker whose source vanished. + if std::fs::rename(path, scratch).is_ok() { + if !Self::scratch_sidecar(scratch).is_file() { + let _ = self.mark_scratch_durable(scratch, meta); + } + let _ = + std::fs::File::open(&self.scratch_dir).and_then(|dir| dir.sync_all()); + return Err(retained(scratch, error)); + } + return Err(retained(path, error)); + } + Ok(path.to_path_buf()) + } + Err(rename_error) => { + let source_marked = + marker_result.is_ok() || Self::scratch_sidecar(scratch).exists(); + if let Err(marker_error) = marker_result { + tracing::warn!( + error = %marker_error, + path = %scratch.display(), + "could not mark rescue source durable before copy fallback" + ); + } + if let Err(error) = std::fs::copy(scratch, path) { + if !source_marked { + let recovery = self.recovery_dir.join( + path.file_name() + .unwrap_or_else(|| std::ffi::OsStr::new("denied-write")), + ); + if std::fs::rename(scratch, &recovery).is_ok() { + let _ = std::fs::File::open(&self.recovery_dir) + .and_then(|dir| dir.sync_all()); + return Err(retained( + &recovery, + Error::Other(format!( + "rename failed ({rename_error}); copy fallback failed ({error})" + )), + )); + } + } + return Err(retained( + scratch, + Error::Other(format!( + "rename failed ({rename_error}); copy fallback failed ({error})" + )), + )); + } + if let Err(error) = std::fs::File::open(path).and_then(|file| file.sync_all()) { + let surviving = if source_marked { scratch } else { path }; + return Err(retained(surviving, error.into())); + } + if let Err(error) = publish_sidecar() { + let _ = std::fs::remove_file(&side_tmp); + let surviving = if source_marked { scratch } else { path }; + return Err(retained(surviving, error)); + } + // Deliberately last: fallback never removes its marked source + // until destination publication is complete. + if let Err(error) = std::fs::remove_file(scratch) { + return Err(retained(path, error.into())); + } + self.clear_scratch_durable(scratch); + Ok(path.to_path_buf()) + } + } + } + /// Drop a staged blob and its sidecar, once its bytes are safely on the /// remote (or the op that owned them has been superseded). Best effort: /// leftovers cost disk, while failing here would strand a drained op. @@ -1906,6 +2088,92 @@ mod tests { ); } + fn preservation_meta() -> StagedWrite { + StagedWrite { + uid: uid("preserved").to_string(), + len: 14, + base_size: 0, + base_mtime: 100, + authored: vec![(0, 14)], + complete: true, + based_on: None, + } + } + + #[test] + fn orphan_preservation_publishes_exact_bytes_then_removes_source() { + use std::io::Write as _; + + let (c, _dir) = cache(); + let (mut file, scratch) = c.create_scratch().unwrap(); + file.write_all(b"accepted bytes").unwrap(); + file.sync_all().unwrap(); + drop(file); + + let preserved = c.preserve_write(&preservation_meta(), &scratch).unwrap(); + + assert!(!scratch.exists()); + assert_eq!(std::fs::read(&preserved).unwrap(), b"accepted bytes"); + let sidecar: StagedWrite = + serde_json::from_slice(&std::fs::read(preserved.with_extension("json")).unwrap()) + .unwrap(); + let expected = preservation_meta(); + assert_eq!(sidecar.uid, expected.uid); + assert_eq!(sidecar.len, expected.len); + assert_eq!(sidecar.authored, expected.authored); + assert_eq!(sidecar.complete, expected.complete); + } + + #[test] + fn orphan_preservation_sidecar_failure_reports_surviving_blob_across_reopen() { + use std::io::Write as _; + + let (c, dir) = cache(); + let (mut file, scratch) = c.create_scratch().unwrap(); + file.write_all(b"only surviving copy").unwrap(); + file.sync_all().unwrap(); + drop(file); + + let destination = c.staging_dir.join("forced-preservation-failure"); + let side = destination.with_extension("json"); + std::fs::create_dir(&side).unwrap(); + + let error = c + .preserve_write_at(&preservation_meta(), &scratch, &destination) + .unwrap_err(); + assert_eq!(error.surviving_path, scratch); + assert!( + scratch.exists(), + "failed sidecar publication must roll the blob back beside its marker" + ); + assert_eq!( + std::fs::read(&error.surviving_path).unwrap(), + b"only surviving copy" + ); + assert_eq!( + error.marker_path, + Some(ContentCache::scratch_sidecar(&scratch)) + ); + assert!( + side.is_dir(), + "the induced sidecar failure remains observable" + ); + + let content = dir.path().join("content"); + let pins = dir.path().join("pins-reopen.json"); + drop(c); + let reopened = + ContentCache::open(content, pins, 0, Arc::new(Db::open_in_memory().unwrap())).unwrap(); + let recovered = reopened.recovered_writes(); + assert_eq!(recovered.len(), 1); + assert_eq!( + std::fs::read(&recovered[0].0).unwrap(), + b"only surviving copy" + ); + assert_eq!(recovered[0].1.uid, preservation_meta().uid); + drop(reopened); + } + /// A2: `fsync(2)` promises the bytes survive a crash, but queueing happens /// at `release`. A scratch file marked durable must therefore outlive the /// open that clears the scratch directory — and an unmarked one must not, diff --git a/crates/pdfs-core/src/db/maintenance.rs b/crates/pdfs-core/src/db/maintenance.rs index 4bd7499..cca4233 100644 --- a/crates/pdfs-core/src/db/maintenance.rs +++ b/crates/pdfs-core/src/db/maintenance.rs @@ -81,6 +81,7 @@ const REPORTED_TABLES: &[&str] = &[ "pending_op", "activity", "local_files", + "share_access", ]; impl Db { diff --git a/crates/pdfs-core/src/db/migrations.rs b/crates/pdfs-core/src/db/migrations.rs index 93d6349..e79853f 100644 --- a/crates/pdfs-core/src/db/migrations.rs +++ b/crates/pdfs-core/src/db/migrations.rs @@ -9,7 +9,7 @@ use super::Db; use crate::Result; /// Current schema version. Bump on every forward migration added below. -pub(super) const SCHEMA_VERSION: i64 = 16; +pub(super) const SCHEMA_VERSION: i64 = 17; impl Db { pub(super) fn migrate(&self) -> Result<()> { @@ -89,6 +89,9 @@ impl Db { if current < 16 { tx.execute_batch(MIGRATION_V16)?; } + if current < 17 { + tx.execute_batch(MIGRATION_V17)?; + } tx.execute( "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1) ON CONFLICT(key) DO UPDATE SET value = excluded.value", @@ -411,3 +414,14 @@ CREATE INDEX IF NOT EXISTS idx_local_files_name_nocase ON local_files(name COLLATE NOCASE); INSERT INTO local_fts(local_fts) VALUES('rebuild'); "; + +/// Schema v17: effective access for roots shared with this account. The role is +/// persisted separately from `nodes.node_json` so a cold/offline mount can +/// enforce the last known permission even when the SDK cannot refresh metadata. +const MIGRATION_V17: &str = " +CREATE TABLE share_access ( + root_uid TEXT PRIMARY KEY, + access TEXT NOT NULL + CHECK (access IN ('owner', 'editor', 'viewer', 'unknown')) +); +"; diff --git a/crates/pdfs-core/src/db/mod.rs b/crates/pdfs-core/src/db/mod.rs index b88cb2a..c90f671 100644 --- a/crates/pdfs-core/src/db/mod.rs +++ b/crates/pdfs-core/src/db/mod.rs @@ -31,6 +31,7 @@ mod nodes; mod ops; mod photos; mod pins; +mod share_access; mod state; mod sync; mod trash; @@ -43,7 +44,7 @@ pub use maintenance::{DbStats, VacuumOutcome}; pub use nodes::{SearchHit, StoredNode}; pub use ops::{ AttachedBlob, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PARK_UNTIL, - PendingCounts, PendingOp, op_supersedes, + PendingCounts, PendingOp, RenameMeta, op_supersedes, }; pub use photos::{StoredPhoto, THUMB_HAVE, THUMB_NONE, THUMB_UNKNOWN}; pub use sync::{StoredSyncEntry, StoredSyncFolder}; diff --git a/crates/pdfs-core/src/db/nodes.rs b/crates/pdfs-core/src/db/nodes.rs index 540ad43..ab33228 100644 --- a/crates/pdfs-core/src/db/nodes.rs +++ b/crates/pdfs-core/src/db/nodes.rs @@ -148,6 +148,32 @@ impl Db { Ok(()) } + /// Atomically retire a landed trash operation and its retained authority. + /// + /// Queued trash keeps the node row so drain-time permission checks can + /// resolve its shared-tree access. Removing the op first could resurrect + /// that row after a crash; removing the row first could strand a now + /// unauthorizable op. One transaction closes both windows. + pub fn complete_trash_op(&self, op_id: i64, uid: &NodeUid) -> Result<()> { + let uid = uid.to_string(); + let mut conn = self.conn.lock(); + let tx = conn.transaction()?; + let rowid: Option = tx + .query_row( + "SELECT rowid FROM nodes WHERE uid = ?1", + params![uid], + |row| row.get(0), + ) + .optional()?; + tx.execute("DELETE FROM nodes WHERE uid = ?1", params![uid])?; + if let Some(rowid) = rowid { + tx.execute("DELETE FROM nodes_fts WHERE rowid = ?1", params![rowid])?; + } + tx.execute("DELETE FROM pending_op WHERE id = ?1", params![op_id])?; + tx.commit()?; + Ok(()) + } + /// Check if a folder node has any non-trashed children in the database. pub fn has_children(&self, parent_uid: &NodeUid) -> Result { let uid_str = parent_uid.to_string(); diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs index b3acab6..e02139a 100644 --- a/crates/pdfs-core/src/db/ops.rs +++ b/crates/pdfs-core/src/db/ops.rs @@ -3,6 +3,7 @@ //! order, so a child never drains before the parent that gives it a real uid. use rusqlite::{OptionalExtension, params}; +use serde::{Deserialize, Serialize}; use super::Db; use crate::Result; @@ -35,6 +36,16 @@ pub const OP_MKDIR: &str = "mkdir"; /// that folder's own drain, exactly as for [`OP_CREATE`]. pub const OP_RENAME: &str = "rename"; +/// Drain-time authority retained by a queued rename. +/// +/// The optimistic local move rewrites the node's persisted ancestry to the +/// destination immediately. Keeping the original parent here prevents that +/// rewrite from erasing the shared tree that admitted the move. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RenameMeta { + pub original_parent_uid: String, +} + /// The `kind` of a [`PendingOp`] that trashes a node the server knows about /// (offline.md Phase 3b). /// @@ -127,6 +138,32 @@ pub struct AttachedBlob { } impl Db { + /// Whether a specific desired-state operation is still queued for a node. + pub fn has_pending_op(&self, uid: &str, kind: &str) -> Result { + let conn = self.conn.lock(); + conn.query_row( + "SELECT EXISTS( + SELECT 1 FROM pending_op WHERE uid = ?1 AND kind = ?2 + )", + params![uid, kind], + |row| row.get(0), + ) + .map_err(Into::into) + } + + /// Read an existing operation's metadata before a superseding enqueue. + pub fn pending_op_meta(&self, uid: &str, kind: &str) -> Result> { + let conn = self.conn.lock(); + conn.query_row( + "SELECT meta_json FROM pending_op WHERE uid = ?1 AND kind = ?2", + params![uid, kind], + |row| row.get(0), + ) + .optional() + .map(|value| value.flatten()) + .map_err(Into::into) + } + pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option)> { let mut conn = self.conn.lock(); let tx = conn.transaction()?; @@ -167,6 +204,50 @@ impl Db { Ok((id, superseded)) } + /// Atomically replace every queued operation for `uid` and its queued + /// descendants with one remote trash intent. + /// + /// Staged blob paths are returned only after the transaction commits. A + /// failed trash insert therefore rolls the deletions back and leaves every + /// prior operation owning its bytes. + pub fn replace_ops_with_trash( + &self, + uid: &str, + name: &str, + created_at: i64, + ) -> Result<(i64, Vec)> { + const SUBTREE: &str = " + WITH RECURSIVE doomed(uid) AS ( + SELECT ?1 + UNION + SELECT p.uid FROM pending_op p JOIN doomed d ON p.parent_uid = d.uid + )"; + let mut conn = self.conn.lock(); + let tx = conn.transaction()?; + let blobs: Vec = { + let mut stmt = tx.prepare(&format!( + "{SUBTREE} + SELECT blob_path FROM pending_op + WHERE uid IN (SELECT uid FROM doomed) AND blob_path IS NOT NULL" + ))?; + let rows = stmt.query_map(params![uid], |row| row.get::<_, String>(0))?; + rows.collect::, _>>()? + }; + tx.execute( + &format!("{SUBTREE} DELETE FROM pending_op WHERE uid IN (SELECT uid FROM doomed)"), + params![uid], + )?; + tx.execute( + "INSERT INTO pending_op + (kind, uid, parent_uid, name, blob_path, meta_json, created_at, next_attempt_at) + VALUES (?1, ?2, NULL, ?3, NULL, NULL, ?4, 0)", + params![OP_TRASH, uid, name, created_at], + )?; + let id = tx.last_insert_rowid(); + tx.commit()?; + Ok((id, blobs)) + } + /// Point a queued create at the bytes that were just written to it, returning /// any blob it previously held so the caller can discard it. /// @@ -509,4 +590,23 @@ impl Db { )?; Ok(()) } + + /// Defer a due operation without recording a failed remote attempt. + /// + /// Access checks happen before the drain calls Proton, so a locally denied + /// operation has not failed remotely and must not consume a retry. Preserve + /// the long-lived transient-create sentinel if this is ever called for one. + pub fn defer_op_without_attempt(&self, id: i64, next_attempt_at: i64) -> Result<()> { + let conn = self.conn.lock(); + conn.execute( + "UPDATE pending_op + SET next_attempt_at = CASE + WHEN next_attempt_at >= ?3 THEN next_attempt_at + ELSE ?2 + END + WHERE id = ?1", + params![id, next_attempt_at, PARK_UNTIL], + )?; + Ok(()) + } } diff --git a/crates/pdfs-core/src/db/share_access.rs b/crates/pdfs-core/src/db/share_access.rs new file mode 100644 index 0000000..8a47bd4 --- /dev/null +++ b/crates/pdfs-core/src/db/share_access.rs @@ -0,0 +1,143 @@ +//! Persisted effective access for roots shared with this account. + +use std::collections::HashMap; + +use proton_drive_rs::proton_sdk::ids::{LinkId, NodeUid, VolumeId}; +use rusqlite::{OptionalExtension, params}; + +use super::Db; +use crate::{Access, Error, Result}; + +impl Db { + /// Load the complete, deliberately small shared-root authority map. + /// + /// Mount state keeps this resident so ordinary node interning and subtree + /// propagation never turn into one SQLite query per inode. The database + /// remains the restart/offline authority and every explicit role change is + /// written through before this cache is updated. + pub fn all_share_access(&self) -> Result> { + let conn = self.conn.lock(); + let mut stmt = conn.prepare("SELECT root_uid, access FROM share_access")?; + let rows = stmt.query_map([], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + })?; + let mut access = HashMap::new(); + for row in rows { + let (uid, value) = row?; + let (volume, link) = uid.split_once('~').ok_or_else(|| { + Error::Other(format!( + "invalid shared-root uid {uid:?} stored in database" + )) + })?; + let access_value = Access::from_db_str(&value).ok_or_else(|| { + Error::Other(format!("invalid share access {value:?} stored for {uid}")) + })?; + access.insert( + NodeUid::new(VolumeId::from(volume), LinkId::from(link)), + access_value, + ); + } + Ok(access) + } + + /// Read the effective access recorded for a shared-tree root. + pub fn share_access(&self, uid: &NodeUid) -> Result> { + let conn = self.conn.lock(); + let value = conn + .query_row( + "SELECT access FROM share_access WHERE root_uid = ?1", + [uid.to_string()], + |row| row.get::<_, String>(0), + ) + .optional()?; + value + .map(|value| { + Access::from_db_str(&value).ok_or_else(|| { + Error::Other(format!("invalid share access {value:?} stored for {uid}")) + }) + }) + .transpose() + } + + /// Insert or replace the effective access for a shared-tree root. + pub fn set_share_access(&self, uid: &NodeUid, access: Access) -> Result<()> { + let conn = self.conn.lock(); + conn.execute( + "INSERT INTO share_access (root_uid, access) VALUES (?1, ?2) + ON CONFLICT(root_uid) DO UPDATE SET access = excluded.access", + params![uid.to_string(), access.as_db_str()], + )?; + Ok(()) + } + + /// Forget a shared-tree root's effective access. + pub fn delete_share_access(&self, uid: &NodeUid) -> Result<()> { + let conn = self.conn.lock(); + conn.execute( + "DELETE FROM share_access WHERE root_uid = ?1", + [uid.to_string()], + )?; + Ok(()) + } + + /// Fail closed every persisted shared-tree authority. + /// + /// Owned and device trees have no row in `share_access`, so this only + /// affects roots already identified as shared with this account. + pub fn downgrade_all_share_access(&self) -> Result { + let conn = self.conn.lock(); + Ok(conn.execute( + "UPDATE share_access SET access = 'viewer' WHERE access != 'viewer'", + [], + )?) + } + + /// Resolve a persisted node's effective access through its nearest recorded + /// share root. `None` means the node itself is not persisted, which callers + /// must not treat as owned: stale handles need to fail closed. + pub fn effective_node_access(&self, uid: &NodeUid) -> Result> { + let conn = self.conn.lock(); + let exists: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM nodes WHERE uid = ?1)", + [uid.to_string()], + |row| row.get(0), + )?; + if !exists { + return Ok(None); + } + let value = conn + .query_row( + "WITH RECURSIVE ancestors(uid, parent_uid, depth, path) AS ( + SELECT uid, parent_uid, 0, char(31) || uid || char(31) + FROM nodes WHERE uid = ?1 + UNION ALL + SELECT a.parent_uid, n.parent_uid, a.depth + 1, + a.path || a.parent_uid || char(31) + FROM ancestors a + LEFT JOIN nodes n ON n.uid = a.parent_uid + WHERE a.parent_uid IS NOT NULL + AND instr( + a.path, + char(31) || a.parent_uid || char(31) + ) = 0 + ) + SELECT sa.access + FROM ancestors a JOIN share_access sa ON sa.root_uid = a.uid + ORDER BY a.depth ASC + LIMIT 1", + [uid.to_string()], + |row| row.get::<_, String>(0), + ) + .optional()?; + value + .map(|value| { + Access::from_db_str(&value).ok_or_else(|| { + Error::Other(format!( + "invalid effective share access {value:?} for {uid}" + )) + }) + }) + .transpose() + .map(|access| Some(access.unwrap_or(Access::Owner))) + } +} diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index 3b19ecf..fedaa7f 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -350,6 +350,34 @@ fn delete_node_removes_row() { assert!(db.load_all().unwrap().is_empty()); } +#[test] +fn completing_trash_atomically_removes_the_op_and_retained_node() { + let db = Db::open_in_memory().unwrap(); + let node = file("trash-me", "parent", "trash-me.txt", 4); + let uid = node.uid.clone(); + db.upsert_node(&node).unwrap(); + db.enqueue_op(&PendingOp { + id: 0, + kind: OP_TRASH.to_string(), + uid: uid.to_string(), + parent_uid: None, + name: Some(node.name.clone()), + blob_path: None, + meta_json: None, + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }) + .unwrap(); + let op = db.pending_ops().unwrap().remove(0); + + db.complete_trash_op(op.id, &uid).unwrap(); + + assert!(db.node_by_uid(&uid.to_string()).unwrap().is_none()); + assert!(db.pending_ops().unwrap().is_empty()); +} + #[test] fn children_if_listed_gated_on_flag() { let db = Db::open_in_memory().unwrap(); @@ -784,6 +812,154 @@ fn opens_and_migrates() { assert_eq!(version, SCHEMA_VERSION.to_string()); } +#[test] +fn share_access_round_trips_updates_and_deletes() { + let db = Db::open_in_memory().unwrap(); + let root = uid("shared-root"); + + assert_eq!(db.share_access(&root).unwrap(), None); + for access in [ + crate::Access::Owner, + crate::Access::Editor, + crate::Access::Viewer, + crate::Access::Unknown, + ] { + db.set_share_access(&root, access).unwrap(); + assert_eq!(db.share_access(&root).unwrap(), Some(access)); + assert_eq!(db.all_share_access().unwrap().get(&root), Some(&access)); + } + db.delete_share_access(&root).unwrap(); + assert_eq!(db.share_access(&root).unwrap(), None); + assert!(db.all_share_access().unwrap().is_empty()); +} + +#[test] +fn effective_node_access_inherits_from_the_persisted_share_root() { + let db = Db::open_in_memory().unwrap(); + db.upsert_node(&folder("root", None, "My Files")).unwrap(); + db.upsert_node(&folder("shared", Some("root"), "Shared")) + .unwrap(); + db.upsert_node(&folder("nested", Some("shared"), "Nested")) + .unwrap(); + db.upsert_node(&file("child", "nested", "child.txt", 1)) + .unwrap(); + db.set_share_access(&uid("shared"), crate::Access::Viewer) + .unwrap(); + + assert_eq!( + db.effective_node_access(&uid("child")).unwrap(), + Some(crate::Access::Viewer) + ); + assert_eq!( + db.effective_node_access(&uid("root")).unwrap(), + Some(crate::Access::Owner) + ); + assert_eq!(db.effective_node_access(&uid("missing")).unwrap(), None); +} + +#[test] +fn effective_node_access_uses_the_nearest_nested_share_root() { + let db = Db::open_in_memory().unwrap(); + db.upsert_node(&folder("root", None, "My Files")).unwrap(); + db.upsert_node(&folder("outer", Some("root"), "Outer")) + .unwrap(); + db.upsert_node(&folder("inner", Some("outer"), "Inner")) + .unwrap(); + db.upsert_node(&file("child", "inner", "child.txt", 1)) + .unwrap(); + db.set_share_access(&uid("outer"), crate::Access::Editor) + .unwrap(); + db.set_share_access(&uid("inner"), crate::Access::Viewer) + .unwrap(); + + assert_eq!( + db.effective_node_access(&uid("child")).unwrap(), + Some(crate::Access::Viewer) + ); +} + +#[test] +fn effective_node_access_keeps_a_missing_share_root_tombstone_authoritative() { + let db = Db::open_in_memory().unwrap(); + db.upsert_node(&folder("root", None, "My Files")).unwrap(); + db.upsert_node(&folder("shared", Some("root"), "Shared")) + .unwrap(); + db.upsert_node(&folder("nested", Some("shared"), "Nested")) + .unwrap(); + db.upsert_node(&file("child", "nested", "child.txt", 1)) + .unwrap(); + db.set_share_access(&uid("shared"), crate::Access::Viewer) + .unwrap(); + db.delete_node(&uid("shared")).unwrap(); + + assert_eq!( + db.effective_node_access(&uid("child")).unwrap(), + Some(crate::Access::Viewer) + ); +} + +#[test] +fn downgrade_all_share_access_leaves_unrecorded_owned_nodes_owned() { + let db = Db::open_in_memory().unwrap(); + db.upsert_node(&folder("root", None, "My Files")).unwrap(); + db.upsert_node(&folder("shared", Some("root"), "Shared")) + .unwrap(); + db.set_share_access(&uid("shared"), crate::Access::Editor) + .unwrap(); + + assert_eq!(db.downgrade_all_share_access().unwrap(), 1); + assert_eq!( + db.effective_node_access(&uid("shared")).unwrap(), + Some(crate::Access::Viewer) + ); + assert_eq!( + db.effective_node_access(&uid("root")).unwrap(), + Some(crate::Access::Owner) + ); +} + +#[test] +fn migration_v17_adds_share_access_to_a_v16_fixture() { + let path = std::env::temp_dir().join(format!( + "pdfs-db-v16-fixture-{}-{}.db", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + { + // V17 only adds `share_access`, so removing it and restoring the version + // stamp produces the exact schema a released V16 database had. + let db = Db::open(&path).unwrap(); + db.set_state_str("fixture_value", "survives").unwrap(); + } + { + let conn = rusqlite::Connection::open(&path).unwrap(); + conn.execute_batch( + "DROP TABLE share_access; + UPDATE sync_state SET value = '16' WHERE key = 'schema_version';", + ) + .unwrap(); + } + + let db = Db::open(&path).unwrap(); + assert_eq!( + db.state_str("fixture_value").unwrap().as_deref(), + Some("survives") + ); + let root = uid("offline-share"); + db.set_share_access(&root, crate::Access::Viewer).unwrap(); + assert_eq!(db.share_access(&root).unwrap(), Some(crate::Access::Viewer)); + assert_eq!( + db.state_str("schema_version").unwrap().as_deref(), + Some("17") + ); + + drop(db); + let _ = std::fs::remove_file(path); +} + #[test] fn refuses_a_schema_newer_than_the_running_build() { let path = std::env::temp_dir().join(format!( @@ -935,6 +1111,82 @@ fn failed_superseding_insert_keeps_the_old_pending_op() { assert_eq!(ops[0].blob_path.as_deref(), Some("/staging/original")); } +#[test] +fn atomic_trash_replacement_returns_owned_blobs_after_commit() { + let db = Db::open_in_memory().unwrap(); + let root = uid("trash-root").to_string(); + let child = uid("trash-child").to_string(); + let revision = |uid: &str, parent_uid: Option<&str>, blob: &str| PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: uid.to_string(), + parent_uid: parent_uid.map(str::to_string), + name: None, + blob_path: Some(blob.to_string()), + meta_json: Some("{}".to_string()), + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }; + db.enqueue_op(&revision(&root, None, "/staging/root")) + .unwrap(); + db.enqueue_op(&revision(&child, Some(&root), "/staging/child")) + .unwrap(); + + let (trash_id, mut blobs) = db.replace_ops_with_trash(&root, "removed.txt", 2).unwrap(); + blobs.sort(); + assert_eq!(blobs, vec!["/staging/child", "/staging/root"]); + + let ops = db.pending_ops().unwrap(); + assert_eq!(ops.len(), 1); + assert_eq!(ops[0].id, trash_id); + assert_eq!(ops[0].kind, OP_TRASH); + assert_eq!(ops[0].uid, root); + assert_eq!(ops[0].name.as_deref(), Some("removed.txt")); + assert!(ops[0].blob_path.is_none()); +} + +#[test] +fn failed_atomic_trash_insert_keeps_prior_revision_and_blob_ownership() { + let db = Db::open_in_memory().unwrap(); + let node = uid("atomic-trash").to_string(); + db.enqueue_op(&PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: node.clone(), + parent_uid: None, + name: None, + blob_path: Some("/staging/still-owned".to_string()), + meta_json: Some("{}".to_string()), + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }) + .unwrap(); + db.with_conn(|conn| { + conn.execute_batch( + "CREATE TRIGGER reject_trash_insert BEFORE INSERT ON pending_op + WHEN NEW.kind = 'trash' + BEGIN SELECT RAISE(ABORT, 'injected trash insert failure'); END;", + )?; + Ok(()) + }) + .unwrap(); + + assert!(db.replace_ops_with_trash(&node, "removed.txt", 2).is_err()); + let ops = db.pending_ops().unwrap(); + assert_eq!(ops.len(), 1); + assert_eq!(ops[0].kind, OP_REVISION); + assert_eq!(ops[0].uid, node); + assert_eq!( + ops[0].blob_path.as_deref(), + Some("/staging/still-owned"), + "the rolled-back revision still owns its staged bytes" + ); +} + /// Deleting a folder that was created offline must take the ops queued /// underneath it with it: they name a placeholder parent that will now never /// become real, so nothing could ever drain them and nothing is left to @@ -1202,6 +1454,63 @@ fn a_failed_op_stays_queued_with_backoff() { assert_eq!(db.pending_op_counts().unwrap().uploads, 0); } +#[test] +fn an_access_blocked_op_is_deferred_without_counting_an_attempt() { + let db = Db::open_in_memory().unwrap(); + db.enqueue_op(&PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: uid("blocked").to_string(), + parent_uid: None, + name: None, + blob_path: Some("/staging/blocked".to_string()), + meta_json: Some("{}".to_string()), + created_at: 1, + attempts: 3, + last_error: Some("earlier network failure".to_string()), + next_attempt_at: 0, + }) + .unwrap(); + let before = db.pending_ops().unwrap(); + let id = before[0].id; + let count = db.pending_op_counts().unwrap(); + + db.defer_op_without_attempt(id, 5_000).unwrap(); + + let after = db.pending_ops().unwrap(); + assert_eq!(after.len(), 1); + let after_count = db.pending_op_counts().unwrap(); + assert_eq!(after_count.uploads, count.uploads); + assert_eq!(after_count.changes, count.changes); + assert_eq!(after[0].attempts, before[0].attempts); + assert_eq!(after[0].last_error, before[0].last_error); + assert_eq!(after[0].next_attempt_at, 5_000); +} + +#[test] +fn access_deferral_does_not_unpark_a_transient_create() { + let db = Db::open_in_memory().unwrap(); + db.enqueue_op(&PendingOp { + id: 0, + kind: OP_CREATE.to_string(), + uid: uid("transient").to_string(), + parent_uid: Some(uid("parent").to_string()), + name: Some(".part".to_string()), + blob_path: Some("/staging/transient".to_string()), + meta_json: Some("{}".to_string()), + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: PARK_UNTIL, + }) + .unwrap(); + let op = db.pending_ops().unwrap().remove(0); + + db.defer_op_without_attempt(op.id, 5_000).unwrap(); + + assert_eq!(db.pending_ops().unwrap()[0].next_attempt_at, PARK_UNTIL); +} + #[test] fn migrate_is_idempotent() { let db = Db::open_in_memory().unwrap(); @@ -1846,6 +2155,7 @@ fn test_db_has_children_and_trashed_filtering() { is_shared: false, is_shared_publicly: false, signature_email: None, + membership: None, verification: Default::default(), }; db.upsert_node(&parent_node).unwrap(); @@ -1869,6 +2179,7 @@ fn test_db_has_children_and_trashed_filtering() { is_shared: false, is_shared_publicly: false, signature_email: None, + membership: None, verification: Default::default(), }; db.upsert_node(&child_node).unwrap(); diff --git a/crates/pdfs-core/src/lib.rs b/crates/pdfs-core/src/lib.rs index 4ef0eed..8d4babb 100644 --- a/crates/pdfs-core/src/lib.rs +++ b/crates/pdfs-core/src/lib.rs @@ -2,6 +2,7 @@ //! identity, XDG paths, and the login/session/keyring layer that every //! front-end (CLI today, daemon + GUI later) shares. +pub mod access; pub mod auth; pub mod cache; pub mod config; @@ -15,6 +16,7 @@ pub mod service; pub mod shell; pub mod syncignore; +pub use access::{Access, access_for, perm_bits}; pub use error::{CoreError, CoreResult, Error, Result}; // Re-export the SDK surface downstream crates need, so they depend on diff --git a/crates/pdfs-fuse/src/background.rs b/crates/pdfs-fuse/src/background.rs index d907b9b..02ffd7b 100644 --- a/crates/pdfs-fuse/src/background.rs +++ b/crates/pdfs-fuse/src/background.rs @@ -27,6 +27,40 @@ fn forget_and_notify(st: &mut State, notifier: Option<&Notifier>, uid: &NodeUid) true } +fn invalidate_access_changes(notifier: Option<&Notifier>, changed: &[u64]) { + let Some(notifier) = notifier else { + return; + }; + for &ino in changed { + let _ = notifier.inval_inode(INodeNo(ino), 0, 0); + } +} + +/// Persist the fail-closed part of an access event and install the same denial +/// in resident state even when SQLite rejects the write. The caller must not +/// acknowledge the event when this returns an error. +fn apply_access_downgrade( + db: &Db, + event: &DriveEvent, + mut deny: impl FnMut(Option<&NodeUid>), +) -> pdfs_core::Result<()> { + match event { + DriveEvent::NodeDeleted { node_uid, .. } => { + let result = db.set_share_access(node_uid, Access::Viewer); + deny(Some(node_uid)); + result + } + DriveEvent::ContinuityLost { .. } + | DriveEvent::ScopeAccessLost { .. } + | DriveEvent::SharedWithMeUpdated { .. } => { + let result = db.downgrade_all_share_access().map(|_| ()); + deny(None); + result + } + _ => Ok(()), + } +} + /// Apply one remote event to the local cache and notify the kernel so it drops /// any stale cached metadata/data for the affected inodes. /// @@ -43,7 +77,16 @@ fn forget_and_notify(st: &mut State, notifier: Option<&Notifier>, uid: &NodeUid) /// read-side twin of `docs/BUGS.md` B74. Inode numbers are per-mount, so each /// mount must be notified through its **own** channel; that pairing is what /// [`Core::for_each_mount`] exists to preserve. -fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) { +fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) -> pdfs_core::Result<()> { + let access_result = apply_access_downgrade(&core.db, event, |root| { + core.for_each_mount(|st, notifier| { + let changed = match root { + Some(uid) => st.downgrade_shared_subtree(uid), + None => st.downgrade_known_shared_access(), + }; + invalidate_access_changes(notifier, &changed); + }); + }); match event { DriveEvent::NodeUpdated { node_uid, @@ -109,11 +152,14 @@ fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) { forget_and_notify(st, notifier, node_uid); }); } - // Continuity or scope was lost: our cached listings may be arbitrarily - // stale, so drop every listing and tell the kernel to forget all - // metadata. Inodes stay stable; dirs simply re-enumerate on next access. - DriveEvent::ContinuityLost { .. } | DriveEvent::ScopeAccessLost { .. } => { - warn!("event continuity lost; dropping all cached listings, resyncing lazily"); + // Losing event continuity or any shared-access signal makes persisted + // shared roots untrustworthy. Keep owned/device trees unchanged, but + // fail every known shared subtree closed until a root refresh carries a + // current membership role. + DriveEvent::ContinuityLost { .. } + | DriveEvent::ScopeAccessLost { .. } + | DriveEvent::SharedWithMeUpdated { .. } => { + warn!("event access continuity lost; downgrading shared trees and resyncing lazily"); core.for_each_mount(|st, notifier| { let dirs: Vec = st.children.keys().copied().collect(); for &ino in &dirs { @@ -126,8 +172,20 @@ fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) { } // No substantive local change; the cursor advance is handled by the // caller persisting the event id. - DriveEvent::CursorAdvanced { .. } | DriveEvent::SharedWithMeUpdated { .. } => {} + DriveEvent::CursorAdvanced { .. } => {} } + access_result +} + +fn acknowledge_applied_event( + db: &Db, + event: &DriveEvent, + applied: pdfs_core::Result<()>, +) -> pdfs_core::Result { + applied?; + let id = event.id().clone(); + db.set_event_cursor(id.as_str())?; + Ok(id) } /// The folders one batch of events changed, collected so the listings are @@ -296,12 +354,17 @@ pub(super) async fn run_event_sync( warn!(error = %e, "sdk cache invalidation for event failed; retaining cursor for retry"); break; } - apply_event(&core, event, &mut dirty); - let applied = event.id().clone(); - if let Err(e) = db.set_event_cursor(applied.as_str()) { - warn!(error = %e, "persist event cursor failed; retaining prior cursor for retry"); - break; - } + let applied = match acknowledge_applied_event( + &db, + event, + apply_event(&core, event, &mut dirty), + ) { + Ok(applied) => applied, + Err(e) => { + warn!(error = %e, "event application or acknowledgment failed; retaining prior cursor for retry"); + break; + } + }; // Advance only after this exact event is durably acknowledged. If a // later event fails, polling resumes from here and replays nothing // that was not already made visible locally. @@ -396,3 +459,108 @@ fn scan_local_once( } indexing.store(false, Ordering::Relaxed); } + +#[cfg(test)] +mod tests { + use super::*; + use std::cell::Cell; + + fn uid(link: &str) -> NodeUid { + NodeUid::new(VolumeId::from("vol"), LinkId::from(link)) + } + + #[test] + fn deleted_share_downgrade_failure_denies_memory_and_retains_cursor() { + let db = Db::open(std::path::Path::new(":memory:")).unwrap(); + let root = uid("shared"); + db.set_share_access(&root, Access::Editor).unwrap(); + db.with_conn(|conn| { + conn.execute_batch( + "CREATE TRIGGER reject_share_update + BEFORE UPDATE ON share_access + BEGIN SELECT RAISE(FAIL, 'forced downgrade failure'); END;", + )?; + Ok(()) + }) + .unwrap(); + let event = DriveEvent::NodeDeleted { + id: DriveEventId::from("event-1"), + node_uid: root, + parent_node_uid: None, + }; + let denied = Cell::new(false); + + let applied = apply_access_downgrade(&db, &event, |root| { + denied.set(root.is_some()); + }); + assert!(applied.is_err()); + assert!(denied.get(), "resident state must fail closed immediately"); + assert!(acknowledge_applied_event(&db, &event, applied).is_err()); + assert!( + db.get_event_cursor().unwrap().is_none(), + "failed downgrade must leave the event unacknowledged" + ); + } + + #[test] + fn deleted_resident_root_without_share_row_is_denied_and_tombstoned() { + let db = Db::open(std::path::Path::new(":memory:")).unwrap(); + let root = uid("resident-without-row"); + assert!(db.share_access(&root).unwrap().is_none()); + let resident = std::cell::RefCell::new(std::collections::HashMap::from([( + root.clone(), + Access::Editor, + )])); + let event = DriveEvent::NodeDeleted { + id: DriveEventId::from("event-missing-row"), + node_uid: root.clone(), + parent_node_uid: None, + }; + + let applied = apply_access_downgrade(&db, &event, |target| { + assert_eq!(target, Some(&root)); + resident.borrow_mut().insert(root.clone(), Access::Viewer); + }); + assert!(applied.is_ok()); + assert_eq!(resident.borrow().get(&root), Some(&Access::Viewer)); + assert_eq!(db.share_access(&root).unwrap(), Some(Access::Viewer)); + + let acknowledged = acknowledge_applied_event(&db, &event, applied).unwrap(); + assert_eq!(acknowledged.as_str(), "event-missing-row"); + assert_eq!( + db.get_event_cursor().unwrap().as_deref(), + Some("event-missing-row") + ); + } + + #[test] + fn continuity_downgrade_failure_denies_memory_and_retains_cursor() { + let db = Db::open(std::path::Path::new(":memory:")).unwrap(); + let root = uid("shared"); + db.set_share_access(&root, Access::Editor).unwrap(); + db.with_conn(|conn| { + conn.execute_batch( + "CREATE TRIGGER reject_all_share_updates + BEFORE UPDATE ON share_access + BEGIN SELECT RAISE(FAIL, 'forced global downgrade failure'); END;", + )?; + Ok(()) + }) + .unwrap(); + let event = DriveEvent::ContinuityLost { + id: DriveEventId::from("event-2"), + }; + let denied_all = Cell::new(false); + + let applied = apply_access_downgrade(&db, &event, |root| { + denied_all.set(root.is_none()); + }); + assert!(applied.is_err()); + assert!( + denied_all.get(), + "all resident shared roots must fail closed immediately" + ); + assert!(acknowledge_applied_event(&db, &event, applied).is_err()); + assert!(db.get_event_cursor().unwrap().is_none()); + } +} diff --git a/crates/pdfs-fuse/src/devices.rs b/crates/pdfs-fuse/src/devices.rs index 5754e2f..018c429 100644 --- a/crates/pdfs-fuse/src/devices.rs +++ b/crates/pdfs-fuse/src/devices.rs @@ -489,6 +489,7 @@ impl Core { /// (devices.md Phase 3). pub(crate) fn fork_state(&self) -> Core { let mut fork = self.clone(); + let share_access = self.state.lock().share_access.clone(); fork.state = Arc::new(Mutex::new(State { entries: HashMap::new(), by_uid: HashMap::new(), @@ -497,6 +498,8 @@ impl Core { active_writes: HashMap::new(), handles: HashMap::new(), next_fh: 1, + access_changes: std::collections::HashSet::new(), + share_access, db: self.db.clone(), })); // A fresh inode space needs a fresh notification channel: this fork's diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index fff2e29..8b794e2 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -27,7 +27,7 @@ use std::time::Duration; use pdfs_core::cache::{Baseline, StagedWrite}; use pdfs_core::control::{ActivityKind, TransferDirection}; -use pdfs_core::db::{OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PendingOp}; +use pdfs_core::db::{OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PendingOp, RenameMeta}; use proton_drive_rs::Node; use proton_drive_rs::proton_sdk::ids::NodeUid; use tracing::{debug, error, info, warn}; @@ -40,6 +40,78 @@ use super::{ now_millis, now_secs, parse_node_uid, }; +const DRAIN_ACCESS_RECHECK: Duration = Duration::from_secs(5); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum DrainDisposition { + Applied, + AccessDeferred, +} + +fn pending_op_authorities(op: &PendingOp) -> Result, Box> { + let uid = || -> Result> { + parse_node_uid(&op.uid) + .ok_or_else(|| Box::::from("pending op has an unparseable uid")) + }; + let parent = || -> Result> { + let parent = op + .parent_uid + .as_deref() + .ok_or("pending op has no parent authority")?; + parse_node_uid(parent).ok_or_else(|| { + Box::::from("pending op has an unparseable parent") + }) + }; + match op.kind.as_str() { + OP_REVISION | OP_TRASH => Ok(vec![uid()?]), + OP_CREATE | OP_MKDIR => Ok(vec![parent()?]), + OP_RENAME => { + let mut authorities = vec![uid()?]; + if let Some(json) = op.meta_json.as_deref() { + let meta: RenameMeta = serde_json::from_str(json)?; + authorities.push( + parse_node_uid(&meta.original_parent_uid) + .ok_or("rename op has an unparseable original parent")?, + ); + } else { + // Rows written by 1.1.1 predate foreign shared mounts. Their + // source tree was necessarily owned, so the node plus desired + // destination are the complete available authority set. + } + authorities.push(parent()?); + authorities.dedup(); + Ok(authorities) + } + other => Err(format!("unknown pending op kind {other:?}").into()), + } +} + +fn run_authorized_drain( + op: &PendingOp, + mut require: impl FnMut(&NodeUid) -> Result<(), fuser::Errno>, + apply: impl FnOnce() -> Result<(), Box>, +) -> Result> { + // This check is the admission/linearization point. A downgrade that lands + // first defers the op; one observed afterwards does not revoke this admitted + // attempt. Every later queued operation rechecks. No permission lock is + // held across remote latency. + for authority in pending_op_authorities(op)? { + match require(&authority) { + Ok(()) => {} + Err(error) if error.code() == libc::EACCES => { + return Ok(DrainDisposition::AccessDeferred); + } + Err(error) => { + return Err( + format!("drain access check failed with errno {}", error.code()).into(), + ); + } + } + } + apply()?; + Ok(DrainDisposition::Applied) +} + impl Core { /// Drain the pending-op queue: the background half of every write /// (offline.md Phase 3). @@ -69,7 +141,30 @@ impl Core { self.wait_for_drain_work_or_due(); continue; }; - if let Err(e) = self.drain_op(&op) { + let outcome = run_authorized_drain( + &op, + |uid| self.require_uid_writable(uid), + || self.drain_op(&op), + ); + if matches!(outcome, Ok(DrainDisposition::AccessDeferred)) { + let next_attempt_at = now_millis() + DRAIN_ACCESS_RECHECK.as_millis() as i64; + if let Err(error) = self.db.defer_op_without_attempt(op.id, next_attempt_at) { + error!( + uid = %op.uid, + %error, + "deferring access-blocked pending operation failed" + ); + self.wait_for_drain_work(); + } else { + debug!( + uid = %op.uid, + next_attempt_at, + "pending operation deferred until access is writable" + ); + } + continue; + } + if let Err(e) = outcome { let attempts = op.attempts + 1; let err_str = e.to_string(); // Never infer that user data is disposable from an error string @@ -252,7 +347,7 @@ impl Core { } Err(e) => return Err(e.into()), } - self.db.delete_op(op.id)?; + self.db.complete_trash_op(op.id, &uid)?; self.own_sealed_revs.lock().remove(&uid); self.invalidate_trash(); self.log_activity(ActivityKind::Trash, &name, "trashed", true); @@ -996,8 +1091,184 @@ fn is_own_self_supersede(complete: bool, remote_rev: Option<&str>, own_rev: Opti #[cfg(test)] mod tests { use super::*; + use pdfs_core::Access; + use pdfs_core::db::Db; use proton_drive_rs::NodeKind; use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId}; + use std::cell::Cell; + + fn pending(kind: &str) -> PendingOp { + PendingOp { + id: 0, + kind: kind.to_string(), + uid: NodeUid::new(VolumeId::from("v"), LinkId::from("node")).to_string(), + parent_uid: Some(NodeUid::new(VolumeId::from("v"), LinkId::from("parent")).to_string()), + name: Some("name".to_string()), + blob_path: (kind == OP_REVISION).then(|| "/staging/blob".to_string()), + meta_json: None, + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + } + } + + #[test] + fn drain_authorities_match_each_operation_kind() { + for kind in [OP_REVISION, OP_TRASH] { + assert_eq!(pending_op_authorities(&pending(kind)).unwrap().len(), 1); + } + for kind in [OP_CREATE, OP_MKDIR] { + let authorities = pending_op_authorities(&pending(kind)).unwrap(); + assert_eq!(authorities.len(), 1); + assert_eq!(authorities[0].to_string(), "v~parent"); + } + let authorities = pending_op_authorities(&pending(OP_RENAME)).unwrap(); + assert_eq!(authorities.len(), 2); + assert_eq!(authorities[0].to_string(), "v~node"); + assert_eq!(authorities[1].to_string(), "v~parent"); + } + + fn folder_node(link: &str, parent: Option<&str>) -> Node { + Node { + uid: NodeUid::new(VolumeId::from("v"), LinkId::from(link)), + parent_uid: parent + .map(|parent| NodeUid::new(VolumeId::from("v"), LinkId::from(parent))), + kind: NodeKind::Folder, + name: link.to_string(), + creation_time: 0, + modification_time: 0, + trashed: false, + is_shared: false, + is_shared_publicly: false, + signature_email: None, + membership: None, + verification: Default::default(), + } + } + + #[test] + fn queued_cross_parent_rename_retains_viewer_source_authority_after_local_move() { + use std::sync::atomic::{AtomicU64, Ordering}; + + static N: AtomicU64 = AtomicU64::new(0); + let dir = std::env::temp_dir().join(format!( + "pdfs-drain-rename-access-{}-{}", + std::process::id(), + N.fetch_add(1, Ordering::Relaxed) + )); + std::fs::create_dir_all(&dir).unwrap(); + let db = Db::open(&dir.join("cache.db")).unwrap(); + let source = folder_node("source", None); + let destination = folder_node("destination", None); + let mut moved = file_node(1, 0, None); + moved.parent_uid = Some(destination.uid.clone()); + for node in [&source, &destination, &moved] { + db.upsert_node(node).unwrap(); + } + db.set_share_access(&source.uid, Access::Viewer).unwrap(); + db.set_share_access(&destination.uid, Access::Editor) + .unwrap(); + + let mut op = pending(OP_RENAME); + op.uid = moved.uid.to_string(); + op.parent_uid = Some(destination.uid.to_string()); + op.meta_json = Some( + serde_json::to_string(&RenameMeta { + original_parent_uid: source.uid.to_string(), + }) + .unwrap(), + ); + let authorities = pending_op_authorities(&op).unwrap(); + assert_eq!(authorities.len(), 3); + assert_eq!(authorities[1], source.uid); + + let applied = Cell::new(false); + let disposition = run_authorized_drain( + &op, + |uid| { + db.effective_node_access(uid) + .unwrap() + .filter(|access| access.writable()) + .map(|_| ()) + .ok_or(fuser::Errno::EACCES) + }, + || { + applied.set(true); + Ok(()) + }, + ) + .unwrap(); + assert_eq!(disposition, DrainDisposition::AccessDeferred); + assert!(!applied.get()); + assert_eq!( + db.effective_node_access(&moved.uid).unwrap(), + Some(Access::Editor), + "the node's optimistic destination ancestry is writable" + ); + drop(db); + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn viewer_deferred_drain_never_applies_or_consumes_an_attempt() { + use std::sync::atomic::{AtomicU64, Ordering}; + + static N: AtomicU64 = AtomicU64::new(0); + let dir = std::env::temp_dir().join(format!( + "pdfs-drain-access-{}-{}", + std::process::id(), + N.fetch_add(1, Ordering::Relaxed) + )); + std::fs::create_dir_all(&dir).unwrap(); + let db = Db::open(&dir.join("cache.db")).unwrap(); + let mut queued = pending(OP_REVISION); + queued.attempts = 2; + db.enqueue_op(&queued).unwrap(); + let op = db.pending_ops().unwrap().remove(0); + let before_count = db.pending_op_counts().unwrap(); + let applied = Cell::new(false); + + let disposition = run_authorized_drain( + &op, + |_| Err(fuser::Errno::EACCES), + || { + applied.set(true); + Ok(()) + }, + ) + .unwrap(); + assert_eq!(disposition, DrainDisposition::AccessDeferred); + assert!(!applied.get(), "no API surrogate may run while read-only"); + db.defer_op_without_attempt(op.id, 5_000).unwrap(); + + let retained = db.pending_ops().unwrap(); + let after_count = db.pending_op_counts().unwrap(); + assert_eq!(retained.len(), 1); + assert_eq!(retained[0].attempts, op.attempts); + assert_eq!(after_count.uploads, before_count.uploads); + assert_eq!(after_count.changes, before_count.changes); + drop(db); + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn owner_and_editor_drain_authorities_apply_normally() { + for access in [Access::Owner, Access::Editor] { + let applied = Cell::new(false); + let disposition = run_authorized_drain( + &pending(OP_REVISION), + |_| access.writable().then_some(()).ok_or(fuser::Errno::EACCES), + || { + applied.set(true); + Ok(()) + }, + ) + .unwrap(); + assert_eq!(disposition, DrainDisposition::Applied); + assert!(applied.get()); + } + } fn file_node(mtime: i64, size: i64, rev: Option<&str>) -> Node { Node { @@ -1019,6 +1290,7 @@ mod tests { is_shared: false, is_shared_publicly: false, signature_email: None, + membership: None, verification: Default::default(), } } diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs index 27e03d0..da8fbe3 100644 --- a/crates/pdfs-fuse/src/filesystem.rs +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -46,7 +46,7 @@ impl Filesystem for ProtonFs { } ) .then(|| (e.parent, e.uid.clone())); - (self.attr(ino.0, &e.node), provisional) + (self.attr(ino.0, &e.node, e.access), provisional) } None => { reply.error(Errno::ENOENT); @@ -79,7 +79,7 @@ impl Filesystem for ProtonFs { let attr = { let st = fs.core.state.lock(); match st.entries.get(&ino.0) { - Some(e) => fs.attr(ino.0, &e.node), + Some(e) => fs.attr(ino.0, &e.node, e.access), // Forgotten while we waited. None => attr, } @@ -110,10 +110,15 @@ impl Filesystem for ProtonFs { } fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) { + let write_requested = flags.acc_mode() != OpenAccMode::O_RDONLY; let (uid, base_mtime, base_size, base_revision_id) = { let mut st = self.core.state.lock(); match st.entries.get_mut(&ino.0) { Some(e) if e.node.is_file() => { + if write_requested && !e.writable() { + reply.error(Errno::EACCES); + return; + } e.open_count = e.open_count.saturating_add(1); ( e.uid.clone(), @@ -315,6 +320,10 @@ impl Filesystem for ProtonFs { reply: ReplyCreate, ) { let parent = parent.0; + if let Err(error) = self.core.require_writable(parent) { + reply.error(error); + return; + } let name = match fuse_name(name) { Ok(name) => name, Err(error) => { @@ -346,6 +355,10 @@ impl Filesystem for ProtonFs { _lock_owner: Option, reply: ReplyWrite, ) { + if let Err(error) = self.core.require_writable(ino.0) { + reply.error(error); + return; + } let fh = fh.0; // Stage the bytes straight into the scratch file (no base download): only // the untouched remainder is pulled from the remote, and only at commit. @@ -413,6 +426,10 @@ impl Filesystem for ProtonFs { _flags: Option, reply: ReplyAttr, ) { + if let Err(error) = self.core.require_writable(ino.0) { + reply.error(error); + return; + } // Only resizes change remote state; everything else (mode/owner/times) is // accepted and ignored so tools that chmod/utimes after writing succeed. if let Some(size) = size { @@ -476,6 +493,10 @@ impl Filesystem for ProtonFs { mode: i32, reply: ReplyEmpty, ) { + if let Err(error) = self.core.require_writable(ino.0) { + reply.error(error); + return; + } let new_end = offset.saturating_add(length); let keep_size = (mode & 1) != 0; // FALLOC_FL_KEEP_SIZE @@ -637,18 +658,7 @@ impl Filesystem for ProtonFs { ) { let (handle, unlinked_uid) = { let mut st = self.core.state.lock(); - let unlinked_uid = if let Some(entry) = st.entries.get_mut(&_ino.0) { - entry.open_count = entry.open_count.saturating_sub(1); - if entry.open_count == 0 && entry.unlinked { - let uid = entry.uid.clone(); - st.forget(&uid); - Some(uid) - } else { - None - } - } else { - None - }; + let unlinked_uid = release_unlinked_entry(&mut st, _ino.0); let ino = st.handles.remove(&fh.0); let h = ino.and_then(|i| { let aw = st.active_writes.get_mut(&i)?; @@ -679,6 +689,10 @@ impl Filesystem for ProtonFs { fn unlink(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) { let parent = parent.0; + if let Err(error) = self.core.require_writable(parent) { + reply.error(error); + return; + } let name_str = match fuse_name(name) { Ok(name) => name, Err(error) => { @@ -695,6 +709,10 @@ impl Filesystem for ProtonFs { fn rmdir(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) { let parent = parent.0; + if let Err(error) = self.core.require_writable(parent) { + reply.error(error); + return; + } let name_str = match fuse_name(name) { Ok(name) => name, Err(error) => { @@ -720,6 +738,10 @@ impl Filesystem for ProtonFs { reply: ReplyEntry, ) { let parent = parent.0; + if let Err(error) = self.core.require_writable(parent) { + reply.error(error); + return; + } let name = match fuse_name(name) { Ok(name) => name, Err(error) => { @@ -748,6 +770,14 @@ impl Filesystem for ProtonFs { ) { let parent = parent.0; let newparent = newparent.0; + if let Err(error) = self + .core + .require_writable(parent) + .and_then(|()| self.core.require_writable(newparent)) + { + reply.error(error); + return; + } let name = match fuse_name(name) { Ok(name) => name, Err(error) => { @@ -771,6 +801,19 @@ impl Filesystem for ProtonFs { }); } + fn access(&self, _req: &Request, ino: INodeNo, mask: AccessFlags, reply: ReplyEmpty) { + let state = self.core.state.lock(); + let Some(entry) = state.entries.get(&ino.0) else { + reply.error(Errno::ENOENT); + return; + }; + if !access_allowed(entry.node.is_folder(), entry.access, mask) { + reply.error(Errno::EACCES); + } else { + reply.ok(); + } + } + fn getxattr(&self, _req: &Request, ino: INodeNo, name: &OsStr, size: u32, reply: ReplyXattr) { // Always off the dispatch loop: a miss goes to the wire, and a lister // that stats a directory issues one of these per file per advertised @@ -935,6 +978,11 @@ pub struct ProtonFs { gid: u32, } +pub(crate) fn access_allowed(is_dir: bool, access: Access, mask: AccessFlags) -> bool { + (!mask.contains(AccessFlags::W_OK) || access.writable()) + && (!mask.contains(AccessFlags::X_OK) || is_dir) +} + impl ProtonFs { /// Build the filesystem rooted at `root` (the user's My Files folder). pub(super) fn new(core: Core, root: Node) -> Self { @@ -944,12 +992,14 @@ impl ProtonFs { warn!(uid = %root.uid, error = %e, "db upsert root failed"); } st.by_uid.insert(root.uid.clone(), ROOT_INO); + let access = st.access_for_node(ROOT_INO, &root); st.entries.insert( ROOT_INO, Entry { uid: root.uid.clone(), parent: ROOT_INO, node: root, + access, lookup_count: 1, open_count: 0, unlinked: false, @@ -991,7 +1041,7 @@ impl ProtonFs { } ) .then(|| (e.parent, e.uid.clone())); - (ino, self.attr(ino, &e.node), provisional) + (ino, self.attr(ino, &e.node, e.access), provisional) }) }) }) @@ -1023,7 +1073,7 @@ impl ProtonFs { let st = self.core.state.lock(); st.entries .get(&ino) - .map_or(attr, |e| self.attr(ino, &e.node)) + .map_or(attr, |e| self.attr(ino, &e.node, e.access)) }; reply.entry(&TTL, &attr, Generation(0)); } @@ -1065,11 +1115,12 @@ impl ProtonFs { reply.ok(); } - fn attr(&self, ino: u64, node: &Node) -> FileAttr { - let (kind, perm) = match node.kind { - NodeKind::Folder => (FileType::Directory, 0o755), - NodeKind::File { .. } => (FileType::RegularFile, 0o644), + fn attr(&self, ino: u64, node: &Node, access: Access) -> FileAttr { + let kind = match node.kind { + NodeKind::Folder => FileType::Directory, + NodeKind::File { .. } => FileType::RegularFile, }; + let perm = perm_bits(node.is_folder(), access); let size = node_size(node); let mtime = unix_secs(node.modification_time); let crtime = unix_secs(node.creation_time); @@ -1097,6 +1148,10 @@ impl ProtonFs { /// subtrees, so an `rmdir` of a non-empty dir behaves the same). /// The body of [`Filesystem::create`], run off the dispatch loop. fn serve_create(&self, parent: u64, name: &str, reply: ReplyCreate) { + if let Err(error) = self.core.require_writable(parent) { + reply.error(error); + return; + } if let Err(e) = self.core.ensure_children(parent) { reply.error(e); return; @@ -1128,6 +1183,10 @@ impl ProtonFs { let node = if !transient && self.core.online.load(Ordering::Relaxed) && !is_local_uid(&parent_uid) { + if let Err(error) = self.core.require_uid_writable(&parent_uid) { + reply.error(error); + return; + } // Create an empty file on the remote so it has a real uid immediately; // written bytes are buffered and sealed as a new revision on close. let new_uid = match self.core.rt.block_on(self.core.client.upload_file( @@ -1211,7 +1270,10 @@ impl ProtonFs { }); aw.open_count += 1; st.handles.insert(fh, ino); - let attr = self.attr(ino, &st.entries.get(&ino).unwrap().node); + let entry = st.entries.get(&ino).unwrap(); + let attr = self.attr(ino, &entry.node, entry.access); + drop(st); + self.core.flush_access_changes(); reply.created( &TTL, &attr, @@ -1223,6 +1285,10 @@ impl ProtonFs { /// The body of [`Filesystem::mkdir`], run off the dispatch loop. fn serve_mkdir(&self, parent: u64, name: &str, reply: ReplyEntry) { + if let Err(error) = self.core.require_writable(parent) { + reply.error(error); + return; + } if let Err(e) = self.core.ensure_children(parent) { reply.error(e); return; @@ -1245,6 +1311,10 @@ impl ProtonFs { // queued — the folder becomes a placeholder that the drain turns into a // real one (offline.md Phase 3b). let node = if self.core.online.load(Ordering::Relaxed) && !is_local_uid(&parent_uid) { + if let Err(error) = self.core.require_uid_writable(&parent_uid) { + reply.error(error); + return; + } let new_uid = match self .core @@ -1292,12 +1362,19 @@ impl ProtonFs { warn!(%uid, error = %e, "db set_listed(true) failed for local folder"); } } - let attr = self.attr(ino, &st.entries.get(&ino).unwrap().node); + let entry = st.entries.get(&ino).unwrap(); + let attr = self.attr(ino, &entry.node, entry.access); + drop(st); + self.core.flush_access_changes(); reply.entry(&TTL, &attr, Generation(0)); } /// The body of [`Filesystem::unlink`], run off the dispatch loop. fn serve_unlink(&self, parent: u64, name_str: &str, reply: ReplyEmpty) { + if let Err(error) = self.core.require_writable(parent) { + reply.error(error); + return; + } if let Ok((ino, _)) = self.core.lookup_child(parent, name_str) { let st = self.core.state.lock(); if let Some(entry) = st.entries.get(&ino) @@ -1312,6 +1389,10 @@ impl ProtonFs { /// The body of [`Filesystem::rmdir`], run off the dispatch loop. fn serve_rmdir(&self, parent: u64, name_str: &str, reply: ReplyEmpty) { + if let Err(error) = self.core.require_writable(parent) { + reply.error(error); + return; + } if let Ok((ino, _)) = self.core.lookup_child(parent, name_str) { { let st = self.core.state.lock(); @@ -1348,10 +1429,18 @@ impl ProtonFs { reply: ReplyEmpty, ) { let was_unlinked = unlinked_uid.is_some(); - if let Some(uid) = unlinked_uid { - self.core.discard_queued_ops(&uid); - self.core.cache.evict(&uid); - self.core.evict_reader(&uid); + if let Some(uid) = unlinked_uid + && release_can_discard_unlinked(&self.core.db, &uid) + { + if let Err(error) = self.core.discard_queued_ops(&uid) { + warn!(%uid, ?error, "queued-op cleanup failed; retaining unlinked state"); + } else { + if let Err(error) = self.core.db.delete_node(&uid) { + warn!(%uid, %error, "deleting released unlinked node state failed"); + } + self.core.cache.evict(&uid); + self.core.evict_reader(&uid); + } } // Hand the bytes to the queue rather than uploading them here: the // scratch file is the only copy of what was just written, and blocking @@ -1366,6 +1455,7 @@ impl ProtonFs { { warn!(path = %h.path.display(), error = %e, "discarding unlinked scratch file failed"); } + self.core.cache.clear_scratch_durable(&h.path); reply.ok(); } (Some(h), false) => match self.core.queue_revision(&h) { @@ -1386,6 +1476,14 @@ impl ProtonFs { flags: RenameFlags, reply: ReplyEmpty, ) { + if let Err(error) = self + .core + .require_writable(parent) + .and_then(|()| self.core.require_writable(newparent)) + { + reply.error(error); + return; + } let (ino, uid) = match self.core.lookup_child(parent, name) { Ok(x) => x, Err(e) => { @@ -1413,11 +1511,11 @@ impl ProtonFs { reply.error(e); return; } - let new_parent_uid = { + let (old_parent_uid, new_parent_uid) = { let st = self.core.state.lock(); - match st.entries.get(&newparent) { - Some(e) => e.uid.clone(), - None => { + match (st.entries.get(&parent), st.entries.get(&newparent)) { + (Some(old), Some(new)) => (old.uid.clone(), new.uid.clone()), + _ => { reply.error(Errno::ENOENT); return; } @@ -1446,7 +1544,7 @@ impl ProtonFs { return; } }; - if let Some((victim_ino, victim_uid)) = &victim { + let replacement_authorized = if let Some((victim_ino, victim_uid)) = &victim { if flags.contains(RenameFlags::RENAME_NOREPLACE) { reply.error(Errno::EEXIST); return; @@ -1490,11 +1588,23 @@ impl ProtonFs { reply.error(e); return; } + if let Err(error) = require_rename_access( + |authority| self.core.require_uid_writable(authority), + &uid, + &old_parent_uid, + &new_parent_uid, + ) { + reply.error(error); + return; + } if let Err(e) = self.core.remove_replaced(victim_uid, newname) { reply.error(e); return; } - } + true + } else { + false + }; // A node whose own creation is still queued has no server-side identity // to rename: the queued op *is* the node, so rewriting its target is the // whole rename. Nothing reaches the API, which is why this works offline @@ -1557,10 +1667,26 @@ impl ProtonFs { newparent != parent, newname != name, ) { - match self - .core - .queue_rename(ino, &uid, newparent, &new_parent_uid, newname) - { + let queued = if replacement_authorized { + self.core.queue_rename_authorized( + ino, + &uid, + &old_parent_uid, + newparent, + &new_parent_uid, + newname, + ) + } else { + self.core.queue_rename( + ino, + &uid, + &old_parent_uid, + newparent, + &new_parent_uid, + newname, + ) + }; + match queued { Ok(()) => { // Send the rename response first. Kernel-side rename cache // updates happen while processing that response; notifying @@ -1588,6 +1714,16 @@ impl ProtonFs { // A failure past this point has already trashed any node the rename was // replacing, so put it back rather than leaving the caller with neither // the source moved nor the destination intact. + if let Err(error) = require_rename_access( + |authority| self.core.require_uid_writable(authority), + &uid, + &old_parent_uid, + &new_parent_uid, + ) { + self.core.restore_replaced(victim.as_ref(), newname); + reply.error(error); + return; + } if newname != name && let Err(e) = self .core @@ -1655,7 +1791,7 @@ impl ProtonFs { let st = self.core.state.lock(); match st.entries.get(&ino) { Some(e) => { - let attr = self.attr(ino, &e.node); + let attr = self.attr(ino, &e.node, e.access); reply.attr(&TTL, &attr); } None => reply.error(Errno::ENOENT), @@ -1663,6 +1799,10 @@ impl ProtonFs { } fn trash_child(&self, parent: u64, name: &str, reply: ReplyEmpty) { + if let Err(error) = self.core.require_writable(parent) { + reply.error(error); + return; + } let (ino, uid) = match self.core.lookup_child(parent, name) { Ok(x) => x, Err(e) => { @@ -1681,8 +1821,9 @@ impl ProtonFs { // it just means its queued creation is no longer wanted. This works // offline, which the remote path below cannot (offline.md Phase 3b). if is_local_uid(&uid) { - if !open_now { - self.core.discard_queued_ops(&uid); + if !open_now && let Err(error) = self.core.discard_queued_ops(&uid) { + reply.error(error); + return; } self.core.state.lock().forget_or_unlink(&uid); debug!(%uid, name, "deleted a node that had not been created remotely yet"); @@ -1694,14 +1835,15 @@ impl ProtonFs { // command returns (offline.md Phase 3b). if !self.core.online.load(Ordering::Relaxed) { match self.core.queue_trash(&uid, name) { - Ok(()) => { - self.core.state.lock().forget_or_unlink(&uid); - reply.ok(); - } + Ok(()) => reply.ok(), Err(e) => reply.error(e), } return; } + if let Err(e) = self.core.require_uid_writable(&uid) { + reply.error(e); + return; + } if let Err(e) = self .core .rt @@ -1718,7 +1860,9 @@ impl ProtonFs { self.core.hidden.lock().insert(uid.clone()); self.core.state.lock().forget_or_unlink(&uid); if !open_now { - self.core.discard_queued_ops(&uid); + if let Err(error) = self.core.discard_queued_ops(&uid) { + warn!(%uid, ?error, "remote trash landed but queued-op cleanup failed"); + } self.core.cache.evict(&uid); self.core.evict_reader(&uid); } diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index fc5270c..3895e49 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -47,9 +47,9 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use fuser::ReplyXattr; use fuser::{ - BackgroundSession, BsdFileFlags, Config, CopyFileRangeFlags, Errno, FileAttr, FileHandle, - FileType, Filesystem, FopenFlags, Generation, INodeNo, IoctlFlags, LockOwner, MountOption, - Notifier, OpenAccMode, OpenFlags, RenameFlags, ReplyAttr, ReplyCreate, ReplyData, + AccessFlags, BackgroundSession, BsdFileFlags, Config, CopyFileRangeFlags, Errno, FileAttr, + FileHandle, FileType, Filesystem, FopenFlags, Generation, INodeNo, IoctlFlags, LockOwner, + MountOption, Notifier, OpenAccMode, OpenFlags, RenameFlags, ReplyAttr, ReplyCreate, ReplyData, ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyIoctl, ReplyLseek, ReplyOpen, ReplyWrite, Request, Session, TimeOrNow, WriteFlags, }; @@ -62,12 +62,12 @@ use pdfs_core::control::{ }; use pdfs_core::db::{ Db, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PARK_UNTIL, PendingOp, - StoredNode, StoredSyncFolder, StoredTrash, + RenameMeta, StoredNode, StoredSyncFolder, StoredTrash, }; use pdfs_core::localindex; use pdfs_core::search::relevance_score; use pdfs_core::syncignore::is_transient_name; -use pdfs_core::{CoreError, CoreResult}; +use pdfs_core::{Access, CoreError, CoreResult, perm_bits}; use proton_drive_rs::proton_sdk::api::ResponseCode; use proton_drive_rs::proton_sdk::error::ProtonError; use proton_drive_rs::proton_sdk::ids::{DriveEventId, LinkId, NodeUid, VolumeId}; @@ -512,8 +512,13 @@ impl Core { /// to all of them is correct because a uid is unique across mounts — at most /// one state has an entry for it, and the rest are no-ops. fn for_each_state(&self, mut apply: impl FnMut(&mut State)) { - for (state, _) in self.states.live() { - apply(&mut state.lock()); + for (state, notifier) in self.states.live() { + let changed = { + let mut state = state.lock(); + apply(&mut state); + state.take_access_changes() + }; + notify_access_changes(notifier.get(), &changed); } } @@ -529,6 +534,131 @@ impl Core { } } + /// Publish access changes accumulated by an intern/root refresh. The State + /// lock is released before notifying the kernel because notifier callbacks + /// may synchronously provoke more filesystem work. + fn flush_access_changes(&self) { + let changed = self.state.lock().take_access_changes(); + notify_access_changes(self.notifier.get(), &changed); + } + + fn require_writable(&self, ino: u64) -> Result<(), Errno> { + self.state.lock().require_writable(ino) + } + + /// Mutation admission point. Persisted and every resident authority must + /// agree that the node is writable. A check completed before a downgrade is + /// admitted; later queued attempts are caught by drain. No global lock is + /// held across a network call. + /// + /// The intersection closes both halves of an access-refresh race: a DB + /// downgrade cannot be hidden by stale live state, and a live downgrade + /// cannot be hidden by a restored DB row. Missing UIDs fail closed. + fn require_uid_writable(&self, uid: &NodeUid) -> Result<(), Errno> { + let mut live_access = Vec::new(); + { + let state = self.state.lock(); + if let Some(access) = state.access_by_uid(uid) { + live_access.push(access); + } + } + for (state, _) in self.states.live() { + if Arc::ptr_eq(&state, &self.state) { + continue; + } + if let Some(access) = state.lock().access_by_uid(uid) { + live_access.push(access); + } + } + require_uid_access(&self.db, uid, &live_access) + } +} + +fn require_uid_access(db: &Db, uid: &NodeUid, live_access: &[Access]) -> Result<(), Errno> { + let persisted = match db.effective_node_access(uid) { + Ok(Some(access)) => access, + Ok(None) => return Err(Errno::EACCES), + Err(error) => { + error!(%uid, %error, "db effective access lookup failed"); + return Err(Errno::EACCES); + } + }; + (persisted.writable() && live_access.iter().all(|access| access.writable())) + .then_some(()) + .ok_or(Errno::EACCES) +} + +fn notify_access_changes(notifier: Option<&Notifier>, changed: &[u64]) { + let Some(notifier) = notifier else { return }; + for &ino in changed { + let _ = notifier.inval_inode(INodeNo(ino), 0, 0); + } +} + +fn require_rename_access( + require: impl Fn(&NodeUid) -> Result<(), Errno>, + uid: &NodeUid, + old_parent_uid: &NodeUid, + new_parent_uid: &NodeUid, +) -> Result<(), Errno> { + for authority in [uid, old_parent_uid, new_parent_uid] { + require(authority)?; + } + Ok(()) +} + +fn require_node_parent_access( + require: impl Fn(&NodeUid) -> Result<(), Errno>, + uid: &NodeUid, + parent_uid: &NodeUid, +) -> Result<(), Errno> { + require(uid)?; + require(parent_uid) +} + +fn preserve_on_access_denied( + access: Result<(), Errno>, + accepted_bytes: bool, + preserve: impl FnOnce(), +) -> Result<(), Errno> { + if let Err(error) = access { + if error.code() == libc::EACCES && accepted_bytes { + preserve(); + } + return Err(error); + } + Ok(()) +} + +/// Retire the last resident reference to an unlinked inode without touching its +/// persisted node row. The namespace path already chose whether that row is +/// disposable or retained as queued-trash authority. +fn release_unlinked_entry(state: &mut State, ino: u64) -> Option { + let entry = state.entries.get_mut(&ino)?; + entry.open_count = entry.open_count.saturating_sub(1); + if entry.open_count != 0 || !entry.unlinked { + return None; + } + let uid = entry.uid.clone(); + state.forget_mem(&uid); + Some(uid) +} + +fn release_must_retain_queued_trash(db: &Db, uid: &NodeUid) -> pdfs_core::Result { + db.has_pending_op(&uid.to_string(), OP_TRASH) +} + +fn release_can_discard_unlinked(db: &Db, uid: &NodeUid) -> bool { + match release_must_retain_queued_trash(db, uid) { + Ok(retain) => !retain, + Err(error) => { + warn!(%uid, %error, "queued-trash lookup failed; retaining unlinked state"); + false + } + } +} + +impl Core { /// Record that this daemon just changed `uid` on the remote, so the event /// feed's echo of that change is recognised instead of re-applied. Called /// from the drain once the remote has actually accepted the change and the @@ -575,11 +705,15 @@ impl Core { .map(|(uid, pr)| (uid.clone(), pr.meta.len)) .collect() }; + // `hydrate_pending` runs first and restores queued-trash tombstones. + // Their node rows remain persisted as drain-time access authority, but + // must not reappear in the mounted namespace after a restart. + let hidden = self.hidden.lock().clone(); let mut st = self.state.lock(); // Pass 1: assign a stable inode to every uid (root is already mapped). for sn in &stored { - if st.by_uid.contains_key(&sn.node.uid) { + if hidden.contains(&sn.node.uid) || st.by_uid.contains_key(&sn.node.uid) { continue; } let ino = st.next_ino; @@ -592,6 +726,9 @@ impl Core { let mut listed_dirs: Vec = Vec::new(); for sn in stored { let StoredNode { mut node, listed } = sn; + if hidden.contains(&node.uid) { + continue; + } let Some(&ino) = st.by_uid.get(&node.uid) else { continue; }; @@ -623,12 +760,14 @@ impl Core { uid, parent, node, + access: pdfs_core::Access::Unknown, lookup_count: 1, open_count: 0, unlinked: false, }, ); } + st.hydrate_access(); // Pass 3: rebuild child listings for fully-enumerated folders. The root // is its own parent (that is what `..` resolves to), so it would match @@ -671,6 +810,9 @@ impl Core { let _ = self.db.delete_op(op.id); continue; }; + if op.kind == OP_TRASH { + self.hidden.lock().insert(uid.clone()); + } restored += 1; // Only a revision must have a blob. A create carries none until // something is written to it (`touch` offline is a legitimate op @@ -916,6 +1058,7 @@ impl Core { } st.children.insert(ino, child_inos); drop(st); + self.flush_access_changes(); // Rows persisted from a cheap enumeration whose upgrade never // ran (a restart in between, say) still owe their real sizes. self.spawn_size_upgrade(ino, needs_size); @@ -987,6 +1130,7 @@ impl Core { warn!(%folder_uid, error = %e, "db set_listed(true) failed"); } drop(st); + self.flush_access_changes(); self.spawn_size_upgrade(ino, needs_size); Ok(()) } @@ -1258,6 +1402,25 @@ impl Core { .map_err(|e| self.errno_error(e, &format!("could not resolve {}", rel.display()))) } + fn source_parent_uid(&self, ino: u64, rel: &Path) -> CoreResult { + let state = self.state.lock(); + let entry = state + .entries + .get(&ino) + .ok_or_else(|| CoreError::not_found(format!("could not resolve {}", rel.display())))?; + entry + .node + .parent_uid + .clone() + .or_else(|| { + state + .entries + .get(&entry.parent) + .map(|parent| parent.uid.clone()) + }) + .ok_or_else(|| CoreError::invalid("the mount root cannot be mutated")) + } + /// Classify a failure that arrived as an `Errno`. /// /// The internal paths speak `Errno` because they also serve FUSE, where a @@ -1324,6 +1487,23 @@ impl Core { /// The scratch file is *moved* into staging, never copied: it is the only /// copy of what the user wrote. fn queue_revision(&self, h: &WriteHandle) -> Result<(), Errno> { + // Closing an untouched writable handle is not a mutation. It remains + // valid after a downgrade and only owes cleanup of its empty scratch. + if !h.dirty { + self.cache.clear_scratch_durable(&h.path); + let _ = std::fs::remove_file(&h.path); + return Ok(()); + } + preserve_on_access_denied(self.require_uid_writable(&h.uid), h.dirty, || { + let meta = self.recovery_meta(h); + self.stage_orphaned_write( + &h.uid, + h.ino, + &h.path, + &meta, + "access changed before the write could be queued", + ); + })?; // The handle is being retired either way, so any durability sidecar an // `fsync` left has done its job: from here the bytes are tracked as a // staged write and a queued op, and a sidecar outliving them would offer @@ -1334,10 +1514,6 @@ impl Core { let _ = std::fs::remove_file(&h.path); return Ok(()); } - if !h.dirty { - let _ = std::fs::remove_file(&h.path); - return Ok(()); - } // Materialize the full content now if we can. A complete blob is // uploadable without the network and, crucially, lets a later write to // the same file supersede this one safely. @@ -1392,6 +1568,35 @@ impl Core { Ok(()) } + /// Describe the locally authored ranges without gap-filling them. + /// + /// Used when a write was accepted by the kernel but a queue-time access + /// check now denies it. Building this metadata is side-effect free, so the + /// permission guard still precedes cache, database, and tree mutation. + fn recovery_meta(&self, h: &WriteHandle) -> StagedWrite { + let authored: Vec<(u64, u64)> = h + .written + .segments(0, h.len) + .into_iter() + .filter(|&(_, _, authored)| authored) + .map(|(start, end, _)| (start, end)) + .collect(); + StagedWrite { + uid: h.uid.to_string(), + len: h.len, + base_size: h.base_size, + base_mtime: h.base_mtime, + complete: authored == [(0, h.len)], + authored, + based_on: self.remote_baseline( + &h.uid, + h.base_mtime, + h.base_size, + h.base_revision_id.clone(), + ), + } + } + /// The remote revision a change to `uid` is being made against, for /// [`StagedWrite::based_on`]. /// @@ -1442,6 +1647,15 @@ impl Core { src: &Path, meta: StagedWrite, ) -> Result<(), Errno> { + preserve_on_access_denied(self.require_uid_writable(uid), true, || { + self.stage_orphaned_write( + uid, + ino, + src, + &meta, + "access changed before the staged write could be queued", + ); + })?; // An incomplete blob's gaps refer to the *remote* base. If an earlier // write to this file is still queued, the remote no longer holds that // base — the previous staged blob does — so superseding it would fill @@ -1450,7 +1664,13 @@ impl Core { // Only reachable offline, editing in place a file whose previous edit // has not drained and whose base is not cached. if !meta.complete && self.pending.lock().contains_key(uid) { - self.stage_orphaned_write(uid, ino, src, &meta); + self.stage_orphaned_write( + uid, + ino, + src, + &meta, + "write overlaps an undrained incomplete edit", + ); return Err(Errno::EIO); } let path = self.cache.stage_write(&meta, src).map_err(|e| { @@ -1552,6 +1772,7 @@ impl Core { None => return Err(Errno::ENOENT), } }; + self.require_uid_writable(&uid)?; if size == base_size { return Ok(()); } @@ -1645,6 +1866,7 @@ impl Core { is_dir: bool, hold: bool, ) -> Result { + self.require_uid_writable(parent_uid)?; let uid = mint_local_uid(); let op = PendingOp { id: 0, @@ -1690,10 +1912,56 @@ impl Core { &self, ino: u64, uid: &NodeUid, + old_parent_uid: &NodeUid, new_parent_ino: u64, new_parent_uid: &NodeUid, new_name: &str, ) -> Result<(), Errno> { + self.require_uid_writable(uid)?; + self.require_uid_writable(old_parent_uid)?; + self.require_uid_writable(new_parent_uid)?; + self.queue_rename_authorized( + ino, + uid, + old_parent_uid, + new_parent_ino, + new_parent_uid, + new_name, + ) + } + + /// Enqueue a rename whose access was linearized immediately before a + /// replacement victim was removed. Rechecking here could reject after that + /// destructive half had already completed. + fn queue_rename_authorized( + &self, + ino: u64, + uid: &NodeUid, + old_parent_uid: &NodeUid, + new_parent_ino: u64, + new_parent_uid: &NodeUid, + new_name: &str, + ) -> Result<(), Errno> { + let original_parent_uid = match self.db.pending_op_meta(&uid.to_string(), OP_RENAME) { + Ok(Some(json)) => serde_json::from_str::(&json) + .map(|meta| meta.original_parent_uid) + .map_err(|error| { + error!(%uid, %error, "queued rename authority metadata is invalid"); + Errno::EIO + })?, + Ok(None) => old_parent_uid.to_string(), + Err(error) => { + error!(%uid, %error, "reading queued rename authority failed"); + return Err(Errno::EIO); + } + }; + let meta_json = serde_json::to_string(&RenameMeta { + original_parent_uid, + }) + .map_err(|error| { + error!(%uid, %error, "serializing queued rename authority failed"); + Errno::EIO + })?; let op = PendingOp { id: 0, kind: OP_RENAME.to_string(), @@ -1701,7 +1969,7 @@ impl Core { parent_uid: Some(new_parent_uid.to_string()), name: Some(new_name.to_string()), blob_path: None, - meta_json: None, + meta_json: Some(meta_json), created_at: now_millis(), attempts: 0, last_error: None, @@ -1729,24 +1997,18 @@ impl Core { /// un-uploaded file means, and the alternative (upload it, then trash it) is /// worse in every way. fn queue_trash(&self, uid: &NodeUid, name: &str) -> Result<(), Errno> { - self.discard_queued_ops(uid); - let op = PendingOp { - id: 0, - kind: OP_TRASH.to_string(), - uid: uid.to_string(), - parent_uid: None, - name: Some(name.to_string()), - blob_path: None, - meta_json: None, - created_at: now_millis(), - attempts: 0, - last_error: None, - next_attempt_at: 0, - }; - self.db.enqueue_op(&op).map_err(|e| { - error!(%uid, error = %e, "queueing trash failed"); - Errno::EIO - })?; + self.require_uid_writable(uid)?; + let (_, blobs) = self + .db + .replace_ops_with_trash(&uid.to_string(), name, now_millis()) + .map_err(|e| { + error!(%uid, error = %e, "queueing trash failed"); + Errno::EIO + })?; + for blob in blobs { + self.cache.discard_staged(Path::new(&blob)); + } + self.pending.lock().remove(uid); self.hidden.lock().insert(uid.clone()); // Withdrawn from every mount, not just the one the unlink came through: // a sync folder maps a remote folder that also exists under My Files, so @@ -1754,7 +2016,7 @@ impl Core { // one would go on serving a file the user just trashed // (`docs/BUGS.md` B74). self.for_each_state(|st| { - st.forget(uid); + st.unlink_mem(uid); }); self.cache.evict(uid); self.evict_reader(uid); @@ -1764,16 +2026,16 @@ impl Core { } /// Drop every op queued against a node, and the staged bytes they own. - fn discard_queued_ops(&self, uid: &NodeUid) { - match self.db.delete_ops_for_uid(&uid.to_string()) { - Ok(blobs) => { - for blob in blobs { - self.cache.discard_staged(Path::new(&blob)); - } - } - Err(e) => error!(%uid, error = %e, "dropping queued ops failed"), + fn discard_queued_ops(&self, uid: &NodeUid) -> Result<(), Errno> { + let blobs = self.db.delete_ops_for_uid(&uid.to_string()).map_err(|e| { + error!(%uid, error = %e, "dropping queued ops failed"); + Errno::EIO + })?; + for blob in blobs { + self.cache.discard_staged(Path::new(&blob)); } self.pending.lock().remove(uid); + Ok(()) } /// Nudge the drain worker to re-examine the queue now. @@ -1785,13 +2047,22 @@ impl Core { /// Keep a write we cannot safely queue, so the bytes are recoverable even /// though the caller is getting an error. See [`Core::queue_revision`]. - fn stage_orphaned_write(&self, uid: &NodeUid, ino: u64, src: &Path, meta: &StagedWrite) { - match self.cache.stage_write(meta, src) { + fn stage_orphaned_write( + &self, + uid: &NodeUid, + ino: u64, + src: &Path, + meta: &StagedWrite, + reason: &str, + ) { + match self.cache.preserve_write(meta, src) { Ok(staged) => { + self.cache.clear_scratch_durable(src); error!( %uid, staged = %staged.display(), - "cannot queue write over an undrained edit; bytes kept in staging" + %reason, + "cannot queue write; bytes kept in staging" ); let name = { let st = self.state.lock(); @@ -1803,20 +2074,23 @@ impl Core { self.log_activity( ActivityKind::Upload, &name, - format!("write not queued; changes kept at {}", staged.display()), + format!( + "write not queued ({reason}); changes kept at {}", + staged.display() + ), false, ); } Err(e) => { - // `src` is the only known copy of bytes the kernel already - // accepted. A failure to publish them into staging must never - // turn into permission to delete them: leave the scratch file - // in place so startup recovery or a human can still salvage it. + let surviving = e.surviving_path.clone(); + let marker = e.marker_path.clone(); error!( %uid, error = %e, - scratch = %src.display(), - "staging write failed; retaining the only scratch copy" + surviving = %surviving.display(), + marker = ?marker, + %reason, + "write rescue was only partially published; bytes retained" ); let name = { let st = self.state.lock(); @@ -1829,8 +2103,8 @@ impl Core { ActivityKind::Upload, &name, format!( - "write not queued and could not be staged; recovery copy retained at {}", - src.display() + "write not queued ({reason}); changes retained at {}", + surviving.display() ), false, ); @@ -2368,7 +2642,15 @@ impl Core { if new_name.is_empty() || new_name.contains('/') { return Err(CoreError::invalid(format!("invalid name: {new_name:?}"))); } - let (_ino, uid) = self.resolve(rel)?; + let (ino, uid) = self.resolve(rel)?; + let old_parent_uid = self.source_parent_uid(ino, rel)?; + require_rename_access( + |authority| self.require_uid_writable(authority), + &uid, + &old_parent_uid, + &old_parent_uid, + ) + .map_err(|error| self.errno_error(error, "rename access"))?; self.rt .block_on(self.client.rename_node(&uid, new_name, None)) .map_err(|e| CoreError::from_api(&e, "rename"))?; @@ -2385,10 +2667,18 @@ impl Core { /// mountpoint-relative. Forgets the node and invalidates both the source and /// destination listings so each re-enumerates on next access. fn move_to(&self, rel: &Path, new_parent_rel: &Path) -> CoreResult { - let (_ino, uid) = self.resolve(rel)?; + let (ino, uid) = self.resolve(rel)?; + let old_parent_uid = self.source_parent_uid(ino, rel)?; let (pino, new_parent_uid) = self .resolve_path(new_parent_rel) .map_err(|e| self.errno_error(e, "resolve new parent"))?; + require_rename_access( + |authority| self.require_uid_writable(authority), + &uid, + &old_parent_uid, + &new_parent_uid, + ) + .map_err(|error| self.errno_error(error, "move access"))?; self.rt .block_on(self.client.move_node(&uid, &new_parent_uid)) .map_err(|e| CoreError::from_api(&e, "move"))?; @@ -2406,7 +2696,14 @@ impl Core { /// Trash a file or folder. `rel` is mountpoint-relative. Forgets the node, /// evicts any cached content, and invalidates the parent listing. fn delete(&self, rel: &Path) -> CoreResult { - let (_ino, uid) = self.resolve(rel)?; + let (ino, uid) = self.resolve(rel)?; + let parent_uid = self.source_parent_uid(ino, rel)?; + require_node_parent_access( + |authority| self.require_uid_writable(authority), + &uid, + &parent_uid, + ) + .map_err(|error| self.errno_error(error, "trash access"))?; self.rt .block_on(self.client.trash_nodes(std::slice::from_ref(&uid))) .map_err(|e| CoreError::from_api(&e, "trash"))?; @@ -2436,13 +2733,14 @@ impl Core { /// it works offline. fn remove_replaced(&self, uid: &NodeUid, name: &str) -> Result<(), Errno> { if is_local_uid(uid) { - self.discard_queued_ops(uid); + self.discard_queued_ops(uid)?; self.for_each_state(|st| { st.forget(uid); }); debug!(%uid, name, "replaced a node whose create was still queued"); return Ok(()); } + self.require_uid_writable(uid)?; if let Err(e) = self .rt .block_on(self.client.trash_nodes(std::slice::from_ref(uid))) @@ -2451,7 +2749,10 @@ impl Core { self.log_activity(ActivityKind::Trash, name, e.to_string(), false); return Err(Errno::EIO); } - self.discard_queued_ops(uid); + if let Err(error) = self.discard_queued_ops(uid) { + error!(%uid, "remote replacement landed but queued-op cleanup failed"); + return Err(error); + } // The node was trashed on the server; withdraw it from every inode // space that had it, not only the mount the rename came through. self.for_each_state(|st| { @@ -2780,6 +3081,8 @@ impl Core { let (pino, parent_uid) = self.resolve(parent_rel)?; self.ensure_children(pino) .map_err(|e| self.errno_error(e, "enumerate"))?; + self.require_uid_writable(&parent_uid) + .map_err(|error| self.errno_error(error, "create folder access"))?; let new_uid = self .rt .block_on( @@ -2797,6 +3100,8 @@ impl Core { { kids.push(ino); } + drop(st); + self.flush_access_changes(); Ok(name.to_string()) } // ---- activity log ----------------------------------------------------- @@ -3084,6 +3389,7 @@ fn local_node(uid: NodeUid, parent_uid: NodeUid, name: String, is_dir: bool) -> is_shared: false, is_shared_publicly: false, signature_email: None, + membership: None, // Nothing signed it: it has never been near the crypto layer. verification: Default::default(), } @@ -3671,6 +3977,7 @@ mod pending_size_tests { is_shared: false, is_shared_publicly: false, signature_email: None, + membership: None, verification: Default::default(), } } @@ -3727,11 +4034,16 @@ mod pending_size_tests { #[cfg(test)] mod tests { use super::{ - HashMap, Intervals, PendingRevision, SELF_CHANGE_TTL_MS, StateRegistry, conflict_name, - copy_pending_for_truncate, fuse_name, is_stale_mount, node_visible, note_self_change, - parse_node_uid, rename_needs_queue, take_self_change, + Access, AccessFlags, Errno, HashMap, Intervals, PendingRevision, SELF_CHANGE_TTL_MS, + StateRegistry, conflict_name, copy_pending_for_truncate, fuse_name, is_stale_mount, + node_visible, note_self_change, parse_node_uid, preserve_on_access_denied, + release_can_discard_unlinked, release_must_retain_queued_trash, release_unlinked_entry, + rename_needs_queue, require_node_parent_access, require_rename_access, require_uid_access, + take_self_change, }; + use crate::filesystem::access_allowed; use pdfs_core::cache::{Baseline, StagedWrite}; + use pdfs_core::db::{OP_REVISION, OP_TRASH, PendingOp}; /// The registry every mount publishes itself into is what lets the daemon's /// single drain thread reach a node living in an on-demand fork's inode @@ -4065,6 +4377,250 @@ mod tests { )); } + #[test] + fn denied_queue_guard_preserves_accepted_bytes_and_pending_count() { + use std::cell::Cell; + + let (mut state, _dir) = state_test_helper(); + let root = state.intern(0, node_helper("root", "none", "root", true)); + let shared = state.intern(root, node_helper("shared", "root", "shared", true)); + state.entries.get_mut(&shared).unwrap().access = Access::Viewer; + let uid = state.entries[&shared].uid.clone(); + let db = state.db.clone(); + let before = db.pending_ops().unwrap().len(); + + let preserved = Cell::new(false); + let result = preserve_on_access_denied( + require_uid_access(&db, &uid, &[Access::Viewer]), + true, + || preserved.set(true), + ); + assert_eq!(result.unwrap_err().code(), libc::EACCES); + assert!(preserved.get(), "accepted bytes must enter recovery"); + assert_eq!(db.pending_ops().unwrap().len(), before); + + let untouched = Cell::new(false); + assert!(preserve_on_access_denied(Ok(()), true, || untouched.set(true)).is_ok()); + assert!( + !untouched.get(), + "a permitted write must continue to the normal queue path" + ); + + // The same fail-closed rule applies before hydration and to a stale UID. + db.set_share_access(&uid, Access::Viewer).unwrap(); + assert_eq!( + require_uid_access(&db, &uid, &[]).unwrap_err().code(), + libc::EACCES + ); + let stale = NodeUid::new(VolumeId::from("vol"), LinkId::from("stale")); + assert_eq!( + require_uid_access(&db, &stale, &[]).unwrap_err().code(), + libc::EACCES + ); + assert_eq!(db.pending_ops().unwrap().len(), before); + } + + #[test] + fn persisted_and_live_queue_authorities_must_both_be_writable() { + let (mut state, _dir) = state_test_helper(); + let root = state.intern(0, node_helper("root", "none", "root", true)); + let node = state.intern(root, node_helper("shared", "root", "shared", true)); + let uid = state.entries[&node].uid.clone(); + let db = state.db.clone(); + + db.set_share_access(&uid, Access::Viewer).unwrap(); + assert_eq!( + require_uid_access(&db, &uid, &[Access::Editor]) + .unwrap_err() + .code(), + libc::EACCES, + "a persisted downgrade must beat stale writable live state" + ); + + db.set_share_access(&uid, Access::Editor).unwrap(); + assert_eq!( + require_uid_access(&db, &uid, &[Access::Viewer]) + .unwrap_err() + .code(), + libc::EACCES, + "a live downgrade must beat restored writable persisted state" + ); + assert!(require_uid_access(&db, &uid, &[Access::Editor]).is_ok()); + } + + fn function_source<'a>(source: &'a str, signature: &str) -> &'a str { + let start = source.find(signature).expect("function signature exists"); + let body = &source[start..]; + let open = body.find('{').expect("function body starts"); + let mut depth = 0usize; + for (offset, byte) in body.as_bytes()[open..].iter().enumerate() { + match byte { + b'{' => depth += 1, + b'}' => { + depth -= 1; + if depth == 0 { + return &body[..open + offset + 1]; + } + } + _ => {} + } + } + panic!("function body closes") + } + + fn assert_before(source: &str, first: &str, second: &str) { + let first = source.find(first).expect("first operation exists"); + let second = source.find(second).expect("second operation exists"); + assert!(first < second, "{first} must precede {second}"); + } + + #[test] + fn production_queue_guards_precede_mutating_work() { + let source = include_str!("lib.rs"); + + let revision = function_source(source, "fn queue_revision("); + assert_before(revision, "if !h.dirty", "preserve_on_access_denied"); + assert_before(revision, "preserve_on_access_denied", "fill_gaps"); + + let staged = function_source(source, "fn enqueue_staged_write("); + assert_before(staged, "preserve_on_access_denied", "self.pending.lock()"); + assert_before( + staged, + "preserve_on_access_denied", + "self.cache.stage_write", + ); + + let truncate = function_source(source, "fn queue_truncate("); + assert_before(truncate, "require_uid_writable", "self.pending.lock()"); + assert_before(truncate, "require_uid_writable", "create_scratch"); + + for (signature, first_side_effect) in [ + ("fn queue_local_node(", "mint_local_uid"), + ("fn queue_rename(", "queue_rename_authorized"), + ("fn queue_trash(", "replace_ops_with_trash"), + ] { + let function = function_source(source, signature); + assert_before(function, "require_uid_writable", first_side_effect); + } + let trash = function_source(source, "fn queue_trash("); + assert_before(trash, "replace_ops_with_trash", "discard_staged"); + + let filesystem = include_str!("filesystem.rs"); + for signature in ["fn serve_unlink(", "fn serve_rmdir("] { + let function = function_source(filesystem, signature); + assert_before(function, "require_writable", "lookup_child"); + } + + let rename = function_source(filesystem, "fn serve_rename("); + assert_before(rename, "require_rename_access", "remove_replaced"); + assert_before(rename, "remove_replaced", "queue_rename_authorized"); + for (signature, gate, remote_call) in [ + ("fn serve_create(", "require_uid_writable", "upload_file"), + ("fn serve_mkdir(", "require_uid_writable", "create_folder"), + ("fn trash_child(", "require_uid_writable", "trash_nodes"), + ("fn serve_rename(", "require_rename_access", "rename_node"), + ] { + let function = function_source(filesystem, signature); + assert_before(function, gate, remote_call); + } + let control_mkdir = function_source(source, "fn create_folder("); + assert_before( + control_mkdir, + "require_uid_writable", + ".create_folder(&parent_uid", + ); + for (signature, gate, remote_call) in [ + ( + "fn rename(&self, rel: &Path", + "require_rename_access", + ".rename_node", + ), + ( + "fn move_to(&self, rel: &Path", + "require_rename_access", + ".move_node", + ), + ( + "fn delete(&self, rel: &Path", + "require_node_parent_access", + ".trash_nodes", + ), + ] { + let function = function_source(source, signature); + assert_before(function, "source_parent_uid", gate); + assert_before(function, gate, remote_call); + } + } + + #[test] + fn replacement_rename_authorizes_source_and_both_parents() { + use std::cell::RefCell; + + let source = NodeUid::new(VolumeId::from("vol"), LinkId::from("source")); + let old_parent = NodeUid::new(VolumeId::from("vol"), LinkId::from("old")); + let new_parent = NodeUid::new(VolumeId::from("vol"), LinkId::from("new")); + let checked = RefCell::new(Vec::new()); + + require_rename_access( + |uid| { + checked.borrow_mut().push(uid.to_string()); + Ok(()) + }, + &source, + &old_parent, + &new_parent, + ) + .unwrap(); + + assert_eq!( + checked.into_inner(), + vec!["vol~source", "vol~old", "vol~new"] + ); + } + + #[test] + fn writable_nested_node_cannot_mutate_a_viewer_parent_namespace() { + let source = NodeUid::new(VolumeId::from("vol"), LinkId::from("nested-editor")); + let viewer_parent = NodeUid::new(VolumeId::from("vol"), LinkId::from("viewer-parent")); + let writable_destination = NodeUid::new(VolumeId::from("vol"), LinkId::from("destination")); + let access = |uid: &NodeUid| { + if uid == &viewer_parent { + Err(Errno::EACCES) + } else { + Ok(()) + } + }; + + assert_eq!( + require_node_parent_access(access, &source, &viewer_parent) + .unwrap_err() + .code(), + libc::EACCES, + "delete must authorize the source namespace" + ); + assert_eq!( + require_rename_access(access, &source, &viewer_parent, &writable_destination,) + .unwrap_err() + .code(), + libc::EACCES, + "rename and move must authorize the original parent" + ); + } + + #[test] + fn access_handler_logic_matches_advertised_permissions() { + assert!(access_allowed(false, Access::Viewer, AccessFlags::F_OK)); + assert!(access_allowed(false, Access::Viewer, AccessFlags::R_OK)); + assert!(!access_allowed(false, Access::Viewer, AccessFlags::W_OK)); + assert!(!access_allowed(false, Access::Owner, AccessFlags::X_OK)); + assert!(access_allowed(true, Access::Viewer, AccessFlags::X_OK)); + assert!(access_allowed( + true, + Access::Editor, + AccessFlags::R_OK | AccessFlags::W_OK | AccessFlags::X_OK + )); + } + use proton_drive_rs::proton_sdk::ids::{LinkId, NodeUid, VolumeId}; use proton_drive_rs::{Node, NodeKind}; @@ -4111,6 +4667,7 @@ mod tests { )); std::fs::create_dir_all(&dir_path).unwrap(); let db = pdfs_core::db::Db::open(&dir_path.join("cache.db")).unwrap(); + let share_access = db.all_share_access().unwrap(); let st = crate::state::State { entries: std::collections::HashMap::new(), by_uid: std::collections::HashMap::new(), @@ -4119,11 +4676,94 @@ mod tests { active_writes: std::collections::HashMap::new(), handles: std::collections::HashMap::new(), next_fh: 1, + access_changes: std::collections::HashSet::new(), + share_access, db: std::sync::Arc::new(db), }; (st, TestDir(dir_path)) } + #[test] + fn queued_trash_of_open_file_survives_last_release_until_atomic_completion() { + let (mut state, _dir) = state_test_helper(); + let parent = state.intern(0, node_helper("parent", "none", "parent", true)); + let file_node = node_helper("open-trash", "parent", "dirty.txt", false); + let uid = file_node.uid.clone(); + let ino = state.intern(parent, file_node); + state.children.insert(parent, vec![ino]); + state.entries.get_mut(&ino).unwrap().open_count = 1; + + // The typed DB operation atomically replaces the old revision with the + // one intent that must survive the open handle's final release. + state + .db + .enqueue_op(&PendingOp { + id: 0, + kind: OP_REVISION.to_string(), + uid: uid.to_string(), + parent_uid: None, + name: None, + blob_path: None, + meta_json: None, + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + }) + .unwrap(); + let (_, blobs) = state + .db + .replace_ops_with_trash(&uid.to_string(), "dirty.txt", 2) + .unwrap(); + assert!(blobs.is_empty()); + let queued_counts = state.db.pending_op_counts().unwrap(); + + state.unlink_mem(&uid); + assert!(state.entries[&ino].unlinked); + assert!(!state.children[&parent].contains(&ino)); + assert!(state.db.node_by_uid(&uid.to_string()).unwrap().is_some()); + + let released = release_unlinked_entry(&mut state, ino).unwrap(); + assert_eq!(released, uid); + assert!(!state.by_uid.contains_key(&uid)); + assert!(release_must_retain_queued_trash(&state.db, &uid).unwrap()); + assert!( + state.db.node_by_uid(&uid.to_string()).unwrap().is_some(), + "last release keeps persisted drain authority" + ); + let pending = state.db.pending_ops().unwrap(); + assert_eq!(pending.len(), 1, "release must not add a revision"); + assert_eq!(pending[0].kind, OP_TRASH); + let after_release = state.db.pending_op_counts().unwrap(); + assert_eq!(after_release.uploads, 0); + assert_eq!(after_release.changes, 1); + assert_eq!(after_release.uploads, queued_counts.uploads); + assert_eq!(after_release.changes, queued_counts.changes); + + state.db.complete_trash_op(pending[0].id, &uid).unwrap(); + assert!(state.db.pending_ops().unwrap().is_empty()); + assert!(state.db.node_by_uid(&uid.to_string()).unwrap().is_none()); + } + + #[test] + fn queued_trash_lookup_failure_conservatively_retains_unlinked_state() { + let (state, _dir) = state_test_helper(); + let uid = NodeUid::new(VolumeId::from("vol"), LinkId::from("uncertain-trash")); + state + .db + .with_conn(|conn| { + conn.execute_batch("DROP TABLE pending_op")?; + Ok(()) + }) + .unwrap(); + + assert!(release_must_retain_queued_trash(&state.db, &uid).is_err()); + assert!( + !release_can_discard_unlinked(&state.db, &uid), + "database uncertainty must retain the tombstone and authority row" + ); + } + fn node_helper(id: &str, parent: &str, name: &str, is_dir: bool) -> Node { let uid = NodeUid::new(VolumeId::from("vol"), LinkId::from(id)); let parent_uid = if parent == "none" { @@ -4154,6 +4794,7 @@ mod tests { is_shared: false, is_shared_publicly: false, signature_email: None, + membership: None, verification: Default::default(), } } diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 23e459e..b35a370 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -179,6 +179,9 @@ pub fn mount( } = options; let (root, online) = fetch_or_recall_root(&client, &rt, &db)?; let scope = root.tree_event_scope_id(); + let share_access = db + .all_share_access() + .map_err(|error| std::io::Error::other(format!("load shared access: {error}")))?; // The folder-sync engine (devices.md Phase 2) runs on its own thread and is // nudged over this channel; the sender lives in Core so control-socket @@ -196,6 +199,8 @@ pub fn mount( active_writes: HashMap::new(), handles: HashMap::new(), next_fh: 1, + access_changes: HashSet::new(), + share_access, db: db.clone(), })), cache: Arc::new(cache), diff --git a/crates/pdfs-fuse/src/state.rs b/crates/pdfs-fuse/src/state.rs index 824c240..4696bb8 100644 --- a/crates/pdfs-fuse/src/state.rs +++ b/crates/pdfs-fuse/src/state.rs @@ -2,15 +2,16 @@ //! the open write handles, and the interval set that tracks which bytes of a //! staged write are authored locally. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet, VecDeque}; use std::fs::File; use std::path::PathBuf; use std::sync::Arc; use pdfs_core::cache::StagedWrite; use pdfs_core::db::Db; +use pdfs_core::{Access, access_for}; use proton_drive_rs::proton_sdk::ids::NodeUid; -use proton_drive_rs::{Node, NodeKind}; +use proton_drive_rs::{MemberRole, Node, NodeKind}; use tracing::warn; /// A node known to the filesystem, addressed by its kernel inode. @@ -18,11 +19,18 @@ pub(crate) struct Entry { pub(crate) uid: NodeUid, pub(crate) parent: u64, pub(crate) node: Node, + pub(crate) access: Access, pub(crate) lookup_count: u64, pub(crate) open_count: u32, pub(crate) unlinked: bool, } +impl Entry { + pub(crate) fn writable(&self) -> bool { + self.access.writable() + } +} + /// A set of non-overlapping `[start, end)` byte ranges, kept sorted and merged. /// Tracks which bytes of a [`WriteHandle`]'s scratch file were authored locally /// (vs. still living only in the remote base), so reads and the commit gap-fill @@ -148,6 +156,14 @@ pub(crate) struct State { /// have no entry here. pub(crate) handles: HashMap, pub(crate) next_fh: u64, + /// Resident inode attributes whose effective access changed during an + /// intern/root refresh. Core drains this set and notifies the matching + /// kernel only after releasing the State lock. + pub(crate) access_changes: HashSet, + /// Persisted shared-root authority, loaded once per inode space and updated + /// on every explicit role/tombstone change. Descendant inheritance reads + /// this map instead of issuing one SQLite query per interned inode. + pub(crate) share_access: HashMap, /// Unified SQLite metadata cache. Every map mutation below writes through to /// it inside the `State` lock so the DB stays the authoritative copy across /// restarts (see plan.md P1). @@ -155,6 +171,306 @@ pub(crate) struct State { } impl State { + pub(crate) fn require_writable(&self, ino: u64) -> Result<(), fuser::Errno> { + match self.entries.get(&ino) { + Some(entry) if entry.writable() => Ok(()), + Some(_) => Err(fuser::Errno::EACCES), + None => Err(fuser::Errno::ENOENT), + } + } + + pub(crate) fn access_by_uid(&self, uid: &NodeUid) -> Option { + self.by_uid + .get(uid) + .and_then(|ino| self.entries.get(ino)) + .map(|entry| entry.access) + } + + pub(crate) fn take_access_changes(&mut self) -> Vec { + self.access_changes.drain().collect() + } + + fn stored_share_access(&self, uid: &NodeUid) -> Option { + self.share_access.get(uid).copied() + } + + pub(crate) fn access_for_node(&mut self, parent: u64, node: &Node) -> Access { + let parent_access = self + .entries + .get(&parent) + .map_or(Access::Owner, |entry| entry.access); + let stored = self.stored_share_access(&node.uid); + + // A persisted row is the offline authority. A live membership on a + // node directly below an owned root identifies a newly discovered + // share root and refreshes that row. Descendants inherit directly. + let is_new_share_root = node.membership.is_some() && parent_access == Access::Owner; + if stored.is_some() || is_new_share_root { + let access = node.membership.as_ref().map_or_else( + || stored.unwrap_or(Access::Viewer), + |membership| access_for(membership.role_exact(), parent_access, true), + ); + if let Err(error) = self.db.set_share_access(&node.uid, access) { + warn!(uid = %node.uid, ?access, %error, "db set_share_access failed"); + } else { + self.share_access.insert(node.uid.clone(), access); + } + return access; + } + parent_access + } + + /// Record a role observed by an authority-bearing shared-root surface. + /// + /// Ordinary node enumeration, especially `enumerate_nodes_light` and old + /// cached JSON, may omit membership and must continue to trust persisted + /// `share_access`. P3 can call this method only when its share-root endpoint + /// explicitly observed a role; explicit absence or an unknown wire role is + /// represented by `None` and fails closed to Viewer. + #[allow(dead_code)] // P3 consumes this explicit-provenance hook. + pub(crate) fn record_observed_share_root_role( + &mut self, + uid: &NodeUid, + role: Option, + ) -> Vec { + let (root, parent_access) = match self.by_uid.get(uid).copied() { + Some(root) => { + let parent_access = self + .entries + .get(&root) + .and_then(|entry| self.entries.get(&entry.parent)) + .map_or(Access::Owner, |entry| entry.access); + (Some(root), parent_access) + } + None => (None, Access::Unknown), + }; + let access = access_for(role, parent_access, true); + if let Err(error) = self.db.set_share_access(uid, access) { + warn!(%uid, ?access, %error, "db set observed share access failed"); + } else { + self.share_access.insert(uid.clone(), access); + } + let Some(root) = root else { + return Vec::new(); + }; + + let mut changed = Vec::new(); + if let Some(entry) = self.entries.get_mut(&root) + && entry.access != access + { + entry.access = access; + changed.push(root); + } + changed.extend(self.recompute_descendant_access(root)); + self.access_changes.extend(changed.iter().copied()); + changed + } + + fn resident_children(&self) -> HashMap> { + let mut children: HashMap> = HashMap::new(); + for (&ino, entry) in &self.entries { + if entry.parent != ino { + children.entry(entry.parent).or_default().push(ino); + } + } + children + } + + /// Recompute all resident descendants after a root's effective access + /// changes. A nested persisted share root remains its own authority. + fn recompute_descendant_access(&mut self, root: u64) -> Vec { + let mut changed = Vec::new(); + let mut queue = VecDeque::from([root]); + let mut seen = HashSet::new(); + let children = self.resident_children(); + while let Some(parent) = queue.pop_front() { + if !seen.insert(parent) { + continue; + } + let parent_access = match self.entries.get(&parent) { + Some(entry) => entry.access, + None => continue, + }; + for &child in children.get(&parent).into_iter().flatten() { + let access = self + .entries + .get(&child) + .and_then(|entry| self.share_access.get(&entry.uid).copied()) + .unwrap_or(parent_access); + if let Some(entry) = self.entries.get_mut(&child) + && entry.access != access + { + entry.access = access; + changed.push(child); + } + queue.push_back(child); + } + } + changed + } + + /// Force a known shared root and every resident descendant read-only. + /// + /// This intentionally ignores nested authorities: losing access to an + /// outer tree must not leave a cached inner entry writable. + pub(crate) fn downgrade_shared_subtree(&mut self, uid: &NodeUid) -> Vec { + self.share_access.insert(uid.clone(), Access::Viewer); + let Some(&root) = self.by_uid.get(uid) else { + return Vec::new(); + }; + let mut changed = Vec::new(); + let mut queue = VecDeque::from([root]); + let mut seen = HashSet::new(); + let children = self.resident_children(); + while let Some(ino) = queue.pop_front() { + if !seen.insert(ino) { + continue; + } + if let Some(entry) = self.entries.get_mut(&ino) + && entry.access != Access::Viewer + { + entry.access = Access::Viewer; + changed.push(ino); + } + queue.extend(children.get(&ino).into_iter().flatten().copied()); + } + changed + } + + /// Downgrade resident shared subtrees identified by persisted authority, + /// live membership, or already-inherited non-owner access. Owned and device + /// entries remain `Owner`. + pub(crate) fn downgrade_known_shared_access(&mut self) -> Vec { + for access in self.share_access.values_mut() { + *access = Access::Viewer; + } + let roots: Vec = self + .entries + .iter() + .filter(|entry| { + let (_, entry) = entry; + entry.access != Access::Owner + || entry.node.membership.is_some() + || self.share_access.contains_key(&entry.uid) + }) + .map(|(&ino, _)| ino) + .collect(); + let children = self.resident_children(); + let mut queue = VecDeque::from(roots); + let mut seen = HashSet::new(); + let mut changed = Vec::new(); + while let Some(ino) = queue.pop_front() { + if !seen.insert(ino) { + continue; + } + if let Some(entry) = self.entries.get_mut(&ino) + && entry.access != Access::Viewer + { + entry.access = Access::Viewer; + changed.push(ino); + } + queue.extend(children.get(&ino).into_iter().flatten().copied()); + } + changed + } + + /// Rebuild effective access after cold hydration. The memo makes each + /// parent edge resolve once, then every child is a direct inheritance. + pub(crate) fn hydrate_access(&mut self) { + fn resolve( + state: &State, + ino: u64, + memo: &mut HashMap, + visiting: &mut HashSet, + ) -> Access { + if let Some(access) = memo.get(&ino) { + return *access; + } + let Some(entry) = state.entries.get(&ino) else { + return Access::Unknown; + }; + if !visiting.insert(ino) { + return Access::Unknown; + } + let stored = state.share_access.get(&entry.uid).copied(); + let access = if let Some(stored) = stored { + stored + } else if entry.parent == ino { + Access::Owner + } else if !state.entries.contains_key(&entry.parent) { + entry + .node + .parent_uid + .as_ref() + .and_then(|parent_uid| state.share_access.get(parent_uid).copied()) + .unwrap_or(Access::Unknown) + } else { + let parent_access = resolve(state, entry.parent, memo, visiting); + if entry.node.membership.is_none() || parent_access != Access::Owner { + parent_access + } else { + access_for( + entry + .node + .membership + .as_ref() + .and_then(|membership| membership.role_exact()), + parent_access, + true, + ) + } + }; + visiting.remove(&ino); + memo.insert(ino, access); + access + } + + let mut memo = HashMap::new(); + let mut visiting = HashSet::new(); + for ino in self.entries.keys().copied().collect::>() { + resolve(self, ino, &mut memo, &mut visiting); + } + for (ino, access) in memo { + if let Some(entry) = self.entries.get_mut(&ino) { + entry.access = access; + } + } + let discovered: Vec<(NodeUid, Access)> = self + .entries + .values() + .filter_map(|entry| { + let parent_access = self + .entries + .get(&entry.parent) + .map_or(Access::Unknown, |parent| parent.access); + (entry.node.membership.is_some() + && parent_access == Access::Owner + && !self.share_access.contains_key(&entry.uid)) + .then(|| { + ( + entry.uid.clone(), + access_for( + entry + .node + .membership + .as_ref() + .and_then(|membership| membership.role_exact()), + parent_access, + true, + ), + ) + }) + }) + .collect(); + for (uid, access) in discovered { + if let Err(error) = self.db.set_share_access(&uid, access) { + warn!(%uid, ?access, %error, "db set hydrated share access failed"); + } else { + self.share_access.insert(uid, access); + } + } + } + pub(crate) fn intern(&mut self, parent: u64, node: Node) -> u64 { if let Err(e) = self.db.upsert_node(&node) { warn!(uid = %node.uid, error = %e, "db upsert_node failed"); @@ -167,12 +483,24 @@ impl State { /// the split exists so a batch can pay for one transaction instead of `n`. pub(crate) fn intern_mem(&mut self, parent: u64, node: Node) -> u64 { if let Some(&ino) = self.by_uid.get(&node.uid) { + let access = self.access_for_node(parent, &node); + let changed = self + .entries + .get(&ino) + .is_some_and(|entry| entry.access != access); if let Some(e) = self.entries.get_mut(&ino) { e.node = node; e.parent = parent; + e.access = access; + } + if changed { + self.access_changes.insert(ino); + let descendants = self.recompute_descendant_access(ino); + self.access_changes.extend(descendants); } return ino; } + let access = self.access_for_node(parent, &node); let ino = self.next_ino; self.next_ino += 1; self.by_uid.insert(node.uid.clone(), ino); @@ -182,6 +510,7 @@ impl State { uid: node.uid.clone(), parent, node, + access, lookup_count: 1, open_count: 0, unlinked: false, @@ -276,15 +605,40 @@ impl State { self.forget(uid) } + /// Remove a dentry while retaining both open inode state and the persisted + /// node row. Queued trash needs the row as drain-time authority; if the node + /// is open, POSIX also requires its inode to survive until the final close. + pub(crate) fn unlink_mem(&mut self, uid: &NodeUid) -> Option<(u64, String)> { + if let Some(&ino) = self.by_uid.get(uid) + && let Some(entry) = self.entries.get_mut(&ino) + && entry.open_count > 0 + { + entry.unlinked = true; + if let Some(kids) = self.children.get_mut(&entry.parent) { + kids.retain(|&kid| kid != ino); + } + return Some((entry.parent, entry.node.name.clone())); + } + self.forget_mem(uid) + } + /// Forget a node entirely: drop its inode, its uid mapping, its own cached /// listing, its slot in its parent's listing, and its DB row. Returns /// `(parent_ino, name)` when the node was known, so the caller can notify /// the kernel. pub(crate) fn forget(&mut self, uid: &NodeUid) -> Option<(u64, String)> { - let ino = self.by_uid.remove(uid)?; if let Err(e) = self.db.delete_node(uid) { warn!(%uid, error = %e, "db delete_node failed"); } + self.forget_mem(uid) + } + + /// Forget a resident node while retaining its persisted row. + /// + /// A queued trash uses the row as its drain-time access authority and only + /// deletes it after the remote mutation lands. + pub(crate) fn forget_mem(&mut self, uid: &NodeUid) -> Option<(u64, String)> { + let ino = self.by_uid.remove(uid)?; let entry = self.entries.remove(&ino)?; self.children.remove(&ino); if let Some(kids) = self.children.get_mut(&entry.parent) { @@ -308,11 +662,21 @@ impl State { new_parent_uid: &NodeUid, name: &str, ) { + let parent_access = self + .entries + .get(&new_parent) + .map_or(Access::Owner, |entry| entry.access); + let access = self + .entries + .get(&ino) + .and_then(|entry| self.share_access.get(&entry.uid).copied()) + .unwrap_or(parent_access); let Some(entry) = self.entries.get_mut(&ino) else { return; }; let old_parent = entry.parent; entry.parent = new_parent; + entry.access = access; entry.node.name = name.to_string(); entry.node.parent_uid = Some(new_parent_uid.clone()); let node = entry.node.clone(); @@ -331,6 +695,7 @@ impl State { if let Err(e) = self.db.upsert_node(&node) { warn!(uid = %node.uid, error = %e, "db upsert_node failed for a queued rename"); } + self.recompute_descendant_access(ino); } /// Drop a directory's cached child listing and mark it unlisted in the DB, @@ -427,7 +792,8 @@ impl State { #[cfg(test)] mod tests { use super::*; - use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId}; + use proton_drive_rs::ShareMembership; + use proton_drive_rs::proton_sdk::ids::{LinkId, ShareId, ShareMembershipId, VolumeId}; fn uid(link: &str) -> NodeUid { NodeUid::new(VolumeId::from("vol"), LinkId::from(link)) @@ -457,10 +823,19 @@ mod tests { is_shared: false, is_shared_publicly: false, signature_email: None, + membership: None, verification: Default::default(), } } + fn membership(permissions: i32) -> ShareMembership { + ShareMembership { + share_id: ShareId::from("share"), + membership_id: ShareMembershipId::from("membership"), + permissions, + } + } + /// A unique temp directory removed on drop; avoids a dev-dependency. struct TempDir(PathBuf); impl TempDir { @@ -488,6 +863,7 @@ mod tests { fn state() -> (State, TempDir) { let dir = TempDir::new(); let db = Db::open(&dir.0.join("cache.db")).unwrap(); + let share_access = db.all_share_access().unwrap(); let st = State { entries: HashMap::new(), by_uid: HashMap::new(), @@ -496,6 +872,8 @@ mod tests { active_writes: HashMap::new(), handles: HashMap::new(), next_fh: 1, + access_changes: HashSet::new(), + share_access, db: Arc::new(db), }; (st, dir) @@ -624,4 +1002,310 @@ mod tests { st.children.remove(&folder); assert!(st.has_children(folder), "has child in db"); } + + #[test] + fn share_root_access_is_persisted_and_children_inherit_in_constant_time() { + let (mut st, _dir) = state(); + let owned_root = st.intern(0, node("root", "none", "My Files", true)); + let mut shared = node("shared", "root", "Shared", true); + shared.membership = Some(membership(6)); + let shared_ino = st.intern(owned_root, shared); + let child = st.intern( + shared_ino, + node("shared-child", "shared", "child.txt", false), + ); + + assert_eq!(st.entries[&shared_ino].access, Access::Editor); + assert_eq!(st.entries[&child].access, Access::Editor); + assert_eq!( + st.db.share_access(&uid("shared")).unwrap(), + Some(Access::Editor) + ); + } + + #[test] + fn persisted_share_access_controls_offline_hydration() { + let (mut st, _dir) = state(); + let root_node = node("root", "none", "My Files", true); + let shared_node = node("offline-share", "root", "Offline share", true); + let child_node = node("offline-child", "offline-share", "nested folder", true); + let leaf_node = node("offline-leaf", "offline-child", "child.txt", false); + for node in [&root_node, &shared_node, &child_node, &leaf_node] { + st.db.upsert_node(node).unwrap(); + } + st.db + .set_share_access(&uid("offline-share"), Access::Viewer) + .unwrap(); + st.share_access = st.db.all_share_access().unwrap(); + + // Reconstruct in deliberately non-topological order, matching the + // database hydration path that first allocates every inode and only + // then materializes entries. + let owned_root = 1; + let shared = 2; + let child = 3; + let leaf = 4; + for (node, ino) in [ + (&root_node, owned_root), + (&shared_node, shared), + (&child_node, child), + (&leaf_node, leaf), + ] { + st.by_uid.insert(node.uid.clone(), ino); + } + for (ino, parent, node) in [ + (leaf, child, leaf_node), + (child, shared, child_node), + (shared, owned_root, shared_node), + (owned_root, owned_root, root_node), + ] { + st.entries.insert( + ino, + Entry { + uid: node.uid.clone(), + parent, + node, + access: Access::Unknown, + lookup_count: 1, + open_count: 0, + unlinked: false, + }, + ); + } + st.next_ino = 5; + st.hydrate_access(); + + assert_eq!(st.entries[&owned_root].access, Access::Owner); + assert_eq!(st.entries[&shared].access, Access::Viewer); + assert_eq!(st.entries[&child].access, Access::Viewer); + assert_eq!(st.entries[&leaf].access, Access::Viewer); + assert_eq!( + st.require_writable(shared).unwrap_err().code(), + libc::EACCES + ); + assert_eq!(st.require_writable(child).unwrap_err().code(), libc::EACCES); + assert_eq!(st.require_writable(leaf).unwrap_err().code(), libc::EACCES); + assert!(st.require_writable(owned_root).is_ok()); + } + + #[test] + fn reinterned_share_root_propagates_a_viewer_downgrade() { + let (mut st, _dir) = state(); + let owned_root = st.intern(0, node("root", "none", "My Files", true)); + let mut shared_node = node("shared", "root", "Shared", true); + shared_node.membership = Some(membership(6)); + let shared = st.intern(owned_root, shared_node.clone()); + let child = st.intern(shared, node("child", "shared", "folder", true)); + let leaf = st.intern(child, node("leaf", "child", "leaf.txt", false)); + assert_eq!(st.entries[&leaf].access, Access::Editor); + + shared_node.membership = Some(membership(4)); + assert_eq!(st.intern(owned_root, shared_node), shared); + + for ino in [shared, child, leaf] { + assert_eq!(st.entries[&ino].access, Access::Viewer); + assert_eq!(st.require_writable(ino).unwrap_err().code(), libc::EACCES); + } + assert_eq!( + st.db.share_access(&uid("shared")).unwrap(), + Some(Access::Viewer) + ); + } + + #[test] + fn missing_light_membership_keeps_persisted_share_authority() { + for persisted in [Access::Viewer, Access::Editor] { + let (mut st, _dir) = state(); + let owned_root = st.intern(0, node("root", "none", "My Files", true)); + st.db.set_share_access(&uid("shared"), persisted).unwrap(); + st.share_access = st.db.all_share_access().unwrap(); + let shared = st.intern( + owned_root, + node("shared", "root", "Shared without membership", true), + ); + let child = st.intern( + shared, + node("child", "shared", "Light child without membership", false), + ); + + assert_eq!(st.entries[&shared].access, persisted); + assert_eq!(st.entries[&child].access, persisted); + assert_eq!(st.db.share_access(&uid("shared")).unwrap(), Some(persisted)); + } + } + + #[test] + fn explicit_missing_share_role_fails_closed_and_marks_kernel_attrs() { + let (mut st, _dir) = state(); + let owned_root = st.intern(0, node("root", "none", "My Files", true)); + st.db + .set_share_access(&uid("shared"), Access::Editor) + .unwrap(); + let shared = st.intern(owned_root, node("shared", "root", "Shared", true)); + let child = st.intern(shared, node("child", "shared", "child.txt", false)); + st.take_access_changes(); + + let changed = st.record_observed_share_root_role(&uid("shared"), None); + + assert_eq!(st.entries[&shared].access, Access::Viewer); + assert_eq!(st.entries[&child].access, Access::Viewer); + assert_eq!( + st.db.share_access(&uid("shared")).unwrap(), + Some(Access::Viewer) + ); + assert!(changed.contains(&shared)); + assert!(changed.contains(&child)); + let pending = st.take_access_changes(); + assert!(pending.contains(&shared)); + assert!(pending.contains(&child)); + } + + #[test] + fn restored_editor_access_marks_root_and_descendants_for_invalidation() { + let (mut st, _dir) = state(); + let owned_root = st.intern(0, node("root", "none", "My Files", true)); + let mut shared_node = node("shared", "root", "Shared", true); + shared_node.membership = Some(membership(4)); + let shared = st.intern(owned_root, shared_node.clone()); + let child = st.intern(shared, node("child", "shared", "child.txt", false)); + st.take_access_changes(); + + shared_node.membership = Some(membership(6)); + assert_eq!(st.intern(owned_root, shared_node), shared); + + assert_eq!(st.entries[&shared].access, Access::Editor); + assert_eq!(st.entries[&child].access, Access::Editor); + let changed = st.take_access_changes(); + assert!(changed.contains(&shared)); + assert!(changed.contains(&child)); + } + + #[test] + fn revoked_root_tombstone_keeps_resident_and_persisted_descendants_read_only() { + let (mut st, _dir) = state(); + let owned_root = st.intern(0, node("root", "none", "My Files", true)); + let mut shared_node = node("shared", "root", "Shared", true); + shared_node.membership = Some(membership(6)); + let shared = st.intern(owned_root, shared_node); + let child = st.intern(shared, node("child", "shared", "folder", true)); + let leaf = st.intern(child, node("leaf", "child", "leaf.txt", false)); + + st.db + .set_share_access(&uid("shared"), Access::Viewer) + .unwrap(); + let changed = st.downgrade_shared_subtree(&uid("shared")); + assert!(changed.contains(&shared)); + assert!(changed.contains(&child)); + assert!(changed.contains(&leaf)); + st.forget(&uid("shared")); + + assert!(!st.entries.contains_key(&shared)); + assert_eq!(st.entries[&child].access, Access::Viewer); + assert_eq!(st.entries[&leaf].access, Access::Viewer); + assert_eq!(st.require_writable(leaf).unwrap_err().code(), libc::EACCES); + assert_eq!( + st.db.effective_node_access(&uid("leaf")).unwrap(), + Some(Access::Viewer) + ); + + // Restart-shaped reconstruction: the deleted root is absent from the + // inode map, so its direct child is attached to the orphan inode. Cold + // hydration must still follow the persisted parent UID to the retained + // share_access tombstone. + let child_node = st.entries[&child].node.clone(); + let leaf_node = st.entries[&leaf].node.clone(); + st.entries.clear(); + st.by_uid.clear(); + st.children.clear(); + st.by_uid.insert(child_node.uid.clone(), child); + st.by_uid.insert(leaf_node.uid.clone(), leaf); + st.entries.insert( + leaf, + Entry { + uid: leaf_node.uid.clone(), + parent: child, + node: leaf_node, + access: Access::Unknown, + lookup_count: 1, + open_count: 0, + unlinked: false, + }, + ); + st.entries.insert( + child, + Entry { + uid: child_node.uid.clone(), + parent: 0, + node: child_node, + access: Access::Unknown, + lookup_count: 1, + open_count: 0, + unlinked: false, + }, + ); + st.hydrate_access(); + + assert_eq!(st.entries[&child].access, Access::Viewer); + assert_eq!(st.entries[&leaf].access, Access::Viewer); + assert_eq!(st.require_writable(leaf).unwrap_err().code(), libc::EACCES); + } + + #[test] + fn global_access_loss_downgrades_only_recorded_shared_subtrees() { + let (mut st, _dir) = state(); + let owned_root = st.intern(0, node("root", "none", "My Files", true)); + let owned_child = st.intern(owned_root, node("owned", "root", "Owned", true)); + let mut shared_node = node("shared", "root", "Shared", true); + shared_node.membership = Some(membership(6)); + let shared = st.intern(owned_root, shared_node); + let shared_child = st.intern(shared, node("child", "shared", "child.txt", false)); + + st.db.downgrade_all_share_access().unwrap(); + let changed = st.downgrade_known_shared_access(); + + assert!(changed.contains(&shared)); + assert!(changed.contains(&shared_child)); + assert_eq!(st.entries[&shared].access, Access::Viewer); + assert_eq!(st.entries[&shared_child].access, Access::Viewer); + assert_eq!(st.entries[&owned_root].access, Access::Owner); + assert_eq!(st.entries[&owned_child].access, Access::Owner); + } + + #[test] + fn global_access_loss_visits_a_deep_shared_tree_once_per_changed_inode() { + let (mut st, _dir) = state(); + let owned_root = st.intern(0, node("root", "none", "My Files", true)); + let mut shared_node = node("shared", "root", "Shared", true); + shared_node.membership = Some(membership(6)); + let shared = st.intern(owned_root, shared_node); + let mut parent = shared; + for index in 0..1_024 { + let link = format!("deep-{index}"); + let parent_link = st.entries[&parent].uid.link_id.to_string(); + parent = st.intern(parent, node(&link, &parent_link, &link, true)); + } + st.db.downgrade_all_share_access().unwrap(); + + let changed = st.downgrade_known_shared_access(); + + let unique: HashSet = changed.iter().copied().collect(); + assert_eq!(changed.len(), 1_025); + assert_eq!(unique.len(), changed.len()); + assert_eq!(st.entries[&owned_root].access, Access::Owner); + assert!( + st.entries + .values() + .all(|entry| { entry.uid == uid("root") || entry.access == Access::Viewer }) + ); + } + + #[test] + fn unknown_membership_permissions_fail_closed() { + let (mut st, _dir) = state(); + let owned_root = st.intern(0, node("root", "none", "My Files", true)); + let mut shared = node("shared", "root", "Shared", true); + shared.membership = Some(membership(38)); + let shared = st.intern(owned_root, shared); + assert_eq!(st.entries[&shared].access, Access::Viewer); + } } diff --git a/crates/pdfs-fuse/src/sweep.rs b/crates/pdfs-fuse/src/sweep.rs index 835c982..83e9e92 100644 --- a/crates/pdfs-fuse/src/sweep.rs +++ b/crates/pdfs-fuse/src/sweep.rs @@ -160,7 +160,10 @@ impl Core { self.for_each_state(|st| { st.forget_or_unlink(uid); }); - self.discard_queued_ops(uid); + if let Err(error) = self.discard_queued_ops(uid) { + warn!(%uid, ?error, "conflict sweep: queued-op cleanup failed"); + return; + } self.cache.evict(uid); self.evict_reader(uid); if let Err(e) = self.db.delete_node(uid) { @@ -346,6 +349,7 @@ mod tests { is_shared: false, is_shared_publicly: false, signature_email: None, + membership: None, verification: Default::default(), } } diff --git a/crates/pdfs-fuse/src/upload.rs b/crates/pdfs-fuse/src/upload.rs index 8211732..3dfd98e 100644 --- a/crates/pdfs-fuse/src/upload.rs +++ b/crates/pdfs-fuse/src/upload.rs @@ -170,6 +170,8 @@ impl Core { { kids.push(ino); } + drop(st); + self.flush_access_changes(); } // Uploaded fine but the metadata refresh failed; it will // appear on the next directory enumeration regardless. @@ -235,6 +237,8 @@ impl Core { { kids.push(ino); } + drop(st); + self.flush_access_changes(); Ok((ino, new_uid)) } diff --git a/docs/BUGS.md b/docs/BUGS.md index d3a5d55..b24675a 100644 --- a/docs/BUGS.md +++ b/docs/BUGS.md @@ -1432,12 +1432,22 @@ added later, but it must retain the row and blob. ## B34 — Writable POSIX Exposure of Read-Only Shared Folders (MED-07) -**Status:** Open +**Status:** Fixed in code by P2 (2026-07-29); live credentialed shared-tree and second-account +acceptance pending P3/P7 **Found:** 2026-07-21, multi-agent permissions audit (`audit_bugs.md` MED-07) **Where:** `crates/pdfs-fuse/src/lib.rs`, `crates/pdfs-fuse/src/sharing.rs` **Cause:** Shared "Viewer" (read-only) folders are exposed via POSIX with write permissions (`0755`). Local writes succeed initially on the mount but fail perpetually during background online drain with `403 Forbidden` errors. +**Fix:** P2 persists share authority in schema V17 `share_access`, inherits it through resident +entries, exposes read-only POSIX modes, and gates mutations at handler, queue, and drain boundaries. +Downgrades fail closed and are not acknowledged past failed durable state updates. + +**Verification:** The local workspace passes formatting, locked checks, clippy with warnings denied, +and all 324 workspace tests, including access inheritance, mutation gating, queued-operation +authorization, and downgrade failure paths. Live credentialed shared-tree and second-account +acceptance cannot run until P3 exposes the tree and remains required in P7. + --- ## B35 — IPC Unix Socket Creation Permission Race (MED-08)