feat(service/media): add S3 support

Cargo.lock

@@ -500,6 +500,7 @@ dependencies = [
  "rusqlite",
  "rust-argon2",
  "rust-rocksdb",
+ "rusty-s3",
  "sd-notify",
  "serde",
  "serde_html_form",
@@ -1733,6 +1734,16 @@ version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94"
 
+[[package]]
+name = "md-5"
+version = "0.10.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d89e7ee0cfbedfc4da3340218492196241d89eefb6dab27de5df917a6d2e78cf"
+dependencies = [
+ "cfg-if",
+ "digest",
+]
+
 [[package]]
 name = "memchr"
 version = "2.7.4"
@@ -2245,6 +2256,16 @@ version = "2.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a993555f31e5a609f617c12db6250dedcac1b0a85076912c436e6fc9b2c8e6a3"
 
+[[package]]
+name = "quick-xml"
+version = "0.37.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "331e97a1af0bf59823e6eadffe373d7b27f485be8748f71471c662c1f269b7fb"
+dependencies = [
+ "memchr",
+ "serde",
+]
+
 [[package]]
 name = "quote"
 version = "1.0.40"
@@ -2804,6 +2825,25 @@ version = "1.0.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "955d28af4278de8121b7ebeb796b6a45735dc01436d898801014aced2773a3d6"
 
+[[package]]
+name = "rusty-s3"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f51a5a6b15f25d3e10c068039ee13befb6110fcb36c2b26317bcbdc23484d96"
+dependencies = [
+ "base64 0.22.1",
+ "hmac",
+ "md-5",
+ "percent-encoding",
+ "quick-xml",
+ "serde",
+ "serde_json",
+ "sha2",
+ "time",
+ "url",
+ "zeroize",
+]
+
 [[package]]
 name = "ryu"
 version = "1.0.20"

Cargo.toml

@@ -154,6 +154,8 @@ tikv-jemallocator = { version = "0.6", features = [
 
 sd-notify = { version = "0.4", optional = true }
 
+rusty-s3 = "0.7.0"
+
 # Used for matrix spec type definitions and helpers
 [dependencies.ruma]
 features = [

src/config/mod.rs

@@ -242,6 +242,32 @@ impl From<IncompleteConfig> for Config {
                     }),
                     directory_structure,
                 },
+                IncompleteMediaBackendConfig::S3 {
+                    endpoint,
+                    bucket,
+                    region,
+                    key,
+                    secret,
+                    duration,
+                    bucket_use_path,
+                } => {
+                    let path_style = if bucket_use_path {
+                        rusty_s3::UrlStyle::Path
+                    } else {
+                        rusty_s3::UrlStyle::VirtualHost
+                    };
+
+                    let bucket = rusty_s3::Bucket::new(endpoint, path_style, bucket, region)
+                        .expect("Invalid S3 config");
+
+                    let credentials = rusty_s3::Credentials::new(key, secret);
+
+                    MediaBackendConfig::S3 {
+                        bucket: bucket,
+                        credentials: credentials,
+                        duration: Duration::from_secs(duration),
+                    }
+                }
             },
             retention: media.retention.into(),
         };
@@ -481,6 +507,17 @@ pub enum IncompleteMediaBackendConfig {
         #[serde(default)]
         directory_structure: DirectoryStructure,
     },
+    S3 {
+        endpoint: Url,
+        bucket: String,
+        region: String,
+        key: String,
+        secret: String,
+        #[serde(default = "default_s3_duration")]
+        duration: u64,
+        #[serde(default = "false_fn")]
+        bucket_use_path: bool,
+    },
 }
 
 impl Default for IncompleteMediaBackendConfig {
@@ -498,6 +535,11 @@ pub enum MediaBackendConfig {
         path: String,
         directory_structure: DirectoryStructure,
     },
+    S3 {
+        bucket: rusty_s3::Bucket,
+        credentials: rusty_s3::Credentials,
+        duration: Duration,
+    },
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -727,3 +769,7 @@ fn default_openid_token_ttl() -> u64 {
 pub fn default_default_room_version() -> RoomVersionId {
     RoomVersionId::V10
 }
+
+fn default_s3_duration() -> u64 {
+    30
+}

src/service/globals/mod.rs

@@ -230,8 +230,6 @@ impl Service {
             shutdown: AtomicBool::new(false),
         };
 
-        // Remove this exception once other media backends are added
-        #[allow(irrefutable_let_patterns)]
         if let MediaBackendConfig::FileSystem { path, .. } = &s.config.media.backend {
             fs::create_dir_all(path)?;
         }

src/service/media/mod.rs

@@ -8,8 +8,9 @@ use ruma::{
     http_headers::{ContentDisposition, ContentDispositionType},
     OwnedServerName, ServerName, UserId,
 };
+use rusty_s3::S3Action;
 use sha2::{digest::Output, Digest, Sha256};
-use tracing::{error, info};
+use tracing::{error, info, warn};
 
 use crate::{
     config::{DirectoryStructure, MediaBackendConfig},
@@ -615,6 +616,25 @@ impl Service {
 
                 file
             }
+            MediaBackendConfig::S3 {
+                bucket,
+                credentials,
+                duration,
+            } => {
+                let url = bucket
+                    .get_object(Some(credentials), &hex::encode(sha256_digest))
+                    .sign(*duration);
+
+                let client = services().globals.default_client();
+                let resp = client.get(url).send().await?;
+
+                if !resp.status().is_success() {
+                    warn!("S3 request error:\n{}", resp.text().await?);
+                    return Err(Error::bad_database("Cannot read media file"));
+                }
+
+                resp.bytes().await?.to_vec()
+            }
         };
 
         if let Some((server_name, media_id)) = original_file_id {
@@ -643,6 +663,23 @@ pub async fn create_file(sha256_hex: &str, file: &[u8]) -> Result<()> {
             let mut f = File::create(path).await?;
             f.write_all(file).await?;
         }
+        MediaBackendConfig::S3 {
+            bucket,
+            credentials,
+            duration,
+        } => {
+            let url = bucket
+                .put_object(Some(credentials), sha256_hex)
+                .sign(*duration);
+
+            let client = services().globals.default_client();
+            let resp = client.put(url).body(file.to_vec()).send().await?;
+
+            if !resp.status().is_success() {
+                warn!("S3 request error:\n{}", resp.text().await?);
+                return Err(Error::bad_database("Cannot write media file"));
+            }
+        }
     }
 
     Ok(())
@@ -713,6 +750,23 @@ async fn delete_file(sha256_hex: &str) -> Result<()> {
                 }
             }
         }
+        MediaBackendConfig::S3 {
+            bucket,
+            credentials,
+            duration,
+        } => {
+            let url = bucket
+                .delete_object(Some(credentials), sha256_hex)
+                .sign(*duration);
+
+            let client = services().globals.default_client();
+            let resp = client.delete(url).send().await?;
+
+            if !resp.status().is_success() {
+                warn!("S3 request error:\n{}", resp.text().await?);
+                return Err(Error::bad_database("Cannot delete media file"));
+            }
+        }
     }
 
     Ok(())