Revoke attachment reads when private channel access is removed

Signed-off-by: Jordan Mecom <jm@squareup.com>

crates/buzz-db/src/runtime/migration.rs

@@ -702,7 +702,7 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 44);
+        assert_eq!(migrations.len(), 45);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]

crates/buzz-db/src/store/deletion.rs

@@ -66,6 +66,7 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[
     "events",
     "git_repo_names",
     "join_policy_acceptances",
+    "media_uploaders",
     "moderation_actions",
     "moderation_reports",
     "parameterized_event_watermarks",
@@ -92,6 +93,7 @@ pub const PURGE_SCOPED_TABLES: &[&str] = &[
     "workflow_runs",
     "push_wake_outbox",
     "join_policy_acceptances",
+    "media_uploaders",
     "moderation_reports",
     "subscriptions",
     "api_tokens",

crates/buzz-db/src/store/media.rs

@@ -0,0 +1,152 @@
+//! Media reads inherit the permissions of events that publish the local URL.
+
+use buzz_core::kind::{AUTHOR_ONLY_KINDS, KIND_AGENT_ENGRAM, P_GATED_KINDS, SHARED_GATED_KINDS};
+use buzz_core::CommunityId;
+use buzz_datastore_tracing::datastore_span;
+
+use crate::{Db, Result};
+
+#[cfg(test)]
+mod tests;
+
+impl Db {
+    /// Record possession only after the upload pipeline has hashed the bytes.
+    /// An idempotent re-upload proves possession again and may add an uploader.
+    #[datastore_span(name = "record_media_upload", system = "postgresql")]
+    pub async fn record_media_upload(
+        &self,
+        community: CommunityId,
+        sha256: &str,
+        pubkey: &[u8],
+    ) -> Result<()> {
+        let mut connection = crate::observability::acquire_writer(
+            &self.pool,
+            crate::observability::WriterOperation::EventWrite,
+        )
+        .await?;
+        sqlx::query(
+            "INSERT INTO media_uploaders (community_id, sha256, pubkey) VALUES ($1, $2, $3) \
+             ON CONFLICT DO NOTHING",
+        )
+        .bind(community.as_uuid())
+        .bind(sha256)
+        .bind(pubkey)
+        .execute(&mut *connection)
+        .await?;
+        Ok(())
+    }
+
+    /// Find local blob references with the same extractor used by the event index.
+    pub async fn local_media_hashes(
+        &self,
+        content: &str,
+        tags: &serde_json::Value,
+        host: &str,
+    ) -> Result<Vec<String>> {
+        let mut connection = crate::observability::acquire_writer(
+            &self.pool,
+            crate::observability::WriterOperation::Authorization,
+        )
+        .await?;
+        Ok(sqlx::query_scalar("SELECT buzz_media_hashes($1, $2, $3)")
+            .bind(content)
+            .bind(tags)
+            .bind(host)
+            .fetch_one(&mut *connection)
+            .await?)
+    }
+
+    /// Authorize a read against live primary-database membership, without caches.
+    /// Unreferenced blobs retain the existing community-level read policy.
+    #[datastore_span(name = "can_read_media", system = "postgresql")]
+    pub async fn can_read_media(
+        &self,
+        community: CommunityId,
+        sha256: &str,
+        pubkey: &[u8],
+    ) -> Result<bool> {
+        self.media_access(community, sha256, pubkey, false).await
+    }
+
+    /// Authorize publication before accepting an event containing a local URL.
+    /// Knowing a hash is insufficient: the author must currently have read
+    /// access or have supplied the actual bytes in a successful upload.
+    #[datastore_span(name = "can_reference_media", system = "postgresql")]
+    pub async fn can_reference_media(
+        &self,
+        community: CommunityId,
+        sha256: &str,
+        pubkey: &[u8],
+    ) -> Result<bool> {
+        self.media_access(community, sha256, pubkey, true).await
+    }
+
+    async fn media_access(
+        &self,
+        community: CommunityId,
+        sha256: &str,
+        pubkey: &[u8],
+        for_publication: bool,
+    ) -> Result<bool> {
+        let mut connection = crate::observability::acquire_writer(
+            &self.pool,
+            crate::observability::WriterOperation::Authorization,
+        )
+        .await?;
+        let as_i32 = |kinds: &[u32]| kinds.iter().map(|kind| *kind as i32).collect::<Vec<_>>();
+        Ok(sqlx::query_scalar(
+            r#"
+            WITH local_refs AS NOT MATERIALIZED (
+                SELECT e.* FROM events e
+                JOIN communities community ON community.id = e.community_id
+                WHERE e.community_id = $1 AND buzz_media_hashes(e.content, e.tags) @> ARRAY[$2]::TEXT[]
+                  AND $2 = ANY(buzz_media_hashes(e.content, e.tags, community.host))
+            )
+            SELECT NOT EXISTS (SELECT 1 FROM local_refs) OR EXISTS (
+                SELECT 1 FROM local_refs e
+                WHERE e.deleted_at IS NULL
+                  AND (e.channel_id IS NULL OR EXISTS (
+                      SELECT 1 FROM channels c
+                      WHERE c.community_id = $1 AND c.id = e.channel_id AND c.deleted_at IS NULL
+                        AND (c.visibility = 'open' OR EXISTS (
+                            SELECT 1 FROM channel_members cm
+                            WHERE cm.community_id = $1 AND cm.channel_id = c.id
+                              AND cm.pubkey = $3 AND cm.removed_at IS NULL
+                        ))
+                  ))
+                  AND (NOT (e.kind = ANY($4)) OR e.pubkey = $3)
+                  AND (NOT (e.kind = ANY($5)) OR EXISTS (
+                      SELECT 1 FROM jsonb_array_elements(e.tags) tag
+                      WHERE tag->>0 = 'p' AND tag->>1 = encode($3, 'hex')
+                  ))
+                  AND (NOT (e.kind = ANY($6)) OR e.pubkey = $3 OR (
+                      SELECT count(*) = 1 AND bool_and(tag = '["shared","true"]'::JSONB)
+                      FROM jsonb_array_elements(e.tags) tag WHERE tag->>0 = 'shared'
+                  ))
+                  AND (e.kind <> $7 OR e.pubkey = $3 OR EXISTS (
+                      SELECT 1 FROM jsonb_array_elements(e.tags) tag
+                      WHERE tag->>0 = 'p' AND tag->>1 = encode($3, 'hex')
+                  ))
+            ) OR EXISTS (
+                SELECT 1 FROM communities c
+                WHERE c.id = $1 AND $2 = ANY(buzz_media_hashes(COALESCE(c.icon, ''), '[]', c.host))
+            ) OR (
+                $8 AND EXISTS (
+                    SELECT 1 FROM media_uploaders
+                    WHERE community_id = $1 AND sha256 = $2 AND pubkey = $3
+                )
+            )
+            "#,
+        )
+        .bind(community.as_uuid())
+        .bind(sha256)
+        .bind(pubkey)
+        .bind(as_i32(AUTHOR_ONLY_KINDS))
+        .bind(as_i32(P_GATED_KINDS))
+        .bind(as_i32(SHARED_GATED_KINDS))
+        .bind(KIND_AGENT_ENGRAM as i32)
+        .bind(for_publication)
+        .fetch_one(&mut *connection)
+        .await?)
+    }
+}

crates/buzz-db/src/store/media/tests.rs

@@ -0,0 +1,302 @@
+use super::*;
+use crate::channel::{ChannelType, ChannelVisibility, MemberRole};
+use nostr::{EventBuilder, Keys, Kind, Tag};
+use serde_json::json;
+use uuid::Uuid;
+
+async fn fixture() -> (Db, CommunityId, String, Keys, Keys, Uuid) {
+    let pool = sqlx::PgPool::connect(&crate::test_support::database_url())
+        .await
+        .expect("test database");
+    crate::migration::run_migrations(&pool)
+        .await
+        .expect("migrate");
+    let db = Db::from_pool(pool);
+    let host = format!("media-{}.example", Uuid::new_v4());
+    let community = db
+        .ensure_configured_community(&host)
+        .await
+        .expect("community")
+        .id;
+    let owner = Keys::generate();
+    let member = Keys::generate();
+    for keys in [&owner, &member] {
+        db.ensure_user(community, keys.public_key().as_bytes())
+            .await
+            .expect("user");
+    }
+    let channel = db
+        .create_channel(
+            community,
+            "private",
+            ChannelType::Stream,
+            ChannelVisibility::Private,
+            None,
+            owner.public_key().as_bytes(),
+            None,
+        )
+        .await
+        .expect("channel")
+        .id;
+    db.add_member(
+        community,
+        channel,
+        member.public_key().as_bytes(),
+        MemberRole::Member,
+        Some(owner.public_key().as_bytes()),
+    )
+    .await
+    .expect("member");
+    (db, community, host, owner, member, channel)
+}
+
+async fn publish(
+    db: &Db,
+    community: CommunityId,
+    owner: &Keys,
+    channel: Option<Uuid>,
+    kind: u16,
+    content: &str,
+    tags: Vec<Tag>,
+) -> nostr::Event {
+    let event = EventBuilder::new(Kind::Custom(kind), content)
+        .tags(tags)
+        .allow_self_tagging()
+        .sign_with_keys(owner)
+        .expect("event");
+    db.insert_event(community, &event, channel)
+        .await
+        .expect("insert event");
+    event
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn existing_attachments_follow_live_membership_and_tenant() {
+    let (db, community, host, owner, member, channel) = fixture().await;
+    let sha = "a".repeat(64);
+    // No uploader row: this also covers attachments published before upgrade.
+    publish(
+        &db,
+        community,
+        &owner,
+        Some(channel),
+        9,
+        &format!("![attachment](https://{host}/media/{sha}.jpg)"),
+        vec![],
+    )
+    .await;
+    let pk = member.public_key().to_bytes();
+    assert!(db
+        .can_read_media(community, &sha, &pk)
+        .await
+        .expect("member read"));
+    assert!(!db
+        .can_read_media(community, &sha, Keys::generate().public_key().as_bytes())
+        .await
+        .expect("stranger read"));
+    db.remove_member(community, channel, &pk, owner.public_key().as_bytes())
+        .await
+        .expect("remove");
+    assert!(!db
+        .can_read_media(community, &sha, &pk)
+        .await
+        .expect("removed read"));
+    assert!(!db
+        .can_reference_media(community, &sha, &pk)
+        .await
+        .expect("removed publication"));
+    assert!(db
+        .can_read_media(community, &sha, owner.public_key().as_bytes())
+        .await
+        .expect("owner read"));
+    let (other_db, other, other_host, other_owner, _, other_channel) = fixture().await;
+    publish(
+        &other_db,
+        other,
+        &other_owner,
+        Some(other_channel),
+        9,
+        &format!("https://{other_host}/media/{sha}.jpg"),
+        vec![],
+    )
+    .await;
+    assert!(!db
+        .can_read_media(other, &sha, owner.public_key().as_bytes())
+        .await
+        .expect("other tenant"));
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn published_attachments_follow_references_without_changing_unposted_uploads() {
+    let (db, community, host, owner, member, channel) = fixture().await;
+    let sha = "b".repeat(64);
+    let pk = member.public_key().to_bytes();
+    db.record_media_upload(community, &sha, &pk)
+        .await
+        .expect("upload");
+    assert!(db
+        .can_read_media(community, &sha, &pk)
+        .await
+        .expect("draft"));
+    assert!(db
+        .can_read_media(community, &sha, owner.public_key().as_bytes())
+        .await
+        .expect("unposted media retains community access"));
+    let url = format!("https://{host}/media/{sha}.txt");
+    publish(&db, community, &member, Some(channel), 9, &url, vec![]).await;
+    db.remove_member(community, channel, &pk, owner.public_key().as_bytes())
+        .await
+        .expect("remove uploader");
+    assert!(!db
+        .can_read_media(community, &sha, &pk)
+        .await
+        .expect("removed uploader"));
+    // Supplying the bytes permits a new publication, even after leaving a channel.
+    assert!(db
+        .can_reference_media(community, &sha, &pk)
+        .await
+        .expect("possessor publication"));
+    let open = db
+        .create_channel(
+            community,
+            "open",
+            ChannelType::Stream,
+            ChannelVisibility::Open,
+            None,
+            owner.public_key().as_bytes(),
+            None,
+        )
+        .await
+        .expect("open channel")
+        .id;
+    publish(&db, community, &owner, Some(open), 9, &url, vec![]).await;
+    assert!(db
+        .can_read_media(community, &sha, &pk)
+        .await
+        .expect("shared in open channel"));
+    db.soft_delete_channel(community, open)
+        .await
+        .expect("delete shared channel");
+    assert!(!db
+        .can_read_media(community, &sha, &pk)
+        .await
+        .expect("deleted channel"));
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn extractor_matches_index_and_rejects_foreign_hosts() {
+    let (db, community, host, owner, _, channel) = fixture().await;
+    let sha = "c".repeat(64);
+    let url = format!("https://{host}/media/{sha}.thumb.jpg");
+    for content in [
+        url.clone(),
+        format!("![image]({url})"),
+        json!({"picture":url}).to_string(),
+        url.replace('/', "\\/"),
+    ] {
+        assert_eq!(
+            db.local_media_hashes(&content, &json!([]), &host)
+                .await
+                .expect("extract"),
+            vec![sha.clone()]
+        );
+    }
+    let tags = json!([["imeta", format!("url {url}"), format!("x {sha}")]]);
+    assert_eq!(
+        db.local_media_hashes("", &tags, &host).await.expect("tags"),
+        vec![sha.clone()]
+    );
+    for content in [
+        format!("https://foreign.example/media/{sha}.jpg"),
+        format!("https://{host}.evil/media/{sha}.jpg"),
+        format!("/media/{sha}a.jpg"),
+    ] {
+        assert!(db
+            .local_media_hashes(&content, &json!([]), &host)
+            .await
+            .expect("foreign")
+            .is_empty());
+    }
+    let event = publish(&db, community, &owner, Some(channel), 9, &url, vec![]).await;
+    let indexed: Vec<String> = sqlx::query_scalar(
+        "SELECT buzz_media_hashes(content, tags) FROM events WHERE community_id = $1 AND id = $2",
+    )
+    .bind(community.as_uuid())
+    .bind(event.id.as_bytes().as_slice())
+    .fetch_one(&db.pool)
+    .await
+    .expect("indexed extractor");
+    assert_eq!(indexed, vec![sha]);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn private_event_kinds_do_not_grant_community_wide_media_access() {
+    let (db, community, host, owner, member, _) = fixture().await;
+    let mut cases = vec![
+        (0_u16, vec![], true),
+        (KIND_AGENT_ENGRAM as u16, vec![], false),
+    ];
+    cases.extend(AUTHOR_ONLY_KINDS.iter().map(|k| (*k as u16, vec![], false)));
+    cases.push((
+        buzz_core::kind::KIND_EVENT_REMINDER as u16,
+        vec![Tag::parse([
+            "not_before",
+            &(nostr::Timestamp::now().as_secs() + 300).to_string(),
+        ])
+        .expect("schedule")],
+        false,
+    ));
+    cases.extend(
+        P_GATED_KINDS
+            .iter()
+            .filter(|k| !buzz_core::kind::is_ephemeral(**k))
+            .map(|k| {
+                (
+                    *k as u16,
+                    vec![Tag::parse(["p", &owner.public_key().to_hex()]).expect("p")],
+                    false,
+                )
+            }),
+    );
+    for kind in SHARED_GATED_KINDS {
+        cases.push((*kind as u16, vec![], false));
+        cases.push((
+            *kind as u16,
+            vec![Tag::parse(["shared", "true"]).expect("shared")],
+            true,
+        ));
+        cases.push((
+            *kind as u16,
+            vec![Tag::parse(["shared", "true", "extra"]).expect("malformed shared")],
+            false,
+        ));
+    }
+    for (i, (kind, tags, allowed)) in cases.into_iter().enumerate() {
+        let sha = format!("{i:064x}");
+        publish(
+            &db,
+            community,
+            &owner,
+            None,
+            kind,
+            &format!("https://{host}/media/{sha}.png"),
+            tags,
+        )
+        .await;
+        assert!(db
+            .can_read_media(community, &sha, owner.public_key().as_bytes())
+            .await
+            .expect("author read"));
+        assert_eq!(
+            db.can_read_media(community, &sha, member.public_key().as_bytes())
+                .await
+                .expect("kind read"),
+            allowed,
+            "kind {kind}, case {i}"
+        );
+    }
+}

crates/buzz-db/src/store/mod.rs

@@ -24,6 +24,8 @@ pub mod event;
 pub mod feed;
 /// Git repository name registry (NIP-34 kind:30617).
 pub mod git_repo;
+/// Blob upload ownership and event-derived read authorization.
+pub mod media;
 /// Community moderation: reports, bans/timeouts, audit actions.
 pub mod moderation;
 /// Monthly table partition management.

crates/buzz-relay/src/api/media.rs

@@ -436,6 +436,21 @@ pub async fn upload_blob(
             }
         })??;
 
+    // Every successful pipeline, including deduplicated uploads, has verified
+    // the supplied bytes against this hash. A signed hash alone proves nothing.
+    state
+        .db
+        .record_media_upload(
+            auth.tenant.community(),
+            &descriptor.sha256,
+            auth.auth_event.pubkey.as_bytes(),
+        )
+        .await
+        .map_err(|error| {
+            tracing::error!(%error, "failed to record media uploader");
+            MediaError::ServiceUnavailable
+        })?;
+
     rewrite_descriptor_urls_for_tenant(
         &mut descriptor,
         &state.config.relay_url,
@@ -546,11 +561,23 @@ async fn authenticate_media_read(
     .await
     .map_err(|_| MediaError::RelayMembershipRequired)?;
 
+    let allowed = state
+        .db
+        .can_read_media(tenant.community(), sha256, auth_event.pubkey.as_bytes())
+        .await
+        .map_err(|error| {
+            tracing::error!(%error, "media authorization lookup failed");
+            MediaError::ServiceUnavailable
+        })?;
+    if !allowed {
+        return Err(MediaError::NotFound);
+    }
+
     Ok(MediaReadAuth { tenant })
 }
 
 fn blob_cache_control() -> &'static str {
-    "private, max-age=31536000, immutable"
+    "private, no-store"
 }
 
 /// Whether a path-segment extension is a safe token.
@@ -1008,6 +1035,10 @@ fn extract_blossom_auth(headers: &HeaderMap) -> Result<nostr::Event, MediaError>
     Ok(event)
 }
 
+#[cfg(test)]
+#[path = "media_access_tests.rs"]
+mod access_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1131,12 +1162,25 @@ mod tests {
     }
 
     async fn test_state() -> Arc<AppState> {
+        test_state_with_media(None).await
+    }
+
+    pub(super) async fn test_state_with_media(endpoint: Option<&str>) -> Arc<AppState> {
         let mut config = crate::config::Config::from_env().expect("default config loads");
+        config.database_url = crate::test_support::database_url();
         config.require_relay_membership = false;
         config.redis_url = "redis://127.0.0.1:1".to_string();
         config.media_uploads_per_minute = 1;
         config.media_max_concurrent_uploads = 2;
         config.media_max_concurrent_uploads_per_pubkey = 1;
+        if let Some(endpoint) = endpoint {
+            config.media.s3_endpoint = endpoint.to_string();
+            config.media.s3_access_key = "test".into();
+            config.media.s3_secret_key = "test".into();
+            config.media.s3_addressing_style = buzz_media::config::S3AddressingStyle::Path;
+            config.media.upload_records_enabled = false;
+            config.media_uploads_per_minute = 100;
+        }
 
         let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool");
         let db = buzz_db::Db::from_pool(pool.clone());

crates/buzz-relay/src/api/media_access_tests.rs

@@ -0,0 +1,409 @@
+use super::*;
+use axum::body::{to_bytes, Body, Bytes};
+use axum::http::{Method, Request};
+use buzz_core::TenantContext;
+use buzz_db::channel::{ChannelType, ChannelVisibility, MemberRole};
+use nostr::{EventBuilder, JsonUtil, Keys, Kind, Tag, Timestamp};
+use sha2::{Digest, Sha256};
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use tokio::sync::Mutex;
+use tower::ServiceExt;
+use uuid::Uuid;
+
+type Objects = Arc<Mutex<HashMap<String, Bytes>>>;
+
+struct ObjectServer {
+    endpoint: String,
+    requests: Arc<AtomicUsize>,
+    task: tokio::task::JoinHandle<()>,
+}
+
+impl Drop for ObjectServer {
+    fn drop(&mut self) {
+        self.task.abort();
+    }
+}
+
+impl ObjectServer {
+    async fn start() -> Self {
+        async fn object(
+            State((objects, requests)): State<(Objects, Arc<AtomicUsize>)>,
+            Path(key): Path<String>,
+            request: Request<Body>,
+        ) -> Response {
+            requests.fetch_add(1, Ordering::SeqCst);
+            if request.method() == Method::PUT {
+                let bytes = to_bytes(request.into_body(), 1024 * 1024)
+                    .await
+                    .expect("object body");
+                objects.lock().await.insert(key, bytes);
+                return StatusCode::OK.into_response();
+            }
+            let Some(bytes) = objects.lock().await.get(&key).cloned() else {
+                return StatusCode::NOT_FOUND.into_response();
+            };
+            let mut status = StatusCode::OK;
+            let mut data = bytes.clone();
+            if let Some(range) = request.headers().get(header::RANGE) {
+                let (start, end) = range
+                    .to_str()
+                    .expect("range")
+                    .strip_prefix("bytes=")
+                    .expect("bytes")
+                    .split_once('-')
+                    .expect("bounds");
+                let start: usize = start.parse().expect("start");
+                let end = end
+                    .parse::<usize>()
+                    .unwrap_or(bytes.len() - 1)
+                    .min(bytes.len() - 1);
+                data = bytes.slice(start..=end);
+                status = StatusCode::PARTIAL_CONTENT;
+            }
+            Response::builder()
+                .status(status)
+                .header(header::CONTENT_LENGTH, data.len())
+                .body(if request.method() == Method::HEAD {
+                    Body::empty()
+                } else {
+                    Body::from(data)
+                })
+                .expect("object response")
+        }
+        let requests = Arc::new(AtomicUsize::new(0));
+        let router = axum::Router::new()
+            .route("/{*key}", axum::routing::any(object))
+            .with_state((Objects::default(), requests.clone()));
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind object server");
+        let endpoint = format!("http://{}", listener.local_addr().expect("address"));
+        let task =
+            tokio::spawn(
+                async move { axum::serve(listener, router).await.expect("object server") },
+            );
+        Self {
+            endpoint,
+            requests,
+            task,
+        }
+    }
+}
+
+fn auth(keys: &Keys, host: &str, verb: &str, sha: &str) -> String {
+    let expiry = (Timestamp::now().as_secs() + 300).to_string();
+    let event = EventBuilder::new(Kind::Custom(24242), "media test")
+        .tags([
+            Tag::parse(["t", verb]).expect("t"),
+            Tag::parse(["x", sha]).expect("x"),
+            Tag::parse(["server", host]).expect("server"),
+            Tag::parse(["expiration", &expiry]).expect("expiration"),
+        ])
+        .sign_with_keys(keys)
+        .expect("auth event");
+    format!(
+        "Nostr {}",
+        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(event.as_json())
+    )
+}
+
+fn request(
+    keys: &Keys,
+    host: &str,
+    sha: &str,
+    path: &str,
+    method: &str,
+    range: bool,
+) -> Request<Body> {
+    let mut request = Request::builder()
+        .method(method)
+        .uri(path)
+        .header(header::HOST, host)
+        .header(header::AUTHORIZATION, auth(keys, host, "get", sha));
+    if range {
+        request = request.header(header::RANGE, "bytes=0-3");
+    }
+    request.body(Body::empty()).expect("request")
+}
+
+async fn state(server: &ObjectServer) -> (Arc<AppState>, TenantContext, Keys, Keys, Uuid) {
+    let pool = sqlx::PgPool::connect(&crate::test_support::database_url())
+        .await
+        .expect("database");
+    buzz_db::migration::run_migrations(&pool)
+        .await
+        .expect("migrate");
+    let state = super::tests::test_state_with_media(Some(&server.endpoint)).await;
+    let host = format!("media-http-{}.example", Uuid::new_v4());
+    state
+        .db
+        .ensure_configured_community(&host)
+        .await
+        .expect("community");
+    let tenant = crate::tenant::bind_community(&state.db, &host)
+        .await
+        .expect("tenant");
+    let owner = Keys::generate();
+    let member = Keys::generate();
+    for key in [&owner, &member] {
+        state
+            .db
+            .ensure_user(tenant.community(), key.public_key().as_bytes())
+            .await
+            .expect("user");
+    }
+    let channel = state
+        .db
+        .create_channel(
+            tenant.community(),
+            "private",
+            ChannelType::Stream,
+            ChannelVisibility::Private,
+            None,
+            owner.public_key().as_bytes(),
+            None,
+        )
+        .await
+        .expect("channel")
+        .id;
+    state
+        .db
+        .add_member(
+            tenant.community(),
+            channel,
+            member.public_key().as_bytes(),
+            MemberRole::Member,
+            Some(owner.public_key().as_bytes()),
+        )
+        .await
+        .expect("add member");
+    (state, tenant, owner, member, channel)
+}
+
+fn router(state: Arc<AppState>) -> axum::Router {
+    axum::Router::new()
+        .route(
+            "/media/{sha256_ext}",
+            axum::routing::get(get_blob).head(head_blob),
+        )
+        .route("/upload", axum::routing::put(upload_blob))
+        .with_state(state)
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn removal_denies_get_head_ranges_and_thumbnails_before_storage() {
+    let server = ObjectServer::start().await;
+    let (state, tenant, owner, member, channel) = state(&server).await;
+    let bytes = Bytes::from_static(b"private attachment");
+    let sha = hex::encode(Sha256::digest(&bytes));
+    let upload = Request::builder()
+        .method("PUT")
+        .uri("/upload")
+        .header(header::HOST, tenant.host())
+        .header("X-SHA-256", &sha)
+        .header(
+            header::AUTHORIZATION,
+            auth(&owner, tenant.host(), "upload", &sha),
+        )
+        .body(Body::from(bytes.clone()))
+        .expect("upload request");
+    let response = router(state.clone()).oneshot(upload).await.expect("upload");
+    assert_eq!(response.status(), StatusCode::OK);
+    let descriptor: buzz_media::BlobDescriptor = serde_json::from_slice(
+        &to_bytes(response.into_body(), 16384)
+            .await
+            .expect("descriptor"),
+    )
+    .expect("json");
+    assert!(state
+        .db
+        .can_reference_media(tenant.community(), &sha, owner.public_key().as_bytes())
+        .await
+        .expect("upload proof"));
+    let path = url::Url::parse(&descriptor.url)
+        .expect("url")
+        .path()
+        .to_string();
+    state
+        .media_storage
+        .put(&format!("{sha}.thumb.jpg"), &bytes, "image/jpeg")
+        .await
+        .expect("thumbnail fixture");
+    let event = EventBuilder::new(Kind::Custom(9), &descriptor.url)
+        .sign_with_keys(&owner)
+        .expect("event");
+    state
+        .db
+        .insert_event(tenant.community(), &event, Some(channel))
+        .await
+        .expect("publish attachment");
+    let paths = [
+        path,
+        format!("/media/{sha}"),
+        format!("/media/{sha}.thumb.jpg"),
+    ];
+    for path in &paths {
+        for (method, range, status) in [
+            ("GET", false, StatusCode::OK),
+            ("HEAD", false, StatusCode::OK),
+            ("GET", true, StatusCode::PARTIAL_CONTENT),
+        ] {
+            let response = router(state.clone())
+                .oneshot(request(&member, tenant.host(), &sha, path, method, range))
+                .await
+                .expect("member request");
+            assert_eq!(response.status(), status, "{method} {path}");
+            assert_eq!(
+                response.headers()[header::CACHE_CONTROL],
+                "private, no-store"
+            );
+        }
+    }
+    state
+        .db
+        .remove_member(
+            tenant.community(),
+            channel,
+            member.public_key().as_bytes(),
+            owner.public_key().as_bytes(),
+        )
+        .await
+        .expect("remove");
+    // Poison the usual caches: media authorization must still consult live membership.
+    state.membership_cache.insert(
+        (
+            tenant.community(),
+            channel,
+            member.public_key().to_bytes().to_vec(),
+        ),
+        true,
+    );
+    state.accessible_channels_cache.insert(
+        (tenant.community(), member.public_key().to_bytes().to_vec()),
+        vec![channel],
+    );
+    let before = server.requests.load(Ordering::SeqCst);
+    for path in &paths {
+        for (method, range) in [("GET", false), ("HEAD", false), ("GET", true)] {
+            let response = router(state.clone())
+                .oneshot(request(&member, tenant.host(), &sha, path, method, range))
+                .await
+                .expect("removed request");
+            assert_eq!(response.status(), StatusCode::NOT_FOUND, "{method} {path}");
+        }
+    }
+    assert_eq!(
+        server.requests.load(Ordering::SeqCst),
+        before,
+        "denied reads must not touch storage"
+    );
+    assert_eq!(
+        router(state.clone())
+            .oneshot(request(
+                &owner,
+                tenant.host(),
+                &sha,
+                &paths[0],
+                "GET",
+                false
+            ))
+            .await
+            .expect("owner")
+            .status(),
+        StatusCode::OK
+    );
+
+    let open = state
+        .db
+        .create_channel(
+            tenant.community(),
+            "open",
+            ChannelType::Stream,
+            ChannelVisibility::Open,
+            None,
+            owner.public_key().as_bytes(),
+            None,
+        )
+        .await
+        .expect("open channel")
+        .id;
+    for (kind, content, tags) in [
+        (
+            0,
+            serde_json::json!({"picture":descriptor.url}).to_string(),
+            vec![],
+        ),
+        (
+            9,
+            descriptor.url.clone(),
+            vec![Tag::parse(["h", &open.to_string()]).expect("channel tag")],
+        ),
+    ] {
+        let forged = EventBuilder::new(Kind::Custom(kind), content)
+            .tags(tags)
+            .sign_with_keys(&member)
+            .expect("copied URL");
+        let result = crate::handlers::ingest::ingest_event(
+            &state,
+            &tenant,
+            forged,
+            crate::handlers::ingest::IngestAuth::Nip42 {
+                pubkey: member.public_key(),
+                scopes: buzz_auth::Scope::all_known(),
+                channel_ids: None,
+                conn_id: Uuid::new_v4(),
+            },
+        )
+        .await;
+        assert!(
+            matches!(result, Err(crate::handlers::ingest::IngestError::Rejected(ref reason)) if reason == "restricted: attachment is not accessible"),
+            "copied URL must not create a read grant: {:?}",
+            result.err()
+        );
+    }
+
+    // Deduplication must still hash the supplied body before recording possession.
+    for (body, status, can_publish) in [
+        (
+            Bytes::from_static(b"wrong bytes"),
+            StatusCode::UNAUTHORIZED,
+            false,
+        ),
+        (bytes, StatusCode::OK, true),
+    ] {
+        let upload = Request::builder()
+            .method("PUT")
+            .uri("/upload")
+            .header(header::HOST, tenant.host())
+            .header("X-SHA-256", &sha)
+            .header(
+                header::AUTHORIZATION,
+                auth(&member, tenant.host(), "upload", &sha),
+            )
+            .body(Body::from(body))
+            .expect("reupload request");
+        assert_eq!(
+            router(state.clone())
+                .oneshot(upload)
+                .await
+                .expect("reupload")
+                .status(),
+            status
+        );
+        assert_eq!(
+            state
+                .db
+                .can_reference_media(tenant.community(), &sha, member.public_key().as_bytes())
+                .await
+                .expect("publication access"),
+            can_publish
+        );
+        assert!(!state
+            .db
+            .can_read_media(tenant.community(), &sha, member.public_key().as_bytes())
+            .await
+            .expect("read access"));
+    }
+}

crates/buzz-relay/src/handlers/ingest.rs

@@ -2283,6 +2283,8 @@ async fn ingest_event_inner(
         )));
     }
 
+    super::media_access::validate_media_references(tenant, state, &event).await?;
+
     // Command kinds are routed AFTER signature verification, timestamp check,
     // pubkey/auth match, and scope validation — never before.
     if buzz_core::kind::is_command_kind(kind_u32) {

crates/buzz-relay/src/handlers/media_access.rs

@@ -0,0 +1,60 @@
+//! Authorize local attachment references before their containing event is stored.
+
+use buzz_core::TenantContext;
+use nostr::Event;
+
+use super::ingest::IngestError;
+use crate::state::AppState;
+
+/// A reference can grant access to additional readers. Check the author's
+/// existing access first so a removed member cannot publish a known private
+/// URL into their profile or another channel to regain download access.
+pub async fn validate_media_references(
+    tenant: &TenantContext,
+    state: &AppState,
+    event: &Event,
+) -> Result<(), IngestError> {
+    // The SQL extractor only recognizes literal /media/ URLs (including JSON
+    // escaped slashes). Most events have none and need no additional DB read.
+    if !event.content.contains("media")
+        && !event
+            .tags
+            .iter()
+            .any(|tag| tag.as_slice().iter().any(|part| part.contains("media")))
+    {
+        return Ok(());
+    }
+    let tags = serde_json::to_value(&event.tags).map_err(|error| {
+        IngestError::Internal(format!(
+            "error: media reference serialization failed: {error}"
+        ))
+    })?;
+    let hashes = state
+        .db
+        .local_media_hashes(&event.content, &tags, tenant.host())
+        .await
+        .map_err(|error| {
+            IngestError::Internal(format!("error: media reference lookup failed: {error}"))
+        })?;
+    if hashes.len() > 64 {
+        return Err(IngestError::Rejected(
+            "invalid: too many media references".into(),
+        ));
+    }
+    let author = super::ingest::effective_message_author(event, &state.relay_keypair.public_key());
+    for sha256 in hashes {
+        let allowed = state
+            .db
+            .can_reference_media(tenant.community(), &sha256, &author)
+            .await
+            .map_err(|error| {
+                IngestError::Internal(format!("error: media access lookup failed: {error}"))
+            })?;
+        if !allowed {
+            return Err(IngestError::Rejected(
+                "restricted: attachment is not accessible".into(),
+            ));
+        }
+    }
+    Ok(())
+}

crates/buzz-relay/src/handlers/mod.rs

@@ -20,6 +20,8 @@ pub mod identity_archive;
 pub mod imeta;
 /// Transport-neutral event ingestion pipeline.
 pub mod ingest;
+/// Local attachment publication authorization.
+pub mod media_access;
 /// Community moderation authorization seam (capability helper).
 pub mod moderation_authz;
 /// Community moderation command handler (kinds 9040–9044).

migrations/0045_media_read_authorization.sql

@@ -0,0 +1,25 @@
+-- Index references in existing events as well as new publications. The optional
+-- host filter keeps a foreign relay URL from granting access to a local blob.
+CREATE FUNCTION buzz_media_hashes(body TEXT, tags JSONB, media_host TEXT DEFAULT NULL)
+RETURNS TEXT[] LANGUAGE SQL IMMUTABLE PARALLEL SAFE AS $$
+    SELECT COALESCE(array_agg(DISTINCT lower(parts[2])), ARRAY[]::TEXT[])
+    FROM regexp_matches(
+        replace(body || ' ' || tags::TEXT, E'\\/', '/'),
+        '(?:https?://([^/[:space:]"<>]+))?/media/([0-9a-f]{64})(?=$|[^0-9a-f])',
+        'g'
+    ) AS parts
+    WHERE media_host IS NULL OR parts[1] IS NULL OR lower(parts[1]) = lower(media_host)
+$$;
+
+CREATE INDEX idx_events_media_hashes ON events USING GIN (buzz_media_hashes(content, tags));
+
+-- A successful upload proves possession of the bytes, not merely knowledge of
+-- their hash. Keep this independently of optional moderation/audit recording.
+CREATE TABLE media_uploaders (
+    community_id UUID NOT NULL REFERENCES communities(id),
+    sha256 TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'),
+    pubkey BYTEA NOT NULL CHECK (length(pubkey) = 32),
+    PRIMARY KEY (community_id, sha256, pubkey)
+);
+
+SELECT attach_community_write_fence('media_uploaders'::regclass);

schema/schema.sql

@@ -200,6 +200,19 @@ CREATE UNIQUE INDEX idx_users_okta ON users (community_id, okta_user_id)
 -- Cross-community dedup: same signed event may exist in two communities;
 -- (community_id, created_at, id) dedupes within one, allows across.
 
+-- Index references in existing events as well as new publications. The optional
+-- host filter keeps a foreign relay URL from granting access to a local blob.
+CREATE FUNCTION buzz_media_hashes(body TEXT, tags JSONB, media_host TEXT DEFAULT NULL)
+RETURNS TEXT[] LANGUAGE SQL IMMUTABLE PARALLEL SAFE AS $$
+    SELECT COALESCE(array_agg(DISTINCT lower(parts[2])), ARRAY[]::TEXT[])
+    FROM regexp_matches(
+        replace(body || ' ' || tags::TEXT, E'\\/', '/'),
+        '(?:https?://([^/[:space:]"<>]+))?/media/([0-9a-f]{64})(?=$|[^0-9a-f])',
+        'g'
+    ) AS parts
+    WHERE media_host IS NULL OR parts[1] IS NULL OR lower(parts[1]) = lower(media_host)
+$$;
+
 CREATE TABLE events (
     community_id UUID NOT NULL REFERENCES communities(id),
     id          BYTEA NOT NULL,
@@ -276,6 +289,9 @@ CREATE INDEX idx_events_not_before ON events (community_id, not_before)
 -- is supplied by the community-leading btree filters above (BitmapAnd), so this
 -- stays a single-column GIN. The search lane confirms the final spelling with
 -- EXPLAIN before its work lands (Quinn option A; Max's index-spelling caveat).
+-- idx_events_media_hashes is created by reconcile-schema-after-pgschema.sql:
+-- pgschema preserves its temporary schema name in function-based index expressions.
+
 CREATE INDEX idx_events_search_tsv ON events USING GIN (search_tsv);
 
 -- ── Event mentions ────────────────────────────────────────────────────────────
@@ -1893,3 +1909,16 @@ CREATE INDEX idx_relay_operator_audit_target
 INSERT INTO _operator_global_tables (table_name, reason) VALUES
     ('relay_operator_audit', 'deployment-global append-only roster mutation audit trail; no community_id intentionally');
 
+
+-- A successful upload proves possession of the bytes, not merely knowledge of
+-- their hash. Keep this independently of optional moderation/audit recording.
+CREATE TABLE media_uploaders (
+    community_id UUID NOT NULL REFERENCES communities(id),
+    sha256 TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'),
+    pubkey BYTEA NOT NULL CHECK (length(pubkey) = 32),
+    PRIMARY KEY (community_id, sha256, pubkey)
+);
+
+CREATE TRIGGER community_write_fence_media_uploaders
+    BEFORE INSERT OR UPDATE OR DELETE ON media_uploaders
+    FOR EACH ROW EXECUTE FUNCTION enforce_community_write_fence();

scripts/reconcile-schema-after-pgschema.sql

@@ -6,8 +6,25 @@
 -- apply caller must run this idempotent script so fresh bootstraps converge on
 -- the same live database contract as migration-managed databases.
 
+-- Function-based indexes retain pgschema's temporary schema qualifier in the
+-- generated plan. Create this index after functions exist in the real schema.
+CREATE INDEX IF NOT EXISTS idx_events_media_hashes
+    ON events USING GIN (buzz_media_hashes(content, tags));
+
 DO $$
 BEGIN
+    IF NOT EXISTS (
+        SELECT 1 FROM pg_index i
+        JOIN pg_class c ON c.oid = i.indexrelid
+        JOIN pg_am am ON am.oid = c.relam
+        WHERE i.indexrelid = 'idx_events_media_hashes'::regclass
+          AND i.indrelid = 'events'::regclass AND i.indisvalid
+          AND am.amname = 'gin'
+          AND pg_get_expr(i.indexprs, i.indrelid) = 'buzz_media_hashes(content, tags)'
+    ) THEN
+        RAISE EXCEPTION 'events must index media references after pgschema apply';
+    END IF;
+
     IF NOT EXISTS (
         SELECT 1 FROM pg_inherits
         WHERE inhparent = 'events'::regclass