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