fix(self-host): unfurl media hosted by the instance itself (#2451)

deploy/self-hosting/docker-compose.yml

@@ -434,6 +434,7 @@ services:
       FLUXER_MEDIA_PROXY_PORT: "8080"
       FLUXER_MEDIA_PROXY_MODE: upload
       FLUXER_MEDIA_PROXY_STORAGE_BACKEND: s3
+      FLUXER_MEDIA_PROXY_PUBLIC_ENDPOINT: ${FLUXER_PUBLIC_ORIGIN:-${FLUXER_PUBLIC_SCHEME:-https}://${FLUXER_DOMAIN}}/media
     depends_on:
       seaweedfs-init: {condition: service_completed_successfully}
       nats: {condition: service_healthy}

fluxer_media_proxy/src/config/mod.rs

@@ -85,6 +85,7 @@ pub struct Config {
     pub bind_host: String,
     pub port: u16,
     pub(crate) secret_key: SecretString,
+    pub public_endpoint: Option<String>,
     pub mode: DeploymentMode,
     pub read_only: bool,
     pub shutdown_grace_ms: u64,
@@ -134,6 +135,8 @@ impl Config {
                 8080,
             )?,
             secret_key,
+            public_endpoint: non_empty(env.get("FLUXER_MEDIA_PROXY_PUBLIC_ENDPOINT"))
+                .map(|endpoint| endpoint.trim_end_matches('/').to_owned()),
             mode,
             read_only: parse_bool(
                 "FLUXER_MEDIA_PROXY_READ_ONLY",

fluxer_media_proxy/src/server/media_operations/input.rs

@@ -3,8 +3,14 @@
 use super::failure::MediaFailure;
 use crate::{
     byte_budget::BudgetedBytes,
-    constants,
+    constants::{self, AssetExtension},
-    server::{external::fetch_external, params::url_filename, state::AppState},
+    server::{
+        external::fetch_external,
+        params::url_filename,
+        self_origin::{self, SelfOrigin},
+        state::AppState,
+    },
+    storage::StorageError,
 };
 use base64::{Engine as _, engine::general_purpose};
 use bytes::Bytes;
@@ -112,17 +118,61 @@ pub(in crate::server) async fn load_media_input(
                 filename: filename.unwrap_or(key),
             })
         }
-        MediaInput::External { url, filename } => {
+        MediaInput::External { url, filename } => match self_origin::resolve(app, &url) {
-            let (fetched_url, data) = fetch_external(app, &url).await.map_err(|err| {
+            Some(SelfOrigin::Stored {
-                let failure = MediaFailure::from(err);
+                bucket,
-                warn!(reason = failure.code(), url = url.as_str(), ?err);
+                key,
-                failure
+                fallback_ext,
-            })?;
+            }) => {
-            Ok(LoadedMediaInput {
+                let data = read_own_object(app, &bucket, &key, fallback_ext)
-                data: retained_input_bytes(data),
+                    .await
-                filename: filename.unwrap_or_else(|| url_filename(&fetched_url)),
+                    .map_err(|err| {
-            })
+                        let failure = MediaFailure::MetadataS3Read;
+                        warn!(reason = failure.code(), bucket = bucket.as_str(), key = key.as_str(), %err);
+                        failure
+                    })?;
+                Ok(LoadedMediaInput {
+                    data,
+                    filename: filename.unwrap_or_else(|| url_filename(&url)),
+                })
+            }
+            Some(SelfOrigin::External { url: target }) => {
+                load_external_url(app, &target, filename).await
+            }
+            None => load_external_url(app, &url, filename).await,
+        },
+    }
+}
+
+async fn load_external_url(
+    app: &AppState,
+    url: &str,
+    filename: Option<String>,
+) -> Result<LoadedMediaInput, MediaFailure> {
+    let (fetched_url, data) = fetch_external(app, url).await.map_err(|err| {
+        let failure = MediaFailure::from(err);
+        warn!(reason = failure.code(), url, ?err);
+        failure
+    })?;
+    Ok(LoadedMediaInput {
+        data: retained_input_bytes(data),
+        filename: filename.unwrap_or_else(|| url_filename(&fetched_url)),
+    })
+}
+
+async fn read_own_object(
+    app: &AppState,
+    bucket: &str,
+    key: &str,
+    fallback_ext: Option<AssetExtension>,
+) -> Result<Bytes, StorageError> {
+    match app.store.read_object(bucket, key).await {
+        Err(StorageError::NotFound) => {
+            let fallback_ext = fallback_ext.ok_or(StorageError::NotFound)?;
+            let fallback_key = format!("{key}.{}", fallback_ext.name());
+            Ok(app.store.read_object(bucket, &fallback_key).await?.data)
         }
+        other => Ok(other?.data),
     }
 }
 

fluxer_media_proxy/src/server/mod.rs

@@ -12,6 +12,7 @@ mod relay;
 mod response;
 mod routes;
 mod runtime;
+mod self_origin;
 mod state;
 mod stored;
 pub mod transform;

fluxer_media_proxy/src/server/self_origin.rs

@@ -0,0 +1,255 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use crate::{
+    config::DeploymentMode,
+    constants::{AssetExtension, AssetKind},
+    external_path, public_net_policy,
+    server::{
+        asset_path::{
+            decode_storage_key, parse_entrance_sound_path, parse_guild_member_asset_path,
+            parse_simple_asset_path, parse_standard_asset_path,
+        },
+        state::AppState,
+    },
+    signing,
+};
+
+pub(in crate::server) enum SelfOrigin {
+    Stored {
+        bucket: String,
+        key: String,
+        fallback_ext: Option<AssetExtension>,
+    },
+    External {
+        url: String,
+    },
+}
+
+pub(in crate::server) fn resolve(app: &AppState, url: &str) -> Option<SelfOrigin> {
+    let endpoint = app.cfg.public_endpoint.as_deref()?;
+    let path = own_path(endpoint, url)?;
+    let cdn = &app.cfg.storage.bucket_cdn;
+    if app.cfg.mode == DeploymentMode::Static {
+        return stored(
+            &app.cfg.storage.bucket_static,
+            decode_storage_key(path).ok()?,
+        );
+    }
+    if let Some(rest) = path.strip_prefix("/external/") {
+        let (signature, proxy_path) = rest.split_once('/')?;
+        if !signing::verify_signature(proxy_path, signature, app.cfg.secret_key.as_bytes()) {
+            return None;
+        }
+        let target = external_path::reconstruct_original_url(proxy_path).ok()?;
+        if own_path(endpoint, &target).is_some() {
+            return None;
+        }
+        return Some(SelfOrigin::External { url: target });
+    }
+    if path.starts_with("/attachments/") || (path.starts_with("/themes/") && path.ends_with(".css"))
+    {
+        return stored(cdn, decode_storage_key(path).ok()?);
+    }
+    if let Some(key) = parse_entrance_sound_path(path) {
+        return stored(cdn, key);
+    }
+    let asset = parse_guild_member_asset_path(path)
+        .or_else(|| parse_simple_asset_path(path, AssetKind::Emoji))
+        .or_else(|| parse_simple_asset_path(path, AssetKind::Sticker))
+        .or_else(|| parse_standard_asset_path(path))?;
+    Some(SelfOrigin::Stored {
+        bucket: cdn.clone(),
+        key: asset.storage_key,
+        fallback_ext: Some(asset.original_ext),
+    })
+}
+
+fn stored(bucket: &str, key: String) -> Option<SelfOrigin> {
+    Some(SelfOrigin::Stored {
+        bucket: bucket.to_owned(),
+        key,
+        fallback_ext: None,
+    })
+}
+
+fn own_path<'a>(endpoint: &str, url: &'a str) -> Option<&'a str> {
+    let base = public_net_policy::parse_url(endpoint).ok()?;
+    let target = public_net_policy::parse_url(url).ok()?;
+    if !base.scheme.eq_ignore_ascii_case(target.scheme) {
+        return None;
+    }
+    if !base.host.eq_ignore_ascii_case(target.host) {
+        return None;
+    }
+    if effective_port(base.scheme, base.port) != effective_port(target.scheme, target.port) {
+        return None;
+    }
+    let prefix = path_of(base.path_query).trim_end_matches('/');
+    let rest = path_of(target.path_query).strip_prefix(prefix)?;
+    rest.starts_with('/').then_some(rest)
+}
+
+fn path_of(path_query: &str) -> &str {
+    path_query
+        .split_once('?')
+        .map(|(path, _)| path)
+        .unwrap_or(path_query)
+}
+
+fn effective_port(scheme: &str, port: Option<u16>) -> u16 {
+    port.unwrap_or(if scheme.eq_ignore_ascii_case("https") {
+        443
+    } else {
+        80
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::Config;
+
+    const ENDPOINT: &str = "https://chat.example.com/media";
+
+    fn app() -> AppState {
+        AppState::for_tests(
+            Config::load_from_iter([
+                (
+                    "FLUXER_MEDIA_PROXY_SECRET_KEY".to_owned(),
+                    "secret".to_owned(),
+                ),
+                (
+                    "FLUXER_MEDIA_PROXY_PUBLIC_ENDPOINT".to_owned(),
+                    format!("{ENDPOINT}/"),
+                ),
+                (
+                    "FLUXER_MEDIA_PROXY_STORAGE_BACKEND".to_owned(),
+                    "local".to_owned(),
+                ),
+            ])
+            .expect("self origin test config"),
+        )
+    }
+
+    #[test]
+    fn own_attachments_read_from_the_cdn_bucket() {
+        let app = app();
+        let resolved = resolve(
+            &app,
+            &format!("{ENDPOINT}/attachments/1544725486800732163/1544971349200470016/cat.gif"),
+        );
+        match resolved {
+            Some(SelfOrigin::Stored { bucket, key, .. }) => {
+                assert_eq!(app.cfg.storage.bucket_cdn, bucket);
+                assert_eq!(
+                    "attachments/1544725486800732163/1544971349200470016/cat.gif",
+                    key
+                );
+            }
+            _ => panic!("own attachment paths resolve to stored objects"),
+        }
+    }
+
+    #[test]
+    fn own_avatars_carry_the_extension_fallback() {
+        let app = app();
+        match resolve(
+            &app,
+            &format!("{ENDPOINT}/avatars/1216100949629702144/a1b2c3d4e5f6.png?size=64"),
+        ) {
+            Some(SelfOrigin::Stored {
+                key, fallback_ext, ..
+            }) => {
+                assert_eq!("avatars/1216100949629702144/a1b2c3d4e5f6", key);
+                assert_eq!(Some(AssetExtension::Png), fallback_ext);
+            }
+            _ => panic!("own asset paths resolve to stored objects"),
+        }
+    }
+
+    #[test]
+    fn own_external_paths_unwrap_to_the_origin_url() {
+        let app = app();
+        let target = "https://static.klipy.com/ii/c8/28/HkAKKCzZ.webp";
+        let proxied = fluxer_common::external_media_path::build_external_media_proxy_url(
+            ENDPOINT,
+            target,
+            app.cfg.secret_key.as_bytes(),
+        )
+        .expect("proxy url");
+        match resolve(&app, &proxied) {
+            Some(SelfOrigin::External { url }) => assert_eq!(target, url),
+            _ => panic!("own external paths unwrap to the origin url"),
+        }
+        let tampered = proxied.replace("/external/", "/external/x");
+        assert!(resolve(&app, &tampered).is_none());
+    }
+
+    #[test]
+    fn foreign_and_unknown_paths_stay_external_fetches() {
+        let app = app();
+        assert!(resolve(&app, "https://static.klipy.com/ii/c8/28/HkAKKCzZ.webp").is_none());
+        assert!(resolve(&app, &format!("{ENDPOINT}/channels/1/2")).is_none());
+    }
+
+    #[test]
+    fn own_path_matches_the_endpoint_prefix() {
+        assert_eq!(
+            own_path(
+                "https://chat.example.com/media",
+                "https://chat.example.com/media/attachments/1/2/cat.gif"
+            ),
+            Some("/attachments/1/2/cat.gif")
+        );
+        assert_eq!(
+            own_path(
+                "https://cdn.example.com",
+                "https://cdn.example.com/avatars/1/abc.png?size=64"
+            ),
+            Some("/avatars/1/abc.png")
+        );
+    }
+
+    #[test]
+    fn own_path_rejects_other_origins_and_sibling_prefixes() {
+        assert_eq!(
+            own_path(
+                "https://chat.example.com/media",
+                "https://evil.example.com/media/attachments/1/2/cat.gif"
+            ),
+            None
+        );
+        assert_eq!(
+            own_path(
+                "https://chat.example.com/media",
+                "http://chat.example.com/media/attachments/1/2/cat.gif"
+            ),
+            None
+        );
+        assert_eq!(
+            own_path(
+                "https://chat.example.com/media",
+                "https://chat.example.com:8443/media/attachments/1/2/cat.gif"
+            ),
+            None
+        );
+        assert_eq!(
+            own_path(
+                "https://chat.example.com/media",
+                "https://chat.example.com/mediafiles/attachments/1/2/cat.gif"
+            ),
+            None
+        );
+    }
+
+    #[test]
+    fn own_path_keeps_explicit_default_ports() {
+        assert_eq!(
+            own_path(
+                "https://chat.example.com:443",
+                "https://chat.example.com/a.png"
+            ),
+            Some("/a.png")
+        );
+    }
+}

fluxer_media_proxy/src/storage/tests/mod.rs

@@ -31,6 +31,7 @@ fn test_config(root: &Path) -> Config {
         bind_host: "127.0.0.1".to_owned(),
         port: 0,
         secret_key: SecretString::new("secret".to_owned()),
+        public_endpoint: None,
         mode: DeploymentMode::Mp,
         read_only: false,
         shutdown_grace_ms: 0,

fluxer_unfurl/src/media_proxy.rs

@@ -112,10 +112,13 @@ impl MediaProxyClient {
         }
     }
 
-    pub fn external_proxy_url(&self, input_url: &str) -> Option<String> {
+    pub fn is_own_url(&self, input_url: &str) -> bool {
-        if input_url == self.public_endpoint
+        input_url == self.public_endpoint
             || input_url.starts_with(&format!("{}/", self.public_endpoint))
-        {
+    }
+
+    pub fn external_proxy_url(&self, input_url: &str) -> Option<String> {
+        if self.is_own_url(input_url) {
             return Some(input_url.to_owned());
         }
         let parsed = Url::parse(input_url).ok()?;
@@ -160,6 +163,20 @@ mod tests {
         );
     }
 
+    #[test]
+    fn own_urls_are_recognised_by_their_public_endpoint() {
+        let mp = MediaProxyClient::new_with_public_endpoint(
+            "http://media-proxy:8080",
+            "secret",
+            Some("https://chat.example.test/media/"),
+            reqwest::Client::new(),
+        );
+        assert!(mp.is_own_url("https://chat.example.test/media"));
+        assert!(mp.is_own_url("https://chat.example.test/media/attachments/1/2/cat.gif"));
+        assert!(!mp.is_own_url("https://chat.example.test/mediafiles/1.gif"));
+        assert!(!mp.is_own_url("https://static.klipy.com/ii/c8/28/HkAKKCzZ.webp"));
+    }
+
     #[test]
     fn external_proxy_url_uses_public_endpoint_and_plain_path() {
         let client = reqwest::Client::new();

fluxer_unfurl/src/resolvers/media.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 use crate::direct_media::{MediaKind, detect_media_kind};
-use crate::media_proxy::{MediaProxyClient, embed_media_flags};
+use crate::media_proxy::{MediaMetadata, MediaProxyClient, embed_media_flags};
 use crate::types::{EmbedMedia, MessageEmbed, NsfwMode};
 use url::Url;
 
@@ -71,9 +71,28 @@ pub async fn build_direct_media_embed(
     nsfw_mode: NsfwMode,
     kind: MediaKind,
 ) -> anyhow::Result<MessageEmbed> {
-    let url_str = url.to_string();
     let nsfw_str = MediaProxyClient::nsfw_mode_str(nsfw_mode);
-    let meta = media_proxy.get_metadata(&url_str, nsfw_str).await?;
+    let meta = media_proxy.get_metadata(url.as_str(), nsfw_str).await?;
+    Ok(media_embed(url, &meta, kind))
+}
+
+pub async fn build_own_media_embed(
+    media_proxy: &MediaProxyClient,
+    url: &Url,
+    nsfw_mode: NsfwMode,
+) -> anyhow::Result<Option<MessageEmbed>> {
+    let nsfw_str = MediaProxyClient::nsfw_mode_str(nsfw_mode);
+    let meta = media_proxy.get_metadata(url.as_str(), nsfw_str).await?;
+    let Some(kind) =
+        media_kind_from_content_type(&meta.content_type).or_else(|| detect_media_kind(url))
+    else {
+        return Ok(None);
+    };
+    Ok(Some(media_embed(url, &meta, kind)))
+}
+
+fn media_embed(url: &Url, meta: &MediaMetadata, kind: MediaKind) -> MessageEmbed {
+    let url_str = url.to_string();
     let media = EmbedMedia {
         url: Some(url_str.clone()),
         content_type: Some(meta.content_type.clone()),
@@ -82,7 +101,7 @@ pub async fn build_direct_media_embed(
         height: meta.height,
         placeholder: meta.placeholder.clone(),
         duration: meta.duration.map(|duration| duration as u32),
-        flags: embed_media_flags(&meta),
+        flags: embed_media_flags(meta),
         ..Default::default()
     };
 
@@ -104,7 +123,7 @@ pub async fn build_direct_media_embed(
         }
     };
     embed.url = Some(url_str);
-    Ok(embed)
+    embed
 }
 
 #[cfg(test)]

fluxer_unfurl/src/shard_impl.rs

@@ -121,6 +121,10 @@ impl UnfurlShard {
     ) -> anyhow::Result<UnfurlResult> {
         let parsed = Url::parse(url_str)?;
 
+        if self.media_proxy.is_own_url(parsed.as_str()) {
+            return Ok(self.resolve_own_media(&parsed, nsfw_mode).await);
+        }
+
         let (fetch_url, matched_resolver_idx) = self.find_transform(&parsed);
 
         let ctx = ResolveContext {
@@ -177,6 +181,25 @@ impl UnfurlShard {
         })
     }
 
+    async fn resolve_own_media(&self, url: &Url, nsfw_mode: NsfwMode) -> UnfurlResult {
+        match resolvers::media::build_own_media_embed(&self.media_proxy, url, nsfw_mode).await {
+            Ok(Some(embed)) => self.finalize_result(ResolverResult {
+                embeds: vec![embed],
+            }),
+            Ok(None) => UnfurlResult {
+                embeds: Vec::new(),
+                cache_ttl_seconds: None,
+            },
+            Err(err) => {
+                tracing::warn!(error = %err, url = %url, "own media unfurl failed");
+                UnfurlResult {
+                    embeds: Vec::new(),
+                    cache_ttl_seconds: None,
+                }
+            }
+        }
+    }
+
     fn find_transform(&self, url: &Url) -> (Url, Option<usize>) {
         for (i, resolver) in self.resolvers.iter().enumerate() {
             if let Some(transformed) = resolver.transform_url(url) {