Merge origin/main into webdav-litmus-compliance

src/application/adapters/caldav_adapter.rs

@@ -96,7 +96,9 @@ impl CalDavAdapter {
                             for attr in e.attributes().flatten() {
                                 let attr_name =
                                     std::str::from_utf8(attr.key.as_ref()).unwrap_or("");
-                                let attr_value = attr.unescape_value().unwrap_or_default();
+                                let attr_value = attr
+                                    .normalized_value(quick_xml::XmlVersion::Implicit1_0)
+                                    .unwrap_or_default();
 
                                 if attr_name == "start" {
                                     // Parse ISO date format with Z for UTC
@@ -161,7 +163,9 @@ impl CalDavAdapter {
                         // Parse time-range attributes
                         for attr in e.attributes().flatten() {
                             let attr_name = std::str::from_utf8(attr.key.as_ref()).unwrap_or("");
-                            let attr_value = attr.unescape_value().unwrap_or_default();
+                            let attr_value = attr
+                                .normalized_value(quick_xml::XmlVersion::Implicit1_0)
+                                .unwrap_or_default();
 
                             if attr_name == "start" {
                                 // Parse ISO date format with Z for UTC

src/application/adapters/plugin_lifecycle_hook.rs

@@ -61,7 +61,10 @@ impl PluginLifecycleHook {
 
             dispatch.dispatch(PluginEvent {
                 name: EVENT_FILE_UPLOADED,
-                user_id: dto.owner_id,
+                // Post-D7 the wire DTO no longer carries `owner_id`;
+                // §14 `created_by` provenance is the equivalent signal
+                // (who put the file in the system).
+                user_id: dto.created_by.map(|u| u.to_string()),
                 invocation_id: Uuid::new_v4().to_string(),
                 payload: serde_json::json!({
                     "path": dto.path,

src/application/adapters/webdav_adapter.rs

@@ -223,10 +223,42 @@ impl NextcloudPropContext {
     }
 }
 
+/// Defense-in-depth cap on attributes per XML element in WebDAV request
+/// bodies. Legitimate PROPFIND / PROPPATCH elements carry a handful of
+/// `xmlns:*` declarations and, occasionally, per-property namespace
+/// bindings — a dozen is already a lot. 100 is generous headroom and
+/// three orders of magnitude below what an attacker would need to
+/// exploit a quadratic parser bug (see quick-xml #969, fixed in 0.41;
+/// this cap fences the same threat model for any future analogous bug
+/// in whatever parser we swap to).
+///
+/// A rejected element yields 400 Bad Request via the ParseError path.
+pub const MAX_ATTRIBUTES_PER_ELEMENT: usize = 100;
+
 /// WebDAV adapter for converting between XML and domain objects
 pub struct WebDavAdapter;
 
 impl WebDavAdapter {
+    /// Refuse elements carrying an unreasonable attribute count.
+    /// See [`MAX_ATTRIBUTES_PER_ELEMENT`] for the reasoning.
+    ///
+    /// `Attributes::count()` is O(N) in the number of attributes (each
+    /// attribute is parsed once), so this check itself is safe even
+    /// against very large elements. The parser may still have paid a
+    /// quadratic cost by the time we get here on a vulnerable version
+    /// of the underlying library — the bump to quick-xml 0.41 closes
+    /// that specific bug; this cap is defense-in-depth against future
+    /// analogous bugs and against adversarially large XML that would
+    /// otherwise reach our downstream code.
+    fn check_attribute_cap(e: &BytesStart) -> Result<()> {
+        if e.attributes().count() > MAX_ATTRIBUTES_PER_ELEMENT {
+            return Err(WebDavError::ParseError(format!(
+                "Element carries more than {MAX_ATTRIBUTES_PER_ELEMENT} attributes"
+            )));
+        }
+        Ok(())
+    }
+
     /// Collect namespace prefix → URI mappings from element attributes.
     /// E.g. `xmlns:D="DAV:"` maps prefix `"D"` to `"DAV:"`.
     pub fn collect_ns_decls(
@@ -236,11 +268,17 @@ impl WebDavAdapter {
         for attr in e.attributes().flatten() {
             let key = std::str::from_utf8(attr.key.as_ref()).unwrap_or("");
             if let Some(prefix) = key.strip_prefix("xmlns:") {
-                let uri = attr.unescape_value().unwrap_or_default().to_string();
+                let uri = attr
+                    .normalized_value(quick_xml::XmlVersion::Implicit1_0)
+                    .unwrap_or_default()
+                    .to_string();
                 ns_map.insert(prefix.to_string(), uri);
             } else if key == "xmlns" {
                 // Default namespace declaration: xmlns="uri"
-                let uri = attr.unescape_value().unwrap_or_default().to_string();
+                let uri = attr
+                    .normalized_value(quick_xml::XmlVersion::Implicit1_0)
+                    .unwrap_or_default()
+                    .to_string();
                 ns_map.insert(String::new(), uri);
             }
         }
@@ -252,7 +290,9 @@ impl WebDavAdapter {
         for attr in e.attributes().flatten() {
             let key = std::str::from_utf8(attr.key.as_ref()).unwrap_or("");
             if key.starts_with("xmlns:") {
-                let uri = attr.unescape_value().unwrap_or_default();
+                let uri = attr
+                    .normalized_value(quick_xml::XmlVersion::Implicit1_0)
+                    .unwrap_or_default();
                 if uri.is_empty() {
                     return Err(WebDavError::ParseError(
                         "Invalid namespace declaration: prefix bound to empty URI".to_string(),
@@ -305,6 +345,7 @@ impl WebDavAdapter {
         loop {
             match xml_reader.read_event_into(&mut buffer) {
                 Ok(Event::Start(ref e)) => {
+                    Self::check_attribute_cap(e)?;
                     Self::collect_ns_decls(e, &mut ns_map);
                     Self::check_ns_decls_valid(e)?;
                     let name = e.name();
@@ -343,6 +384,7 @@ impl WebDavAdapter {
                     }
                 }
                 Ok(Event::Empty(ref e)) => {
+                    Self::check_attribute_cap(e)?;
                     Self::collect_ns_decls(e, &mut ns_map);
                     Self::check_ns_decls_valid(e)?;
                     let name = e.name();
@@ -981,6 +1023,7 @@ impl WebDavAdapter {
         loop {
             match xml_reader.read_event_into(&mut buffer) {
                 Ok(Event::Start(ref e)) => {
+                    Self::check_attribute_cap(e)?;
                     Self::collect_ns_decls(e, &mut ns_map);
                     let name = e.name();
                     let name_str = std::str::from_utf8(name.as_ref()).unwrap_or("");
@@ -1063,6 +1106,7 @@ impl WebDavAdapter {
                     }
                 }
                 Ok(Event::Empty(ref e)) => {
+                    Self::check_attribute_cap(e)?;
                     Self::collect_ns_decls(e, &mut ns_map);
                     let name = e.name();
                     let name_str = std::str::from_utf8(name.as_ref()).unwrap_or("");

src/application/dtos/address_book_dto.rs

@@ -61,16 +61,3 @@ pub struct UpdateAddressBookDto {
     pub is_public: Option<bool>,
     pub user_id: String, // Current user making the update
 }
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct ShareAddressBookDto {
-    pub address_book_id: String,
-    pub user_id: String,
-    pub can_write: bool,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct UnshareAddressBookDto {
-    pub address_book_id: String,
-    pub user_id: String,
-}

src/application/dtos/favorites_dto.rs

@@ -55,11 +55,6 @@ pub struct FavoriteItemDto {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub item_path: Option<String>,
 
-    /// UUID of the file/folder's actual owner (may differ from `user_id` when
-    /// the item was shared and then favourited by another user).
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub owner_id: Option<String>,
-
     // ── Pre-computed display fields ──
     /// FontAwesome icon CSS class (e.g. "fas fa-file-image", "fas fa-folder")
     pub icon_class: String,
@@ -124,7 +119,6 @@ pub struct FavoriteResourceRow {
     pub size: i64,
     pub resource_created_at: DateTime<Utc>,
     pub modified_at: DateTime<Utc>,
-    pub owner_id: Uuid,
     /// Drive that owns this row. Surfaced on the favorites listing
     /// so a UI can tell when a favorited item lives in a different
     /// drive than the user's home (post-D6 cross-drive moves +

src/application/dtos/file_dto.rs

@@ -54,10 +54,6 @@ pub struct FileDto {
     /// Human-readable formatted size (e.g. "3.27 MB")
     pub size_formatted: String,
 
-    /// Owner user ID (omitted from JSON when None)
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub owner_id: Option<String>,
-
     /// Sort date for Photos timeline — COALESCE(EXIF captured_at, created_at).
     /// Only populated by the /api/photos endpoint.
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -102,7 +98,7 @@ impl From<File> for FileDto {
         let content_hash = file.content_hash().to_string();
 
         // Consume the entity by moving all fields — zero heap allocations
-        // for id, name, path, folder_id, owner_id (previously 5× .to_string()).
+        // for id, name, path, folder_id (previously 4× .to_string()).
         let parts = file.into_parts();
 
         let icon_class = Arc::from(icon_class_for(&parts.name, &parts.mime_type));
@@ -124,7 +120,6 @@ impl From<File> for FileDto {
             icon_special_class,
             category,
             size_formatted,
-            owner_id: parts.owner_id.map(|u| u.to_string()),
             sort_date: None,
             content_hash,
             etag,
@@ -157,9 +152,8 @@ impl FileDto {
     ///
     /// Used when a file is returned to a share recipient: `path` reveals the
     /// full folder hierarchy above the file which the recipient may not have
-    /// access to.  `folder_id` and `owner_id` are intentionally kept — the
-    /// former is needed for sub-folder navigation (covered by the cascade
-    /// grant), and the latter is harmless metadata.
+    /// access to.  `folder_id` is intentionally kept — it's needed for
+    /// sub-folder navigation (covered by the cascade grant).
     #[must_use]
     pub fn without_hierarchy_info(self) -> Self {
         Self {
@@ -183,7 +177,6 @@ impl FileDto {
             icon_special_class: Arc::from(""),
             category: Arc::from("Document"),
             size_formatted: "0 Bytes".to_string(),
-            owner_id: None,
             content_hash: String::new(),
             etag: String::new(),
             sort_date: None,

src/application/dtos/folder_dto.rs

@@ -48,10 +48,6 @@ pub struct FolderDto {
     /// Parent folder ID
     pub parent_id: Option<String>,
 
-    /// Owner user ID (scopes visibility per user)
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub owner_id: Option<String>,
-
     /// Drive that owns this folder. The scope axis for path-based
     /// lookups across REST / WebDAV / NextCloud / CalDAV / CardDAV.
     /// Post-D0 `storage.folders.drive_id` is `NOT NULL`; stub /
@@ -111,7 +107,6 @@ impl From<Folder> for FolderDto {
             name: folder.name().to_string(),
             path: folder.path_string().to_string(),
             parent_id: folder.parent_id().map(String::from),
-            owner_id: folder.owner_id().map(|u| u.to_string()),
             drive_id: folder.drive_id(),
             created_at: folder.created_at(),
             modified_at: folder.modified_at(),
@@ -147,9 +142,8 @@ impl FolderDto {
     ///
     /// Used when a folder is returned to a share recipient: `path` reveals the
     /// full folder hierarchy above the shared folder which the recipient may
-    /// not have access to.  `parent_id` and `owner_id` are intentionally kept
-    /// — the former is needed for sub-folder navigation (covered by the
-    /// cascade grant), and the latter is harmless metadata.
+    /// not have access to.  `parent_id` is intentionally kept — it's needed
+    /// for sub-folder navigation (covered by the cascade grant).
     #[must_use]
     pub fn without_hierarchy_info(self) -> Self {
         Self {
@@ -165,7 +159,6 @@ impl FolderDto {
             name: "stub-folder".to_string(),
             path: "/stub/path".to_string(),
             parent_id: None,
-            owner_id: None,
             drive_id: Uuid::nil(),
             created_at: 0,
             modified_at: 0,
@@ -205,7 +198,6 @@ pub struct FolderResourceRow {
     pub size: i64,
     pub created_at: DateTime<Utc>,
     pub modified_at: DateTime<Utc>,
-    pub owner_id: Uuid,
     /// Drive that owns this row. Same column as
     /// `storage.folders.drive_id` / `storage.files.drive_id`. Surfaced
     /// on the listing so a UI can tell when a child lives in a

src/application/dtos/grant_dto.rs

@@ -55,11 +55,14 @@ impl From<Subject> for SubjectDto {
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
-#[serde(rename_all = "lowercase")]
+#[serde(rename_all = "snake_case")]
 pub enum ResourceTypeDto {
     Folder,
     File,
     Drive,
+    Calendar,
+    AddressBook,
+    Playlist,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
@@ -75,6 +78,9 @@ impl From<ResourceDto> for Resource {
             ResourceTypeDto::Folder => Resource::Folder(dto.id),
             ResourceTypeDto::File => Resource::File(dto.id),
             ResourceTypeDto::Drive => Resource::Drive(dto.id),
+            ResourceTypeDto::Calendar => Resource::Calendar(dto.id),
+            ResourceTypeDto::AddressBook => Resource::AddressBook(dto.id),
+            ResourceTypeDto::Playlist => Resource::Playlist(dto.id),
         }
     }
 }
@@ -85,6 +91,9 @@ impl From<Resource> for ResourceDto {
             Resource::Folder(id) => (ResourceTypeDto::Folder, id),
             Resource::File(id) => (ResourceTypeDto::File, id),
             Resource::Drive(id) => (ResourceTypeDto::Drive, id),
+            Resource::Calendar(id) => (ResourceTypeDto::Calendar, id),
+            Resource::AddressBook(id) => (ResourceTypeDto::AddressBook, id),
+            Resource::Playlist(id) => (ResourceTypeDto::Playlist, id),
         };
         ResourceDto { kind, id }
     }

src/application/dtos/recent_dto.rs

@@ -104,7 +104,6 @@ pub struct RecentResourceRow {
     pub size: i64,
     pub resource_created_at: DateTime<Utc>,
     pub modified_at: DateTime<Utc>,
-    pub owner_id: Uuid,
     /// Drive that owns this row. Surfaced on the recent listing
     /// so a UI can tell when a recently-accessed item lives in a
     /// different drive than the user's home (post-D6 cross-drive

src/application/dtos/trash_dto.rs

@@ -60,7 +60,6 @@ pub struct TrashResourceRow {
     pub size: i64,
     pub resource_created_at: DateTime<Utc>,
     pub modified_at: DateTime<Utc>,
-    pub owner_id: Uuid,
     /// Drive the trashed item belongs to. Surfaced verbatim on the wire
     /// (`TrashResourceItemDto.drive_id`) so the `/trash` UI can group by
     /// drive without an extra lookup per row. D2b: filtering by drive is

src/application/ports/authorization_ports.rs

@@ -62,6 +62,9 @@ pub trait AuthorizationEngine: Send + Sync + 'static {
                 Resource::Folder(id) => ("Folder", id),
                 Resource::File(id) => ("File", id),
                 Resource::Drive(id) => ("Drive", id),
+                Resource::Calendar(id) => ("Calendar", id),
+                Resource::AddressBook(id) => ("AddressBook", id),
+                Resource::Playlist(id) => ("Playlist", id),
             };
             // Audit-worthy: denials are the interesting signal. Routed
             // through the `audit` tracing target so log aggregators can

src/application/ports/calendar_ports.rs

@@ -25,38 +25,11 @@ pub trait CalendarStoragePort: Send + Sync + 'static {
         &self,
         owner_id: Uuid,
     ) -> Result<Vec<CalendarDto>, DomainError>;
-    async fn list_calendars_shared_with_user(
-        &self,
-        user_id: Uuid,
-    ) -> Result<Vec<CalendarDto>, DomainError>;
     async fn list_public_calendars(
         &self,
         limit: i64,
         offset: i64,
     ) -> Result<Vec<CalendarDto>, DomainError>;
-    async fn check_calendar_access(
-        &self,
-        calendar_id: &str,
-        user_id: Uuid,
-    ) -> Result<bool, DomainError>;
-
-    // Calendar sharing
-    async fn share_calendar(
-        &self,
-        calendar_id: &str,
-        user_id: Uuid,
-        access_level: &str,
-    ) -> Result<(), DomainError>;
-    async fn remove_calendar_sharing(
-        &self,
-        calendar_id: &str,
-        user_id: Uuid,
-    ) -> Result<(), DomainError>;
-    async fn get_calendar_shares(
-        &self,
-        calendar_id: &str,
-    ) -> Result<Vec<(String, String)>, DomainError>;
-
     // Calendar properties
     async fn set_calendar_property(
         &self,
@@ -146,33 +119,12 @@ pub trait CalendarUseCase: Send + Sync + 'static {
         user_id: Uuid,
     ) -> Result<CalendarDto, DomainError>;
     async fn list_my_calendars(&self, user_id: Uuid) -> Result<Vec<CalendarDto>, DomainError>;
-    async fn list_shared_calendars(&self, user_id: Uuid) -> Result<Vec<CalendarDto>, DomainError>;
     async fn list_public_calendars(
         &self,
         limit: Option<i64>,
         offset: Option<i64>,
     ) -> Result<Vec<CalendarDto>, DomainError>;
 
-    // Calendar sharing
-    async fn share_calendar(
-        &self,
-        calendar_id: &str,
-        target_user_id: Uuid,
-        access_level: &str,
-        caller_user_id: Uuid,
-    ) -> Result<(), DomainError>;
-    async fn remove_calendar_sharing(
-        &self,
-        calendar_id: &str,
-        target_user_id: Uuid,
-        caller_user_id: Uuid,
-    ) -> Result<(), DomainError>;
-    async fn get_calendar_shares(
-        &self,
-        calendar_id: &str,
-        user_id: Uuid,
-    ) -> Result<Vec<(String, String)>, DomainError>;
-
     // Event operations
     async fn create_event(
         &self,

src/application/ports/carddav_ports.rs

@@ -1,16 +1,105 @@
 use crate::application::dtos::address_book_dto::{
-    AddressBookDto, CreateAddressBookDto, ShareAddressBookDto, UnshareAddressBookDto,
-    UpdateAddressBookDto,
+    AddressBookDto, CreateAddressBookDto, UpdateAddressBookDto,
 };
 use crate::application::dtos::contact_dto::{
     ContactDto, ContactGroupDto, CreateContactDto, CreateContactGroupDto, CreateContactVCardDto,
     GroupMembershipDto, UpdateContactDto, UpdateContactGroupDto,
 };
 use crate::common::errors::DomainError;
+use crate::domain::entities::contact::{AddressBook, Contact, ContactGroup};
 use uuid::Uuid;
 
 pub type CardDavRepositoryError = DomainError;
 
+/// Low-level storage port for CardDAV resources. Post-Round-3 the
+/// port covers ONLY raw storage operations — everything that used
+/// to be routed through it for sharing (`share_address_book`,
+/// `unshare_address_book`, `get_address_book_shares`) or
+/// scope-listing (`get_address_books_by_owner`,
+/// `get_shared_address_books`) is gone. Access decisions live in
+/// `AuthorizationEngine`; sharing state lives in
+/// `storage.role_grants`. The service layer (`ContactService`) gates
+/// each call, then reaches through this port for storage.
+///
+/// Symmetric with `CalendarStoragePort`. Implemented by
+/// `ContactStorageAdapter` against Postgres today; a future backend
+/// (external CardDAV, LDAP directory, in-memory test mock) would
+/// implement the same trait and swap in via DI.
+pub trait ContactStoragePort: Send + Sync + 'static {
+    // ── Address books ────────────────────────────────────────────
+    async fn create_address_book(
+        &self,
+        address_book: AddressBook,
+    ) -> Result<AddressBook, DomainError>;
+    async fn update_address_book(
+        &self,
+        address_book: AddressBook,
+    ) -> Result<AddressBook, DomainError>;
+    async fn delete_address_book(&self, id: &Uuid) -> Result<(), DomainError>;
+    async fn get_address_book_by_id(&self, id: &Uuid) -> Result<Option<AddressBook>, DomainError>;
+    async fn get_public_address_books(&self) -> Result<Vec<AddressBook>, DomainError>;
+
+    // ── Contacts ─────────────────────────────────────────────────
+    async fn create_contact(&self, contact: Contact) -> Result<Contact, DomainError>;
+    async fn update_contact(&self, contact: Contact) -> Result<Contact, DomainError>;
+    async fn delete_contact(&self, id: &Uuid) -> Result<(), DomainError>;
+    async fn get_contact_by_id(&self, id: &Uuid) -> Result<Option<Contact>, DomainError>;
+    /// Indexed single-row lookup by vCard UID within a specific book.
+    async fn get_contact_by_uid(
+        &self,
+        address_book_id: &Uuid,
+        uid: &str,
+    ) -> Result<Option<Contact>, DomainError>;
+    /// Indexed batch lookup by vCard UID within a specific book.
+    async fn get_contacts_by_uids(
+        &self,
+        address_book_id: &Uuid,
+        uids: &[String],
+    ) -> Result<Vec<Contact>, DomainError>;
+    async fn get_contacts_by_address_book(
+        &self,
+        address_book_id: &Uuid,
+    ) -> Result<Vec<Contact>, DomainError>;
+    async fn get_contacts_by_address_book_paginated(
+        &self,
+        address_book_id: &Uuid,
+        limit: i64,
+        offset: i64,
+    ) -> Result<Vec<Contact>, DomainError>;
+    async fn search_contacts(
+        &self,
+        address_book_id: &Uuid,
+        query: &str,
+    ) -> Result<Vec<Contact>, DomainError>;
+
+    // ── Contact groups ───────────────────────────────────────────
+    async fn create_group(&self, group: ContactGroup) -> Result<ContactGroup, DomainError>;
+    async fn update_group(&self, group: ContactGroup) -> Result<ContactGroup, DomainError>;
+    async fn delete_group(&self, id: &Uuid) -> Result<(), DomainError>;
+    async fn get_group_by_id(&self, id: &Uuid) -> Result<Option<ContactGroup>, DomainError>;
+    async fn get_groups_by_address_book(
+        &self,
+        address_book_id: &Uuid,
+    ) -> Result<Vec<ContactGroup>, DomainError>;
+
+    // ── Group membership ─────────────────────────────────────────
+    async fn add_contact_to_group(
+        &self,
+        group_id: &Uuid,
+        contact_id: &Uuid,
+    ) -> Result<(), DomainError>;
+    async fn remove_contact_from_group(
+        &self,
+        group_id: &Uuid,
+        contact_id: &Uuid,
+    ) -> Result<(), DomainError>;
+    async fn get_contacts_in_group(&self, group_id: &Uuid) -> Result<Vec<Contact>, DomainError>;
+    async fn get_groups_for_contact(
+        &self,
+        contact_id: &Uuid,
+    ) -> Result<Vec<ContactGroup>, DomainError>;
+}
+
 pub trait AddressBookUseCase: Send + Sync + 'static {
     // Address Book operations
     async fn create_address_book(
@@ -37,23 +126,6 @@ pub trait AddressBookUseCase: Send + Sync + 'static {
         user_id: Uuid,
     ) -> Result<Vec<AddressBookDto>, DomainError>;
     async fn list_public_address_books(&self) -> Result<Vec<AddressBookDto>, DomainError>;
-
-    // Address Book sharing
-    async fn share_address_book(
-        &self,
-        dto: ShareAddressBookDto,
-        user_id: Uuid,
-    ) -> Result<(), DomainError>;
-    async fn unshare_address_book(
-        &self,
-        dto: UnshareAddressBookDto,
-        user_id: Uuid,
-    ) -> Result<(), DomainError>;
-    async fn get_address_book_shares(
-        &self,
-        address_book_id: &str,
-        user_id: Uuid,
-    ) -> Result<Vec<(String, bool)>, DomainError>;
 }
 
 pub trait ContactUseCase: Send + Sync + 'static {

src/application/ports/file_ports.rs

@@ -75,7 +75,12 @@ pub trait FileUploadUseCase: Send + Sync + 'static {
     /// `updated_by` column reflects the principal that performed the
     /// PUT — not the file's existing owner (D2 shared drives let
     /// non-owners overwrite content).
-    async fn update_file_streaming(
+    /// `_with_perms` suffix (AGENTS.md AuthZ convention): the
+    /// implementation calls `authz.require(caller, Update, File(id))`
+    /// on the overwrite branch and `authz.require(caller, Create,
+    /// Folder|Drive(id))` on the new-file branch. Handlers just plumb
+    /// `caller_id` through — no protocol-layer authz.
+    async fn update_file_streaming_with_perms(
         &self,
         path: &str,
         drive_id: Uuid,
@@ -241,23 +246,21 @@ pub trait FileRetrievalUseCase: Send + Sync + 'static {
             .collect())
     }
 
-    /// Like [`list_files_batch`], but scoped to a specific owner.
+    /// Like [`list_files_batch`], but scoped to a specific caller.
     ///
-    /// Used by streaming WebDAV PROPFIND so that each user only sees their
-    /// own files, even in shared folder_id namespaces.
+    /// Used by streaming WebDAV PROPFIND. Post-D7 the concrete
+    /// implementation in `FileRetrievalService` uses drive-membership
+    /// grants; this default falls back to the unscoped listing (the
+    /// caller passes through `owner_id` for interface parity but the
+    /// stub can't apply a real filter without a repo lookup).
     async fn list_files_batch_with_perms(
         &self,
         folder_id: Option<&str>,
-        owner_id: Uuid,
+        _owner_id: Uuid,
         offset: i64,
         limit: i64,
     ) -> Result<Vec<FileDto>, DomainError> {
-        let all = self.list_files_batch(folder_id, offset, limit).await?;
-        let owner_str = owner_id.to_string();
-        Ok(all
-            .into_iter()
-            .filter(|f| f.owner_id.as_deref().is_some_and(|o| o == owner_str))
-            .collect())
+        self.list_files_batch(folder_id, offset, limit).await
     }
 }
 

src/application/ports/storage_ports.rs

@@ -1,6 +1,5 @@
 use bytes::Bytes;
 use futures::Stream;
-use serde_json::Value;
 use std::path::PathBuf;
 use std::pin::Pin;
 use uuid::Uuid;
@@ -31,40 +30,9 @@ pub trait FileReadPort: Send + Sync + 'static {
 
     async fn get_file_or_trashed(&self, id: &str) -> Result<File, DomainError>;
 
-    /// Gets a file by its ID, scoped to a specific owner.
-    ///
-    /// Returns `NotFound` if the file does not exist **or** belongs to a
-    /// different user.  This is the primary IDOR-safe accessor — handlers
-    /// serving end-user requests should always prefer this over `get_file`.
-    async fn get_file_for_owner(&self, id: &str, owner_id: Uuid) -> Result<File, DomainError>;
-
-    /// Verifies that the file identified by `id` belongs to `owner_id`.
-    ///
-    /// Returns `Ok(())` on success or `NotFound` when the file does not
-    /// exist or belongs to another user.
-    async fn verify_file_owner(&self, id: &str, owner_id: Uuid) -> Result<(), DomainError> {
-        self.get_file_for_owner(id, owner_id).await.map(|_| ())
-    }
-
     /// Lists files in a folder.
     async fn list_files(&self, folder_id: Option<&str>) -> Result<Vec<File>, DomainError>;
 
-    /// Lists files in a folder scoped to a specific owner (SQL-level).
-    ///
-    /// Default falls back to `list_files` + in-memory filter.
-    /// Repositories should override with a direct `AND user_id = $N` query.
-    async fn list_files_for_owner(
-        &self,
-        folder_id: Option<&str>,
-        owner_id: Uuid,
-    ) -> Result<Vec<File>, DomainError> {
-        let all = self.list_files(folder_id).await?;
-        Ok(all
-            .into_iter()
-            .filter(|f| f.owner_id() == Some(owner_id))
-            .collect())
-    }
-
     /// Gets content as a stream (ideal for large files).
     async fn get_file_stream(
         &self,
@@ -155,25 +123,6 @@ pub trait FileReadPort: Send + Sync + 'static {
         Ok(all.into_iter().skip(start).take(end - start).collect())
     }
 
-    /// Like [`list_files_batch`], but only returns files owned by `owner_id`.
-    ///
-    /// Used by streaming WebDAV PROPFIND to list files scoped to the
-    /// authenticated user, preventing cross-user data leakage.
-    async fn list_files_batch_for_owner(
-        &self,
-        folder_id: Option<&str>,
-        owner_id: Uuid,
-        offset: i64,
-        limit: i64,
-    ) -> Result<Vec<File>, DomainError> {
-        // Default: filter in-memory (repos should override with SQL)
-        let all = self.list_files_batch(folder_id, offset, limit).await?;
-        Ok(all
-            .into_iter()
-            .filter(|f| f.owner_id() == Some(owner_id))
-            .collect())
-    }
-
     /// Streams every file in the subtree rooted at `folder_id`.
     ///
     /// Uses an ltree `<@` join against `storage.folders` so the entire
@@ -195,7 +144,10 @@ pub trait FileReadPort: Send + Sync + 'static {
     /// # Arguments
     /// * `folder_id` - Optional folder ID to scope the search (for recursive search, pass None)
     /// * `criteria` - Search criteria including name_contains, file_types, date ranges, size ranges
-    /// * `user_id` - User ID for ownership filtering
+    /// * `caller_id` - Caller user id — scoped by drive-membership grants
+    ///   (`role_grants` on `resource_type='drive'`) rather than the legacy
+    ///   `files.user_id` column. Group memberships (direct + transitive)
+    ///   are expanded inline via `storage.caller_group_ids($caller)`.
     ///
     /// # Returns
     /// A tuple of (files, total_count) where files are paginated and filtered
@@ -203,36 +155,39 @@ pub trait FileReadPort: Send + Sync + 'static {
         &self,
         folder_id: Option<&str>,
         criteria: &SearchCriteriaDto,
-        user_id: Uuid,
+        caller_id: Uuid,
     ) -> Result<(Vec<File>, usize), DomainError>;
 
     /// Search files recursively in a folder subtree using ltree.
     ///
     /// When `root_folder_id` is Some, uses ltree descendant queries to find
-    /// all files within the subtree rooted at that folder. When None, searches
-    /// all files for the user. This replaces the O(N) recursive spawn-per-folder
-    /// approach with O(1) SQL queries.
+    /// all files within the subtree rooted at that folder. When None,
+    /// delegates to `search_files_paginated`.
+    ///
+    /// Post-PR-B: scoped by drive-membership grants (same semantics as
+    /// `search_files_paginated`), not by `files.user_id`.
     ///
     /// Returns a tuple of (matching files, total count for pagination).
     async fn search_files_in_subtree(
         &self,
         root_folder_id: Option<&str>,
         criteria: &SearchCriteriaDto,
-        user_id: Uuid,
+        caller_id: Uuid,
     ) -> Result<(Vec<File>, usize), DomainError> {
         // Default: delegate to paginated search (non-recursive fallback)
-        self.search_files_paginated(root_folder_id, criteria, user_id)
+        self.search_files_paginated(root_folder_id, criteria, caller_id)
             .await
     }
 
     /// Count files matching the search criteria (without loading them).
     ///
     /// Used for pagination metadata without fetching the actual files.
+    /// Same drive-membership scoping as `search_files_paginated`.
     async fn count_files(
         &self,
         folder_id: Option<&str>,
         criteria: &SearchCriteriaDto,
-        user_id: Uuid,
+        caller_id: Uuid,
     ) -> Result<usize, DomainError>;
 
     /// Return up to `limit` files whose name contains `query` (case-insensitive).
@@ -490,9 +445,3 @@ pub trait StorageUsagePort: Send + Sync + 'static {
         additional_bytes: u64,
     ) -> Result<(), DomainError>;
 }
-
-/// Generic storage service interface for calendar and contact services
-pub trait StorageUseCase: Send + Sync + 'static {
-    /// Handle a request with the specified action and parameters
-    async fn handle_request(&self, action: &str, params: Value) -> Result<Value, DomainError>;
-}

src/application/services/calendar_service.rs

@@ -1,4 +1,5 @@
 use chrono::{DateTime, Utc};
+use std::collections::HashSet;
 use std::sync::Arc;
 use uuid::Uuid;
 
@@ -6,17 +7,80 @@ use crate::application::dtos::calendar_dto::{
     CalendarDto, CalendarEventDto, CreateCalendarDto, CreateEventDto, CreateEventICalDto,
     UpdateCalendarDto, UpdateEventDto,
 };
+use crate::application::ports::authorization_ports::AuthorizationEngine;
 use crate::application::ports::calendar_ports::{CalendarStoragePort, CalendarUseCase};
 use crate::common::errors::{DomainError, ErrorKind};
+use crate::domain::services::authorization::{Permission, Resource, Role, Subject};
 use crate::infrastructure::adapters::calendar_storage_adapter::CalendarStorageAdapter;
+use crate::infrastructure::services::pg_acl_engine::PgAclEngine;
 
+/// Calendar service — the CalDAV / REST entry point for every calendar
+/// or event operation. Every method routes through `AuthorizationEngine`;
+/// the pre-Round-3 `check_calendar_access` bespoke helper is gone.
+///
+/// Ownership + sharing live entirely in `storage.role_grants`
+/// (`resource_type='calendar'`). `caldav.calendars.owner_id` stays for
+/// provenance and legacy queries but is no longer consulted for access
+/// decisions.
 pub struct CalendarService {
     calendar_storage: Arc<CalendarStorageAdapter>,
+    /// ReBAC engine — every user-facing method calls `authz.require`
+    /// with the appropriate `Permission`. `create_calendar` also
+    /// uses it to seed an Owner grant for the caller so the common
+    /// "owning my own calendar" case takes a single indexed
+    /// role_grants lookup.
+    authz: Arc<PgAclEngine>,
 }
 
 impl CalendarService {
-    pub fn new(calendar_storage: Arc<CalendarStorageAdapter>) -> Self {
-        Self { calendar_storage }
+    pub fn new(calendar_storage: Arc<CalendarStorageAdapter>, authz: Arc<PgAclEngine>) -> Self {
+        Self {
+            calendar_storage,
+            authz,
+        }
+    }
+
+    /// Parse `calendar_id` and enforce `permission` on `Resource::Calendar(uuid)`.
+    /// On denial `authz.require` returns `NotFound` (anti-enum — same
+    /// shape as "no such calendar") and emits the `authz.denied` audit
+    /// line. Returns the parsed UUID on success so the caller doesn't
+    /// have to parse it a second time.
+    async fn require_calendar_perm(
+        &self,
+        calendar_id: &str,
+        caller_id: Uuid,
+        permission: Permission,
+    ) -> Result<Uuid, DomainError> {
+        let uuid = Uuid::parse_str(calendar_id)
+            .map_err(|_| DomainError::new(ErrorKind::InvalidInput, "Calendar", "Invalid ID"))?;
+        self.authz
+            .require(
+                Subject::User(caller_id),
+                permission,
+                Resource::Calendar(uuid),
+            )
+            .await?;
+        Ok(uuid)
+    }
+
+    /// Check `permission` on a calendar without throwing. Used by the
+    /// read paths that also allow a public-calendar bypass — they need
+    /// a bool, not a `Result<(), NotFound>`.
+    async fn has_calendar_perm(
+        &self,
+        calendar_id: &str,
+        caller_id: Uuid,
+        permission: Permission,
+    ) -> Result<bool, DomainError> {
+        let uuid = Uuid::parse_str(calendar_id)
+            .map_err(|_| DomainError::new(ErrorKind::InvalidInput, "Calendar", "Invalid ID"))?;
+        self.authz
+            .check(
+                Subject::User(caller_id),
+                permission,
+                Resource::Calendar(uuid),
+            )
+            .await
     }
 }
 
@@ -26,9 +90,30 @@ impl CalendarUseCase for CalendarService {
         calendar: CreateCalendarDto,
         user_id: Uuid,
     ) -> Result<CalendarDto, DomainError> {
-        self.calendar_storage
+        // No pre-write gate: creating a calendar is a personal act
+        // (like creating a folder in your own drive). Storage stamps
+        // `owner_id = user_id`; we then seed an Owner role_grant so
+        // the engine's cache warms on first-read.
+        let created = self
+            .calendar_storage
             .create_calendar(calendar, user_id)
-            .await
+            .await?;
+        let calendar_uuid = Uuid::parse_str(&created.id).map_err(|_| {
+            DomainError::internal_error("Calendar", "storage returned invalid calendar id")
+        })?;
+        // `set_role` is idempotent on the `(subject, resource)` unique
+        // key — a re-run (rare — only if storage retried) is a no-op.
+        // `granted_by = user_id` is the self-seeded creation event.
+        self.authz
+            .set_role(
+                user_id,
+                Subject::User(user_id),
+                Role::Owner,
+                Resource::Calendar(calendar_uuid),
+                None,
+            )
+            .await?;
+        Ok(created)
     }
 
     async fn update_calendar(
@@ -37,35 +122,28 @@ impl CalendarUseCase for CalendarService {
         update: UpdateCalendarDto,
         user_id: Uuid,
     ) -> Result<CalendarDto, DomainError> {
-        let has_access = self
-            .calendar_storage
-            .check_calendar_access(calendar_id, user_id)
+        self.require_calendar_perm(calendar_id, user_id, Permission::Update)
             .await?;
-        if !has_access {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "You don't have permission to update this calendar",
-            ));
-        }
         self.calendar_storage
             .update_calendar(calendar_id, update)
             .await
     }
 
     async fn delete_calendar(&self, calendar_id: &str, user_id: Uuid) -> Result<(), DomainError> {
-        let has_access = self
-            .calendar_storage
-            .check_calendar_access(calendar_id, user_id)
+        let uuid = self
+            .require_calendar_perm(calendar_id, user_id, Permission::Delete)
             .await?;
-        if !has_access {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "You don't have permission to delete this calendar",
-            ));
-        }
-        self.calendar_storage.delete_calendar(calendar_id).await
+        self.calendar_storage.delete_calendar(calendar_id).await?;
+        // Wipe every grant on this calendar so a re-used UUID (impossible
+        // today but cheap to defend against) doesn't inherit stale ACLs.
+        // The storage DELETE won't cascade to `storage.role_grants` — the
+        // legacy `caldav.calendar_shares` had an FK, `role_grants`
+        // doesn't (it's cross-schema).
+        let _ = self
+            .authz
+            .revoke_all_for_resource(Resource::Calendar(uuid))
+            .await;
+        Ok(())
     }
 
     async fn get_calendar(
@@ -74,28 +152,52 @@ impl CalendarUseCase for CalendarService {
         user_id: Uuid,
     ) -> Result<CalendarDto, DomainError> {
         let calendar = self.calendar_storage.get_calendar(calendar_id).await?;
-        let has_access = self
-            .calendar_storage
-            .check_calendar_access(calendar_id, user_id)
-            .await?;
-        if !has_access && !calendar.is_public {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "You don't have permission to view this calendar",
-            ));
+        // Public-calendar bypass: anonymous-ish read. `check` returns
+        // bool (no throw); combine with the public flag before
+        // deciding.
+        let allowed = calendar.is_public
+            || self
+                .has_calendar_perm(calendar_id, user_id, Permission::Read)
+                .await?;
+        if !allowed {
+            return Err(DomainError::not_found("Calendar", calendar_id));
         }
         Ok(calendar)
     }
 
     async fn list_my_calendars(&self, user_id: Uuid) -> Result<Vec<CalendarDto>, DomainError> {
-        self.calendar_storage.list_calendars_by_owner(user_id).await
-    }
+        // Post-Round-3 semantics: every calendar the caller has any
+        // grant on — owned + shared, one union. The pre-Round-3
+        // `list_calendars_by_owner` returned owner-only; shared
+        // calendars never surfaced through this method. See
+        // `docs/plan/caldav-carddav-migration-to-authz.md`.
+        let grants = self
+            .authz
+            .list_incoming_grants(Subject::User(user_id))
+            .await?;
 
-    async fn list_shared_calendars(&self, user_id: Uuid) -> Result<Vec<CalendarDto>, DomainError> {
-        self.calendar_storage
-            .list_calendars_shared_with_user(user_id)
-            .await
+        // Deduplicate — a user can hold multiple grants on the same
+        // calendar (direct + group-inherited). We only need one DTO
+        // per resource.
+        let calendar_ids: HashSet<Uuid> = grants
+            .into_iter()
+            .filter_map(|g| match g.resource {
+                Resource::Calendar(id) => Some(id),
+                _ => None,
+            })
+            .collect();
+
+        // Hydrate DTOs. `get_calendar` misses on trashed / deleted
+        // calendars — those are dropped from the listing rather than
+        // erroring, so a lifecycle-race doesn't turn a PROPFIND into
+        // a 5xx.
+        let mut out = Vec::with_capacity(calendar_ids.len());
+        for id in calendar_ids {
+            if let Ok(dto) = self.calendar_storage.get_calendar(&id.to_string()).await {
+                out.push(dto);
+            }
+        }
+        Ok(out)
     }
 
     async fn list_public_calendars(
@@ -103,6 +205,8 @@ impl CalendarUseCase for CalendarService {
         limit: Option<i64>,
         offset: Option<i64>,
     ) -> Result<Vec<CalendarDto>, DomainError> {
+        // No caller gate: public listing by definition. Storage
+        // filters on `is_public = true`.
         let limit = limit.unwrap_or(100);
         let offset = offset.unwrap_or(0);
         self.calendar_storage
@@ -110,90 +214,13 @@ impl CalendarUseCase for CalendarService {
             .await
     }
 
-    async fn share_calendar(
-        &self,
-        calendar_id: &str,
-        target_user_id: Uuid,
-        access_level: &str,
-        caller_user_id: Uuid,
-    ) -> Result<(), DomainError> {
-        let calendar = self.calendar_storage.get_calendar(calendar_id).await?;
-        if calendar.owner_id != caller_user_id.to_string() {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "Only the calendar owner can change sharing settings",
-            ));
-        }
-        match access_level {
-            "read" | "write" | "owner" => {}
-            _ => {
-                return Err(DomainError::new(
-                    ErrorKind::InvalidInput,
-                    "Calendar",
-                    format!(
-                        "Invalid access level: {}. Valid values are: read, write, owner",
-                        access_level
-                    ),
-                ));
-            }
-        }
-        self.calendar_storage
-            .share_calendar(calendar_id, target_user_id, access_level)
-            .await
-    }
-
-    async fn remove_calendar_sharing(
-        &self,
-        calendar_id: &str,
-        target_user_id: Uuid,
-        caller_user_id: Uuid,
-    ) -> Result<(), DomainError> {
-        let calendar = self.calendar_storage.get_calendar(calendar_id).await?;
-        if calendar.owner_id != caller_user_id.to_string() {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "Only the calendar owner can change sharing settings",
-            ));
-        }
-        self.calendar_storage
-            .remove_calendar_sharing(calendar_id, target_user_id)
-            .await
-    }
-
-    async fn get_calendar_shares(
-        &self,
-        calendar_id: &str,
-        user_id: Uuid,
-    ) -> Result<Vec<(String, String)>, DomainError> {
-        let calendar = self.calendar_storage.get_calendar(calendar_id).await?;
-        if calendar.owner_id != user_id.to_string() {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "Only the calendar owner can view sharing settings",
-            ));
-        }
-        self.calendar_storage.get_calendar_shares(calendar_id).await
-    }
-
     async fn create_event(
         &self,
         event: CreateEventDto,
         user_id: Uuid,
     ) -> Result<CalendarEventDto, DomainError> {
-        let has_access = self
-            .calendar_storage
-            .check_calendar_access(&event.calendar_id, user_id)
+        self.require_calendar_perm(&event.calendar_id, user_id, Permission::Create)
             .await?;
-        if !has_access {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "You don't have permission to add events to this calendar",
-            ));
-        }
         self.calendar_storage.create_event(event).await
     }
 
@@ -202,17 +229,8 @@ impl CalendarUseCase for CalendarService {
         event: CreateEventICalDto,
         user_id: Uuid,
     ) -> Result<CalendarEventDto, DomainError> {
-        let has_access = self
-            .calendar_storage
-            .check_calendar_access(&event.calendar_id, user_id)
+        self.require_calendar_perm(&event.calendar_id, user_id, Permission::Create)
             .await?;
-        if !has_access {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "You don't have permission to add events to this calendar",
-            ));
-        }
         self.calendar_storage.create_event_from_ical(event).await
     }
 
@@ -223,33 +241,15 @@ impl CalendarUseCase for CalendarService {
         user_id: Uuid,
     ) -> Result<CalendarEventDto, DomainError> {
         let event = self.calendar_storage.get_event(event_id).await?;
-        let has_access = self
-            .calendar_storage
-            .check_calendar_access(&event.calendar_id, user_id)
+        self.require_calendar_perm(&event.calendar_id, user_id, Permission::Update)
             .await?;
-        if !has_access {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "You don't have permission to update events in this calendar",
-            ));
-        }
         self.calendar_storage.update_event(event_id, update).await
     }
 
     async fn delete_event(&self, event_id: &str, user_id: Uuid) -> Result<(), DomainError> {
         let event = self.calendar_storage.get_event(event_id).await?;
-        let has_access = self
-            .calendar_storage
-            .check_calendar_access(&event.calendar_id, user_id)
+        self.require_calendar_perm(&event.calendar_id, user_id, Permission::Delete)
             .await?;
-        if !has_access {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "You don't have permission to delete events in this calendar",
-            ));
-        }
         self.calendar_storage.delete_event(event_id).await
     }
 
@@ -259,20 +259,17 @@ impl CalendarUseCase for CalendarService {
         user_id: Uuid,
     ) -> Result<CalendarEventDto, DomainError> {
         let event = self.calendar_storage.get_event(event_id).await?;
-        let has_access = self
-            .calendar_storage
-            .check_calendar_access(&event.calendar_id, user_id)
-            .await?;
         let calendar = self
             .calendar_storage
             .get_calendar(&event.calendar_id)
             .await?;
-        if !has_access && !calendar.is_public {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "You don't have permission to view events in this calendar",
-            ));
+        // Same public-calendar bypass as `get_calendar`.
+        let allowed = calendar.is_public
+            || self
+                .has_calendar_perm(&event.calendar_id, user_id, Permission::Read)
+                .await?;
+        if !allowed {
+            return Err(DomainError::not_found("Event", event_id));
         }
         Ok(event)
     }
@@ -283,17 +280,13 @@ impl CalendarUseCase for CalendarService {
         ical_uid: &str,
         user_id: Uuid,
     ) -> Result<Option<CalendarEventDto>, DomainError> {
-        let has_access = self
-            .calendar_storage
-            .check_calendar_access(calendar_id, user_id)
-            .await?;
         let calendar = self.calendar_storage.get_calendar(calendar_id).await?;
-        if !has_access && !calendar.is_public {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "You don't have permission to view events in this calendar",
-            ));
+        let allowed = calendar.is_public
+            || self
+                .has_calendar_perm(calendar_id, user_id, Permission::Read)
+                .await?;
+        if !allowed {
+            return Err(DomainError::not_found("Calendar", calendar_id));
         }
         self.calendar_storage
             .find_event_by_ical_uid(calendar_id, ical_uid)
@@ -306,17 +299,13 @@ impl CalendarUseCase for CalendarService {
         ical_uids: &[String],
         user_id: Uuid,
     ) -> Result<Vec<CalendarEventDto>, DomainError> {
-        let has_access = self
-            .calendar_storage
-            .check_calendar_access(calendar_id, user_id)
-            .await?;
         let calendar = self.calendar_storage.get_calendar(calendar_id).await?;
-        if !has_access && !calendar.is_public {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "You don't have permission to view events in this calendar",
-            ));
+        let allowed = calendar.is_public
+            || self
+                .has_calendar_perm(calendar_id, user_id, Permission::Read)
+                .await?;
+        if !allowed {
+            return Err(DomainError::not_found("Calendar", calendar_id));
         }
         if ical_uids.is_empty() {
             return Ok(Vec::new());
@@ -333,17 +322,13 @@ impl CalendarUseCase for CalendarService {
         offset: Option<i64>,
         user_id: Uuid,
     ) -> Result<Vec<CalendarEventDto>, DomainError> {
-        let has_access = self
-            .calendar_storage
-            .check_calendar_access(calendar_id, user_id)
-            .await?;
         let calendar = self.calendar_storage.get_calendar(calendar_id).await?;
-        if !has_access && !calendar.is_public {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "You don't have permission to view events in this calendar",
-            ));
+        let allowed = calendar.is_public
+            || self
+                .has_calendar_perm(calendar_id, user_id, Permission::Read)
+                .await?;
+        if !allowed {
+            return Err(DomainError::not_found("Calendar", calendar_id));
         }
         if limit.is_some() || offset.is_some() {
             let limit = limit.unwrap_or(100);
@@ -365,17 +350,13 @@ impl CalendarUseCase for CalendarService {
         end: DateTime<Utc>,
         user_id: Uuid,
     ) -> Result<Vec<CalendarEventDto>, DomainError> {
-        let has_access = self
-            .calendar_storage
-            .check_calendar_access(calendar_id, user_id)
-            .await?;
         let calendar = self.calendar_storage.get_calendar(calendar_id).await?;
-        if !has_access && !calendar.is_public {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Calendar",
-                "You don't have permission to view events in this calendar",
-            ));
+        let allowed = calendar.is_public
+            || self
+                .has_calendar_perm(calendar_id, user_id, Permission::Read)
+                .await?;
+        if !allowed {
+            return Err(DomainError::not_found("Calendar", calendar_id));
         }
         self.calendar_storage
             .get_events_in_time_range(calendar_id, &start, &end)

src/application/services/drive_management_service.rs

@@ -249,6 +249,20 @@ impl DriveManagementService {
             .set_role(caller_id, subject, role, resource, expires_at)
             .await?;
 
+        // Drop the entire drive-role cache for this drive so the new
+        // grant is visible on the very next `check` — without this, a
+        // caller that gets Owner via `POST /api/drives/{id}/members`
+        // then immediately acts on drive content (WebDAV cross-drive
+        // MOVE, admin-driven cleanup, drive management) hits the
+        // stale "no role for this subject on this drive" entry
+        // seeded at some earlier `check`. TTL rescues eventually,
+        // but the storage_cleanup_check.sh drain pattern hits this
+        // race within a single test-second and fails on `authz.denied`
+        // for admin's cascade to files inside.
+        self.authz
+            .invalidate_drive_role_cache_for_drive(drive_id)
+            .await;
+
         // D6 §11: canonical `drive.member_added` audit event covers
         // every successful membership write (add + role-refresh, since
         // the underlying `set_role` is UPSERT — distinguishing the two
@@ -312,6 +326,15 @@ impl DriveManagementService {
 
         self.authz.clear_role(subject, resource).await?;
 
+        // Mirror of `set_member_role`'s cache invalidation: after
+        // clearing a role we MUST drop the `drive_role_cache` entries
+        // targeting this drive, otherwise the just-removed subject's
+        // former role stays visible until TTL expires. Same anti-drift
+        // reason as the sibling add path above.
+        self.authz
+            .invalidate_drive_role_cache_for_drive(drive_id)
+            .await;
+
         // D6 §11: canonical `drive.member_removed` audit event covers
         // every successful removal (owner-driven or admin bypass).
         // `via_admin` replaces the separate
@@ -448,7 +471,7 @@ impl DriveManagementService {
         &self,
         caller_id: Uuid,
         drive_id: Uuid,
-        partial: crate::domain::entities::drive::DrivePolicies,
+        partial: serde_json::Value,
     ) -> Result<crate::domain::entities::drive::DrivePolicies, DomainError> {
         let merged = self
             .drive_repo
@@ -474,6 +497,8 @@ impl DriveManagementService {
             forbid_public_links = merged.forbid_public_links,
             forbid_cross_drive_move = merged.forbid_cross_drive_move,
             forbid_owner_role_change = merged.forbid_owner_role_change,
+            include_in_photo_index = merged.include_in_photo_index,
+            include_in_music_index = merged.include_in_music_index,
             "📜 drive policies updated",
         );
         Ok(merged)

src/application/services/favorites_service.rs

@@ -9,10 +9,12 @@ use crate::application::dtos::favorites_dto::{
     BatchFavoritesResult, BatchFavoritesStats, FavoriteItemDto, FavoriteResourceRow,
     FavoritesCursor,
 };
+use crate::application::ports::authorization_ports::AuthorizationEngine;
 use crate::application::ports::favorites_ports::{FavoritesRepositoryPort, FavoritesUseCase};
-use crate::common::errors::{DomainError, ErrorKind, Result};
-use crate::domain::services::authorization::ResourceKind;
+use crate::common::errors::Result;
+use crate::domain::services::authorization::{Permission, Resource, ResourceKind, Subject};
 use crate::infrastructure::repositories::pg::FavoritesPgRepository;
+use crate::infrastructure::services::pg_acl_engine::PgAclEngine;
 
 /// Implementation of the FavoritesUseCase for managing user favorites.
 ///
@@ -20,12 +22,22 @@ use crate::infrastructure::repositories::pg::FavoritesPgRepository;
 /// accessing the database directly, following hexagonal architecture.
 pub struct FavoritesService {
     repo: Arc<FavoritesPgRepository>,
+    /// ReBAC engine — enforces `Permission::Read` on the referenced
+    /// file/folder before enrolling it into a user's favorites.
+    /// Without this gate the write path is an information oracle:
+    /// listing endpoints JOIN back to `storage.files/folders` and
+    /// return name/mime/size/drive_id for any UUID the caller was
+    /// able to enroll. See `docs/plan/authz_audit/rest_storage.md`.
+    authorization: Arc<PgAclEngine>,
 }
 
 impl FavoritesService {
     /// Create a new FavoritesService with the given repository port
-    pub fn new(repo: Arc<FavoritesPgRepository>) -> Self {
-        Self { repo }
+    pub fn new(repo: Arc<FavoritesPgRepository>, authorization: Arc<PgAclEngine>) -> Self {
+        Self {
+            repo,
+            authorization,
+        }
     }
 
     /// Subset of `(item_id, item_type)` pairs the user has favorited — used to
@@ -60,13 +72,15 @@ impl FavoritesUseCase for FavoritesService {
             item_type, item_id, user_id
         );
 
-        if item_type != "file" && item_type != "folder" {
-            return Err(DomainError::new(
-                ErrorKind::InvalidInput,
-                "Favorites",
-                "Item type must be 'file' or 'folder'",
-            ));
-        }
+        // AuthZ pre-write: caller must have Read on the referenced
+        // resource. Denial routes through `require` → NotFound
+        // (anti-enum, matches the listing shape) + `authz.denied`
+        // audit line. Without this gate the write path was an
+        // information oracle over the whole tenant.
+        let resource = Resource::parse(item_type, item_id)?;
+        self.authorization
+            .require(Subject::User(user_id), Permission::Read, resource)
+            .await?;
 
         self.repo.add_favorite(user_id, item_id, item_type).await?;
         info!(
@@ -125,18 +139,17 @@ impl FavoritesUseCase for FavoritesService {
             user_id
         );
 
-        // Validate all item types
+        // AuthZ pre-write: caller must have Read on every referenced
+        // resource. Fail the whole batch on the first denial so the
+        // response shape doesn't tell an attacker which items were
+        // valid (partial success would leak the same oracle we
+        // closed on the single-item path). See
+        // `docs/plan/authz_audit/rest_storage.md`.
         for (item_id, item_type) in items {
-            if item_type != "file" && item_type != "folder" {
-                return Err(DomainError::new(
-                    ErrorKind::InvalidInput,
-                    "Favorites",
-                    format!(
-                        "Item type must be 'file' or 'folder' for item '{}'",
-                        item_id
-                    ),
-                ));
-            }
+            let resource = Resource::parse(item_type, item_id)?;
+            self.authorization
+                .require(Subject::User(user_id), Permission::Read, resource)
+                .await?;
         }
 
         let requested = items.len();

src/application/services/file_management_service.rs

@@ -43,6 +43,13 @@ pub struct FileManagementService {
     /// that case the cross-drive move check is skipped (the policy
     /// is silently off). Production DI wires it in.
     drive_repo: Option<Arc<dyn crate::domain::repositories::drive_repository::DriveRepository>>,
+    /// Storage-usage service — used to pre-check the destination
+    /// drive's `used_bytes + delta ≤ quota_bytes` invariant on
+    /// cross-drive MOVE, matching the pre-write check the upload path
+    /// already performs. Without it, the check is silently skipped
+    /// (stub/test builders); production DI wires it in.
+    storage_usage:
+        Option<Arc<crate::application::services::storage_usage_service::StorageUsageService>>,
 }
 
 impl FileManagementService {
@@ -67,6 +74,7 @@ impl FileManagementService {
             file_lifecycle_hook: None,
             resource_access_hook: None,
             drive_repo: None,
+            storage_usage: None,
         }
     }
 
@@ -100,6 +108,18 @@ impl FileManagementService {
         self
     }
 
+    /// Wires the storage-usage service so `move_file_with_perms` can
+    /// pre-check the destination drive's quota on cross-drive moves.
+    pub fn with_storage_usage(
+        mut self,
+        storage_usage: Arc<
+            crate::application::services::storage_usage_service::StorageUsageService,
+        >,
+    ) -> Self {
+        self.storage_usage = Some(storage_usage);
+        self
+    }
+
     /// Engine check for a file resource. Parses the id into a `Uuid` and
     /// requires the specified permission.
     async fn require_file_perm(
@@ -338,12 +358,42 @@ impl FileManagementUseCase for FileManagementService {
                         dst_drive_id,
                     },
                 )?;
+                // Destination drive quota: same pre-write check the
+                // upload path already runs (`file_upload_service.rs`
+                // `check_storage_quota`), applied here so a caller
+                // can't sneak content past the drive cap via MOVE.
+                // Denial → `DomainError::QuotaExceeded` → 507
+                // Insufficient Storage. Skipped when `storage_usage`
+                // isn't wired (stub builders) — same shape as the
+                // upload path's skip semantics.
+                if let Some(storage_usage) = &self.storage_usage
+                    && let Some(size_bytes) = storage_usage.file_bytes(file_uuid).await?
+                    && let Ok(size_u64) = u64::try_from(size_bytes)
+                {
+                    storage_usage
+                        .check_drive_quota(dst_drive_id, size_u64)
+                        .await?;
+                }
                 cross_drive = Some((src_drive_id, dst_drive_id));
             }
         }
 
         let dto = self.move_file(file_id, folder_id, caller_id).await?;
 
+        // Cross-drive move invalidates the file's `owner_cache` entry
+        // in the authz engine — the cache assumed drive_id stability
+        // that no longer holds. Without this call the drive-role
+        // precheck at `check_inner` steers to the (stale) source
+        // drive and legitimate Delete/Update by a destination-drive
+        // role-holder returns 404 for up to the cache TTL.
+        if cross_drive.is_some()
+            && let Ok(file_uuid) = Uuid::parse_str(file_id)
+        {
+            self.authz
+                .invalidate_owner_cache_for_resource(Resource::File(file_uuid))
+                .await;
+        }
+
         // D6 §11 audit: emit only when the move actually crossed a
         // drive boundary. Same-drive moves are too noisy to audit at
         // info — operators care about the cross-drive case for
@@ -375,6 +425,31 @@ impl FileManagementUseCase for FileManagementService {
             .await?;
         self.require_target_folder_perm(target_folder_id.as_deref(), Permission::Create, caller_id)
             .await?;
+
+        // Destination drive quota: COPY creates a new file row that
+        // counts against the destination drive's `used_bytes` even
+        // though blob dedup means no new bytes hit the store. Same
+        // pre-flight shape the delta-upload path already uses.
+        // Skipped when `storage_usage` isn't wired (stub builders) or
+        // `target_folder_id` is None (root namespace — same-drive
+        // semantics inherit the source's cap coverage). Denial →
+        // `QuotaExceeded` → 507.
+        if let (Some(storage_usage), Some(target_folder)) =
+            (&self.storage_usage, target_folder_id.as_deref())
+        {
+            let file_uuid =
+                Uuid::parse_str(file_id).map_err(|_| DomainError::not_found("File", file_id))?;
+            let target_folder_uuid = Uuid::parse_str(target_folder)
+                .map_err(|_| DomainError::not_found("Folder", target_folder))?;
+            if let Some(size_bytes) = storage_usage.file_bytes(file_uuid).await?
+                && let Ok(size_u64) = u64::try_from(size_bytes)
+            {
+                storage_usage
+                    .check_drive_quota_by_folder(target_folder_uuid, size_u64)
+                    .await?;
+            }
+        }
+
         self.copy_file(file_id, target_folder_id, new_name.as_deref(), caller_id)
             .await
     }
@@ -453,6 +528,26 @@ impl FileManagementUseCase for FileManagementService {
             .await?;
         self.require_target_folder_perm(target_parent_id.as_deref(), Permission::Create, caller_id)
             .await?;
+
+        // Destination drive quota: sum the subtree's non-trashed files
+        // and refuse if the destination couldn't hold them. Skipped
+        // when `storage_usage` isn't wired or the target is root
+        // (same rationale as `copy_file_with_perms`).
+        if let (Some(storage_usage), Some(target_parent)) =
+            (&self.storage_usage, target_parent_id.as_deref())
+        {
+            let source_uuid = Uuid::parse_str(source_folder_id)
+                .map_err(|_| DomainError::not_found("Folder", source_folder_id))?;
+            let target_parent_uuid = Uuid::parse_str(target_parent)
+                .map_err(|_| DomainError::not_found("Folder", target_parent))?;
+            let subtree_bytes = storage_usage.folder_subtree_bytes(source_uuid).await?;
+            if let Ok(subtree_u64) = u64::try_from(subtree_bytes) {
+                storage_usage
+                    .check_drive_quota_by_folder(target_parent_uuid, subtree_u64)
+                    .await?;
+            }
+        }
+
         self.copy_folder_tree(source_folder_id, target_parent_id, dest_name)
             .await
     }

src/application/services/file_retrieval_service.rs

@@ -343,19 +343,17 @@ impl FileRetrievalUseCase for FileRetrievalService {
         folder_id: Option<&str>,
         owner_id: Uuid,
     ) -> Result<Vec<FileDto>, DomainError> {
-        if folder_id.is_some() {
-            // folder id is defined, check permissions
-            self.require_target_folder_perm(folder_id, Permission::Read, owner_id)
-                .await?;
-            self.list_files(folder_id).await
-        } else {
-            // no folder id, get owners's files' root
-            let files = self
-                .file_read
-                .list_files_for_owner(folder_id, owner_id)
-                .await?;
-            Ok(files.into_iter().map(FileDto::from).collect())
+        // Files always have a `folder_id` in the D0+ model — there is no
+        // longer any concept of "root-level files". A `None` from the
+        // caller means the query string was missing `folder_id`; reject
+        // with a clear error rather than returning an empty set from a
+        // meaningless root-level query.
+        if folder_id.is_none() {
+            return Err(DomainError::validation_error("folder_id is required"));
         }
+        self.require_target_folder_perm(folder_id, Permission::Read, owner_id)
+            .await?;
+        self.list_files(folder_id).await
     }
 
     async fn get_file_stream(
@@ -470,20 +468,21 @@ impl FileRetrievalUseCase for FileRetrievalService {
         offset: i64,
         limit: i64,
     ) -> Result<Vec<FileDto>, DomainError> {
-        if folder_id.is_some() {
-            // folder id is defined, check permissions
-            self.require_target_folder_perm(folder_id, Permission::Read, owner_id)
-                .await?;
-            let files = self
-                .file_read
-                .list_files_batch(folder_id, offset, limit)
-                .await?;
-            return Ok(files.into_iter().map(FileDto::from).collect());
-        }
-
+        // Post-D0: every file lives in a folder — `storage.files.folder_id`
+        // is NOT NULL. `folder_id = None` means the caller is asking for
+        // "root-level files", which by design return an empty set: the
+        // WebDAV synthetic root only lists drive-root folders as
+        // children. Skip the DB round-trip and the pre-D7 owner-fallback
+        // query (which used to hit `_for_owner` and would have driven
+        // the `files.user_id` filter this refactor is retiring).
+        let Some(_) = folder_id else {
+            return Ok(Vec::new());
+        };
+        self.require_target_folder_perm(folder_id, Permission::Read, owner_id)
+            .await?;
         let files = self
             .file_read
-            .list_files_batch_for_owner(folder_id, owner_id, offset, limit)
+            .list_files_batch(folder_id, offset, limit)
             .await?;
         Ok(files.into_iter().map(FileDto::from).collect())
     }

src/application/services/file_upload_service.rs

@@ -42,6 +42,16 @@ pub struct FileUploadService {
     /// `(file_id, blob_hash, content_type)`; the recording side needs the
     /// `caller_id` the service already has in hand.
     resource_access_hook: Option<Arc<dyn ResourceAccessHook>>,
+    /// ReBAC engine — enforces `Permission::Update` on
+    /// overwrite-existing and `Permission::Create` on new-file paths
+    /// inside `update_file_streaming_with_perms`. Optional at the
+    /// struct level for the minimal test constructors (`new`,
+    /// `new_with_read`) but the WebDAV/NC/WOPI put paths refuse
+    /// (fail-closed internal error) if this isn't wired. Set by
+    /// either `with_instant_upload` or `with_authorization` — both
+    /// stash the same Arc so DI callers wiring instant upload get
+    /// the streaming gate for free.
+    authorization: Option<Arc<PgAclEngine>>,
     /// Dependencies of the instant-upload path
     /// (`create_file_from_owned_blob_with_perms`); `None` in minimal test
     /// wiring.
@@ -66,6 +76,7 @@ impl FileUploadService {
             content_cache: None,
             file_lifecycle_hook: None,
             resource_access_hook: None,
+            authorization: None,
             instant_upload: None,
         }
     }
@@ -82,18 +93,34 @@ impl FileUploadService {
             content_cache: None,
             file_lifecycle_hook: None,
             resource_access_hook: None,
+            authorization: None,
             instant_upload: None,
         }
     }
 
+    /// Wires the authorization engine used by
+    /// `update_file_streaming_with_perms` on the WebDAV / NC / WOPI
+    /// PUT path. Independent of `with_instant_upload` so callers can
+    /// enable the streaming gate without also opting into the
+    /// dedup-instant-upload check (test wiring, minimal deployments).
+    pub fn with_authorization(mut self, authz: Arc<PgAclEngine>) -> Self {
+        self.authorization = Some(authz);
+        self
+    }
+
     /// Wires the authorization engine, dedup index and quota service that
     /// power the instant-upload path.
+    ///
+    /// Also stashes the `authz` handle in `self.authorization` so
+    /// DI callers wiring instant upload get the streaming-put gate
+    /// for free — a single `Arc` clone, no behavioural coupling.
     pub fn with_instant_upload(
         mut self,
         authz: Arc<PgAclEngine>,
         dedup: Arc<DedupService>,
         quota: Arc<StorageUsageService>,
     ) -> Self {
+        self.authorization = Some(authz.clone());
         self.instant_upload = Some(InstantUploadDeps {
             authz,
             dedup,
@@ -296,7 +323,6 @@ impl FileUploadService {
             parts.folder_id,
             parts.created_at,
             updated_at as u64,
-            parts.owner_id,
             new_hash,
         )
         .map_err(|e| DomainError::internal_error("FileUpload", format!("rebuild entity: {e}")))?;
@@ -317,19 +343,22 @@ impl FileUploadService {
     /// Incremental (`+size`, O(1)) and fire-and-forget on a background task, so
     /// it adds neither latency nor a `SUM(size)` over the user's whole library
     /// to the upload path (the previous full recompute was O(N) per upload,
-    /// O(N²) for a bulk upload). Keyed by the file's `owner_id`; drift — e.g.
-    /// deletes, which don't decrement — is reconciled by the periodic sweep. A
-    /// DTO without a resolvable owner is simply left to that sweep.
-    fn maybe_update_storage_usage(&self, file: &FileDto) {
+    /// O(N²) for a bulk upload). Drift — e.g. deletes, which don't decrement —
+    /// is reconciled by the periodic sweep.
+    ///
+    /// Post-D7: `file.owner_id` is now nullable and unpopulated on new
+    /// rows, so the envelope owner comes from `caller_id` (the user who
+    /// just did the upload). The user-side delta is guarded by
+    /// `add_user_storage_usage_delta_if_personal` — it only fires when
+    /// the target drive is `kind='personal'`, so a shared-drive upload
+    /// still doesn't touch any user envelope.
+    fn maybe_update_storage_usage(&self, file: &FileDto, caller_id: Uuid) {
         let Some(storage_service) = &self.storage_usage_service else {
             return;
         };
         let delta = file.size as i64;
 
-        let owner = file
-            .owner_id
-            .as_deref()
-            .and_then(|s| Uuid::parse_str(s).ok());
+        let owner = Some(caller_id);
         let folder = file
             .folder_id
             .as_deref()
@@ -410,7 +439,7 @@ impl FileUploadUseCase for FileUploadService {
             "📡 STREAMING UPLOAD: {} ({} bytes, ID: {})",
             name, blob.size, dto.id
         );
-        self.maybe_update_storage_usage(&dto);
+        self.maybe_update_storage_usage(&dto, caller_id);
         if let Some(hook) = &self.file_lifecycle_hook {
             hook.on_file_created(&dto.id, &dto.content_hash, &dto.mime_type, blob.is_new_blob);
         }
@@ -422,7 +451,16 @@ impl FileUploadUseCase for FileUploadService {
 
     /// Swap the content of the file at `path` to an already-ingested blob,
     /// creating the file when it doesn't exist (WebDAV/NextCloud/WOPI PUT).
-    async fn update_file_streaming(
+    ///
+    /// AuthZ (post-Drive audit Round 2 fix): overwrite path requires
+    /// `Update` on the target file; new-file path requires `Create`
+    /// on the parent folder (or on the drive when writing at drive
+    /// root). Fail-closed if the engine wasn't wired — this method
+    /// is the last line of defence between a Viewer/Commenter drive
+    /// member and cross-tenant PUT. See
+    /// `docs/plan/authz_audit/nextcloud.md` and the sibling native
+    /// `/webdav/*` handler.
+    async fn update_file_streaming_with_perms(
         &self,
         path: &str,
         drive_id: Uuid,
@@ -431,10 +469,33 @@ impl FileUploadUseCase for FileUploadService {
         modified_at: Option<i64>,
         caller_id: Uuid,
     ) -> Result<FileDto, DomainError> {
+        let Some(authz) = &self.authorization else {
+            return Err(DomainError::internal_error(
+                "FileUpload",
+                "update_file_streaming_with_perms called without authorization engine wired",
+            ));
+        };
+
         // Try to find the existing file first
         if let Some(file_read) = &self.file_read
             && let Some(file) = file_read.find_file_by_path(path, drive_id).await?
         {
+            // Overwrite branch — caller must have `Update` on the
+            // target file. Denial routes through `require` → 404
+            // (anti-enum, matches read-side shape). Before the D7
+            // audit this whole branch ran unchecked; Viewer members
+            // of shared drives could PUT freely.
+            let file_uuid = Uuid::parse_str(file.id()).map_err(|_| {
+                DomainError::internal_error("FileUpload", "invalid file id from repository")
+            })?;
+            authz
+                .require(
+                    Subject::User(caller_id),
+                    Permission::Update,
+                    Resource::File(file_uuid),
+                )
+                .await?;
+
             let file_id = file.id().to_string();
             let (new_hash, updated_at) = self
                 .file_write
@@ -464,7 +525,6 @@ impl FileUploadUseCase for FileUploadService {
                 parts.folder_id,
                 parts.created_at,
                 updated_at as u64,
-                parts.owner_id,
                 new_hash,
             )
             .map_err(|e| {
@@ -504,6 +564,32 @@ impl FileUploadUseCase for FileUploadService {
             None
         };
 
+        // Create branch — caller must have `Create` on the parent
+        // scope. Two cases:
+        //   * `parent_id.is_some()` → caller needs Create on the
+        //     parent Folder resource.
+        //   * `parent_id.is_none()` → the write lands at the drive
+        //     root (either the path was single-segment, or the
+        //     parent-folder lookup failed). We require Create on
+        //     the Drive itself — bundled with owner/editor/contributor
+        //     role_grants, refused for viewer/commenter.
+        let create_resource = match &parent_id {
+            Some(pid) => {
+                let uuid = Uuid::parse_str(pid).map_err(|_| {
+                    DomainError::internal_error("FileUpload", "invalid parent folder id")
+                })?;
+                Resource::Folder(uuid)
+            }
+            None => Resource::Drive(drive_id),
+        };
+        authz
+            .require(
+                Subject::User(caller_id),
+                Permission::Create,
+                create_resource,
+            )
+            .await?;
+
         let is_new_blob = blob.is_new_blob;
         let created = self
             .file_write

src/application/services/folder_service.rs

@@ -31,6 +31,11 @@ pub struct FolderService {
     /// that case the cross-drive move check is skipped (the policy is
     /// silently off). Production DI wires it via `with_drive_repo`.
     drive_repo: Option<Arc<dyn crate::domain::repositories::drive_repository::DriveRepository>>,
+    /// Storage-usage service — used to pre-check the destination
+    /// drive's `used_bytes + subtree_bytes ≤ quota_bytes` invariant
+    /// on cross-drive MOVE. Silently skipped when unwired (stubs).
+    storage_usage:
+        Option<Arc<crate::application::services::storage_usage_service::StorageUsageService>>,
 }
 
 impl FolderService {
@@ -45,6 +50,7 @@ impl FolderService {
             authz,
             file_lifecycle,
             drive_repo: None,
+            storage_usage: None,
         }
     }
 
@@ -60,6 +66,19 @@ impl FolderService {
         self
     }
 
+    /// Wires the storage-usage service so `move_folder_with_perms`
+    /// can pre-check the destination drive's quota on cross-drive
+    /// folder moves.
+    pub fn with_storage_usage(
+        mut self,
+        storage_usage: Arc<
+            crate::application::services::storage_usage_service::StorageUsageService,
+        >,
+    ) -> Self {
+        self.storage_usage = Some(storage_usage);
+        self
+    }
+
     /// Batch counterpart of `get_folder`: resolve many folder ids in ONE
     /// query instead of one per id. Like `get_folder` it performs no
     /// per-folder authorization — both current callers (ACL grant listing,
@@ -358,18 +377,18 @@ impl FolderUseCase for FolderService {
                 .await?;
             return self.list_folders(parent_id).await;
         }
-        // No parent → list the user's root folders.
+        // No parent → list the caller's readable root folders. The
+        // predicate scopes by drive-membership grants (post-PR-B),
+        // closing the pre-D7 gap where the legacy `user_id` filter
+        // surfaced admin-created folders that admin had no role on.
         let folders = self
             .folder_storage
-            .list_folders_by_owner(parent_id, caller_id)
+            .list_root_folders_for_caller(caller_id)
             .await
             .map_err(|e| {
                 DomainError::internal_error(
                     "FolderStorage",
-                    format!(
-                        "Failed to list folders for owner '{}' in parent {:?}: {}",
-                        caller_id, parent_id, e
-                    ),
+                    format!("Failed to list root folders for caller '{caller_id}': {e}"),
                 )
             })?;
         Ok(folders.into_iter().map(FolderDto::from).collect())
@@ -431,24 +450,23 @@ impl FolderUseCase for FolderService {
             return self.list_folders_paginated(parent_id, &pagination).await;
         } else {
             let (folders, total_items) = self
-            .folder_storage
-            .list_folders_by_owner_paginated(
-                parent_id,
-                owner_id,
-                pagination.offset(),
-                pagination.limit(),
-                true,
-            )
-            .await
-            .map_err(|e| {
-                DomainError::internal_error(
-                    "FolderStorage",
-                    format!(
-                        "Failed to list folders for owner '{}' with pagination in parent {:?}: {}",
-                        owner_id, parent_id, e
-                    ),
+                .folder_storage
+                .list_root_folders_for_caller_paginated(
+                    owner_id,
+                    pagination.offset(),
+                    pagination.limit(),
+                    true,
                 )
-            })?;
+                .await
+                .map_err(|e| {
+                    DomainError::internal_error(
+                        "FolderStorage",
+                        format!(
+                            "Failed to list root folders for caller '{}' with pagination: {}",
+                            owner_id, e
+                        ),
+                    )
+                })?;
 
             let total = total_items.unwrap_or(folders.len());
 
@@ -594,6 +612,19 @@ impl FolderUseCase for FolderService {
                         dst_drive_id,
                     },
                 )?;
+                // Destination drive quota: sum the moved subtree's
+                // non-trashed files and refuse if the destination
+                // couldn't hold them. Same 507 shape as the file
+                // path + upload path — DomainError::QuotaExceeded
+                // maps at the AppError boundary.
+                if let Some(storage_usage) = &self.storage_usage {
+                    let subtree_bytes = storage_usage.folder_subtree_bytes(src_folder_uuid).await?;
+                    if let Ok(subtree_u64) = u64::try_from(subtree_bytes) {
+                        storage_usage
+                            .check_drive_quota(dst_drive_id, subtree_u64)
+                            .await?;
+                    }
+                }
                 cross_drive = Some((src_drive_id, dst_drive_id));
             }
         }
@@ -610,6 +641,17 @@ impl FolderUseCase for FolderService {
                 )
             })?;
 
+        // Cross-drive move flushes the authz engine's `owner_cache`
+        // — every descendant's cached `Resource → drive_id` mapping
+        // just got stale via the cascade trigger, and we don't (yet)
+        // walk the subtree to invalidate individually. Small perf
+        // cost (single JOIN per resource touched over the next
+        // minute) versus a stale-authz bug where destination-drive
+        // Owner cascades don't apply to moved content.
+        if cross_drive.is_some() {
+            self.authz.invalidate_owner_cache_all().await;
+        }
+
         // D6 audit: only emit when the move crossed a drive boundary.
         // The cascade trigger has already propagated drive_id to the
         // subtree at this point (see migration
@@ -1082,17 +1124,18 @@ mod cascade_hook_integration_tests {
         let blob_hash = blake3::hash(format!("cascade-{label}-{}", Uuid::new_v4()).as_bytes())
             .to_hex()
             .to_string();
+        // Post-D7: `user_id` omitted — the column is nullable and
+        // provenance flows through `created_by` / `updated_by`.
         sqlx::query_scalar(
             "INSERT INTO storage.files
-                 (name, user_id, drive_id, folder_id, blob_hash, size, created_by, updated_by)
-             VALUES ($1, $2, $3, $4, $5, $6, $7, $7)
+                 (name, drive_id, folder_id, blob_hash, size, created_by, updated_by)
+             VALUES ($1, $2, $3, $4, $5, $6, $6)
              RETURNING id",
         )
         .bind(format!(
             "rust-test-cascade-{label}-{}",
             &Uuid::new_v4().to_string()[..8]
         ))
-        .bind(user_id)
         .bind(drive_id)
         .bind(folder_id)
         .bind(&blob_hash)

src/application/services/idor_protection_test.rs

@@ -1,430 +0,0 @@
-//! Tests for IDOR (Insecure Direct Object Reference) protection.
-//!
-//! Verifies that ownership checks at the repository and service layers
-//! correctly reject access when the caller is not the file owner.
-
-use bytes::Bytes;
-use futures::Stream;
-use std::collections::HashMap;
-use std::path::PathBuf;
-use std::pin::Pin;
-use std::sync::Mutex;
-use uuid::Uuid;
-
-use crate::application::ports::storage_ports::{FileReadPort, FileWritePort};
-use crate::common::errors::DomainError;
-use crate::domain::entities::file::File;
-use crate::domain::services::path_service::StoragePath;
-
-// ═══════════════════════════════════════════════════════════════════════════
-// Mock repositories
-// ═══════════════════════════════════════════════════════════════════════════
-
-/// A simple in-memory mock that maps (file_id → (File, owner_id)).
-struct MockFileReadPort {
-    /// file_id → (File, owner_id)
-    files: Mutex<HashMap<String, (File, Uuid)>>,
-}
-
-impl MockFileReadPort {
-    fn new() -> Self {
-        Self {
-            files: Mutex::new(HashMap::new()),
-        }
-    }
-
-    /// Insert a test file owned by `owner_id`.
-    fn insert(&self, id: &str, name: &str, owner_id: Uuid) {
-        let file = File::new(
-            id.to_string(),
-            name.to_string(),
-            StoragePath::from_string(&format!("/{}", name)),
-            42,
-            "text/plain".to_string(),
-            None,
-        )
-        .unwrap();
-        self.files
-            .lock()
-            .unwrap()
-            .insert(id.to_string(), (file, owner_id));
-    }
-}
-
-impl FileReadPort for MockFileReadPort {
-    async fn get_file(&self, id: &str) -> Result<File, DomainError> {
-        let files = self.files.lock().unwrap();
-        files
-            .get(id)
-            .map(|(f, _)| f.clone())
-            .ok_or_else(|| DomainError::not_found("File", id.to_string()))
-    }
-
-    async fn get_file_or_trashed(&self, id: &str) -> Result<File, DomainError> {
-        let files = self.files.lock().unwrap();
-        files
-            .get(id)
-            .map(|(f, _)| f.clone())
-            .ok_or_else(|| DomainError::not_found("File", id.to_string()))
-    }
-
-    async fn get_file_for_owner(&self, id: &str, owner_id: Uuid) -> Result<File, DomainError> {
-        let files = self.files.lock().unwrap();
-        match files.get(id) {
-            Some((file, actual_owner)) if *actual_owner == owner_id => Ok(file.clone()),
-            // Return NotFound regardless — do not leak existence
-            _ => Err(DomainError::not_found("File", id.to_string())),
-        }
-    }
-
-    async fn list_files(&self, _folder_id: Option<&str>) -> Result<Vec<File>, DomainError> {
-        Ok(Vec::new())
-    }
-
-    async fn get_file_stream(
-        &self,
-        _id: &str,
-    ) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
-        unimplemented!()
-    }
-
-    async fn get_file_range_stream(
-        &self,
-        _id: &str,
-        _start: u64,
-        _end: Option<u64>,
-    ) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
-        unimplemented!()
-    }
-
-    async fn get_file_path(&self, _id: &str) -> Result<StoragePath, DomainError> {
-        unimplemented!()
-    }
-
-    async fn get_parent_folder_id(
-        &self,
-        _path: &str,
-        _drive_id: Uuid,
-    ) -> Result<String, DomainError> {
-        unimplemented!()
-    }
-
-    async fn get_blob_hash(&self, _file_id: &str) -> Result<String, DomainError> {
-        Ok(String::new())
-    }
-
-    async fn search_files_paginated(
-        &self,
-        _folder_id: Option<&str>,
-        _criteria: &crate::application::dtos::search_dto::SearchCriteriaDto,
-        _user_id: Uuid,
-    ) -> Result<(Vec<File>, usize), DomainError> {
-        Ok((Vec::new(), 0))
-    }
-
-    async fn count_files(
-        &self,
-        _folder_id: Option<&str>,
-        _criteria: &crate::application::dtos::search_dto::SearchCriteriaDto,
-        _user_id: Uuid,
-    ) -> Result<usize, DomainError> {
-        Ok(0)
-    }
-
-    async fn get_folder_id_by_path(
-        &self,
-        _folder_path: &str,
-        _drive_id: Uuid,
-    ) -> Result<String, DomainError> {
-        unimplemented!()
-    }
-
-    async fn stream_files_in_subtree(
-        &self,
-        _folder_id: &str,
-    ) -> Result<Pin<Box<dyn Stream<Item = Result<File, DomainError>> + Send>>, DomainError> {
-        Ok(Box::pin(futures::stream::empty()))
-    }
-}
-
-/// Minimal mock write port — only `move_file` and `rename_file` need real logic.
-#[allow(dead_code)]
-struct MockFileWritePort {
-    files: Mutex<HashMap<String, File>>,
-}
-
-impl MockFileWritePort {
-    #[allow(dead_code)]
-    fn new() -> Self {
-        Self {
-            files: Mutex::new(HashMap::new()),
-        }
-    }
-
-    #[allow(dead_code)]
-    fn insert(&self, id: &str, name: &str) {
-        let file = File::new(
-            id.to_string(),
-            name.to_string(),
-            StoragePath::from_string(&format!("/{}", name)),
-            42,
-            "text/plain".to_string(),
-            None,
-        )
-        .unwrap();
-        self.files.lock().unwrap().insert(id.to_string(), file);
-    }
-}
-
-impl FileWritePort for MockFileWritePort {
-    async fn save_file_with_blob(
-        &self,
-        _name: String,
-        _folder_id: Option<String>,
-        _content_type: String,
-        _blob_hash: &str,
-        _size: u64,
-        _caller_id: Uuid,
-    ) -> Result<File, DomainError> {
-        unimplemented!()
-    }
-
-    async fn move_file(
-        &self,
-        file_id: &str,
-        _target_folder_id: Option<String>,
-        _caller_id: Uuid,
-    ) -> Result<File, DomainError> {
-        let files = self.files.lock().unwrap();
-        files
-            .get(file_id)
-            .cloned()
-            .ok_or_else(|| DomainError::not_found("File", file_id.to_string()))
-    }
-
-    async fn rename_file(
-        &self,
-        file_id: &str,
-        _new_name: &str,
-        _caller_id: Uuid,
-    ) -> Result<File, DomainError> {
-        let files = self.files.lock().unwrap();
-        files
-            .get(file_id)
-            .cloned()
-            .ok_or_else(|| DomainError::not_found("File", file_id.to_string()))
-    }
-
-    async fn delete_file(&self, _id: &str) -> Result<(), DomainError> {
-        Ok(())
-    }
-
-    async fn update_file_content_with_blob(
-        &self,
-        _file_id: &str,
-        _blob_hash: &str,
-        _size: u64,
-        _modified_at: Option<i64>,
-        _caller_id: Uuid,
-    ) -> Result<(String, i64), DomainError> {
-        Ok((String::new(), 0))
-    }
-
-    async fn register_file_deferred(
-        &self,
-        _name: String,
-        _folder_id: Option<String>,
-        _content_type: String,
-        _size: u64,
-        _caller_id: Uuid,
-    ) -> Result<(File, PathBuf), DomainError> {
-        unimplemented!()
-    }
-
-    async fn copy_file(
-        &self,
-        _file_id: &str,
-        _target_folder_id: Option<String>,
-        _new_name: Option<&str>,
-        _caller_id: Uuid,
-    ) -> Result<File, DomainError> {
-        unimplemented!()
-    }
-
-    async fn move_to_trash(&self, _file_id: &str, _caller_id: Uuid) -> Result<(), DomainError> {
-        Ok(())
-    }
-
-    async fn restore_from_trash(
-        &self,
-        _file_id: &str,
-        _original_path: &str,
-        _caller_id: Uuid,
-    ) -> Result<(), DomainError> {
-        Ok(())
-    }
-
-    async fn delete_file_permanently(&self, _file_id: &str) -> Result<(), DomainError> {
-        Ok(())
-    }
-}
-
-// ═══════════════════════════════════════════════════════════════════════════
-// Tests — FileReadPort::get_file_for_owner (Repository layer, Solution C)
-// ═══════════════════════════════════════════════════════════════════════════
-
-#[tokio::test]
-async fn get_file_for_owner_returns_file_for_correct_owner() {
-    let alice_id = Uuid::new_v4();
-    let repo = MockFileReadPort::new();
-    repo.insert("file-1", "secret.txt", alice_id);
-
-    let result = repo.get_file_for_owner("file-1", alice_id).await;
-    assert!(result.is_ok(), "owner should be able to read own file");
-    assert_eq!(result.unwrap().id(), "file-1");
-}
-
-#[tokio::test]
-async fn get_file_for_owner_rejects_wrong_owner() {
-    let alice_id = Uuid::new_v4();
-    let bob_id = Uuid::new_v4();
-    let repo = MockFileReadPort::new();
-    repo.insert("file-1", "secret.txt", alice_id);
-
-    let result = repo.get_file_for_owner("file-1", bob_id).await;
-    assert!(result.is_err(), "non-owner should be rejected");
-
-    // Must be NotFound, NOT Forbidden — avoids leaking existence
-    let err = result.unwrap_err();
-    let msg = format!("{}", err);
-    assert!(
-        msg.contains("not found") || msg.contains("NotFound"),
-        "error must be NotFound, got: {}",
-        msg
-    );
-}
-
-#[tokio::test]
-async fn get_file_for_owner_returns_not_found_for_missing_file() {
-    let alice_id = Uuid::new_v4();
-    let repo = MockFileReadPort::new();
-
-    let result = repo.get_file_for_owner("nonexistent", alice_id).await;
-    assert!(result.is_err());
-}
-
-#[tokio::test]
-async fn verify_file_owner_uses_default_impl() {
-    let alice_id = Uuid::new_v4();
-    let bob_id = Uuid::new_v4();
-    let repo = MockFileReadPort::new();
-    repo.insert("file-1", "secret.txt", alice_id);
-
-    // Default impl delegates to get_file_for_owner and maps to ()
-    assert!(repo.verify_file_owner("file-1", alice_id).await.is_ok());
-    assert!(repo.verify_file_owner("file-1", bob_id).await.is_err());
-}
-
-// ═══════════════════════════════════════════════════════════════════════════
-// Tests — FileManagementService _owned methods (Service layer, Solution B)
-// ═══════════════════════════════════════════════════════════════════════════
-//
-// Note: FileManagementService::with_trash takes concrete types for the write
-// repository (Arc<FileBlobWriteRepository>). We cannot construct real PG repos
-// without a database. Instead, we test the verify_owner logic indirectly by
-// testing the mock-based trait interactions at the port level, and document
-// that integration tests hitting the real DB are the ultimate verification.
-//
-// The tests below verify the *contract*: _owned methods must call
-// verify_owner before delegating, and verify_owner must fail-closed when
-// no read repo is available.
-
-#[tokio::test]
-async fn verify_file_owner_delegates_to_read_port() {
-    // This test verifies the FileReadPort contract that verify_file_owner
-    // returns Ok for the correct owner and Err for others.
-    let user_id = Uuid::new_v4();
-    let attacker_id = Uuid::new_v4();
-    let read = MockFileReadPort::new();
-    read.insert("abc-123", "report.pdf", user_id);
-
-    // Same user → Ok
-    let ok = read.verify_file_owner("abc-123", user_id).await;
-    assert!(ok.is_ok(), "correct owner should pass verify_file_owner");
-
-    // Different user → Err
-    let err = read.verify_file_owner("abc-123", attacker_id).await;
-    assert!(err.is_err(), "wrong owner should fail verify_file_owner");
-}
-
-#[tokio::test]
-async fn owned_methods_require_ownership_check_first() {
-    // Simulate what the _owned methods do: verify_owner then delegate.
-    // We test with the mock read port to prove the sequence.
-    let owner_id = Uuid::new_v4();
-    let attacker_id = Uuid::new_v4();
-    let read = MockFileReadPort::new();
-    read.insert("file-1", "data.csv", owner_id);
-
-    // Step 1: verify_owner for correct owner → Ok
-    let step1 = read.verify_file_owner("file-1", owner_id).await;
-    assert!(step1.is_ok());
-
-    // Step 2: verify_owner for attacker → Err, so the move/rename never executes
-    let step2 = read.verify_file_owner("file-1", attacker_id).await;
-    assert!(step2.is_err());
-}
-
-// ═══════════════════════════════════════════════════════════════════════════
-// Tests — Trait-level _owned method stubs (StubFileManagementUseCase)
-// ═══════════════════════════════════════════════════════════════════════════
-
-use crate::application::ports::file_ports::FileManagementUseCase;
-use crate::common::stubs::StubFileManagementUseCase;
-
-#[tokio::test]
-async fn stub_move_file_owned_returns_ok() {
-    let user_id = Uuid::new_v4();
-    let stub = StubFileManagementUseCase;
-    let result = stub
-        .move_file_with_perms("file-1", user_id, Some("folder-2".to_string()))
-        .await;
-    assert!(result.is_ok(), "stub should return Ok for move_file_owned");
-}
-
-#[tokio::test]
-async fn stub_rename_file_owned_returns_ok() {
-    let user_id = Uuid::new_v4();
-    let stub = StubFileManagementUseCase;
-    let result = stub
-        .rename_file_with_perms("file-1", user_id, "new-name.txt")
-        .await;
-    assert!(
-        result.is_ok(),
-        "stub should return Ok for rename_file_owned"
-    );
-}
-
-use crate::application::ports::file_ports::FileRetrievalUseCase;
-use crate::common::stubs::StubFileRetrievalUseCase;
-
-#[tokio::test]
-async fn stub_get_file_owned_returns_ok() {
-    let user_id = Uuid::new_v4();
-    let stub = StubFileRetrievalUseCase;
-    let result = stub.get_file_with_perms("file-1", user_id).await;
-    assert!(result.is_ok(), "stub should return Ok for get_file_owned");
-}
-
-#[tokio::test]
-async fn stub_get_file_optimized_owned_returns_ok() {
-    let user_id = Uuid::new_v4();
-    let stub = StubFileRetrievalUseCase;
-    let result = stub
-        .get_file_optimized_with_perms("file-1", user_id, true, false)
-        .await;
-    assert!(
-        result.is_ok(),
-        "stub should return Ok for get_file_optimized_owned"
-    );
-}

src/application/services/magic_link_invite_service.rs

@@ -307,20 +307,30 @@ impl MagicLinkInviteService {
         let (kind, resource_id) = match resource {
             Resource::Folder(id) => (MagicLinkResourceKind::Folder, id),
             Resource::File(id) => (MagicLinkResourceKind::File, id),
-            // Drive sharing — and therefore drive magic-link invitations —
-            // land in D2. The grant DTOs accept `Resource::Drive` from the
-            // wire today (see ResourceTypeDto) but no public API path
-            // actually grants on a drive in D0, so this arm is
-            // defensively unreachable. Treating it as an audit-logged
-            // no-op (grant is in place, mail suppressed) matches the
-            // ineligible-recipient branch above.
-            Resource::Drive(_) => {
+            // Drive / Calendar / AddressBook / Playlist sharing is
+            // out-of-band for the magic-link flow. Drive shares land
+            // through `/api/drives/{id}/members`; Calendar /
+            // AddressBook shares through the Round-3
+            // `/api/(calendars|address-books)/{id}/shares` endpoints;
+            // Playlist shares through `/api/playlists/{id}/share`.
+            // The DTOs accept every `Resource` variant on the wire
+            // (see `ResourceTypeDto`) but only file/folder grants
+            // trigger an invitation email. Treating the other arms
+            // as audit-logged suppressed no-ops keeps the grant in
+            // place while matching the ineligible-recipient branch
+            // above.
+            Resource::Drive(_)
+            | Resource::Calendar(_)
+            | Resource::AddressBook(_)
+            | Resource::Playlist(_) => {
                 tracing::info!(
                     target: "audit",
                     event = "magic_link.invitation_suppressed",
-                    reason = "drive_resource_unsupported",
+                    reason = "resource_kind_unsupported",
                     user_id = %recipient.id(),
-                    "📭 magic-link invitation suppressed: drive resources aren't invitable until D2",
+                    resource_kind = %resource.type_str(),
+                    "📭 magic-link invitation suppressed: {} resources aren't invitable via email",
+                    resource.type_str(),
                 );
                 return Ok(());
             }
@@ -347,10 +357,13 @@ impl MagicLinkInviteService {
             Resource::Folder(_) => "server.magic_link.email.kind_folder",
             Resource::File(_) => "server.magic_link.email.kind_file",
             // Unreachable — the early-return above exits before we get
-            // here for a Drive resource. The arm exists only to satisfy
-            // exhaustiveness; if you find this firing, the early-return
-            // was bypassed.
-            Resource::Drive(_) => "server.magic_link.email.kind_folder",
+            // here for Drive / Calendar / AddressBook / Playlist
+            // resources. The arms exist only to satisfy exhaustiveness;
+            // if you find any firing, the early-return was bypassed.
+            Resource::Drive(_)
+            | Resource::Calendar(_)
+            | Resource::AddressBook(_)
+            | Resource::Playlist(_) => "server.magic_link.email.kind_folder",
         };
         // PR C: render in the recipient's preferred locale (set by UI
         // switcher, OIDC JIT claim, or inviter inheritance at row

src/application/services/mod.rs

@@ -39,8 +39,6 @@ pub mod wopi_token_service;
 #[cfg(test)]
 mod batch_operations_test;
 #[cfg(test)]
-mod idor_protection_test;
-#[cfg(test)]
 mod trash_service_test;
 
 // Re-exportar para facilitar acceso

src/application/services/music_service.rs

@@ -1,3 +1,4 @@
+use std::collections::HashSet;
 use std::sync::Arc;
 use uuid::Uuid;
 
@@ -5,17 +6,80 @@ use crate::application::dtos::playlist_dto::{
     AddTracksDto, AudioMetadataDto, CreatePlaylistDto, PlaylistDto, PlaylistItemDto,
     PlaylistQueryDto, PlaylistShareInfoDto, ReorderTracksDto, SharePlaylistDto, UpdatePlaylistDto,
 };
+use crate::application::ports::authorization_ports::AuthorizationEngine;
 use crate::application::ports::music_ports::{MusicStoragePort, MusicUseCase};
 use crate::common::errors::{DomainError, ErrorKind};
+use crate::domain::services::authorization::{Permission, Resource, Role, Subject};
 use crate::infrastructure::adapters::music_storage_adapter::MusicStorageAdapter;
+use crate::infrastructure::services::pg_acl_engine::PgAclEngine;
 
+/// Music service — the REST entry point for every playlist or audio
+/// metadata operation. Every method routes through
+/// `AuthorizationEngine`; the pre-Round-3 `user_has_access` /
+/// `user_can_write` bespoke helpers on `MusicStorageAdapter` are no
+/// longer consulted for access decisions.
+///
+/// Ownership + sharing live entirely in `storage.role_grants`
+/// (`resource_type='playlist'`). `audio.playlists.owner_id` stays for
+/// provenance and legacy queries; `audio.playlist_shares` is
+/// backfilled and slated for removal in a follow-up migration.
 pub struct MusicService {
     storage: Arc<MusicStorageAdapter>,
+    /// ReBAC engine — every user-facing method calls `authz.require`
+    /// with the appropriate `Permission`. `create_playlist` also uses
+    /// it to seed an Owner grant for the caller, so the common
+    /// "owning my own playlist" case takes a single indexed
+    /// role_grants lookup on subsequent reads.
+    authz: Arc<PgAclEngine>,
 }
 
 impl MusicService {
-    pub fn new(storage: Arc<MusicStorageAdapter>) -> Self {
-        Self { storage }
+    pub fn new(storage: Arc<MusicStorageAdapter>, authz: Arc<PgAclEngine>) -> Self {
+        Self { storage, authz }
+    }
+
+    /// Parse `playlist_id` and enforce `permission` on
+    /// `Resource::Playlist(uuid)`. On denial `authz.require` returns
+    /// `NotFound` (anti-enum — same shape as "no such playlist") and
+    /// emits the `authz.denied` audit line. Returns the parsed UUID
+    /// on success so the caller doesn't have to parse it a second
+    /// time.
+    async fn require_playlist_perm(
+        &self,
+        playlist_id: &str,
+        caller_id: Uuid,
+        permission: Permission,
+    ) -> Result<Uuid, DomainError> {
+        let uuid = Uuid::parse_str(playlist_id)
+            .map_err(|_| DomainError::new(ErrorKind::InvalidInput, "Playlist", "Invalid ID"))?;
+        self.authz
+            .require(
+                Subject::User(caller_id),
+                permission,
+                Resource::Playlist(uuid),
+            )
+            .await?;
+        Ok(uuid)
+    }
+
+    /// Check `permission` on a playlist without throwing. Used by the
+    /// read paths that also allow a public-playlist bypass — they
+    /// need a bool, not a `Result<(), NotFound>`.
+    async fn has_playlist_perm(
+        &self,
+        playlist_id: &str,
+        caller_id: Uuid,
+        permission: Permission,
+    ) -> Result<bool, DomainError> {
+        let uuid = Uuid::parse_str(playlist_id)
+            .map_err(|_| DomainError::new(ErrorKind::InvalidInput, "Playlist", "Invalid ID"))?;
+        self.authz
+            .check(
+                Subject::User(caller_id),
+                permission,
+                Resource::Playlist(uuid),
+            )
+            .await
     }
 }
 
@@ -25,7 +89,26 @@ impl MusicUseCase for MusicService {
         dto: CreatePlaylistDto,
         user_id: Uuid,
     ) -> Result<PlaylistDto, DomainError> {
-        self.storage.create_playlist(dto, user_id).await
+        // No pre-write gate: creating a playlist is a personal act.
+        // Storage stamps `owner_id = user_id`; we then seed an Owner
+        // role_grant so subsequent reads hit the same
+        // `storage.role_grants` fast path used everywhere else.
+        let created = self.storage.create_playlist(dto, user_id).await?;
+        let playlist_uuid = Uuid::parse_str(&created.id).map_err(|_| {
+            DomainError::internal_error("Playlist", "storage returned invalid playlist id")
+        })?;
+        // `set_role` is idempotent on the `(subject, resource)` unique
+        // key. `granted_by = user_id` is the self-seeded creation event.
+        self.authz
+            .set_role(
+                user_id,
+                Subject::User(user_id),
+                Role::Owner,
+                Resource::Playlist(playlist_uuid),
+                None,
+            )
+            .await?;
+        Ok(created)
     }
 
     async fn update_playlist(
@@ -34,45 +117,25 @@ impl MusicUseCase for MusicService {
         dto: UpdatePlaylistDto,
         user_id: Uuid,
     ) -> Result<PlaylistDto, DomainError> {
-        let has_access = self.storage.user_has_access(playlist_id, user_id).await?;
-        if !has_access {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "You don't have permission to update this playlist",
-            ));
-        }
-        let can_write = self.storage.user_can_write(playlist_id, user_id).await?;
-        if !can_write {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "You need write access to update this playlist",
-            ));
-        }
+        self.require_playlist_perm(playlist_id, user_id, Permission::Update)
+            .await?;
         self.storage.update_playlist(playlist_id, dto).await
     }
 
     async fn delete_playlist(&self, playlist_id: &str, user_id: Uuid) -> Result<(), DomainError> {
-        let playlist = self.storage.get_playlist(playlist_id).await?;
-        let playlist = match playlist {
-            Some(p) => p,
-            None => {
-                return Err(DomainError::new(
-                    ErrorKind::NotFound,
-                    "Playlist",
-                    "Playlist not found",
-                ));
-            }
-        };
-        if playlist.owner_id != user_id.to_string() {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "Only the owner can delete this playlist",
-            ));
-        }
-        self.storage.delete_playlist(playlist_id).await
+        let uuid = self
+            .require_playlist_perm(playlist_id, user_id, Permission::Delete)
+            .await?;
+        self.storage.delete_playlist(playlist_id).await?;
+        // Wipe every grant on this playlist so a re-used UUID
+        // (impossible today but cheap to defend against) doesn't
+        // inherit stale ACLs. The storage DELETE won't cascade to
+        // `storage.role_grants` — it's cross-schema.
+        let _ = self
+            .authz
+            .revoke_all_for_resource(Resource::Playlist(uuid))
+            .await;
+        Ok(())
     }
 
     async fn get_playlist(
@@ -80,23 +143,22 @@ impl MusicUseCase for MusicService {
         playlist_id: &str,
         user_id: Uuid,
     ) -> Result<PlaylistDto, DomainError> {
-        let has_access = self.storage.user_has_access(playlist_id, user_id).await?;
-        if !has_access {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "You don't have permission to view this playlist",
-            ));
-        }
         let playlist = self.storage.get_playlist(playlist_id).await?;
-        match playlist {
-            Some(p) => Ok(p),
-            None => Err(DomainError::new(
-                ErrorKind::NotFound,
-                "Playlist",
-                "Playlist not found",
-            )),
+        let playlist = match playlist {
+            Some(p) => p,
+            None => return Err(DomainError::not_found("Playlist", playlist_id)),
+        };
+        // Public-playlist bypass: anonymous-ish read. `check` returns
+        // bool (no throw); combine with the public flag before
+        // deciding.
+        let allowed = playlist.is_public
+            || self
+                .has_playlist_perm(playlist_id, user_id, Permission::Read)
+                .await?;
+        if !allowed {
+            return Err(DomainError::not_found("Playlist", playlist_id));
         }
+        Ok(playlist)
     }
 
     async fn list_playlists(
@@ -109,17 +171,38 @@ impl MusicUseCase for MusicService {
         let limit = query.limit.unwrap_or(100);
         let offset = query.offset.unwrap_or(0);
 
-        let mut playlists = Vec::new();
+        // Post-Round-3 semantics: playlists the caller has any grant
+        // on come from `list_incoming_grants` — one union of owned +
+        // shared. The pre-Round-3 code fetched them via two separate
+        // queries (`list_playlists_by_owner` + `list_shared_with_user`)
+        // that each read a different table.
+        let grants = self
+            .authz
+            .list_incoming_grants(Subject::User(user_id))
+            .await?;
 
-        let owned = self.storage.list_playlists_by_owner(user_id).await?;
-        playlists.extend(owned);
+        // Deduplicate — a user can hold multiple grants on the same
+        // playlist (direct + group-inherited). We only need one DTO
+        // per resource.
+        let mut playlist_ids: HashSet<Uuid> = grants
+            .into_iter()
+            .filter_map(|g| match g.resource {
+                Resource::Playlist(id) => Some(id),
+                _ => None,
+            })
+            .collect();
 
-        if include_shared {
-            let shared = self.storage.list_shared_with_user(user_id).await?;
-            for s in shared {
-                if !playlists.iter().any(|p: &PlaylistDto| p.id == s.id) {
-                    playlists.push(s);
-                }
+        // `include_shared=false` narrows the listing to owned playlists
+        // only. Owner is a grant like any other in `role_grants`, so we
+        // filter the aggregated set against the owner_id stamped on
+        // each row after hydration — cheaper than a second SQL round-trip.
+        let mut playlists: Vec<PlaylistDto> = Vec::with_capacity(playlist_ids.len());
+        let user_str = user_id.to_string();
+        for id in playlist_ids.drain() {
+            if let Ok(Some(p)) = self.storage.get_playlist(&id.to_string()).await
+                && (include_shared || p.owner_id == user_str)
+            {
+                playlists.push(p);
             }
         }
 
@@ -141,26 +224,9 @@ impl MusicUseCase for MusicService {
         dto: AddTracksDto,
         user_id: Uuid,
     ) -> Result<Vec<PlaylistItemDto>, DomainError> {
-        let playlist_uuid = Uuid::parse_str(playlist_id).map_err(|_| {
-            DomainError::new(ErrorKind::InvalidInput, "Playlist", "Invalid playlist ID")
-        })?;
-
-        let has_access = self.storage.user_has_access(playlist_id, user_id).await?;
-        if !has_access {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "You don't have permission to modify this playlist",
-            ));
-        }
-        let can_write = self.storage.user_can_write(playlist_id, user_id).await?;
-        if !can_write {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "You need write access to add tracks",
-            ));
-        }
+        let playlist_uuid = self
+            .require_playlist_perm(playlist_id, user_id, Permission::Update)
+            .await?;
 
         let file_ids: Result<Vec<Uuid>, _> =
             dto.file_ids.iter().map(|id| Uuid::parse_str(id)).collect();
@@ -177,30 +243,12 @@ impl MusicUseCase for MusicService {
         file_id: &str,
         user_id: Uuid,
     ) -> Result<(), DomainError> {
-        let playlist_uuid = Uuid::parse_str(playlist_id).map_err(|_| {
-            DomainError::new(ErrorKind::InvalidInput, "Playlist", "Invalid playlist ID")
-        })?;
+        let playlist_uuid = self
+            .require_playlist_perm(playlist_id, user_id, Permission::Update)
+            .await?;
         let file_uuid = Uuid::parse_str(file_id).map_err(|_| {
             DomainError::new(ErrorKind::InvalidInput, "Playlist", "Invalid file ID")
         })?;
-
-        let has_access = self.storage.user_has_access(playlist_id, user_id).await?;
-        if !has_access {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "You don't have permission to modify this playlist",
-            ));
-        }
-        let can_write = self.storage.user_can_write(playlist_id, user_id).await?;
-        if !can_write {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "You need write access to remove tracks",
-            ));
-        }
-
         self.storage.remove_track(&playlist_uuid, &file_uuid).await
     }
 
@@ -210,26 +258,9 @@ impl MusicUseCase for MusicService {
         dto: ReorderTracksDto,
         user_id: Uuid,
     ) -> Result<(), DomainError> {
-        let playlist_uuid = Uuid::parse_str(playlist_id).map_err(|_| {
-            DomainError::new(ErrorKind::InvalidInput, "Playlist", "Invalid playlist ID")
-        })?;
-
-        let has_access = self.storage.user_has_access(playlist_id, user_id).await?;
-        if !has_access {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "You don't have permission to modify this playlist",
-            ));
-        }
-        let can_write = self.storage.user_can_write(playlist_id, user_id).await?;
-        if !can_write {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "You need write access to reorder tracks",
-            ));
-        }
+        let playlist_uuid = self
+            .require_playlist_perm(playlist_id, user_id, Permission::Update)
+            .await?;
 
         let item_ids: Result<Vec<Uuid>, _> =
             dto.item_ids.iter().map(|id| Uuid::parse_str(id)).collect();
@@ -248,16 +279,21 @@ impl MusicUseCase for MusicService {
         let playlist_uuid = Uuid::parse_str(playlist_id).map_err(|_| {
             DomainError::new(ErrorKind::InvalidInput, "Playlist", "Invalid playlist ID")
         })?;
-
-        let has_access = self.storage.user_has_access(playlist_id, user_id).await?;
-        if !has_access {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "You don't have permission to view this playlist",
-            ));
+        // Public-playlist bypass mirrors `get_playlist`: readers of a
+        // public playlist can see its tracks. Fetch the playlist row
+        // to inspect `is_public` before deciding.
+        let playlist = self
+            .storage
+            .get_playlist(playlist_id)
+            .await?
+            .ok_or_else(|| DomainError::not_found("Playlist", playlist_id))?;
+        let allowed = playlist.is_public
+            || self
+                .has_playlist_perm(playlist_id, user_id, Permission::Read)
+                .await?;
+        if !allowed {
+            return Err(DomainError::not_found("Playlist", playlist_id));
         }
-
         self.storage.list_playlist_tracks(&playlist_uuid).await
     }
 
@@ -267,36 +303,33 @@ impl MusicUseCase for MusicService {
         dto: SharePlaylistDto,
         caller_id: Uuid,
     ) -> Result<(), DomainError> {
-        let playlist = self.storage.get_playlist(playlist_id).await?;
-        let playlist = match playlist {
-            Some(p) => p,
-            None => {
-                return Err(DomainError::new(
-                    ErrorKind::NotFound,
-                    "Playlist",
-                    "Playlist not found",
-                ));
-            }
-        };
-        if playlist.owner_id != caller_id.to_string() {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "Only the owner can share this playlist",
-            ));
-        }
-
-        let playlist_uuid = Uuid::parse_str(playlist_id).map_err(|_| {
-            DomainError::new(ErrorKind::InvalidInput, "Playlist", "Invalid playlist ID")
-        })?;
+        let playlist_uuid = self
+            .require_playlist_perm(playlist_id, caller_id, Permission::Share)
+            .await?;
         let target_user_id = Uuid::parse_str(&dto.user_id).map_err(|_| {
             DomainError::new(ErrorKind::InvalidInput, "Playlist", "Invalid user ID")
         })?;
-        let can_write = dto.can_write.unwrap_or(false);
-
-        self.storage
-            .share_playlist(&playlist_uuid, target_user_id, can_write)
-            .await
+        // Legacy `can_write` boolean maps into the role bundle system:
+        //   - false → Viewer (Read only)
+        //   - true  → Editor (Read + Update)
+        // The endpoint stays boolean-shaped for API back-compat; new
+        // integrations should switch to the unified `/api/grants` API
+        // which exposes the full role set.
+        let role = if dto.can_write.unwrap_or(false) {
+            Role::Editor
+        } else {
+            Role::Viewer
+        };
+        self.authz
+            .set_role(
+                caller_id,
+                Subject::User(target_user_id),
+                role,
+                Resource::Playlist(playlist_uuid),
+                None,
+            )
+            .await?;
+        Ok(())
     }
 
     async fn remove_share(
@@ -305,33 +338,18 @@ impl MusicUseCase for MusicService {
         target_user_id: &str,
         caller_id: Uuid,
     ) -> Result<(), DomainError> {
-        let playlist = self.storage.get_playlist(playlist_id).await?;
-        let playlist = match playlist {
-            Some(p) => p,
-            None => {
-                return Err(DomainError::new(
-                    ErrorKind::NotFound,
-                    "Playlist",
-                    "Playlist not found",
-                ));
-            }
-        };
-        if playlist.owner_id != caller_id.to_string() {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "Only the owner can manage sharing",
-            ));
-        }
-
-        let playlist_uuid = Uuid::parse_str(playlist_id).map_err(|_| {
-            DomainError::new(ErrorKind::InvalidInput, "Playlist", "Invalid playlist ID")
-        })?;
+        let playlist_uuid = self
+            .require_playlist_perm(playlist_id, caller_id, Permission::Share)
+            .await?;
         let target_uuid = Uuid::parse_str(target_user_id).map_err(|_| {
             DomainError::new(ErrorKind::InvalidInput, "Playlist", "Invalid user ID")
         })?;
-
-        self.storage.remove_share(&playlist_uuid, target_uuid).await
+        self.authz
+            .clear_role(
+                Subject::User(target_uuid),
+                Resource::Playlist(playlist_uuid),
+            )
+            .await
     }
 
     async fn get_playlist_shares(
@@ -339,35 +357,26 @@ impl MusicUseCase for MusicService {
         playlist_id: &str,
         user_id: Uuid,
     ) -> Result<Vec<PlaylistShareInfoDto>, DomainError> {
-        let playlist = self.storage.get_playlist(playlist_id).await?;
-        let playlist = match playlist {
-            Some(p) => p,
-            None => {
-                return Err(DomainError::new(
-                    ErrorKind::NotFound,
-                    "Playlist",
-                    "Playlist not found",
-                ));
-            }
-        };
-        if playlist.owner_id != user_id.to_string() {
-            return Err(DomainError::new(
-                ErrorKind::AccessDenied,
-                "Playlist",
-                "Only the owner can view sharing info",
-            ));
-        }
-
-        let playlist_uuid = Uuid::parse_str(playlist_id).map_err(|_| {
-            DomainError::new(ErrorKind::InvalidInput, "Playlist", "Invalid playlist ID")
-        })?;
-
-        let shares = self.storage.get_shares(&playlist_uuid).await?;
-        Ok(shares
+        let playlist_uuid = self
+            .require_playlist_perm(playlist_id, user_id, Permission::Share)
+            .await?;
+        // `list_grants_on_resource` returns every role_grant row for
+        // the playlist. Drop the Owner self-grant seeded at creation
+        // (the caller already knows they own it) and collapse the
+        // role bundle back to a boolean `can_write` for the legacy
+        // DTO shape.
+        let grants = self
+            .authz
+            .list_grants_on_resource(Resource::Playlist(playlist_uuid))
+            .await?;
+        Ok(grants
             .into_iter()
-            .map(|(uid, can_write)| PlaylistShareInfoDto {
-                user_id: uid.to_string(),
-                can_write,
+            .filter_map(|g| match g.subject {
+                Subject::User(uid) if g.role != Role::Owner => Some(PlaylistShareInfoDto {
+                    user_id: uid.to_string(),
+                    can_write: g.role.expand().contains(&Permission::Update),
+                }),
+                _ => None,
             })
             .collect())
     }
@@ -375,10 +384,23 @@ impl MusicUseCase for MusicService {
     async fn get_audio_metadata(
         &self,
         file_id: &str,
-        _user_id: Uuid,
+        caller_id: Uuid,
     ) -> Result<Option<AudioMetadataDto>, DomainError> {
         let file_uuid = Uuid::parse_str(file_id)
             .map_err(|_| DomainError::new(ErrorKind::InvalidInput, "Music", "Invalid file ID"))?;
+        // AuthZ pre-read: caller must have `Read` on the underlying
+        // audio file. Before this check the endpoint returned
+        // metadata for any known file id (cross-tenant IDOR — the
+        // `_user_id` parameter was deliberately unused). `require`
+        // returns 404 on denial to match the anti-enum shape used
+        // everywhere else.
+        self.authz
+            .require(
+                Subject::User(caller_id),
+                Permission::Read,
+                Resource::File(file_uuid),
+            )
+            .await?;
         self.storage.get_audio_metadata(&file_uuid).await
     }
 }

src/application/services/places_service.rs

@@ -9,10 +9,12 @@ use crate::infrastructure::repositories::pg::FileBlobReadRepository;
 /// "Places" use case: the caller's geotagged photos aggregated into map
 /// clusters.
 ///
-/// Strictly user-scoped — the repository filters `WHERE fi.user_id = $1`, so,
-/// like [`RecentService`](super::recent_service::RecentService) and the photos
-/// timeline, it needs no `AuthorizationEngine` check: the `caller_id`
-/// parameter *is* the access scope.
+/// Post-§15 the surface follows the Photos scope: drives where the
+/// caller has Read AND `policies.include_in_photo_index = true`
+/// (default personal drives materialise the flag at creation).
+/// Group-membership expansion is handled inline by
+/// `storage.caller_group_ids(caller)` inside the repo's SQL, so this
+/// service is a thin coordinate-math wrapper — no engine dependency.
 pub struct PlacesService {
     file_read: Arc<FileBlobReadRepository>,
 }
@@ -30,7 +32,8 @@ impl PlacesService {
         360.0 / (2_f64.powi(z) * 4.0)
     }
 
-    /// Clustered geotagged photos for `caller_id` within `bounds`.
+    /// Clustered geotagged photos in the caller's Photos-scope drive set,
+    /// within `bounds`.
     pub async fn clusters(
         &self,
         caller_id: Uuid,

src/application/services/recent_service.rs

@@ -1,10 +1,12 @@
 use crate::application::dtos::cursor::PageCursor;
 use crate::application::dtos::recent_dto::{RecentCursor, RecentItemDto, RecentResourceRow};
+use crate::application::ports::authorization_ports::AuthorizationEngine;
 use crate::application::ports::recent_ports::{RecentItemsRepositoryPort, RecentItemsUseCase};
 use crate::application::ports::resource_access_hook::ResourceAccessHook;
-use crate::common::errors::{DomainError, ErrorKind, Result};
-use crate::domain::services::authorization::ResourceKind;
+use crate::common::errors::{DomainError, Result};
+use crate::domain::services::authorization::{Permission, Resource, ResourceKind, Subject};
 use crate::infrastructure::repositories::pg::RecentItemsPgRepository;
+use crate::infrastructure::services::pg_acl_engine::PgAclEngine;
 use std::sync::{Arc, OnceLock};
 use tracing::info;
 use uuid::Uuid;
@@ -16,6 +18,13 @@ use uuid::Uuid;
 pub struct RecentService {
     repo: Arc<RecentItemsPgRepository>,
     max_recent_items: i32,
+    /// ReBAC engine — enforces `Permission::Read` on the referenced
+    /// file/folder before enrolling it into a user's Recent list.
+    /// The listing side JOINs back to `storage.files/folders` and
+    /// returns name/mime/size/drive_id for any enrolled UUID, so
+    /// the write path is an information oracle without this gate.
+    /// See `docs/plan/authz_audit/rest_storage.md`.
+    authorization: Arc<PgAclEngine>,
     /// Set after construction via [`Self::set_resource_access_hook`].
     /// The hook is built FROM this service (it wraps an `Arc<Self>`), so
     /// we can't take it as a constructor arg without circular ownership;
@@ -28,10 +37,15 @@ pub struct RecentService {
 
 impl RecentService {
     /// Create a new recent items service
-    pub fn new(repo: Arc<RecentItemsPgRepository>, max_recent_items: i32) -> Self {
+    pub fn new(
+        repo: Arc<RecentItemsPgRepository>,
+        authorization: Arc<PgAclEngine>,
+        max_recent_items: i32,
+    ) -> Self {
         Self {
             repo,
             max_recent_items: max_recent_items.clamp(1, 100),
+            authorization,
             resource_access_hook: OnceLock::new(),
         }
     }
@@ -53,6 +67,41 @@ impl RecentService {
             hook.on_recents_cleared(user_id);
         }
     }
+
+    /// Record access to an item WITHOUT the pre-write `authz.require`
+    /// gate. Callers must have gated the caller's Read upstream — this
+    /// method exists for the `RecentRecordingHook` fast path: writes
+    /// that reach the hook have already passed a `_with_perms` service
+    /// method (uploads, streams, GETs, etc.), so re-checking here
+    /// would be pure duplicate work AND widen the race window between
+    /// the POST response and the `tokio::spawn`ed upsert (
+    /// `tests/api/recent.hurl` step 7 hits this — the extra SQL
+    /// round-trip pushes the upsert past the client's immediate
+    /// `GET /api/recent/resources`).
+    ///
+    /// **Do NOT call this from an externally-reachable handler.** The
+    /// REST endpoint goes through the trait method `record_item_access`
+    /// below, which enforces the Read gate per AGENTS.md convention.
+    pub async fn record_item_access_internal(
+        &self,
+        user_id: Uuid,
+        item_id: &str,
+        item_type: &str,
+    ) -> Result<()> {
+        // Type validation only — no authz, no resource parse for the
+        // engine (the hook path is already resource-typed by construction).
+        if item_type != "file" && item_type != "folder" {
+            return Err(DomainError::new(
+                crate::common::errors::ErrorKind::InvalidInput,
+                "RecentItems",
+                "Item type must be 'file' or 'folder'",
+            ));
+        }
+
+        self.repo.upsert_access(user_id, item_id, item_type).await?;
+        self.repo.prune(user_id, self.max_recent_items).await?;
+        Ok(())
+    }
 }
 
 impl RecentItemsUseCase for RecentService {
@@ -87,16 +136,24 @@ impl RecentItemsUseCase for RecentService {
             item_type, item_id, user_id
         );
 
-        if item_type != "file" && item_type != "folder" {
-            return Err(DomainError::new(
-                ErrorKind::InvalidInput,
-                "RecentItems",
-                "Item type must be 'file' or 'folder'",
-            ));
-        }
+        // AuthZ pre-write: caller must have Read on the referenced
+        // resource. Denial routes through `require` → NotFound
+        // (anti-enum) + `authz.denied` audit line. Without this
+        // gate the write path was an information oracle over the
+        // whole tenant via the listing endpoint's JOIN back to
+        // storage.files/folders.
+        //
+        // Internal hook callers (RecentRecordingHook) bypass the
+        // trait entry point and call `record_item_access_internal`
+        // directly — Read has already been enforced upstream on
+        // whatever `_with_perms` service produced the access event.
+        let resource = Resource::parse(item_type, item_id)?;
+        self.authorization
+            .require(Subject::User(user_id), Permission::Read, resource)
+            .await?;
 
-        self.repo.upsert_access(user_id, item_id, item_type).await?;
-        self.repo.prune(user_id, self.max_recent_items).await?;
+        self.record_item_access_internal(user_id, item_id, item_type)
+            .await?;
 
         info!(
             "Successfully recorded access to {} '{}' for user {}",

src/application/services/recipient_notification_service.rs

@@ -472,11 +472,14 @@ impl RecipientNotificationService {
         let kind_key = match resource {
             Resource::Folder(_) => "server.magic_link.email.kind_folder",
             Resource::File(_) => "server.magic_link.email.kind_file",
-            // Drives don't generate share notifications in D0 — drive
-            // sharing lands in D2 and gets its own template key. Fall
-            // back to the folder label so any path that does reach
-            // here produces a readable, if generic, mail body.
-            Resource::Drive(_) => "server.magic_link.email.kind_folder",
+            // Drive / Calendar / AddressBook / Playlist shares don't
+            // produce email notifications through this path. Fall
+            // back to the folder label so any code that does reach
+            // here still produces a readable (if generic) mail body.
+            Resource::Drive(_)
+            | Resource::Calendar(_)
+            | Resource::AddressBook(_)
+            | Resource::Playlist(_) => "server.magic_link.email.kind_folder",
         };
         let kind_label = self.i18n_or(kind_key, &locale, &[]).await;
         // Short form for the subject, long form (with email) for the

src/application/services/search_service.rs

@@ -283,20 +283,10 @@ impl SearchService {
             return Vec::new();
         };
 
-        // Resolve the caller's accessible drive set via the engine
-        // (handles group-mediated drive grants) + the repo lookup.
-        let caller = Subject::User(user_id);
-        let (subject_types, subject_ids) = match authz.expand_subject_for_listing(caller).await {
-            Ok(pair) => pair,
-            Err(e) => {
-                tracing::warn!("Content-index: subject expansion failed — degrading to empty: {e}");
-                return Vec::new();
-            }
-        };
-        let accessible_drives: Vec<Uuid> = match drive_repo
-            .list_for_subjects(&subject_types, &subject_ids)
-            .await
-        {
+        // Resolve the caller's accessible drive set. Group-mediated
+        // grants are honoured inline by `storage.caller_group_ids` on
+        // the SQL side, so no Rust-side subject expansion here.
+        let accessible_drives: Vec<Uuid> = match drive_repo.list_readable_by(user_id).await {
             Ok(drives) => drives.into_iter().map(|d| d.drive.id).collect(),
             Err(e) => {
                 tracing::warn!("Content-index: drive lookup failed — degrading to empty: {e}");
@@ -338,7 +328,11 @@ impl SearchService {
                 }
             };
             match authz
-                .check(caller, Permission::Read, Resource::File(file_uuid))
+                .check(
+                    Subject::User(user_id),
+                    Permission::Read,
+                    Resource::File(file_uuid),
+                )
                 .await
             {
                 Ok(true) => verified.push(hit),

src/application/services/share_service.rs

@@ -6,7 +6,7 @@ use uuid::Uuid;
 
 use crate::domain::repositories::drive_repository::DriveRepository;
 use crate::domain::repositories::folder_repository::FolderRepository;
-use crate::domain::services::authorization::{Resource, Role, Subject};
+use crate::domain::services::authorization::{Permission, Resource, Role, Subject};
 use crate::infrastructure::repositories::pg::DrivePgRepository;
 use crate::infrastructure::repositories::pg::SharePgRepository;
 use crate::infrastructure::repositories::pg::file_blob_read_repository::FileBlobReadRepository;
@@ -243,6 +243,28 @@ impl ShareUseCase for ShareService {
 
         self.verify_item_exists(&dto.item_id, &item_type).await?;
 
+        // AuthZ: only callers with `Share` on the resource may mint a
+        // public link. Without this gate, an ex-Viewer who kept a
+        // guessed UUID could launder a temporary read into a
+        // permanent anonymous URL that survives their own grant
+        // revocation. `Permission::Share` is bundled with the
+        // `owner` and `editor` role_grants only. `require` returns
+        // `not_found` on denial (anti-enum, matches the shape used
+        // by every other share route). See `docs/plan/authz_audit/`.
+        let item_uuid_for_authz = Uuid::parse_str(&dto.item_id)
+            .map_err(|_| ShareServiceError::Validation("Invalid item UUID".to_string()))?;
+        let resource_for_authz = match item_type {
+            ShareItemType::File => Resource::File(item_uuid_for_authz),
+            ShareItemType::Folder => Resource::Folder(item_uuid_for_authz),
+        };
+        self.authorization
+            .require(
+                Subject::User(user_id),
+                Permission::Share,
+                resource_for_authz,
+            )
+            .await?;
+
         // D5: `forbid_public_links` policy gate. The drive owner can
         // disable anonymous-link creation on every resource in their
         // drive without per-resource intervention. Lookup is one JOIN
@@ -928,14 +950,6 @@ mod tests {
         > {
             Ok(Box::pin(futures::stream::empty()))
         }
-
-        async fn get_file_for_owner(
-            &self,
-            id: &str,
-            _owner_id: Uuid,
-        ) -> Result<crate::domain::entities::file::File, DomainError> {
-            self.get_file(id).await
-        }
     }
 
     impl FolderRepository for MockFolderRepository {
@@ -983,10 +997,9 @@ mod tests {
             unimplemented!()
         }
 
-        async fn list_folders_by_owner(
+        async fn list_root_folders_for_caller(
             &self,
-            _parent_id: Option<&str>,
-            _owner_id: Uuid,
+            _caller_id: Uuid,
         ) -> Result<Vec<crate::domain::entities::folder::Folder>, DomainError> {
             unimplemented!()
         }
@@ -1002,10 +1015,9 @@ mod tests {
             unimplemented!()
         }
 
-        async fn list_folders_by_owner_paginated(
+        async fn list_root_folders_for_caller_paginated(
             &self,
-            _parent_id: Option<&str>,
-            _owner_id: Uuid,
+            _caller_id: Uuid,
             _offset: usize,
             _limit: usize,
             _include_total: bool,

src/application/services/storage_usage_service.rs

@@ -213,6 +213,47 @@ impl StorageUsageService {
         Ok(())
     }
 
+    /// Return the size in bytes of a single non-trashed file. `None`
+    /// if the file is trashed or absent. Used by cross-drive MOVE to
+    /// know how many bytes will land on the destination drive so the
+    /// pre-move `check_drive_quota` call can fire.
+    pub async fn file_bytes(&self, file_id: Uuid) -> Result<Option<i64>, DomainError> {
+        let row: Option<(i64,)> = sqlx::query_as(
+            "SELECT size::bigint FROM storage.files WHERE id = $1 AND NOT is_trashed",
+        )
+        .bind(file_id)
+        .fetch_optional(self.pool.as_ref())
+        .await
+        .map_err(|e| DomainError::internal_error("StorageUsage", format!("file_bytes: {e}")))?;
+        Ok(row.map(|(s,)| s))
+    }
+
+    /// Sum the sizes of every non-trashed file whose parent folder is
+    /// `folder_id` itself or a descendant of it via the `lpath` ltree.
+    /// Used by cross-drive MOVE to know how many bytes would land on
+    /// the destination drive — necessary for the pre-move
+    /// `check_drive_quota` call.
+    ///
+    /// Returns 0 for an empty subtree AND for a non-existent
+    /// `folder_id` (the JOIN silently drops); callers that need to
+    /// distinguish those two cases must probe the folder separately.
+    pub async fn folder_subtree_bytes(&self, folder_id: Uuid) -> Result<i64, DomainError> {
+        let (bytes,): (Option<i64>,) = sqlx::query_as(
+            "SELECT COALESCE(SUM(f.size), 0)::bigint
+               FROM storage.files f
+               JOIN storage.folders fo ON fo.id = f.folder_id
+              WHERE fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1)
+                AND NOT f.is_trashed",
+        )
+        .bind(folder_id)
+        .fetch_one(self.pool.as_ref())
+        .await
+        .map_err(|e| {
+            DomainError::internal_error("StorageUsage", format!("folder_subtree_bytes: {e}"))
+        })?;
+        Ok(bytes.unwrap_or(0))
+    }
+
     /// Same as [`Self::add_drive_storage_usage_delta`] but resolves
     /// the drive id from a parent folder id in a single statement.
     /// Avoids a separate `SELECT drive_id FROM storage.folders` round

src/application/services/trash_service.rs

@@ -786,13 +786,9 @@ impl TrashService {
     /// keeps the two HTTP surfaces semantically consistent and avoids
     /// duplicating the subject-expansion plumbing.
     async fn drives_with_delete_for(&self, user_id: Uuid) -> Result<Vec<Uuid>> {
-        let (subject_types, subject_ids) = self
-            .authz
-            .expand_subject_for_listing(Subject::User(user_id))
-            .await?;
         let drives = self
             .drive_repo
-            .list_for_subjects(&subject_types, &subject_ids)
+            .list_readable_by(user_id)
             .await
             .map_err(|e| {
                 DomainError::internal_error(
@@ -900,15 +896,7 @@ impl TrashService {
         // D2b: scope by drives the caller can read (resolved through
         // role_grants on resource_type='drive', including group-mediated
         // grants). Empty set → empty page without a SQL round-trip.
-        let (subject_types, subject_ids) = self
-            .authz
-            .expand_subject_for_listing(Subject::User(user_id))
-            .await?;
-        let drive_ids: Vec<Uuid> = match self
-            .drive_repo
-            .list_for_subjects(&subject_types, &subject_ids)
-            .await
-        {
+        let drive_ids: Vec<Uuid> = match self.drive_repo.list_readable_by(user_id).await {
             Ok(drives) => drives.into_iter().map(|d| d.drive.id).collect(),
             Err(e) => {
                 return Err(DomainError::internal_error(
@@ -974,7 +962,6 @@ fn row_to_item_dto(row: TrashResourceRow) -> TrashResourceItemDto {
             name: row.name.clone(),
             path,
             parent_id: row.parent_id.map(|u| u.to_string()),
-            owner_id: Some(row.owner_id.to_string()),
             // D2b: the trash listing query now SELECTs `drive_id` (the
             // unified view exposes it). Surfaced so per-drive grouping
             // in the `/trash` UI doesn't need an extra lookup per row.
@@ -1025,7 +1012,6 @@ fn row_to_item_dto(row: TrashResourceRow) -> TrashResourceItemDto {
             icon_special_class: std::sync::Arc::from(icon_special_class_for(&row.name, mime)),
             category: std::sync::Arc::from(category_for(&row.name, mime)),
             size_formatted: format_file_size(size_bytes),
-            owner_id: Some(row.owner_id.to_string()),
             sort_date: None,
             content_hash,
             etag,

src/application/services/trash_service_test.rs

@@ -554,15 +554,6 @@ impl FileReadPort for MockFileRepository {
     > {
         Ok(Box::pin(futures::stream::empty()))
     }
-
-    async fn get_file_for_owner(
-        &self,
-        id: &str,
-        _owner_id: Uuid,
-    ) -> std::result::Result<File, DomainError> {
-        // In this mock, ignore ownership — trash tests don't focus on ownership
-        self.get_file(id).await
-    }
 }
 
 impl FileWritePort for MockFileRepository {
@@ -745,10 +736,9 @@ impl FolderRepository for MockFolderRepository {
         Ok(vec![])
     }
 
-    async fn list_folders_by_owner(
+    async fn list_root_folders_for_caller(
         &self,
-        _parent_id: Option<&str>,
-        _owner_id: Uuid,
+        _caller_id: Uuid,
     ) -> std::result::Result<Vec<Folder>, DomainError> {
         Ok(vec![])
     }
@@ -763,10 +753,9 @@ impl FolderRepository for MockFolderRepository {
         Ok((vec![], Some(0)))
     }
 
-    async fn list_folders_by_owner_paginated(
+    async fn list_root_folders_for_caller_paginated(
         &self,
-        _parent_id: Option<&str>,
-        _owner_id: Uuid,
+        _caller_id: Uuid,
         _offset: usize,
         _limit: usize,
         _include_total: bool,