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