diff --git a/Cargo.lock b/Cargo.lock index 5aa78909a..5c53881c9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -945,6 +945,7 @@ dependencies = [ "futures-core", "futures-util", "hex", + "http", "image", "imagesize", "infer", diff --git a/crates/buzz-media/Cargo.toml b/crates/buzz-media/Cargo.toml index 8f5d206d7..5aff3b66d 100644 --- a/crates/buzz-media/Cargo.toml +++ b/crates/buzz-media/Cargo.toml @@ -20,6 +20,7 @@ hex = { workspace = true } chrono = { workspace = true } axum = { workspace = true } s3 = { version = "0.37", package = "rust-s3", default-features = false, features = ["tokio-rustls-tls", "fail-on-err", "tags"] } +http = "1" infer = "0.19" image = { version = "0.25", default-features = false, features = ["jpeg", "png", "gif", "webp"] } blurhash = "0.2" diff --git a/crates/buzz-media/src/storage.rs b/crates/buzz-media/src/storage.rs index 743ebdcb0..b2661b1b9 100644 --- a/crates/buzz-media/src/storage.rs +++ b/crates/buzz-media/src/storage.rs @@ -77,6 +77,32 @@ impl MediaStorage { Ok(()) } + /// Store an object from a byte slice with `x-amz-meta-*` object metadata. + /// + /// `metadata` keys are bare names (e.g. `buzz-uploader-id`); the S3 client + /// adds the `x-amz-meta-` prefix. Used for media blobs that carry upload + /// attribution so out-of-band consumers can read it from a HEAD without + /// touching relay internals. + pub async fn put_with_metadata( + &self, + key: &str, + bytes: &[u8], + content_type: &str, + metadata: &[(&str, &str)], + ) -> Result<(), MediaError> { + let mut builder = self + .bucket + .put_object_builder(key, bytes) + .with_content_type(content_type); + for (k, v) in metadata { + builder = builder + .with_metadata(k, v) + .map_err(|e| MediaError::StorageError(e.to_string()))?; + } + builder.execute().await?; + Ok(()) + } + /// Stream a file from disk into S3 without loading it into RAM. /// /// Uses rust-s3's `put_object_stream_with_content_type` which reads from @@ -87,6 +113,24 @@ impl MediaStorage { key: &str, path: &Path, content_type: &str, + ) -> Result<(), MediaError> { + self.put_file_with_metadata(key, path, content_type, &[]) + .await + } + + /// Stream a file from disk into S3 with `x-amz-meta-*` object metadata. + /// + /// Metadata is attached via bucket-level `extra_headers` (not the stream + /// builder's `with_metadata`) because rust-s3's streaming multipart path + /// only forwards builder headers on the small-file (single PUT) branch; + /// bucket `extra_headers` are applied to `InitiateMultipartUpload` too, so + /// metadata survives files larger than the 8 MiB chunk threshold. + pub async fn put_file_with_metadata( + &self, + key: &str, + path: &Path, + content_type: &str, + metadata: &[(&str, &str)], ) -> Result<(), MediaError> { const BUF: usize = 8 * 1024 * 1024; // 8 MiB read buffer @@ -95,7 +139,19 @@ impl MediaStorage { .map_err(|e| MediaError::Io(e.to_string()))?; let mut reader = tokio::io::BufReader::with_capacity(BUF, file); - self.bucket + if metadata.is_empty() { + self.bucket + .put_object_stream_with_content_type(&mut reader, key, content_type) + .await?; + return Ok(()); + } + + let headers = build_amz_meta_headers(metadata)?; + let bucket = self + .bucket + .with_extra_headers(headers) + .map_err(|e| MediaError::StorageError(e.to_string()))?; + bucket .put_object_stream_with_content_type(&mut reader, key, content_type) .await?; Ok(()) @@ -336,6 +392,71 @@ mod tests { "video/mp4" ); } + + #[test] + fn amz_meta_headers_are_prefixed_and_validated() { + let headers = build_amz_meta_headers(&[ + ("buzz-uploader-id", "aabbcc"), + ("buzz-community-id", "0000-1111"), + ]) + .unwrap(); + assert_eq!( + headers.get("x-amz-meta-buzz-uploader-id").unwrap(), + "aabbcc" + ); + assert_eq!( + headers.get("x-amz-meta-buzz-community-id").unwrap(), + "0000-1111" + ); + + // Control characters in values are rejected, not silently mangled. + assert!(build_amz_meta_headers(&[("buzz-uploader-id", "bu\nzz")]).is_err()); + // Invalid header-name characters in the key are rejected. + assert!(build_amz_meta_headers(&[("bad key", "v")]).is_err()); + } + + /// Old sidecars (written before upload attribution) must still parse, and + /// new fields must round-trip. + #[test] + fn sidecar_attribution_fields_are_backward_compatible() { + // Pre-attribution sidecar JSON — no uploader_id/community_id keys. + let old = r#"{"dim":"800x600","blurhash":"","thumb_url":"","ext":"jpg","mime_type":"image/jpeg","size":123,"uploaded_at":1700000000}"#; + let meta: BlobMeta = serde_json::from_str(old).unwrap(); + assert_eq!(meta.uploader_id, None); + assert_eq!(meta.community_id, None); + + // Absent attribution is omitted from serialized output (not null). + let json = serde_json::to_value(&meta).unwrap(); + assert!(json.get("uploader_id").is_none()); + assert!(json.get("community_id").is_none()); + + // Populated attribution round-trips. + let meta = BlobMeta { + uploader_id: Some("aa".repeat(32)), + community_id: Some("6b8e1c2a-0000-0000-0000-000000000000".to_string()), + ..meta + }; + let round: BlobMeta = serde_json::from_str(&serde_json::to_string(&meta).unwrap()).unwrap(); + assert_eq!(round.uploader_id, meta.uploader_id); + assert_eq!(round.community_id, meta.community_id); + } +} + +/// Build an `x-amz-meta-*` [`http::HeaderMap`] from bare metadata key/value +/// pairs. Keys must be valid header-name characters; values must be valid +/// header values (S3 object metadata is US-ASCII). +fn build_amz_meta_headers(metadata: &[(&str, &str)]) -> Result { + let mut headers = http::HeaderMap::new(); + for (k, v) in metadata { + let name: http::HeaderName = format!("x-amz-meta-{k}") + .parse() + .map_err(|_| MediaError::StorageError(format!("invalid metadata key: {k}")))?; + let value: http::HeaderValue = v + .parse() + .map_err(|_| MediaError::StorageError(format!("invalid metadata value for {k}")))?; + headers.insert(name, value); + } + Ok(headers) } /// Metadata returned by HEAD — just enough for BUD-01 response headers. @@ -364,4 +485,14 @@ pub struct BlobMeta { /// Video duration in seconds. `None` for non-video blobs. #[serde(default, skip_serializing_if = "Option::is_none")] pub duration_secs: Option, + /// Authenticated uploader pubkey (hex). Upload attribution for + /// out-of-band consumers; `None` on sidecars written before attribution + /// existed. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub uploader_id: Option, + /// Host-resolved community id (UUID string). Mirrors the community segment + /// of the sidecar key so attribution survives even if the object is copied + /// out of its keyed location; `None` on pre-attribution sidecars. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub community_id: Option, } diff --git a/crates/buzz-media/src/upload.rs b/crates/buzz-media/src/upload.rs index cbf0760f0..26ae329ae 100644 --- a/crates/buzz-media/src/upload.rs +++ b/crates/buzz-media/src/upload.rs @@ -86,6 +86,17 @@ where // Compute uploaded_at once — single source of truth for sidecar and response. let uploaded_at = chrono::Utc::now().timestamp(); + // Upload attribution: the authenticated uploader pubkey and the + // host-resolved community, stamped as S3 object metadata so out-of-band + // consumers can attribute a blob from a HEAD + // alone. The community comes from the server-resolved `TenantContext`, + // never from client input. Note the blob is shared CAS across communities: + // if the same bytes are later uploaded under another tenant, the re-put + // overwrites this metadata with the most recent uploader — the + // community-scoped sidecar remains the authoritative per-tenant record. + let uploader_id = auth_event.pubkey.to_hex(); + let community_id = ctx.community().to_string(); + // Store blob first, then metadata. // On failure we intentionally do NOT delete the orphan blob — concurrent // uploads of the same hash could race and delete a blob that another @@ -93,7 +104,17 @@ where // content-addressed and bounded by the upload size limit, so the storage // cost is negligible. A V2 background GC job can sweep blobs with no // matching sidecar after a grace period. - storage.put(&key, &body, &mime).await?; + storage + .put_with_metadata( + &key, + &body, + &mime, + &[ + ("buzz-uploader-id", uploader_id.as_str()), + ("buzz-community-id", community_id.as_str()), + ], + ) + .await?; let meta_result = store_metadata(MetadataInput { sha256: sha256.clone(), @@ -101,6 +122,8 @@ where mime: mime.clone(), body: body.clone(), uploaded_at, + uploader_id, + community_id, }) .await; @@ -131,6 +154,10 @@ struct MetadataInput { mime: String, body: Bytes, uploaded_at: i64, + /// Authenticated uploader pubkey (hex), mirrored into the sidecar. + uploader_id: String, + /// Host-resolved community id, mirrored into the sidecar. + community_id: String, } /// Process an upload end-to-end: validate, store, thumbnail, return descriptor. @@ -195,6 +222,8 @@ pub async fn process_file_upload( mime_type: input.mime, uploaded_at: input.uploaded_at, duration_secs: None, + uploader_id: Some(input.uploader_id), + community_id: Some(input.community_id), }; storage.put_sidecar(ctx, &input.sha256, &meta).await?; Ok(meta) @@ -370,8 +399,23 @@ pub async fn process_video_upload( let uploaded_at = chrono::Utc::now().timestamp(); + // Upload attribution — see process_buffered_upload for the rationale and + // the shared-CAS re-put caveat. + let uploader_id = auth_event.pubkey.to_hex(); + let community_id = ctx.community().to_string(); + // --- 6. Stream blob from temp file to S3 --- - storage.put_file(&key, &tmp_path, &mime).await?; + storage + .put_file_with_metadata( + &key, + &tmp_path, + &mime, + &[ + ("buzz-uploader-id", uploader_id.as_str()), + ("buzz-community-id", community_id.as_str()), + ], + ) + .await?; drop(tmp); // Free temp file disk space immediately after S3 upload. // --- 7. Write sidecar (no thumbnail for video — desktop handles that) --- @@ -384,6 +428,8 @@ pub async fn process_video_upload( size: file_size, uploaded_at, duration_secs: Some(video_meta.duration_secs), + uploader_id: Some(uploader_id), + community_id: Some(community_id), }; storage.put_sidecar(ctx, &sha256_hex, &meta).await?; @@ -418,10 +464,30 @@ async fn generate_and_store_metadata( .map_err(|_| MediaError::Internal)??; meta.uploaded_at = input.uploaded_at; + meta.uploader_id = Some(input.uploader_id); + meta.community_id = Some(input.community_id); if let Some(ref tb) = thumb_bytes { + // The thumbnail is a derived object with its own S3 key, so it carries + // the same attribution metadata as the source blob. let thumb_key = format!("{}.thumb.jpg", input.sha256); - storage.put(&thumb_key, tb, "image/jpeg").await?; + storage + .put_with_metadata( + &thumb_key, + tb, + "image/jpeg", + &[ + ( + "buzz-uploader-id", + meta.uploader_id.as_deref().unwrap_or_default(), + ), + ( + "buzz-community-id", + meta.community_id.as_deref().unwrap_or_default(), + ), + ], + ) + .await?; } storage.put_sidecar(ctx, &input.sha256, &meta).await?; @@ -484,6 +550,8 @@ mod tests { size: 5_000_000, uploaded_at: 1700000000, duration_secs: Some(29.5), + uploader_id: None, + community_id: None, }; let desc = build_descriptor( @@ -542,6 +610,8 @@ mod tests { size: 100_000, uploaded_at: 1700000000, duration_secs: None, + uploader_id: None, + community_id: None, }; let desc = build_descriptor(