5 header::{HeaderName, ACCEPT_ENCODING, HOST},
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;
19 use reqwest_middleware::{ClientWithMiddleware, RequestBuilder};
20 use serde::{Deserialize, Serialize};
23 cfg: &mut web::ServiceConfig,
24 client: ClientWithMiddleware,
25 rate_limit: &RateLimitCell,
28 .app_data(web::Data::new(client))
30 web::resource("/pictrs/image")
31 .wrap(rate_limit.image())
32 .route(web::post().to(upload)),
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)));
39 #[derive(Debug, Serialize, Deserialize)]
45 #[derive(Debug, Serialize, Deserialize)]
48 files: Option<Vec<Image>>,
51 #[derive(Deserialize)]
53 format: Option<String>,
54 thumbnail: Option<i32>,
57 #[derive(Deserialize)]
58 enum PictrsPurgeParams {
59 #[serde(rename = "file")]
61 #[serde(rename = "alias")]
66 request: &HttpRequest,
67 client: &ClientWithMiddleware,
70 // remove accept-encoding header so that pictrs doesnt compress the response
71 const INVALID_HEADERS: &[HeaderName] = &[ACCEPT_ENCODING, HOST];
73 let client_request = client
74 .request(request.method().clone(), url)
75 .timeout(REQWEST_TIMEOUT);
80 .fold(client_request, |client_req, (key, value)| {
81 if INVALID_HEADERS.contains(key) {
84 client_req.header(key, value)
92 client: web::Data<ClientWithMiddleware>,
93 context: web::Data<LemmyContext>,
94 ) -> Result<HttpResponse, Error> {
95 // TODO: check rate limit here
98 .expect("No auth header for picture upload");
100 if Claims::decode(jwt.value(), &context.secret().jwt_secret).is_err() {
101 return Ok(HttpResponse::Unauthorized().finish());
104 let pictrs_config = context.settings().pictrs_config()?;
105 let image_url = format!("{}image", pictrs_config.url);
107 let mut client_req = adapt_request(&req, &client, image_url);
109 if let Some(addr) = req.head().peer_addr {
110 client_req = client_req.header("X-Forwarded-For", addr.to_string())
114 .body(Body::wrap_stream(make_send(body)))
117 .map_err(error::ErrorBadRequest)?;
119 let status = res.status();
120 let images = res.json::<Images>().await.map_err(error::ErrorBadRequest)?;
122 Ok(HttpResponse::build(status).json(images))
126 filename: web::Path<String>,
127 web::Query(params): web::Query<PictrsParams>,
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())
135 .map_err(error::ErrorBadRequest)?;
136 if local_site.private_instance {
139 .expect("No auth header for picture access");
140 if get_local_user_view_from_jwt(jwt.value(), context.pool(), context.secret())
144 return Ok(HttpResponse::Unauthorized().finish());
147 let name = &filename.into_inner();
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,)
154 // Take file type from name, or jpg if nothing is given
157 .unwrap_or_else(|| name.split('.').last().unwrap_or("jpg").to_string());
159 let mut url = format!("{}image/process.{}?src={}", pictrs_config.url, format, name,);
161 if let Some(size) = params.thumbnail {
162 url = format!("{}&thumbnail={}", url, size,);
167 image(url, req, client).await
173 client: web::Data<ClientWithMiddleware>,
174 ) -> Result<HttpResponse, Error> {
175 let mut client_req = adapt_request(&req, &client, url);
177 if let Some(addr) = req.head().peer_addr {
178 client_req = client_req.header("X-Forwarded-For", addr.to_string());
181 if let Some(addr) = req.head().peer_addr {
182 client_req = client_req.header("X-Forwarded-For", addr.to_string());
185 let res = client_req.send().await.map_err(error::ErrorBadRequest)?;
187 if res.status() == StatusCode::NOT_FOUND {
188 return Ok(HttpResponse::NotFound().finish());
191 let mut client_res = HttpResponse::build(res.status());
193 for (name, value) in res.headers().iter().filter(|(h, _)| *h != "connection") {
194 client_res.insert_header((name.clone(), value.clone()));
197 Ok(client_res.body(BodyStream::new(res.bytes_stream())))
201 components: web::Path<(String, String)>,
203 client: web::Data<ClientWithMiddleware>,
204 context: web::Data<LemmyContext>,
205 ) -> Result<HttpResponse, Error> {
206 let (token, file) = components.into_inner();
208 let pictrs_config = context.settings().pictrs_config()?;
209 let url = format!("{}image/delete/{}/{}", pictrs_config.url, &token, &file);
211 let mut client_req = adapt_request(&req, &client, url);
213 if let Some(addr) = req.head().peer_addr {
214 client_req = client_req.header("X-Forwarded-For", addr.to_string());
217 let res = client_req.send().await.map_err(error::ErrorBadRequest)?;
219 Ok(HttpResponse::build(res.status()).body(BodyStream::new(res.bytes_stream())))
222 fn make_send<S>(mut stream: S) -> impl Stream<Item = S::Item> + Send + Unpin + 'static
224 S: Stream + Unpin + 'static,
227 // NOTE: the 8 here is arbitrary
228 let (tx, rx) = tokio::sync::mpsc::channel(8);
230 // NOTE: spawning stream into a new task can potentially hit this bug:
231 // - https://github.com/actix/actix-web/issues/1679
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() {
246 struct SendStream<T> {
247 rx: tokio::sync::mpsc::Receiver<T>,
250 impl<T> Stream for SendStream<T>
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)