diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 5901512504..ecd48bef59 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/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] diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 0e184e00d8..6ba1a6a59a 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/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", diff --git a/crates/buzz-db/src/store/media.rs b/crates/buzz-db/src/store/media.rs new file mode 100644 index 0000000000..8193dbd1f2 --- /dev/null +++ b/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> { + 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 { + 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 { + self.media_access(community, sha256, pubkey, true).await + } + + async fn media_access( + &self, + community: CommunityId, + sha256: &str, + pubkey: &[u8], + for_publication: bool, + ) -> Result { + 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::>(); + 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?) + } +} diff --git a/crates/buzz-db/src/store/media/tests.rs b/crates/buzz-db/src/store/media/tests.rs new file mode 100644 index 0000000000..12958dd912 --- /dev/null +++ b/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, + kind: u16, + content: &str, + tags: Vec, +) -> 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 = 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}" + ); + } +} diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 1fa1273eb0..faffb95440 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/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. diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs index 780532ec5d..ed074a5ae4 100644 --- a/crates/buzz-relay/src/api/media.rs +++ b/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 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 { + test_state_with_media(None).await + } + + pub(super) async fn test_state_with_media(endpoint: Option<&str>) -> Arc { 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()); diff --git a/crates/buzz-relay/src/api/media_access_tests.rs b/crates/buzz-relay/src/api/media_access_tests.rs new file mode 100644 index 0000000000..12be5a2615 --- /dev/null +++ b/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>>; + +struct ObjectServer { + endpoint: String, + requests: Arc, + 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)>, + Path(key): Path, + request: Request, + ) -> 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::() + .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 { + 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, 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) -> 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")); + } +} diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index ee1d0312be..738f9e7889 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/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) { diff --git a/crates/buzz-relay/src/handlers/media_access.rs b/crates/buzz-relay/src/handlers/media_access.rs new file mode 100644 index 0000000000..48cd2a55c5 --- /dev/null +++ b/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(()) +} diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 2f4aa00b59..3e594abfc6 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/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). diff --git a/migrations/0045_media_read_authorization.sql b/migrations/0045_media_read_authorization.sql new file mode 100644 index 0000000000..66f39ca1fe --- /dev/null +++ b/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); diff --git a/schema/schema.sql b/schema/schema.sql index 0950812562..d9a9080ce3 100644 --- a/schema/schema.sql +++ b/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(); diff --git a/scripts/reconcile-schema-after-pgschema.sql b/scripts/reconcile-schema-after-pgschema.sql index 7c3a7be871..25ac26ebf0 100644 --- a/scripts/reconcile-schema-after-pgschema.sql +++ b/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