crates/routes/src/images.rs

   1 use actix_web::{
   2   body::BodyStream,
   3   error,
   4   http::{
   5     header::{HeaderName, ACCEPT_ENCODING, HOST},
   6     StatusCode,
   7   },
   8   web,
   9   Error,
  10   HttpRequest,
  11   HttpResponse,
  12 };
  13 use anyhow::anyhow;
  14 use futures::stream::{Stream, StreamExt};
  15 use lemmy_utils::{claims::Claims, rate_limit::RateLimit, LemmyError};
  16 use lemmy_websocket::LemmyContext;
  17 use reqwest::Body;
  18 use reqwest_middleware::{ClientWithMiddleware, RequestBuilder};
  19 use serde::{Deserialize, Serialize};
  20 use std::time::Duration;
  21
  22 pub fn config(cfg: &mut web::ServiceConfig, client: ClientWithMiddleware, rate_limit: &RateLimit) {
  23   cfg
  24     .app_data(web::Data::new(client))
  25     .service(
  26       web::resource("/pictrs/image")
  27         .wrap(rate_limit.image())
  28         .route(web::post().to(upload)),
  29     )
  30     // This has optional query params: /image/{filename}?format=jpg&thumbnail=256
  31     .service(web::resource("/pictrs/image/{filename}").route(web::get().to(full_res)))
  32     .service(web::resource("/pictrs/image/delete/{token}/{filename}").route(web::get().to(delete)));
  33 }
  34
  35 #[derive(Debug, Serialize, Deserialize)]
  36 struct Image {
  37   file: String,
  38   delete_token: String,
  39 }
  40
  41 #[derive(Debug, Serialize, Deserialize)]
  42 struct Images {
  43   msg: String,
  44   files: Option<Vec<Image>>,
  45 }
  46
  47 #[derive(Deserialize)]
  48 struct PictrsParams {
  49   format: Option<String>,
  50   thumbnail: Option<String>,
  51 }
  52
  53 fn adapt_request(
  54   request: &HttpRequest,
  55   client: &ClientWithMiddleware,
  56   url: String,
  57 ) -> RequestBuilder {
  58   // remove accept-encoding header so that pictrs doesnt compress the response
  59   const INVALID_HEADERS: &[HeaderName] = &[ACCEPT_ENCODING, HOST];
  60
  61   let client_request = client
  62     .request(request.method().clone(), url)
  63     .timeout(Duration::from_secs(30));
  64
  65   request
  66     .headers()
  67     .iter()
  68     .fold(client_request, |client_req, (key, value)| {
  69       if INVALID_HEADERS.contains(key) {
  70         client_req
  71       } else {
  72         client_req.header(key, value)
  73       }
  74     })
  75 }
  76
  77 async fn upload(
  78   req: HttpRequest,
  79   body: web::Payload,
  80   client: web::Data<ClientWithMiddleware>,
  81   context: web::Data<LemmyContext>,
  82 ) -> Result<HttpResponse, Error> {
  83   // TODO: check rate limit here
  84   let jwt = req
  85     .cookie("jwt")
  86     .expect("No auth header for picture upload");
  87
  88   if Claims::decode(jwt.value(), &context.secret().jwt_secret).is_err() {
  89     return Ok(HttpResponse::Unauthorized().finish());
  90   };
  91
  92   let image_url = format!("{}/image", pictrs_url(context.settings().pictrs_url)?);
  93
  94   let mut client_req = adapt_request(&req, &client, image_url);
  95
  96   if let Some(addr) = req.head().peer_addr {
  97     client_req = client_req.header("X-Forwarded-For", addr.to_string())
  98   };
  99
 100   let res = client_req
 101     .body(Body::wrap_stream(make_send(body)))
 102     .send()
 103     .await
 104     .map_err(error::ErrorBadRequest)?;
 105
 106   let status = res.status();
 107   let images = res.json::<Images>().await.map_err(error::ErrorBadRequest)?;
 108
 109   Ok(HttpResponse::build(status).json(images))
 110 }
 111
 112 async fn full_res(
 113   filename: web::Path<String>,
 114   web::Query(params): web::Query<PictrsParams>,
 115   req: HttpRequest,
 116   client: web::Data<ClientWithMiddleware>,
 117   context: web::Data<LemmyContext>,
 118 ) -> Result<HttpResponse, Error> {
 119   let name = &filename.into_inner();
 120
 121   // If there are no query params, the URL is original
 122   let pictrs_url_settings = context.settings().pictrs_url;
 123   let url = if params.format.is_none() && params.thumbnail.is_none() {
 124     format!(
 125       "{}/image/original/{}",
 126       pictrs_url(pictrs_url_settings)?,
 127       name,
 128     )
 129   } else {
 130     // Use jpg as a default when none is given
 131     let format = params.format.unwrap_or_else(|| "jpg".to_string());
 132
 133     let mut url = format!(
 134       "{}/image/process.{}?src={}",
 135       pictrs_url(pictrs_url_settings)?,
 136       format,
 137       name,
 138     );
 139
 140     if let Some(size) = params.thumbnail {
 141       url = format!("{}&thumbnail={}", url, size,);
 142     }
 143     url
 144   };
 145
 146   image(url, req, client).await
 147 }
 148
 149 async fn image(
 150   url: String,
 151   req: HttpRequest,
 152   client: web::Data<ClientWithMiddleware>,
 153 ) -> Result<HttpResponse, Error> {
 154   let mut client_req = adapt_request(&req, &client, url);
 155
 156   if let Some(addr) = req.head().peer_addr {
 157     client_req = client_req.header("X-Forwarded-For", addr.to_string());
 158   }
 159
 160   if let Some(addr) = req.head().peer_addr {
 161     client_req = client_req.header("X-Forwarded-For", addr.to_string());
 162   }
 163
 164   let res = client_req.send().await.map_err(error::ErrorBadRequest)?;
 165
 166   if res.status() == StatusCode::NOT_FOUND {
 167     return Ok(HttpResponse::NotFound().finish());
 168   }
 169
 170   let mut client_res = HttpResponse::build(res.status());
 171
 172   for (name, value) in res.headers().iter().filter(|(h, _)| *h != "connection") {
 173     client_res.insert_header((name.clone(), value.clone()));
 174   }
 175
 176   Ok(client_res.body(BodyStream::new(res.bytes_stream())))
 177 }
 178
 179 async fn delete(
 180   components: web::Path<(String, String)>,
 181   req: HttpRequest,
 182   client: web::Data<ClientWithMiddleware>,
 183   context: web::Data<LemmyContext>,
 184 ) -> Result<HttpResponse, Error> {
 185   let (token, file) = components.into_inner();
 186
 187   let url = format!(
 188     "{}/image/delete/{}/{}",
 189     pictrs_url(context.settings().pictrs_url)?,
 190     &token,
 191     &file
 192   );
 193
 194   let mut client_req = adapt_request(&req, &client, url);
 195
 196   if let Some(addr) = req.head().peer_addr {
 197     client_req = client_req.header("X-Forwarded-For", addr.to_string());
 198   }
 199
 200   let res = client_req.send().await.map_err(error::ErrorBadRequest)?;
 201
 202   Ok(HttpResponse::build(res.status()).body(BodyStream::new(res.bytes_stream())))
 203 }
 204
 205 fn pictrs_url(pictrs_url: Option<String>) -> Result<String, LemmyError> {
 206   pictrs_url.ok_or_else(|| anyhow!("images_disabled").into())
 207 }
 208
 209 fn make_send<S>(mut stream: S) -> impl Stream<Item = S::Item> + Send + Unpin + 'static
 210 where
 211   S: Stream + Unpin + 'static,
 212   S::Item: Send,
 213 {
 214   // NOTE: the 8 here is arbitrary
 215   let (tx, rx) = tokio::sync::mpsc::channel(8);
 216
 217   // NOTE: spawning stream into a new task can potentially hit this bug:
 218   // - https://github.com/actix/actix-web/issues/1679
 219   //
 220   // Since 4.0.0-beta.2 this issue is incredibly less frequent. I have not personally reproduced it.
 221   // That said, it is still technically possible to encounter.
 222   actix_web::rt::spawn(async move {
 223     while let Some(res) = stream.next().await {
 224       if tx.send(res).await.is_err() {
 225         break;
 226       }
 227     }
 228   });
 229
 230   SendStream { rx }
 231 }
 232
 233 struct SendStream<T> {
 234   rx: tokio::sync::mpsc::Receiver<T>,
 235 }
 236
 237 impl<T> Stream for SendStream<T>
 238 where
 239   T: Send,
 240 {
 241   type Item = T;
 242
 243   fn poll_next(
 244     mut self: std::pin::Pin<&mut Self>,
 245     cx: &mut std::task::Context<'_>,
 246   ) -> std::task::Poll<Option<Self::Item>> {
 247     std::pin::Pin::new(&mut self.rx).poll_recv(cx)
 248   }
 249 }