feat(media): write per-upload-event records for moderation (#1551)

Signed-off-by: Bradley Axen <baxen@squareup.com>
Signed-off-by: npub1dccv64krpcpse5cmkzfeh998cftungyatw3djt8jwdw6g43f7fyqzzmrf7 <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1ty04d6th3jzvggd25za6xmqrh99g42kvkvqsmf7ydv793fgyxa3s3t2lw7 <591f56e9778c84c421aaa0bba36c03b94a8aaaccb3010da7c46b3c58a5043763@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1dccv64krpcpse5cmkzfeh998cftungyatw3djt8jwdw6g43f7fyqzzmrf7 <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@sprout-oss.stage.blox.sqprod.co>

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"

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"

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<String>,
+    /// 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<String>,
 }
 
 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:?}");
+        }
+    }
+}

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};

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,
         }
     }
 

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<V, M, Fut>(
-    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<UploadAttribution>,
+}
+
+async fn process_buffered_upload<V, M, Fut>(
+    input: BufferedUploadInput<'_>,
     validate: V,
-    store_metadata: M,
+    prepare_metadata: M,
 ) -> Result<BlobDescriptor, MediaError>
 where
     V: FnOnce(&Bytes, &MediaConfig) -> Result<(String, String), MediaError> + Send + 'static,
     M: FnOnce(MetadataInput) -> Fut,
     Fut: std::future::Future<Output = Result<BlobMeta, MediaError>>,
 {
+    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<UploadAttribution>,
 ) -> Result<BlobDescriptor, MediaError> {
     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<UploadAttribution>,
 ) -> Result<BlobDescriptor, MediaError> {
     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<Item = Result<Bytes, axum::Error>> + Send + 'static,
     content_length: Option<u64>,
+    attribution: Option<UploadAttribution>,
 ) -> Result<BlobDescriptor, MediaError> {
     // --- 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<BlobMeta, MediaError> {
     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,
         }
     }
 

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<String>,
+    /// 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<String>,
+    /// 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<u16>,
+}
+
+/// 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<IpAddr>,
+    /// Source port of the uploader, or `None`. Ignored when `ip` is `None`.
+    pub port: Option<u16>,
+}
+
+/// 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<String>,
+    /// 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<IpAddr> {
+    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<u16> {
+    raw.trim().parse::<u16>().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);
+    }
+}

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,
         }
     }
 

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,
     }
 }
 

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<Arc<AppState>> 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<UploadAttribution> {
+    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<String>| {
+        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!(

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")

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 }} }

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