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