diff --git a/Cargo.lock b/Cargo.lock index afb474a2d..de7fa3aa9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -959,6 +959,7 @@ dependencies = [ "tokio", "tokio-util", "tracing", + "ulid", "uuid", ] @@ -8598,6 +8599,16 @@ dependencies = [ "windows-sys 0.60.2", ] +[[package]] +name = "ulid" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "470dbf6591da1b39d43c14523b2b469c86879a53e8b758c8e090a470fe7b1fbe" +dependencies = [ + "rand 0.9.4", + "web-time", +] + [[package]] name = "unarray" version = "0.1.4" diff --git a/crates/buzz-media/Cargo.toml b/crates/buzz-media/Cargo.toml index 8f5d206d7..aee5fdbdf 100644 --- a/crates/buzz-media/Cargo.toml +++ b/crates/buzz-media/Cargo.toml @@ -18,6 +18,7 @@ thiserror = { workspace = true } sha2 = { workspace = true } hex = { workspace = true } chrono = { workspace = true } +ulid = "1" axum = { workspace = true } s3 = { version = "0.37", package = "rust-s3", default-features = false, features = ["tokio-rustls-tls", "fail-on-err", "tags"] } infer = "0.19" diff --git a/crates/buzz-media/src/config.rs b/crates/buzz-media/src/config.rs index 5b6ca96df..047c08475 100644 --- a/crates/buzz-media/src/config.rs +++ b/crates/buzz-media/src/config.rs @@ -43,6 +43,22 @@ pub struct MediaConfig { pub max_file_bytes: u64, /// Public base URL for media URLs in BlobDescriptor (must include `/media` path). pub public_base_url: String, + /// Whether to write per-upload-event records under `_uploads/` + /// (moderation side channel). Off by default; set via + /// `BUZZ_MEDIA_UPLOAD_RECORDS=true`. + #[serde(default)] + pub upload_records_enabled: bool, + /// Trusted edge header to read the uploader's public IP from (e.g. + /// `cf-connecting-ip`). Unset (default) → no IP is read or recorded. + /// Only consulted when `upload_records_enabled` is true; the value is + /// validated as a public IP and dropped otherwise (fail-empty). + #[serde(default)] + pub upload_ip_header: Option, + /// Trusted edge header to read the uploader's source port from. Standard + /// edges don't emit one, so this is usually unset; a port is only + /// recorded alongside a valid IP. + #[serde(default)] + pub upload_port_header: Option, } impl MediaConfig { @@ -72,6 +88,100 @@ impl MediaConfig { if self.max_file_bytes == 0 { return Err("max_file_bytes must be > 0".to_string()); } + // Fail startup on incoherent collection config instead of silently + // recording nothing — an operator who set an IP header believes they + // are meeting a reporting obligation. + if self.upload_ip_header.is_some() && !self.upload_records_enabled { + return Err( + "BUZZ_MEDIA_UPLOAD_IP_HEADER is set but BUZZ_MEDIA_UPLOAD_RECORDS is not \ + enabled — the IP would never be recorded. Enable upload records or unset \ + the header." + .to_string(), + ); + } + if self.upload_port_header.is_some() && self.upload_ip_header.is_none() { + return Err( + "BUZZ_MEDIA_UPLOAD_PORT_HEADER is set without BUZZ_MEDIA_UPLOAD_IP_HEADER — \ + a port is only recorded alongside an IP. Set the IP header or unset the \ + port header." + .to_string(), + ); + } + for (name, value) in [ + ("BUZZ_MEDIA_UPLOAD_IP_HEADER", &self.upload_ip_header), + ("BUZZ_MEDIA_UPLOAD_PORT_HEADER", &self.upload_port_header), + ] { + if let Some(h) = value { + if axum::http::HeaderName::from_bytes(h.as_bytes()).is_err() { + return Err(format!("{name} is not a valid header name: {h:?}")); + } + } + } Ok(()) } } + +#[cfg(test)] +mod tests { + use super::MediaConfig; + + fn valid_config() -> MediaConfig { + MediaConfig { + s3_endpoint: "http://localhost:9000".to_string(), + s3_access_key: "k".to_string(), + s3_secret_key: "s".to_string(), + s3_bucket: "buzz-media".to_string(), + s3_region: "us-east-1".to_string(), + max_image_bytes: 1, + max_gif_bytes: 1, + max_video_bytes: 1, + max_file_bytes: 1, + public_base_url: "http://localhost:3000/media".to_string(), + upload_records_enabled: false, + upload_ip_header: None, + upload_port_header: None, + } + } + + #[test] + fn upload_record_knobs_default_off_and_validate() { + assert!(valid_config().validate().is_ok()); + + let mut on = valid_config(); + on.upload_records_enabled = true; + assert!(on.validate().is_ok()); + + on.upload_ip_header = Some("cf-connecting-ip".to_string()); + assert!(on.validate().is_ok()); + + on.upload_port_header = Some("x-client-port".to_string()); + assert!(on.validate().is_ok()); + } + + #[test] + fn ip_header_without_records_fails_startup() { + // An operator who set the header believes IPs are being recorded — + // fail loudly instead of silently collecting nothing. + let mut cfg = valid_config(); + cfg.upload_ip_header = Some("cf-connecting-ip".to_string()); + assert!(cfg.validate().is_err()); + } + + #[test] + fn port_header_without_ip_header_fails_startup() { + let mut cfg = valid_config(); + cfg.upload_records_enabled = true; + cfg.upload_port_header = Some("x-client-port".to_string()); + assert!(cfg.validate().is_err()); + } + + #[test] + fn malformed_header_names_fail_startup() { + let mut cfg = valid_config(); + cfg.upload_records_enabled = true; + for bad in ["with space", "colon:name", "bad/header", "bad,header", ""] { + cfg.upload_ip_header = Some(bad.to_string()); + assert!(cfg.validate().is_err(), "should reject header name {bad:?}"); + } + } +} diff --git a/crates/buzz-media/src/lib.rs b/crates/buzz-media/src/lib.rs index f382c21f8..e62972d58 100644 --- a/crates/buzz-media/src/lib.rs +++ b/crates/buzz-media/src/lib.rs @@ -9,6 +9,7 @@ pub mod storage; pub mod thumbnail; pub mod types; pub mod upload; +pub mod upload_record; pub mod validation; pub use config::MediaConfig; @@ -16,4 +17,8 @@ pub use error::MediaError; pub use storage::{BlobHeadMeta, BlobMeta, ByteStream, MediaStorage}; pub use types::BlobDescriptor; pub use upload::{process_file_upload, process_upload, process_video_upload}; +pub use upload_record::{ + parse_port, parse_public_ip, upload_record_key, UploadAttribution, UploadNetworkInfo, + UploadRecord, UPLOAD_RECORD_VERSION, +}; pub use validation::{serve_inline, validate_video_file, VideoMeta}; diff --git a/crates/buzz-media/src/storage.rs b/crates/buzz-media/src/storage.rs index 743ebdcb0..a469826df 100644 --- a/crates/buzz-media/src/storage.rs +++ b/crates/buzz-media/src/storage.rs @@ -256,6 +256,9 @@ mod tests { max_video_bytes: 524_288_000, max_file_bytes: 104_857_600, public_base_url: "http://localhost:3000/media".to_string(), + upload_records_enabled: false, + upload_ip_header: None, + upload_port_header: None, } } diff --git a/crates/buzz-media/src/upload.rs b/crates/buzz-media/src/upload.rs index cbf0760f0..32f8d09ba 100644 --- a/crates/buzz-media/src/upload.rs +++ b/crates/buzz-media/src/upload.rs @@ -11,6 +11,7 @@ use crate::error::MediaError; use crate::storage::{BlobMeta, MediaStorage}; use crate::thumbnail::generate_image_metadata_sync; use crate::types::BlobDescriptor; +use crate::upload_record::{record_upload_event, UploadAttribution, UploadEventFacts}; use crate::validation::{ mime_to_ext, validate_content, validate_file_content, validate_video_file, }; @@ -21,30 +22,53 @@ use crate::validation::{ /// - `validate`: a CPU-bound check (run inside `spawn_blocking`) that returns /// the `(mime, ext)` pair for the body. Images derive `ext` from the MIME; /// generic files get both from the deny-list validator. -/// - `store_metadata`: stores the sidecar (and any derived artifacts such as a -/// thumbnail) and returns the resulting [`BlobMeta`]. Images run the full -/// image-metadata pipeline; generic files write a minimal sidecar. It -/// receives the already-computed `(sha256, ext, mime, uploaded_at)` so no -/// work is repeated. +/// - `prepare_metadata`: builds metadata and stores any derived artifacts such +/// as a thumbnail, but deliberately does not write the sidecar. The sidecar +/// is the media serve gate and is published only after the moderation record +/// succeeds. It receives the already-computed +/// `(sha256, ext, mime, uploaded_at)` so no work is repeated. /// /// Everything else — hash, Blossom auth (10-minute window), content-addressed /// key, the both-exist idempotency short-circuit, blob store, orphan-blob /// handling, and descriptor build — is common. The streaming video path stays /// separate (see [`process_video_upload`]) because it never buffers in RAM. -async fn process_buffered_upload( - storage: &MediaStorage, - config: &MediaConfig, - ctx: &TenantContext, - auth_event: &nostr::Event, +/// +/// `attribution` is `Some` when per-event upload records are enabled +/// (`BUZZ_MEDIA_UPLOAD_RECORDS`): a record is then written for **every** +/// accepted upload — including the idempotent short-circuit, which does no +/// blob PUT and would otherwise be invisible to the moderation pipeline. +/// For fresh uploads, the record is written after the blob and derived +/// artifacts but before the sidecar. This preserves both contracts: record +/// existence implies referenced objects are readable, while a record failure +/// cannot publish media without triggering moderation. +struct BufferedUploadInput<'a> { + storage: &'a MediaStorage, + config: &'a MediaConfig, + ctx: &'a TenantContext, + auth_event: &'a nostr::Event, body: Bytes, + attribution: Option, +} + +async fn process_buffered_upload( + input: BufferedUploadInput<'_>, validate: V, - store_metadata: M, + prepare_metadata: M, ) -> Result where V: FnOnce(&Bytes, &MediaConfig) -> Result<(String, String), MediaError> + Send + 'static, M: FnOnce(MetadataInput) -> Fut, Fut: std::future::Future>, { + let BufferedUploadInput { + storage, + config, + ctx, + auth_event, + body, + attribution, + } = input; + // CPU-bound: validate content, compute hash, verify auth. let auth = auth_event.clone(); let bytes = body.clone(); @@ -72,6 +96,26 @@ where let blob_exists = storage.head(&key).await?; if sidecar_exists && blob_exists { let meta = storage.get_sidecar(ctx, &sha256).await?; + // A re-upload of known bytes is still a distinct upload *event*: no + // blob PUT happens, so without this record the uploader would be + // invisible to the moderation pipeline (and takedown re-uploads + // would go unscanned). + if let Some(attribution) = &attribution { + record_upload_event( + storage, + ctx, + &auth_event.pubkey, + attribution, + UploadEventFacts { + sha256: &sha256, + ext: &ext, + mime: &mime, + size: body.len() as u64, + uploaded_at: chrono::Utc::now().timestamp(), + }, + ) + .await?; + } return Ok(build_descriptor( config, &sha256, @@ -95,30 +139,52 @@ where // matching sidecar after a grace period. storage.put(&key, &body, &mime).await?; - let meta_result = store_metadata(MetadataInput { + let meta = match prepare_metadata(MetadataInput { sha256: sha256.clone(), ext: ext.clone(), mime: mime.clone(), body: body.clone(), uploaded_at, }) - .await; - - match meta_result { - Ok(meta) => Ok(build_descriptor( - config, - &sha256, - &ext, - &mime, - body.len() as u64, - Some(&meta), - uploaded_at, - )), + .await + { + Ok(meta) => meta, Err(e) => { tracing::warn!(sha256 = %sha256, error = %e, "metadata generation failed; orphan blob left for GC"); - Err(e) + return Err(e); } + }; + + // The moderation record precedes the sidecar publish gate. If this write + // fails, the blob and any thumbnail remain orphaned but the media cannot be + // served. Conversely, record existence still implies those objects exist. + if let Some(attribution) = &attribution { + record_upload_event( + storage, + ctx, + &auth_event.pubkey, + attribution, + UploadEventFacts { + sha256: &sha256, + ext: &ext, + mime: &mime, + size: body.len() as u64, + uploaded_at, + }, + ) + .await?; } + storage.put_sidecar(ctx, &sha256, &meta).await?; + + Ok(build_descriptor( + config, + &sha256, + &ext, + &mime, + body.len() as u64, + Some(&meta), + uploaded_at, + )) } /// Inputs handed to a buffered-upload metadata builder, after the shared @@ -143,19 +209,23 @@ pub async fn process_upload( ctx: &TenantContext, auth_event: &nostr::Event, body: Bytes, + attribution: Option, ) -> Result { process_buffered_upload( - storage, - config, - ctx, - auth_event, - body, + BufferedUploadInput { + storage, + config, + ctx, + auth_event, + body, + attribution, + }, |bytes, cfg| { let mime = validate_content(bytes, cfg)?; let ext = mime_to_ext(&mime).to_string(); Ok((mime, ext)) }, - |input| async move { generate_and_store_metadata(storage, config, ctx, input).await }, + |input| async move { prepare_image_metadata(storage, config, input).await }, ) .await } @@ -176,13 +246,17 @@ pub async fn process_file_upload( ctx: &TenantContext, auth_event: &nostr::Event, body: Bytes, + attribution: Option, ) -> Result { process_buffered_upload( - storage, - config, - ctx, - auth_event, - body, + BufferedUploadInput { + storage, + config, + ctx, + auth_event, + body, + attribution, + }, |bytes, cfg| validate_file_content(bytes, cfg), |input| async move { // Minimal sidecar — no thumbnail/dim/blurhash/duration for generic files. @@ -196,7 +270,6 @@ pub async fn process_file_upload( uploaded_at: input.uploaded_at, duration_secs: None, }; - storage.put_sidecar(ctx, &input.sha256, &meta).await?; Ok(meta) }, ) @@ -221,6 +294,7 @@ pub async fn process_video_upload( auth_event: &nostr::Event, body_stream: impl futures_core::Stream> + Send + 'static, content_length: Option, + attribution: Option, ) -> Result { // --- 1. Stream body to temp file, compute SHA-256 incrementally --- let tmp = tempfile::NamedTempFile::new().map_err(|e| MediaError::Io(e.to_string()))?; @@ -357,6 +431,24 @@ pub async fn process_video_upload( let blob_exists = storage.head(&key).await?; if sidecar_exists && blob_exists { let meta = storage.get_sidecar(ctx, &sha256_hex).await?; + // Re-upload of known bytes: still a distinct upload event — see the + // buffered path's short-circuit for the rationale. + if let Some(attribution) = &attribution { + record_upload_event( + storage, + ctx, + &auth_event.pubkey, + attribution, + UploadEventFacts { + sha256: &sha256_hex, + ext, + mime: &mime, + size: file_size, + uploaded_at: chrono::Utc::now().timestamp(), + }, + ) + .await?; + } return Ok(build_descriptor( config, &sha256_hex, @@ -374,7 +466,7 @@ pub async fn process_video_upload( storage.put_file(&key, &tmp_path, &mime).await?; drop(tmp); // Free temp file disk space immediately after S3 upload. - // --- 7. Write sidecar (no thumbnail for video — desktop handles that) --- + // --- 7. Build metadata (no thumbnail for video — desktop handles that) --- let meta = BlobMeta { dim: format!("{}x{}", video_meta.width, video_meta.height), blurhash: String::new(), @@ -385,6 +477,24 @@ pub async fn process_video_upload( uploaded_at, duration_secs: Some(video_meta.duration_secs), }; + + // Record before publishing the sidecar serve gate. See the buffered path. + if let Some(attribution) = &attribution { + record_upload_event( + storage, + ctx, + &auth_event.pubkey, + attribution, + UploadEventFacts { + sha256: &sha256_hex, + ext, + mime: &mime, + size: file_size, + uploaded_at, + }, + ) + .await?; + } storage.put_sidecar(ctx, &sha256_hex, &meta).await?; Ok(build_descriptor( @@ -398,12 +508,11 @@ pub async fn process_video_upload( )) } -/// Generate thumbnail, blurhash, and sidecar metadata, then store them. +/// Generate thumbnail and metadata without publishing the sidecar serve gate. /// Returns the completed [`BlobMeta`] on success. -async fn generate_and_store_metadata( +async fn prepare_image_metadata( storage: &MediaStorage, config: &MediaConfig, - ctx: &TenantContext, input: MetadataInput, ) -> Result { let body_ref = input.body.clone(); @@ -424,7 +533,6 @@ async fn generate_and_store_metadata( storage.put(&thumb_key, tb, "image/jpeg").await?; } - storage.put_sidecar(ctx, &input.sha256, &meta).await?; Ok(meta) } @@ -467,6 +575,9 @@ mod tests { max_video_bytes: 524_288_000, max_file_bytes: 104_857_600, public_base_url: "https://media.example.com".to_string(), + upload_records_enabled: false, + upload_ip_header: None, + upload_port_header: None, } } diff --git a/crates/buzz-media/src/upload_record.rs b/crates/buzz-media/src/upload_record.rs new file mode 100644 index 000000000..42f57fbd2 --- /dev/null +++ b/crates/buzz-media/src/upload_record.rs @@ -0,0 +1,419 @@ +//! Per-upload-event records — the moderation side channel. +//! +//! Content-addressed storage keys facts about *bytes*; moderation and legal +//! reporting (e.g. NCMEC CyberTipline) need facts about *upload events*: who +//! uploaded, when, from which network address. This module writes one small +//! append-only JSON record per **accepted** upload — including the idempotent +//! re-upload short-circuit, which does no blob PUT and is therefore invisible +//! to any blob-creation-driven pipeline. +//! +//! Key layout (alongside the existing `_meta/` sidecar convention): +//! +//! ```text +//! _uploads/{community}/{sha256}/{event_id}.json +//! ``` +//! +//! `event_id` is a ULID — unique and time-sortable, one record per accepted +//! upload event. Records are unreachable through the media serve path by +//! construction (`validate_media_path` requires a bare 64-hex first segment), +//! and the bucket is only accessible via the relay's IAM role. +//! +//! The whole feature is **off by default** and gated behind +//! `BUZZ_MEDIA_UPLOAD_RECORDS`. IP collection is a second, independent opt-in +//! (`BUZZ_MEDIA_UPLOAD_IP_HEADER`) and is *fail-empty*: a missing, malformed, +//! or non-public address records nothing — a wrong IP is worse than no IP, +//! so absent is always preferable. The IP goes only into this +//! record — never blob metadata, never the upload response, never the +//! hash-chained audit log. +//! +//! ## Consumer contract (buzz-moderation) +//! +//! The moderation pipeline triggers on `ObjectCreated` events under the +//! `_uploads/` prefix and parses this record instead of HEADing blobs: +//! +//! - For fresh uploads, the record is written after the blob and derived +//! artifacts but before the sidecar serve gate. Record existence therefore +//! implies the scan inputs are readable, while record failure cannot leave +//! unscanned media publicly servable. +//! - `ext`, `mime_type`, and `size` are always present so the consumer can +//! derive the blob key (`{sha256}.{ext}`) and scan eligibility without +//! extra round-trips. +//! - `uploader_name`, `ip`, and `port` are omitted (never `null`) when +//! unknown or when collection is disabled. +//! - Consumers must tolerate unknown fields; `version` bumps only on +//! breaking changes to existing fields. + +use std::net::IpAddr; + +use buzz_core::tenant::TenantContext; +use serde::{Deserialize, Serialize}; + +/// Current record schema version. Additive fields do not bump this. +pub const UPLOAD_RECORD_VERSION: u32 = 1; + +/// One accepted upload event. See module docs for the consumer contract. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct UploadRecord { + /// Schema version ([`UPLOAD_RECORD_VERSION`]). + pub version: u32, + /// ULID — unique per accepted upload, time-sortable. Also the key suffix. + pub event_id: String, + /// Content hash of the uploaded bytes (64 lowercase hex chars). + pub sha256: String, + /// Canonical extension — consumers derive the blob key `{sha256}.{ext}`. + pub ext: String, + /// Sniffed MIME type of the uploaded bytes. + pub mime_type: String, + /// Size of the uploaded bytes. + pub size: u64, + /// Unix seconds when the relay accepted *this* upload event. On an + /// idempotent re-upload this is the re-upload time, not the original + /// blob's `uploaded_at`. + pub uploaded_at: i64, + /// Server-resolved community id (UUID). Never client-supplied. + pub community_id: String, + /// Server-resolved tenant host the upload was bound to. + pub community_host: String, + /// Authenticated uploader pubkey (64 lowercase hex chars). + pub uploader_id: String, + /// Same pubkey, bech32 `npub` encoding. + pub uploader_npub: String, + /// Uploader's configured display name at upload time. Best-effort label; + /// `uploader_id` is authoritative. Omitted when unset. + #[serde(skip_serializing_if = "Option::is_none")] + pub uploader_name: Option, + /// Uploader's public IP as reported by the configured edge header. + /// Present only when `BUZZ_MEDIA_UPLOAD_IP_HEADER` is set AND the header + /// held a valid public address (fail-empty). Omitted, never `null`. + #[serde(skip_serializing_if = "Option::is_none")] + pub ip: Option, + /// Uploader's source port as reported by the configured edge header. + /// Standard edge headers don't carry the client port, so this is + /// best-effort and usually absent. Only recorded alongside `ip`. + #[serde(skip_serializing_if = "Option::is_none")] + pub port: Option, +} + +/// Network address facts extracted from trusted edge headers by the HTTP +/// handler, already validated (fail-empty). `Default` is "nothing collected". +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct UploadNetworkInfo { + /// Validated public IP of the uploader, or `None`. + pub ip: Option, + /// Source port of the uploader, or `None`. Ignored when `ip` is `None`. + pub port: Option, +} + +/// Per-event upload attribution passed into the upload pipeline by the +/// handler. Only consulted when upload records are enabled. +#[derive(Debug, Clone, Default)] +pub struct UploadAttribution { + /// Uploader's configured display name, if known (sanitized, bounded). + pub uploader_name: Option, + /// Validated network facts from trusted edge headers. + pub net: UploadNetworkInfo, +} + +/// Facts about the accepted upload, computed by the upload pipeline. +#[derive(Debug, Clone, Copy)] +pub struct UploadEventFacts<'a> { + /// Content hash (64 lowercase hex chars). + pub sha256: &'a str, + /// Canonical extension. + pub ext: &'a str, + /// Sniffed MIME type. + pub mime: &'a str, + /// Uploaded byte size. + pub size: u64, + /// Unix seconds this upload event was accepted. + pub uploaded_at: i64, +} + +/// Build and store the per-event record for one accepted upload. +/// +/// Called after blob and derived-artifact durability but before the sidecar +/// publish gate on fresh uploads; called on the existing published state for +/// idempotent re-uploads. A write failure propagates and fails the upload. The +/// record's `ObjectCreated` event is the moderation pipeline's only scan +/// trigger, so no newly published media may exist without a record. +pub async fn record_upload_event( + storage: &crate::storage::MediaStorage, + ctx: &TenantContext, + uploader: &nostr::PublicKey, + attribution: &UploadAttribution, + facts: UploadEventFacts<'_>, +) -> Result<(), crate::error::MediaError> { + use nostr::ToBech32; + + let event_id = ulid::Ulid::new().to_string(); + // Ports are only meaningful next to the address they were observed with. + let ip = attribution.net.ip; + let port = ip.and(attribution.net.port); + let record = UploadRecord { + version: UPLOAD_RECORD_VERSION, + event_id: event_id.clone(), + sha256: facts.sha256.to_string(), + ext: facts.ext.to_string(), + mime_type: facts.mime.to_string(), + size: facts.size, + uploaded_at: facts.uploaded_at, + community_id: ctx.community().to_string(), + community_host: ctx.host().to_string(), + uploader_id: uploader.to_hex(), + uploader_npub: uploader.to_bech32().map_err(|e| { + // Unreachable for a valid pubkey; surfaced rather than unwrapped. + crate::error::MediaError::StorageError(format!("npub encoding failed: {e}")) + })?, + uploader_name: attribution.uploader_name.clone(), + ip: ip.map(|addr| addr.to_string()), + port, + }; + let key = upload_record_key(ctx, facts.sha256, &event_id); + let json = serde_json::to_vec(&record)?; + storage.put(&key, &json, "application/json").await +} + +/// Build the per-event record key: +/// `_uploads/{community}/{sha256}/{event_id}.json`. +/// +/// `ctx` must be the server-resolved request tenant — same fence as the +/// `_meta/` sidecar key (see [`crate::storage::MediaStorage::sidecar_key`]). +pub fn upload_record_key(ctx: &TenantContext, sha256: &str, event_id: &str) -> String { + format!("_uploads/{}/{sha256}/{event_id}.json", ctx.community()) +} + +/// Parse an IP header value, accepting only public addresses (fail-empty). +/// +/// Returns `None` — record nothing — for anything that is not a single, +/// syntactically valid, public IP: garbage, comma lists, private ranges, +/// loopback, link-local, CGNAT, multicast, documentation, ULA, etc. Never +/// guesses and never falls back to the socket address. +pub fn parse_public_ip(raw: &str) -> Option { + let ip: IpAddr = raw.trim().parse().ok()?; + is_public_ip(&ip).then_some(ip) +} + +/// Parse a port header value: a single decimal u16, non-zero. +pub fn parse_port(raw: &str) -> Option { + raw.trim().parse::().ok().filter(|&p| p != 0) +} + +/// Conservative "is this a public, routable address" check. +/// +/// `IpAddr::is_global` is unstable, so this enumerates the reserved ranges +/// explicitly. Anything not recognizably public is rejected — the cost of a +/// false negative is an absent field; the cost of a false positive is a wrong +/// address in a federal report. +fn is_public_ip(ip: &IpAddr) -> bool { + match ip { + IpAddr::V4(v4) => { + let octets = v4.octets(); + !(v4.is_private() + || v4.is_loopback() + || v4.is_link_local() + || v4.is_broadcast() + || v4.is_documentation() + || v4.is_multicast() + || v4.is_unspecified() + // This network 0.0.0.0/8 (RFC 1122) + || octets[0] == 0 + // CGNAT 100.64.0.0/10 (RFC 6598) + || (octets[0] == 100 && (octets[1] & 0b1100_0000) == 64) + // Reserved 240.0.0.0/4 (RFC 1112) — is_broadcast covers .255 only + || octets[0] >= 240 + // Benchmarking 198.18.0.0/15 (RFC 2544) + || (octets[0] == 198 && (octets[1] & 0xFE) == 18) + // IETF protocol assignments 192.0.0.0/24 (RFC 6890) + || (octets[0] == 192 && octets[1] == 0 && octets[2] == 0)) + } + IpAddr::V6(v6) => { + let seg = v6.segments(); + !(v6.is_loopback() + || v6.is_multicast() + || v6.is_unspecified() + // Unique local fc00::/7 (RFC 4193) + || (seg[0] & 0xFE00) == 0xFC00 + // Link-local fe80::/10 (RFC 4291) + || (seg[0] & 0xFFC0) == 0xFE80 + // Discard-only 100::/64 (RFC 6666) + || (seg[0] == 0x0100 && seg[1..4] == [0, 0, 0]) + // Teredo 2001::/32 (RFC 4380) + || (seg[0] == 0x2001 && seg[1] == 0) + // Benchmarking 2001:2::/48 (RFC 5180) + || (seg[0] == 0x2001 && seg[1] == 2 && seg[2] == 0) + // Documentation 2001:db8::/32 (RFC 3849) + || (seg[0] == 0x2001 && seg[1] == 0x0DB8) + // 6to4 2002::/16 (RFC 3056) + || seg[0] == 0x2002 + // Documentation 3fff::/20 (RFC 9637) + || (seg[0] & 0xFFF0) == 0x3FF0 + // IPv4-mapped ::ffff:0:0/96 — the embedded v4 was already + // rejected above if it arrived as dotted quad; reject the + // mapped form outright rather than re-deriving it. + || v6.to_ipv4_mapped().is_some()) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use buzz_core::tenant::CommunityId; + + fn tenant() -> TenantContext { + TenantContext::resolved( + CommunityId::from_uuid(uuid::Uuid::from_u128(7)), + "chat.example.com", + ) + } + + #[test] + fn record_key_matches_spec_layout() { + let ctx = tenant(); + let sha = "a".repeat(64); + let key = upload_record_key(&ctx, &sha, "01J9W3ULIDULIDULIDULIDULID"); + assert_eq!( + key, + format!( + "_uploads/{}/{sha}/01J9W3ULIDULIDULIDULIDULID.json", + ctx.community() + ) + ); + } + + #[test] + fn record_serializes_full_shape() { + let record = UploadRecord { + version: UPLOAD_RECORD_VERSION, + event_id: "01J9W3TEST".into(), + sha256: "b".repeat(64), + ext: "png".into(), + mime_type: "image/png".into(), + size: 12345, + uploaded_at: 1_783_358_352, + community_id: uuid::Uuid::from_u128(7).to_string(), + community_host: "chat.example.com".into(), + uploader_id: "c".repeat(64), + uploader_npub: "npub1example".into(), + uploader_name: Some("alice".into()), + ip: Some("203.0.113.7".into()), + port: Some(51234), + }; + let json = serde_json::to_value(&record).unwrap(); + assert_eq!(json["version"], 1); + assert_eq!(json["ext"], "png"); + assert_eq!(json["mime_type"], "image/png"); + assert_eq!(json["size"], 12345); + assert_eq!(json["ip"], "203.0.113.7"); + assert_eq!(json["port"], 51234); + assert_eq!(json["uploader_name"], "alice"); + } + + #[test] + fn record_omits_absent_optionals_entirely() { + let record = UploadRecord { + version: UPLOAD_RECORD_VERSION, + event_id: "01J9W3TEST".into(), + sha256: "b".repeat(64), + ext: "mp4".into(), + mime_type: "video/mp4".into(), + size: 1, + uploaded_at: 0, + community_id: "cid".into(), + community_host: "h".into(), + uploader_id: "c".repeat(64), + uploader_npub: "npub1example".into(), + uploader_name: None, + ip: None, + port: None, + }; + let json = serde_json::to_value(&record).unwrap(); + // Omitted, not null — the consumer contract. + assert!(json.get("uploader_name").is_none()); + assert!(json.get("ip").is_none()); + assert!(json.get("port").is_none()); + } + + #[test] + fn record_deserialization_tolerates_unknown_fields() { + // Forward-compat: additive relay fields must not break older parsers + // of the same version (mirrors the consumer's requirement). + let json = r#"{ + "version": 1, "event_id": "01J", "sha256": "ab", "ext": "png", + "mime_type": "image/png", "size": 1, "uploaded_at": 2, + "community_id": "c", "community_host": "h", + "uploader_id": "u", "uploader_npub": "n", + "some_future_field": {"nested": true} + }"#; + let record: UploadRecord = serde_json::from_str(json).unwrap(); + assert_eq!(record.version, 1); + assert_eq!(record.ip, None); + } + + #[test] + fn public_ips_accepted() { + for raw in [ + "8.8.8.8", + "1.1.1.1", + " 93.184.216.34 ", // trims whitespace + "2600:1f18::1", + "2a00:1450:4009:81f::200e", + ] { + assert!(parse_public_ip(raw).is_some(), "should accept {raw}"); + } + } + + #[test] + fn non_public_ips_fail_empty() { + for raw in [ + "", + "not-an-ip", + "10.0.0.1", // private + "172.16.0.1", // private + "192.168.1.1", // private + "127.0.0.1", // loopback + "169.254.1.1", // link-local + "100.64.0.1", // CGNAT + "100.127.255.255", // CGNAT upper edge + "0.0.0.0", // unspecified + "0.1.2.3", // this network 0.0.0.0/8 + "255.255.255.255", // broadcast + "224.0.0.1", // multicast + "240.0.0.1", // reserved + "198.18.0.1", // benchmarking + "192.0.0.1", // IETF assignments + "203.0.113.7", // documentation (TEST-NET-3) + "198.51.100.1", // documentation (TEST-NET-2) + "192.0.2.1", // documentation (TEST-NET-1) + "::1", // v6 loopback + "::", // v6 unspecified + "fe80::1", // v6 link-local + "fc00::1", // v6 ULA + "fd12:3456::1", // v6 ULA + "ff02::1", // v6 multicast + "100::1", // v6 discard-only + "2001::1", // Teredo + "2001:2::1", // v6 benchmarking + "2001:db8::1", // v6 documentation + "2002::1", // 6to4 + "3fff::1", // v6 documentation + "::ffff:8.8.8.8", // v4-mapped — reject the mapped form + "8.8.8.8, 1.1.1.1", // comma list — not a single IP + "8.8.8.8:443", // ip:port — not a bare IP + ] { + assert!(parse_public_ip(raw).is_none(), "should reject {raw:?}"); + } + } + + #[test] + fn port_parses_single_nonzero_u16() { + assert_eq!(parse_port("51234"), Some(51234)); + assert_eq!(parse_port(" 443 "), Some(443)); + assert_eq!(parse_port("0"), None); + assert_eq!(parse_port("65536"), None); + assert_eq!(parse_port("-1"), None); + assert_eq!(parse_port("443, 444"), None); + assert_eq!(parse_port("abc"), None); + assert_eq!(parse_port(""), None); + } +} diff --git a/crates/buzz-media/src/validation.rs b/crates/buzz-media/src/validation.rs index 328516bb1..74b4ab0d9 100644 --- a/crates/buzz-media/src/validation.rs +++ b/crates/buzz-media/src/validation.rs @@ -422,6 +422,9 @@ mod tests { max_video_bytes: 524_288_000, max_file_bytes: 104_857_600, public_base_url: String::new(), + upload_records_enabled: false, + upload_ip_header: None, + upload_port_header: None, } } diff --git a/crates/buzz-media/tests/static_creds_minio.rs b/crates/buzz-media/tests/static_creds_minio.rs index 38dc6618c..d7591238c 100644 --- a/crates/buzz-media/tests/static_creds_minio.rs +++ b/crates/buzz-media/tests/static_creds_minio.rs @@ -35,6 +35,9 @@ fn minio_config() -> MediaConfig { max_video_bytes: 524_288_000, max_file_bytes: 104_857_600, public_base_url: "http://localhost:3000/media".to_string(), + upload_records_enabled: false, + upload_ip_header: None, + upload_port_header: None, } } diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs index 28074611b..c5498cc5e 100644 --- a/crates/buzz-relay/src/api/media.rs +++ b/crates/buzz-relay/src/api/media.rs @@ -18,7 +18,7 @@ use axum::{ use base64::Engine; use buzz_audit::{AuditAction, NewAuditEntry}; use buzz_core::tenant::TenantContext; -use buzz_media::{BlobDescriptor, MediaError}; +use buzz_media::{BlobDescriptor, MediaError, UploadAttribution, UploadNetworkInfo}; use crate::state::AppState; @@ -209,6 +209,52 @@ impl FromRequestParts> for AuthenticatedUpload { } } +/// Build per-event upload attribution when upload records are enabled +/// (`BUZZ_MEDIA_UPLOAD_RECORDS`). Returns `None` when the feature is off — +/// the upload pipeline then writes no `_uploads/` record at all. +/// +/// - `uploader_name` is the uploader's current display name in the bound +/// community (best-effort label; lookup failure degrades to absent). +/// - `net.ip` is read from the operator-configured trusted edge header +/// (`BUZZ_MEDIA_UPLOAD_IP_HEADER`) and validated as a public IP — +/// fail-empty: missing/malformed/non-public values record nothing. The +/// socket address is never used; behind a sidecar it is meaningless, and a +/// wrong address is worse than none. +/// - `net.port` (optional companion header) is only kept alongside a valid IP. +async fn upload_attribution( + state: &AppState, + auth: &AuthenticatedUpload, + headers: &HeaderMap, +) -> Option { + let cfg = &state.config.media; + if !cfg.upload_records_enabled { + return None; + } + + let uploader_name = state + .db + .get_user(auth.tenant.community(), &auth.auth_event.pubkey.to_bytes()) + .await + .ok() + .flatten() + .and_then(|profile| profile.display_name) + .map(|name| name.trim().to_string()) + .filter(|name| !name.is_empty()); + + let header_value = |name: &Option| { + name.as_deref() + .and_then(|h| headers.get(h)) + .and_then(|v| v.to_str().ok()) + }; + let ip = header_value(&cfg.upload_ip_header).and_then(buzz_media::parse_public_ip); + let port = ip.and(header_value(&cfg.upload_port_header).and_then(buzz_media::parse_port)); + + Some(UploadAttribution { + uploader_name, + net: UploadNetworkInfo { ip, port }, + }) +} + /// PUT /media/upload — Blossom BUD-02 upload. /// /// Auth is validated via the [`AuthenticatedUpload`] extractor BEFORE the body @@ -238,6 +284,8 @@ pub async fn upload_blob( .and_then(|v| v.to_str().ok()) .unwrap_or(""); + let attribution = upload_attribution(&state, &auth, &headers).await; + let mut descriptor = if content_type.starts_with("video/") { // Video path: stream body directly to disk — never fully buffered in RAM. let content_length = headers @@ -251,6 +299,7 @@ pub async fn upload_blob( &auth.auth_event, body.into_data_stream(), content_length, + attribution, ) .await? } else { @@ -280,6 +329,7 @@ pub async fn upload_blob( &auth.tenant, &auth.auth_event, bytes, + attribution, ) .await? } else { @@ -289,6 +339,7 @@ pub async fn upload_blob( &auth.tenant, &auth.auth_event, bytes, + attribution, ) .await? } @@ -914,6 +965,20 @@ mod tests { assert!(validate_media_path("").is_err()); } + #[test] + fn test_validate_media_path_rejects_upload_record_keys() { + // The `_uploads/` per-event records (and `_meta/` sidecars) must be + // unreachable through the serve path. Axum's single path segment + // can't even contain `/`, but assert the validator rejects these + // shapes outright so the property survives any routing change. + assert!(validate_media_path("_uploads").is_err()); + assert!(validate_media_path(&format!("_uploads/c/{VALID_HASH}/01J.json")).is_err()); + assert!(validate_media_path("_meta").is_err()); + assert!(validate_media_path(&format!("_meta/c/{VALID_HASH}.json")).is_err()); + // Suffix-style metadata keys are also non-servable (>3 segments / bad ext). + assert!(validate_media_path(&format!("{VALID_HASH}.png.metadata")).is_err()); + } + #[test] fn media_base_url_for_tenant_uses_tenant_host_and_http_scheme() { assert_eq!( diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 7e7df2019..f9f4d0805 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -363,6 +363,20 @@ impl Config { .unwrap_or(100 * 1024 * 1024), public_base_url: std::env::var("BUZZ_MEDIA_BASE_URL") .unwrap_or_else(|_| "http://localhost:3000/media".to_string()), + // Per-upload-event records (`_uploads/` moderation side channel). + // Off by default; coherence between the three knobs is enforced in + // MediaConfig::validate at startup. + upload_records_enabled: std::env::var("BUZZ_MEDIA_UPLOAD_RECORDS") + .map(|v| v == "true" || v == "1") + .unwrap_or(false), + upload_ip_header: std::env::var("BUZZ_MEDIA_UPLOAD_IP_HEADER") + .ok() + .map(|s| s.trim().to_lowercase()) + .filter(|s| !s.is_empty()), + upload_port_header: std::env::var("BUZZ_MEDIA_UPLOAD_PORT_HEADER") + .ok() + .map(|s| s.trim().to_lowercase()) + .filter(|s| !s.is_empty()), }; let media_max_concurrent_uploads: usize = std::env::var("BUZZ_MEDIA_MAX_CONCURRENT_UPLOADS") diff --git a/deploy/charts/buzz/templates/deployment.yaml b/deploy/charts/buzz/templates/deployment.yaml index 9f334197b..806e24144 100644 --- a/deploy/charts/buzz/templates/deployment.yaml +++ b/deploy/charts/buzz/templates/deployment.yaml @@ -85,6 +85,15 @@ spec: {{- if gt (.Values.relay.ephemeralTtlOverride | int) 0 }} - { name: BUZZ_EPHEMERAL_TTL_OVERRIDE, value: {{ .Values.relay.ephemeralTtlOverride | quote }} } {{- end }} + {{- if .Values.relay.uploadRecords }} + - { name: BUZZ_MEDIA_UPLOAD_RECORDS, value: "true" } + {{- end }} + {{- if .Values.relay.uploadIpHeader }} + - { name: BUZZ_MEDIA_UPLOAD_IP_HEADER, value: {{ .Values.relay.uploadIpHeader | quote }} } + {{- end }} + {{- if .Values.relay.uploadPortHeader }} + - { name: BUZZ_MEDIA_UPLOAD_PORT_HEADER, value: {{ .Values.relay.uploadPortHeader | quote }} } + {{- end }} # ── Owner ──────────────────────────────────────────────── - { name: RELAY_OWNER_PUBKEY, value: {{ .Values.ownerPubkey | quote }} } diff --git a/deploy/charts/buzz/values.yaml b/deploy/charts/buzz/values.yaml index 81d2e7c34..c118cdae2 100644 --- a/deploy/charts/buzz/values.yaml +++ b/deploy/charts/buzz/values.yaml @@ -84,6 +84,17 @@ relay: # accepts/owns the external multi-pod audio/SFU behavior. huddleAudioAvailable: null ephemeralTtlOverride: 0 + # Per-upload-event records (`_uploads/` moderation side channel). Off by + # default. Operators hosting communities for other people may have legal + # obligations (e.g. NCMEC reporting for US-serving providers) that require + # recording the network address of an upload; Buzz never collects IPs unless + # uploadIpHeader is set. When set (e.g. "cf-connecting-ip"), the connecting + # address reported by YOUR trusted edge is stored in the per-event record + # only — never served to clients, never in event data. uploadRecords must be + # true for uploadIpHeader to be valid (startup-checked). + uploadRecords: false + uploadIpHeader: "" + uploadPortHeader: "" livenessProbe: httpGet: @@ -211,7 +222,7 @@ postgresql: enabled: true size: 10Gi externalPostgresql: - url: "" # postgres://user:pass@host:5432/db + url: "" # postgres://user:pass@host:5432/db — placeholder example, sadscan:disable np.postgres.1 # ── Redis ──────────────────────────────────────────────────────────────────── # Eval-only CloudPirates subchart (standalone). REDIS_URL is composed in the