5 header::{HeaderName, ACCEPT_ENCODING, HOST},
13 use futures::stream::{Stream, StreamExt};
14 use lemmy_utils::{claims::Claims, rate_limit::RateLimit, REQWEST_TIMEOUT};
15 use lemmy_websocket::LemmyContext;
17 use reqwest_middleware::{ClientWithMiddleware, RequestBuilder};
18 use serde::{Deserialize, Serialize};
20 pub fn config(cfg: &mut web::ServiceConfig, client: ClientWithMiddleware, rate_limit: &RateLimit) {
22 .app_data(web::Data::new(client))
24 web::resource("/pictrs/image")
25 .wrap(rate_limit.image())
26 .route(web::post().to(upload)),
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)));
34 #[derive(Debug, Serialize, Deserialize)]
40 #[derive(Debug, Serialize, Deserialize)]
43 files: Option<Vec<Image>>,
46 #[derive(Deserialize)]
48 format: Option<String>,
49 thumbnail: Option<String>,
52 #[derive(Deserialize)]
53 enum PictrsPurgeParams {
54 #[serde(rename = "file")]
56 #[serde(rename = "alias")]
61 request: &HttpRequest,
62 client: &ClientWithMiddleware,
65 // remove accept-encoding header so that pictrs doesnt compress the response
66 const INVALID_HEADERS: &[HeaderName] = &[ACCEPT_ENCODING, HOST];
68 let client_request = client
69 .request(request.method().clone(), url)
70 .timeout(REQWEST_TIMEOUT);
75 .fold(client_request, |client_req, (key, value)| {
76 if INVALID_HEADERS.contains(key) {
79 client_req.header(key, value)
87 client: web::Data<ClientWithMiddleware>,
88 context: web::Data<LemmyContext>,
89 ) -> Result<HttpResponse, Error> {
90 // TODO: check rate limit here
93 .expect("No auth header for picture upload");
95 if Claims::decode(jwt.value(), &context.secret().jwt_secret).is_err() {
96 return Ok(HttpResponse::Unauthorized().finish());
99 let pictrs_config = context.settings().pictrs_config()?;
100 let image_url = format!("{}/image", pictrs_config.url);
102 let mut client_req = adapt_request(&req, &client, image_url);
104 if let Some(addr) = req.head().peer_addr {
105 client_req = client_req.header("X-Forwarded-For", addr.to_string())
109 .body(Body::wrap_stream(make_send(body)))
112 .map_err(error::ErrorBadRequest)?;
114 let status = res.status();
115 let images = res.json::<Images>().await.map_err(error::ErrorBadRequest)?;
117 Ok(HttpResponse::build(status).json(images))
121 filename: web::Path<String>,
122 web::Query(params): web::Query<PictrsParams>,
124 client: web::Data<ClientWithMiddleware>,
125 context: web::Data<LemmyContext>,
126 ) -> Result<HttpResponse, Error> {
127 let name = &filename.into_inner();
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,)
134 // Use jpg as a default when none is given
135 let format = params.format.unwrap_or_else(|| "jpg".to_string());
137 let mut url = format!(
138 "{}/image/process.{}?src={}",
139 pictrs_config.url, format, name,
142 if let Some(size) = params.thumbnail {
143 url = format!("{}&thumbnail={}", url, size,);
148 image(url, req, client).await
154 client: web::Data<ClientWithMiddleware>,
155 ) -> Result<HttpResponse, Error> {
156 let mut client_req = adapt_request(&req, &client, url);
158 if let Some(addr) = req.head().peer_addr {
159 client_req = client_req.header("X-Forwarded-For", addr.to_string());
162 if let Some(addr) = req.head().peer_addr {
163 client_req = client_req.header("X-Forwarded-For", addr.to_string());
166 let res = client_req.send().await.map_err(error::ErrorBadRequest)?;
168 if res.status() == StatusCode::NOT_FOUND {
169 return Ok(HttpResponse::NotFound().finish());
172 let mut client_res = HttpResponse::build(res.status());
174 for (name, value) in res.headers().iter().filter(|(h, _)| *h != "connection") {
175 client_res.insert_header((name.clone(), value.clone()));
178 Ok(client_res.body(BodyStream::new(res.bytes_stream())))
182 components: web::Path<(String, String)>,
184 client: web::Data<ClientWithMiddleware>,
185 context: web::Data<LemmyContext>,
186 ) -> Result<HttpResponse, Error> {
187 let (token, file) = components.into_inner();
189 let pictrs_config = context.settings().pictrs_config()?;
190 let url = format!("{}/image/delete/{}/{}", pictrs_config.url, &token, &file);
192 let mut client_req = adapt_request(&req, &client, url);
194 if let Some(addr) = req.head().peer_addr {
195 client_req = client_req.header("X-Forwarded-For", addr.to_string());
198 let res = client_req.send().await.map_err(error::ErrorBadRequest)?;
200 Ok(HttpResponse::build(res.status()).body(BodyStream::new(res.bytes_stream())))
204 web::Query(params): web::Query<PictrsPurgeParams>,
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),
214 let pictrs_config = context.settings().pictrs_config()?;
215 let url = format!("{}/internal/purge?{}", pictrs_config.url, &purge_string);
217 let mut client_req = adapt_request(&req, &client, url);
219 if let Some(addr) = req.head().peer_addr {
220 client_req = client_req.header("X-Forwarded-For", addr.to_string())
223 // Add the API token, X-Api-Token header
224 client_req = client_req.header("x-api-token", pictrs_config.api_key);
226 let res = client_req.send().await.map_err(error::ErrorBadRequest)?;
228 Ok(HttpResponse::build(res.status()).body(BodyStream::new(res.bytes_stream())))
231 fn make_send<S>(mut stream: S) -> impl Stream<Item = S::Item> + Send + Unpin + 'static
233 S: Stream + Unpin + 'static,
236 // NOTE: the 8 here is arbitrary
237 let (tx, rx) = tokio::sync::mpsc::channel(8);
239 // NOTE: spawning stream into a new task can potentially hit this bug:
240 // - https://github.com/actix/actix-web/issues/1679
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() {
255 struct SendStream<T> {
256 rx: tokio::sync::mpsc::Receiver<T>,
259 impl<T> Stream for SendStream<T>
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)