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